orchestrata_ultimate.postman.envelope

Reference › orchestrata_ultimate › postman › envelope

Creation and validation of the message envelope, the routing metadata that travels with every message through the broker (SPEC-002 story FR-001, REQ-002-FR1-007 to -019). The frozen MessageEnvelope dataclass is defined in types; EnvelopeHandler implements IMessageEnvelope from interfaces. Its importer is delivery, which validates every envelope before queuing it. Limits come from constants, and errors are InvalidEnvelopeError from exceptions. The body of a message is handled separately by payload.

create builds an envelope and rejects bad input with InvalidEnvelopeError: the sender and recipient must be non-empty, the sender must match the agent pattern (U: followed by a capital and lower-case letters), the content type must be one of the four supported types, the priority must be 0 to 4, and a time-to-live must be positive and no more than 24 hours (the default is one hour). The new envelope is then run through validate. validate applies six checks in order and returns the first error: message id is a valid UUID, sender, recipient (an agent id or a lower-case topic name), time-to-live range, timezone-aware creation time, and valid priority and message-type enums. serialize and deserialize convert to and from bytes, is_expired compares the creation time plus time-to-live with now, get_routing_key returns the recipient, extend_ttl returns a new envelope with more time, and validate_content_type checks the content type.

At a glance

Class diagram

Dependency diagram

API

Classes

EnvelopeHandler

Bases: IMessageEnvelope · Decorators: traces(spec='002', story='FR-001', reqs=['REQ-002-FR1-019', 'REQ-002-FR1-010', 'REQ-002-FR1-016', 'REQ-002-FR1-015', 'R… · Line 49

Handler for message envelope operations.

Source

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

"""
Message Envelope Handler Implementation.

Implements the IMessageEnvelope interface for creating, validating,
serializing, and deserializing message envelopes.

Author: Mohamed (Stream A, Subagent 1)
"""

import json
import logging
import re
import uuid
from datetime import UTC, datetime, timedelta

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.postman.constants import (
    CONTENT_TYPE_BINARY,
    CONTENT_TYPE_COMPRESSED,
    CONTENT_TYPE_JSON,
    CONTENT_TYPE_TEXT,
    DEFAULT_MESSAGE_TTL_SECONDS,
    MAX_ENVELOPE_TTL_SECONDS,
)
from orchestrata_ultimate.postman.exceptions import (
    InvalidEnvelopeError,
    InvalidPayloadError,
)
from orchestrata_ultimate.postman.interfaces import IMessageEnvelope
from orchestrata_ultimate.postman.types import MessageEnvelope, MessageType, Priority

# Spec: 002 - Postman Message Broker


logger = logging.getLogger(__name__)


@traces(
    spec="002",
    story="FR-001",
    reqs=[
        "REQ-002-FR1-019",
        "REQ-002-FR1-010",
        "REQ-002-FR1-016",
        "REQ-002-FR1-015",
        "REQ-002-FR1-017",
    ],
)
class EnvelopeHandler(IMessageEnvelope):
    """Handler for message envelope operations."""

    @traces(
        spec="002",
        story="FR-001",
        reqs=[
            "REQ-002-FR1-019",
            "REQ-002-FR1-010",
            "REQ-002-FR1-012",
            "REQ-002-FR1-014",
            "REQ-002-FR1-007",
        ],
    )
    def create(
        self,
        sender: str,
        recipient: str,
        content_type: str = CONTENT_TYPE_JSON,
        priority: int = 2,
        ttl_seconds: int | None = None,
        correlation_id: str | None = None,
        reply_to: str | None = None,
    ) -> MessageEnvelope:
        """Create a new message envelope.

        Args:
            sender: Sender agent ID (must match pattern "U:[A-Za-z]+")
            recipient: Recipient agent ID or topic
            content_type: MIME type of payload (default: application/json)
            priority: Priority level (0=CRITICAL to 4=BACKGROUND)
            ttl_seconds: Time-to-live in seconds (default: 3600)
            correlation_id: For request-response correlation
            reply_to: Reply destination

        Returns:
            New MessageEnvelope instance

        Raises:
            InvalidEnvelopeError: If validation fails during creation
        """
        # Validate inputs
        if not sender or not sender.strip():
            raise InvalidEnvelopeError("Sender cannot be empty")

        if not recipient or not recipient.strip():
            raise InvalidEnvelopeError("Recipient cannot be empty")

        # Validate sender pattern: U:[A-Z][a-z]+
        agent_pattern = re.compile(r"^U:[A-Z][a-z]+$")
        if not agent_pattern.match(sender):
            raise InvalidEnvelopeError(
                f"Sender '{sender}' must match pattern U:[A-Z][a-z]+ (e.g., U:Ox, U:PS)"
            )

        # Validate content type
        if not self.validate_content_type(content_type):
            raise InvalidEnvelopeError(
                f"Unsupported content type: {content_type}. "
                f"Supported types: {CONTENT_TYPE_JSON}, {CONTENT_TYPE_TEXT}, "
                f"{CONTENT_TYPE_BINARY}, {CONTENT_TYPE_COMPRESSED}"
            )

        # Validate priority
        try:
            priority_enum = Priority(priority)
        except ValueError:
            raise InvalidEnvelopeError(
                f"Invalid priority value: {priority}. Must be 0-4"
            )

        # Validate TTL
        if ttl_seconds is not None:
            if ttl_seconds <= 0:
                raise InvalidEnvelopeError(f"TTL must be positive, got {ttl_seconds}")
            if ttl_seconds > MAX_ENVELOPE_TTL_SECONDS:  # 24 hours max
                raise InvalidEnvelopeError(
                    f"TTL cannot exceed {MAX_ENVELOPE_TTL_SECONDS} seconds (24 hours), got {ttl_seconds}"
                )

        # Default TTL
        actual_ttl = (
            ttl_seconds if ttl_seconds is not None else DEFAULT_MESSAGE_TTL_SECONDS
        )

        # Create envelope
        # MessageEnvelope fields: to, from_, thread_id, subject, message_id,
        # priority, created_at, ttl_seconds, correlation_id, reply_to, message_type
        # Note: content_type belongs to MessagePayload, not MessageEnvelope — and
        # always has (MessageEnvelope has never carried a content_type field; see
        # its twin-compatibility mapping docstring, ADR-0038). correlation_id DOES
        # belong here (REQ-002-FR1-007) and is threaded through below — a
        # first-time addition to MessageEnvelope, not a restored one (see the
        # same docstring for why "restored" would be the wrong word).
        envelope = MessageEnvelope(
            from_=sender,
            to=recipient,
            priority=priority_enum,
            ttl_seconds=actual_ttl,
            correlation_id=correlation_id,
            reply_to=reply_to,
            # message_id auto-generated by dataclass default_factory
            # created_at auto-set by dataclass default_factory
            # message_type defaults to INFORM
        )

        # Validate the created envelope
        is_valid, error_msg = self.validate(envelope)
        if not is_valid:
            raise InvalidEnvelopeError(
                f"Created envelope failed validation: {error_msg}"
            )

        return envelope

    @traces(
        spec="002",
        story="FR-001",
        reqs=[
            "REQ-002-FR1-010",
            "REQ-002-FR1-011",
            "REQ-002-FR1-012",
            "REQ-002-FR1-014",
        ],
    )
    def validate(self, envelope: MessageEnvelope) -> tuple[bool, str | None]:
        """Validate envelope fields.

        Args:
            envelope: Envelope to validate

        Returns:
            Tuple of (is_valid, error_message)
        """
        for check in (
            self._validate_message_id,
            self._validate_sender,
            self._validate_recipient,
            self._validate_ttl,
            self._validate_timestamp,
            self._validate_enums,
        ):
            error = check(envelope)
            if error is not None:
                return False, error

        return True, None

    @staticmethod
    @traces(spec="002", story="FR-001", reqs=["REQ-002-FR1-011"])
    def _validate_message_id(envelope: MessageEnvelope) -> str | None:
        """Return an error if message_id is not a valid UUID, else None."""
        try:
            uuid.UUID(envelope.message_id)
        except ValueError:
            return f"Invalid message_id format: {envelope.message_id}"
        return None

    @staticmethod
    def _validate_sender(envelope: MessageEnvelope) -> str | None:
        """Return an error if from_ is empty or malformed, else None."""
        if not envelope.from_ or not envelope.from_.strip():
            return "Sender cannot be empty"
        if not re.match(r"^U:[A-Z][a-z]+$", envelope.from_):
            return f"Sender must match pattern U:[A-Z][a-z]+: {envelope.from_}"
        return None

    @staticmethod
    def _validate_recipient(envelope: MessageEnvelope) -> str | None:
        """Return an error if recipient is empty or not an agent/topic, else None."""
        if not envelope.to or not envelope.to.strip():
            return "Recipient cannot be empty"
        recipient = envelope.to.strip()
        # Agent pattern: U:[A-Z][a-z]+; topic pattern: lowercase/digits/._-
        agent_pattern = re.compile(r"^U:[A-Z][a-z]+$")
        topic_pattern = re.compile(r"^[a-z0-9._-]+$")
        if not (agent_pattern.match(recipient) or topic_pattern.match(recipient)):
            return f"Recipient must be either agent ID (U:Xx) or topic pattern: {recipient}"
        return None

    @staticmethod
    def _validate_ttl(envelope: MessageEnvelope) -> str | None:
        """Return an error if TTL is non-positive or over the maximum, else None."""
        if envelope.ttl_seconds <= 0:
            return f"TTL must be positive: {envelope.ttl_seconds}"
        if envelope.ttl_seconds > MAX_ENVELOPE_TTL_SECONDS:
            return (
                f"TTL cannot exceed {MAX_ENVELOPE_TTL_SECONDS} seconds: "
                f"{envelope.ttl_seconds}"
            )
        return None

    @staticmethod
    def _validate_timestamp(envelope: MessageEnvelope) -> str | None:
        """Return an error if created_at is naive (no tzinfo), else None."""
        if envelope.created_at.tzinfo is None:
            return "Timestamp must have timezone info"
        return None

    @staticmethod
    def _validate_enums(envelope: MessageEnvelope) -> str | None:
        """Return an error if priority or message_type is invalid, else None."""
        try:
            Priority(envelope.priority)
        except ValueError:
            return f"Invalid priority value: {envelope.priority}"
        try:
            MessageType(envelope.message_type)
        except ValueError:
            return f"Invalid message_type: {envelope.message_type}"
        return None

    @traces(spec="002", story="FR-001", reqs=["REQ-002-FR1-016"])
    def serialize(self, envelope: MessageEnvelope) -> bytes:
        """Serialize envelope to bytes.

        Args:
            envelope: Envelope to serialize

        Returns:
            Serialized bytes

        Raises:
            InvalidPayloadError: If serialization fails
        """
        try:
            # Use envelope's to_dict method
            envelope_dict = envelope.to_dict()

            # Convert to JSON bytes
            json_str = json.dumps(envelope_dict, separators=(",", ":"))
            return json_str.encode("utf-8")

        except (TypeError, ValueError, AttributeError) as e:
            raise InvalidPayloadError(f"Failed to serialize envelope: {e}") from e

    @traces(spec="002", story="FR-001", reqs=["REQ-002-FR1-015"])
    def deserialize(self, data: bytes) -> MessageEnvelope:
        """Deserialize envelope from bytes.

        Args:
            data: Serialized envelope data

        Returns:
            Deserialized MessageEnvelope

        Raises:
            InvalidPayloadError: If deserialization fails
            ValueError: If data is malformed
        """
        try:
            # Decode JSON
            json_str = data.decode("utf-8")
            envelope_dict = json.loads(json_str)

            # Use MessageEnvelope.from_dict which handles field mapping correctly
            # (maps "from" key -> from_ field, "created_at" string -> datetime, etc.)
            return MessageEnvelope.from_dict(envelope_dict)

        except (
            json.JSONDecodeError,
            UnicodeDecodeError,
            KeyError,
            ValueError,
            TypeError,
        ) as e:
            raise InvalidPayloadError(
                f"Failed to deserialize envelope data: {e}"
            ) from e

    def is_expired(self, envelope: MessageEnvelope) -> bool:
        """Check if message envelope has expired based on TTL.

        Args:
            envelope: Envelope to check for expiration

        Returns:
            True if expired, False otherwise
        """
        # Calculate expiry time: created_at + TTL
        expiry_time = envelope.created_at + timedelta(seconds=envelope.ttl_seconds)

        # Compare with current UTC time
        current_time = datetime.now(UTC)

        return current_time > expiry_time

    @traces(spec="002", story="FR-001", reqs=["REQ-002-FR1-017"])
    def get_routing_key(self, envelope: MessageEnvelope) -> str:
        """Extract routing key from envelope.

        The routing key is used to determine where the message is routed.
        For direct agent-to-agent messages, returns the recipient agent ID.
        For topic-based messages, returns the topic name.

        Args:
            envelope: Envelope to extract routing key from

        Returns:
            Routing key (recipient or topic)

        Raises:
            InvalidEnvelopeError: If recipient is empty or invalid
        """
        if not envelope.to or not envelope.to.strip():
            raise InvalidEnvelopeError("Cannot extract routing key: recipient is empty")

        # For topic-based routing (e.g., "events.user.created")
        # For agent-based routing (e.g., "U:Ox", "U:PS")
        # Both are valid routing keys - return to field as-is
        return envelope.to.strip()

    @traces(spec="002", story="US-002", reqs=["REQ-002-US-002-016"])
    def extend_ttl(self, envelope: MessageEnvelope, seconds: int) -> MessageEnvelope:
        """Create new envelope with extended TTL.

        Creates a copy of the envelope with TTL extended by the specified seconds.
        Does not modify the original envelope (immutable pattern).

        Args:
            envelope: Original envelope
            seconds: Number of seconds to extend TTL by (must be > 0)

        Returns:
            New MessageEnvelope with extended TTL

        Raises:
            InvalidEnvelopeError: If seconds <= 0 or new TTL exceeds maximum
        """
        if seconds <= 0:
            raise InvalidEnvelopeError(f"TTL extension must be positive, got {seconds}")

        new_ttl = envelope.ttl_seconds + seconds

        # Check maximum TTL (24 hours)
        if new_ttl > MAX_ENVELOPE_TTL_SECONDS:
            raise InvalidEnvelopeError(
                f"Extended TTL ({new_ttl}) would exceed maximum ({MAX_ENVELOPE_TTL_SECONDS} seconds)"
            )

        # Create new envelope with extended TTL
        # Use the dataclass replace pattern
        from dataclasses import replace

        return replace(envelope, ttl_seconds=new_ttl)

    @traces(spec="002", story="FR-001", reqs=["REQ-002-FR1-012"])
    def validate_content_type(self, content_type: str) -> bool:
        """Validate that content type is supported.

        Args:
            content_type: MIME type to validate

        Returns:
            True if supported, False otherwise
        """
        supported_types = {
            CONTENT_TYPE_JSON,
            CONTENT_TYPE_TEXT,
            CONTENT_TYPE_BINARY,
            CONTENT_TYPE_COMPRESSED,
        }
        return content_type in supported_types