orchestrata_ultimate.kernel.router

Reference › orchestrata_ultimate › kernel › router

The kernel's CXP message router, the first stop for a packet inside the kernel. CXPRouter decides who should receive a CXPPacket and reports the outcome; it does not deliver. Delivery is the job of dispatcher, and undeliverable packets go to queue_manager. The router depends on the volk registry (agents.volk_registry) to know who exists, and serves SPEC-001 story US-002 (REQ-001-US-002-001 to -040). The curated kernel page shows where it sits in the message plane. The separate postman router routes inside the broker.

route runs four checks in order. _validate_packet_schema calls the packet's own validation and returns a TOTAL_FAILURE result when it raises. validate_sender requires the sender to be registered and not terminated, else the result is INVALID_SENDER. _resolve_and_validate_recipients expands the TO field through resolve_recipients: a name starting with U: must be a registered alias, anything else is matched case-insensitively against volk roles, and unknown names or empty roles raise an error that becomes an INVALID_RECIPIENT result. _determine_routing_mode picks BROADCAST for * or broadcast, MULTICAST for several recipients or a comma, otherwise UNICAST.

Last, recipients that are not online are collected. _determine_status_and_track yields OFFLINE_RECIPIENT when all are offline, PARTIAL_FAILURE when some are, otherwise SUCCESS; it records a PENDING to QUEUED (or stays PENDING) transition in a MessageStatus, and adds a warning that offline recipients will be queued. The router updates RoutingMetrics (count, running average latency, success rate) and appends a trace entry, trimmed to max_trace_entries (10000 by default). get_message_status, get_message_trace and get_metrics expose the records. The router does not itself enqueue packets.

At a glance

Class diagram

Dependency diagram

API

Classes

RoutingMode (enum)

Bases: Enum · Decorators: traces(spec='001', story='US-002', reqs=['REQ-001-US-002-003', 'REQ-001-US-002-004', 'REQ-001-US-002-005']) · Line 33

Routing delivery mode.

RoutingStatus (enum)

Bases: Enum · Decorators: traces(spec='001', story='US-002', reqs=['REQ-001-US-002-008', 'REQ-001-US-002-009']) · Line 42

Status of routing operation.

MessageStatus (dataclass)

Decorators: dataclass, traces(spec='001', story='US-002', reqs=['REQ-001-US-002-019', 'REQ-001-US-002-031', 'REQ-001-US-002-032', 'REQ-001-US-… · Line 67

Tracks message delivery status and transitions.

RoutingMetrics (dataclass)

Decorators: dataclass, traces(spec='001', story='US-002', reqs=['REQ-001-US-002-040']) · Line 101

Routing performance metrics.

RoutingResult (dataclass)

Decorators: dataclass, traces(spec='001', story='US-002', reqs=['REQ-001-US-002-009']) · Line 141

Result of routing operation.

CXPRouter

Decorators: traces(spec='001', story='US-002', reqs=['REQ-001-US-002-001', 'REQ-001-US-002-003', 'REQ-001-US-002-004', 'REQ-001-US-… · Line 179

Routes CXP packets to appropriate recipients.

Source

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

# mypy: ignore-errors
"""CXP Message Router for Orchestration Kernel.

This module implements CXP packet routing with support for unicast,
multicast, and broadcast delivery modes as specified in Stream 2A.

Requirements Implemented:
- REQ-001-US-002-001 to REQ-001-US-002-010: Message routing with validation
- REQ-001-US-002-031 to REQ-001-US-002-040: Status tracking and metrics
"""

import logging
from dataclasses import dataclass, field
from datetime import UTC, datetime
from enum import Enum
from typing import Any

from orchestrata_ultimate.agents.volk_registry import VolkRegistry, VolkStatus
from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.packet import CXPPacket, DeliveryStatus

logger = logging.getLogger(__name__)

# Trace log eviction policy
DEFAULT_MAX_TRACE_ENTRIES = 10000


@traces(
    spec="001",
    story="US-002",
    reqs=["REQ-001-US-002-003", "REQ-001-US-002-004", "REQ-001-US-002-005"],
)
class RoutingMode(Enum):
    """Routing delivery mode."""

    UNICAST = "unicast"  # Single recipient
    MULTICAST = "multicast"  # Multiple specific recipients
    BROADCAST = "broadcast"  # All registered Volks


@traces(spec="001", story="US-002", reqs=["REQ-001-US-002-008", "REQ-001-US-002-009"])
class RoutingStatus(Enum):
    """Status of routing operation."""

    SUCCESS = "success"  # Successfully routed
    PARTIAL_FAILURE = "partial"  # Some recipients failed
    TOTAL_FAILURE = "failed"  # All recipients failed
    INVALID_SENDER = "invalid_sender"
    INVALID_RECIPIENT = "invalid_recipient"
    OFFLINE_RECIPIENT = "offline"


@dataclass
@traces(
    spec="001",
    story="US-002",
    reqs=[
        "REQ-001-US-002-019",
        "REQ-001-US-002-031",
        "REQ-001-US-002-032",
        "REQ-001-US-002-033",
        "REQ-001-US-002-034",
        "REQ-001-US-002-035",
        "REQ-001-US-002-037",
    ],
)
class MessageStatus:
    """Tracks message delivery status and transitions."""

    message_id: str
    current_status: DeliveryStatus
    transitions: list[dict[str, Any]] = field(default_factory=list)
    created_at: datetime = field(default_factory=lambda: datetime.now(UTC))

    @traces(spec="001", story="US-002", reqs=["REQ-001-US-002-037"])
    def add_transition(
        self, from_status: DeliveryStatus, to_status: DeliveryStatus, reason: str = ""
    ) -> None:
        """Record status transition."""
        transition = {
            "from": from_status.value,
            "to": to_status.value,
            "timestamp": datetime.now(UTC).isoformat(),
            "reason": reason,
        }
        self.transitions.append(transition)
        self.current_status = to_status

    def to_dict(self) -> dict[str, Any]:
        """Convert to dictionary."""
        return {
            "message_id": self.message_id,
            "current_status": self.current_status.value,
            "transitions": self.transitions,
            "created_at": self.created_at.isoformat(),
        }


@dataclass
@traces(spec="001", story="US-002", reqs=["REQ-001-US-002-040"])
class RoutingMetrics:
    """Routing performance metrics."""

    total_routed: int = 0
    successful_deliveries: int = 0
    failed_deliveries: int = 0
    average_latency_ms: float = 0.0

    @traces(spec="001", story="US-002", reqs=["REQ-001-US-002-040"])
    def update_latency(self, new_latency_ms: float) -> None:
        """Update average latency with new measurement."""
        total = self.successful_deliveries + self.failed_deliveries
        if total > 0:
            self.average_latency_ms = (
                self.average_latency_ms * (total - 1) + new_latency_ms
            ) / total
        else:
            self.average_latency_ms = new_latency_ms

    @traces(spec="001", story="US-002", reqs=["REQ-001-US-002-040"])
    def success_rate(self) -> float:
        """Calculate delivery success rate."""
        total = self.successful_deliveries + self.failed_deliveries
        if total == 0:
            return 0.0
        return (self.successful_deliveries / total) * 100.0

    def to_dict(self) -> dict[str, Any]:
        """Convert to dictionary."""
        return {
            "total_routed": self.total_routed,
            "successful_deliveries": self.successful_deliveries,
            "failed_deliveries": self.failed_deliveries,
            "average_latency_ms": self.average_latency_ms,
            "success_rate": self.success_rate(),
        }


@dataclass
@traces(spec="001", story="US-002", reqs=["REQ-001-US-002-009"])
class RoutingResult:
    """Result of routing operation."""

    recipients: list[str]
    routing_mode: RoutingMode
    status: RoutingStatus
    message_id: str
    errors: list[str] = field(default_factory=list)
    warnings: list[str] = field(default_factory=list)
    timestamp: datetime = field(default_factory=lambda: datetime.now(UTC))

    def to_dict(self) -> dict[str, Any]:
        """Convert to dictionary."""
        return {
            "recipients": self.recipients,
            "routing_mode": self.routing_mode.value,
            "status": self.status.value,
            "message_id": self.message_id,
            "errors": self.errors,
            "warnings": self.warnings,
            "timestamp": self.timestamp.isoformat(),
        }


@traces(
    spec="001",
    story="US-002",
    reqs=[
        "REQ-001-US-002-001",
        "REQ-001-US-002-003",
        "REQ-001-US-002-004",
        "REQ-001-US-002-005",
        "REQ-001-US-002-010",
        "REQ-001-US-002-036",
        "REQ-001-US-002-039",
        "REQ-001-US-002-040",
    ],
)
class CXPRouter:
    """Routes CXP packets to appropriate recipients.

    Handles unicast, multicast, and broadcast routing with
    validation and status tracking.
    """

    def __init__(
        self,
        volk_registry: VolkRegistry,
        max_trace_entries: int = DEFAULT_MAX_TRACE_ENTRIES,
    ) -> None:
        """Initialize router with Volk registry.

        Args:
            volk_registry: Registry for looking up Volk information
            max_trace_entries: Maximum number of trace entries to keep
        """
        self.registry = volk_registry
        self.message_statuses: dict[str, MessageStatus] = {}
        self.metrics = RoutingMetrics()
        self._trace_log: list[dict[str, Any]] = []
        self._max_trace_entries: int = max_trace_entries

    @traces(spec="001", story="US-002", reqs=["REQ-001-US-002-010"])
    def _validate_packet_schema(self, packet: CXPPacket) -> RoutingResult | None:
        """Validate packet schema, returning a failure result if invalid.

        REQ-001-US-002-010: Validate message schema before routing

        Args:
            packet: CXP packet to validate

        Returns:
            RoutingResult with TOTAL_FAILURE if invalid, None if valid.
        """
        try:
            packet._validate()
            return None
        except ValueError as e:
            return RoutingResult(
                recipients=[],
                routing_mode=RoutingMode.UNICAST,
                status=RoutingStatus.TOTAL_FAILURE,
                message_id=packet.header.packet_id,
                errors=[f"Schema validation failed: {str(e)}"],
            )

    @traces(spec="001", story="US-002", reqs=["REQ-001-US-002-009"])
    async def _resolve_and_validate_recipients(
        self, packet: CXPPacket
    ) -> tuple[list[str] | None, RoutingResult | None]:
        """Resolve recipients and validate all are registered.

        Args:
            packet: CXP packet with recipient field to resolve

        Returns:
            Tuple of (recipients list, failure RoutingResult).
            On success: (recipients, None). On failure: (None, error_result).
        """
        try:
            recipients = await self.resolve_recipients(packet.header.recipient)
        except Exception as e:
            return None, RoutingResult(
                recipients=[],
                routing_mode=RoutingMode.UNICAST,
                status=RoutingStatus.INVALID_RECIPIENT,
                message_id=packet.header.packet_id,
                errors=[f"Failed to resolve recipients: {str(e)}"],
            )

        invalid_recipients = [
            r for r in recipients if not self.registry.is_registered(r)
        ]
        if invalid_recipients:
            return None, RoutingResult(
                recipients=[],
                routing_mode=RoutingMode.UNICAST,
                status=RoutingStatus.INVALID_RECIPIENT,
                message_id=packet.header.packet_id,
                errors=[f"Invalid recipients: {', '.join(invalid_recipients)}"],
            )

        return recipients, None

    @traces(
        spec="001",
        story="US-002",
        reqs=["REQ-001-US-002-008", "REQ-001-US-002-031"],
    )
    def _determine_status_and_track(
        self,
        packet_id: str,
        recipients: list[str],
        offline_recipients: list[str],
    ) -> tuple[RoutingStatus, list[str]]:
        """Determine routing status, track message, and collect warnings.

        Args:
            packet_id: Packet ID for status tracking
            recipients: All resolved recipients
            offline_recipients: Recipients that are currently offline

        Returns:
            Tuple of (RoutingStatus, warnings list).
        """
        if not recipients:
            status = RoutingStatus.INVALID_RECIPIENT
        elif len(offline_recipients) == len(recipients):
            status = RoutingStatus.OFFLINE_RECIPIENT
        elif offline_recipients:
            status = RoutingStatus.PARTIAL_FAILURE
        else:
            status = RoutingStatus.SUCCESS

        # Track message status (REQ-001-US-002-031)
        initial_delivery_status = (
            DeliveryStatus.QUEUED if offline_recipients else DeliveryStatus.PENDING
        )
        self._track_message_status(
            packet_id,
            DeliveryStatus.PENDING,
            initial_delivery_status,
            f"Routing initiated - {len(recipients)} recipients ({len(offline_recipients)} offline)",
        )

        warnings = (
            [f"Offline recipients will be queued: {', '.join(offline_recipients)}"]
            if offline_recipients
            else []
        )
        return status, warnings

    @traces(
        spec="001",
        story="US-002",
        reqs=["REQ-001-US-002-001", "REQ-001-US-002-009", "REQ-001-US-002-010"],
    )
    async def route(self, packet: CXPPacket) -> RoutingResult:
        """Route a CXP packet to appropriate recipients.

        REQ-001-US-002-001: Route message based on TO field
        REQ-001-US-002-010: Validate message schema before routing

        Args:
            packet: CXP packet to route

        Returns:
            RoutingResult with routing outcome
        """
        start_time = datetime.now(UTC)

        # REQ-001-US-002-010: Validate schema
        schema_failure = self._validate_packet_schema(packet)
        if schema_failure is not None:
            return schema_failure

        # REQ-001-US-002-002: Validate sender
        if not await self.validate_sender(packet):
            return RoutingResult(
                recipients=[],
                routing_mode=RoutingMode.UNICAST,
                status=RoutingStatus.INVALID_SENDER,
                message_id=packet.header.packet_id,
                errors=[f"Invalid sender: {packet.header.sender}"],
            )

        # REQ-001-US-002-001: Resolve and validate recipients
        recipients, resolve_failure = await self._resolve_and_validate_recipients(
            packet
        )
        if resolve_failure is not None:
            return resolve_failure
        # _resolve_and_validate_recipients returns (recipients, None) on
        # success and (None, failure) on failure — this documents that
        # invariant for pyright now that the failure branch has returned.
        assert recipients is not None

        # Determine routing mode
        routing_mode = self._determine_routing_mode(
            packet.header.recipient, len(recipients)
        )

        # Check for offline recipients (REQ-001-US-002-008)
        offline_recipients = [r for r in recipients if not self._is_volk_online(r)]

        status, warnings = self._determine_status_and_track(
            packet.header.packet_id, recipients, offline_recipients
        )

        # Update metrics
        self.metrics.total_routed += 1
        latency_ms = (datetime.now(UTC) - start_time).total_seconds() * 1000
        self.metrics.update_latency(latency_ms)

        # REQ-001-US-002-039: Add trace entry
        self._add_trace_entry(
            packet.header.packet_id,
            "routed",
            {
                "recipients": recipients,
                "mode": routing_mode.value,
                "status": status.value,
                "latency_ms": latency_ms,
            },
        )

        return RoutingResult(
            recipients=recipients,
            routing_mode=routing_mode,
            status=status,
            message_id=packet.header.packet_id,
            warnings=warnings,
        )

    @traces(spec="001", story="US-002", reqs=["REQ-001-US-002-002"])
    async def validate_sender(self, packet: CXPPacket) -> bool:
        """Validate that FROM field matches a registered Volk.

        REQ-001-US-002-002: Validate FROM field matches sender identity

        Args:
            packet: CXP packet to validate

        Returns:
            True if sender is valid and registered
        """
        sender_alias = packet.header.sender

        # Check if sender is registered
        if not self.registry.is_registered(sender_alias):
            logger.warning(f"Unknown sender: {sender_alias}")
            return False

        # Get sender info
        sender_info = self.registry.get_volk(sender_alias)
        if not sender_info:
            return False

        # Check if sender is not terminated
        if sender_info.status == VolkStatus.TERMINATED:
            logger.warning(f"Sender is terminated: {sender_alias}")
            return False

        return True

    @traces(
        spec="001",
        story="US-002",
        reqs=["REQ-001-US-002-004", "REQ-001-US-002-007"],
    )
    async def resolve_recipients(self, to_field: str) -> list[str]:
        """Resolve recipients from TO field.

        REQ-001-US-002-003: Support unicast (single recipient)
        REQ-001-US-002-004: Support multicast (multiple recipients)
        REQ-001-US-002-005: Support broadcast (all Volks)
        REQ-001-US-002-006: Route to specific Volk by alias
        REQ-001-US-002-007: Route to Volk role
        REQ-001-US-002-009: Handle unknown recipients

        Args:
            to_field: Recipient specification (alias, role, or broadcast)

        Returns:
            List of Volk aliases to deliver to

        Raises:
            ValueError: If recipient specification is invalid
        """
        # REQ-001-US-002-005: Broadcast to all
        if to_field == "*" or to_field.lower() == "broadcast":
            return self._get_all_volks()

        # REQ-001-US-002-004: Multicast (comma-separated)
        if "," in to_field:
            aliases = [a.strip() for a in to_field.split(",")]
            resolved = []
            errors = []
            for alias in aliases:
                try:
                    recipients = await self._resolve_single_recipient(alias)
                    resolved.extend(recipients)
                except ValueError as e:
                    errors.append(str(e))

            if errors and not resolved:
                # All recipients failed to resolve
                raise ValueError(
                    f"Failed to resolve all recipients: {'; '.join(errors)}"
                )
            elif errors:
                # Some recipients resolved, log warnings but continue
                logger.warning(
                    f"Some recipients failed to resolve: {'; '.join(errors)}"
                )

            return list(set(resolved))  # Remove duplicates

        # REQ-001-US-002-003/006/007: Unicast (single alias or role)
        try:
            return await self._resolve_single_recipient(to_field)
        except ValueError as e:
            raise ValueError(f"Failed to resolve recipient '{to_field}': {str(e)}")

    @traces(
        spec="001",
        story="US-002",
        reqs=["REQ-001-US-002-003", "REQ-001-US-002-006", "REQ-001-US-002-007"],
    )
    async def _resolve_single_recipient(self, recipient: str) -> list[str]:
        """Resolve a single recipient specification.

        Args:
            recipient: Single recipient (alias or role)

        Returns:
            List of matching Volk aliases

        Raises:
            ValueError: If recipient alias is unknown or no Volks found with specified role
        """
        # REQ-001-US-002-006: Direct alias (U:Ox format)
        if recipient.startswith("U:"):
            if self.registry.is_registered(recipient):
                return [recipient]
            else:
                # REQ-001-US-002-009: Unknown recipient
                raise ValueError(f"Unknown recipient: {recipient}")

        # REQ-001-US-002-007: Role-based routing (e.g., "qa", "coder")
        # Find Volks by role
        all_volks = self.registry.list_volks()
        matching_volks = [
            v for v in all_volks if v.definition.role.lower() == recipient.lower()
        ]

        if not matching_volks:
            raise ValueError(f"No Volks found with role: {recipient}")

        return [v.definition.alias for v in matching_volks]

    @traces(
        spec="001",
        story="US-002",
        reqs=["REQ-001-US-002-003", "REQ-001-US-002-004", "REQ-001-US-002-005"],
    )
    def _determine_routing_mode(
        self, to_field: str, recipient_count: int
    ) -> RoutingMode:
        """Determine routing mode from TO field and recipient count."""
        if to_field == "*" or to_field.lower() == "broadcast":
            return RoutingMode.BROADCAST
        elif recipient_count > 1 or "," in to_field:
            return RoutingMode.MULTICAST
        else:
            return RoutingMode.UNICAST

    @traces(spec="001", story="US-002", reqs=["REQ-001-US-002-005"])
    def _get_all_volks(self) -> list[str]:
        """Get all registered Volk aliases."""
        all_registrations = self.registry.list_volks()
        # Filter out terminated Volks
        return [
            reg.definition.alias
            for reg in all_registrations
            if reg.status != VolkStatus.TERMINATED
        ]

    @traces(spec="001", story="US-002", reqs=["REQ-001-US-002-008"])
    def _is_volk_online(self, alias: str) -> bool:
        """Check if Volk is currently online.

        REQ-001-US-002-008: Handle offline Volks
        """
        volk_registration = self.registry.get_volk(alias)
        if not volk_registration:
            return False

        return volk_registration.status == VolkStatus.ACTIVE

    @traces(
        spec="001",
        story="US-002",
        reqs=["REQ-001-US-002-037"],
    )
    def _track_message_status(
        self,
        message_id: str,
        from_status: DeliveryStatus,
        to_status: DeliveryStatus,
        reason: str = "",
    ) -> None:
        """Track message status transition.

        REQ-001-US-002-031 to REQ-001-US-002-037: Status tracking

        Note: Preserves transition history. If message is new, creates initial
        status with from_status, then adds transition to to_status.
        """
        if message_id not in self.message_statuses:
            # Create new status with initial from_status
            self.message_statuses[message_id] = MessageStatus(
                message_id=message_id, current_status=from_status
            )

        # Add transition (preserves history via append in MessageStatus.add_transition)
        self.message_statuses[message_id].add_transition(from_status, to_status, reason)

    @traces(spec="001", story="US-002", reqs=["REQ-001-US-002-039"])
    def _add_trace_entry(
        self, message_id: str, event: str, data: dict[str, Any]
    ) -> None:
        """Add tracing entry for message.

        REQ-001-US-002-039: Support message tracing across hops

        Note: Implements FIFO eviction policy. When trace log exceeds
        _max_trace_entries (default 10000), oldest entries are removed.
        """
        entry = {
            "message_id": message_id,
            "event": event,
            "timestamp": datetime.now(UTC).isoformat(),
            "data": data,
        }
        self._trace_log.append(entry)

        # Apply eviction policy if trace log exceeds maximum
        if len(self._trace_log) > self._max_trace_entries:
            # Remove oldest entries (FIFO)
            entries_to_remove = len(self._trace_log) - self._max_trace_entries
            self._trace_log = self._trace_log[entries_to_remove:]
            logger.debug(
                f"Trace log eviction: removed {entries_to_remove} oldest entries"
            )

    @traces(
        spec="001",
        story="US-002",
        reqs=["REQ-001-US-002-036"],
    )
    def get_message_status(self, message_id: str) -> MessageStatus | None:
        """Get status of a message.

        REQ-001-US-002-036: Provide message status query API
        """
        return self.message_statuses.get(message_id)

    @traces(
        spec="001",
        story="US-002",
        reqs=["REQ-001-US-002-039"],
    )
    def get_message_trace(self, message_id: str) -> list[dict[str, Any]]:
        """Get trace history for a message.

        REQ-001-US-002-039: Support message tracing
        """
        return [entry for entry in self._trace_log if entry["message_id"] == message_id]

    @traces(spec="001", story="US-002", reqs=["REQ-001-US-002-040"])
    def get_metrics(self) -> RoutingMetrics:
        """Get routing metrics.

        REQ-001-US-002-040: Generate routing metrics
        """
        return self.metrics