# Extensible Framework Design > Designs robust extensible software frameworks with plugin architectures, configuration-driven extension points, and validation patterns — enabling third-party contributors while maintaining core stability and API contract guarantees. - Skill: `paulpas/extensible-framework-design` (Agent Skill) - Install (CLI): `npx skillmds@latest add paulpas/extensible-framework-design` - Raw SKILL.md: https://api.skillmd.com/api/skills/paulpas/extensible-framework-design/raw - Safety review: pending - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: Integrations & APIs - License: MIT - Author: paulpas (https://skillmd.com/u/paulpas) - Updated: 2026-09-17 - Page: https://skillmd.com/skills/paulpas/extensible-framework-design --- # Extensible Framework Design Makes the model design robust extensible software frameworks with plugin architectures, configuration-driven extension points, and validation patterns. When loaded, this skill enforces disciplined separation between core runtime behavior and pluggable extensions — ensuring third-party contributors can extend functionality without breaking backward compatibility or circumventing lifecycle hooks. ## TL;DR Checklist - [ ] Identify all extension points by cataloging where external code must inject behavior - [ ] Define plugin interfaces with strict type contracts — no duck typing at boundaries - [ ] Write configuration schemas (JSON Schema / Pydantic models) that validate every extension's config before loading - [ ] Implement a lifecycle manager that enforces ordered hook execution with error isolation - [ ] Add validation enforcement so misconfigured plugins fail fast during registration, not at runtime - [ ] Draft a backward compatibility plan documenting how plugin interfaces evolve across versions --- ## When to Use Use this skill when: - Building an internal framework that multiple teams or external contributors will extend via plugins or hooks - Designing a middleware chain where processing stages must be independently pluggable and orderable - Creating a configuration-driven system where behavior changes based on deployed extension bundles - Migrating a monolithic application into a plugin-based architecture with clear separation of concerns - Establishing an internal package registry where third-party developers submit extensions for review and distribution --- ## When NOT to Use Avoid this skill for: - **Simple applications** — A 200-line script or microservice with one responsibility needs no plugin architecture - **When extension requirements are unknown** — Premature abstraction creates unnecessary complexity; wait until at least two distinct extensions are needed - **Performance-critical hot paths** — Dynamic plugin resolution adds overhead; use static composition in latency-sensitive code --- ## Core Workflow ### 1. Extension Point Identification — Catalog Where External Code Must Inject Behavior Walk the framework's core domain model and identify every location where behavior varies based on context, customer, or configuration. Each variation point becomes a potential extension point. Record each as an interface with a clear contract: inputs, outputs, side effects, and error semantics. ```python from __future__ import annotations import logging from abc import ABC, abstractmethod from dataclasses import dataclass, field from enum import Enum from typing import Any, Protocol logger = logging.getLogger(__name__) class ExtensionPhase(Enum): """Lifecycle phase at which an extension hook fires.""" BEFORE_INPUT = "before_input" VALIDATE = "validate" PROCESS = "process" POST_PROCESS = "post_process" ERROR = "error" CLEANUP = "cleanup" @dataclass(frozen=True) class ExtensionPoint: """Describes a single location where plugins can inject behavior. Attributes: name: Unique identifier for this extension point within the framework. phase: Lifecycle phase when registered hooks execute. input_type: Expected type of data passed to hooks at this point. output_type: Expected return type from hooks (can be None for side-effect-only hooks). priority_range: Valid range for hook priority values. Lower runs first. error_policy: How framework handles exceptions thrown by registered hooks. """ name: str phase: ExtensionPhase input_type: type | None = None output_type: type | None = None priority_range: tuple[int, int] = (-100, 100) error_policy: str = "isolate" # "isolate", "abort", "continue" def validate_priority(self, priority: int) -> bool: """Check if a priority value falls within the allowed range.""" low, high = self.priority_range return low <= priority <= high class ExtensionHook(Protocol): """Contract that all plugin hooks must satisfy at an extension point.""" def __call__(self, context: Any) -> Any: ... @property def priority(self) -> int: ... @property def name(self) -> str: ... def catalog_extension_points(domain_model: type) -> list[ExtensionPoint]: """Analyze a domain model and identify natural extension points. Scans the provided class for methods that delegate to strategies, read from configuration-driven behavior tables, or contain branching logic based on external input — all indicators of natural plugin boundaries. Args: domain_model: The core framework class to analyze. Returns: List of identified extension points with inferred contracts. """ import inspect from types import FunctionType points: list[ExtensionPoint] = [] analyzed_methods: set[str] = set() for name, method in inspect.getmembers(domain_model, predicate=inspect.isfunction): if name.startswith("_") and name != "__init__": continue if name in analyzed_methods: continue source = inspect.getsource(method) if hasattr(inspect, 'getsource') else "" # Heuristic: methods with strategy dispatch or config lookups are extension points is_extension_candidate = ( "strategy" in source.lower() or "config." in source.lower() or "plugin" in source.lower() or any(keyword in source for keyword in [".get(", "dict[", "case ", "if mode"]) ) if is_extension_candidate: analyzed_methods.add(name) sig = inspect.signature(method) input_type = None output_type = None # Infer input from first parameter (usually self is skipped) params = list(sig.parameters.values()) if len(params) > 1: param = params[1] if param.annotation != inspect.Parameter.empty: input_type = param.annotation if sig.return_annotation != inspect.Parameter.empty: output_type = sig.return_annotation phase = _infer_phase_from_name(name, source) points.append(ExtensionPoint( name=f"{domain_model.__name__}.{name}", phase=phase, input_type=input_type, output_type=output_type, )) return sorted(points, key=lambda p: (p.phase.value, p.name)) def _infer_phase_from_name(method_name: str, source: str) -> ExtensionPhase: """Heuristically infer the lifecycle phase from a method's name or body.""" lower = method_name.lower() if any(kw in lower for kw in ("pre_", "before_", "init_", "prepare_")): return ExtensionPhase.BEFORE_INPUT if any(kw in lower for kw in ("valid", "check_", "verify_")): return ExtensionPhase.VALIDATE if any(kw in lower for kw in ("transform", "process_", "handle_", "execute_")): return ExtensionPhase.PROCESS if any(kw in lower for kw in ("post_", "after_", "cleanup_", "finish_")): return ExtensionPhase.POST_PROCESS return ExtensionPhase.PROCESS # default ``` **Checkpoint:** Every method in your domain model that exhibits branching based on external input has an identified extension point. If a method has no corresponding `ExtensionPoint`, justify why it should remain non-extensible. --- ### 2. Plugin Interface Design — Define Strict Type Contracts Each plugin interface must be a Protocol or ABC with explicit type annotations, documented side effects, and clearly defined error conditions. No duck typing at extension boundaries — interfaces are contracts, not suggestions. ```python from __future__ import annotations import logging from abc import ABC, abstractmethod from dataclasses import dataclass, field from typing import Any, Generic, Protocol, TypeVar logger = logging.getLogger(__name__) T = TypeVar("T") U = TypeVar("U") @dataclass(frozen=True) class PluginMetadata: """Immutable metadata describing a loaded plugin. Attributes: name: Unique plugin identifier (namespace/name format). version: Semantic version string. author: Organization or developer name. supported_versions: Framework version range this plugin supports. requires_config_keys: Set of top-level config keys this plugin expects. """ name: str version: str = "0.0.0" author: str = "" supported_versions: tuple[str, str] = ("*", "*") requires_config_keys: frozenset[str] = field(default_factory=frozenset) class PluginError(Exception): """Raised when a plugin violates its interface contract.""" def __init__(self, plugin_name: str, violation: str, context: dict[str, Any] | None = None) -> None: self.plugin_name = plugin_name self.violation = violation self.context = context or {} super().__init__(f"PluginError[{plugin_name}]: {violation}") class Plugin(ABC, Generic[T, U]): """Abstract base for all framework plugins. Subclasses must implement the lifecycle methods and register themselves with the PluginRegistry upon instantiation via the metaclass hook. Attributes: metadata: Immutable plugin identity and capability description. _initialized: Flag indicating whether on_load has completed successfully. """ def __init__(self, metadata: PluginMetadata) -> None: self.metadata = metadata self._initialized = False @property def name(self) -> str: return self.metadata.name @abstractmethod async def on_load(self, config: dict[str, Any]) -> None: """Called when the plugin is registered. Validate configuration and initialize resources. Args: config: Configuration dictionary provided by the framework loader. Raises: PluginError: If configuration is invalid or required resources are unavailable. """ ... @abstractmethod async def execute(self, input_data: T) -> U: """Execute the plugin's primary transformation logic. Args: input_data: The data payload to process according to the plugin's contract. Returns: Transformed output matching the plugin's declared output type. Raises: PluginError: If the input violates the plugin's preconditions. RuntimeError: For unexpected failures during execution. """ ... @abstractmethod async def on_unload(self) -> None: """Called when the plugin is being removed. Release all held resources.""" ... def validate_input_type(self, data: Any, expected_type: type[T]) -> bool: """Validate that input_data matches the expected plugin input type. Args: data: The actual input to check. expected_type: The type declared by the plugin interface. Returns: True if data is an instance of expected_type. """ if expected_type is Any: return True return isinstance(data, expected_type) # --- Concrete Plugin Example --- @dataclass class TransformInput: raw_value: str metadata: dict[str, str] = field(default_factory=dict) @dataclass class TransformOutput: processed_value: str tags: list[str] = field(default_factory=list) class TextTransformer(Plugin[TransformInput, TransformOutput]): """Example plugin that transforms text with configurable rules. Subclasses override _apply_transform to provide custom logic while inheriting input validation, error handling, and lifecycle management. """ def __init__(self, metadata: PluginMetadata, config: dict[str, Any] | None = None) -> None: super().__init__(metadata) self._config = config or {} self._rules: list[str] = self._config.get("rules", ["uppercase", "strip_whitespace"]) async def on_load(self, config: dict[str, Any]) -> None: if not isinstance(config, dict): raise PluginError( self.name, "Configuration must be a dictionary", {"received_type": type(config).__name__} ) known_keys = {"rules", "encoding", "fallback"} unexpected = set(config.keys()) - known_keys if unexpected: raise PluginError( self.name, f"Unknown configuration keys: {unexpected}", context={"unexpected_keys": list(unexpected)} ) self._rules = config.get("rules", ["uppercase"]) self._initialized = True async def execute(self, input_data: TransformInput) -> TransformOutput: if not self._initialized: raise PluginError(self.name, "Plugin not initialized — call on_load first") if not isinstance(input_data, TransformInput): raise PluginError( self.name, f"Invalid input type", context={"expected": TransformInput.__name__, "received": type(input_data).__name__} ) result = input_data.raw_value for rule in self._rules: if rule == "uppercase": result = result.upper() elif rule == "lowercase": result = result.lower() elif rule == "strip_whitespace": result = result.strip() return TransformOutput(processed_value=result, tags=self._rules) async def on_unload(self) -> None: self._initialized = False self._rules.clear() ``` **Checkpoint:** Every plugin interface has explicit type annotations on all abstract methods, documented side effects in docstrings, and raises `PluginError` for contract violations — not generic exceptions. --- ### 3. Configuration Schema Definition — Validate Extensions Before Loading Every extension point must have an accompanying JSON Schema or Pydantic model that validates the configuration before a plugin is registered. Reject invalid configs at load time, never at runtime. ```python from __future__ import annotations import json import logging from dataclasses import dataclass, field from typing import Any logger = logging.getLogger(__name__) @dataclass(frozen=True) class SchemaValidationError: """Describes a single configuration validation failure.""" key_path: str # Dot-separated path to the invalid field message: str received_value: Any class ConfigSchemaValidator: """Validates plugin configurations against declared schemas. Supports Pydantic models, JSON Schema dictionaries, and custom validators. Rejects configurations that violate the schema before any plugin code runs. """ def __init__(self) -> None: self._schemas: dict[str, Any] = {} self._validators: dict[str, Any] = {} def register_schema(self, extension_name: str, schema: Any, validator_fn: Any | None = None) -> None: """Register a validation schema for a specific extension point. Args: extension_name: The name of the extension this schema applies to. schema: Either a Pydantic BaseModel class or a JSON Schema dictionary. validator_fn: Optional custom validation function taking (config) -> list[SchemaValidationError]. """ self._schemas[extension_name] = schema if validator_fn is not None: self._validators[extension_name] = validator_fn def validate(self, extension_name: str, config: dict[str, Any]) -> list[SchemaValidationError]: """Validate a configuration against the registered schema. Args: extension_name: The extension whose schema to use. config: Configuration dictionary to validate. Returns: List of validation errors (empty if valid). """ if extension_name not in self._schemas: return [SchemaValidationError("", f"No schema registered for extension '{extension_name}'", config)] schema = self._schemas[extension_name] validator_fn = self._validators.get(extension_name) # Custom validator takes priority if validator_fn is not None: try: errors = validator_fn(config) return errors if isinstance(errors, list) else [] except Exception as exc: return [SchemaValidationError("", f"Custom validator failed: {exc}", config)] # JSON Schema validation if isinstance(schema, dict) and "$schema" in schema: return self._validate_json_schema(schema, config) # Pydantic model validation try: model_cls = schema if hasattr(schema, "model_validate") else None if model_cls is not None: model_cls.model_validate(config) # type: ignore[attr-defined] return [] except Exception: pass return [SchemaValidationError("", f"Unsupported schema type: {type(schema).__name__}", config)] def _validate_json_schema(self, schema: dict[str, Any], config: dict[str, Any]) -> list[SchemaValidationError]: """Validate config against a JSON Schema definition. Uses a lightweight approach without external dependencies — checks required fields and basic type constraints. """ errors: list[SchemaValidationError] = [] required_fields = schema.get("required", []) properties = schema.get("properties", {}) for req_key in required_fields: if req_key not in config: errors.append(SchemaValidationError( key_path=req_key, message=f"Required field missing: {req_key}", received_value=None )) # Type checking for provided fields for key, value in config.items(): if key in properties: expected_type_spec = properties[key].get("type") if expected_type_spec and isinstance(value, dict): # Nested object — simple check continue return errors # --- Usage Example --- def create_text_transformer_schema() -> dict[str, Any]: """Define the JSON Schema for the TextTransformer plugin.""" return { "$schema": "http://json-schema.org/draft-07/schema#", "type": "object", "required": ["rules"], "properties": { "rules": { "type": "array", "items": {"type": "string"}, "minItems": 1, "enum": ["uppercase", "lowercase", "strip_whitespace", "title_case"], }, "encoding": { "type": "string", "default": "utf-8", "enum": ["utf-8", "ascii", "latin-1"], }, "fallback": { "type": "string", "default": "", }, }, } # Demonstrate validation if __name__ == "__main__": validator = ConfigSchemaValidator() validator.register_schema("text_transformer", create_text_transformer_schema()) # Valid config valid_config = {"rules": ["uppercase", "strip_whitespace"], "encoding": "utf-8"} errors = validator.validate("text_transformer", valid_config) print(f"Valid config errors: {errors}") # [] # Invalid config — missing required field invalid_config = {"encoding": "ascii"} errors = validator.validate("text_transformer", invalid_config) print(f"Invalid config errors: {[e.message for e in errors]}") ``` **Checkpoint:** Every plugin has a registered schema, and no plugin passes the `on_load` phase without first passing schema validation. Invalid configurations are rejected with specific error messages pointing to exact field paths. --- ### 4. Lifecycle Management — Enforce Ordered Hook Execution with Error Isolation The framework must manage the full plugin lifecycle (load → execute → unload) and ensure hooks at each phase execute in priority order. Exceptions from one hook must not prevent other hooks from running — errors are collected and reported, never swallowed silently. ```python from __future__ import annotations import asyncio import logging from dataclasses import dataclass, field from typing import Any, Callable logger = logging.getLogger(__name__) @dataclass(frozen=True) class HookExecutionResult: """Result of executing a single hook in the lifecycle pipeline.""" hook_name: str phase: str priority: int succeeded: bool output: Any | None = None error: Exception | None = None class LifecycleManager: """Manages ordered execution of hooks across all extension phases. Each phase maintains a sorted list of hook functions by priority. Execution is sequential within a phase but isolated per hook so that one failure does not prevent subsequent hooks from running. Attributes: _phases: Mapping of phase name → list of (priority, callable) tuples, sorted ascending. _registered_hooks: Registry tracking which hooks are active per extension. """ def __init__(self) -> None: self._phases: dict[str, list[tuple[int, Callable[..., Any], str]]] = {} self._registered_hooks: dict[str, list[str]] = {} def register_hook( self, phase_name: str, priority: int, hook_fn: Callable[..., Any], extension_name: str, description: str = "" ) -> None: """Register a hook function for execution at a specific lifecycle phase. Hooks are sorted by priority (ascending) within each phase before execution. Args: phase_name: The lifecycle phase to attach this hook to. priority: Execution order — lower values run first. Must be in the extension point's range. hook_fn: Async callable accepting a shared context dict. extension_name: Owning extension identifier for lifecycle tracking. description: Human-readable name of this hook for logging. """ if phase_name not in self._phases: self._phases[phase_name] = [] # Validate priority range against the registered extension point from framework_extension import get_extension_point # type: ignore try: ep = get_extension_point(phase_name) if not ep.validate_priority(priority): raise ValueError( f"Priority {priority} out of range [{ep.priority_range}] for phase '{phase_name}'" ) except Exception: # If we can't resolve the extension point, log warning but allow registration logger.warning("Could not validate priority %s for phase %s — registering anyway", priority, phase_name) self._phases[phase_name].append((priority, hook_fn, description or f"hook_at_{phase_name}_{len(self._phases[phase_name])}")) self._phases[phase_name].sort(key=lambda x: x[0]) if extension_name not in self._registered_hooks: self._registered_hooks[extension_name] = [] hook_desc = description or f"hook_at_{phase_name}" if hook_desc not in self._registered_hooks[extension_name]: self._registered_hooks[extension_name].append(hook_desc) def unregister_hook(self, extension_name: str) -> None: """Remove all hooks registered by a specific extension. Args: extension_name: The extension whose hooks to remove from all phases. """ for phase_hooks in self._phases.values(): self._phases[phase_name] = [ (p, fn, desc) for p, fn, desc in phase_hooks if desc not in (self._registered_hooks.get(extension_name, [])) ] self._registered_hooks.pop(extension_name, None) async def execute_phase(self, phase_name: str, context: dict[str, Any]) -> list[HookExecutionResult]: """Execute all hooks registered for a phase in priority order. Each hook runs in isolation — exceptions are caught and recorded but do not prevent subsequent hooks from executing. Args: phase_name: The lifecycle phase to execute. context: Shared mutable context dict passed to every hook. Returns: List of HookExecutionResult objects, one per executed hook. """ hooks = self._phases.get(phase_name, []) results: list[HookExecutionResult] = [] for priority, hook_fn, description in hooks: try: output = await hook_fn(context) if asyncio.iscoroutinefunction(hook_fn) else hook_fn(context) # type: ignore[misc] results.append(HookExecutionResult( hook_name=description, phase=phase_name, priority=priority, succeeded=True, output=output )) except Exception as exc: logger.error("Hook '%s' in phase '%s' failed (priority=%d): %s", description, phase_name, priority, exc) results.append(HookExecutionResult( hook_name=description, phase=phase_name, priority=priority, succeeded=False, error=exc )) return results async def execute_all_phases(self, context: dict[str, Any]) -> dict[str, list[HookExecutionResult]]: """Execute all registered phases in their defined order. Args: context: Shared context passed to every phase and hook. Returns: Mapping of phase name → list of execution results. """ ordered_phases = [ ExtensionPhase.BEFORE_INPUT, ExtensionPhase.VALIDATE, ExtensionPhase.PROCESS, ExtensionPhase.POST_PROCESS, ] results: dict[str, list[HookExecutionResult]] = {} for phase in ordered_phases: phase_results = await self.execute_phase(phase.value, context) results[phase.value] = phase_results # Check for abort policy — if any hook failed with error_policy "abort" if any(not r.succeeded and getattr(r.error, "error_policy", "isolate") == "abort" for phase_res in results.values() for r in phase_res): logger.warning("Abort triggered at phase %s — skipping remaining phases", phase.value) break return results # --- Demonstration --- if __name__ == "__main__": async def demo() -> None: manager = LifecycleManager() async def logging_hook(ctx: dict[str, Any]) -> dict[str, Any]: print(f"[HOOK] Logging: {ctx.get('action', 'unknown')}") return {"logged": True} async def transform_hook(ctx: dict[str, Any]) -> dict[str, Any]: value = ctx.get("value", "") ctx["transformed"] = value.upper() return {"transformed": True} # Register hooks with priorities manager.register_hook(ExtensionPhase.VALIDATE.value, priority=10, hook_fn=logging_hook, extension_name="logging_plugin", description="validate-logger") manager.register_hook(ExtensionPhase.PROCESS.value, priority=5, hook_fn=transform_hook, extension_name="transformer_plugin", description="data-transformer") # Execute all phases context = {"value": "hello world", "action": "test_request"} results = await manager.execute_all_phases(context) for phase, hooks in results.items(): print(f"\nPhase: {phase}") for h in hooks: status = "OK" if h.succeeded else f"FAIL ({h.error})" print(f" [{status}] {h.hook_name} (priority={h.priority})") asyncio.run(demo()) ``` **Checkpoint:** Hook execution results are collected and inspectable after each phase. Failed hooks log errors but do not crash the pipeline unless the error policy explicitly requires aborting. Every hook has a unique description for traceability in production logs. --- ### 5. Validation Enforcement — Fail Fast During Registration, Not at Runtime Register every plugin through a central registry that validates: type conformance against the declared interface, schema compliance, and uniqueness of extension point registrations. Plugins that fail validation are rejected with detailed error messages during startup, never silently loaded. ```python from __future__ import annotations import importlib.metadata import logging from dataclasses import dataclass, field from typing import Any logger = logging.getLogger(__name__) @dataclass(frozen=True) class RegistrationResult: """Outcome of a plugin registration attempt. Attributes: success: Whether the plugin was registered successfully. plugin_name: Name of the attempted plugin. errors: List of human-readable error messages (empty on success). warnings: Non-fatal observations that don't block registration. """ success: bool plugin_name: str errors: list[str] = field(default_factory=list) warnings: list[str] = field(default_factory=list) class PluginRegistry: """Central registry for framework plugins with validation enforcement. All plugins must go through this registry before they can be loaded or executed. The registry enforces type conformance, schema validation, and registration uniqueness. Attributes: _plugins: Mapping of plugin name → Plugin instance. _schemas: Mapping of extension point name → validation schema. _validators: ConfigSchemaValidator instance for config validation. """ def __init__(self) -> None: self._plugins: dict[str, Plugin] = {} self._schemas: dict[str, Any] = {} self._validators = ConfigSchemaValidator() def register_schema(self, extension_point_name: str, schema: Any) -> None: """Register a validation schema for an extension point. Args: extension_point_name: Name of the extension point. schema: JSON Schema dict or Pydantic model class. """ self._schemas[extension_point_name] = schema self._validators.register_schema(extension_point_name, schema) def register_plugin(self, plugin: Plugin, config: dict[str, Any] | None = None) -> RegistrationResult: """Validate and register a plugin through the central registry. Performs three validation gates in order: 1. Type conformance — does the plugin implement the Plugin ABC? 2. Schema validation — does its config match the extension's schema? 3. Uniqueness — is there no existing plugin with the same name? Args: plugin: The Plugin instance to register. config: Configuration dictionary for this plugin. Returns: RegistrationResult with detailed success/failure information. """ errors: list[str] = [] warnings: list[str] = [] plugin_name = plugin.name # Gate 1: Type conformance check if not isinstance(plugin, Plugin): errors.append( f"Plugin '{plugin_name}' does not implement the Plugin ABC. " f"Missing abstract methods: {self._missing_abstract_methods(type(plugin))}" ) # Gate 2: Schema validation if config is not None: schema_errors = self._validators.validate(plugin_name, config) for err in schema_errors: errors.append(f"Config validation failed for '{plugin_name}': [{err.key_path}] {err.message}") # Gate 3: Uniqueness check if plugin_name in self._plugins: existing = self._plugins[plugin_name] errors.append( f"Plugin name '{plugin_name}' is already registered " f"(version {existing.metadata.version})" ) # Gate 4: on_load execution (if all gates passed) if not errors and plugin_name not in self._plugins: import asyncio try: loop = asyncio.get_running_loop() except RuntimeError: loop = None if loop: asyncio.run_coroutine_threadsafe(plugin.on_load(config or {}), loop) else: try: import asyncio asyncio.run(plugin.on_load(config or {})) except Exception as load_exc: errors.append(f"Plugin on_load failed for '{plugin_name}': {load_exc}") if errors: return RegistrationResult(success=False, plugin_name=plugin_name, errors=errors, warnings=warnings) self._plugins[plugin_name] = plugin return RegistrationResult(success=True, plugin_name=plugin_name, errors=[], warnings=warnings) def get_plugin(self, name: str) -> Plugin | None: """Retrieve a registered plugin by name. Args: name: The plugin's unique identifier. Returns: The Plugin instance or None if not found. """ return self._plugins.get(name) def list_plugins(self) -> list[str]: """Return names of all currently registered plugins, sorted alphabetically.""" return sorted(self._plugins.keys()) def unregister_plugin(self, name: str) -> RegistrationResult: """Remove a plugin from the registry and call its on_unload hook. Args: name: Plugin identifier to remove. Returns: RegistrationResult indicating success or failure. """ if name not in self._plugins: return RegistrationResult( success=False, plugin_name=name, errors=[f"No plugin registered with name '{name}'"] ) plugin = self._plugins.pop(name) try: import asyncio asyncio.run(plugin.on_unload()) except Exception as exc: return RegistrationResult( success=False, plugin_name=name, errors=[f"Plugin on_unload failed for '{name}': {exc}"] ) return RegistrationResult(success=True, plugin_name=name) def _missing_abstract_methods(self, cls: type) -> list[str]: """Identify which abstract methods are not implemented by a class.""" if not hasattr(ABC, '__abstractmethods__'): return [] try: from abc import ABC missing = [] for attr in getattr(cls, '__abstractmethods__', set()): if not hasattr(cls, attr) or isinstance(getattr(cls, attr), property): missing.append(attr) return missing except Exception: return [] ``` **Checkpoint:** The registry rejects misconfigured plugins at registration time with specific error messages. No plugin can execute unless it passes all four validation gates (type conformance, schema validation, uniqueness, on_load success). --- ### 6. Backward Compatibility Planning — Document How Plugin Interfaces Evolve Across Versions Create a compatibility contract that defines what changes are allowed between major/minor/patch versions. Use version ranges in plugin metadata to declare supported framework versions. Provide migration guides for deprecated interfaces. ```python from __future__ import annotations import re from dataclasses import dataclass, field from enum import Enum from typing import Any class VersionBump(Enum): MAJOR = "major" # Breaking change — plugins may break MINOR = "minor" # New feature — backward compatible PATCH = "patch" # Bug fix — backward compatible @dataclass(frozen=True) class CompatibilityRule: """Defines what kinds of changes are allowed in a version bump. Attributes: plugin_interface_name: The interface this rule applies to. allowed_modifications: Operations permitted without breaking plugins. deprecated_operations: Operations that work but emit warnings. removed_operations: Operations no longer available after this version. migration_guide_url: URL to documentation on how to migrate plugins. """ plugin_interface_name: str allowed_modifications: list[str] = field(default_factory=list) deprecated_operations: list[str] = field(default_factory=list) removed_operations: list[str] = field(default_factory=list) migration_guide_url: str = "" class CompatibilityChecker: """Validates plugin compatibility against declared framework version constraints. Tracks semantic versions of both the framework core and each plugin, enforcing that plugins only load when their supported version range includes the current framework version. """ VERSION_PATTERN = re.compile(r"^(\d+)\.(\d+)\.(\d+)$") def __init__(self) -> None: self._rules: dict[str, CompatibilityRule] = {} self._current_framework_version: str = "0.0.0" def set_framework_version(self, version: str) -> None: """Set the current framework core version for compatibility checks. Args: version: Semantic version string (e.g., "3.2.1"). """ self._validate_semver(version) self._current_framework_version = version def register_rule(self, rule: CompatibilityRule) -> None: """Register a compatibility rule for a plugin interface. Args: rule: The compatibility rule defining allowed/forbidden changes. """ self._rules[rule.plugin_interface_name] = rule def check_plugin_compatibility( self, plugin: Plugin, framework_version: str | None = None ) -> list[str]: """Check whether a plugin is compatible with the current framework version. Args: plugin: The plugin to check. framework_version: Optional override for the framework version (defaults to set value). Returns: List of compatibility issues (empty if fully compatible). """ fw_ver = framework_version or self._current_framework_version issues: list[str] = [] # Parse versions fw_parts = self._parse_semver(fw_ver) plugin_parts = self._parse_semver(plugin.metadata.version) # Check major version — breaking change if plugin targets different major fw_major, fw_minor, _ = fw_parts plugin_major, plugin_minor, _ = plugin_parts if fw_major != plugin_major and plugin_major > 0: issues.append( f"Plugin '{plugin.name}' targets major v{plugin_major}, " f"but framework is at v{fw_major}. Breaking changes likely." ) # Check if the plugin's interface has been modified since its version rule = self._rules.get(plugin.__class__.__name__) if rule: for removed in rule.removed_operations: if hasattr(plugin, removed): issues.append( f"Plugin '{plugin.name}' uses deprecated method '{removed}' " f"which was removed in framework {fw_ver}. " f"Migrate: {rule.migration_guide_url}" ) for deprecated in rule.deprecated_operations: if hasattr(plugin, deprecated): import warnings warnings.warn( f"Plugin '{plugin.name}' uses method '{deprecated}' " f"which is deprecated in framework {fw_ver}.", DeprecationWarning, stacklevel=2 ) return issues def _validate_semver(self, version: str) -> bool: """Validate that a string conforms to semantic versioning …(truncated)