orchestrata_ultimate.plugin.manager

Reference › orchestrata_ultimate › plugin › manager

PluginManager is the lifecycle orchestrator of the Plugin SDK (SPEC-015). It implements the manager protocol from interfaces, so it is also a lifecycle component. sdk wraps it as the public facade, and the kernel's command-line entry point imports it. The plugin narrative gives the overall design.

The constructor builds a PluginRegistry, a PluginLoader configured with the plugin directories and allowed module prefixes, and a PluginGovernance. It accepts optional registries for Volks, tools, providers and channels and an event bus; without a registry, the matching registration calls are recorded on the plugin context but projected nowhere. initialize merges a runtime configuration dictionary through PluginConfig.from_dict, rebuilds the loader and governance, and runs discovery when auto_discover is set.

load_plugin(name) is the central operation. It raises PluginNotFoundError for an undiscovered name and CapabilityDeniedError when governance refuses the manifest. Otherwise it sets the status to LOADING, imports and instantiates the plugin class through the loader, builds a PluginContext with the plugin's own configuration section, calls the plugin's initialize(context) and registers it as active. Any failure sets the status to ERROR with the message and surfaces as PluginLoadError. start starts every active plugin, marking one that fails as ERROR; stop stops active plugins in reverse order and logs errors. discover_plugins, unload_plugin, reload_plugin, enable_plugin, disable_plugin, install_plugin and uninstall_plugin delegate to the functions in discovery; installs from git land in the first configured plugin directory.

At a glance

Class diagram

Dependency diagram

API

Classes

PluginManager

Bases: IPluginManager · Decorators: traces(spec='015', story='US-010', reqs=['REQ-015-126']), traces(spec='015', story='US-014', reqs=['REQ-015-188']) · Line 49

Concrete plugin manager implementing full lifecycle.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/plugin/manager.py

"""PluginManager — lifecycle orchestrator for all plugins.

Implements IPluginManager to manage the full plugin lifecycle:
discover, validate, load, initialize, start, stop, unload.
Coordinates PluginLoader, PluginRegistry, and PluginGovernance.

Discovery, install, enable/disable are delegated to
orchestrata.plugin.discovery.
"""

import logging
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.protocols import (
    ComponentMetadata,
    ComponentType,
    HealthStatus,
)
from orchestrata_ultimate.plugin.config import PluginConfig
from orchestrata_ultimate.plugin.context import PluginContext
from orchestrata_ultimate.plugin.discovery import disable_plugin as _disable
from orchestrata_ultimate.plugin.discovery import discover_plugins as _discover
from orchestrata_ultimate.plugin.discovery import enable_plugin as _enable
from orchestrata_ultimate.plugin.discovery import install_plugin as _install
from orchestrata_ultimate.plugin.discovery import reload_plugin as _reload
from orchestrata_ultimate.plugin.discovery import uninstall_plugin as _uninstall
from orchestrata_ultimate.plugin.discovery import unload_plugin as _unload
from orchestrata_ultimate.plugin.exceptions import (
    CapabilityDeniedError,
    PluginLoadError,
    PluginNotFoundError,
)
from orchestrata_ultimate.plugin.governance import PluginGovernance
from orchestrata_ultimate.plugin.interfaces import (
    IPluginManager,
    PluginInfo,
    PluginStatus,
)
from orchestrata_ultimate.plugin.loader import PluginLoader
from orchestrata_ultimate.plugin.manifest import PluginManifest
from orchestrata_ultimate.plugin.registry import PluginRegistry

_LOG = logging.getLogger(__name__)


@traces(spec="015", story="US-010", reqs=["REQ-015-126"])
@traces(spec="015", story="US-014", reqs=["REQ-015-188"])
class PluginManager(IPluginManager):
    """Concrete plugin manager implementing full lifecycle.

    Coordinates loading, initialization, start/stop, and
    governance enforcement. Discovery and install operations
    are delegated to orchestrata.plugin.discovery.

    Attributes:
        config: Plugin manager configuration.
        registry: Plugin registry instance.
        loader: Plugin loader for discovery and import.
        governance: PAM++ governance checker.
    """

    def __init__(
        self,
        config: PluginConfig | None = None,
        di_container: Any = None,
        *,
        volk_registry: Any = None,
        tool_registry: Any = None,
        provider_registry: Any = None,
        channel_registry: Any = None,
        event_bus: Any = None,
    ) -> None:
        """Initialize the plugin manager.

        Args:
            config: Plugin manager configuration. Uses defaults
                if not provided.
            di_container: The DI container for creating scoped
                plugin contexts.
            volk_registry: Optional ``IVolkRegistry`` that plugin-registered
                Volks are wired into. When omitted, a plugin's ``register_volk``
                is recorded on the context but not projected anywhere.
            tool_registry: Optional ``IToolRegistry`` for plugin tools.
            provider_registry: Optional ``IProviderRegistry`` for LLM providers.
            channel_registry: Optional ``IChannelRegistry`` for channels.
            event_bus: Optional ``IEventBus`` for plugin event wiring.
        """
        self._config = config or PluginConfig()
        self._di_container = di_container
        self._volk_registry = volk_registry
        self._tool_registry = tool_registry
        self._provider_registry = provider_registry
        self._channel_registry = channel_registry
        self._event_bus = event_bus
        self.registry = PluginRegistry()
        self._loader = PluginLoader(
            plugin_dirs=self._config.plugin_dirs,
            allowed_prefixes=self._config.allowed_prefixes,
        )
        self.governance = PluginGovernance(self._config)
        self._initialized = False
        self._started = False

    @traces(
        spec="015",
        story="US-010",
        reqs=["REQ-015-128"],
    )
    async def initialize(self, config: dict[str, Any]) -> None:
        """Initialize the plugin manager.

        Args:
            config: Runtime configuration dict.
        """
        if config:
            merged = PluginConfig.from_dict(config)
            self._config = merged
            self._loader = PluginLoader(
                plugin_dirs=merged.plugin_dirs,
                allowed_prefixes=merged.allowed_prefixes,
            )
            self.governance = PluginGovernance(merged)
        self._initialized = True
        _LOG.info("PluginManager initialized")
        if self._config.auto_discover:
            await self.discover_plugins()

    @traces(
        spec="015",
        story="US-010",
        reqs=["REQ-015-129"],
    )
    async def start(self) -> None:
        """Start the plugin manager and all active plugins."""
        self._started = True
        for info in self.registry.list_plugins():
            if info.status == PluginStatus.ACTIVE:
                plugin = self.registry.get_plugin(info.name)
                if plugin is not None:
                    try:
                        await plugin.start()
                        _LOG.info(
                            "Started plugin '%s'",
                            info.name,
                        )
                    except Exception as exc:
                        _LOG.error(
                            "Failed to start plugin '%s': %s",
                            info.name,
                            exc,
                        )
                        self.registry.set_status(
                            info.name,
                            PluginStatus.ERROR,
                            error=str(exc),
                        )
        _LOG.info("PluginManager started")

    @traces(
        spec="015",
        story="US-010",
        reqs=["REQ-015-130"],
    )
    async def stop(self) -> None:
        """Stop the plugin manager and all active plugins."""
        plugins = list(
            reversed(self.registry.list_plugins()),
        )
        for info in plugins:
            if info.status == PluginStatus.ACTIVE:
                plugin = self.registry.get_plugin(info.name)
                if plugin is not None:
                    try:
                        await plugin.stop()
                        _LOG.info(
                            "Stopped plugin '%s'",
                            info.name,
                        )
                    except Exception as exc:
                        _LOG.error(
                            "Error stopping plugin '%s': %s",
                            info.name,
                            exc,
                        )
        self._started = False
        _LOG.info("PluginManager stopped")

    @traces(
        spec="015",
        story="US-014",
        reqs=["REQ-015-200"],
    )
    async def get_health(self) -> HealthStatus:
        """Get the health status of the plugin manager."""
        if not self._initialized:
            return HealthStatus.UNHEALTHY
        error_count = sum(
            1 for p in self.registry.list_plugins() if p.status == PluginStatus.ERROR
        )
        if error_count > 0:
            return HealthStatus.DEGRADED
        return HealthStatus.HEALTHY

    def get_metadata(self) -> ComponentMetadata:
        """Get plugin manager metadata."""
        return ComponentMetadata(
            id="plugin-manager",
            name="PluginManager",
            version="0.7.0",
            component_type=ComponentType.SERVICE,
            description="Plugin lifecycle orchestrator",
        )

    @traces(spec="015", story="US-001", reqs=["REQ-015-006"])
    async def discover_plugins(
        self,
    ) -> list[PluginManifest]:
        """Discover available plugins from all sources."""
        return _discover(self._loader, self.registry)

    @traces(
        spec="015",
        story="US-014",
        reqs=["REQ-015-193"],
    )
    async def load_plugin(self, name: str) -> None:
        """Load and initialize a plugin by name.

        Args:
            name: Plugin name from manifest.

        Raises:
            PluginNotFoundError: If not discovered.
            CapabilityDeniedError: If governance denies loading.
            PluginLoadError: If import or init fails.
        """
        manifest = self.registry.get_manifest(name)
        if manifest is None:
            raise PluginNotFoundError(
                f"Plugin '{name}' not discovered",
                context={"name": name},
            )
        if not self.governance.is_plugin_allowed(manifest):
            raise CapabilityDeniedError(
                f"Governance denied loading plugin '{name}'",
                context={"name": name},
            )
        self.registry.set_status(name, PluginStatus.LOADING)
        try:
            plugin_cls = self._loader.load_plugin_class(
                manifest,
            )
            plugin_instance = plugin_cls()
            plugin_config = self._config.plugin_configs.get(
                name,
                {},
            )
            context = PluginContext(
                plugin_id=name,
                di_container=self._di_container,
                config=plugin_config,
                manifest=manifest,
                plugin_dir=manifest.source_path,
                volk_registry=self._volk_registry,
                tool_registry=self._tool_registry,
                provider_registry=self._provider_registry,
                channel_registry=self._channel_registry,
                event_bus=self._event_bus,
            )
            await plugin_instance.initialize(context)
            source = f"local:{manifest.source_path}" if manifest.source_path else "pip"
            self.registry.register(
                name=name,
                plugin=plugin_instance,
                manifest=manifest,
                context=context,
                source=source,
            )
            _LOG.info(
                "Loaded plugin '%s' v%s",
                name,
                manifest.version,
            )
        except (
            PluginLoadError,
            CapabilityDeniedError,
        ) as exc:
            self.registry.set_status(
                name,
                PluginStatus.ERROR,
                error=str(exc),
            )
            raise
        except Exception as exc:
            self.registry.set_status(
                name,
                PluginStatus.ERROR,
                error=str(exc),
            )
            raise PluginLoadError(
                f"Failed to load plugin '{name}': {exc}",
                context={"name": name},
            ) from exc

    @traces(
        spec="015",
        story="US-014",
        reqs=["REQ-015-194"],
    )
    async def unload_plugin(self, name: str) -> None:
        """Stop and unload a plugin."""
        await _unload(name, self.registry)

    @traces(
        spec="015",
        story="US-014",
        reqs=["REQ-015-195"],
    )
    async def reload_plugin(self, name: str) -> None:
        """Hot-reload a plugin."""
        await _reload(
            name,
            self.registry,
            self.load_plugin,
        )

    @traces(
        spec="015",
        story="US-014",
        reqs=["REQ-015-192"],
    )
    def get_plugin(self, name: str) -> PluginInfo | None:
        """Get plugin info by name."""
        return self.registry.get_info(name)

    @traces(
        spec="015",
        story="US-014",
        reqs=["REQ-015-191"],
    )
    def list_plugins(self) -> list[PluginInfo]:
        """List all known plugins with status."""
        return self.registry.list_plugins()

    @traces(
        spec="015",
        story="US-007",
        reqs=["REQ-015-085"],
    )
    async def enable_plugin(self, name: str) -> None:
        """Enable a disabled plugin."""
        await _enable(
            name,
            self.registry,
            self.load_plugin,
        )

    @traces(
        spec="015",
        story="US-007",
        reqs=["REQ-015-086"],
    )
    async def disable_plugin(self, name: str) -> None:
        """Disable a plugin without uninstalling."""
        await _disable(name, self.registry)

    @traces(
        spec="015",
        story="US-001",
        reqs=["REQ-015-002", "REQ-015-003"],
    )
    async def install_plugin(
        self,
        source: str,
    ) -> PluginManifest:
        """Install a plugin from a local directory or a git repository.

        Git clones land in the manager's first configured plugin directory, so
        the installed plugin is on the loader's discovery path.
        """
        plugin_dirs = self._config.plugin_dirs
        install_dir = plugin_dirs[0] if plugin_dirs else None
        return _install(source, self.registry, install_dir=install_dir)

    @traces(
        spec="015",
        story="US-007",
        reqs=["REQ-015-088"],
    )
    async def uninstall_plugin(self, name: str) -> None:
        """Uninstall a plugin completely."""
        await _uninstall(name, self.registry)