orchestrata_ultimate.cxp.servicer

Reference › orchestrata_ultimate › cxp › servicer

servicer is the transport-neutral service core of CXP: CXPServicer implements the CXP operations once, and direct_cxp_adapter exposes them over the cxpt transport. The agent runtime, the kernel CLI and the knowledge-graph server also import it. See cxp.

• CXPServerConfig and CXPStats hold the server settings (host, port from endpoint_config, worker count) and counters of packets sent and received.
• send_packet converts the protobuf request to a dictionary with _proto_to_dict, hands it to the configured packet handler (which may be sync or async), publishes any reply_packet in the result through enqueue_reply, and returns a CXPReceipt. An exception becomes a receipt with success=False and the error text, not a raised error. send_packet_stream does the same for a stream of packets.
• enqueue_reply is the single reply publication point and passes the packet to a reply_router. receive_packet and subscribe stream from the caller's own queue, keyed by agent id, using a drain lease so the queue is not evicted while in use, and stop when the requested timeout elapses; _matches_subscription filters by subject and sender.
• get_status asks the packet handler for the volk roster with a status subject and returns a StatusReport (alias, role, status, summary, current thread, busy flag).
• health_check and acknowledge answer the remaining calls.
• run_serve_lifecycle starts a server, serves until stopped, and always calls stop in a finally, so a server dropped without stopping cannot leak sockets or threads.

The reply queue settings are read from CXP_REPLY_QUEUE_TTL_S, CXP_REPLY_QUEUE_SWEEP_S and CXP_REPLY_QUEUE_MAX_CONSUMERS. Helpers convert between status strings and the protobuf enums.

At a glance

Class diagram

Dependency diagram

API

Classes

CXPServerConfig (dataclass)

Decorators: dataclass · Line 165

Configuration for CXP gRPC server.

CXPStats (dataclass)

Decorators: dataclass · Line 175

Statistics for CXP gRPC server.

CXPServicer

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

Implementation of CXP gRPC service.

Functions

Source

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

# mypy: ignore-errors
"""Transport-neutral CXP service core.

:class:`CXPServicer` implements the CXP protocol itself -- packet handling,
receipt construction, subscriptions, acknowledgements, and the proto<->dict
conversions either side of them. It carries no transport dependency: a
transport supplies the request objects, the servicer answers them.

Carved out of the retired ``cxp/grpc_server.py`` on 2026-08-07 when
``grpc.aio`` was removed (SPEC-023: its EventEngine threads re-enable the
GIL). The carve-out is a MOVE, not a rewrite -- every servicer method already
took its ``context`` argument as an unused annotation (a ``context.`` grep
over the original returned zero hits), which is exactly why
``direct_cxp_adapter`` has been driving this same servicer over the cxpt
direct transport, in production, all along. ``context`` is typed ``Any``
here: it is whatever the calling transport passes, and nothing reads it.

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

import asyncio
import logging
import os
import time
from collections.abc import (
    AsyncGenerator,
    AsyncIterator,
    Awaitable,
    Callable,
    Mapping,
)
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 PacketType
from orchestrata_ultimate.core.tracing.tracer import Tracer
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 (
    CXPAck,
    CXPHealthRequest,
    CXPHealthResponse,
    CXPPacket,
    CXPReceipt,
    CXPReceiveRequest,
    CXPSubscription,
    StatusReport,
    StatusRequest,
    VolkStatus,
)
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.reply_router import ReplyRouter
from orchestrata_ultimate.cxp.transport import CXPServerLike
from orchestrata_ultimate.cxp.work_status import WorkStatus

logger = logging.getLogger(__name__)


def _env_float(
    name: str, default: float, environ: Mapping[str, str] | None = None
) -> float:
    """Resolve a float policy tunable from the environment at call time.

    Mirrors the house pattern in ``agents/kernel_handler.py``'s
    ``shutdown()`` (``KERNEL_BG_DRAIN_GRACE_S``): this is a tunable, not an
    infra endpoint, so a literal *default* is permitted (RUL-003's
    inverse); an empty or unparseable value falls back to *default* with a
    traced warning instead of raising -- a misconfigured env must not
    crash server construction. ``environ=`` is the injection seam
    (``gateway/config.py`` form) so tests exercise the fallback without
    mutating process-global state.
    """
    env = os.environ if environ is None else environ
    raw = (env.get(name, str(default)) or str(default)).strip()
    try:
        return float(raw)
    except ValueError:
        logger.warning("invalid %s=%r; falling back to %s", name, raw, default)
        return default


def _env_int(name: str, default: int, environ: Mapping[str, str] | None = None) -> int:
    """Integer counterpart to :func:`_env_float` -- same fallback contract."""
    env = os.environ if environ is None else environ
    raw = (env.get(name, str(default)) or str(default)).strip()
    try:
        return int(raw)
    except ValueError:
        logger.warning("invalid %s=%r; falling back to %s", name, raw, default)
        return default


def _enum_token(value: Any) -> str:
    """Extract the bare enum token from a string or enum value.

    Accepts ``"delivered"``, ``DeliveryStatus.DELIVERED``, or a raw proto int's
    string form and returns an uppercase token (e.g. ``"DELIVERED"``). Empty or
    ``None`` values yield ``""`` so callers can apply a default.
    """
    if value is None:
        return ""
    if hasattr(value, "value") and not isinstance(value, (str, int)):
        value = value.value
    return str(value).strip().upper()


def _delivery_str_to_proto(value: Any) -> DeliveryStatusProto:
    """Inverse of _proto_to_dict's delivery_status decode.

    Falls back to ``DELIVERY_STATUS_QUEUED`` when *value* is absent/empty or
    does not name a known enum member.
    """
    token = _enum_token(value)
    if not token:
        return DeliveryStatusProto.DELIVERY_STATUS_QUEUED
    name = f"DELIVERY_STATUS_{token}"
    if name in DeliveryStatusProto.keys():
        # EnumTypeWrapper.Value() is typed -> int in protobuf's stubs.
        return cast(DeliveryStatusProto, DeliveryStatusProto.Value(name))
    return DeliveryStatusProto.DELIVERY_STATUS_QUEUED


def _work_str_to_proto(value: Any) -> WorkStatusProto:
    """Inverse of _proto_to_dict's work_status decode.

    Falls back to ``WORK_STATUS_PENDING`` when *value* is absent/empty or does
    not name a known enum member.
    """
    token = _enum_token(value)
    if not token:
        return WorkStatusProto.WORK_STATUS_PENDING
    name = f"WORK_STATUS_{token}"
    if name in WorkStatusProto.keys():
        # EnumTypeWrapper.Value() is typed -> int in protobuf's stubs.
        return cast(WorkStatusProto, WorkStatusProto.Value(name))
    return WorkStatusProto.WORK_STATUS_PENDING


def _packet_type_str_to_proto(value: Any) -> PacketTypeProto:
    """Inverse of _proto_to_dict's packet_type decode.

    PacketType.ACKNOWLEDGEMENT has value ``"ack"`` but its proto enum is
    ``PACKET_TYPE_ACKNOWLEDGEMENT``; we resolve via the enum *name* when given a
    ``PacketType`` so that case round-trips. Falls back to
    ``PACKET_TYPE_MESSAGE`` when absent/empty or unknown.
    """
    if isinstance(value, PacketType):
        token = value.name.upper()
    else:
        token = _enum_token(value)
    if not token:
        return PacketTypeProto.PACKET_TYPE_MESSAGE
    name = f"PACKET_TYPE_{token}"
    if name in PacketTypeProto.keys():
        # EnumTypeWrapper.Value() is typed -> int in protobuf's stubs.
        return cast(PacketTypeProto, PacketTypeProto.Value(name))
    return PacketTypeProto.PACKET_TYPE_MESSAGE


@dataclass
class CXPServerConfig:
    """Configuration for CXP gRPC server."""

    host: str = "0.0.0.0"
    port: int = field(default_factory=cxp_grpc_port)
    max_workers: int = 10
    enable_reflection: bool = True


@dataclass
class CXPStats:
    """Statistics for CXP gRPC server."""

    packets_sent: int = 0
    packets_received: int = 0
    start_time: float = field(default_factory=time.time)


@traces(spec="001", story="US-002", reqs=["REQ-001-US-002-012"])
class CXPServicer:
    """Implementation of CXP gRPC service."""

    def __init__(
        self,
        packet_handler: (
            Callable[[dict[str, Any]], dict[str, Any] | Awaitable[dict[str, Any]]]
            | None
        ) = None,
        reply_router: ReplyRouter | None = None,
    ) -> None:
        self._packet_handler = packet_handler
        self._stats = CXPStats()
        self._acks: dict[str, asyncio.Event] = {}
        # Per-consumer reply routing (closes G3/G4/G7 -- keyed queues
        # replace the single unkeyed queue every consumer used to compete
        # for). Injectable for tests; defaults to a process Tracer-backed
        # instance when not composed explicitly. F7 hardening (D-H10):
        # the eviction tunables are policy, not infra endpoints, so a
        # literal default is permitted (RUL-003's inverse) -- resolved at
        # CALL time (not import time) so deploy tooling and the acceptance
        # probe can override them without a code change.
        self._reply_router: ReplyRouter = reply_router or ReplyRouter(
            Tracer.default(),
            idle_ttl_s=_env_float("CXP_REPLY_QUEUE_TTL_S", 900.0),
            sweep_interval_s=_env_float("CXP_REPLY_QUEUE_SWEEP_S", 60.0),
            max_consumers=_env_int("CXP_REPLY_QUEUE_MAX_CONSUMERS", 1024),
        )

    @property
    def stats(self) -> CXPStats:
        return self._stats

    @traces(
        spec="001", story="US-002", reqs=["REQ-001-US-002-019", "REQ-001-US-002-038"]
    )
    def enqueue_reply(self, packet: dict[str, Any]) -> None:
        """Publish a reply packet for streaming consumers (receive_packet/subscribe).

        The single reply-publication point: every reply -- including
        ``send_packet``'s inline ``reply_packet`` branch below -- routes
        through here, which delegates to the composed :class:`ReplyRouter`
        so the packet lands on its addressee's own queue instead of one
        shared, unkeyed queue every consumer used to compete for (closes
        G3/G4; see ``plans/kernel-dispatch-reply-design.md`` §4.1 D6/D7).
        """
        self._reply_router.route(packet)

    async def send_packet(self, request: CXPPacket, context: Any) -> CXPReceipt:
        """Handle incoming CXP packet."""
        self._stats.packets_received += 1

        try:
            packet_dict = self._proto_to_dict(request)

            if self._packet_handler:
                handler_result = self._packet_handler(packet_dict)
                if asyncio.iscoroutine(handler_result):
                    result = await handler_result
                else:
                    result = cast("dict[str, Any]", handler_result)
            else:
                result: dict[str, Any] = {"success": True}

            # The unary CXPReceipt only carries an ack. If the handler produced a
            # reply packet, enqueue it so streaming consumers (receive_packet /
            # subscribe) deliver the volk's reply out-of-band.
            if isinstance(result, dict) and result.get("reply_packet"):
                self.enqueue_reply(result["reply_packet"])

            self._stats.packets_sent += 1

            from google.protobuf.timestamp_pb2 import Timestamp

            received_at = Timestamp()
            received_at.GetCurrentTime()

            return CXPReceipt(
                message_id=request.message_id,
                success=result.get("success", True),
                error_message=(
                    str(result.get("error", "")) if result.get("error") else ""
                ),
                received_at=received_at,
                delivery_latency_ms=0,
            )

        except Exception as e:
            logger.error(f"Error handling packet: {e}")
            return CXPReceipt(
                message_id=request.message_id,
                success=False,
                error_message=str(e),
            )

    async def send_packet_stream(
        self,
        request_iterator: AsyncIterator[CXPPacket],
        context: Any,
    ) -> CXPReceipt:
        """Handle streaming CXP packets."""
        count = 0
        async for packet in request_iterator:
            result = await self.send_packet(packet, context)
            if result.success:
                count += 1

        return CXPReceipt(
            message_id="stream",
            success=True,
            error_message="",
        )

    async def health_check(
        self, request: CXPHealthRequest, context: Any
    ) -> CXPHealthResponse:
        """Handle health check."""
        return CXPHealthResponse(
            healthy=True,
            status="Serving",
            uptime_seconds=int(time.time() - self._stats.start_time),
            packets_sent=self._stats.packets_sent,
            packets_received=self._stats.packets_received,
        )

    async def get_status(self, request: StatusRequest, context: Any) -> StatusReport:
        """Return the live volk roster (synchronous status query).

        Reuses the ``status`` CXP subject the kernel handler already answers, so
        the gateway/orchestrator can read every volk's status on demand without
        correlating an asynchronous stream reply.
        """
        self._stats.packets_received += 1
        volks: list[dict[str, Any]] = []
        if self._packet_handler is not None:
            status_packet = {
                "message_id": "",
                "from_agent": request.requester or "GATEWAY",
                "to_agent": "SYSTEM",
                "subject": "status",
                "thread_id": "",
                "payload": {},
            }
            result = self._packet_handler(status_packet)
            if asyncio.iscoroutine(result):
                result = await result
            if isinstance(result, dict):
                payload = result.get("reply_payload") or {}
                if isinstance(payload, dict):
                    volks = payload.get("volks") or []
        return StatusReport(
            volks=[
                VolkStatus(
                    alias=str(v.get("alias", "")),
                    role=str(v.get("role", "")),
                    status=str(v.get("status", "")),
                    summary=str(v.get("summary", "")),
                    current_thread=str(v.get("current_thread", "") or ""),
                    busy=bool(v.get("busy", False)),
                )
                for v in volks
            ]
        )

    async def receive_packet(
        self, request: CXPReceiveRequest, context: Any
    ) -> AsyncGenerator[CXPPacket]:
        """Stream packets to client, from THIS consumer's own reply queue.

        Keyed by ``request.agent_id`` via the composed ``ReplyRouter``
        (closes G3: every consumer used to compete for one shared queue,
        measured at 50/50 reply loss across concurrent gateways). Wrapped
        in an RAII drain lease (F7 hardening, D-H9): while this generator
        is live, the router's idle-TTL sweep (D-H10) can never evict this
        consumer's queue out from under it, whatever else is going idle.
        """
        queue = self._reply_router.open_drain(request.agent_id)
        try:
            while True:
                try:
                    packet = await asyncio.wait_for(
                        queue.get(), timeout=request.timeout_seconds
                    )
                    yield self._dict_to_proto(packet)
                except TimeoutError:
                    break
        finally:
            self._reply_router.close_drain(request.agent_id)

    async def subscribe(
        self, request: CXPSubscription, context: Any
    ) -> AsyncGenerator[CXPPacket]:
        """Subscribe to packets matching criteria, from THIS consumer's own queue.

        Keyed by ``request.agent_id`` via the composed ``ReplyRouter``
        (closes G4: ``subscribe`` used to drain the single shared queue and
        silently discard -- not requeue -- any packet failing its filter,
        destroying replies owned by other consumers). A packet from this
        consumer's OWN queue that fails the filter is still dropped, not
        requeued -- it was addressed here, so today's semantics for that
        case are unchanged. Wrapped in an RAII drain lease (F7 hardening,
        D-H9): while this generator is live, the router's idle-TTL sweep
        (D-H10) can never evict this consumer's queue out from under it.
        """
        queue = self._reply_router.open_drain(request.agent_id)
        try:
            while True:
                try:
                    packet = await asyncio.wait_for(queue.get(), timeout=60)
                    if self._matches_subscription(packet, request):
                        yield self._dict_to_proto(packet)
                except TimeoutError:
                    break
        finally:
            self._reply_router.close_drain(request.agent_id)

    async def acknowledge(self, request: CXPAck, context: Any) -> CXPReceipt:
        """Handle packet acknowledgment."""
        message_id = request.message_id
        if message_id in self._acks:
            self._acks[message_id].set()

        return CXPReceipt(
            message_id=message_id,
            success=True,
        )

    def _matches_subscription(
        self, packet: dict[str, Any], subscription: CXPSubscription
    ) -> bool:
        """Check if packet matches subscription filters."""
        if subscription.subjects and packet.get("subject") not in subscription.subjects:
            return False
        if (
            subscription.from_agents
            and packet.get("from_agent") not in subscription.from_agents
        ):
            return False
        return True

    def _proto_to_dict(self, proto: CXPPacket) -> dict[str, Any]:
        """Convert protobuf to dictionary."""
        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),
            "ttl_seconds": proto.ttl_seconds,
        }

    def _dict_to_proto(self, data: dict[str, Any]) -> CXPPacket:
        """Convert dictionary to protobuf."""
        from google.protobuf.struct_pb2 import Struct

        packet = CXPPacket()
        packet.message_id = data.get("message_id", "")
        packet.from_agent = data.get("from_agent", "")
        packet.to_agent = data.get("to_agent", "")
        packet.thread_id = data.get("thread_id", "")
        packet.subject = data.get("subject", "")

        if data.get("payload"):
            payload = Struct()
            payload.update(data["payload"])
            packet.payload = payload

        if data.get("timestamp"):
            ts_value = data["timestamp"]
            # _proto_to_dict emits an ISO-8601 string (see its "timestamp" key
            # below); parse it back the same way _packet_results.py does so a
            # dict produced by this class's own encoder round-trips. Mutate the
            # field's own sub-message via FromDatetime(), the same pattern
            # proto_codec.py's _packet_to_proto uses — assigning a standalone
            # Timestamp() instance to packet.timestamp instead (the previous
            # code) hits protobuf's well-known-type descriptor magic, which
            # treats the RHS as a raw datetime and raises AttributeError
            # regardless of whether it started life as a string.
            if isinstance(ts_value, str):
                ts_value = datetime.fromisoformat(ts_value.replace("Z", "+00:00"))
            packet.timestamp.FromDatetime(ts_value)

        packet.priority = data.get("priority", 5)
        packet.protocol_version = data.get("protocol_version", "1.0.0")
        packet.session_id = data.get("session_id", "")
        packet.ttl_seconds = data.get("ttl_seconds", 300)

        if data.get("metadata"):
            packet.metadata.update(data["metadata"])

        # Honor the enum values carried in *data* (the inverse of _proto_to_dict's
        # decode), falling back to CXPPacket model defaults only when absent/empty.
        packet.delivery_status = _delivery_str_to_proto(data.get("delivery_status"))
        packet.work_status = _work_str_to_proto(data.get("work_status"))
        packet.packet_type = _packet_type_str_to_proto(data.get("packet_type"))

        return packet


async def run_serve_lifecycle(server: CXPServerLike, *, label: str, port: int) -> None:
    """Run ``server.start()`` -> ``server.serve_forever()`` with a guaranteed stop.

    Shared by every ``_serve()`` entry point that owns a CXP server
    (``knowledge_graph/server.py``, ``agents/runtime_main.py``). ``stop()``
    ALWAYS runs before this coroutine returns or raises: on the graceful
    signal-driven path (a signal handler's fire-and-forget
    ``asyncio.ensure_future(server.stop())`` task can still be mid-flight
    when ``serve_forever()`` resolves) and on any exception path (e.g. a
    ``KeyboardInterrupt`` racing the caller's prelude, before any signal
    handler fires).

    The guarantee is transport-neutral, and it is not decoration. Under the
    retired ``grpc.aio`` transport an abandoned half-stopped server would
    later schedule a coroutine on a closed loop from ``__del__``, corrupt
    grpc's engine bookkeeping, and hang the process forever at interpreter
    exit on threads that never join. The cxpt transport has its own
    version of the same hazard: a server dropped without ``stop()`` leaves
    its acceptor thread and accepted sockets alive until deallocation --
    the exact shape of the 2026-08-07 kernel fd leak. Whatever the
    transport, an owner that stops exporting a server must stop it.

    ``stop()`` is idempotent, so calling it again here after the graceful
    path already stopped it is a safe no-op. The teardown call is itself
    best-effort: a coroutine cancelled mid-``start()`` can leave the server
    in a transitional state where ``stop()`` raises instead of completing (a
    different case from the leak above -- it never finished starting, so it
    was never left half-stopped); swallowing that keeps it from masking the
    original exception already propagating out of the ``try``.

    Args:
        server: The (not-yet-started) :class:`CXPServer` to run.
        label: Human-readable server name for the "started" log line.
        port: The bound port, for the same log line.
    """
    try:
        await server.start()
        logger.info("%s started on 0.0.0.0:%d", label, port)
        await server.serve_forever()
    finally:
        try:
            await server.stop()
        except Exception:  # noqa: BLE001 — best-effort shutdown during teardown
            logger.debug("server.stop() during teardown failed", exc_info=True)