orchestrata_ultimate.cxp.delivery_status

Reference › orchestrata_ultimate › cxp › delivery_status

delivery_status tracks the delivery lifecycle of packets (Spec 004 US-004). validate_delivery_transition checks a move between two DeliveryStatus values against the DELIVERY_TRANSITIONS table defined with the packet model in core.packet. DeliveryTracker applies the rules in memory, keyed by message id. set_status rejects an invalid move with StateTransitionError, otherwise sets the status on the packet and appends it, with a UTC timestamp, to the history that get_history returns. record_failure and get_failure_reasons keep failure reasons. increment_retry raises the retry count and the packet's delivery_attempts, and raises MaxRetriesExceededError once the count has reached the limit; can_retry reports whether retries remain. The limit and retry delay default to MAX_DELIVERY_ATTEMPTS and DEFAULT_RETRY_DELAY_SECONDS from constants. dispatcher drives it during delivery, and servicer and proto_codec use its status types. The companion for work state is work_status. See cxp.

At a glance

Class diagram

Dependency diagram

API

Classes

DeliveryTracker

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

Tracks CXP packet delivery status with history and failure recording.

Functions

Source

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

"""
CXP Delivery Status Tracking.

Implements IDeliveryTracker for tracking packet delivery status,
validating state transitions against the CXP state machine,
and maintaining delivery history with failure tracking.

Spec: 004-cxp, Section US-004
Author: Mohamed
"""

from datetime import UTC, datetime

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.packet import (
    DELIVERY_TRANSITIONS,
    CXPPacket,
    DeliveryStatus,
)
from orchestrata_ultimate.cxp.constants import (
    DEFAULT_RETRY_DELAY_SECONDS,
    MAX_DELIVERY_ATTEMPTS,
)
from orchestrata_ultimate.cxp.exceptions import (
    MaxRetriesExceededError,
    StateTransitionError,
)

__all__ = [
    "DeliveryTracker",
    "validate_delivery_transition",
]


@traces(
    spec="004",
    story="US-004",
    reqs=[
        "REQ-004-US-004-006",
        "REQ-004-US-004-007",
        "REQ-004-US-004-008",
        "REQ-004-US-004-009",
        "REQ-004-US-004-010",
        "REQ-004-US-004-011",
    ],
)
def validate_delivery_transition(
    current: DeliveryStatus, target: DeliveryStatus
) -> bool:
    """Check whether a delivery status transition is valid.

    Uses DELIVERY_TRANSITIONS from the CXP packet module to determine
    whether moving from ``current`` to ``target`` is allowed by the
    state machine defined in Spec 004.

    Args:
        current: The current delivery status.
        target: The desired target delivery status.

    Returns:
        True if the transition is valid, False otherwise.
    """
    allowed = DELIVERY_TRANSITIONS.get(current.value, frozenset())
    return target.value in allowed


@traces(
    spec="004",
    story="US-004",
    reqs=["REQ-004-US-004-018", "REQ-004-US-004-020", "REQ-004-US-004-021"],
)
class DeliveryTracker:
    """Tracks CXP packet delivery status with history and failure recording.

    Implements the IDeliveryTracker protocol interface defined in
    ``src.protocols.cxp.interfaces``.

    State machine (Spec 004):
        QUEUED -> DELIVERED | FAILED
        DELIVERED -> ACKNOWLEDGED | FAILED
        ACKNOWLEDGED -> (terminal)
        FAILED -> (terminal)

    Args:
        max_attempts: Maximum delivery attempts before raising
            MaxRetriesExceededError. Defaults to MAX_DELIVERY_ATTEMPTS (3).
        retry_delay: Default delay in seconds between retries.
            Defaults to DEFAULT_RETRY_DELAY_SECONDS (60).
    """

    def __init__(
        self,
        max_attempts: int = MAX_DELIVERY_ATTEMPTS,
        retry_delay: int = DEFAULT_RETRY_DELAY_SECONDS,
    ) -> None:
        self.max_attempts: int = max_attempts
        self.retry_delay: int = retry_delay

        # Keyed by message_id
        self._history: dict[str, list[tuple[DeliveryStatus, datetime]]] = {}
        self._failure_reasons: dict[str, list[str]] = {}
        self._retry_counts: dict[str, int] = {}

    # ------------------------------------------------------------------
    # IDeliveryTracker interface
    # ------------------------------------------------------------------

    @traces(
        spec="004",
        story="US-004",
        reqs=["REQ-004-US-004-012"],
    )
    def set_status(self, packet: CXPPacket, status: DeliveryStatus) -> None:
        """Update delivery status on a packet.

        Validates the transition against DELIVERY_TRANSITIONS before
        applying. Records the new status with a UTC timestamp in the
        packet's history.

        Args:
            packet: The CXP packet to update.
            status: The target delivery status.

        Raises:
            StateTransitionError: If the transition is not valid per
                the CXP state machine.
        """
        current = packet.delivery_status

        if not validate_delivery_transition(current, status):
            raise StateTransitionError(
                current_state=current.value,
                target_state=status.value,
                state_type="delivery",
            )

        packet.delivery_status = status

        mid = packet.message_id
        if mid not in self._history:
            self._history[mid] = []
        self._history[mid].append((status, datetime.now(UTC)))

    @traces(
        spec="004",
        story="US-004",
        reqs=["REQ-004-US-004-018"],
    )
    def get_history(self, packet: CXPPacket) -> list[tuple[DeliveryStatus, datetime]]:
        """Return the delivery status history for a packet.

        Args:
            packet: The CXP packet to query.

        Returns:
            A list of (DeliveryStatus, datetime) tuples in chronological
            order. Returns an empty list if no history exists.
        """
        return list(self._history.get(packet.message_id, []))

    @traces(
        spec="004",
        story="US-004",
        reqs=["REQ-004-US-004-020"],
    )
    def record_failure(self, packet: CXPPacket, reason: str) -> None:
        """Record a delivery failure with a reason string.

        This does **not** change the packet's delivery_status. Use
        ``set_status`` to transition to FAILED after recording.

        Args:
            packet: The CXP packet that failed delivery.
            reason: Human-readable failure reason.
        """
        mid = packet.message_id
        if mid not in self._failure_reasons:
            self._failure_reasons[mid] = []
        self._failure_reasons[mid].append(reason)

    @traces(spec="004", story="US-004", reqs=["REQ-004-US-004-021"])
    def increment_retry(self, packet: CXPPacket) -> int:
        """Increment the retry count for a packet and return the new value.

        Also increments ``packet.delivery_attempts`` to keep the packet
        object in sync.

        Args:
            packet: The CXP packet being retried.

        Returns:
            The new retry count after incrementing.

        Raises:
            MaxRetriesExceededError: If the retry count has reached
                ``max_attempts``.
        """
        mid = packet.message_id
        current_count = self._retry_counts.get(mid, 0)

        if current_count >= self.max_attempts:
            raise MaxRetriesExceededError(
                packet_id=mid,
                attempts=current_count,
                max_attempts=self.max_attempts,
            )

        new_count = current_count + 1
        self._retry_counts[mid] = new_count
        packet.delivery_attempts = new_count

        return new_count

    # ------------------------------------------------------------------
    # Extended methods (beyond IDeliveryTracker interface)
    # ------------------------------------------------------------------

    @traces(spec="004", story="US-004", reqs=["REQ-004-US-004-022"])
    def can_retry(self, packet: CXPPacket) -> bool:
        """Check whether the packet has retries remaining.

        Args:
            packet: The CXP packet to check.

        Returns:
            True if the retry count is below ``max_attempts``,
            False otherwise.
        """
        mid = packet.message_id
        current_count = self._retry_counts.get(mid, 0)
        return current_count < self.max_attempts

    def get_failure_reasons(self, packet: CXPPacket) -> list[str]:
        """Return all recorded failure reasons for a packet.

        Args:
            packet: The CXP packet to query.

        Returns:
            A list of failure reason strings in recording order.
            Returns an empty list if no failures have been recorded.
        """
        return list(self._failure_reasons.get(packet.message_id, []))