orchestrata_ultimate.gateway.handlers.cxp

Reference › orchestrata_ultimate › gateway › handlers › cxp

CxpHandler serves the cxp.send, cxp.subscribe, cxp.unsubscribe and cxp.status RPCs, which connect UI clients to Cross-agent eXchange Protocol traffic (see the cxp package and gateway). rpc creates it; it uses the pubsub manager and the CXPPacket model in core.packet. Collaborators arrive through setters: the message processor, the kernel client, the reply consumer id and the orchestrator handler.

• handle_send accepts the target as target, to or to_agent, and raises RPCError with code -32602 if none is given. It builds a packet dictionary from the caller's connection id, subject, payload and thread id (a new UUID if none is passed), broadcasts it on cxp.traffic, records it through publish_traffic, and returns the status, thread id and payload size.
• _maybe_reply_as_concierge handles chat-style subjects sent to a concierge alias. When a kernel client is wired it sends the intent to the Concierge volk as a CXPPacket, stamped with this gateway's reply consumer id, and the real reply later arrives through the kernel reply pump in server. If no kernel is wired or the dispatch fails, _build_unavailable_reply produces an explicit "assistant unavailable" notice. The handler never invents a reply on the volk's behalf (SPEC-032).
• _record_event keeps a bounded per-thread event log, and handle_status reports the latest delivery status and events for a thread, or unknown if none were seen.
• handle_subscribe and handle_unsubscribe manage a subscription to the cxp.traffic topic, with optional filters.

At a glance

Class diagram

Dependency diagram

API

Classes

CxpHandler

Decorators: traces(spec='014', story='US-001', reqs=['REQ-014-001', 'REQ-014-008', 'REQ-014-011']), traces(spec='014', story='US-005', reqs=['REQ-014-071']) · Line 42

CXP packet sending and subscription methods.

Source

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

"""CXP RPC handlers -- send, subscribe, unsubscribe, status.

Implements SPEC-014 FR-015 CXP protocol methods.
"""

import json
import logging
import uuid
from datetime import UTC, datetime
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.gateway.exceptions import RPCError
from orchestrata_ultimate.gateway.pubsub import PubSubManager

logger = logging.getLogger(__name__)

# Chat gadgets address the concierge and tag messages with these subjects.
# A send to the concierge always gets a reply packet so the conversation
# round-trips -- either the real ConciergeVolk's answer (kernel reachable) or
# an explicit "assistant unavailable" notice (SPEC-032 G243/REQ-032-053/054);
# the gateway never fabricates a conversational-looking reply on its behalf.
_CONCIERGE_ALIASES = frozenset({"U:Cx"})
_CHAT_SUBJECTS = frozenset({"intent", "response"})

# Machine-detectable status the caller can check to tell an "assistant
# unavailable" notice apart from a genuine ConciergeVolk reply (SPEC-032 G243).
_ASSISTANT_STATUS_UNAVAILABLE = "unavailable"
_ASSISTANT_UNAVAILABLE_TEXT = (
    "The Concierge assistant isn't reachable right now (starting up or "
    "disconnected) -- this is an automated notice, not a live reply. "
    "Please try again shortly."
)

# Bounds for the in-memory per-thread traffic log backing cxp.status.
_MAX_EVENTS_PER_THREAD = 100
_MAX_TRACKED_THREADS = 1000


@traces(spec="014", story="US-001", reqs=["REQ-014-001", "REQ-014-008", "REQ-014-011"])
@traces(spec="014", story="US-005", reqs=["REQ-014-071"])
class CxpHandler:
    """CXP packet sending and subscription methods."""

    def __init__(
        self,
        pubsub: PubSubManager,
        processor: Any = None,
    ) -> None:
        """Wire handler to CXP subsystems.

        Args:
            pubsub: Pub/sub subscription manager.
            processor: Message processor for broadcast/push.
        """
        self._pubsub = pubsub
        self._processor = processor
        self._orchestrator: Any = None
        self._kernel_client: Any = None
        # P-B3 (design D6): this gateway's minted reply-consumer id, stamped
        # onto every kernel-bound chat packet's metadata.reply_to so the
        # kernel's ReplyRouter (P-B1) can key this gateway's private queue.
        # None (the default) is the back-compat path -- unstamped send,
        # consume under the legacy "HUMAN" key -- for callers that never call
        # set_reply_consumer_id (e.g. tests constructing CxpHandler directly).
        self._reply_consumer_id: str | None = None
        # Per-thread traffic log (bounded) backing cxp.status — every packet
        # published through publish_traffic is recorded here.
        self._thread_events: dict[str, list[dict[str, Any]]] = {}

    def set_orchestrator_handler(self, handler: Any) -> None:
        """Inject the orchestrator handler for concierge replies.

        Also hands the orchestrator this handler's pub/sub manager so it can
        subscribe the UI to ``pipeline.<thread_id>`` before dispatching an
        intent (closing the subscribe-after-publish race on early stage events).

        Args:
            handler: OrchestratorHandler used to turn chat intents into tasks.
        """
        self._orchestrator = handler
        if hasattr(handler, "set_pubsub"):
            handler.set_pubsub(self._pubsub)

    @traces(spec="032", story="US-032-G243", reqs=["REQ-032-053"])
    def set_kernel_client(self, client: Any) -> None:
        """Inject the CXP gRPC client used to route chat packets to the live
        volk kernel (the out-of-process agent-runtime). When set, ``cxp.send``
        dispatches to the real ConciergeVolk through it (SPEC-032 G243); when
        unset (or dispatch fails), an explicit assistant-unavailable notice is
        returned instead of a fabricated reply."""
        self._kernel_client = client

    @traces(
        spec="001",
        story="US-002",
        reqs=["REQ-001-US-002-001", "REQ-001-US-002-003"],
    )
    def set_reply_consumer_id(self, consumer_id: str) -> None:
        """Inject this gateway's minted reply-consumer id (design D6).

        Stamped onto every kernel-bound chat packet's ``metadata.reply_to`` so
        the kernel's per-consumer ``ReplyRouter`` (P-B1) routes the volk's
        reply back to THIS gateway's private queue instead of the single
        legacy ``"HUMAN"`` queue every gateway used to compete over (G3).

        Args:
            consumer_id: This gateway's minted id (``HUMAN@<hex>``).
        """
        self._reply_consumer_id = consumer_id

    def set_processor(self, processor: Any) -> None:
        """Inject the message processor after construction.

        Args:
            processor: MessageProcessor instance.
        """
        self._processor = processor

    @traces(
        spec="016",
        story="US-005",
        reqs=[
            "REQ-016-099",
            "REQ-016-100",
            "REQ-016-102",
        ],
    )
    @traces(
        spec="014",
        story="US-005",
        reqs=["REQ-014-071"],
    )
    async def handle_send(
        self,
        connection_id: str,
        params: dict[str, Any],
    ) -> dict[str, Any]:
        """Handle cxp.send -- send CXP packet (stub).

        Args:
            connection_id: Caller connection ID.
            params: CXP packet fields (target, payload).

        Returns:
            Send status with generated thread_id.
        """
        # Accept both the gateway-native "target" and the client-native "to".
        target = (
            params.get("target") or params.get("to") or params.get("to_agent") or ""
        )
        payload = params.get("payload", {})
        if not isinstance(payload, dict):
            payload = {}
        if not target:
            raise RPCError(
                "Missing 'target'/'to' parameter",
                rpc_code=-32602,
            )
        # Subject may be top-level (chat client) or nested in the payload.
        subject = params.get("subject") or payload.get("subject") or "message"
        # Honour the caller's thread_id so subscribers filtering by it (e.g. the
        # chat gadget) actually receive this packet and its reply.
        thread_id = (
            params.get("thread_id") or params.get("threadId") or str(uuid.uuid4())
        )

        sender = connection_id[:8]
        packet = {
            "from": sender,
            "to": target,
            "from_agent": sender,
            "to_agent": target,
            "thread_id": thread_id,
            "subject": subject,
            "delivery_status": "delivered",
            "payload": payload,
        }
        if self._processor:
            await self._processor.broadcast("cxp.traffic", packet)
        await self.publish_traffic(packet)

        await self._maybe_reply_as_concierge(target, subject, thread_id, payload)

        return {
            "status": "sent",
            "thread_id": thread_id,
            "target": target,
            "to": target,
            "payload_size": len(json.dumps(payload)),
        }

    @traces(
        spec="032",
        story="US-032-G243",
        reqs=["REQ-032-053", "REQ-032-054"],
    )
    @traces(
        spec="001",
        story="US-002",
        reqs=["REQ-001-US-002-001", "REQ-001-US-002-003"],
    )
    async def _maybe_reply_as_concierge(
        self,
        target: str,
        subject: str,
        thread_id: str,
        payload: dict[str, Any],
    ) -> None:
        """Route a chat intent to the real ConciergeVolk, or report unavailable.

        Only fires for chat-style subjects addressed to a concierge alias.
        When the live volk kernel is reachable, the intent is dispatched to it
        for real (send_packet is ack-only; the volk's genuine reply streams
        back separately via the gateway's reply consumer, broadcast on
        ``cxp.traffic`` with this same ``thread_id``). When the kernel is
        unwired or dispatch fails, this emits an explicit, machine-detectable
        "assistant unavailable" notice instead -- P0 fix for SPEC-032 G243:
        the gateway must never fabricate a conversational-looking reply on
        the real ConciergeVolk's behalf (e.g. during the ~3s gRPC readiness
        window at startup, before the kernel is proven reachable).

        Args:
            target: Recipient alias the client addressed.
            subject: Message subject ("intent"/"response" for chat).
            thread_id: Conversation thread id to reply on.
            payload: Original message payload (expects ``text``).
        """
        if subject not in _CHAT_SUBJECTS or target not in _CONCIERGE_ALIASES:
            return

        text = payload.get("text", "") if isinstance(payload, dict) else ""

        # When the live kernel is wired, dispatch the chat intent to the real
        # Concierge volk. send_packet is ack-only; the volk's reply returns via
        # the gateway's reply consumer, broadcast on cxp.traffic with this same
        # thread_id so the chat gadget renders it.
        if self._kernel_client is not None:
            try:
                from orchestrata_ultimate.core.packet import CXPPacket

                kernel_subject = "response" if subject == "response" else "intent"
                # P-B3 (design D6): stamp this gateway's reply-consumer id so
                # the kernel's ReplyRouter (P-B1/P-B2) addresses the reply
                # back to THIS gateway. Back-compat: when unset (no
                # set_reply_consumer_id call), send unstamped -- byte-
                # identical to pre-P-B3 output -- and the kernel/ReplyRouter
                # fall back to the legacy "HUMAN" key.
                metadata = (
                    {"reply_to": self._reply_consumer_id}
                    if self._reply_consumer_id
                    else None
                )
                receipt = await self._kernel_client.send_packet(
                    CXPPacket(
                        from_agent="HUMAN",
                        to_agent="U:Cx",
                        subject=kernel_subject,
                        payload={"text": text},
                        thread_id=thread_id,
                        metadata=metadata,
                    )
                )
                if receipt.get("success") is False:
                    logger.warning(
                        "Kernel chat dispatch reported failure; reporting "
                        "assistant-unavailable (SPEC-032 G243 -- never a "
                        "fabricated reply)"
                    )
                else:
                    return
            except Exception:
                logger.warning(
                    "Kernel chat dispatch failed; reporting assistant-unavailable "
                    "(SPEC-032 G243 -- never a fabricated reply)"
                )
        else:
            logger.info(
                "No live Concierge kernel wired; reporting assistant-unavailable "
                "(SPEC-032 G243 -- never a fabricated reply)"
            )

        reply = self._build_unavailable_reply(target, thread_id)
        if self._processor:
            await self._processor.broadcast("cxp.traffic", reply)
        await self.publish_traffic(reply)

    def _build_unavailable_reply(self, target: str, thread_id: str) -> dict[str, Any]:
        """Build an explicit "assistant unavailable" reply packet.

        Used whenever the real, live ConciergeVolk isn't reachable (unwired,
        still inside the gRPC readiness window, or dispatch failed), so the
        caller can never mistake this for a genuine assistant reply: both the
        top-level ``assistant_status`` field and the payload flag it, and the
        text itself says plainly that it is an automated notice.

        Args:
            target: Concierge alias the client addressed (echoed as sender).
            thread_id: Conversation thread id to reply on.

        Returns:
            CXP-shaped reply packet dict ready for ``publish_traffic``.
        """
        return {
            "from": target,
            "to": "HUMAN",
            "from_agent": target,
            "to_agent": "HUMAN",
            "thread_id": thread_id,
            "subject": "response",
            "work_status": _ASSISTANT_STATUS_UNAVAILABLE,
            "delivery_status": "delivered",
            "assistant_status": _ASSISTANT_STATUS_UNAVAILABLE,
            "timestamp": datetime.now(UTC).isoformat(),
            "payload": {
                "text": _ASSISTANT_UNAVAILABLE_TEXT,
                "assistant_status": _ASSISTANT_STATUS_UNAVAILABLE,
            },
        }

    @traces(spec="014", story="US-001", reqs=["REQ-014-002"])
    async def publish_traffic(self, packet: dict[str, Any]) -> None:
        """Broadcast a CXP packet to all cxp.traffic subscribers and record it in
        the per-thread traffic log (so cxp.status can report real delivery).

        Args:
            packet: CXP packet dict.
        """
        self._record_event(packet)
        await self._pubsub.publish("cxp.traffic", packet)

    def _record_event(self, packet: dict[str, Any]) -> None:
        """Append a bounded delivery event to the thread's traffic log."""
        thread_id = packet.get("thread_id") or ""
        if not thread_id:
            return
        event = {
            "from": packet.get("from") or packet.get("from_agent", ""),
            "to": packet.get("to") or packet.get("to_agent", ""),
            "subject": packet.get("subject", ""),
            "delivery_status": packet.get("delivery_status", "delivered"),
            "timestamp": packet.get("timestamp") or datetime.now(UTC).isoformat(),
        }
        # SPEC-032 G243: surface the assistant-unavailable signal through
        # cxp.status too, so a polling caller (not just a live cxp.traffic
        # subscriber) can detect a reply wasn't from the real ConciergeVolk.
        if "assistant_status" in packet:
            event["assistant_status"] = packet["assistant_status"]
        events = self._thread_events.setdefault(thread_id, [])
        events.append(event)
        if len(events) > _MAX_EVENTS_PER_THREAD:
            del events[:-_MAX_EVENTS_PER_THREAD]
        # Bound the number of tracked threads (evict oldest-inserted).
        if len(self._thread_events) > _MAX_TRACKED_THREADS:
            oldest = next(iter(self._thread_events))
            if oldest != thread_id:
                self._thread_events.pop(oldest, None)

    @traces(
        spec="016",
        story="US-003",
        reqs=["REQ-016-051"],
    )
    @traces(
        spec="014",
        story="US-001",
        reqs=["REQ-014-001"],
    )
    async def handle_subscribe(
        self,
        connection_id: str,
        params: dict[str, Any],
    ) -> dict[str, Any]:
        """Handle cxp.subscribe -- subscribe to CXP traffic.

        Args:
            connection_id: Caller connection ID.
            params: Optional filters dict.

        Returns:
            subscription_id, topic, and filters.
        """
        filters = params.get("filters", {})
        sub_id = await self._pubsub.subscribe(connection_id, "cxp.traffic", filters)
        return {
            "subscription_id": sub_id,
            "topic": "cxp.traffic",
            "filters": filters,
        }

    @traces(spec="016", story="US-003", reqs=["REQ-016-053"])
    @traces(
        spec="014",
        story="US-001",
        reqs=["REQ-014-008"],
    )
    async def handle_unsubscribe(
        self,
        connection_id: str,
        params: dict[str, Any],
    ) -> dict[str, Any]:
        """Handle cxp.unsubscribe -- remove CXP subscription.

        Args:
            connection_id: Caller connection ID.
            params: subscription_id to remove.

        Returns:
            Unsubscribe status.
        """
        sub_id = params.get("subscription_id") or params.get("subscriptionId", "")
        if not sub_id:
            raise RPCError(
                "Missing 'subscription_id' parameter",
                rpc_code=-32602,
            )
        found = await self._pubsub.unsubscribe_by_id(sub_id)
        return {
            "status": "unsubscribed",
            "subscription_id": sub_id,
            "found": found,
        }

    @traces(
        spec="014",
        story="US-001",
        reqs=["REQ-014-011"],
    )
    async def handle_status(
        self,
        connection_id: str,
        params: dict[str, Any],
    ) -> dict[str, Any]:
        """Handle cxp.status -- real delivery status for a thread.

        Reports the traffic the gateway has seen for ``thread_id`` (recorded by
        ``publish_traffic``): the per-packet events and the latest delivery
        status. A thread with no observed traffic reports ``"unknown"``.

        Args:
            connection_id: Caller connection ID.
            params: thread_id to query.

        Returns:
            ``thread_id``, latest ``status``, and the list of ``events``.
        """
        thread_id = params.get("thread_id") or params.get("threadId", "")
        if not thread_id:
            raise RPCError(
                "Missing 'thread_id' parameter",
                rpc_code=-32602,
            )
        events = self._thread_events.get(thread_id, [])
        status = events[-1]["delivery_status"] if events else "unknown"
        return {
            "thread_id": thread_id,
            "status": status,
            "events": list(events),
        }