orchestrata_ultimate.gateway.handlers.volk

Reference › orchestrata_ultimate › gateway › handlers › volk

VolkHandler serves the volk.* RPCs: volk.list, volk.get, volk.register, volk.deregister and volk.lifecycle. It manages the gateway's roster of volks (the specialised agents), not the agent processes, which the agent-runtime kernel starts. rpc creates it and the pubsub manager carries its events. See gateway.

load_volk_registry reads the roster from a configs/volks.yaml file located by _find_volks_yaml, and falls back to a built-in default registry if the file is missing or invalid. handle_list returns alias, name, role, health, current task and status for each volk. When a kernel client is bound through set_kernel_client, _live_status merges the live status from the CXP status call, so callers see real busy and idle state; if that query fails the static roster is used. handle_register and handle_deregister change the roster and are admin-gated: _require_admin compares a supplied admin token against the one set with set_admin_token. A duplicate alias or a missing alias raises RPCError. handle_lifecycle subscribes the connection to volk.lifecycle, and publish_lifecycle_event emits registered and deregistered events.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

VolkHandler

Decorators: traces(spec='014', story='US-002', reqs=['REQ-014-016', 'REQ-014-018', 'REQ-014-026', 'REQ-014-028']) · Line 179

Volk registry query and lifecycle subscription.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/gateway/handlers/volk.py

"""Volk RPC handlers -- list, get, register, deregister, lifecycle.

Implements SPEC-014 FR-016 volk management methods.
Loads the volk registry from configs/volks.yaml when available,
falling back to a hardcoded default.
"""

import logging
from datetime import UTC, datetime
from pathlib import Path
from typing import Any

import yaml

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.gateway.exceptions import RPCError
from orchestrata_ultimate.gateway.pubsub import PubSubManager

logger = logging.getLogger(__name__)

_DEFAULT_VOLK_REGISTRY: list[dict[str, Any]] = [
    {
        "alias": "U:Ox",
        "name": "Orchestrator",
        "role": "Executive",
        "status": "active",
        "capabilities": [
            "routing",
            "governance",
            "escalation",
        ],
    },
    {
        "alias": "U:PS",
        "name": "Postman",
        "role": "Router",
        "status": "active",
        "capabilities": [
            "message_delivery",
            "queue_management",
        ],
    },
    {
        "alias": "U:Vz",
        "name": "VizInt",
        "role": "Coder",
        "status": "active",
        "capabilities": [
            "code_generation",
            "artifact_creation",
        ],
    },
    {
        "alias": "U:Ux",
        "name": "UX",
        "role": "Designer",
        "status": "active",
        "capabilities": [
            "ux_review",
            "design_patterns",
        ],
    },
    {
        "alias": "U:PM",
        "name": "PM",
        "role": "Manager",
        "status": "active",
        "capabilities": [
            "timeline",
            "process",
            "planning",
        ],
    },
    {
        "alias": "U:Ar",
        "name": "Architect",
        "role": "Architect",
        "status": "active",
        "capabilities": [
            "architecture",
            "patterns",
            "diagrams",
        ],
    },
    {
        "alias": "U:Qa",
        "name": "QA",
        "role": "Tester",
        "status": "active",
        "capabilities": [
            "test_plans",
            "code_review",
        ],
    },
    {
        "alias": "U:Cx",
        "name": "Concierge",
        "role": "Intake",
        "status": "active",
        "capabilities": [
            "requirements",
            "clarifications",
        ],
    },
    {
        "alias": "U:KR",
        "name": "Kernel",
        "role": "System",
        "status": "active",
        "capabilities": [
            "orchestration",
            "lifecycle",
        ],
    },
]


def load_volk_registry(
    yaml_path: str | Path | None = None,
) -> list[dict[str, Any]]:
    """Load volk definitions from YAML, with fallback.

    Searches for configs/volks.yaml relative to the project
    root.  If the file is missing or invalid, returns the
    hardcoded default.

    Args:
        yaml_path: Explicit path override for testing.

    Returns:
        List of volk definition dicts.
    """
    if yaml_path is None:
        candidate = _find_volks_yaml()
    else:
        candidate = Path(yaml_path)

    if candidate and candidate.is_file():
        try:
            with open(candidate) as fh:
                data = yaml.safe_load(fh)
            volks = data.get("volks", [])
            if volks:
                logger.info(
                    "Loaded %d volks from %s",
                    len(volks),
                    candidate,
                )
                return list(volks)
        except Exception:
            logger.warning(
                "Failed to load %s, using defaults",
                candidate,
            )
    return list(_DEFAULT_VOLK_REGISTRY)


def _find_volks_yaml() -> Path | None:
    """Walk up from this file to find configs/volks.yaml.

    Returns:
        Path to the YAML file, or None.
    """
    current = Path(__file__).resolve()
    for parent in current.parents:
        candidate = parent / "configs" / "volks.yaml"
        if candidate.is_file():
            return candidate
        if (parent / ".git").exists():
            return candidate
    return None


@traces(
    spec="014",
    story="US-002",
    reqs=["REQ-014-016", "REQ-014-018", "REQ-014-026", "REQ-014-028"],
)
class VolkHandler:
    """Volk registry query and lifecycle subscription."""

    def __init__(
        self,
        pubsub: PubSubManager,
        yaml_path: str | Path | None = None,
        processor: Any = None,
    ) -> None:
        """Wire handler to volk subsystems.

        Args:
            pubsub: Pub/sub subscription manager.
            yaml_path: Override path for volks.yaml (testing).
            processor: Message processor for broadcast/push.
        """
        self._pubsub = pubsub
        self._registry = load_volk_registry(yaml_path)
        self._admin_token: str | None = None
        self._kernel_client: Any = None
        self._processor = processor

    def set_processor(self, processor: Any) -> None:
        """Inject the message processor after construction.

        Args:
            processor: MessageProcessor instance.
        """
        self._processor = processor

    @property
    def registry(self) -> list[dict[str, Any]]:
        """Expose the loaded volk registry."""
        return self._registry

    def set_kernel_client(self, client: Any) -> None:
        """Bind the kernel gRPC client so ``handle_list`` can serve live volk
        status (via its GetStatus RPC) instead of the static roster."""
        self._kernel_client = client

    @traces(spec="014", story="US-002", reqs=["REQ-014-016", "REQ-014-017"])
    async def _live_status(self) -> dict[str, dict[str, Any]]:
        """Live per-volk status from the kernel, keyed by alias.

        Best-effort: returns ``{}`` when no kernel client is bound or the query
        fails, so ``handle_list`` falls back to the static roster.
        """
        if self._kernel_client is None:
            return {}
        try:
            roster = await self._kernel_client.get_status(requester="gateway")
        except Exception:  # a status query must never break the roster listing
            return {}
        return {v.get("alias"): v for v in roster if v.get("alias")}

    @traces(spec="014", story="US-002", reqs=["REQ-014-026", "REQ-014-028"])
    def set_admin_token(self, token: str | None) -> None:
        """Inject the admin token that gates register/deregister. When unset,
        those operations are denied (no admin scope)."""
        self._admin_token = token

    @traces(spec="014", story="US-002", reqs=["REQ-014-026", "REQ-014-028"])
    def _require_admin(self, params: dict[str, Any]) -> None:
        """Raise unless a configured admin token matches ``params['admin_token']``."""
        if not self._admin_token or params.get("admin_token") != self._admin_token:
            raise RPCError("volk admin operation requires admin scope")

    @traces(
        spec="014",
        story="US-002",
        reqs=["REQ-014-016", "REQ-014-017"],
    )
    async def handle_list(
        self,
        connection_id: str,
        params: dict[str, Any],
    ) -> dict[str, Any]:
        """Handle volk.list -- list registered volks.

        The static registry supplies the roster (alias/name/role); when a kernel
        client is bound, each volk's *live* status (from the CXP GetStatus RPC)
        is merged in so callers see real work state (busy/current_task/status)
        rather than the static ``active``/``idle`` placeholder. If the live query
        is unavailable it falls back cleanly to the static roster.

        Args:
            connection_id: Caller connection ID.
            params: Optional filters and limit.

        Returns:
            List of volk info dicts.
        """
        live = await self._live_status()
        volks = []
        for volk in self._registry:
            alias = volk.get("alias", "")
            entry = {
                "alias": alias,
                "name": volk.get("name", alias),
                "role": volk.get("role", ""),
                "health": "healthy" if volk.get("status") == "active" else "unknown",
                "current_task": None,
                "status": volk.get("status", "idle"),
                "registered_at": datetime.now(UTC).isoformat(),
            }
            snapshot = live.get(alias)
            if snapshot:
                entry["status"] = snapshot.get("status") or entry["status"]
                entry["health"] = "busy" if snapshot.get("busy") else "healthy"
                entry["current_task"] = snapshot.get("summary") or None
                if snapshot.get("current_thread"):
                    entry["current_thread"] = snapshot["current_thread"]
                if snapshot.get("role"):
                    entry["role"] = entry["role"] or snapshot["role"]
            volks.append(entry)
        return {"volks": volks}

    @traces(
        spec="014",
        story="US-002",
        reqs=["REQ-014-018"],
    )
    async def handle_get(
        self,
        connection_id: str,
        params: dict[str, Any],
    ) -> dict[str, Any]:
        """Handle volk.get -- get volk by alias.

        Args:
            connection_id: Caller connection ID.
            params: volk_id or alias (camelCase volkId accepted).

        Returns:
            Volk detail dict.
        """
        alias = params.get("volk_id") or params.get("volkId") or params.get("alias", "")
        if not alias:
            raise RPCError(
                "Missing 'volk_id' or 'alias' parameter",
                rpc_code=-32602,
            )
        for volk in self._registry:
            if volk["alias"] == alias:
                return {"volk": dict(volk)}
        raise RPCError(f"Volk not found: {alias}")

    @traces(
        spec="014",
        story="US-002",
        reqs=["REQ-014-026"],
    )
    async def handle_register(
        self,
        connection_id: str,
        params: dict[str, Any],
    ) -> dict[str, Any]:
        """Handle volk.register -- add a volk to the gateway roster (admin only).

        Manages the gateway's volk roster (what ``volk.list`` reports). The
        actual volk *processes* are spawned by the agent-runtime kernel; this
        registers the volk's presence with the gateway. Admin-gated via an
        injected admin token.

        Args:
            connection_id: Caller connection ID.
            params: ``alias`` (required), optional ``name``/``role``/``status``,
                and ``admin_token``.

        Returns:
            ``{"registered": True, "volk": <entry>}``.
        """
        self._require_admin(params)
        alias = params.get("alias", "")
        if not alias:
            raise RPCError("Missing 'alias' parameter", rpc_code=-32602)
        if any(v.get("alias") == alias for v in self._registry):
            raise RPCError(f"Volk already registered: {alias}")
        entry = {
            "alias": alias,
            "name": params.get("name", alias),
            "role": params.get("role", ""),
            "status": params.get("status", "idle"),
        }
        self._registry.append(entry)
        await self.publish_lifecycle_event(
            "registered",
            alias,
            name=entry["name"],
            role=entry["role"],
            status=entry["status"],
        )
        return {"registered": True, "volk": dict(entry)}

    @traces(
        spec="014",
        story="US-002",
        reqs=["REQ-014-028"],
    )
    async def handle_deregister(
        self,
        connection_id: str,
        params: dict[str, Any],
    ) -> dict[str, Any]:
        """Handle volk.deregister -- remove a volk from the gateway roster (admin).

        Admin-gated via an injected admin token. Returns whether a roster entry
        was removed.

        Args:
            connection_id: Caller connection ID.
            params: ``alias`` (required) and ``admin_token``.

        Returns:
            ``{"deregistered": bool, "alias": <alias>}``.
        """
        self._require_admin(params)
        alias = params.get("alias", "")
        if not alias:
            raise RPCError("Missing 'alias' parameter", rpc_code=-32602)
        before = len(self._registry)
        self._registry[:] = [v for v in self._registry if v.get("alias") != alias]
        deregistered = len(self._registry) < before
        if deregistered:
            await self.publish_lifecycle_event("deregistered", alias)
        return {"deregistered": deregistered, "alias": alias}

    @traces(
        spec="014",
        story="US-002",
        reqs=["REQ-014-021"],
    )
    async def handle_lifecycle(
        self,
        connection_id: str,
        params: dict[str, Any],
    ) -> dict[str, Any]:
        """Handle volk.lifecycle -- subscribe to lifecycle events.

        Args:
            connection_id: Caller connection ID.
            params: Optional filters.

        Returns:
            subscription_id and topic.
        """
        filters = params.get("filters", {})
        sub_id = await self._pubsub.subscribe(connection_id, "volk.lifecycle", filters)
        return {
            "subscription_id": sub_id,
            "topic": "volk.lifecycle",
        }

    @traces(spec="014", story="US-002", reqs=["REQ-014-023"])
    async def publish_lifecycle_event(
        self,
        event_type: str,
        alias: str,
        **kwargs: Any,
    ) -> None:
        """Publish a lifecycle event to all subscribers.

        Delivers through the message processor's real broadcast (the same
        path ``cxp.traffic`` uses), so subscribed connections actually
        receive the event over their websocket. A no-op when no processor
        has been injected yet (e.g. constructed before gateway boot wires
        it via ``set_processor``).

        Args:
            event_type: Type of lifecycle event.
            alias: Volk alias.
            **kwargs: Additional event fields.
        """
        event = {
            "type": event_type,
            "alias": alias,
            "timestamp": datetime.now(UTC).isoformat(),
            **kwargs,
        }
        if self._processor:
            await self._processor.broadcast("volk.lifecycle", event)