orchestrata_ultimate.cxp.dispatcher

Reference › orchestrata_ultimate › cxp › dispatcher

CXPDispatcher combines routing with delivery for CXP packets (Spec 004). It is built from an optional router, an optional DeliveryTracker from delivery_status and a retry limit that defaults to MAX_DELIVERY_ATTEMPTS in constants. See cxp.

dispatch asks the router for a decision when one is configured. A REJECT result is recorded as a failed DeliveryResult with code ROUTE_REJECTED; otherwise the first routing target is the destination, or the packet's own recipient when there is no router. _deliver then makes one attempt: it increments the packet's attempt count, sets its status to DELIVERED through the tracker (which enforces the state machine) or directly, and returns a successful DeliveryResult with the latency. Any exception becomes a failed result with code DELIVERY_FAILED through _record_failure. dispatch_to delivers to a named recipient, and broadcast gives each recipient an independent attempt so that one failure does not block the others. get_dispatch_history and get_dispatch_count expose the results kept in memory.

At a glance

Class diagram

Dependency diagram

API

Classes

CXPDispatcher

Decorators: traces(spec='004', story='US-004', reqs=['REQ-004-US-004-020']) · Line 43

CXP-level packet dispatcher.

Source

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

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

Responsible for actually SENDING a CXP packet: routes via the
CXPRouter, updates delivery status, records results in the
DeliveryTracker, and maintains dispatch history.

Spec: 004-cxp
Author: Mohamed
"""

import logging
from datetime import UTC, datetime
from typing import TYPE_CHECKING

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.packet import CXPPacket, DeliveryResult, DeliveryStatus
from orchestrata_ultimate.cxp.constants import MAX_DELIVERY_ATTEMPTS
from orchestrata_ultimate.cxp.types import RoutingDecision, RoutingResult

if TYPE_CHECKING:
    from orchestrata_ultimate.cxp.delivery_status import DeliveryTracker
    from orchestrata_ultimate.cxp.router import CXPRouter

# Spec: 004 - CXP Protocol


logger = logging.getLogger(__name__)

__all__ = [
    "CXPDispatcher",
]


@traces(
    spec="004",
    story="US-004",
    reqs=[
        "REQ-004-US-004-020",
    ],
)
class CXPDispatcher:
    """CXP-level packet dispatcher.

    Combines routing (via :class:`CXPRouter`) with delivery tracking to
    actually send packets and record the outcome.

    The dispatch flow:
    1. Route the packet via the router to determine the target(s).
    2. Attempt delivery (mark DELIVERED on success, FAILED on failure).
    3. Record the result in the delivery tracker (if available).
    4. Append the :class:`DeliveryResult` to the dispatch history.

    Args:
        router: An optional CXPRouter instance. If ``None``, dispatch_to
            and broadcast still work (they skip the routing step).
        delivery_tracker: An optional DeliveryTracker for recording
            delivery state transitions.
        max_retries: Maximum delivery attempts before giving up.
            Defaults to :data:`MAX_DELIVERY_ATTEMPTS` (3).
    """

    def __init__(
        self,
        router: CXPRouter | None = None,
        delivery_tracker: DeliveryTracker | None = None,
        max_retries: int = MAX_DELIVERY_ATTEMPTS,
    ) -> None:
        self._router = router
        self._delivery_tracker = delivery_tracker
        self._max_retries: int = max_retries
        self._history: list[DeliveryResult] = []

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

    async def dispatch(self, packet: CXPPacket) -> DeliveryResult:
        """Route and deliver a CXP packet.

        Uses the router to determine the destination, then delivers.
        If no router is configured, delivers directly to ``packet.to_agent``.

        Args:
            packet: The CXP packet to dispatch.

        Returns:
            A DeliveryResult describing the outcome.

        Raises:
            DeliveryError: If delivery fails after all retry attempts.
        """
        destination = packet.to_agent

        # Route via the router if available.
        if self._router is not None:
            routing_result: RoutingResult = await self._router.route(packet)

            if routing_result.decision == RoutingDecision.REJECT:
                return self._record_failure(
                    packet=packet,
                    destination=destination,
                    error_message=routing_result.reason or "Routing rejected",
                    error_code="ROUTE_REJECTED",
                )

            if routing_result.targets:
                destination = routing_result.targets[0]

        return await self._deliver(packet, destination)

    async def dispatch_to(self, packet: CXPPacket, recipient: str) -> DeliveryResult:
        """Deliver a packet to a specific recipient, bypassing the router.

        Args:
            packet: The CXP packet to deliver.
            recipient: Target Volk alias (e.g. ``"U:Cd"``).

        Returns:
            A DeliveryResult describing the outcome.
        """
        return await self._deliver(packet, recipient)

    async def broadcast(
        self, packet: CXPPacket, recipients: list[str]
    ) -> list[DeliveryResult]:
        """Deliver a packet to multiple recipients.

        Each recipient gets an independent delivery attempt. Failures
        for one recipient do not block delivery to others.

        Args:
            packet: The CXP packet to broadcast.
            recipients: List of target Volk aliases.

        Returns:
            A list of DeliveryResult objects, one per recipient.
        """
        results: list[DeliveryResult] = []
        for recipient in recipients:
            result = await self._deliver(packet, recipient)
            results.append(result)
        return results

    @traces(spec="004", story="US-004", reqs=["REQ-004-US-004-018"])
    def get_dispatch_history(self) -> list[DeliveryResult]:
        """Return all recorded dispatch results.

        Returns:
            List of DeliveryResult objects in chronological order.
        """
        return list(self._history)

    def get_dispatch_count(self) -> int:
        """Return the total number of dispatches performed.

        Returns:
            Integer count of dispatch attempts.
        """
        return len(self._history)

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

    async def _deliver(self, packet: CXPPacket, destination: str) -> DeliveryResult:
        """Attempt to deliver a packet to a single destination.

        On success the packet's delivery_status is set to DELIVERED.
        On failure it is set to FAILED.  The delivery tracker is updated
        if one is configured.

        Args:
            packet: The CXP packet to deliver.
            destination: Target Volk alias.

        Returns:
            A DeliveryResult for this delivery attempt.
        """
        start = datetime.now(UTC)

        try:
            # Mark the packet as delivered.
            packet.delivery_attempts += 1

            # Record in tracker (which also sets packet status).
            if self._delivery_tracker is not None:
                self._delivery_tracker.set_status(packet, DeliveryStatus.DELIVERED)
            else:
                packet.delivery_status = DeliveryStatus.DELIVERED

            elapsed = (datetime.now(UTC) - start).total_seconds()

            result = DeliveryResult(
                success=True,
                packet_id=packet.message_id,
                destination=destination,
                timestamp=datetime.now(UTC),
                delivery_latency=elapsed,
            )
            self._history.append(result)
            return result

        except Exception as exc:
            return self._record_failure(
                packet=packet,
                destination=destination,
                error_message=str(exc),
                error_code="DELIVERY_FAILED",
            )

    def _record_failure(
        self,
        packet: CXPPacket,
        destination: str,
        error_message: str,
        error_code: str,
    ) -> DeliveryResult:
        """Record a failed delivery attempt.

        Sets the packet status to FAILED, records in tracker, and
        appends to history.

        Args:
            packet: The packet that failed delivery.
            destination: Intended target.
            error_message: Human-readable failure reason.
            error_code: Machine-readable error code.

        Returns:
            A DeliveryResult marked as failed.
        """

        if self._delivery_tracker is not None:
            self._delivery_tracker.record_failure(packet, error_message)
            try:
                self._delivery_tracker.set_status(packet, DeliveryStatus.FAILED)
            except Exception as exc:
                logger.debug(
                    "Tracker state transition to FAILED skipped for packet %s: %s",
                    getattr(packet, "message_id", "unknown"),
                    exc,
                )  # Tracker state transition may fail if already FAILED.
        else:
            packet.delivery_status = DeliveryStatus.FAILED

        result = DeliveryResult(
            success=False,
            packet_id=packet.message_id,
            destination=destination,
            timestamp=datetime.now(UTC),
            error_message=error_message,
            error_code=error_code,
        )
        self._history.append(result)
        return result