orchestrata_ultimate.cxp.postman_adapter

Reference › orchestrata_ultimate › cxp › postman_adapter

PostmanAdapter bridges CXP packets into the Postman message broker's per-recipient priority queues (see the postman package). The agents package also imports it. Queues are keyed by canonical alias: _resolve_recipient maps a display name such as Postman to its alias (U:PS) and raises RoutingError for an empty recipient. See cxp.

enqueue first calls _assert_enqueueable, which rejects a packet without a recipient (RoutingError), an expired packet, or one already acknowledged or failed (DeliveryError). It then sets the status to QUEUED, appends the packet to the recipient's queue, sorts the queue with _sort_key (priority order from priority) and returns a queue-position id. dequeue takes the next packet for a recipient, peek looks at it, and get_queue_size, get_all_queued, active_queues and history report state. acknowledge requires the packet to be in the DELIVERED state and marks it acknowledged; reject stores a reason in the packet metadata and marks it permanently failed. Both add the packet to the history. Routing results come from types.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

PostmanAdapter

Decorators: traces(spec='004', story='INTG', reqs=['REQ-004-INTG-010', 'REQ-004-INTG-012']) · Line 86

Adapter between CXP packets and the Postman message-routing subsystem.

Functions

Source

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

"""
CXP Postman Adapter.

Bridges CXP packet format (flat 11 fields) to the Postman queue/routing
system.  Provides in-memory priority-ordered queues per recipient and
delivery-status lifecycle management.

Spec: 004-cxp
Author: Mohamed
"""

import logging
import uuid
from collections import defaultdict
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.packet import CXPPacket, DeliveryStatus
from orchestrata_ultimate.cxp.exceptions import DeliveryError, RoutingError
from orchestrata_ultimate.cxp.types import VOLK_ALIASES

# Spec: 004 - CXP Protocol


logger = logging.getLogger(__name__)

__all__ = ["PostmanAdapter"]


# Reverse lookup: alias -> name  (e.g. "U:PS" -> "Postman")
_ALIAS_TO_NAME: dict[str, str] = {v: k for k, v in VOLK_ALIASES.items()}

# All known recipient identifiers (names + aliases)
_KNOWN_RECIPIENTS: frozenset = frozenset(
    list(VOLK_ALIASES.keys()) + list(VOLK_ALIASES.values())
)


@traces(spec="002", story="FR3", reqs=["REQ-002-FR3-006"])
def _resolve_recipient(recipient: str) -> str:
    """Normalise a recipient to its canonical alias (e.g. "Postman" -> "U:PS").

    Accepts both display names and aliases.  Returns the alias form so
    queue keys are always consistent.

    Raises:
        RoutingError: If *recipient* is empty.
    """
    if not recipient:
        raise RoutingError("Recipient cannot be empty")

    # Already an alias?
    if recipient in VOLK_ALIASES.values():
        return recipient

    # Display name?
    alias = VOLK_ALIASES.get(recipient)
    if alias is not None:
        return alias

    # Unknown — still allow it (forward-compatible with dynamic Volks),
    # but use the raw value as the queue key.
    return recipient


@traces(
    spec="004",
    story="US-009",
    reqs=["REQ-004-US-009-005", "REQ-004-US-009-006"],
)
def _sort_key(packet: CXPPacket) -> tuple:
    """Sort key for priority-ordered, FIFO-within-priority queues.

    Higher priority values are dequeued first.  Within the same priority
    packets are ordered by timestamp (earliest first).
    """
    # Negate priority so that higher numeric values sort first.
    return (-int(packet.priority), packet.timestamp)


@traces(
    spec="004",
    story="INTG",
    reqs=["REQ-004-INTG-010", "REQ-004-INTG-012"],
)
class PostmanAdapter:
    """Adapter between CXP packets and the Postman message-routing subsystem.

    Manages per-recipient priority queues in memory.  An optional external
    *queue_manager* (e.g. ``orchestrata.postman.queue_manager.QueueManager``)
    can be injected; when present the adapter delegates storage to it.
    When absent, the adapter uses its own in-memory structures.

    Parameters
    ----------
    queue_manager : optional
        External queue manager instance.  If ``None`` the adapter uses
        built-in ``dict[str, list[CXPPacket]]`` queues.
    """

    def __init__(self, queue_manager: Any = None) -> None:
        self._external_qm = queue_manager
        self._queues: dict[str, list[CXPPacket]] = defaultdict(list)
        self._history: list[CXPPacket] = []

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

    @traces(
        spec="004",
        story="US-009",
        reqs=["REQ-004-INTG-010", "REQ-004-US-009-008"],
    )
    async def enqueue(self, packet: CXPPacket) -> str:
        """Enqueue a CXP packet for delivery.

        The packet's ``to_agent`` field determines the destination queue.
        A unique queue-position identifier is returned.

        Parameters
        ----------
        packet : CXPPacket
            The packet to enqueue.

        Returns
        -------
        str
            A unique queue-position ID for tracking.

        Raises
        ------
        RoutingError
            If the packet has no ``to_agent`` set.
        DeliveryError
            If the packet is expired or already in a terminal state.
        """
        self._assert_enqueueable(packet)

        recipient_key = _resolve_recipient(packet.to_agent)
        packet.delivery_status = DeliveryStatus.QUEUED

        queue = self._queues[recipient_key]
        queue.append(packet)
        queue.sort(key=_sort_key)

        return f"q-{recipient_key}-{uuid.uuid4().hex[:8]}"

    @traces(spec="004", reqs=["REQ-004-INTG-010"])
    def _assert_enqueueable(self, packet: CXPPacket) -> None:
        """Raise if the packet cannot be enqueued.

        Checks for missing recipient, expiry, and terminal delivery state.

        Parameters
        ----------
        packet : CXPPacket
            The packet to check.

        Raises
        ------
        RoutingError
            If ``to_agent`` is empty.
        DeliveryError
            If the packet is expired or in a terminal delivery state.
        """
        if not packet.to_agent:
            raise RoutingError(
                "Cannot enqueue packet without a recipient (to_agent is empty)",
                packet_id=packet.message_id,
            )

        if packet.is_expired():
            raise DeliveryError(
                f"Cannot enqueue expired packet {packet.message_id}",
                packet_id=packet.message_id,
            )

        if packet.delivery_status in (
            DeliveryStatus.ACKNOWLEDGED,
            DeliveryStatus.FAILED,
        ):
            raise DeliveryError(
                f"Cannot enqueue packet in terminal state "
                f"'{packet.delivery_status.value}'",
                packet_id=packet.message_id,
            )

    @traces(
        spec="004",
        story="US-009",
        reqs=["REQ-004-INTG-012", "REQ-004-US-009-008"],
    )
    async def dequeue(self, recipient: str) -> CXPPacket | None:
        """Dequeue the highest-priority packet for *recipient*.

        Marks the packet as DELIVERED before returning it.  Expired
        packets are silently discarded.

        Returns ``None`` when the queue is empty.
        """
        recipient_key = _resolve_recipient(recipient)
        queue = self._queues.get(recipient_key)
        if not queue:
            return None

        # Skip and discard expired packets.
        while queue:
            packet = queue[0]
            if packet.is_expired():
                queue.pop(0)
                packet.mark_failed(permanent=True)
                self._history.append(packet)
                continue
            break
        else:
            return None

        packet = queue.pop(0)
        packet.mark_delivered()
        self._history.append(packet)
        return packet

    @traces(spec="002", story="US3", reqs=["REQ-002-US-003-013"])
    async def peek(self, recipient: str) -> CXPPacket | None:
        """Return the next packet for *recipient* without removing it.

        Expired packets at the front of the queue are discarded first.
        Returns ``None`` when the queue is empty.
        """
        recipient_key = _resolve_recipient(recipient)
        queue = self._queues.get(recipient_key)
        if not queue:
            return None

        # Discard expired head entries.
        while queue:
            if queue[0].is_expired():
                expired = queue.pop(0)
                expired.mark_failed(permanent=True)
                self._history.append(expired)
                continue
            return queue[0]

        return None

    @traces(spec="004", story="INTG", reqs=["REQ-004-INTG-012"])
    async def get_queue_size(self, recipient: str) -> int:
        """Return the number of packets currently queued for *recipient*.

        Expired packets are **not** pruned by this call — it returns the
        raw queue length for speed.
        """
        recipient_key = _resolve_recipient(recipient)
        return len(self._queues.get(recipient_key, []))

    async def get_all_queued(self, recipient: str) -> list[CXPPacket]:
        """Return a snapshot of all queued packets for *recipient*.

        The returned list is a shallow copy; callers may iterate freely
        without affecting the internal queue.
        """
        recipient_key = _resolve_recipient(recipient)
        return list(self._queues.get(recipient_key, []))

    @traces(
        spec="004",
        story="INTG",
        reqs=["REQ-004-INTG-011", "REQ-004-INTG-017"],
    )
    async def acknowledge(self, packet: CXPPacket) -> None:
        """Mark *packet* as acknowledged.

        Moves the packet to history if it was previously delivered.

        Raises
        ------
        DeliveryError
            If the packet is not in ``DELIVERED`` state.
        """
        if packet.delivery_status != DeliveryStatus.DELIVERED:
            raise DeliveryError(
                f"Cannot acknowledge packet in state "
                f"'{packet.delivery_status.value}'; expected 'delivered'",
                packet_id=packet.message_id,
            )

        packet.mark_acknowledged()

        # Ensure it lands in history (may already be there from dequeue).
        if packet not in self._history:
            self._history.append(packet)

    @traces(
        spec="004",
        story="INTG",
        reqs=["REQ-004-INTG-011", "REQ-004-INTG-013"],
    )
    async def reject(self, packet: CXPPacket, reason: str) -> None:
        """Mark *packet* as failed with the given *reason*.

        The reason is stored in ``packet.metadata["reject_reason"]``.

        Raises
        ------
        DeliveryError
            If the packet is already in a terminal state.
        """
        if packet.delivery_status in (
            DeliveryStatus.ACKNOWLEDGED,
            DeliveryStatus.FAILED,
        ):
            raise DeliveryError(
                f"Cannot reject packet already in terminal state "
                f"'{packet.delivery_status.value}'",
                packet_id=packet.message_id,
            )

        packet.metadata["reject_reason"] = reason
        packet.mark_failed(permanent=True)

        if packet not in self._history:
            self._history.append(packet)

    # ------------------------------------------------------------------
    # Introspection helpers
    # ------------------------------------------------------------------

    @property
    def history(self) -> list[CXPPacket]:
        """Shallow copy of all packets that have left the queues."""
        return list(self._history)

    @property
    @traces(spec="004", story="INTG", reqs=["REQ-004-INTG-012"])
    def active_queues(self) -> dict[str, int]:
        """Mapping of recipient alias -> current queue depth."""
        return {k: len(v) for k, v in self._queues.items() if v}