orchestrata_ultimate.cxp.handler

Reference › orchestrata_ultimate › cxp › handler

CXPHandler is the legacy single-object packet handler, kept for backward compatibility; the current design splits its job between router and dispatcher. handle_packet reads the recipient from the packet header and returns a DeliveryResult with success=False and the error "Missing recipient" when there is none, otherwise a successful result for that recipient. route_packet simply calls handle_packet. It performs no delivery of its own. See cxp.

At a glance

Class diagram

Dependency diagram

API

Classes

CXPHandler

Line 29

Handles CXP packet processing.

Source

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

"""
CXP Handler - Processes CXP packets.

Spec: 004 - CXP Protocol

Status: Minimal stub implementation.

This module provides a thin routing layer that delegates delivery tracking to
the Postman subsystem (packages/orchestrata-postman). It is intentionally
lightweight: full protocol-specific dispatch (gRPC, WebSocket, HTTP) lives in
orchestrata.kernel.dispatcher.MessageDispatcher.

The two public methods here cover the in-process path:
- handle_packet: validates recipient presence, returns DeliveryResult
- route_packet: delegates to handle_packet (hook point for future middleware)

For production multi-protocol dispatch, use MessageDispatcher directly.
"""

import logging
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.packet import CXPPacket, DeliveryResult

logger = logging.getLogger(__name__)


class CXPHandler:
    """
    Handles CXP packet processing.

    Manages the lifecycle of packets including routing,
    delivery tracking, and response handling.
    """

    def __init__(self, config: dict[str, Any] | None = None) -> None:
        """
        Initialize CXP Handler.

        Args:
            config: Handler configuration
        """
        self._config = config or {}
        self._handlers: dict[str, Any] = {}
        logger.info("CXPHandler initialized")

    @traces(spec="001", story="US-002", reqs=["REQ-001-US-002-001"])
    async def handle_packet(self, packet: CXPPacket) -> DeliveryResult:
        """
        Handle incoming CXP packet.

        Args:
            packet: CXP packet to handle

        Returns:
            Delivery result
        """
        recipient = packet.header.recipient

        if not recipient:
            return DeliveryResult(
                success=False,
                packet_id=packet.header.packet_id,
                destination="",
                error_message="Missing recipient",
            )

        logger.debug(f"Handling packet for {recipient}")

        return DeliveryResult(
            success=True, packet_id=packet.header.packet_id, destination=recipient
        )

    @traces(spec="001", story="US-002", reqs=["REQ-001-US-002-001"])
    async def route_packet(self, packet: CXPPacket) -> DeliveryResult:
        """
        Route packet to destination.

        Args:
            packet: CXP packet to route

        Returns:
            Delivery result
        """
        return await self.handle_packet(packet)