orchestrata_ultimate.plugin.registration

Reference › orchestrata_ultimate › plugin › registration

The functions that carry out the registrations a plugin requests through its PluginContext, which is their only importer. Each takes the plugin id, the arguments, the context's mutable tracking list, the optional target registry or bus (typed by the protocols in interfaces) and the plugin's logger.

register_volk raises PluginRegistrationError if the alias is already tracked for the plugin, forwards to the Volk registry when one is wired, and otherwise logs a warning that registration is deferred. register_tool qualifies the name as plugin.<plugin id>.<name>, rejects duplicates in the same way, and forwards to the tool registry. register_provider and register_channel follow the same pattern for providers and channels. subscribe_event subscribes to an event-bus topic and returns the subscription id; with no bus it generates an id and logs a deferral. emit_event publishes through the bus. register_rules and register_kg_schema record or log the request only: the first appends the rule file names to the tracking list and the second logs the number of node and relationship types; neither loads rules or changes the graph in this module.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Functions

Source

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

"""Registration and event helpers for PluginContext.

Extracted from context.py to keep PluginContext under 300 lines.
Provides module-level async functions for registering Volks, tools,
providers, channels, governance rules, KG schema extensions, and
event subscription/emission.
"""

import logging
import uuid
from collections.abc import Callable
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.plugin.exceptions import PluginRegistrationError
from orchestrata_ultimate.plugin.interfaces import (
    IChannelRegistry,
    IEventBus,
    IProviderRegistry,
    IToolRegistry,
    IVolkRegistry,
)


@traces(
    spec="015",
    story="US-002",
    reqs=["REQ-015-019"],
)
async def register_volk(
    plugin_id: str,
    alias: str,
    capabilities: list[str],
    handler: Callable,
    registered_volks: list[str],
    volk_registry: IVolkRegistry | None,
    log: logging.Logger,
) -> None:
    """Register a Volk (agent) with the kernel.

    Args:
        plugin_id: Owning plugin identifier.
        alias: Unique Volk alias (e.g., "U:BW").
        capabilities: List of Volk capability strings.
        handler: Async callable that handles CXP packets.
        registered_volks: Mutable list tracking registrations.
        volk_registry: Optional registry to forward to.
        log: Plugin-scoped logger.

    Raises:
        PluginRegistrationError: If alias is already taken.
    """
    if alias in registered_volks:
        raise PluginRegistrationError(
            f"Volk '{alias}' already registered by plugin '{plugin_id}'",
            context={"plugin_id": plugin_id, "alias": alias},
        )
    if volk_registry is not None:
        await volk_registry.register_volk(
            alias,
            capabilities,
            handler,
        )
    else:
        log.warning(
            "Volk '%s' registration deferred: no volk_registry wired for plugin '%s'",
            alias,
            plugin_id,
        )
    registered_volks.append(alias)
    log.info(
        "Registered Volk '%s' for plugin '%s'",
        alias,
        plugin_id,
    )


@traces(
    spec="015",
    story="US-003",
    reqs=["REQ-015-031"],
)
async def register_tool(
    plugin_id: str,
    name: str,
    description: str,
    parameters_schema: dict,
    handler: Callable,
    registered_tools: list[str],
    tool_registry: IToolRegistry | None,
    log: logging.Logger,
) -> None:
    """Register a tool in the tool registry.

    Args:
        plugin_id: Owning plugin identifier.
        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.
        registered_tools: Mutable list tracking registrations.
        tool_registry: Optional registry to forward to.
        log: Plugin-scoped logger.

    Raises:
        PluginRegistrationError: If tool name is already taken.
    """
    qualified_name = f"plugin.{plugin_id}.{name}"
    if qualified_name in registered_tools:
        raise PluginRegistrationError(
            f"Tool '{qualified_name}' already registered",
            context={
                "plugin_id": plugin_id,
                "tool_name": qualified_name,
            },
        )
    if tool_registry is not None:
        await tool_registry.register_tool(
            qualified_name,
            description,
            parameters_schema,
            handler,
        )
    else:
        log.warning(
            "Tool '%s' registration deferred: no tool_registry wired for plugin '%s'",
            qualified_name,
            plugin_id,
        )
    registered_tools.append(qualified_name)
    log.info(
        "Registered tool '%s' for plugin '%s'",
        qualified_name,
        plugin_id,
    )


@traces(
    spec="015",
    story="US-004",
    reqs=["REQ-015-046"],
)
async def register_provider(
    plugin_id: str,
    name: str,
    provider: Any,
    registered_providers: list[str],
    provider_registry: IProviderRegistry | None,
    log: logging.Logger,
) -> None:
    """Register an LLM provider with the ProviderManager.

    Args:
        plugin_id: Owning plugin identifier.
        name: Provider name.
        provider: Provider instance implementing IProvider.
        registered_providers: Mutable list tracking registrations.
        provider_registry: Optional registry to forward to.
        log: Plugin-scoped logger.

    Raises:
        PluginRegistrationError: If provider name is taken.
    """
    if name in registered_providers:
        raise PluginRegistrationError(
            f"Provider '{name}' already registered by plugin '{plugin_id}'",
            context={
                "plugin_id": plugin_id,
                "provider_name": name,
            },
        )
    if provider_registry is not None:
        await provider_registry.register_provider(
            name,
            provider,
        )
    else:
        log.warning(
            "Provider '%s' registration deferred: "
            "no provider_registry wired for plugin '%s'",
            name,
            plugin_id,
        )
    registered_providers.append(name)
    log.info(
        "Registered provider '%s' for plugin '%s'",
        name,
        plugin_id,
    )


@traces(
    spec="015",
    story="US-012",
    reqs=["REQ-015-156"],
)
async def register_channel(
    plugin_id: str,
    channel_id: str,
    inbound_handler: Callable,
    outbound_handler: Callable,
    registered_channels: list[str],
    channel_registry: IChannelRegistry | None,
    log: logging.Logger,
) -> None:
    """Register a messaging channel adapter.

    Args:
        plugin_id: Owning plugin identifier.
        channel_id: Unique channel identifier.
        inbound_handler: Async callable for inbound messages.
        outbound_handler: Async callable for outbound messages.
        registered_channels: Mutable list tracking registrations.
        channel_registry: Optional registry to forward to.
        log: Plugin-scoped logger.

    Raises:
        PluginRegistrationError: If channel ID is taken.
    """
    if channel_id in registered_channels:
        raise PluginRegistrationError(
            f"Channel '{channel_id}' already registered",
            context={
                "plugin_id": plugin_id,
                "channel_id": channel_id,
            },
        )
    if channel_registry is not None:
        await channel_registry.register_channel(
            channel_id,
            inbound_handler,
            outbound_handler,
        )
    else:
        log.warning(
            "Channel '%s' registration deferred: "
            "no channel_registry wired for plugin '%s'",
            channel_id,
            plugin_id,
        )
    registered_channels.append(channel_id)
    log.info(
        "Registered channel '%s' for plugin '%s'",
        channel_id,
        plugin_id,
    )


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

    Args:
        plugin_id: Owning plugin identifier.
        rule_files: Paths relative to plugin directory.
        registered_rules: Mutable list tracking registrations.
        log: Plugin-scoped logger.
    """
    for rule_file in rule_files:
        registered_rules.append(rule_file)
    log.info(
        "Registered %d Prolog rules for plugin '%s'",
        len(rule_files),
        plugin_id,
    )


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

    Args:
        plugin_id: Owning plugin identifier.
        node_types: New node labels (auto-prefixed).
        relationship_types: New relationship types.
        indexes: Index definitions.
        log: Plugin-scoped logger.
    """
    log.info(
        "Registered KG schema extension for plugin '%s': "
        "%d node types, %d relationship types",
        plugin_id,
        len(node_types),
        len(relationship_types),
    )


@traces(
    spec="015",
    story="US-011",
    reqs=["REQ-015-146"],
)
async def subscribe_event(
    plugin_id: str,
    topic: str,
    handler: Callable,
    event_subscriptions: list[str],
    event_bus: IEventBus | None,
    log: logging.Logger,
) -> str:
    """Subscribe to a kernel EventBus topic.

    Args:
        plugin_id: Owning plugin identifier.
        topic: Event topic with optional wildcard.
        handler: Async callable for event handling.
        event_subscriptions: Mutable list tracking subscriptions.
        event_bus: Optional event bus to forward to.
        log: Plugin-scoped logger.

    Returns:
        Subscription ID for later unsubscription.
    """
    if event_bus is not None:
        sub_id = await event_bus.subscribe(topic, handler)
    else:
        sub_id = str(uuid.uuid4())
        log.warning(
            "Event subscription for '%s' deferred: no event_bus wired for plugin '%s'",
            topic,
            plugin_id,
        )
    event_subscriptions.append(sub_id)
    log.info(
        "Subscribed to '%s' for plugin '%s' (sub=%s)",
        topic,
        plugin_id,
        sub_id,
    )
    return sub_id


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

    Args:
        plugin_id: Owning plugin identifier.
        event_type: Event type name (auto-prefixed).
        payload: Event payload dict.
        event_bus: Optional event bus to forward to.
        log: Plugin-scoped logger.
    """
    namespaced = f"plugin.{plugin_id}.{event_type}"
    if event_bus is not None:
        await event_bus.emit(namespaced, payload)
    else:
        log.warning(
            "Event '%s' emission deferred: no event_bus wired for plugin '%s'",
            namespaced,
            plugin_id,
        )
    log.info(
        "Emitting event '%s' from plugin '%s'",
        namespaced,
        plugin_id,
    )