orchestrata_ultimate.plugin.context

Reference › orchestrata_ultimate › plugin › context

PluginContext is what a plugin receives in its initialize() method: its identity, its own configuration section, its manifest, its directory, a logger and the registration API. The PluginManager builds one per plugin and the PluginRegistry keeps it.

get_config(key, default) reads from the plugin's configuration. resolve(interface_type) returns a component from the DI container, but only if the manifest declares the di_access capability; otherwise it raises CapabilityDeniedError. The registration methods are register_volk, register_tool, register_provider, register_channel, register_rules, register_kg_schema, subscribe_event and emit_event. Each one forwards to the matching function in registration, together with the context's own bookkeeping list and the registry or event bus that the manager wired in. cleanup deregisters every Volk this context registered from the wired Volk registry, so a stopped plugin's alias is no longer dispatchable, and clears all tracking lists; the registry calls it when a plugin is removed.

At a glance

Class diagram

Dependency diagram

API

Classes

PluginContext

Line 40

Runtime context given to each plugin during initialization.

Source

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

"""PluginContext — DI-backed context for plugin authors.

Provides the public API surface that plugins use to register
Volks, tools, providers, channels, governance rules, and KG
schema extensions. Wraps the DI container in a scoped proxy
that enforces declared capabilities.

Registration logic is delegated to orchestrata.plugin.registration.
"""

import logging
from collections.abc import Callable
from pathlib import Path
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.plugin.exceptions import (
    CapabilityDeniedError,
)
from orchestrata_ultimate.plugin.interfaces import (
    IChannelRegistry,
    IEventBus,
    IProviderRegistry,
    IToolRegistry,
    IVolkRegistry,
)
from orchestrata_ultimate.plugin.manifest import PluginManifest
from orchestrata_ultimate.plugin.registration import emit_event as _emit_event
from orchestrata_ultimate.plugin.registration import register_channel as _reg_channel
from orchestrata_ultimate.plugin.registration import (
    register_kg_schema as _reg_kg_schema,
)
from orchestrata_ultimate.plugin.registration import register_provider as _reg_provider
from orchestrata_ultimate.plugin.registration import register_rules as _reg_rules
from orchestrata_ultimate.plugin.registration import register_tool as _reg_tool
from orchestrata_ultimate.plugin.registration import register_volk as _reg_volk
from orchestrata_ultimate.plugin.registration import subscribe_event as _sub_event


class PluginContext:
    """Runtime context given to each plugin during initialization.

    Provides DI resolution scoped to the plugin's declared
    capabilities, configuration access, and registration APIs
    for Volks, tools, providers, channels, and governance rules.

    Registration calls are delegated to module-level helpers in
    orchestrata.plugin.registration.

    Attributes:
        plugin_id: Unique plugin name.
        config: Plugin-specific configuration dict.
        manifest: The plugin's validated manifest.
        plugin_dir: Filesystem path to the plugin directory.
    """

    def __init__(
        self,
        plugin_id: str,
        di_container: Any,
        config: dict[str, Any],
        manifest: PluginManifest,
        plugin_dir: Path | None = None,
        volk_registry: IVolkRegistry | None = None,
        tool_registry: IToolRegistry | None = None,
        provider_registry: IProviderRegistry | None = None,
        channel_registry: IChannelRegistry | None = None,
        event_bus: IEventBus | None = None,
    ) -> None:
        """Initialize the plugin context.

        Args:
            plugin_id: Unique plugin name.
            di_container: The DI container (or scoped proxy).
            config: Plugin-specific configuration dict.
            manifest: Validated plugin manifest.
            plugin_dir: Filesystem path to the plugin directory.
            volk_registry: Optional registry for wiring Volks.
            tool_registry: Optional registry for wiring tools.
            provider_registry: Optional registry for wiring
                LLM providers.
            channel_registry: Optional registry for wiring
                messaging channels.
            event_bus: Optional event bus for wiring event
                subscriptions and emissions.
        """
        self.plugin_id = plugin_id
        self._di_container = di_container
        self.config = config
        self.manifest = manifest
        self.plugin_dir = plugin_dir or Path(".")
        self.logger = logging.getLogger(
            f"orchestrata_ultimate.plugin.{plugin_id}",
        )
        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._registered_volks: list[str] = []
        self._registered_tools: list[str] = []
        self._registered_providers: list[str] = []
        self._registered_channels: list[str] = []
        self._registered_rules: list[str] = []
        self._event_subscriptions: list[str] = []

    @traces(
        spec="015",
        story="US-009",
        reqs=["REQ-015-107", "REQ-015-108"],
    )
    def resolve(self, interface_type: type) -> Any:
        """Resolve a component from the scoped DI container.

        Args:
            interface_type: The interface type to resolve.

        Returns:
            The resolved component instance.

        Raises:
            CapabilityDeniedError: If the plugin lacks the
                di_access capability.
        """
        if "di_access" not in self.manifest.capabilities:
            raise CapabilityDeniedError(
                f"Plugin '{self.plugin_id}' lacks 'di_access' "
                f"capability required to resolve DI components",
                context={
                    "plugin_id": self.plugin_id,
                    "interface": interface_type.__name__,
                },
            )
        return self._di_container.resolve(interface_type)

    @traces(
        spec="015",
        story="US-008",
        reqs=["REQ-015-099"],
    )
    def get_config(self, key: str, default: Any = None) -> Any:
        """Get a configuration value by key.

        Args:
            key: Configuration key.
            default: Value to return if key is not found.

        Returns:
            The configuration value or default.
        """
        return self.config.get(key, default)

    @traces(
        spec="015",
        story="US-002",
        reqs=["REQ-015-016"],
    )
    async def register_volk(
        self,
        alias: str,
        capabilities: list[str],
        handler: Callable,
    ) -> None:
        """Register a Volk (agent) with the kernel.

        Args:
            alias: Unique Volk alias (e.g., "U:BW").
            capabilities: List of Volk capability strings.
            handler: Async callable that handles CXP packets.
        """
        await _reg_volk(
            self.plugin_id,
            alias,
            capabilities,
            handler,
            self._registered_volks,
            self._volk_registry,
            self.logger,
        )

    @traces(
        spec="015",
        story="US-003",
        reqs=["REQ-015-031"],
    )
    async def register_tool(
        self,
        name: str,
        description: str,
        parameters_schema: dict,
        handler: Callable,
    ) -> None:
        """Register a tool in the tool registry.

        Args:
            name: Tool name (without namespace prefix).
            description: Human-readable tool description.
            parameters_schema: JSON Schema for tool parameters.
            handler: Async callable that executes the tool.
        """
        await _reg_tool(
            self.plugin_id,
            name,
            description,
            parameters_schema,
            handler,
            self._registered_tools,
            self._tool_registry,
            self.logger,
        )

    @traces(
        spec="015",
        story="US-004",
        reqs=["REQ-015-046"],
    )
    async def register_provider(
        self,
        name: str,
        provider: Any,
    ) -> None:
        """Register an LLM provider with the ProviderManager.

        Args:
            name: Provider name.
            provider: Provider instance implementing IProvider.
        """
        await _reg_provider(
            self.plugin_id,
            name,
            provider,
            self._registered_providers,
            self._provider_registry,
            self.logger,
        )

    @traces(
        spec="015",
        story="US-012",
        reqs=["REQ-015-156"],
    )
    async def register_channel(
        self,
        channel_id: str,
        inbound_handler: Callable,
        outbound_handler: Callable,
    ) -> None:
        """Register a messaging channel adapter.

        Args:
            channel_id: Unique channel identifier.
            inbound_handler: Async callable for inbound messages.
            outbound_handler: Async callable for outbound messages.
        """
        await _reg_channel(
            self.plugin_id,
            channel_id,
            inbound_handler,
            outbound_handler,
            self._registered_channels,
            self._channel_registry,
            self.logger,
        )

    @traces(
        spec="015",
        story="US-005",
        reqs=["REQ-015-061"],
    )
    async def register_rules(
        self,
        rule_files: list[str],
    ) -> None:
        """Load Prolog governance rule files.

        Args:
            rule_files: Paths relative to plugin directory.
        """
        await _reg_rules(
            self.plugin_id,
            rule_files,
            self._registered_rules,
            self.logger,
        )

    @traces(
        spec="015",
        story="US-006",
        reqs=["REQ-015-071"],
    )
    async def register_kg_schema(
        self,
        node_types: list[str],
        relationship_types: list[str],
        indexes: list[str],
    ) -> None:
        """Extend the knowledge graph schema.

        Args:
            node_types: New node labels (auto-prefixed).
            relationship_types: New relationship types.
            indexes: Index definitions.
        """
        await _reg_kg_schema(
            self.plugin_id,
            node_types,
            relationship_types,
            indexes,
            self.logger,
        )

    @traces(
        spec="015",
        story="US-011",
        reqs=["REQ-015-146"],
    )
    async def subscribe_event(
        self,
        topic: str,
        handler: Callable,
    ) -> str:
        """Subscribe to a kernel EventBus topic.

        Args:
            topic: Event topic with optional wildcard.
            handler: Async callable for event handling.

        Returns:
            Subscription ID for later unsubscription.
        """
        return await _sub_event(
            self.plugin_id,
            topic,
            handler,
            self._event_subscriptions,
            self._event_bus,
            self.logger,
        )

    @traces(
        spec="015",
        story="US-011",
        reqs=["REQ-015-151"],
    )
    async def emit_event(
        self,
        event_type: str,
        payload: dict,
    ) -> None:
        """Emit a custom event namespaced to this plugin.

        Args:
            event_type: Event type name (auto-prefixed).
            payload: Event payload dict.
        """
        await _emit_event(
            self.plugin_id,
            event_type,
            payload,
            self._event_bus,
            self.logger,
        )

    @traces(spec="015", story="US-002", reqs=["REQ-015-017"])
    def cleanup(self) -> None:
        """Remove all registrations made by this plugin.

        Actually deregisters every Volk this plugin registered from the
        wired ``IVolkRegistry`` (not just this context's local bookkeeping
        lists), so a stopped plugin's Volk alias is no longer dispatchable.
        """
        if self._volk_registry is not None:
            for alias in self._registered_volks:
                self._volk_registry.unregister(alias)
        self._registered_volks.clear()
        self._registered_tools.clear()
        self._registered_providers.clear()
        self._registered_channels.clear()
        self._registered_rules.clear()
        self._event_subscriptions.clear()
        self.logger.info(
            "Cleaned up all registrations for plugin '%s'",
            self.plugin_id,
        )