orchestrata_ultimate.plugin.registries

Reference › orchestrata_ultimate › plugin › registries

Concrete registries that give a plugin's registration calls somewhere real to land. The protocols they satisfy are in interfaces; without them a registration is only recorded on the PluginContext.

VolkRegistry is an in-memory store of RegisteredVolk records (alias, capabilities, handler, plugin id) keyed by alias, guarded by a lock because plugins may register from any thread. register_volk registers or replaces an alias, unregister is idempotent, and get_volk, list_volks, __contains__ and __len__ query it. sdk creates one and wires it into the manager.

ToolRegistryAdapter adapts the tool registry of the tools package to the plugin tool protocol, so a plugin's tool is registered in the same registry that tool search and dispatch already query. Each registration wraps the plugin's handler in _PluginToolAdapter, a BaseTool whose metadata uses the utility category and whose execute calls the handler and turns a non-ToolResult return value into a successful ToolResult holding its string form. Registration replaces an existing tool of the same name.

At a glance

Class diagram

Dependency diagram

API

Classes

RegisteredVolk (dataclass)

Decorators: traces(spec='015', story='US-002', reqs=['REQ-015-016']), dataclass(frozen=True) · Line 29

A Volk a plugin registered through the plugin context.

VolkRegistry

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

In-memory IVolkRegistry: stores plugin-registered Volks by alias.

_PluginToolAdapter

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

Wraps one plugin register_tool call as a dispatchable BaseTool.

ToolRegistryAdapter

Decorators: traces(spec='015', story='US-003', reqs=['REQ-015-031']) · Line 135

Adapts the SPEC-062 concrete ToolRegistry to the plugin SDK's IToolRegistry protocol.

Source

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

"""Concrete in-memory capability registries that plugins register into.

The plugin SDK defines registry *protocols* (``interfaces.py``); a plugin's
``initialize`` calls ``context.register_volk(...)`` etc. Until now no concrete
registry existed and ``PluginManager`` never wired one in, so those calls were a
silent no-op. ``VolkRegistry`` is a real, queryable implementation: a loaded
plugin's Volk registrations land here and can be looked up. (Projecting a
registry's contents onto the live kernel dispatcher is a further step.)
"""

from __future__ import annotations

import threading
from dataclasses import dataclass
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.tools.base import (
    BaseTool,
    ToolCategory,
    ToolMetadata,
    ToolResult,
)
from orchestrata_ultimate.tools.registry import ToolRegistry


@traces(spec="015", story="US-002", reqs=["REQ-015-016"])
@dataclass(frozen=True)
class RegisteredVolk:
    """A Volk a plugin registered through the plugin context."""

    alias: str
    capabilities: tuple[str, ...]
    handler: Any
    plugin_id: str | None = None


@traces(spec="015", story="US-002", reqs=["REQ-015-017"])
class VolkRegistry:
    """In-memory ``IVolkRegistry``: stores plugin-registered Volks by alias.

    Thread-safe: every access is serialized through a single lock, since
    plugins may register/unregister Volks from arbitrary threads under
    free-threaded Python.
    """

    def __init__(self) -> None:
        self._volks: dict[str, RegisteredVolk] = {}
        self._lock = threading.Lock()

    @traces(spec="015", story="US-002", reqs=["REQ-015-016"])
    async def register_volk(
        self,
        alias: str,
        capabilities: list[str],
        handler: Any,
        plugin_id: str | None = None,
    ) -> None:
        """Register (or replace) a Volk under ``alias``."""
        with self._lock:
            self._volks[alias] = RegisteredVolk(
                alias=alias,
                capabilities=tuple(capabilities),
                handler=handler,
                plugin_id=plugin_id,
            )

    @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 isn't (or is no longer)
        registered is not an error.
        """
        with self._lock:
            self._volks.pop(alias, None)

    def get_volk(self, alias: str) -> RegisteredVolk | None:
        """Look up a registered Volk by alias (``None`` if absent)."""
        with self._lock:
            return self._volks.get(alias)

    def list_volks(self) -> list[str]:
        """List the aliases of every registered Volk."""
        with self._lock:
            return list(self._volks)

    def __contains__(self, alias: str) -> bool:
        with self._lock:
            return alias in self._volks

    def __len__(self) -> int:
        with self._lock:
            return len(self._volks)


@traces(spec="015", story="US-003", reqs=["REQ-015-034"])
class _PluginToolAdapter(BaseTool):
    """Wraps one plugin ``register_tool`` call as a dispatchable ``BaseTool``.

    A plugin's ``context.register_tool(name, description, parameters_schema,
    handler)`` carries a bare async ``handler(arguments)``; this adapter gives
    it the ``BaseTool`` shape (``metadata`` + ``execute``) the SPEC-062
    ``ToolRegistry`` requires, normalizing the handler's return value to a
    ``ToolResult`` when it isn't already one.
    """

    def __init__(
        self,
        name: str,
        description: str,
        parameters_schema: dict,
        handler: Any,
    ) -> None:
        self._metadata = ToolMetadata(
            name=name,
            description=description,
            category=ToolCategory.UTILITY,
            parameters=parameters_schema or {},
        )
        self._handler = handler

    @property
    def metadata(self) -> ToolMetadata:
        return self._metadata

    async def execute(self, arguments: dict[str, Any]) -> ToolResult:
        result = await self._handler(arguments)
        if isinstance(result, ToolResult):
            return result
        return ToolResult(self._metadata.name, True, str(result))


@traces(spec="015", story="US-003", reqs=["REQ-015-031"])
class ToolRegistryAdapter:
    """Adapts the SPEC-062 concrete ``ToolRegistry`` to the plugin SDK's
    ``IToolRegistry`` protocol.

    Until now no concrete ``IToolRegistry`` implementation existed either, so a
    plugin's ``register_tool`` call was recorded on the ``PluginContext`` but
    projected nowhere (the same gap ``VolkRegistry`` closed for Volks). Backing
    it with ``tools.registry.default_registry()`` means a plugin's tool lands
    in the SAME registry ``ToolSearchTool``/dispatch already query.
    """

    def __init__(self, tool_registry: ToolRegistry) -> None:
        self._tool_registry = tool_registry

    @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 ``handler`` as a real, dispatchable tool named ``name``."""
        self._tool_registry.register(
            _PluginToolAdapter(name, description, parameters_schema, handler),
            replace=True,
        )