orchestrata_ultimate.cxp.router

Reference › orchestrata_ultimate › cxp › router

CXPRouter decides where a CXP packet may go and whether it is allowed to go there. dispatcher is its caller, and the optional prolog_bridge supplies governance. It returns the RoutingResult and RoutingDecision types from types. See cxp.

The router holds a routing table mapping each sender alias to a set of allowed recipients. _seed_default_routes fills it from the known volk aliases: every volk may send to Postman (U:PS) and the Orchestrator (U:Ox), and those two may send to every other volk. register_route and unregister_route change the table; is_valid_route, get_routes_for and get_all_routes read it.

route first raises RoutingError if the packet lacks a sender or recipient, then NoRouteError if the sender has no route to the recipient. If a Prolog bridge was supplied, _check_governance asks it to validate the route, and a refusal produces a REJECT result with the reason. An error while consulting the bridge is logged and also counts as a refusal. Otherwise the result is DELIVER with the recipient as the single target.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

CXPRouter

Line 52

CXP-level packet router.

Source

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

# mypy: ignore-errors
"""
CXP Packet Router.

Determines WHERE a CXP packet goes based on the routing table
and governance rules. Each Volk has a set of allowed destinations;
the router resolves the correct target(s) for a given packet.

Spec: 004-cxp
Author: Mohamed
"""

import logging
from typing import TYPE_CHECKING

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.packet import CXPPacket
from orchestrata_ultimate.cxp.exceptions import NoRouteError, RoutingError
from orchestrata_ultimate.cxp.types import (
    VOLK_ALIASES,
    RoutingDecision,
    RoutingResult,
)

if TYPE_CHECKING:
    from orchestrata_ultimate.cxp.prolog_bridge import PrologBridge

logger = logging.getLogger(__name__)

__all__ = [
    "CXPRouter",
]


# Agents that act as global hubs: they can send to anyone.
_HUB_AGENTS: frozenset = frozenset(
    {
        VOLK_ALIASES["Orchestrator"],  # U:Ox
        VOLK_ALIASES["Postman"],  # U:PS
    }
)

# Default destinations available to every non-hub agent.
_DEFAULT_DESTINATIONS: frozenset = frozenset(
    {
        VOLK_ALIASES["Postman"],  # U:PS
        VOLK_ALIASES["Orchestrator"],  # U:Ox
    }
)


class CXPRouter:
    """CXP-level packet router.

    Maintains a routing table that maps each ``from_agent`` to a set of
    allowed ``to_agent`` destinations.  The table is pre-populated with
    default routes:

    * Every Volk can send to Postman (U:PS) and Orchestrator (U:Ox).
    * Orchestrator and Postman can send to any registered Volk.

    An optional *prolog_bridge* can be supplied for governance-level
    validation (e.g. PAM++ ownership checks).

    Args:
        prolog_bridge: Optional bridge to a Prolog governance engine.
    """

    def __init__(self, prolog_bridge: PrologBridge | None = None) -> None:
        self._routes: dict[str, set[str]] = {}
        self._prolog_bridge = prolog_bridge

        # Seed the routing table with defaults.
        self._seed_default_routes()

    # ------------------------------------------------------------------
    # Public API
    # ------------------------------------------------------------------

    @traces(spec="001", story="US-002", reqs=["REQ-001-US-002-001"])
    async def route(self, packet: CXPPacket) -> RoutingResult:
        """Determine routing for a CXP packet.

        Resolution order:
        1. Validate that *from_agent* has a route to *to_agent*.
        2. If a prolog_bridge is present, perform governance validation.
        3. Return a :class:`RoutingResult` with decision DELIVER (or REJECT
           when no valid route exists).

        Args:
            packet: The CXP packet to route.

        Returns:
            A RoutingResult describing where the packet goes.

        Raises:
            NoRouteError: If no route exists from the sender to the
                intended recipient.
            RoutingError: If governance validation fails.
        """
        from_agent: str = packet.from_agent
        to_agent: str = packet.to_agent

        if not from_agent or not to_agent:
            raise RoutingError(
                message="Packet missing from_agent or to_agent",
                packet_id=packet.message_id,
            )

        # Check the routing table.
        if not self.is_valid_route(from_agent, to_agent):
            raise NoRouteError(
                destination=to_agent,
                message=f"No route from {from_agent} to {to_agent}",
                details={"from_agent": from_agent, "to_agent": to_agent},
            )

        # Governance check via Prolog bridge (if available).
        if self._prolog_bridge is not None:
            governance_ok = await self._check_governance(packet)
            if not governance_ok:
                return RoutingResult(
                    decision=RoutingDecision.REJECT,
                    targets=[to_agent],
                    reason=f"Governance rejected route {from_agent} -> {to_agent}",
                )

        return RoutingResult(
            decision=RoutingDecision.DELIVER,
            targets=[to_agent],
            reason=f"Routed {from_agent} -> {to_agent}",
        )

    def register_route(self, from_agent: str, to_agent: str) -> None:
        """Register an allowed route between two agents.

        Args:
            from_agent: Sender Volk alias (e.g. ``"U:Cd"``).
            to_agent: Receiver Volk alias (e.g. ``"U:Qa"``).
        """
        if from_agent not in self._routes:
            self._routes[from_agent] = set()
        self._routes[from_agent].add(to_agent)

    def unregister_route(self, from_agent: str, to_agent: str) -> None:
        """Remove a previously registered route.

        Args:
            from_agent: Sender Volk alias.
            to_agent: Receiver Volk alias.
        """
        if from_agent in self._routes:
            self._routes[from_agent].discard(to_agent)

    @traces(spec="001", story="US-002", reqs=["REQ-001-US-002-009"])
    def is_valid_route(self, from_agent: str, to_agent: str) -> bool:
        """Check whether a route from *from_agent* to *to_agent* exists.

        Args:
            from_agent: Sender Volk alias.
            to_agent: Receiver Volk alias.

        Returns:
            True if the route is registered, False otherwise.
        """
        destinations = self._routes.get(from_agent, set())
        return to_agent in destinations

    def get_routes_for(self, agent: str) -> list[str]:
        """Return all agents reachable from *agent*.

        Args:
            agent: The source Volk alias.

        Returns:
            Sorted list of destination aliases.
        """
        return sorted(self._routes.get(agent, set()))

    def get_all_routes(self) -> dict[str, list[str]]:
        """Return the full routing table.

        Returns:
            Dictionary mapping each source agent to its sorted list of
            allowed destinations.
        """
        return {
            agent: sorted(destinations)
            for agent, destinations in sorted(self._routes.items())
        }

    # ------------------------------------------------------------------
    # Internal helpers
    # ------------------------------------------------------------------

    def _seed_default_routes(self) -> None:
        """Populate the routing table with default routes.

        Default policy:
        * All Volks can reach Postman (U:PS) and Orchestrator (U:Ox).
        * Orchestrator and Postman can reach every Volk.
        """
        all_aliases: list[str] = list(VOLK_ALIASES.values())

        for alias in all_aliases:
            if alias not in self._routes:
                self._routes[alias] = set()

            if alias in _HUB_AGENTS:
                # Hub agents can reach everyone.
                self._routes[alias] = set(all_aliases) - {alias}
            else:
                # Non-hub agents get the default destinations.
                self._routes[alias].update(_DEFAULT_DESTINATIONS - {alias})

    @traces(spec="001", story="US-003", reqs=["REQ-001-US-003-001"])
    async def _check_governance(self, packet: CXPPacket) -> bool:
        """Validate routing against governance rules via the Prolog bridge.

        Args:
            packet: The packet whose routing is being validated.

        Returns:
            True if governance permits the route, False otherwise.
        """
        try:
            if self._prolog_bridge is None:
                # Only called (route()) when a bridge is configured; guard
                # narrows the type for the call below without changing
                # behavior for the one real caller.
                return True
            result = self._prolog_bridge.validate_route_via_prolog(
                from_agent=packet.from_agent,
                to_agent=packet.to_agent,
            )
            return bool(result)
        except Exception as exc:
            # Governance bridge failure is treated as rejection.
            logger.warning(
                "Prolog route validation failed for %s->%s: %s — rejecting route",
                packet.from_agent,
                packet.to_agent,
                exc,
            )
            return False