Orchestrata Ultimate › Plans › kernel-dispatch-reply-design

Kernel Dispatch Ack-Only + Reply Addressing — Design

Source: plans/kernel-dispatch-reply-design.md at 8a2cb42ad · last changed 2026-08-19 · first committed 2026-08-05 · 450 lines.

Prose pending.

Plan text

Kernel Dispatch Ack-Only + Reply Addressing — Design

Status: DESIGN ONLY (RUL-022 — no implementation in this artifact) Branch: feat/cxp-direct-transport · Date: 2026-08-05 · Author: analyst (design phase) Defects: (A) kernel violates its documented "send_packet is ack-only" contract — measured 13.9s/21.1s acks + silently dropped volk failures; (B) kernel reply stream has no per-consumer addressing — measured 50/50 reply loss across concurrent gateways.

Every as-built claim below was grounded via the AST toolkit (scripts/serialize_codebase_ast.py {symbols,refs,source}), never by hand-reading .py files. Specs are authoritative (RUL-013). Rulings consulted: RUL-002 (no mocks), RUL-003 (env from profile, injected seams; tunables may default, infra endpoints may not), RUL-011 (ports from env), RUL-020 (async contract), RUL-022 (design before build), RUL-023 (feature means wired), RUL-024 (pre-existing is not a disposition).

Coordination: the volk-comms epic (plans/volk-comms-heartbeat-design.md P1–P8) is landing on this branch NOW. This design touches a DISJOINT surface (kernel_handler / grpc_server / gateway server + cxp handler); the two overlap only as merge-adjacency in kernel/pipeline.py and kernel/cli_main.py — see §7 sequencing. Nothing here duplicates P1–P8; one optional follow-up composes its DeliveryOutcome (§4.1 D8).

1. AS-BUILT INVENTORY

1.1 Defect A path — UI send → gateway → kernel → volk (all AST-verified)

1.2 Defect B path — kernel reply stream → gateway consumer(s)

1.3 Boot/wiring seams (found via refs, not guessed)

• build_kernel_packet_handler (agents/kernel_handler.py:250) has exactly two production use sites (per refs): agents/runtime_main.py:81 (inside _serve:59-96, server = create_cxp_server(config, packet_handler=build_kernel_packet_handler())) and kernel/cli_main.py:594 (import at 578).
• CXPServer.__init__ (cxp/grpc_server.py:417-427) stores self._servicer but exposes no accessor; DirectCXPServer.servicer (direct_cxp_adapter.py:158) is the property precedent to mirror.
• Gateway kernel-client construction: gateway/server.py:512-625 (_init_kernel) → create_cxp_client(config=GRPCClientConfig(host, port)), then self._rpc.set_kernel_client(kernel_client) — the pass-through is gateway/rpc.py:214 RPCHandler.set_kernel_client → gateway/handlers/cxp.py:79 CxpHandler.set_kernel_client. A consumer-id setter mirrors this exact chain.
• Reply-consumer config seam: gateway/config.py:173 (kernel_reply_timeout_s), :179 (kernel_reply_backoff_base_s).
• Deployment: kernel + gateway run as systemd services in the orchestrata-test app container (10.93.53.13) via deploy/orchestrata-test/ — they roll together (version-skew note in §4.1 D7).

1.4 Existing tests on these paths (regression floor)

• tests/gateway/test_kernel_reply_consumer.py — test_pump_broadcasts_volk_reply_to_cxp_traffic, test_pump_raises_when_server_unreachable, test_pump_recovers_after_server_restart.
• tests/gateway/handlers/test_cxp_kernel_dispatch.py — test_cxp_chat_dispatches_to_kernel_with_correlated_thread.
• tests/cxp/test_grpc_kernel_endpoint.py — endpoint resolution + test_server_registers_both_services, test_get_status_returns_live_roster.
• tests/cxp/test_direct_transport.py, test_direct_cxp_adapter.py, test_direct_transport_liveness.py — cxpt transport floor.

2. SPEC CONTRACT REVIEW (specs are authoritative — RUL-013)

SPEC-001 US-002 "Message Routing and Dispatch" — specs/001-kernel/checklists/requirements.md:75-130

(NOTE: SPEC-001's checklist numbers this story's requirements REQ-002-NNN, distinct from the postman checklist's REQ-002-USn-NNN series.)

• REQ-001-US-002-001 — Route message based on TO field in CXP packet.
• REQ-001-US-002-003 — Support unicast (single recipient) routing.
• REQ-001-US-002-008 — Handle routing to offline Volks.
• REQ-001-US-002-009 — Handle routing to unknown recipients.
• REQ-001-US-002-019 — Provide delivery acknowledgments.
• REQ-001-US-002-030 — Handle queue overflow gracefully.
• REQ-001-US-002-031..035 — Track message status: queued / dispatched / delivered / acknowledged / failed.
• REQ-001-US-002-037 — Store status transitions with timestamps.
• REQ-001-US-002-038 — Notify sender of status changes.

SPEC-002 (postman) — specs/002-postman/checklists/requirements.md

• REQ-002-US-001-011 (line 35) — Support for NOTIFY message type.
• REQ-002-US-001-026 (line 54) — Notify sender of permanent failures.

SPEC-014 (gateway) — specs/014-websocket-gateway/checklists/requirements.md

• REQ-014-001/002 — Client subscribes to CXP traffic; server pushes CXP packets as they flow.
• REQ-014-071 — Client sends CXP packet via cxp.send (the RPC that currently blocks).

SPEC-032 G243 — specs/032-concierge-volk/spec.md:380,422 + checklists/requirements.md:85-89

• REQ-032-053 — gateway MUST dispatch to the real live ConciergeVolk when reachable and MUST NOT fabricate a reply.
• REQ-032-054 — when not reachable or dispatch fails, gateway MUST return an explicit, machine-detectable "assistant unavailable" response — never a conversational-looking fallback.

The G243 distinction this design must honor: G243 forbids fabricated conversational replies (a canned text pretending to be the assistant). An honest, machine-generated FAILURE status packet — sent from the KERNEL's own identity, flagged with a machine-detectable status field, never conversational prose posing as U:Cx — is the opposite of a fabricated reply: it is exactly the REQ-001-US-002-035/038 + REQ-002-US-001-026 sender notification, in the same assistant_status convention _build_unavailable_reply (gateway/handlers/cxp.py:239-270) already established (_ASSISTANT_STATUS_UNAVAILABLE = "unavailable", flag both top-level and in payload, text plainly labelled an automated notice).

Status vocabulary already exists — cxp/_packet_enums.py:45 DeliveryStatus (state machine QUEUED → DELIVERED|FAILED; DELIVERED → ACKNOWLEDGED|FAILED) and PacketType.ERROR (_packet_enums.py:20-30). No new enums are needed.

3. GAP LIST (as-built vs spec, each with evidence)

4. CONTRACT DESIGN

4.1 Key decisions

Defect A — ack-immediately + background intent + honest terminal failure (G1, G2, G6):

1. D1 — Ack-only __call__, background the round (closes G1). For chat subjects (intent/response), _KernelPacketHandler.__call__ runs ONLY the synchronous pre-flight — _govern (governance still sees every intent before ack; a denial returns success=True with reply_subject="intent_denied" — CORRECTED 2026-08-05 during P-A3: the original text here said success=False, but the as-built contract, pinned by test_kernel_handler_governance.py::test_govern_denies_forbidden_intent, returns a SUCCESSFUL receipt whose reply is the denial) — then mints/keeps the thread id and schedules asyncio.create_task(self._process_in_background(...)), returning the ack immediately: the existing key shape (success/thread_id/reply/reply_payload/ reply_subject/reply_packet — reply fields empty) plus accepted: True. This is the receipt = QUEUED/DISPATCHED acknowledgment (REQ-001-US-002-019/031/032). subject == "status" stays fully synchronous (_status_reply:119). Precedent: PipelineKernelClient.send_packet:130-158 (_spawn + return) — this design makes the out-of-process path match the in-process one. Threading posture (free-threaded 3.14t): create_task lands on the servicer's own loop — grpc.aio's loop for the grpc transport, the DirectCXPServer loop for cxpt (_dispatch:206-249 executes servicer coroutines there via run_coroutine_threadsafe) — and the blocking engine call stays on asyncio.to_thread exactly as _run_intent:158-175 does today. The handler holds strong refs (_bg_tasks: set, discard-on-done); tasks die with the server loop on stop() (bounded lifecycle, no daemon threads).
2. D2 — Reply sink seam (makes D1 possible). The background job cannot return a reply_packet through the receipt, so the handler gains set_reply_sink(sink: Callable[[dict], None]) and publishes replies itself. The sink is CXPServicer.enqueue_reply — a NEW public method that is also Defect B's single routing point. CXPServer gains a servicer property (mirror of DirectCXPServer.servicer:158; self._servicer already exists at grpc_server.py:426). Boot order at both construction sites: handler = build_kernel_packet_handler(); server = create_cxp_server(config, packet_handler=handler); handler.set_reply_sink(server.servicer.enqueue_reply). Back-compat: send_packet:158-205's existing inline result.get("reply_packet") branch delegates to enqueue_reply too, so any handler still returning inline replies gets identical routing.
3. D3 — Honest terminal FAILURE packet (closes G2, respects G243). The background job wraps the round; on exception (the verified ProviderError propagation path) it sinks _build_failure_packet(thread_id, error, reply_to): from_agent="KERNEL" (never the volk's alias — no impersonation), to_agent="HUMAN", same thread_id, subject="intent_failed", packet_type=PacketType.ERROR, delivery_status=DeliveryStatus.FAILED, timestamp (REQ-001-US-002-037), payload {"text": <plainly-labelled automated failure notice>, "error": str, "failure": true, "assistant_status": "failed"} — the same machine-detectable convention as _build_unavailable_reply:239-270, so the chat gadget detects it with the field it already checks. This is REQ-001-US-002-035 (track failed) + REQ-001-US-002-038/REQ-002-US-001-026 (notify sender); it is NOT a fabricated reply under G243 because it never claims to be the assistant and is machine-flagged end-to-end. Terminal contract: one terminal packet per accepted intent — the genuine reply (DELIVERED) or the failure packet (FAILED). The pre-existing "ran fine but produced no human-directed reply" case (produced=False, queued-for-offline path _route_sync_deliver:85-106) intentionally keeps today's no-packet semantics (REQ-001-US-002-018 queueing is Postman's job, not a failure).
4. D4 — Gateway checks the receipt it already has (closes G2's sync half). _maybe_reply_as_concierge inspects the send receipt: success=False (transport/handler-level failure — NOT governance denial, which returns success=True + reply_subject="intent_denied"; corrected 2026-08-05, see D1) → emit the existing _build_unavailable_reply notice, same as a raised exception. One conditional; zero new machinery; REQ-032-054's "or dispatch fails" clause becomes true for the receipt path.
5. D5 — Thread-correct reply selection (closes G6). PipelineRunner.get_last_human_message gains an optional thread_id: str | None = None; when given, scan human_inbox in reverse for the first packet with that thread_id. _build_reply passes the thread. Unfiltered call keeps byte-identical behavior (back-compat with every existing caller).

Defect B — reply addressing (G3, G4, G5, G7):

6. D6 — Chosen: reply-to metadata stamped at send + per-consumer queues, keyed drain via the EXISTING agent_id field. Three candidates were evaluated:

   Option	Verdict	Rationale
   (a) consumer-id in receive_packets alone	REJECTED as sole mechanism	The servicer cannot know which consumer owns a reply unless the reply is tagged; a key on the drain side without a tag on the supply side routes nothing. It IS the right drain half, and CXPReceiveRequest.agent_id (proto:93-97, transmitted by both clients, ignored today) already carries it — no proto change.
   (b) thread→consumer registry in Redis	REJECTED	Adds an infra hop + registration race (reply can beat the registration write) + TTL/cleanup lifecycle for dead gateways + Redis access from inside both transport servers. The sender already KNOWS the answer at send time; a registry re-derives it expensively.
   (c) reply-to metadata stamped at send	CHOSEN (combined with (a)'s keyed drain)	Zero proto changes (metadata map field 14 verified round-trip in all four codec directions, §1.2); zero new infra; one routing point (enqueue_reply) covers grpc AND cxpt because both share the one CXPServicer (§1.2 row 7); Defect A's failure packets inherit correct addressing for free.
   Mechanics: each gateway mints _reply_consumer_id = f"HUMAN@{uuid4().hex[:12]}"
   at construction; stamps metadata={"reply_to": <id>} on every kernel-bound
   chat packet; consumes with receive_packets(agent_id=<id>). The kernel
   handler propagates inbound metadata["reply_to"] onto the reply AND failure
   packet. ReplyRouter._pick_key: metadata.reply_to if present, else
   to_agent (legacy unstamped traffic keeps queueing under "HUMAN" —
   backward compatible with every existing test/client), else a default key.

7. D7 — ReplyRouter (new, cxp/reply_router.py), composed into CXPServicer (closes G3/G4/G7). Per-key asyncio.Queues created on demand; route(packet) -> str picks the key and enqueues with bounded-overflow drop-oldest + traced warning (max_queue ctor param, default 256 — a tunable, not an infra endpoint, so a literal default is permitted under RUL-003's inverse rule). receive_packet drains queue_for(request.agent_id); subscribe drains queue_for(request.agent_id) too (its CXPSubscription.agent_id proto field exists, proto:99-103), which ends G4's cross-consumer destruction. All servicer coroutines execute on one loop per server (grpc.aio / DirectCXPServer), so a plain dict of queues needs no additional locking; the composed Tracer records route/overflow decisions. Dead-consumer cost is bounded at max_queue dicts per departed id (TTL eviction: explicit non-goal, below). Version-skew caveat: a stamping gateway against an old kernel (no propagation) would wait on an empty private queue. Kernel + gateway deploy together in the app container (deploy/orchestrata-test/), and Group B packets land kernel-side propagation (P-B2) before the gateway flip (P-B3), so the skew window does not occur in this fleet.
8. D8 — Coordination with the comms epic (compose, don't duplicate). Nothing in P1–P8 touches these files' defect sites; conversely this design does NOT add delivery-outcome types (P3 owns DeliveryOutcome/aroute_outcome), does not touch heartbeats, and does not modify status_report. One optional follow-up (NOT in these packets): once P3 lands, a route_sync_outcome twin would let the background job distinguish NO_ROUTE (REQ-001-US-002-009) from HANDLER_ERROR without exception inference; the measured defect (exception drop) does not need it, so Group A ships independent of P3's landing time.

Non-goals (MVP discipline): consumer-queue TTL eviction; ACKNOWLEDGED lifecycle wiring (the acknowledge RPC exists at grpc_server.py:303 but no production consumer acks — REQ-001-US-002-034 stays with its current owner); UI/chat gadget rendering of intent_failed; multi-kernel fan-in; changing PipelineKernelClient (already contract-compliant); mapping the model's unused reply_to field into the proto (metadata is the carrier; a field-level mapping is a SPEC-004 decision).

4.2 Contract diagram (also at plans/contracts/kernel-dispatch-reply.mmd)

Diagram source shown as text: Mermaid could not parse it (Error: Lexical error on line 22. Unrecognized text.).

classDiagram
    direction LR

    %% ═══ KERNEL DISPATCH ACK-ONLY + REPLY ADDRESSING — contract (analyst, 2026-08-05) ═══
    %% REUSE (compose, never regenerate): cxp.grpc_server.CXPServicer (send_packet:158,
    %% receive_packet:278, subscribe:291, codecs :329/:365 both metadata-capable),
    %% CXPServer:414 (_servicer:426), DirectCXPServer:114 (shares the SAME CXPServicer via
    %% .servicer:158 — one fix covers both transports), agents.kernel_handler
    %% ._KernelPacketHandler:42, kernel.pipeline.PipelineRunner (submit_intent:294,
    %% respond:390, get_last_human_message:524), gateway.handlers.cxp.CxpHandler
    %% (_build_unavailable_reply:239 — machine-detectable precedent), gateway.server
    %% .WebSocketGateway (_consume_kernel_replies:792), orchestrator.pipeline_kernel
    %% .PipelineKernelClient.send_packet:130 (ack-then-background precedent),
    %% cxp._packet_enums.DeliveryStatus + PacketType.ERROR, core.tracing.Tracer.
    %% New module homes:
    %%   cxp/reply_router.py -> ReplyRouter (new)
    %%   edits: cxp/grpc_server.py, agents/kernel_handler.py, kernel/pipeline.py,
    %%          agents/runtime_main.py, kernel/cli_main.py, gateway/handlers/cxp.py,
    %%          gateway/rpc.py, gateway/server.py

    class ReplyRouter {
        <<class>>
        +__init__(tracer: Tracer, max_queue: int) None
        +route(packet: dict) str
        +queue_for(consumer_id: str) Queue
        -_pick_key(packet: dict) str
        -_queues: dict
        -_max_queue: int
    }
    %% traces spec=001 story=US-002 reqs=REQ-001-US-002-001,REQ-001-US-002-003,REQ-001-US-002-008,REQ-001-US-002-030
    ReplyRouter o-- Tracer : composes (DI)

    class CXPServicer {
        <<existing — MODIFIED  cxp/grpc_server.py>>
        +__init__(packet_handler: Callable, reply_router: ReplyRouter | None) None
        +enqueue_reply(packet: dict) None
        +receive_packet(request: CXPReceiveRequest, context: Any) AsyncIterator
        +subscribe(request: CXPSubscription, context: Any) AsyncIterator
    }
    %% traces spec=001 story=US-002 reqs=REQ-001-US-002-019,REQ-001-US-002-033,REQ-001-US-002-001
    CXPServicer o-- ReplyRouter : composes (DI; keyed queues replace the single _packet_queue)

    class CXPServer {
        <<existing — MODIFIED  cxp/grpc_server.py>>
        +servicer() CXPServicer
    }
    %% traces spec=001 story=US-002 reqs=REQ-001-US-002-019
    CXPServer o-- CXPServicer : composes (already; property mirrors DirectCXPServer.servicer:158)

    class KernelPacketHandler {
        <<existing — MODIFIED  agents/kernel_handler.py>>
        +__call__(packet: dict) dict
        +set_reply_sink(sink: Callable) None
        -_process_in_background(subject: str, thread_id: str, text: str, reply_to: str) None
        -_build_failure_packet(thread_id: str, error: str, reply_to: str) dict
        -_reply_sink: Callable | None
        -_bg_tasks: set
    }
    %% traces spec=001 story=US-002 reqs=REQ-001-US-002-019,REQ-001-US-002-031,REQ-001-US-002-032,REQ-001-US-002-033,REQ-001-US-002-035,REQ-001-US-002-038
    KernelPacketHandler ..> CXPServicer : reply_sink = servicer.enqueue_reply (wired at boot)
    KernelPacketHandler ..> KernelPipelineRunner : submit_intent / respond via to_thread (background)

    class KernelPipelineRunner {
        <<existing — MODIFIED  kernel/pipeline.py>>
        +get_last_human_message(thread_id: str | None) CXPPacket | None
    }
    %% traces spec=001 story=US-002 reqs=REQ-001-US-002-033

    class CxpHandler {
        <<existing — MODIFIED  gateway/handlers/cxp.py>>
        +set_reply_consumer_id(consumer_id: str) None
        -_maybe_reply_as_concierge(target: str, subject: str, thread_id: str, payload: dict) None
        -_reply_consumer_id: str
    }
    %% traces spec=032 story=US-032-G243 reqs=REQ-032-053,REQ-032-054
    CxpHandler ..> KernelPacketHandler : send_packet stamps metadata reply_to + receipt.success now CHECKED

    class WebSocketGateway {
        <<existing — MODIFIED  gateway/server.py>>
        -_reply_consumer_id: str
        -_consume_kernel_replies() None
    }
    %% traces spec=014 story=US-001 reqs=REQ-014-001,REQ-014-002
    WebSocketGateway ..> CXPServicer : receive_packets(agent_id=_reply_consumer_id)
    WebSocketGateway o-- CxpHandler : composes (via RPCHandler.set_kernel_client:214 precedent)

patternRole map for gen-skeleton: ReplyRouter=strategy (routing policy, fully generatable shell; _pick_key+overflow=leaf) · enqueue_reply=leaf · CXPServer.servicer=boilerplate property · _process_in_background/ _build_failure_packet=leaf (builder-written) · everything else = edits to existing classes (builder/wirer tasks, not generated).

4.3 Wiring plan (RUL-023 / Hard Rule 8 — named production seams, found via refs)

Transport constraint respected: metadata + CXPReceiveRequest.agent_id are the only carriers, both already present in the proto (cxp.proto:69, 93-97) and serialized by both transports — zero proto changes, zero new RPCs; the test profile's CXP_TRANSPORT=cxpt path is covered by the shared-servicer fact.

5. TDD TEST PLAN (tests-first; real infra only — Redis 10.93.53.11, graph 10.93.53.10, kernel 10.93.53.13:50053, env profile test; no mocks/monkeypatch per RUL-002; slow/failing volks are REAL registered plugin-volk handlers via _PluginVolkAdapter (kernel/pipeline.py:50-94, SPEC-015 production feature) — no LLM in the loop, no test doubles)

Runner discipline: scripts/run_tests.py --affected / targeted modules only, live-observed (observe_pytest attach, RUL-009), 5-min cap per run; kernel required (never --skip kernel).

6. BUILD PACKETS (each <5 min for a builder agent; tests-first inside each; Group A ships without Group B)

Group A — ack-only + honest failure (G1, G2, G6)

Group B — reply addressing (G3, G4, G5, G7)

Sequencing vs the comms epic: no shared defect sites, but kernel/pipeline.py (P-A1 here vs comms-P7 status_report) and kernel/cli_main.py (P-A4 here vs comms-P8 boot wiring) are merge-adjacent — land each file's packets serially, either order. P-A2/P-B1 are gen-skeleton-friendly (property/leaf-method shells + traced test stubs); the rest are builder/wirer edits to existing classes. Deploy note (RUL-012): Group A and Group B each end with redeploying the app container's kernel+gateway services (deploy/orchestrata-test/) and re-running the paired chat probe that measured the defect — B's acceptance is 2/2 paired probes delivered to their own gateways, A's is ack <1s + a failure packet observed on a forced provider error.

Related

• Plans
• Orchestrata Ultimate — project index
• Workflow
• Tasks