orchestrata_ultimate.plugin.interfaces

Reference › orchestrata_ultimate › plugin › interfaces

The contracts of the plugin system. The registry protocols that a plugin's registrations flow into are IVolkRegistry, IToolRegistry, IProviderRegistry, IChannelRegistry and IEventBus (subscribe returns a subscription id; emit publishes an event). registries implements the first two. The enums are PluginCapability (the nine capabilities a manifest can request: di_access, redis, neo4j, network, filesystem, prolog, event_bus, provider, gateway), PluginStatus (DISCOVERED, LOADING, ACTIVE, ERROR, DISABLED) and ResourceType (memory bytes, API calls per hour, events per hour). PluginResourceUsage and PluginInfo are the runtime records; PluginInfo holds the name, version, status, source, load time, last error and the aliases and names of what the plugin registered.

IPlugin is the protocol every plugin implements: initialize(context), start, stop and get_manifest. IPluginManager extends the core component protocol with discover, load, unload, reload, get, list, enable, disable, install and uninstall, and is implemented by the PluginManager.

At a glance

Class diagram

Dependency diagram

API

Classes

IVolkRegistry (protocol)

Bases: Protocol · Decorators: traces(spec='015', story='US-002', reqs=['REQ-015-017']) · Line 21

Protocol for registering Volk agents with the kernel.

IToolRegistry (protocol)

Bases: Protocol · Decorators: traces(spec='015', story='US-003', reqs=['REQ-015-034']) · Line 58

Protocol for registering tools in the tool registry.

IProviderRegistry (protocol)

Bases: Protocol · Decorators: traces(spec='015', story='US-004', reqs=['REQ-015-048']) · Line 85

Protocol for registering LLM providers.

IChannelRegistry (protocol)

Bases: Protocol · Decorators: traces(spec='015', story='US-012', reqs=['REQ-015-157']) · Line 108

Protocol for registering messaging channel adapters.

IEventBus (protocol)

Bases: Protocol · Line 132

Protocol for kernel event bus access.

PluginCapability (enum)

Bases: Enum · Decorators: traces(spec='015', story='US-009', reqs=['REQ-015-110']) · Line 172

Capabilities a plugin can request access to.

PluginStatus (enum)

Bases: Enum · Line 202

Plugin lifecycle status.

ResourceType (enum)

Bases: StrEnum · Decorators: traces(spec='015', story='US-009', reqs=['REQ-015-116', 'REQ-015-117', 'REQ-015-118']) · Line 221

Resource types for plugin resource limiting.

PluginResourceUsage (dataclass)

Decorators: traces(spec='015', story='US-009', reqs=['REQ-015-116', 'REQ-015-117', 'REQ-015-118', 'REQ-015-122']), dataclass · Line 245

Tracks resource consumption for a single plugin.

PluginInfo (dataclass)

Decorators: traces(spec='015', story='US-007', reqs=['REQ-015-082', 'REQ-015-084']), dataclass · Line 263

Runtime information about a loaded or discovered plugin.

IPlugin (protocol)

Bases: Protocol · Line 305

Protocol that all plugins must implement.

IPluginManager (protocol)

Bases: IComponent, Protocol · Decorators: traces(spec='015', story='US-010', reqs=['REQ-015-126']) · Line 354

Protocol for plugin lifecycle and registry management.

Source

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

"""Plugin SDK interfaces and protocols.

Defines the IPlugin protocol that all plugins must implement,
the IPluginManager abstract base class for lifecycle management,
registry protocols for wiring plugin registrations to real systems,
and supporting enums/dataclasses for plugin state.
"""

from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum, StrEnum
from typing import Any, Protocol

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.protocols import (
    IComponent,
)


@traces(spec="015", story="US-002", reqs=["REQ-015-017"])
class IVolkRegistry(Protocol):
    """Protocol for registering Volk agents with the kernel.

    Implementations wire plugin-registered Volks into the
    actual agent runtime.
    """

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

        Args:
            alias: Unique Volk alias (e.g., "U:BW").
            capabilities: List of capability strings.
            handler: Async callable that handles CXP packets.
        """
        ...

    @traces(spec="015", story="US-002", reqs=["REQ-015-017"])
    def unregister(self, alias: str) -> None:
        """Remove a Volk registration by alias.

        Idempotent: unregistering an alias that is not registered is not
        an error.

        Args:
            alias: Volk alias to remove.
        """
        ...


@traces(spec="015", story="US-003", reqs=["REQ-015-034"])
class IToolRegistry(Protocol):
    """Protocol for registering tools in the tool registry.

    Implementations wire plugin-registered tools into the
    actual tool execution system.
    """

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

        Args:
            name: Qualified tool name.
            description: Human-readable description.
            parameters_schema: JSON Schema for parameters.
            handler: Async callable that executes the tool.
        """
        ...


@traces(spec="015", story="US-004", reqs=["REQ-015-048"])
class IProviderRegistry(Protocol):
    """Protocol for registering LLM providers.

    Implementations wire plugin-registered providers into
    the ProviderManager.
    """

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

        Args:
            name: Provider name.
            provider: Provider instance implementing IProvider.
        """
        ...


@traces(spec="015", story="US-012", reqs=["REQ-015-157"])
class IChannelRegistry(Protocol):
    """Protocol for registering messaging channel adapters.

    Implementations wire plugin-registered channels into
    the messaging subsystem.
    """

    @traces(spec="015", story="US-012", reqs=["REQ-015-156"])
    async def register_channel(
        self,
        channel_id: str,
        inbound_handler: Any,
        outbound_handler: Any,
    ) -> 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.
        """
        ...


class IEventBus(Protocol):
    """Protocol for kernel event bus access.

    Implementations wire plugin event subscriptions and
    emissions into the actual kernel EventBus.
    """

    @traces(spec="015", story="US-011", reqs=["REQ-015-146"])
    async def subscribe(
        self,
        topic: str,
        handler: Any,
    ) -> str:
        """Subscribe to an event topic.

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

        Returns:
            Subscription ID for later unsubscription.
        """
        ...

    @traces(spec="015", story="US-011", reqs=["REQ-015-151"])
    async def emit(
        self,
        event_type: str,
        payload: dict,
    ) -> None:
        """Emit an event.

        Args:
            event_type: Namespaced event type.
            payload: Event payload dict.
        """
        ...


@traces(spec="015", story="US-009", reqs=["REQ-015-110"])
class PluginCapability(Enum):
    """Capabilities a plugin can request access to.

    Each capability gates access to a specific system resource
    or DI component family. Plugins must declare required
    capabilities in their manifest.

    Attributes:
        DI_ACCESS: Access to the DI container.
        REDIS: Direct Redis access.
        NEO4J: Direct Neo4j access.
        NETWORK: Outbound network access.
        FILESYSTEM: Filesystem read/write.
        PROLOG: Prolog knowledge base access.
        EVENT_BUS: Kernel EventBus access.
        PROVIDER: IProviderManager access.
        GATEWAY: IGatewayServer access.
    """

    DI_ACCESS = "di_access"
    REDIS = "redis"
    NEO4J = "neo4j"
    NETWORK = "network"
    FILESYSTEM = "filesystem"
    PROLOG = "prolog"
    EVENT_BUS = "event_bus"
    PROVIDER = "provider"
    GATEWAY = "gateway"


class PluginStatus(Enum):
    """Plugin lifecycle status.

    Attributes:
        DISCOVERED: Manifest found, not yet loaded.
        LOADING: Currently being loaded.
        ACTIVE: Successfully loaded and running.
        ERROR: Plugin encountered an error.
        DISABLED: Explicitly disabled by operator.
    """

    DISCOVERED = "discovered"
    LOADING = "loading"
    ACTIVE = "active"
    ERROR = "error"
    DISABLED = "disabled"


@traces(spec="015", story="US-009", reqs=["REQ-015-116", "REQ-015-117", "REQ-015-118"])
class ResourceType(StrEnum):
    """Resource types for plugin resource limiting.

    Used by PluginGovernance.get_resource_limit to ensure
    callers pass valid resource identifiers instead of
    arbitrary strings.

    Attributes:
        MEMORY_BYTES: Memory consumption in bytes.
        API_CALLS_PER_HOUR: API call rate limit.
        EVENTS_PER_HOUR: Event emission rate limit.
    """

    MEMORY_BYTES = "memory_bytes"
    API_CALLS_PER_HOUR = "api_calls_per_hour"
    EVENTS_PER_HOUR = "events_per_hour"


@traces(
    spec="015",
    story="US-009",
    reqs=["REQ-015-116", "REQ-015-117", "REQ-015-118", "REQ-015-122"],
)
@dataclass
class PluginResourceUsage:
    """Tracks resource consumption for a single plugin.

    Attributes:
        memory_bytes: Current memory usage in bytes.
        api_calls_last_hour: API calls in the last hour.
        events_emitted_last_hour: Events emitted in the last hour.
        errors_last_hour: Errors encountered in the last hour.
    """

    memory_bytes: int = 0
    api_calls_last_hour: int = 0
    events_emitted_last_hour: int = 0
    errors_last_hour: int = 0


@traces(spec="015", story="US-007", reqs=["REQ-015-082", "REQ-015-084"])
@dataclass
class PluginInfo:
    """Runtime information about a loaded or discovered plugin.

    Attributes:
        name: Unique plugin name.
        version: Semantic version string.
        description: Human-readable description.
        author: Plugin author.
        status: Current lifecycle status.
        source: Where the plugin was found.
        loaded_at: Timestamp when the plugin was loaded.
        error: Last error message, if any.
        registered_volks: Aliases of Volks registered by this plugin.
        registered_tools: Names of tools registered by this plugin.
        registered_providers: Names of providers registered.
        registered_channels: IDs of channels registered.
        resource_usage: Current resource consumption.
    """

    name: str
    version: str
    description: str = ""
    author: str = ""
    status: PluginStatus = PluginStatus.DISCOVERED
    source: str = ""
    loaded_at: datetime | None = None
    error: str | None = None
    registered_volks: list[str] = field(default_factory=list)
    registered_tools: list[str] = field(default_factory=list)
    registered_providers: list[str] = field(default_factory=list)
    registered_channels: list[str] = field(default_factory=list)
    resource_usage: PluginResourceUsage = field(
        default_factory=PluginResourceUsage,
    )


# ── Forward reference for PluginManifest ──
# Avoids circular import; manifest module defines the full class.
# We import it here only for type annotations in the protocol.
from orchestrata_ultimate.plugin.manifest import PluginManifest  # noqa: E402


class IPlugin(Protocol):
    """Protocol that all plugins must implement.

    Plugins are Python classes that implement this protocol to
    extend Orchestrata with new Volks, tools, providers, channels,
    governance rules, or knowledge graph extensions.
    """

    @traces(spec="015", story="US-010", reqs=["REQ-015-128"])
    async def initialize(self, context: Any) -> None:
        """Initialize the plugin with a PluginContext.

        Called after the plugin is loaded. Register components
        (Volks, tools, providers, channels, rules) here.

        Args:
            context: PluginContext providing DI, config, and
                registration APIs.
        """
        ...

    @traces(spec="015", story="US-010", reqs=["REQ-015-129"])
    async def start(self) -> None:
        """Start the plugin.

        Called after all plugins have been initialized.
        Begin background tasks or long-running operations here.
        """
        ...

    @traces(spec="015", story="US-010", reqs=["REQ-015-130"])
    async def stop(self) -> None:
        """Stop the plugin.

        Called on system shutdown or plugin disable.
        Clean up resources, cancel tasks, close connections.
        """
        ...

    def get_manifest(self) -> PluginManifest:
        """Return the plugin manifest.

        Returns:
            PluginManifest describing this plugin.
        """
        ...


@traces(spec="015", story="US-010", reqs=["REQ-015-126"])
class IPluginManager(IComponent, Protocol):
    """Protocol for plugin lifecycle and registry management.

    Manages the full plugin lifecycle: discover, validate, load,
    initialize, start, stop, unload, install, and uninstall.
    """

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

        Returns:
            List of discovered plugin manifests.
        """
        ...

    @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: Unique plugin name from its manifest.
        """
        ...

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

        Args:
            name: Plugin name to unload.
        """
        ...

    @traces(spec="015", story="US-014", reqs=["REQ-015-195"])
    async def reload_plugin(self, name: str) -> None:
        """Hot-reload a plugin (stop, unload, load, init, start).

        Args:
            name: Plugin name to reload.
        """
        ...

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

        Args:
            name: Plugin name to look up.

        Returns:
            PluginInfo or None if not found.
        """
        ...

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

        Returns:
            List of PluginInfo for all known plugins.
        """
        ...

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

        Args:
            name: Plugin name to enable.
        """
        ...

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

        Args:
            name: Plugin name to disable.
        """
        ...

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

        Args:
            source: Pip package spec, local path, or git URL.

        Returns:
            The installed plugin's manifest.
        """
        ...

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

        Stops, unloads, and removes the plugin.

        Args:
            name: Plugin name to uninstall.
        """
        ...