orchestrata_ultimate.postman.router

Reference › orchestrata_ultimate › postman › router

The routing engine of the broker, SPEC-002 story FR-003 (REQ-002-FR3-001 to -011), also traced to SPEC-001 message routing. RoutingEngine implements IRoutingEngine from interfaces and turns the recipient field of a MessageEnvelope into a list of destination agent ids. delivery is its importer. It is the broker-level counterpart of the kernel's CXPRouter, which routes from the volk registry instead.

get_routing_type classifies the recipient: empty raises RoutingError; *, broadcast or all is broadcast; a comma means multicast; a pattern like U:Cd (letter U, colon, a capital and a lower-case letter) is direct; a name containing ., * or ? is a topic; anything else is direct. resolve first looks in a route cache (the cache key combines sender and recipient; the cache time-to-live is 60 seconds by default), then calls _check_governance, then dispatches to _resolve_direct, _resolve_topic, _resolve_broadcast or _resolve_multicast, caches the answer and returns it. The built-in governance rule is simple: the orchestrator U:Ox may send anywhere, only the orchestrator may broadcast, and topic, multicast and direct routes are allowed; a denial raises RoutingError. Topic resolution unions the static routes with the subscribers of every matching pattern (fnmatch) and raises NoRouteFoundError when there are none. Multicast resolution requires every listed agent to be in the registered-agent set and raises NoRouteFoundError otherwise.

add_route and remove_route manage static routes, subscribe_to_topic and unsubscribe_from_topic manage topic patterns, and invalidate_cache clears cached routes, either all or those matching a pattern.

At a glance

Class diagram

Dependency diagram

API

Classes

RoutingEngine

Bases: IRoutingEngine · Decorators: traces(spec='002', story='FR-003', reqs=['REQ-002-FR3-001', 'REQ-002-FR3-002', 'REQ-002-FR3-004', 'REQ-002-FR3-005']) · Line 37

Routes messages to recipients based on routing rules.

Source

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

"""Routing engine for Postman message broker.

Implements the IRoutingEngine interface with support for:
- Direct routing to specific agents
- Topic-based routing to subscribers
- Broadcast routing to all agents
- Multicast routing to multiple agents
- PAM++ governance integration for authority checks
- Route caching with TTL
"""

import asyncio
import fnmatch
import logging
import re
import time

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.postman.constants import ROUTER_CACHE_TTL_SECONDS
from orchestrata_ultimate.postman.exceptions import NoRouteFoundError, RoutingError
from orchestrata_ultimate.postman.interfaces import IRoutingEngine
from orchestrata_ultimate.postman.types import MessageEnvelope

logger = logging.getLogger(__name__)


@traces(
    spec="002",
    story="FR-003",
    reqs=[
        "REQ-002-FR3-001",
        "REQ-002-FR3-002",
        "REQ-002-FR3-004",
        "REQ-002-FR3-005",
    ],
)
class RoutingEngine(IRoutingEngine):
    """Routes messages to recipients based on routing rules.

    Supports direct, topic-based, broadcast, and multicast routing
    with PAM++ governance integration for authority checks.
    """

    # Known agent IDs for broadcast routing
    REGISTERED_AGENTS = {
        "U:Ox",  # Orchestrator
        "U:PS",  # Postman
        "U:Cd",  # CodeDirector
        "U:Ux",  # UX
        "U:PM",  # PM
        "U:Ar",  # Architect
        "U:Qa",  # QA
        "U:Cx",  # Concierge
    }

    # Routing type constants
    ROUTING_DIRECT = "direct"
    ROUTING_TOPIC = "topic"
    ROUTING_BROADCAST = "broadcast"
    ROUTING_MULTICAST = "multicast"

    @traces(spec="002", story="FR-003", reqs=["REQ-002-FR3-014"])
    def __init__(self, cache_ttl: int = ROUTER_CACHE_TTL_SECONDS) -> None:
        """Initialize routing engine.

        Args:
            cache_ttl: Time-to-live for cached routes in seconds
        """
        self._cache_ttl = cache_ttl
        self._route_cache: dict[str, tuple[list[str], float]] = {}
        self._cache_lock = asyncio.Lock()

        # Static routes: pattern -> list of destinations
        self._static_routes: dict[str, list[str]] = {}
        self._routes_lock = asyncio.Lock()

        # Topic subscriptions: topic_pattern -> set of subscriber agent IDs
        self._topic_subscriptions: dict[str, set[str]] = {}
        self._subscriptions_lock = asyncio.Lock()

    @traces(
        spec="002",
        story="FR-003",
        reqs=[
            "REQ-002-FR3-001",
            "REQ-002-FR3-002",
            "REQ-002-FR3-004",
            "REQ-002-FR3-005",
        ],
    )
    def get_routing_type(self, recipient: str) -> str:
        """Determine routing type from recipient pattern.

        Args:
            recipient: Recipient pattern from envelope

        Returns:
            Routing type constant (ROUTING_DIRECT, ROUTING_TOPIC, etc.)
        """
        if not recipient:
            raise RoutingError("Recipient cannot be empty")

        # Broadcast patterns
        if recipient in ("*", "broadcast", "all"):
            return self.ROUTING_BROADCAST

        # Multicast (comma-separated agent IDs)
        if "," in recipient:
            return self.ROUTING_MULTICAST

        # Direct agent ID (U:XX format)
        if re.match(r"^U:[A-Z][a-z]$", recipient):
            return self.ROUTING_DIRECT

        # Topic pattern (contains dots or wildcards)
        if "." in recipient or "*" in recipient or "?" in recipient:
            return self.ROUTING_TOPIC

        # Default to direct
        return self.ROUTING_DIRECT

    @traces(spec="001", story="PAM-ENFORCEMENT", reqs=["REQ-001-US-003-011"])
    async def _check_governance(self, sender: str, recipient: str) -> tuple[bool, str]:
        """Check if sender has authority to send to recipient.

        This method implements basic governance rules and can be extended
        to integrate with PolicyEvaluator when available.

        Basic rules:
        - U:Ox (Orchestrator) can send to anyone
        - Agents can send to their known recipients
        - Topics are generally allowed
        - Broadcast requires orchestrator authority

        Args:
            sender: Sender agent ID
            recipient: Recipient pattern or agent ID

        Returns:
            Tuple of (authorized, reason)
        """
        # Orchestrator has universal authority
        if sender == "U:Ox":
            return True, "Orchestrator has universal authority"

        # Determine routing type
        routing_type = self.get_routing_type(recipient)

        # Broadcast requires orchestrator authority
        if routing_type == self.ROUTING_BROADCAST:
            return False, f"Only orchestrator can broadcast, not {sender}"

        # Topics are generally allowed
        if routing_type == self.ROUTING_TOPIC:
            return True, "Topic routing allowed"

        # Multicast allowed for any agent
        if routing_type == self.ROUTING_MULTICAST:
            return True, "Multicast routing allowed"

        # Direct routing - allow it, let _resolve_direct handle registration check
        if routing_type == self.ROUTING_DIRECT:
            return True, "Direct routing allowed"

        # Default deny
        return False, f"No governance rule matches for {sender} -> {recipient}"

    @traces(spec="001", story="FR-003", reqs=["REQ-001-US-002-003"])
    @traces(spec="002", story="FR-003", reqs=["REQ-002-FR3-006"])
    async def _resolve_direct(self, recipient: str) -> list[str]:
        """Resolve direct routing to single agent.

        Args:
            recipient: Agent ID

        Returns:
            List containing single agent ID

        Raises:
            NoRouteFoundError: If agent is not registered
        """
        if recipient not in self.REGISTERED_AGENTS:
            raise NoRouteFoundError(f"Agent {recipient} is not registered")
        return [recipient]

    @traces(spec="002", story="FR-003", reqs=["REQ-002-FR3-007"])
    async def _resolve_topic(self, topic: str) -> list[str]:
        """Resolve topic routing to all subscribers.

        Args:
            topic: Topic pattern

        Returns:
            List of subscriber agent IDs

        Raises:
            NoRouteFoundError: If no subscribers found
        """
        subscribers: set[str] = set()

        async with self._subscriptions_lock:
            # Check static routes first
            if topic in self._static_routes:
                subscribers.update(self._static_routes[topic])

            # Match against subscription patterns
            for pattern, subs in self._topic_subscriptions.items():
                if fnmatch.fnmatch(topic, pattern):
                    subscribers.update(subs)

        if not subscribers:
            raise NoRouteFoundError(f"No subscribers for topic {topic}")

        return list(subscribers)

    @traces(spec="001", story="MESSAGE-ROUTING", reqs=["REQ-001-US-002-005"])
    async def _resolve_broadcast(self) -> list[str]:
        """Resolve broadcast routing to all registered agents.

        Returns:
            List of all registered agent IDs
        """
        return list(self.REGISTERED_AGENTS)

    @traces(spec="001", story="MESSAGE-ROUTING", reqs=["REQ-001-US-002-004"])
    async def _resolve_multicast(self, recipient: str) -> list[str]:
        """Resolve multicast routing to multiple agents.

        Args:
            recipient: Comma-separated list of agent IDs

        Returns:
            List of agent IDs

        Raises:
            NoRouteFoundError: If any agent is not registered
        """
        agent_ids = [aid.strip() for aid in recipient.split(",")]

        # Validate all agents are registered
        for agent_id in agent_ids:
            if agent_id not in self.REGISTERED_AGENTS:
                raise NoRouteFoundError(
                    f"Agent {agent_id} in multicast list is not registered"
                )

        return agent_ids

    @traces(spec="002", story="FR-003", reqs=["REQ-002-FR3-011"])
    async def _get_cached_route(self, cache_key: str) -> list[str] | None:
        """Get route from cache if not expired.

        Args:
            cache_key: Cache key for route

        Returns:
            Cached route list or None if expired/missing
        """
        async with self._cache_lock:
            if cache_key in self._route_cache:
                destinations, timestamp = self._route_cache[cache_key]
                if time.time() - timestamp < self._cache_ttl:
                    return destinations
                # Expired - remove from cache
                del self._route_cache[cache_key]
        return None

    @traces(spec="002", story="FR-003", reqs=["REQ-002-FR3-011"])
    async def _set_cached_route(self, cache_key: str, destinations: list[str]) -> None:
        """Store route in cache with timestamp.

        Args:
            cache_key: Cache key for route
            destinations: List of destination agent IDs
        """
        async with self._cache_lock:
            self._route_cache[cache_key] = (destinations, time.time())

    @traces(spec="001", story="MESSAGE-ROUTING", reqs=["REQ-001-US-002-001"])
    @traces(
        spec="002", story="MESSAGE-ROUTING", reqs=["REQ-002-FR3-010", "REQ-002-FR3-011"]
    )
    async def resolve(self, envelope: MessageEnvelope) -> list[str]:
        """Resolve envelope to destination agent IDs.

        Performs governance checks and routes based on recipient pattern.

        Args:
            envelope: Message envelope to route

        Returns:
            List of destination agent IDs

        Raises:
            RoutingError: If governance check fails
            NoRouteFoundError: If no route can be found
        """
        recipient = envelope.to
        sender = envelope.from_

        # Check cache first
        cache_key = f"{sender}|{recipient}"
        cached = await self._get_cached_route(cache_key)
        if cached is not None:
            return cached

        # Governance check
        authorized, reason = await self._check_governance(sender, recipient)
        if not authorized:
            raise RoutingError(
                f"Governance denied: {reason} (sender={sender}, recipient={recipient})"
            )

        # Resolve based on routing type
        routing_type = self.get_routing_type(recipient)

        if routing_type == self.ROUTING_DIRECT:
            destinations = await self._resolve_direct(recipient)
        elif routing_type == self.ROUTING_TOPIC:
            destinations = await self._resolve_topic(recipient)
        elif routing_type == self.ROUTING_BROADCAST:
            destinations = await self._resolve_broadcast()
        elif routing_type == self.ROUTING_MULTICAST:
            destinations = await self._resolve_multicast(recipient)
        else:
            raise RoutingError(f"Unknown routing type: {routing_type}")

        # Cache the resolved route
        await self._set_cached_route(cache_key, destinations)

        return destinations

    @traces(spec="002", story="US-005", reqs=["REQ-002-US-005-007"])
    async def add_route(self, pattern: str, destinations: list[str]) -> None:
        """Add static route.

        Args:
            pattern: Route pattern (topic or agent ID)
            destinations: List of destination agent IDs
        """
        async with self._routes_lock:
            self._static_routes[pattern] = destinations.copy()

        # Invalidate cache for this pattern
        self.invalidate_cache(pattern)

    @traces(spec="002", story="US-005", reqs=["REQ-002-US-005-014"])
    async def remove_route(self, pattern: str) -> bool:
        """Remove route.

        Args:
            pattern: Route pattern to remove

        Returns:
            True if route was removed, False if not found
        """
        async with self._routes_lock:
            if pattern in self._static_routes:
                del self._static_routes[pattern]
                self.invalidate_cache(pattern)
                return True
        return False

    @traces(spec="002", story="FR-003", reqs=["REQ-002-FR3-012"])
    def invalidate_cache(self, pattern: str | None = None) -> None:
        """Invalidate route cache.

        Args:
            pattern: Specific pattern to invalidate, or None for all
        """
        if pattern is None:
            # Clear entire cache
            self._route_cache.clear()
        else:
            # Remove cache entries matching pattern
            # Cache key format is "sender|recipient"
            # For topic patterns, we need to match the recipient part
            keys_to_remove = []
            for key in self._route_cache.keys():
                # Extract recipient from cache key (format: "sender|recipient")
                parts = key.split("|", 1)
                if len(parts) == 2:
                    recipient = parts[1]
                    # Check if pattern matches recipient using fnmatch
                    if fnmatch.fnmatch(recipient, pattern) or pattern in recipient:
                        keys_to_remove.append(key)
                elif pattern in key:
                    keys_to_remove.append(key)

            for key in keys_to_remove:
                del self._route_cache[key]

    @traces(spec="002", story="US-002", reqs=["REQ-002-US-002-001"])
    async def subscribe_to_topic(self, agent_id: str, topic_pattern: str) -> None:
        """Subscribe an agent to a topic pattern.

        Args:
            agent_id: Agent ID to subscribe
            topic_pattern: Topic pattern to subscribe to
        """
        async with self._subscriptions_lock:
            if topic_pattern not in self._topic_subscriptions:
                self._topic_subscriptions[topic_pattern] = set()
            self._topic_subscriptions[topic_pattern].add(agent_id)

        # Invalidate cache for this topic
        self.invalidate_cache(topic_pattern)

    @traces(
        spec="002",
        story="US-002",
        reqs=["REQ-002-US-002-002", "REQ-002-FR5-011"],
    )
    async def unsubscribe_from_topic(self, agent_id: str, topic_pattern: str) -> bool:
        """Unsubscribe an agent from a topic pattern.

        Args:
            agent_id: Agent ID to unsubscribe
            topic_pattern: Topic pattern to unsubscribe from

        Returns:
            True if unsubscribed, False if not found
        """
        async with self._subscriptions_lock:
            if topic_pattern in self._topic_subscriptions:
                if agent_id in self._topic_subscriptions[topic_pattern]:
                    self._topic_subscriptions[topic_pattern].remove(agent_id)
                    # Clean up empty subscription sets
                    if not self._topic_subscriptions[topic_pattern]:
                        del self._topic_subscriptions[topic_pattern]
                    self.invalidate_cache(topic_pattern)
                    return True
        return False