orchestrata_ultimate.cxp.headers

Reference › orchestrata_ultimate › cxp › headers

headers handles the 11 core header fields of a CXP packet (Spec 004 US-002): message id, sender, recipient, thread id, subject, payload, timestamp, priority, delivery status, work status and protocol version. The module-level validators check single values: validate_uuid, validate_volk_alias (the pattern from constants) and validate_subject, plus private validators for each field. extract_headers returns the fields as a dict.

HeaderManager is a stateless class of classmethods. get_headers reads all 11 fields with enum values normalised, get_immutable_headers returns message id, sender, recipient, timestamp and protocol version, and get_mutable_headers returns priority and the two statuses. validate_header checks one field and raises FieldValidationError for an unknown name; validate_all_headers returns a list of error strings, empty when the headers are valid. set_priority validates the range, and set_delivery_status and set_work_status apply a state change only if the state machine allows it, raising StateTransitionError otherwise. See validator, exceptions and cxp.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

HeaderManager

Decorators: traces(spec='004', story='US-002', reqs=['REQ-004-US-002-001', 'REQ-004-US-002-013']) · Line 261

Stateless manager for CXP packet header operations.

Functions

Source

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

"""
CXP Header Management.

Extracts, validates, and mutates CXP packet header fields per
Spec 004 Section US-002.

Performance target: header extraction < 1ms for all 11 core fields.

Author: Mohamed
"""

# Spec: 004 - CXP Protocol

from collections.abc import Callable
from datetime import datetime
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.packet import (
    DELIVERY_TRANSITIONS,
    WORK_TRANSITIONS,
    CXPPacket,
    DeliveryStatus,
    Priority,
    WorkStatus,
)
from orchestrata_ultimate.cxp.constants import (
    PRIORITY_MAX,
    PRIORITY_MIN,
    SUBJECT_MAX_LENGTH,
    UUID_REGEX,
    VOLK_ALIAS_REGEX,
)
from orchestrata_ultimate.cxp.exceptions import (
    FieldValidationError,
    StateTransitionError,
)

__all__ = [
    "HeaderManager",
    "extract_headers",
    "validate_volk_alias",
    "validate_uuid",
    "validate_subject",
]


# ============================================================================
# Core header field definitions
# ============================================================================

# The 11 core header fields per Spec 004.
CORE_HEADER_FIELDS: tuple = (
    "message_id",
    "from_agent",
    "to_agent",
    "thread_id",
    "subject",
    "payload",
    "timestamp",
    "priority",
    "delivery_status",
    "work_status",
    "protocol_version",
)

IMMUTABLE_FIELDS: frozenset = frozenset(
    {
        "message_id",
        "from_agent",
        "to_agent",
        "timestamp",
        "protocol_version",
    }
)

MUTABLE_FIELDS: frozenset = frozenset(
    {
        "priority",
        "delivery_status",
        "work_status",
    }
)

# Fields that are core but neither in the immutable nor mutable governance set.
# They are still extracted by get_headers but do not have setter guards.
INFO_FIELDS: frozenset = frozenset(
    {
        "thread_id",
        "subject",
        "payload",
    }
)


# ============================================================================
# Module-level convenience functions
# ============================================================================


@traces(
    spec="004",
    story="US-001",
    reqs=["REQ-004-US-001-013"],
)
def validate_uuid(value: str) -> bool:
    """Validate that *value* matches the CXP UUID pattern.

    Returns True for a valid UUID v4 hex string, False otherwise.
    """
    if not isinstance(value, str):
        return False
    return UUID_REGEX.match(value) is not None


@traces(
    spec="004",
    story="US-001",
    reqs=["REQ-004-US-001-014", "REQ-004-US-001-015"],
)
def validate_volk_alias(alias: str) -> bool:
    """Validate a Volk alias against the CXP alias pattern ``U:Xx``.

    Returns True if *alias* matches ``^U:[A-Z][a-z]$``, False otherwise.
    """
    if not isinstance(alias, str):
        return False
    return VOLK_ALIAS_REGEX.match(alias) is not None


@traces(
    spec="004",
    story="US-001",
    reqs=[
        "REQ-004-US-001-017",
        "REQ-004-US-001-018",
        "REQ-004-US-007-005",
    ],
)
def validate_subject(subject: str) -> bool:
    """Validate a subject line.

    A valid subject is a non-empty string no longer than SUBJECT_MAX_LENGTH.
    """
    if not isinstance(subject, str):
        return False
    return 0 < len(subject) <= SUBJECT_MAX_LENGTH


# ============================================================================
# Per-field header validators
#
# Each function validates one core header field independently and returns a
# bool.  ``_HEADER_FIELD_VALIDATORS`` dispatches by field name; this keeps
# ``HeaderManager.validate_header`` a flat table lookup rather than a long
# if/elif chain.
# ============================================================================


def _validate_agent_field(value: Any) -> bool:
    """Validate a from_agent/to_agent value (Volk alias or non-empty string)."""
    if not isinstance(value, str) or not value:
        return False
    return validate_volk_alias(value) or validate_uuid(value) or len(value) > 0


def _validate_thread_id(value: Any) -> bool:
    """Validate a thread_id (non-empty string)."""
    return isinstance(value, str) and len(value) > 0


def _validate_priority(value: Any) -> bool:
    """Validate a priority (Priority enum or int within [MIN, MAX])."""
    if isinstance(value, Priority):
        return True
    if isinstance(value, int):
        return PRIORITY_MIN <= value <= PRIORITY_MAX
    return False


def _validate_delivery_status(value: Any) -> bool:
    """Validate a delivery_status (DeliveryStatus enum or valid enum string)."""
    if isinstance(value, DeliveryStatus):
        return True
    if isinstance(value, str):
        try:
            DeliveryStatus(value)
            return True
        except ValueError:
            return False
    return False


def _validate_work_status(value: Any) -> bool:
    """Validate a work_status (WorkStatus enum or valid enum string)."""
    if isinstance(value, WorkStatus):
        return True
    if isinstance(value, str):
        try:
            WorkStatus(value)
            return True
        except ValueError:
            return False
    return False


def _validate_protocol_version(value: Any) -> bool:
    """Validate a protocol_version (loose semver: ``digits.digits.digits``)."""
    if not isinstance(value, str):
        return False
    parts = value.split(".")
    if len(parts) != 3:
        return False
    return all(part.isdigit() for part in parts)


# Maps each core header field name to its independent validator.  ``payload``
# accepts any type; ``timestamp`` must be a datetime.
# Typed as Callable[..., bool], not Any: every value here is a one-arg
# predicate, and `dict[str, Any]` let that Any leak through
# `validate_header`'s `return validator(value)` — mypy no-any-return, so a
# validator that stopped returning bool would not have been caught.
_HEADER_FIELD_VALIDATORS: dict[str, Callable[[Any], bool]] = {
    "message_id": validate_uuid,
    "from_agent": _validate_agent_field,
    "to_agent": _validate_agent_field,
    "thread_id": _validate_thread_id,
    "subject": validate_subject,
    "payload": lambda value: True,
    "timestamp": lambda value: isinstance(value, datetime),
    "priority": _validate_priority,
    "delivery_status": _validate_delivery_status,
    "work_status": _validate_work_status,
    "protocol_version": _validate_protocol_version,
}


@traces(
    spec="004",
    story="US-002",
    reqs=["REQ-004-US-002-001"],
)
def extract_headers(packet: CXPPacket) -> dict[str, Any]:
    """Convenience wrapper: extract all 11 core header fields from *packet*.

    Equivalent to ``HeaderManager.get_headers(packet)``.
    """
    return HeaderManager.get_headers(packet)


# ============================================================================
# HeaderManager
# ============================================================================


@traces(
    spec="004",
    story="US-002",
    reqs=["REQ-004-US-002-001", "REQ-004-US-002-013"],
)
class HeaderManager:
    """Stateless manager for CXP packet header operations.

    All methods are classmethods so no instantiation is required,
    though the class can be instantiated if a caller prefers OOP style.
    """

    # -- Extraction -----------------------------------------------------------

    @traces(
        spec="004",
        story="US-002",
        reqs=[
            "REQ-004-US-002-001",
            "REQ-004-US-002-002",
            "REQ-004-US-002-003",
            "REQ-004-US-002-004",
            "REQ-004-US-002-005",
            "REQ-004-US-002-006",
            "REQ-004-US-002-007",
            "REQ-004-US-002-008",
            "REQ-004-US-002-009",
            "REQ-004-US-002-010",
            "REQ-004-US-002-011",
            "REQ-004-US-002-012",
        ],
    )
    @classmethod
    def get_headers(cls, packet: CXPPacket) -> dict[str, Any]:
        """Extract all 11 core header fields as a dict.

        Performance target: < 1ms.  This is a direct attribute read with
        enum-value normalisation, no serialization overhead.
        """
        delivery = packet.delivery_status
        work = packet.work_status

        return {
            "message_id": packet.message_id,
            "from_agent": packet.from_agent,
            "to_agent": packet.to_agent,
            "thread_id": packet.thread_id,
            "subject": packet.subject,
            "payload": packet.payload,
            "timestamp": packet.timestamp,
            "priority": packet.priority,
            "delivery_status": (
                delivery.value if isinstance(delivery, DeliveryStatus) else delivery
            ),
            "work_status": work.value if isinstance(work, WorkStatus) else work,
            "protocol_version": packet.protocol_version,
        }

    @classmethod
    def get_immutable_headers(cls, packet: CXPPacket) -> dict[str, Any]:
        """Extract only the immutable header fields."""
        return {
            "message_id": packet.message_id,
            "from_agent": packet.from_agent,
            "to_agent": packet.to_agent,
            "timestamp": packet.timestamp,
            "protocol_version": packet.protocol_version,
        }

    @classmethod
    def get_mutable_headers(cls, packet: CXPPacket) -> dict[str, Any]:
        """Extract only the mutable header fields."""
        delivery = packet.delivery_status
        work = packet.work_status
        return {
            "priority": packet.priority,
            "delivery_status": (
                delivery.value if isinstance(delivery, DeliveryStatus) else delivery
            ),
            "work_status": work.value if isinstance(work, WorkStatus) else work,
        }

    # -- Validation -----------------------------------------------------------

    @traces(
        spec="004",
        story="US-002",
        reqs=[
            "REQ-004-US-002-014",
            "REQ-004-US-007-002",
        ],
    )
    @classmethod
    def validate_header(cls, field_name: str, value: Any) -> bool:
        """Validate a single header field value.

        Args:
            field_name: One of the 11 core header field names.
            value:      The value to validate.

        Returns:
            True if valid, False otherwise.

        Raises:
            FieldValidationError: If *field_name* is not a recognised header.
        """
        validator = _HEADER_FIELD_VALIDATORS.get(field_name)
        if validator is None:
            raise FieldValidationError(
                field=field_name,
                message=f"Unknown header field: {field_name}",
                constraint="must be one of the 11 core header fields",
            )
        return validator(value)

    @traces(
        spec="004",
        story="US-002",
        reqs=[
            "REQ-004-US-002-013",
            "REQ-004-US-002-015",
            "REQ-004-US-002-016",
            "REQ-004-US-002-017",
        ],
    )
    @classmethod
    def validate_all_headers(cls, packet: CXPPacket) -> list[str]:
        """Validate every core header field on *packet*.

        Returns:
            A list of human-readable error strings.  An empty list means
            the packet headers are fully valid.
        """
        errors: list[str] = []
        headers = cls.get_headers(packet)

        for field_name in CORE_HEADER_FIELDS:
            value = headers[field_name]
            try:
                if not cls.validate_header(field_name, value):
                    errors.append(f"{field_name}: invalid value {value!r}")
            except FieldValidationError as exc:
                errors.append(f"{field_name}: {exc.message}")

        return errors

    # -- Mutation (mutable fields only) ----------------------------------------

    @traces(
        spec="004",
        story="US-002",
        reqs=[
            "REQ-004-US-002-021",
            "REQ-004-US-009-009",
            "REQ-004-US-009-010",
            "REQ-004-US-009-011",
        ],
    )
    @classmethod
    def set_priority(cls, packet: CXPPacket, priority: int) -> None:
        """Set the priority on *packet* after range validation.

        Args:
            packet:   Target CXP packet.
            priority: Integer in [PRIORITY_MIN, PRIORITY_MAX].

        Raises:
            FieldValidationError: If *priority* is out of range.
        """
        if not isinstance(priority, int) or not (
            PRIORITY_MIN <= priority <= PRIORITY_MAX
        ):
            raise FieldValidationError(
                field="priority",
                message=f"Priority must be an integer in [{PRIORITY_MIN}, {PRIORITY_MAX}], got {priority!r}",
                constraint=f"{PRIORITY_MIN}-{PRIORITY_MAX}",
            )
        packet.priority = priority

    @traces(
        spec="004",
        story="US-002",
        reqs=[
            "REQ-004-US-002-019",
            "REQ-004-US-002-022",
        ],
    )
    @classmethod
    def set_delivery_status(cls, packet: CXPPacket, status: DeliveryStatus) -> None:
        """Transition the delivery status with state-machine validation.

        Args:
            packet: Target CXP packet.
            status: New delivery status.

        Raises:
            StateTransitionError: If the transition is not permitted.
        """
        if not isinstance(status, DeliveryStatus):
            try:
                status = DeliveryStatus(status)
            except ValueError:
                raise FieldValidationError(
                    field="delivery_status",
                    message=f"Invalid delivery status: {status!r}",
                    constraint="must be a valid DeliveryStatus value",
                )

        current = packet.delivery_status
        current_value = (
            current.value if isinstance(current, DeliveryStatus) else current
        )
        target_value = status.value

        allowed = DELIVERY_TRANSITIONS.get(current_value, frozenset())
        if target_value not in allowed:
            raise StateTransitionError(
                current_state=current_value,
                target_state=target_value,
                state_type="delivery",
            )

        packet.delivery_status = status

    @traces(
        spec="004",
        story="US-002",
        reqs=["REQ-004-US-002-020", "REQ-004-US-002-022"],
    )
    @classmethod
    def set_work_status(cls, packet: CXPPacket, status: WorkStatus) -> None:
        """Transition the work status with state-machine validation.

        Args:
            packet: Target CXP packet.
            status: New work status.

        Raises:
            StateTransitionError: If the transition is not permitted.
        """
        if not isinstance(status, WorkStatus):
            try:
                status = WorkStatus(status)
            except ValueError:
                raise FieldValidationError(
                    field="work_status",
                    message=f"Invalid work status: {status!r}",
                    constraint="must be a valid WorkStatus value",
                )

        current = packet.work_status
        current_value = current.value if isinstance(current, WorkStatus) else current
        target_value = status.value

        allowed = WORK_TRANSITIONS.get(current_value, frozenset())
        if target_value not in allowed:
            raise StateTransitionError(
                current_state=current_value,
                target_state=target_value,
                state_type="work",
            )

        packet.work_status = status