orchestrata_ultimate.agents.kernel_handler

Reference › orchestrata_ultimate › agents › kernel_handler

kernel_handler builds the packet handler that connects live CXP gRPC traffic to the volk engine. build_kernel_packet_handler returns a _KernelPacketHandler bound to a booted kernel PipelineRunner (booted over all the volks when none is passed), an optional memory engine and an optional intent broker. runtime_main uses it, as does the kernel CLI; it binds run memory through memory_context.

The handler is an async callable taking a packet dictionary. A packet with subject status is answered at once by _status_reply without running the pipeline. Every other packet first passes the governance gate _govern, which returns a rejection reply when the intent is denied. For chat subjects, when a reply sink has been set with set_reply_sink, the handler acknowledges immediately with accepted true and runs the round in the background (_process_in_background), which publishes exactly one terminal packet through the sink. _run_intent runs the blocking engine off the gRPC loop, _bind_memory binds the memory engine to the thread, _remember stores the turn best-effort, and _build_reply reads the engine's last human-inbox message. Terminal packets are built by _build_terminal_packet and its variants for failure, no reply and draining; _stamp_reply_to copies the sender's reply target onto the reply. shutdown drains in-flight rounds; after it begins, new chat packets are rejected synchronously with a draining receipt.

At a glance

Class diagram

Dependency diagram

API

Classes

_KernelPacketHandler

Line 71

Async-callable packet_handler bound to one booted PipelineRunner (+ optional memory engine).

Functions

Source

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

"""Packet handler that drives the volk engine for the live CXP gRPC kernel.

``agents/runtime_main.py`` starts a :class:`CXPServer` for the out-of-process
kernel (the ``agent-runtime`` service). This module builds the ``packet_handler``
that the server invokes for each received packet: it boots a
:class:`~orchestrata_ultimate.kernel.pipeline.PipelineRunner` (all 9 volks on a
real ``Postman``) **once**, then routes each incoming packet's intent/response
text through the engine and returns the volk's reply.

Reply transport note: the unary ``send_packet`` gRPC RPC returns only an ack
(``CXPReceipt``). The ``reply_packet`` produced here is enqueued by the servicer
onto the streaming channel and pumped to the gateway's UI subscribers by
``WebSocketGateway._consume_kernel_replies``; the handler's return value also
carries the reply for direct callers/tests.

Ack-only mode (P-A3, ``plans/kernel-dispatch-reply-design.md`` D1/D2/D3;
hardened by P-H4, ``plans/kernel-dispatch-reply-hardening.md`` D-H5): once
``set_reply_sink`` has been called (production wiring happens at boot --
packet P-A4, NOT this module), chat subjects (``intent``/``response``) are
acknowledged immediately and the volk round runs in the background,
publishing EXACTLY ONE terminal packet through the sink instead of the
receipt: the genuine reply, the ``intent_no_reply`` terminal when the round
completed but produced nothing human-directed, an honest ``intent_failed``
failure packet on exception, or (P-H5, D-H6) the same failure packet when
:meth:`_KernelPacketHandler.shutdown` cancels a round still in flight past
its drain grace period. Without a sink, this handler is unchanged: fully
synchronous, reply inline in the receipt.

Once :meth:`_KernelPacketHandler.shutdown` begins draining (P-H10,
``plans/kernel-dispatch-reply-hardening.md`` §11.2 D-H11), a fresh
chat-subject packet is never backgrounded at all: it gets a synchronous
``intent_draining``/``KERNEL_DRAINING`` rejection receipt instead, returned
the same way a synchronous governance denial already is, so the servicer's
existing plumbing enqueues it onto the still-open reply stream.
"""

from __future__ import annotations

import asyncio
import logging
import os
import uuid
from collections.abc import Awaitable, Callable
from datetime import UTC, datetime
from typing import TYPE_CHECKING, Any

from orchestrata_ultimate.agents.volks.memory_context import (
    register_memory,
    unregister_memory,
)
from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.packet import DeliveryStatus, PacketType

if TYPE_CHECKING:
    # Type-only: a module-level runtime import here is an upward layer
    # violation (agents layer 3 -> kernel layer 4); the runtime import is
    # function-scoped at the construction site (accepted pattern, see
    # architecture-standards.md).
    from orchestrata_ultimate.kernel.pipeline import PipelineRunner

logger = logging.getLogger(__name__)

KernelPacketHandler = Callable[[dict[str, Any]], Awaitable[dict[str, Any]]]

# Chat subjects eligible for the ack-only background path (D1) once a reply
# sink is wired. "status" always stays synchronous (answered without running
# the pipeline); every other subject keeps today's fully synchronous round.
_CHAT_SUBJECTS = frozenset({"intent", "response"})


class _KernelPacketHandler:
    """Async-callable ``packet_handler`` bound to one booted ``PipelineRunner``
    (+ optional memory engine).

    Extracted from :func:`build_kernel_packet_handler` so each piece of the
    per-packet logic (status report, memory binding, intent dispatch, reply
    assembly) is its own small method instead of one large closure.
    """

    def __init__(
        self, runner: PipelineRunner, mem: Any | None, broker: Any | None = None
    ) -> None:
        self._runner = runner
        self._mem = mem
        self._mem_ready = False
        # SPEC-003 Option A: the broker is the GOVERNED FRONT DOOR. When wired,
        # each new work intent is classified + validated (Prolog governance,
        # fail-closed) before the PipelineRunner executes it. The broker's own
        # route/dispatch is unused here — execution stays with the pipeline.
        self._broker = broker
        # P-A3 (plans/kernel-dispatch-reply-design.md D1/D2): unset by default,
        # which keeps __call__ fully synchronous (byte-for-byte with the
        # pre-P-A3 behavior). set_reply_sink activates the ack-only background
        # path for chat subjects.
        self._reply_sink: Callable[[dict[str, Any]], None] | None = None
        # Strong refs to in-flight background rounds so they aren't
        # garbage-collected mid-flight; discarded via the done callback.
        self._bg_tasks: set[asyncio.Task[None]] = set()
        # D-H11 (F-3 hardening, plans/kernel-dispatch-reply-hardening.md
        # §11.2): flipped True as shutdown()'s FIRST action, before the
        # background-task snapshot, so accept-reject (__call__) and the
        # drain snapshot (shutdown) can never interleave a stray spawn.
        # __call__ and shutdown both run on the kernel's single event loop
        # (never concurrent OS threads, and neither awaits between reading
        # and branching on this flag), so plain attribute read/write
        # visibility is sufficient -- no lock is needed here.
        self._draining: bool = False

    def set_reply_sink(self, sink: Callable[[dict[str, Any]], None]) -> None:
        """Wire the reply-publication sink (D2).

        Production wiring passes ``server.servicer.enqueue_reply`` at boot
        (packet P-A4, not this module). Once set, ``__call__`` acks chat
        subjects (``intent``/``response``) immediately and runs the volk
        round in the background, publishing its terminal packet through this
        sink instead of the (now-empty) receipt.
        """
        self._reply_sink = sink

    @staticmethod
    def _stamp_reply_to(
        reply_packet: dict[str, Any], incoming: dict[str, Any]
    ) -> dict[str, Any]:
        """Stamp *reply_packet*'s ``metadata.reply_to`` from *incoming*'s
        (P-B2/P-B7, D6) -- the ONE mechanism every reply-packet producer in
        this handler funnels through (``_govern``'s denial, ``_status_reply``,
        ``_build_reply``, ``_build_failure_packet``). A no-op -- no
        ``metadata`` key invented -- when *incoming* carries no non-empty
        ``metadata.reply_to``, so unstamped/legacy callers get byte-identical
        output to before this helper existed.
        """
        reply_to = (incoming.get("metadata") or {}).get("reply_to", "")
        if reply_to:
            reply_packet["metadata"] = {"reply_to": reply_to}
        return reply_packet

    async def _govern(
        self,
        subject: str,
        thread_id: str,
        text: str,
        incoming: dict[str, Any] | None = None,
    ) -> dict[str, Any] | None:
        """Governance gate. Returns a rejection reply dict when the intent is
        DENIED (caller must not run the pipeline), or ``None`` to proceed.

        Only new work intents are gated — never ``status`` polls or in-thread
        ``response`` follow-ups (which the concierge already owns). A gate
        *error* (infra/classification failure, not a policy denial) logs and
        proceeds: governance must not reduce availability on its own fault.

        *incoming* (P-B7, D6) is the raw inbound packet; when it carries
        ``metadata.reply_to``, the denial reply is stamped with it via
        :meth:`_stamp_reply_to` so a DENIED intent still routes back to the
        exact consumer that sent it. Optional (defaults to ``None``, treated
        as unstamped) so direct unit calls need not construct one.
        """
        broker = self._broker
        if broker is None or subject in ("status", "response") or not text:
            return None
        evaluator = getattr(broker, "evaluator", None)
        if evaluator is None:
            return None
        try:
            intent = evaluator.evaluate(text).intent
            validation = await broker.validate_only(intent)
            if getattr(validation.outcome, "value", "") == "allowed":
                return None
            reason = getattr(validation, "reason", "") or "governance policy: denied"
            logger.info("intent DENIED by governance: %s (%s)", text[:80], reason)
            denial = {
                # Real id, same contract as _build_terminal_packet: consumers
                # key dedup/claim-checks on message_id (H-B probe finding).
                "message_id": str(uuid.uuid4()),
                "from_agent": "SYSTEM",
                "to_agent": "HUMAN",
                "thread_id": thread_id or "",
                "subject": "intent_denied",
                "payload": {"reason": reason, "intent_id": intent.id},
            }
            self._stamp_reply_to(denial, incoming or {})
            return {
                "success": True,
                "thread_id": thread_id or "",
                "reply": reason,
                "reply_payload": {"reason": reason, "intent_id": intent.id},
                "reply_subject": "intent_denied",
                "reply_packet": denial,
            }
        except Exception:
            logger.exception("governance gate error (non-fatal); allowing intent")
            return None

    async def _remember(self, content: str, volk_id: str, session_id: str) -> None:
        """Persist one conversation turn to long-term memory (best-effort)."""
        if self._mem is None or not content or not session_id:
            return
        try:
            if not self._mem_ready:
                await self._mem.initialize({})
                self._mem_ready = True
            await self._mem.store(content, volk_id=volk_id, session_id=session_id)
        except Exception:  # auxiliary — never break the orchestration reply
            logger.exception("kernel packet_handler: memory store failed (non-fatal)")

    def _status_reply(self, packet: dict[str, Any], thread_id: str) -> dict[str, Any]:
        """Live volk status — answered without running the pipeline, so the
        orchestrator/gateway can poll every volk's status at any time. The
        report is emitted as a reply_packet so the gRPC servicer enqueues it
        onto the streaming channel (the unary receipt is ack-only); in-process
        callers can also read reply_payload directly.

        *packet*'s ``metadata.reply_to`` (P-B7, D6), when present, is stamped
        onto ``status_reply`` via :meth:`_stamp_reply_to` so the poll's reply
        routes back to the exact consumer that asked.
        """
        report = self._runner.status_report()
        status_reply = {
            # Real id for the same reason as every other reply this module
            # emits: message_id is the consumer's dedup/claim-check key.
            "message_id": str(uuid.uuid4()),
            "from_agent": "SYSTEM",
            "to_agent": packet.get("from_agent") or "HUMAN",
            "thread_id": thread_id or "",
            "subject": "status_report",
            "payload": {"volks": report},
        }
        self._stamp_reply_to(status_reply, packet)
        return {
            "success": True,
            "thread_id": thread_id or "",
            "reply": "",
            "reply_payload": {"volks": report},
            "reply_subject": "status_report",
            "reply_packet": status_reply,
        }

    async def _bind_memory(self, thread_id: str) -> bool:
        """Bind the memory engine for this conversation so volks can recall
        relevant prior turns while the pipeline runs (best-effort; only for a
        continuing thread, which is exactly when stored history exists).
        """
        if self._mem is None or not thread_id:
            return False
        try:
            await self._mem.initialize({})
        except Exception:
            logger.debug("memory initialize failed (non-fatal)", exc_info=True)
        register_memory(thread_id, self._mem)
        return True

    async def _run_intent(
        self, subject: str, thread_id: str, text: str
    ) -> tuple[str, bool]:
        """Run submit_intent/respond off the gRPC loop (blocking-sync engine
        calls); return the (possibly new) thread_id and whether a reply was
        produced.
        """
        if subject == "response" and thread_id:
            await asyncio.to_thread(self._runner.respond, thread_id, text)
            return thread_id, True
        # Honour the caller's thread_id so the volk reply correlates to the
        # gateway/UI conversation (and a later respond() resolves it).
        new_tid = await asyncio.to_thread(
            self._runner.submit_intent, text, thread_id or None
        )
        if new_tid is not None:
            return new_tid, True
        return thread_id, False

    @traces(
        spec="001",
        story="US-002",
        reqs=["REQ-001-US-002-001", "REQ-001-US-002-033", "REQ-001-US-002-039"],
    )
    async def _build_reply(
        self, text: str, thread_id: str, reply_to: str = ""
    ) -> tuple[str, dict[str, Any], str, dict[str, Any]]:
        """Read the engine's last human-inbox message (if any) and persist this
        turn (human intent + volk reply) to long-term memory under the
        conversation thread, when an engine is wired.

        The lookup is filtered to *thread_id* (G6, P-A1's
        ``get_last_human_message(thread_id=...)``), so a reply from an older,
        unrelated conversation still sitting in the shared human-inbox can
        never be mis-attributed to this round. It also consumes the match
        (P-H3, D-H3, ``consume=True``) -- a claim-check: once THIS round has
        matched and returned a reply, that inbox entry is gone, so a LATER
        round on the same thread that produces nothing of its own can never
        re-match it and re-emit it as its own terminal (REQ-001-US-002-033/039).

        *reply_to* (P-B2, D6) is the inbound packet's ``metadata.reply_to`` --
        when non-empty it is stamped onto the built ``reply_packet`` (via the
        shared :meth:`_stamp_reply_to`) so ``ReplyRouter._pick_key`` routes
        the terminal reply to the exact consumer that sent the request.
        Empty (the unstamped/legacy default) leaves ``reply_packet`` with no
        ``metadata`` key at all -- byte-identical to the pre-P-B2 dict.
        """
        reply_text, reply_payload, reply_subject = "", {}, ""
        reply_packet: dict[str, Any] = {}
        reply = self._runner.get_last_human_message(thread_id=thread_id, consume=True)
        if reply is not None and isinstance(reply.payload, dict):
            reply_payload = reply.payload
            reply_text = reply.payload.get("text", "")
            reply_subject = reply.subject
            # A _dict_to_proto-compatible reply packet that the servicer
            # enqueues onto the streaming channel for the gateway.
            reply_packet = {
                "message_id": getattr(reply, "message_id", "") or "",
                "from_agent": reply.from_agent,
                "to_agent": reply.to_agent,
                "thread_id": reply.thread_id or thread_id,
                "subject": reply.subject,
                "payload": reply_payload,
            }
            self._stamp_reply_to(reply_packet, {"metadata": {"reply_to": reply_to}})

        await self._remember(text, "HUMAN", thread_id)
        if reply is not None and reply_text:
            await self._remember(reply_text, reply.from_agent, thread_id)
        return reply_text, reply_payload, reply_subject, reply_packet

    @traces(
        spec="001",
        story="US-002",
        reqs=[
            "REQ-001-US-002-001",
            "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",
            "REQ-001-US-002-039",
        ],
    )
    async def _process_in_background(
        self, subject: str, thread_id: str, text: str, reply_to: str = ""
    ) -> None:
        """Run the volk round off the ack path (D1) and publish EXACTLY ONE
        terminal packet through the sink on every exit path (D-H5, F5 fix —
        amends the parent design's D3): the genuine reply on success, the
        new ``intent_no_reply`` terminal when the round completed but
        produced nothing human-directed, or an honest ``intent_failed``
        packet on exception.

        Before this fix, a ``produced=False`` round (governance-blocked
        ``submit_intent``, or a ``respond`` that matched no inbox entry) sank
        NOTHING — indistinguishable from packet loss to a sender who was
        already told (via the ack) that a background round was running. Once
        the ack promises a reply-or-terminal, silence is a hole, not a
        feature (REQ-001-US-002-035/038, REQ-002-US-001-026).

        *reply_to* (P-B2, D6) is the inbound packet's stamped consumer id,
        forwarded unchanged to ``_build_reply``, ``_build_no_reply_packet``,
        and ``_build_failure_packet`` so whichever terminal packet this round
        produces carries the same addressing.
        """
        sink = self._reply_sink
        if sink is None:  # defensive — only ever scheduled when a sink is set
            return
        memory_bound = False
        try:
            memory_bound = await self._bind_memory(thread_id)
            thread_id, produced = await self._run_intent(subject, thread_id, text)
            reply_packet: dict[str, Any] = {}
            if produced:
                _, _, _, reply_packet = await self._build_reply(
                    text, thread_id, reply_to
                )
            if reply_packet:
                sink(reply_packet)
            else:
                sink(self._build_no_reply_packet(thread_id, reply_to))
        except asyncio.CancelledError:
            # D-H6 (F5 fix): shutdown() cancels stragglers past the drain
            # grace period. CancelledError is a BaseException -- it never
            # matches the `except Exception` arm below, so without this arm
            # a cancelled round sank nothing (the RUL-020 violation this
            # closes). Sink an honest failure terminal, THEN re-raise so the
            # cancellation still propagates (never swallow it).
            logger.warning(
                "kernel packet_handler background round cancelled (shutdown)"
            )
            sink(
                self._build_failure_packet(thread_id, "kernel shutting down", reply_to)
            )
            raise
        except Exception as exc:  # the verified ProviderError propagation path
            logger.exception("kernel packet_handler background round failed")
            sink(self._build_failure_packet(thread_id, str(exc), reply_to))
        finally:
            if memory_bound:
                unregister_memory(thread_id)

    @traces(
        spec="001",
        story="US-002",
        reqs=[
            "REQ-001-US-002-033",
            "REQ-001-US-002-035",
            "REQ-001-US-002-037",
            "REQ-001-US-002-038",
            "REQ-001-US-002-039",
        ],
    )
    @traces(
        spec="032",
        story="US-032-G243",
        reqs=["REQ-032-053", "REQ-032-054"],
    )
    def _build_terminal_packet(
        self,
        thread_id: str,
        subject: str,
        delivery_status: DeliveryStatus,
        payload: dict[str, Any],
        reply_to: str = "",
    ) -> dict[str, Any]:
        """Shared terminal-packet core (D-H5): the one dict shape every
        background-round terminal (failure, no-reply) is built from —
        ``_build_failure_packet`` and ``_build_no_reply_packet`` are thin
        wrappers over this.

        Always sent from the KERNEL's own identity — never a volk alias — so
        a terminal packet can never be mistaken for a fabricated
        conversational reply (SPEC-032 G243); machine-flagged the same way
        ``_build_unavailable_reply`` (``gateway/handlers/cxp.py``) already
        flags an unreachable assistant — *payload*'s ``assistant_status``
        (when present) is mirrored onto the packet's TOP LEVEL too (backlog
        [403]): the gateway's ``cxp.py::_record_event`` only reads
        ``packet["assistant_status"]``, never ``packet["payload"][...]``, so
        without this mirror a kernel terminal's status was invisible to that
        top-level-only surfacing even though the payload carried it all
        along. ``packet_type`` follows *delivery_status*: ``FAILED`` rounds
        are ``PacketType.ERROR``, everything else (the round completed,
        there's simply no reply) is ``PacketType.NOTIFICATION``.

        *reply_to* (P-B2, D6): stamped onto the packet's ``metadata`` (via
        the shared :meth:`_stamp_reply_to`) when non-empty, so the terminal
        routes back to the exact consumer that sent the request instead of
        the ``to_agent``/default fallback.
        """
        packet: dict[str, Any] = {
            # A real id, not "": terminal packets are consumer-visible CXP
            # packets — an empty message_id defeats client-side dedup/
            # distinctness checks (caught live by the H-B acceptance probe,
            # 2026-08-05: two rounds' terminals were indistinguishable).
            "message_id": str(uuid.uuid4()),
            "from_agent": "KERNEL",
            "to_agent": "HUMAN",
            "thread_id": thread_id or "",
            "subject": subject,
            "packet_type": (
                PacketType.ERROR
                if delivery_status == DeliveryStatus.FAILED
                else PacketType.NOTIFICATION
            ),
            "delivery_status": delivery_status,
            "timestamp": datetime.now(UTC).isoformat(),
            "payload": payload,
        }
        # Backlog [403]: mirror payload.assistant_status onto the TOP LEVEL,
        # the same dual placement `_build_unavailable_reply` already uses —
        # see the docstring note above for why the top-level copy matters.
        if "assistant_status" in payload:
            packet["assistant_status"] = payload["assistant_status"]
        self._stamp_reply_to(packet, {"metadata": {"reply_to": reply_to}})
        return packet

    @traces(
        spec="001",
        story="US-002",
        reqs=[
            "REQ-001-US-002-001",
            "REQ-001-US-002-035",
            "REQ-001-US-002-038",
            "REQ-001-US-002-039",
        ],
    )
    def _build_failure_packet(
        self, thread_id: str, error: str, reply_to: str = ""
    ) -> dict[str, Any]:
        """Build the honest terminal FAILURE packet for a background round
        that raised (D3, closes G2). Thin wrapper over
        :meth:`_build_terminal_packet` (D-H5 extraction).
        """
        return self._build_terminal_packet(
            thread_id,
            "intent_failed",
            DeliveryStatus.FAILED,
            {
                "text": (
                    "[automated notice] the assistant failed to process this "
                    "request."
                ),
                "error": error,
                "failure": True,
                "assistant_status": "failed",
            },
            reply_to,
        )

    @traces(
        spec="001",
        story="US-002",
        reqs=[
            "REQ-001-US-002-033",
            "REQ-001-US-002-037",
            "REQ-001-US-002-038",
            "REQ-001-US-002-039",
        ],
    )
    @traces(
        spec="032",
        story="US-032-G243",
        reqs=["REQ-032-053", "REQ-032-054"],
    )
    def _build_no_reply_packet(
        self, thread_id: str, reply_to: str = ""
    ) -> dict[str, Any]:
        """Build the honest terminal NO-REPLY packet (D-H5, amends the
        parent design's D3): a background round that completed without
        producing a human-directed reply — governance blocked the
        ``submit_intent`` (``produced=False``), or the round's outcome never
        landed a matching packet in the human inbox — must still tell the
        sender something happened; the ack already promised a background
        round, so silence would be indistinguishable from loss.

        Thin wrapper over :meth:`_build_terminal_packet` (D-H5 extraction).
        Explicitly non-failure: the round COMPLETED, there is simply no
        reply (``delivery_status=DELIVERED``, ``assistant_status="no_reply"``,
        ``failure=False``) — G243-honest (SPEC-032): machine-detectable,
        never conversational, never a volk alias.
        """
        return self._build_terminal_packet(
            thread_id,
            "intent_no_reply",
            DeliveryStatus.DELIVERED,
            {
                "text": (
                    "[automated notice] the assistant completed processing "
                    "without a reply."
                ),
                "failure": False,
                "assistant_status": "no_reply",
            },
            reply_to,
        )

    @traces(
        spec="001",
        story="US-002",
        reqs=[
            "REQ-001-US-002-030",
            "REQ-001-US-002-035",
            "REQ-001-US-002-038",
            "REQ-001-US-002-039",
        ],
    )
    @traces(
        spec="002",
        story="US-001",
        reqs=["REQ-002-US-001-026"],
    )
    def _build_draining_packet(
        self, thread_id: str, reply_to: str = ""
    ) -> dict[str, Any]:
        """Build the SYNCHRONOUS ``KERNEL_DRAINING`` rejection terminal for a
        chat-subject packet that arrives after :meth:`shutdown` began
        draining (D-H11, F-3 hardening -- amends the parent design at
        ``plans/kernel-dispatch-reply-hardening.md`` §11.2).

        Unlike every other terminal this module builds, this one is never
        handed to the reply sink: it is returned directly from ``__call__``
        as the packet_handler result's ``reply_packet``, which the
        servicer's existing ``send_packet``/``receive_packet`` plumbing
        enqueues onto the still-open reply stream -- exactly the mechanism a
        synchronous governance denial already uses (see ``_govern``). Thin
        wrapper over :meth:`_build_terminal_packet` (D-H5 extraction).

        Explicit failure (``delivery_status=FAILED``, ``assistant_status=
        "draining"``, ``payload.error_code="KERNEL_DRAINING"``): the request
        was never accepted, so the caller's one clear move is to retry
        against the restarted kernel. *thread_id*/*reply_to* are the
        ORIGINAL values from the rejected packet (never minted), stamped
        through the shared :meth:`_stamp_reply_to` exactly like every other
        terminal here.
        """
        return self._build_terminal_packet(
            thread_id,
            "intent_draining",
            DeliveryStatus.FAILED,
            {
                "text": (
                    "[automated notice] the kernel is shutting down and "
                    "cannot accept new requests; retry after it restarts."
                ),
                "error": "kernel is draining (shutdown in progress)",
                "error_code": "KERNEL_DRAINING",
                "failure": True,
                "assistant_status": "draining",
            },
            reply_to,
        )

    @traces(
        spec="001",
        story="US-002",
        reqs=["REQ-001-US-002-035", "REQ-001-US-002-038", "REQ-001-US-002-039"],
    )
    async def shutdown(self, grace: float | None = None) -> None:
        """Graceful drain (D-H6, F5 fix): give every in-flight background
        round up to *grace* seconds to finish normally -- sinking its
        genuine reply or ``intent_no_reply`` terminal exactly as it would
        outside shutdown -- then cancel any stragglers, which trips the
        ``except asyncio.CancelledError`` arm in
        :meth:`_process_in_background`: an honest ``intent_failed`` "kernel
        shutting down" terminal is sunk before the cancellation propagates.

        Call this BEFORE stopping the transport (D-H7 wiring, not this
        module) so still-connected consumers can receive the final
        terminals. A no-op when nothing is in flight.

        *grace* defaults to ``KERNEL_BG_DRAIN_GRACE_S`` (env, default
        ``"30"`` -- a tunable with a permitted literal default, RUL-003's
        inverse: this is not an infra endpoint), resolved at call time (not
        import time) so tests and deploy tooling can override it without a
        code change. 30s covers the measured 11.5-21.1s round and sits
        inside systemd's stop window. An unparseable value (empty string,
        garbage) falls back to 30.0 with a warning instead of raising --
        a misconfigured env must not turn a graceful shutdown into a crash
        (F-1 hardening).

        Cancelled stragglers get a bounded 5s window to react to
        cancellation and sink their failure terminal; any still pending
        after that are abandoned with a warning rather than blocking
        shutdown indefinitely (F-5 hardening).

        FIRST action (D-H11, F-3 hardening): flip ``_draining`` True before
        touching ``_bg_tasks`` at all, so a packet accepted by ``__call__``
        right up to the instant the snapshot below is taken either lands in
        the snapshot (and gets drained/cancelled normally) or is rejected
        with a synchronous ``KERNEL_DRAINING`` receipt at the accept seam --
        never both silently missed.
        """
        self._draining = True
        if grace is None:
            raw = (os.environ.get("KERNEL_BG_DRAIN_GRACE_S", "30") or "30").strip()
            try:
                grace = float(raw)
            except ValueError:
                logger.warning(
                    "invalid KERNEL_BG_DRAIN_GRACE_S=%r; falling back to 30.0s",
                    raw,
                )
                grace = 30.0
        tasks = set(self._bg_tasks)
        if not tasks:
            return
        _, pending = await asyncio.wait(tasks, timeout=grace)
        if pending:
            for task in pending:
                task.cancel()
            _, still_pending = await asyncio.wait(pending, timeout=5.0)
            if still_pending:
                logger.warning(
                    "%d background round(s) did not react to cancellation "
                    "within 5.0s during shutdown; abandoning them",
                    len(still_pending),
                )

    async def __call__(self, packet: dict[str, Any]) -> dict[str, Any]:
        payload = packet.get("payload")
        text = payload.get("text", "") if isinstance(payload, dict) else ""
        subject = packet.get("subject", "")
        thread_id = packet.get("thread_id", "") or ""
        # P-B2: the sender-stamped consumer id (D6, ReplyRouter._pick_key's
        # first-precedence key). Propagated onto the terminal reply/failure
        # packet below so a multi-gateway deployment's reply routes back to
        # the exact consumer that sent this packet, not just "HUMAN". Absent
        # for unstamped/legacy callers -- reply_to stays "" and the terminal
        # packets carry no metadata key, byte-identical to pre-P-B2 output.
        metadata = packet.get("metadata")
        reply_to = (
            str(metadata["reply_to"])
            if isinstance(metadata, dict) and metadata.get("reply_to")
            else ""
        )

        memory_bound = False
        try:
            if subject == "status":
                return self._status_reply(packet, thread_id)

            denial = await self._govern(subject, thread_id, text, packet)
            if denial is not None:
                return denial

            if self._reply_sink is not None and subject in _CHAT_SUBJECTS:
                if self._draining:
                    # D-H11 (F-3 hardening): this is the single accept seam
                    # where a packet would enqueue background work. shutdown()
                    # has already begun draining -- its task snapshot was (or
                    # is about to be) taken and will never await a round
                    # spawned after that point, stranding its terminal behind
                    # a transport that's about to close. Do NOT spawn: return
                    # a synchronous rejection receipt instead, addressed with
                    # the ORIGINAL thread_id/reply_to so the still-open reply
                    # stream delivers it (same mechanism as a governance
                    # denial, see _govern). Non-spawning subjects (status,
                    # a denied/synchronous round) never reach this branch and
                    # pass through unchanged while draining.
                    draining_packet = self._build_draining_packet(thread_id, reply_to)
                    return {
                        "success": True,
                        "thread_id": thread_id or "",
                        "reply": "",
                        "reply_payload": draining_packet["payload"],
                        "reply_subject": draining_packet["subject"],
                        "reply_packet": draining_packet,
                        "accepted": False,
                    }
                # D1: ack immediately, run the round in the background, and
                # let _process_in_background publish the terminal packet
                # through the sink (D2/D3) instead of returning it inline.
                thread_id = thread_id or str(uuid.uuid4())
                task = asyncio.create_task(
                    self._process_in_background(subject, thread_id, text, reply_to)
                )
                self._bg_tasks.add(task)
                task.add_done_callback(self._bg_tasks.discard)
                return {
                    "success": True,
                    "thread_id": thread_id,
                    "reply": "",
                    "reply_payload": {},
                    "reply_subject": "",
                    "reply_packet": {},
                    "accepted": True,
                }

            memory_bound = await self._bind_memory(thread_id)

            thread_id, produced = await self._run_intent(subject, thread_id, text)

            reply_text: str = ""
            reply_payload: dict[str, Any] = {}
            reply_subject: str = ""
            reply_packet: dict[str, Any] = {}
            if produced:
                reply_text, reply_payload, reply_subject, reply_packet = (
                    await self._build_reply(text, thread_id, reply_to)
                )

            return {
                "success": True,
                "thread_id": thread_id or "",
                "reply": reply_text,
                "reply_payload": reply_payload,
                "reply_subject": reply_subject,
                "reply_packet": reply_packet,
            }
        except Exception as exc:  # surface engine errors as a failed receipt
            logger.exception("kernel packet_handler failed")
            return {"success": False, "error": str(exc)}
        finally:
            if memory_bound:
                unregister_memory(thread_id)


@traces(spec="001", story="US-002", reqs=["REQ-001-US-002-012"])
def build_kernel_packet_handler(
    kernel: PipelineRunner | None = None,
    memory_engine: Any | None = None,
    intent_broker: Any | None = None,
) -> _KernelPacketHandler:
    # Return type is the concrete class (not the bare-callable
    # KernelPacketHandler alias): boot sites need the full surface —
    # set_reply_sink (P-A4 wiring) — and no external caller imports the
    # alias, so narrowing to Callable only forced cast(Any) at both sites.
    """Return an async ``packet_handler(packet_dict) -> reply_dict`` that drives
    the volk engine.

    The engine (a booted ``PipelineRunner`` with all 9 volks) is constructed
    once. Pass *kernel* to reuse an already-booted runner (tests); otherwise a
    fresh one is booted here.

    The returned dict carries ``success`` (the handler ran without raising),
    ``thread_id`` (empty if the intent was blocked/not accepted), ``reply`` /
    ``reply_payload`` / ``reply_subject`` (the volk's reply, empty if none).

    When *memory_engine* (a :class:`~orchestrata_ultimate.memory.memory_engine.
    MemoryEngine`) is supplied, each accepted conversation turn — the human
    intent and the volk's reply — is persisted to long-term memory under the
    conversation's ``thread_id`` (``session_id``). This is where the kernel
    *uses* the memory engine: intents and replies become recallable rather than
    only living in the transient human-inbox. The async handler runs on one
    long-lived loop, so the engine is initialized once there and reused (no
    per-call ``asyncio.run`` and no cross-loop client reuse). Memory writes are
    auxiliary: a memory failure is logged and never breaks the orchestration
    reply.

    Limitation (single-flow): the reply is read from the engine's shared
    human-inbox; concurrent independent conversations through one kernel can
    interleave. Per-thread reply tracking is a later increment.
    """
    if kernel is None:
        from orchestrata_ultimate.kernel.pipeline import PipelineRunner

        kernel = PipelineRunner()
        kernel.boot()

    return _KernelPacketHandler(kernel, memory_engine, intent_broker)