orchestrata_ultimate.cxp.work_status

Reference › orchestrata_ultimate › cxp › work_status

work_status tracks the work lifecycle of a packet, separate from its delivery (Spec 004 US-005). validate_work_transition checks a move between two WorkStatus values against the WORK_TRANSITIONS table from core.packet. WorkTracker keeps a history and a metadata record per message id. set_status raises StateTransitionError, listing the allowed targets, for an invalid move; otherwise it updates the packet, records the transition, and updates lifecycle metadata: a start time for IN_PROGRESS, a resolved time for RESOLVED, an archived time for ARCHIVED, and for PENDING a requeue count with the start and resolve times cleared. The helpers start_work, resolve_work, archive_work and requeue_work wrap common transitions. get_work_duration returns the seconds between start and resolution, or None if the packet has not been both started and resolved. get_history and get_metadata return the records. The codec proto_codec and servicer use its status types. See delivery_status and cxp.

At a glance

Class diagram

Dependency diagram

API

Classes

WorkTracker

Line 47

Tracks work status transitions and lifecycle metadata for CXP packets.

Functions

Source

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

# mypy: ignore-errors
"""CXP Work Status Tracking per Spec 004 Section US-005.

Tracks work status transitions for CXP packets with full lifecycle
metadata and history. Enforces valid transitions defined in WORK_TRANSITIONS.

Transition graph:
    PENDING -> IN_PROGRESS -> RESOLVED -> ARCHIVED
                   |
                   +-------> PENDING  (re-queue)
"""

from datetime import UTC, datetime
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.packet import WORK_TRANSITIONS, CXPPacket, WorkStatus
from orchestrata_ultimate.cxp.exceptions import StateTransitionError


@traces(
    spec="004",
    story="US-005",
    reqs=[
        "REQ-004-US-005-006",
        "REQ-004-US-005-007",
        "REQ-004-US-005-008",
        "REQ-004-US-005-009",
        "REQ-004-US-005-010",
        "REQ-004-US-005-011",
    ],
)
def validate_work_transition(current: WorkStatus, target: WorkStatus) -> bool:
    """Check whether a work status transition is allowed.

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

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


class WorkTracker:
    """Tracks work status transitions and lifecycle metadata for CXP packets.

    Each packet is identified by its ``message_id``. The tracker maintains
    a full history of status changes and records timestamps for key lifecycle
    events (started, resolved, archived).

    Internal state:
        _history:  Dict mapping message_id to ordered list of
                   (WorkStatus, datetime) tuples.
        _metadata: Dict mapping message_id to lifecycle metadata dict
                   containing started_at, resolved_at, archived_at, and
                   requeue_count.
    """

    def __init__(self) -> None:
        self._history: dict[str, list[tuple[WorkStatus, datetime]]] = {}
        self._metadata: dict[str, dict[str, Any]] = {}

    @traces(
        spec="004", story="US-005", reqs=["REQ-004-US-005-012", "REQ-004-US-005-022"]
    )
    def _ensure_tracked(self, packet: CXPPacket) -> None:
        """Register a packet if it is not already being tracked."""
        mid = packet.message_id
        if mid not in self._history:
            now = datetime.now(UTC)
            self._history[mid] = [(packet.work_status, now)]
            self._metadata[mid] = {
                "started_at": None,
                "resolved_at": None,
                "archived_at": None,
                "requeue_count": 0,
            }

    @traces(
        spec="004",
        story="US-005",
        reqs=[
            "REQ-004-US-005-012",
            "REQ-004-US-005-022",
        ],
    )
    def set_status(self, packet: CXPPacket, status: WorkStatus) -> None:
        """Validate and apply a work status transition on *packet*.

        Updates the packet's ``work_status`` attribute, records the
        transition in history, and updates lifecycle metadata timestamps.

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

        Raises:
            StateTransitionError: If the transition from the packet's current
                work_status to *status* is not permitted.
        """
        self._ensure_tracked(packet)

        current = packet.work_status
        if not validate_work_transition(current, status):
            allowed = sorted(WORK_TRANSITIONS.get(current.value, frozenset()))
            raise StateTransitionError(
                current_state=current.value,
                target_state=status.value,
                state_type="work",
                details={"allowed_targets": allowed},
            )

        now = datetime.now(UTC)
        packet.work_status = status
        self._history[packet.message_id].append((status, now))

        mid = packet.message_id
        if status == WorkStatus.IN_PROGRESS:
            self._metadata[mid]["started_at"] = now
        elif status == WorkStatus.RESOLVED:
            self._metadata[mid]["resolved_at"] = now
        elif status == WorkStatus.ARCHIVED:
            self._metadata[mid]["archived_at"] = now
        elif status == WorkStatus.PENDING:
            # Re-queue: increment counter, clear started_at for fresh cycle
            self._metadata[mid]["requeue_count"] += 1
            self._metadata[mid]["started_at"] = None
            self._metadata[mid]["resolved_at"] = None

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

        Args:
            packet: The CXP packet to query.

        Returns:
            Ordered list of (WorkStatus, datetime) tuples. Returns an empty
            list if the packet has never been tracked.
        """
        return list(self._history.get(packet.message_id, []))

    @traces(spec="004", story="US-005", reqs=["REQ-004-US-005-007"])
    def start_work(self, packet: CXPPacket) -> None:
        """Transition *packet* to IN_PROGRESS.

        Raises:
            StateTransitionError: If the transition is not valid from the
                packet's current status.
        """
        self.set_status(packet, WorkStatus.IN_PROGRESS)

    @traces(
        spec="004",
        story="US-005",
        reqs=["REQ-004-US-005-008"],
    )
    def resolve_work(self, packet: CXPPacket) -> None:
        """Transition *packet* to RESOLVED.

        Raises:
            StateTransitionError: If the transition is not valid from the
                packet's current status.
        """
        self.set_status(packet, WorkStatus.RESOLVED)

    @traces(spec="004", story="US-005", reqs=["REQ-004-US-005-009"])
    def archive_work(self, packet: CXPPacket) -> None:
        """Transition *packet* to ARCHIVED.

        Raises:
            StateTransitionError: If the transition is not valid from the
                packet's current status.
        """
        self.set_status(packet, WorkStatus.ARCHIVED)

    @traces(spec="004", story="US-005", reqs=["REQ-004-US-005-010"])
    def requeue_work(self, packet: CXPPacket) -> None:
        """Transition *packet* back to PENDING (re-queue).

        Only valid from IN_PROGRESS status. Increments the requeue counter
        and clears started_at/resolved_at for a fresh work cycle.

        Raises:
            StateTransitionError: If the transition is not valid from the
                packet's current status.
        """
        self.set_status(packet, WorkStatus.PENDING)

    @traces(spec="004", story="US-005", reqs=["REQ-004-US-005-023"])
    def get_work_duration(self, packet: CXPPacket) -> float | None:
        """Return work duration in seconds from start to resolution.

        Args:
            packet: The CXP packet to query.

        Returns:
            Duration in seconds as a float, or None if the packet has not
            been both started and resolved.
        """
        meta = self._metadata.get(packet.message_id)
        if meta is None:
            return None

        started = meta.get("started_at")
        resolved = meta.get("resolved_at")
        if started is None or resolved is None:
            return None

        delta = resolved - started
        return delta.total_seconds()

    def get_metadata(self, packet: CXPPacket) -> dict[str, Any]:
        """Return work lifecycle metadata for *packet*.

        Args:
            packet: The CXP packet to query.

        Returns:
            Dict containing:
                - started_at: datetime or None
                - resolved_at: datetime or None
                - archived_at: datetime or None
                - requeue_count: int
                - current_status: str (current work_status value)
                - transition_count: int (number of transitions recorded)
        """
        self._ensure_tracked(packet)
        mid = packet.message_id
        meta = dict(self._metadata[mid])
        meta["current_status"] = packet.work_status.value
        meta["transition_count"] = len(self._history[mid])
        return meta