orchestrata_ultimate.cxp.proto_codec

Reference › orchestrata_ultimate › cxp › proto_codec

proto_codec converts between the in-memory CXP packet model and its protobuf message, and defines the client configuration. CXPClientConfig holds host, port (from endpoint_config), timeout_seconds and max_retries; the gateway server builds one to connect to the kernel. CXPPacketCodec.packet_to_proto copies the identity fields, payload (as a protobuf struct, with a non-dict payload stored as a string under data), timestamp, priority, delivery and work status (via _delivery_to_proto and _work_to_proto, using the types of delivery_status and work_status), protocol version, packet type and metadata. proto_to_packet reverses this to a dictionary. direct_cxp_adapter and transport use it. The messages come from proto. See cxp.

At a glance

Class diagram

Dependency diagram

API

Classes

CXPClientConfig (dataclass)

Decorators: traces(spec='004', story='US-004', reqs=['REQ-004-US-004-022']), dataclass · Line 43

Configuration for a CXP client (host/port/timeout/retries).

CXPPacketCodec

Line 52

Stateless CXP packet <-> protobuf conversion.

Functions

Source

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

# mypy: ignore-errors
"""CXP packet <-> protobuf codec, and the CXP client configuration.

Both were previously private members of the retired ``cxp/grpc_client.py``.
Neither ever touched ``grpc``: the wire format is protobuf (``cxp_pb2``),
which is independent of the RPC runtime that used to carry it. When
``grpc.aio`` was retired on 2026-08-07 (SPEC-023: its EventEngine threads
re-enable the GIL) they were extracted here rather than deleted, because the
cxpt direct transport encodes exactly the same packets.

``DirectCXPClient`` had been reaching into ``GRPCClient._packet_to_proto`` /
``._proto_to_packet`` by constructing an unconnected client purely to borrow
them -- a private-member dependency that only worked because
``GRPCClient.__init__`` performed no I/O. :class:`CXPPacketCodec` makes that
a real, stateless API instead.

See ``plans/grpc-retirement-design.md``.
"""

from __future__ import annotations

import logging
from dataclasses import dataclass, field
from datetime import datetime
from typing import Any, cast

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.packet import CXPPacket as CXPPacketModel
from orchestrata_ultimate.core.packet import PacketType
from orchestrata_ultimate.cxp.delivery_status import DeliveryStatus
from orchestrata_ultimate.cxp.endpoint_config import cxp_grpc_port
from orchestrata_ultimate.cxp.proto.cxp_pb2 import CXPPacket
from orchestrata_ultimate.cxp.proto.cxp_pb2 import DeliveryStatus as DeliveryStatusProto
from orchestrata_ultimate.cxp.proto.cxp_pb2 import PacketType as PacketTypeProto
from orchestrata_ultimate.cxp.proto.cxp_pb2 import WorkStatus as WorkStatusProto
from orchestrata_ultimate.cxp.work_status import WorkStatus

logger = logging.getLogger(__name__)


@traces(spec="004", story="US-004", reqs=["REQ-004-US-004-022"])
@dataclass
class CXPClientConfig:
    """Configuration for a CXP client (host/port/timeout/retries)."""

    host: str = "localhost"
    port: int = field(default_factory=cxp_grpc_port)
    timeout_seconds: int = 30
    max_retries: int = 3


class CXPPacketCodec:
    """Stateless CXP packet <-> protobuf conversion.

    Holds no connection and no configuration; every transport that speaks
    the CXP wire format encodes through this.
    """

    def packet_to_proto(self, packet: CXPPacketModel) -> CXPPacket:
        """Convert CXP packet model to protobuf."""
        from google.protobuf.struct_pb2 import Struct

        proto = CXPPacket()
        proto.message_id = packet.message_id
        proto.from_agent = packet.from_agent
        proto.to_agent = packet.to_agent
        proto.thread_id = packet.thread_id
        proto.subject = packet.subject

        if packet.payload:
            payload = Struct()
            if isinstance(packet.payload, dict):
                payload.update(packet.payload)
            else:
                payload.update({"data": str(packet.payload)})
            proto.payload = payload

        if packet.timestamp:
            # Check if it's a datetime object
            if isinstance(packet.timestamp, datetime):
                # Convert datetime to Timestamp
                proto.timestamp.FromDatetime(packet.timestamp)
            else:
                # It's already a Timestamp object
                proto.timestamp.CopyFrom(packet.timestamp)

        proto.priority = packet.priority
        proto.delivery_status = _delivery_to_proto(packet.delivery_status)
        proto.work_status = _work_to_proto(packet.work_status)
        proto.protocol_version = packet.protocol_version

        # packet_type: map model PacketType enum -> proto enum. Use the enum
        # *name* (not value) because PacketType.ACKNOWLEDGEMENT.value == "ack"
        # while the proto member is PACKET_TYPE_ACKNOWLEDGEMENT.
        # cast: protobuf's generated EnumTypeWrapper.Value() stub returns
        # plain `int` (not parametrized per concrete enum), though at
        # runtime it is the correct PacketType-typed value.
        proto.packet_type = cast(
            PacketTypeProto,
            PacketTypeProto.Value(f"PACKET_TYPE_{packet.packet_type.name.upper()}"),
        )

        # CXPPacketModel has no session_id attribute (only thread_id /
        # correlation_id), so there is nothing to copy onto proto.session_id.

        if packet.metadata:
            proto.metadata.update(packet.metadata)

        return proto

    def proto_to_packet(self, proto: CXPPacket) -> dict[str, Any]:
        """Convert protobuf to CXP packet dict."""
        return {
            "message_id": proto.message_id,
            "from_agent": proto.from_agent,
            "to_agent": proto.to_agent,
            "thread_id": proto.thread_id,
            "subject": proto.subject,
            "payload": dict(proto.payload) if proto.payload else {},
            "timestamp": (
                proto.timestamp.ToDatetime().isoformat() if proto.timestamp else ""
            ),
            "priority": proto.priority,
            "delivery_status": DeliveryStatus(
                DeliveryStatusProto.Name(proto.delivery_status)
                .lower()
                .replace("delivery_status_", "")
            ),
            "work_status": WorkStatus(
                WorkStatusProto.Name(proto.work_status)
                .lower()
                .replace("work_status_", "")
            ),
            "protocol_version": proto.protocol_version,
            "packet_type": PacketType(
                PacketTypeProto.Name(proto.packet_type)
                .lower()
                .replace("packet_type_", "")
                if proto.packet_type != PacketTypeProto.PACKET_TYPE_UNSPECIFIED
                else "message"
            ),
            "session_id": proto.session_id,
            "metadata": dict(proto.metadata),
        }


def _delivery_to_proto(status: DeliveryStatus) -> DeliveryStatusProto:
    """Convert DeliveryStatus enum to protobuf."""
    mapping = {
        DeliveryStatus.QUEUED: DeliveryStatusProto.DELIVERY_STATUS_QUEUED,
        DeliveryStatus.DELIVERED: DeliveryStatusProto.DELIVERY_STATUS_DELIVERED,
        DeliveryStatus.ACKNOWLEDGED: DeliveryStatusProto.DELIVERY_STATUS_ACKNOWLEDGED,
        DeliveryStatus.FAILED: DeliveryStatusProto.DELIVERY_STATUS_FAILED,
    }
    return mapping.get(status, DeliveryStatusProto.DELIVERY_STATUS_UNSPECIFIED)


def _work_to_proto(status: WorkStatus) -> WorkStatusProto:
    """Convert WorkStatus enum to protobuf."""
    mapping = {
        WorkStatus.PENDING: WorkStatusProto.WORK_STATUS_PENDING,
        WorkStatus.IN_PROGRESS: WorkStatusProto.WORK_STATUS_IN_PROGRESS,
        WorkStatus.RESOLVED: WorkStatusProto.WORK_STATUS_RESOLVED,
        WorkStatus.ARCHIVED: WorkStatusProto.WORK_STATUS_ARCHIVED,
    }
    return mapping.get(status, WorkStatusProto.WORK_STATUS_UNSPECIFIED)