"""Plugin registry — manages discovered, installed, and active plugins.""" from __future__ import annotations import importlib import logging import uuid from pathlib import Path from typing import Any from fastapi import FastAPI from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession from app.core.event_bus import EventBus, get_event_bus from app.core.service_container import ServiceContainer, get_container from app.models.plugin import Plugin as PluginModel from app.plugins.base import BasePlugin from app.plugins.migration_runner import MigrationRunner logger = logging.getLogger(__name__) class PluginRegistry: """Registry for managing plugin lifecycle. Maintains in-memory state for discovered plugins and their runtime instances, backed by database records for persistent status tracking. """ def __init__(self) -> None: # name -> BasePlugin instance (discovered/loaded) self._plugins: dict[str, BasePlugin] = {} # name -> PluginModel DB record (installed status) self._db_status: dict[str, PluginModel] = {} self._engine: AsyncEngine | None = None self._event_bus: EventBus = get_event_bus() self._container: ServiceContainer = get_container() self._migration_runner: MigrationRunner | None = None self._app: FastAPI | None = None self._initialized = False def initialize(self, engine: AsyncEngine, app: FastAPI | None = None) -> None: """Initialize the registry with the DB engine and optional FastAPI app.""" self._engine = engine self._app = app self._migration_runner = MigrationRunner(engine) self._initialized = True @property def migration_runner(self) -> MigrationRunner: if self._migration_runner is None: raise RuntimeError("Registry not initialized — call initialize() first") return self._migration_runner @property def engine(self) -> AsyncEngine: if self._engine is None: raise RuntimeError("Registry not initialized — call initialize() first") return self._engine # ── Discovery ── def discover_builtins(self) -> list[str]: """Discover and load built-in plugins from app.plugins.builtins. Scans the builtins package for modules that export a BasePlugin subclass. Returns list of discovered plugin names. """ discovered: list[str] = [] try: builtins_pkg = importlib.import_module("app.plugins.builtins") except ImportError: return discovered pkg_path = getattr(builtins_pkg, "__path__", None) if pkg_path is None: return discovered import pkgutil for _importer, modname, _ispkg in pkgutil.iter_modules(pkg_path): if modname.startswith("_"): continue try: full_name = f"app.plugins.builtins.{modname}" module = importlib.import_module(full_name) # Look for BasePlugin subclass in module for attr_name in dir(module): attr = getattr(module, attr_name) if ( isinstance(attr, type) and issubclass(attr, BasePlugin) and attr is not BasePlugin ): instance = attr() if instance.name not in self._plugins: self._plugins[instance.name] = instance discovered.append(instance.name) logger.info(f"Discovered plugin: {instance.name} v{instance.version}") except Exception as exc: logger.warning(f"Failed to load builtin plugin module {modname}: {exc}") return discovered def discover_external(self) -> list[str]: """Discover plugins from an external plugins/ directory. Scans the directory specified by EXTERNAL_PLUGINS_PATH env var (default: 'plugins/') for plugin packages. Returns list of discovered plugin names. """ import os import sys discovered: list[str] = [] external_dir = Path(os.environ.get("EXTERNAL_PLUGINS_PATH", "plugins")) if not external_dir.exists(): return discovered for plugin_dir in sorted(external_dir.iterdir()): if not plugin_dir.is_dir() or plugin_dir.name.startswith("_"): continue # Look for plugin.py or __init__.py plugin_file = plugin_dir / "plugin.py" init_file = plugin_dir / "__init__.py" if not plugin_file.exists() and not init_file.exists(): continue try: # Add to sys.path temporarily str_dir = str(external_dir) if str_dir not in sys.path: sys.path.insert(0, str_dir) module_name = f"{plugin_dir.name}.plugin" if plugin_file.exists() else plugin_dir.name module = importlib.import_module(module_name) # Look for BasePlugin subclass for attr_name in dir(module): attr = getattr(module, attr_name) if ( isinstance(attr, type) and issubclass(attr, BasePlugin) and attr is not BasePlugin ): instance = attr() if instance.name not in self._plugins: self._plugins[instance.name] = instance discovered.append(instance.name) logger.info(f"Discovered external plugin: {instance.name} v{instance.version}") except Exception as exc: logger.warning(f"Failed to load external plugin {plugin_dir.name}: {exc}") return discovered def discover_all(self) -> list[str]: """Discover built-in AND external plugins.""" discovered = self.discover_builtins() discovered.extend(self.discover_external()) return discovered def register_plugin(self, plugin: BasePlugin) -> None: """Manually register a plugin instance.""" self._plugins[plugin.name] = plugin logger.info(f"Registered plugin: {plugin.name} v{plugin.version}") def get_plugin(self, name: str) -> BasePlugin | None: """Get a registered plugin instance by name.""" return self._plugins.get(name) def list_discovered(self) -> list[str]: """List all discovered plugin names.""" return list(self._plugins.keys()) # ── Notification Type Sync ── async def sync_notification_types(self, db: AsyncSession) -> None: """Sync notification types from all active plugins to DB. For each active plugin, calls get_notification_types() and upserts the returned types into the notification_types table. Removes types belonging to plugins that are no longer active. """ from app.models.notification import NotificationType # Collect all notification types from active plugins active_types: dict[str, dict[str, Any]] = {} # type_key -> type_data for name, plugin in self._plugins.items(): record = await self._get_plugin_record(db, name) if record is None or not record.active: continue for nt in plugin.get_notification_types(): key = nt.get("type_key", "") if not key: continue active_types[key] = { "type_key": key, "plugin_name": name, "category": nt.get("category", "general"), "label": nt.get("label", key), "description": nt.get("description"), "is_enabled_by_default": nt.get("is_enabled_by_default", True), } # Fetch existing types from DB result = await db.execute(select(NotificationType)) existing = {row.type_key: row for row in result.scalars().all()} # Upsert active types for key, data in active_types.items(): if key in existing: row = existing[key] row.plugin_name = data["plugin_name"] row.category = data["category"] row.label = data["label"] row.description = data["description"] row.is_enabled_by_default = data["is_enabled_by_default"] else: new_row = NotificationType( type_key=key, plugin_name=data["plugin_name"], category=data["category"], label=data["label"], description=data["description"], is_enabled_by_default=data["is_enabled_by_default"], ) db.add(new_row) # Remove types from plugins that are no longer active active_keys = set(active_types.keys()) for key, row in existing.items(): if key not in active_keys: # Check if the plugin is still active plugin_name = row.plugin_name plugin_record = await self._get_plugin_record(db, plugin_name) if plugin_record is None or not plugin_record.active: await db.delete(row) await db.flush() logger.info("Synced %d notification types from active plugins", len(active_types)) # ── DB Status Sync ── async def sync_db_status(self, db: AsyncSession) -> None: """Load plugin status records from the database.""" result = await db.execute(select(PluginModel)) self._db_status = {row.name: row for row in result.scalars().all()} def get_db_status(self, name: str) -> PluginModel | None: """Get the DB status record for a plugin.""" return self._db_status.get(name) # ── Dependency Resolution ── async def _check_dependencies_installed(self, db: AsyncSession, name: str) -> None: """Verify that all declared dependencies of a plugin are installed. Raises ValueError listing missing dependencies. """ plugin = self.get_plugin(name) if plugin is None: raise ValueError(f"Plugin '{name}' not found in registry") dependencies = plugin.manifest.dependencies if not dependencies: return missing: list[str] = [] for dep_name in dependencies: dep_record = await self._get_plugin_record(db, dep_name) if dep_record is None or not dep_record.installed: missing.append(dep_name) if missing: raise ValueError( f"Plugin '{name}' has uninstalled dependencies: {', '.join(missing)}. " f"Install them first." ) async def _check_dependencies_active(self, db: AsyncSession, name: str) -> None: """Verify that all declared dependencies of a plugin are active. Raises ValueError listing inactive dependencies. """ plugin = self.get_plugin(name) if plugin is None: raise ValueError(f"Plugin '{name}' not found in registry") dependencies = plugin.manifest.dependencies if not dependencies: return inactive: list[str] = [] for dep_name in dependencies: dep_record = await self._get_plugin_record(db, dep_name) if dep_record is None or not dep_record.active: inactive.append(dep_name) if inactive: raise ValueError( f"Plugin '{name}' has inactive dependencies: {', '.join(inactive)}. " f"Activate them first." ) # ── Topological Sort & Dependents ── def resolve_load_order(self) -> list[str]: """Return plugin names in dependency-aware load order using Kahn's algorithm. Core plugins (is_core=True) are emitted first, then non-core plugins in topological order based on declared dependencies. Raises RuntimeError on circular dependencies or missing dependencies. """ names = set(self._plugins.keys()) # Validate that all declared dependencies exist in the registry for name in names: plugin = self._plugins[name] for dep in plugin.manifest.dependencies: if dep not in names: raise RuntimeError( f"Plugin '{name}' depends on '{dep}' which is not discovered. " f"Available plugins: {', '.join(sorted(names))}" ) # Build adjacency: dep -> list of plugins that depend on it # and in-degree: plugin -> count of unsatisfied dependencies in_degree: dict[str, int] = {name: 0 for name in names} dependents_map: dict[str, list[str]] = {name: [] for name in names} for name in names: plugin = self._plugins[name] for dep in plugin.manifest.dependencies: dependents_map[dep].append(name) in_degree[name] += 1 # Single Kahn pass with priority: core plugins before non-core at each level. # Queue entries are tuples: (priority, name) where priority 0 = core, 1 = non-core. import heapq heap: list[tuple[int, str]] = [] for name in names: if in_degree[name] == 0: priority = 0 if self._plugins[name].manifest.is_core else 1 heapq.heappush(heap, (priority, name)) result: list[str] = [] while heap: _, current = heapq.heappop(heap) result.append(current) for dependent in dependents_map[current]: in_degree[dependent] -= 1 if in_degree[dependent] == 0: priority = 0 if self._plugins[dependent].manifest.is_core else 1 heapq.heappush(heap, (priority, dependent)) if len(result) != len(names): unresolved = names - set(result) raise RuntimeError( f"Circular dependency detected among plugins: {', '.join(sorted(unresolved))}" ) return result def get_dependents(self, name: str) -> list[str]: """Return list of discovered plugin names that depend on the given plugin. This examines manifest.dependencies of all discovered plugins. """ dependents: list[str] = [] for plugin_name, plugin in self._plugins.items(): if name in plugin.manifest.dependencies: dependents.append(plugin_name) return sorted(dependents) # ── Permission Validation (soft check) ── def _check_permissions(self, name: str) -> list[str]: """Soft-check that declared permissions exist in the system. Returns a list of warning messages for permissions that are not recognised in the available system permissions set. Does not raise — this is a warning-only check. """ plugin = self.get_plugin(name) if plugin is None: return [] declared = plugin.manifest.permissions if not declared: return [] # Build a set of all known permission strings from every discovered plugin available: set[str] = set() for p in self._plugins.values(): available.update(p.manifest.permissions) warnings: list[str] = [] for perm in declared: if perm not in available: warnings.append( f"Plugin '{name}' declares permission '{perm}' " f"which is not found in any discovered plugin's permissions." ) return warnings # ── Plugin Contribution Registration ── async def _register_contributions_for_plugin(self, db: AsyncSession, name: str, plugin: BasePlugin) -> None: """Register agent_definitions, automation_templates, cron_jobs, and heartbeat_configs from a plugin's manifest with the automation plugin.""" automation_plugin = self._plugins.get("automation") if automation_plugin is None: return # Check if automation plugin is active automation_record = await self._get_plugin_record(db, "automation") if automation_record is None or not automation_record.active: return # Check if the activating plugin has any contributions manifest = plugin.manifest has_contributions = bool( manifest.agent_definitions or manifest.automation_templates or manifest.cron_jobs or manifest.heartbeat_configs ) if not has_contributions: return try: await automation_plugin.register_plugin_contributions(db, name, manifest) logger.info( "Registered contributions from plugin '%s' with automation plugin", name, ) except Exception: logger.exception( "Failed to register contributions from plugin '%s' with automation plugin", name, ) async def _unregister_contributions_for_plugin(self, db: AsyncSession, name: str) -> None: """Unregister contributions from a plugin being deactivated.""" automation_plugin = self._plugins.get("automation") if automation_plugin is None: return try: await automation_plugin.unregister_plugin_contributions(db, name) logger.info( "Unregistered contributions from plugin '%s' with automation plugin", name, ) except Exception: logger.exception( "Failed to unregister contributions from plugin '%s' with automation plugin", name, ) # ── Version Comparison & Update Path ── async def _check_and_run_version_migrations( self, db: AsyncSession, name: str, record: PluginModel ) -> bool: """Compare manifest version with DB version and run migrations if different. If the plugin's manifest version differs from the stored DB version, re-run all migrations to bring the schema up to date, then update the DB record's version field. Returns True if migrations were run, False otherwise. """ plugin = self.get_plugin(name) if plugin is None: return False manifest_version = plugin.manifest.version db_version = record.version if manifest_version == db_version: return False logger.info( f"Plugin '{name}' version mismatch: DB={db_version}, manifest={manifest_version}. " f"Running migrations to update." ) # Re-run migrations to apply any new migration files — use migration engine (crm_migration) for DDL if plugin.manifest.migrations: from app.core.db import get_migration_session_factory mig_factory = get_migration_session_factory() async with mig_factory() as mig_db: await self.migration_runner.run_all_migrations(mig_db, name, plugin.manifest.migrations) await mig_db.commit() # Update DB version to match manifest record.version = manifest_version await db.flush() self._db_status[name] = record logger.info(f"Plugin '{name}' updated to version {manifest_version}.") return True # ── Install / Activate / Deactivate / Uninstall ── async def install(self, db: AsyncSession, name: str) -> PluginModel: """Install a plugin: run migrations and create DB record. Idempotent: if already installed, checks for version updates and returns existing record. Checks that all declared dependencies are installed first. """ plugin = self.get_plugin(name) if plugin is None: raise ValueError(f"Plugin '{name}' not found in registry") # Check if already installed in DB existing = await self._get_plugin_record(db, name) if existing is not None: # Check for version update — run migrations if version changed await self._check_and_run_version_migrations(db, name, existing) return existing # Check app version compatibility from app.config import get_settings from app.plugins.semver import SemVer settings = get_settings() app_version = getattr(settings, "app_version", "0.0.0") min_version = plugin.manifest.min_app_version if min_version and min_version != "0.0.0": try: if not SemVer.parse(app_version).is_compatible_with(SemVer.parse(min_version)): raise ValueError( f"Plugin '{name}' requires LeoCRM >= {min_version}, " f"but current version is {app_version}" ) except ValueError as e: if "Invalid semver" in str(e): pass # Skip check if version is not valid SemVer else: raise # Check dependencies are installed await self._check_dependencies_installed(db, name) # Run migrations — use migration engine (crm_migration) for DDL if plugin.manifest.migrations: from app.core.db import get_migration_session_factory mig_factory = get_migration_session_factory() async with mig_factory() as mig_db: await self.migration_runner.run_all_migrations(mig_db, name, plugin.manifest.migrations) await mig_db.commit() # Call on_install hook await plugin.on_install(db, self._container) # Create DB record record = PluginModel( name=name, display_name=plugin.manifest.display_name, version=plugin.manifest.version, status="installed", installed=True, active=False, is_core=plugin.manifest.is_core, ) db.add(record) await db.flush() self._db_status[name] = record return record async def activate(self, db: AsyncSession, name: str) -> PluginModel: """Activate a plugin: register routes, event listeners, set status=active. Idempotent: if already active, checks for version updates and returns existing record without error. Checks that all declared dependencies are active first. Performs a soft permission check and logs warnings for unknown permissions. """ plugin = self.get_plugin(name) if plugin is None: raise ValueError(f"Plugin '{name}' not found in registry") record = await self._get_plugin_record(db, name) if record is None: raise ValueError(f"Plugin '{name}' is not installed — install first") # Check for version update — run migrations if version changed await self._check_and_run_version_migrations(db, name, record) # Idempotent: already active if record.active and record.status == "active": return record # Check dependencies are active await self._check_dependencies_active(db, name) # Soft permission check — log warnings for unknown permissions perm_warnings = self._check_permissions(name) for warning in perm_warnings: logger.warning(warning) # Register permissions and entity models BEFORE on_activate (ARCH-001 fix): # the activation hook may already rely on its own permissions/entities being # resolvable (e.g. tools declaring required_permission). from app.core.permission_registry import ( get_permission_registry, register_plugin_permissions, ) if plugin.manifest.permissions: register_plugin_permissions(name, plugin.manifest.permissions) get_permission_registry()._active_plugins.add(name) for entity_type, model_class in plugin.get_entity_models().items(): from app.services.entity_permission_service import register_entity_model register_entity_model(entity_type, model_class, plugin_name=name) # Call on_activate hook (registers event listeners) await plugin.on_activate(db, self._container, self._event_bus) # Register plugin contributions (agent_definitions, automation_templates, etc.) # with the automation plugin if it is active await self._register_contributions_for_plugin(db, name, plugin) # NOTE: Routes are registered statically in main.py at app creation time. # Activation status is enforced per-request via require_active_plugin(). # No dynamic route registration here — prevents duplicate routes. # Update DB record FIRST — sync_notification_types() only picks up types # from plugins whose record is already active; running it before the # status update silently skipped the freshly activated plugin's types. record.status = "active" record.active = True await db.flush() self._db_status[name] = record # Sync notification types from this plugin (AFTER status update) await self.sync_notification_types(db) # Invalidate Redis cache for this plugin across all tenants try: from app.core.redis import get_redis redis = get_redis() if redis is not None: async for key in redis.scan_iter(f"plugin-activation:*:{name}"): await redis.delete(key) except Exception: pass # Cache invalidation is best-effort return record async def deactivate(self, db: AsyncSession, name: str) -> PluginModel: """Deactivate a plugin: unregister event listeners, set status=inactive. Idempotent: if already inactive, returns existing record without error. """ plugin = self.get_plugin(name) if plugin is None: raise ValueError(f"Plugin '{name}' not found in registry") record = await self._get_plugin_record(db, name) if record is None: raise ValueError(f"Plugin '{name}' is not installed") # Idempotent: already inactive if not record.active and record.status == "inactive": return record # Reject deactivation of core plugins if plugin.manifest.is_core or record.is_core: raise ValueError( f"Plugin '{name}' is a core plugin and cannot be deactivated" ) # Reject deactivation if other active plugins depend on it dependents = self.get_dependents(name) active_dependents: list[str] = [] for dep_name in dependents: dep_record = await self._get_plugin_record(db, dep_name) if dep_record is not None and dep_record.active: active_dependents.append(dep_name) if active_dependents: raise ValueError( f"Plugin '{name}' cannot be deactivated because active plugins " f"depend on it: {', '.join(active_dependents)}" ) # Unregister plugin contributions from automation plugin await self._unregister_contributions_for_plugin(db, name) # Call on_deactivate hook (unregisters event listeners) await plugin.on_deactivate(db, self._container, self._event_bus) # NOTE: Route removal is not implemented — routes stay mounted at runtime. # The Gate model is used: require_active_plugin() returns 403 for inactive # plugins. This is documented and intentional (see architecture-cleanup-plan.md). # Update DB record record.status = "inactive" record.active = False await db.flush() self._db_status[name] = record # Re-sync notification types so entries of the deactivated plugin are # removed from the DB (ARCH-015) — must run AFTER the status update. try: await self.sync_notification_types(db) except Exception: logger.exception("Failed to sync notification types after deactivating '%s'", name) # Invalidate Redis cache for this plugin across all tenants try: from app.core.redis import get_redis redis = get_redis() if redis is not None: async for key in redis.scan_iter(f"plugin-activation:*:{name}"): await redis.delete(key) except Exception: pass # Cache invalidation is best-effort return record async def uninstall( self, db: AsyncSession, name: str, remove_data: bool = False ) -> PluginModel: """Uninstall a plugin: deactivate, optionally drop tables, remove DB record. Args: db: Async database session. name: Plugin name to uninstall. remove_data: If True, drop all plugin-created tables. Returns: The plugin record before deletion (for response). Raises: ValueError if plugin not installed. """ plugin = self.get_plugin(name) if plugin is None: raise ValueError(f"Plugin '{name}' not found in registry") record = await self._get_plugin_record(db, name) if record is None: raise ValueError(f"Plugin '{name}' is not installed") # Deactivate first if active if record.active: await self.deactivate(db, name) # Refetch record after deactivate record = await self._get_plugin_record(db, name) if record is None: raise ValueError(f"Plugin '{name}' disappeared during uninstall") # Call on_uninstall hook await plugin.on_uninstall(db, self._container) # Optionally drop plugin tables — use migration engine (crm_migration) for DDL dropped_tables: list[str] = [] if remove_data: from app.core.db import get_migration_session_factory mig_factory = get_migration_session_factory() async with mig_factory() as mig_db: dropped_tables = await self.migration_runner.drop_plugin_tables(mig_db, name) await mig_db.commit() # Remove DB record await db.delete(record) await db.flush() self._db_status.pop(name, None) # Return a detached copy for response record_dropped_tables = dropped_tables record.status = "uninstalled" record.dropped_tables = record_dropped_tables return record async def list_plugins(self, db: AsyncSession) -> list[dict[str, Any]]: """List all plugins with their current status. Merges discovered (in-memory) plugins with installed (DB) records. """ result = await db.execute(select(PluginModel)) db_records = {row.name: row for row in result.scalars().all()} plugins_list: list[dict[str, Any]] = [] for name, plugin in self._plugins.items(): record = db_records.get(name) if record is not None: plugins_list.append( { "name": name, "display_name": record.display_name, "version": record.version, "status": record.status, "installed": record.installed, "active": record.active, "description": plugin.manifest.description, "dependencies": plugin.manifest.dependencies, "events": plugin.manifest.events, "migrations": plugin.manifest.migrations, "permissions": plugin.manifest.permissions, } ) else: plugins_list.append( { "name": name, "display_name": plugin.manifest.display_name, "version": plugin.version, "status": "discovered", "installed": False, "active": False, "description": plugin.manifest.description, "dependencies": plugin.manifest.dependencies, "events": plugin.manifest.events, "migrations": plugin.manifest.migrations, "permissions": plugin.manifest.permissions, } ) # Also include DB-only records (plugins that were installed but no longer discovered) for name, record in db_records.items(): if name not in self._plugins: plugins_list.append( { "name": name, "display_name": record.display_name, "version": record.version, "status": record.status, "installed": record.installed, "active": record.active, "description": "", "dependencies": [], "events": [], "migrations": [], "permissions": [], } ) return plugins_list # ── Active UI Manifests (Phase 3) ── async def get_active_manifests( self, db: AsyncSession, tenant_id: uuid.UUID | None = None ) -> list[dict[str, Any]]: """Return UI manifests for all active plugins. Each entry contains the plugin name and its frontend UI contributions (menu_items, page_routes, detail_tabs, settings_pages, dashboard_widgets). Audit P1 (tenant manifests): when *tenant_id* is given, plugins that are deactivated for that tenant (tenant_plugin_activation.is_active=False) are excluded — the manifest output must mirror require_active_plugin() semantics so the UI never offers menus/routes the backend then blocks with 403. No tenant row = default active (same as the API gate). """ result = await db.execute(select(PluginModel).where(PluginModel.active.is_(True))) active_records = {row.name: row for row in result.scalars().all()} # Per-tenant deactivations (same table/semantics as deps.require_active_plugin) tenant_disabled: set[str] = set() if tenant_id is not None: from sqlalchemy import text as sa_text rows = await db.execute( sa_text( "SELECT plugin_name FROM tenant_plugin_activation " "WHERE tenant_id = :tid AND is_active = false" ), {"tid": tenant_id}, ) tenant_disabled = {row[0] for row in rows} manifests: list[dict[str, Any]] = [] for name, plugin in self._plugins.items(): if name not in active_records: continue if name in tenant_disabled: continue m = plugin.manifest manifests.append( { "name": m.name, "display_name": m.display_name, "version": m.version, "is_core": m.is_core, "menu_items": [item.model_dump() for item in m.menu_items], "page_routes": [route.model_dump() for route in m.page_routes], "detail_tabs": [tab.model_dump() for tab in m.detail_tabs], "settings_pages": [page.model_dump() for page in m.settings_pages], "dashboard_widgets": [widget.model_dump() for widget in m.dashboard_widgets], "custom_fields": [cf.model_dump() for cf in m.custom_fields], } ) return manifests # ── Internal Helpers ── async def _get_plugin_record(self, db: AsyncSession, name: str) -> PluginModel | None: """Fetch a plugin record from DB by name.""" result = await db.execute(select(PluginModel).where(PluginModel.name == name)) return result.scalar_one_or_none() # Global registry instance _registry: PluginRegistry | None = None def get_registry() -> PluginRegistry: """Get the global plugin registry.""" global _registry if _registry is None: _registry = PluginRegistry() return _registry def reset_registry_for_testing() -> PluginRegistry: """Create a fresh registry for testing.""" global _registry _registry = PluginRegistry() return _registry