orchestrata_ultimate.kernel.pipeline

Reference › orchestrata_ultimate › kernel › pipeline

The product-level flow of the kernel: how a human intent enters the system, travels through the volk agents and comes back. PipelineRunner boots the built-in volks, registers each with the postman, and drives conversations by creating CXP packets and routing them. The monolithic deployment in cli_main, the REPL in pipeline_cli and agents.kernel_handler all import it. The curated kernel page describes the pipeline; note that the class docstring and the volk table below name 11 built-in volks.

PipelineRunner is composed of four mixins: session, knowledge graph, audit and escalation. Its constructor creates the collaborators: a Postman (see the postman package), a PrologGate (see prolog_gate), a PipelineEventBus and PipelineCheckpoint sharing the postman's Redis client, an artifact store, and a bounded human_inbox deque (256 entries by default) that holds replies addressed to the human.

• boot validates configuration, loads the Prolog rules (the boot line reports permissive mode if loading fails), creates the workspace directory, starts session and audit services, calls _register_volks and _register_plugin_volks, syncs volk and component nodes to the knowledge graph and emits a boot event.
• _register_volks instantiates the volks under the aliases U:Cx, U:PM, U:Ux, U:Ar, U:Cd, J:Junior, U:Qa, U:BA, U:Cp, U:Do and U:SD, registers each process method with the postman, registers a HUMAN handler, and hands each volk the postman, runtime writer, context injector and optional health probe. _PluginVolkAdapter wraps plugin-registered handlers, sync or async, to look like a volk.
• submit_intent begins a session, builds a CXPPacket from HUMAN to U:Cx with subject intent, and asks the Prolog gate whether HUMAN may submit_intent; if not, it returns None. Otherwise it emits an intent event, binds a memory engine to the thread on a best-effort basis, syncs the intent to the graph, calls the postman's route_sync, and files a reply for the human in the inbox or routes an escalation reply on to the next volk. A clarification reply also asserts an open question in Prolog.
• respond finds the volk that last spoke on the thread, retracts the open question and routes a response packet to it with the prior human answers. resume reloads checkpointed stage outputs and re-dispatches from the stage after the last completed one. get_status, status_report, review_volk and get_last_human_message support the CLIs.

At a glance

Class diagram

Dependency diagram

Truncated at 25 nodes: showing 19 of 29 imports and 5 of 5 importers.

API

Classes

_PluginVolkAdapter

Line 58

Presents a plugin-registered Volk handler with the full volk interface the kernel expects.

PipelineRunner

Bases: PipelineSessionMixin, PipelineKGMixin, PipelineAuditMixin, PipelineEscalationMixin · Line 106

Production pipeline runner for multi-volk orchestration.

Source

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

# mypy: ignore-errors
"""PipelineRunner — boots the 11-volk pipeline and routes commands.

Static-boot pipeline engine for the out-of-process agent runtime and
monolithic boot; the gateway's live path runs
``orchestrator.pipeline_runner``. Composes session, knowledge-graph, and
escalation mixins, and routes CXP packets through the production
``Postman`` (sync router). Used by
``orchestrata_ultimate.kernel.pipeline_cli``.
"""

import inspect
import logging
import os
import threading
from collections import deque
from collections.abc import Awaitable, Callable, Mapping

from orchestrata_ultimate.agents.volks.analyst.agent import AnalystAgent
from orchestrata_ultimate.agents.volks.architect.volk import ArchitectVolk
from orchestrata_ultimate.agents.volks.base import BaseAgent
from orchestrata_ultimate.agents.volks.codedirector.volk import CodeDirectorVolk
from orchestrata_ultimate.agents.volks.compliance import ComplianceVolk
from orchestrata_ultimate.agents.volks.concierge.volk import ConciergeVolk
from orchestrata_ultimate.agents.volks.documentor.volk import DocumentorVolk
from orchestrata_ultimate.agents.volks.junior import JuniorVolk
from orchestrata_ultimate.agents.volks.pm.volk import PMVolk
from orchestrata_ultimate.agents.volks.qa.volk import QaVolk
from orchestrata_ultimate.agents.volks.senior_dev.agent import SeniorDevAgent
from orchestrata_ultimate.agents.volks.ux.volk import UXVolk
from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.infra_config import WORKSPACE_DIR, validate
from orchestrata_ultimate.core.packet import CXPPacket, WorkStatus
from orchestrata_ultimate.core.protocols.decoupling import PluginVolkRegistryProtocol
from orchestrata_ultimate.core.protocols.volk_health import VolkHealthProbe
from orchestrata_ultimate.governance.prolog_gate import PrologGate
from orchestrata_ultimate.kernel.pipeline_artifacts import ArtifactStore
from orchestrata_ultimate.kernel.pipeline_audit import PipelineAuditMixin
from orchestrata_ultimate.kernel.pipeline_checkpoint import PipelineCheckpoint
from orchestrata_ultimate.kernel.pipeline_escalation import PipelineEscalationMixin
from orchestrata_ultimate.kernel.pipeline_events import PipelineEventBus
from orchestrata_ultimate.kernel.pipeline_kg import PipelineKGMixin
from orchestrata_ultimate.kernel.pipeline_session import PipelineSessionMixin
from orchestrata_ultimate.memory.memory_context import (
    register_memory,
    unregister_memory,
)
from orchestrata_ultimate.postman.postman import Postman

# The volk roster mixes built-in BaseAgent subclasses with plugin-projected
# adapters (_PluginVolkAdapter, below) — both share the same process/
# set_postman/status/summary/role/current_thread/get_output_display surface,
# so this union (not `object`) types `self.volks` precisely enough for that
# shared surface to resolve without per-access `getattr`/`hasattr` guards.
type _Volk = BaseAgent | _PluginVolkAdapter


class _PluginVolkAdapter:
    """Presents a plugin-registered Volk handler with the full volk interface the
    kernel expects.

    Beyond ``process``/``set_postman``, it exposes the ``status``/``summary``/
    ``role``/``current_thread``/``get_output_display`` surface that the roster and
    status paths (``get_status``, ``status_report`` → the CXP GetStatus RPC,
    ``review_volk_output``) read — so a registered plugin volk never crashes those
    with a missing attribute. The plugin handler may be a sync or async callable
    taking a ``CXPPacket``; ``process`` awaits it when needed so the Postman
    dispatch treats it like any built-in volk.
    """

    def __init__(
        self,
        alias: str,
        handler: Callable[[CXPPacket], CXPPacket | None | Awaitable[CXPPacket | None]],
        capabilities: tuple[str, ...] = (),
    ) -> None:
        self.alias = alias
        self._handler = handler
        self.postman: object = None
        # Roster/status surface — plugin volks are stateless request handlers, so
        # they report a stable idle status rather than tracking per-thread work.
        self.role = "Plugin"
        self.status = WorkStatus.PENDING
        self.summary = "Plugin volk"
        self.current_thread: str | None = None
        # Declared capabilities let a plugin volk participate in capability-routed
        # hops (PipelineEscalationMixin.resolve_capability discovers them).
        self.capabilities = capabilities

    @traces(spec="015", story="US-015-002", reqs=["REQ-015-021", "REQ-015-022"])
    async def process(self, packet: CXPPacket) -> CXPPacket | None:
        result = self._handler(packet)
        if inspect.isawaitable(result):
            return await result
        return result

    @traces(spec="015", story="US-015-002", reqs=["REQ-015-018"])
    def set_postman(self, postman: object) -> None:
        self.postman = postman

    @traces(spec="000", story="US-006", reqs=["REQ-000-051"])
    def get_output_display(self) -> str:
        return f"Plugin volk {self.alias} (no captured output)."


class PipelineRunner(
    PipelineSessionMixin, PipelineKGMixin, PipelineAuditMixin, PipelineEscalationMixin
):
    # Mixin order matters for the type checker: PipelineAuditMixin's concrete
    # _audit_fire must precede PipelineEscalationMixin, whose TYPE_CHECKING
    # _EscalationHost declares the same method as a bodyless stub — with the
    # stub first in the MRO, pyright reports the class as abstract.
    """Production pipeline runner for multi-volk orchestration."""

    @traces(
        spec="001",
        story="US-001",
        reqs=["REQ-001-018", "REQ-001-015", "REQ-001-US-002-030"],
    )
    def __init__(
        self,
        plugin_volk_registry: PluginVolkRegistryProtocol | None = None,
        *,
        health_probe: VolkHealthProbe | None = None,
        human_inbox_max: int = 256,
    ) -> None:
        """
        Args:
            plugin_volk_registry: Optional plugin ``VolkRegistry`` whose
                registered Volks are projected onto the live dispatch at boot,
                so packets addressed to their aliases reach the plugin handlers
                exactly like the built-in volks. When omitted, only the 11
                built-in volks are dispatchable.
            health_probe: Optional ``VolkHealthProbe`` (e.g. ``VolkRegistry``)
                consulted by :meth:`status_report` to enrich each volk row with
                heartbeat health (REQ-001-018). When omitted, ``status_report``
                is byte-identical to its pre-heartbeat shape.
            human_inbox_max: Cap on ``human_inbox``'s size (D-H4,
                plans/kernel-dispatch-reply-hardening.md §3.2). A tunable, not
                an infra endpoint (RUL-003 inverse) — mirrors ReplyRouter's
                own drop-oldest 256. Steady-state size stays near zero once
                ``get_last_human_message(consume=True)`` claims each matched
                reply (D-H3); this cap only bounds unclaimed residue (e.g.
                session-mixin terminals like ``qa_complete``).
        """
        self.postman = Postman()
        self.prolog = PrologGate()
        self.events = PipelineEventBus(self.postman._redis)
        self._init_context_injector()
        # Queryable IN-MEMORY artifact collection, fed from stage-completion
        # hooks in the session/escalation mixins. The kernel engine serves
        # many projects from one long-lived process, so there is no single
        # sanctioned base dir for the store's JSONL layer here — durability
        # comes from PipelineKGMixin's graph sync and the on-demand
        # ``write_to_disk``; per-project durable stores are the orchestrator
        # engine's job (PipelineRunner(project_dir=...)).
        self.artifacts = ArtifactStore()
        # Per-stage output checkpoint store enabling resume (small state →
        # Redis, large blobs → disk + pointer). Reuses the postman's Redis.
        self.checkpoint = PipelineCheckpoint(redis_client=self.postman._redis)
        self.volks: dict[str, _Volk] = {}
        self.thread_ids: set = set()
        # Bounded deque (D-H4): append/reversed/[-1]/remove are all
        # deque-native, so every existing human_inbox usage form is
        # unchanged; only the drop-oldest cap and consume-on-match
        # (get_last_human_message) are new.
        self.human_inbox: deque[CXPPacket] = deque(maxlen=human_inbox_max)
        # Guards every human_inbox read-scan and mutation. deque iteration
        # RAISES RuntimeError on concurrent append, and deque.remove RAISES
        # IndexError when the deque mutates during its __eq__-dispatching
        # scan (CXPPacket.__eq__ is Python-level, so the scan yields) — both
        # reachable when a background round's consume races another round's
        # append (reviewer findings 1-2, 2026-08-05). One lock over the
        # three append sites and the snapshot+match+remove makes the
        # claim-check atomic; sections are tiny and never await.
        self._human_inbox_lock = threading.Lock()
        self.session = None
        self.audit = None
        self.booted = False
        self._plugin_volk_registry = plugin_volk_registry
        self._health_probe = health_probe

    @traces(spec="000", story="US-002", reqs=["REQ-000-009"])
    @traces(spec="001", story="US-008", reqs=["REQ-001-US-008-020"])
    def boot(self, env: Mapping[str, str] | None = None) -> bool:
        """Boot the kernel — load Prolog, register volks, connect services.

        Args:
            env: Injection seam for both the SPEC-001 US-008 audit-logger's
                enable/disable check and session-manager init (defaults to
                ``os.environ``; tests pass an explicit mapping instead of
                monkeypatching the process environment — see
                ``_init_audit_logger`` / ``_init_session_manager``).

        Returns:
            True when boot completes successfully.
        """
        print("[Kernel] Booting...")
        warnings = validate()
        for w in warnings:
            print(f"  [Config] WARNING: {w}")

        prolog_ok = self.prolog.load_rules()
        os.makedirs(WORKSPACE_DIR, exist_ok=True)
        print(f"  [Workspace] {WORKSPACE_DIR}")

        self._init_session_manager(env)
        self._init_audit_logger(env)
        self._start_audit_background()
        self._register_volks()
        self._register_plugin_volks()

        print(f"  [Volks] Registered: {', '.join(self.volks.keys())}")
        print(f"  [Prolog] {'Loaded' if prolog_ok else 'FAILED — permissive mode'}")
        print("[Kernel] Ready.")
        print()

        self.booted = True
        # Publish the standing roster into the KG so the semantic layer is
        # queryable with live Volk nodes (best-effort; no-op without the graph).
        synced = self.sync_volks_to_kg()
        if synced:
            print(f"  [KG Volks] {synced} volk nodes synced")
        components_synced = self.sync_components_to_kg()
        if components_synced:
            print(f"  [KG Components] {components_synced} component nodes synced")
        self.events.emit_boot(list(self.volks.keys()))
        return True

    def _register_volks(self) -> None:
        """Instantiate all volks and register them with Postman."""
        cx = ConciergeVolk()
        pm = PMVolk()
        ux = UXVolk()
        ar = ArchitectVolk()
        vz = CodeDirectorVolk()
        jr = JuniorVolk()
        # QaVolk's mixins (agents/volks/qa/_*.py) use the "mixin self-type"
        # trick: they inherit `_QaHost` (a package-local Protocol declaring
        # `_call_llm` as abstract) only under TYPE_CHECKING, purely so pyright
        # can resolve `self.<attr>` accesses inside the mixins. At runtime the
        # mixins inherit plain `object`, so QaVolk (which also inherits the
        # concrete `BaseAgent._call_llm`) is genuinely instantiable — proven
        # by every QaVolk() call across tests/agents/volks/qa/. Pyright's
        # static MRO still resolves `_call_llm` to the abstract Protocol stub
        # ahead of BaseAgent's concrete override, a documented false positive
        # of that typing-only trick (out of scope here: agents/volks/qa/*.py).
        qa = QaVolk()  # pyright: ignore[reportAbstractUsage]
        ba = AnalystAgent()
        cp = ComplianceVolk()
        do = DocumentorVolk()
        sd = SeniorDevAgent()

        self.volks = {
            "U:Cx": cx,
            "U:PM": pm,
            "U:Ux": ux,
            "U:Ar": ar,
            "U:Cd": vz,
            "J:Junior": jr,
            "U:Qa": qa,
            "U:BA": ba,
            "U:Cp": cp,
            "U:Do": do,
            "U:SD": sd,
        }
        for alias, volk in self.volks.items():
            self.postman.register(alias, volk.process)
        self.postman.register("HUMAN", self._human_handler)
        for volk in self.volks.values():
            volk.set_postman(self.postman)
            # set_runtime_writer/set_context_injector are BaseAgent-only hooks
            # (_PluginVolkAdapter is a deliberately thinner surface — plugin
            # volks don't participate in KG runtime-signal writing or the
            # validator context-injection loop), so narrow with isinstance
            # rather than hasattr (which doesn't narrow union membership).
            if isinstance(volk, BaseAgent):
                volk.set_runtime_writer(getattr(self, "_runtime_writer", None))
                volk.set_context_injector(getattr(self, "context_injector", None))
                # G5 (plans/volk-comms-heartbeat-design.md §4.1 decision 7):
                # mirrors orchestrator/pipeline_runner.py:786 -- bind the
                # injected health probe into every rostered volk so its tool
                # calls (message_volk health pre-check, get_volk_status) can
                # query peer health, exactly like the orchestrator's live
                # path already does. Only when a probe was actually injected
                # (self._health_probe is Optional) -- previously the kernel
                # path never bound a registry into its volks at all.
                if self._health_probe is not None:
                    volk.set_volk_registry(self._health_probe)

    @traces(
        spec="015",
        story="US-015-002",
        reqs=["REQ-015-017", "REQ-015-021"],
    )
    def _register_plugin_volks(self) -> None:
        """Project plugin-registered Volks onto the live dispatch.

        Iterates the injected plugin ``VolkRegistry`` and registers each Volk's
        handler with the same ``Postman`` map the built-in volks use, wrapped so
        it presents the volk interface (``process``/``set_postman``). After this,
        a CXP packet addressed to a plugin Volk's alias is dispatched to the
        plugin handler — the SPEC-015 registration becomes operational, not just
        recorded.
        """
        registry = self._plugin_volk_registry
        if registry is None:
            return
        for alias in registry.list_volks():
            rv = registry.get_volk(alias)
            if rv is None:
                continue
            adapter = _PluginVolkAdapter(
                alias, rv.handler, capabilities=tuple(getattr(rv, "capabilities", ()))
            )
            adapter.set_postman(self.postman)
            self.volks[alias] = adapter
            self.postman.register(alias, adapter.process)

    def _bind_memory_engine(self, thread_id: str) -> bool:
        """Best-effort: bind a memory engine for this thread so volks recall
        prior context (mirrors the orchestrator path). Returns False and no-ops
        when memory or its backing store is unavailable — the kernel pipeline
        never breaks on it.

        The orchestrator path is async and calls ``MemoryEngine.initialize`` in
        the runner's event loop; the kernel path is synchronous, so this helper
        bridges the async initialization using the same thread-pool pattern
        ``postman.sync_router`` uses for async handlers.
        """
        try:
            from orchestrata_ultimate.agents.volks.memory_context import _run_async_sync
            from orchestrata_ultimate.memory.memory_engine import MemoryEngine

            engine = MemoryEngine()
            _run_async_sync(engine.initialize({}))
            register_memory(thread_id, engine)
            return True
        except Exception as exc:
            logging.getLogger(__name__).debug(
                "memory engine unavailable for kernel thread %s: %s", thread_id, exc
            )
            return False

    @traces(
        spec="000",
        story="US-004",
        reqs=[
            "REQ-000-033",
            "REQ-000-035",
            "REQ-000-036",
            "REQ-000-037",
            "REQ-000-041",
        ],
    )
    def submit_intent(self, text: str, thread_id: str | None = None) -> str | None:
        """Submit an intent — returns thread_id.

        Args:
            text: Human intent text.
            thread_id: Optional caller-supplied conversation id. When provided
                (e.g. the gateway dispatching a UI intent over gRPC) it is
                honoured so the volk's reply correlates to the caller's
                conversation and follow-up ``respond(thread_id, ...)`` resolves
                the same thread. When omitted a fresh id is minted.

        Returns:
            The thread ID for the new conversation, or None if blocked.
        """
        if self.session:
            try:
                self.session.begin(purpose=text)
                print("  [Session] Started", flush=True)
            except Exception as exc:
                print(f"  [Session] Begin failed: {exc}", flush=True)

        packet = CXPPacket(
            from_agent="HUMAN",
            to_agent="U:Cx",
            subject="intent",
            payload={"text": text},
            # CXPPacket.thread_id is `str = ""` (falsy-checked, not Optional);
            # `or ""` preserves the existing "mint a fresh id when omitted"
            # behaviour (`"" or uuid4()` == `None or uuid4()` at runtime).
            thread_id=thread_id or "",
        )
        thread_id = packet.thread_id

        allowed, reason = self.prolog.may_execute("HUMAN", "submit_intent")
        print(f"  [Prolog] may_execute(HUMAN, submit_intent): {reason}")
        if not allowed:
            print(f"  [BLOCKED] {reason}")
            return None

        self.events.emit_intent(thread_id, text)
        # Bind a memory engine for this thread so volks recall prior context on
        # the kernel path too (mirrors the orchestrator path). Best-effort/no-op
        # when Redis/graph is unavailable. Round-scoped: released in the
        # `finally` below (P-H2, plans/kernel-dispatch-reply-hardening.md D-H2)
        # so this round's bind is always balanced — under the refcounted
        # registry (P-H1, memory/memory_context.py) an unreleased bind here
        # would leak a permanent +1 count on the thread, and a same-thread
        # concurrent round's own binding must survive this round finishing
        # first.
        memory_bound = self._bind_memory_engine(thread_id)
        try:
            # Record the intent + its pipeline run in the semantic graph so the
            # provenance chain (intent → run → volk → artifact) is queryable. The
            # run's EXECUTED_BY / artifact GENERATED_BY edges are added as stages
            # produce artifacts (sync_artifacts_to_kg). Best-effort — never blocks.
            self.sync_intent_to_kg(thread_id, text)
            response = self.postman.route_sync(packet)
            self.thread_ids.add(thread_id)

            if response and response.to_agent == "HUMAN":
                # F8 fix: the very first hop can itself be a clarification
                # request (e.g. ConciergeVolk asking a follow-up question)
                # before any escalation routing ever runs — mark the open
                # question here too, same as _route_escalation does for every
                # later hop, so may_complete's no_open_questions gate sees it.
                if response.subject == "clarification":
                    try:
                        self.prolog.assert_open_question(
                            response.from_agent, f"task_{thread_id}"
                        )
                    except Exception as exc:
                        print(f"  [Prolog] assert_open_question failed: {exc}")
                with self._human_inbox_lock:
                    self.human_inbox.append(response)
            if (
                response
                and response.to_agent in self.volks
                and response.to_agent != "HUMAN"
            ):
                self._route_escalation(response)

            return thread_id
        finally:
            if memory_bound:
                unregister_memory(thread_id)

    @traces(
        spec="000",
        story="US-004",
        reqs=["REQ-000-039", "REQ-000-040"],
    )
    def respond(self, thread_id: str, text: str) -> None:
        """Human responds to a volk on a thread.

        Args:
            thread_id: The active thread identifier.
            text: The human's response text.
        """
        if thread_id not in self.thread_ids:
            print(f"  [Error] Thread {thread_id} not found")
            return

        thread_packets = self.postman.get_thread_packets(thread_id)
        target = "U:Cx"
        if thread_packets:
            last_packet = thread_packets[-1]
            target = last_packet.from_agent
            if target == "HUMAN":
                for p in reversed(thread_packets):
                    if p.from_agent != "HUMAN":
                        target = p.from_agent
                        break

        prior_answers = [
            p.payload.get("text", "")
            for p in thread_packets
            if p.from_agent == "HUMAN" and p.subject == "response"
        ]

        # F8 fix: the human answering resolves whatever open question
        # `target` raised (assert_open_question, set when `target` sent a
        # "clarification" packet — see submit_intent/_route_escalation).
        # Without this, an open question could be asserted but never
        # retracted, permanently denying may_complete for this task —
        # the mirror-image bug of the self-fulfilling one this fix removes.
        try:
            self.prolog.retract_open_question(target, f"task_{thread_id}")
        except Exception as exc:
            print(f"  [Prolog] retract_open_question failed: {exc}")

        packet = CXPPacket(
            from_agent="HUMAN",
            to_agent=target,
            thread_id=thread_id,
            subject="response",
            payload={"text": text, "prior_answers": prior_answers},
        )
        response = self.postman.route_sync(packet)
        if response:
            if response.to_agent == "HUMAN":
                with self._human_inbox_lock:
                    self.human_inbox.append(response)
            elif response.to_agent in self.volks:
                self._route_escalation(response)

    @traces(spec="000", story="US-002", reqs=["REQ-000-023"])
    def get_status(self) -> list[str]:
        """Return status lines for all registered volks.

        Returns:
            List of status strings, one per volk.
        """
        return [
            f"[{volk.status.value.upper()}] {alias}: {volk.summary}"
            for alias, volk in self.volks.items()
        ]

    @traces(spec="000", story="US-002", reqs=["REQ-000-023"])
    @traces(spec="001", story="US-001", reqs=["REQ-001-018", "REQ-001-015"])
    def status_report(self) -> list[dict[str, object]]:
        """Structured live status for every registered volk.

        The machine-readable counterpart to :meth:`get_status` — this is what the
        orchestrator/gateway query to know each volk's status at any point in
        time (``busy`` is derived from the volk holding a ``current_thread``).

        When a health probe is wired in (``self._health_probe``, REQ-001-015),
        each row is additionally enriched with heartbeat health — ``healthy``,
        ``heartbeat_age_s``, ``last_heartbeat`` (REQ-001-018). A failure
        enriching any one row is swallowed so it never breaks the report
        (REQ-001-017); that row simply omits the three health keys. With no
        probe injected, rows are byte-identical to the pre-heartbeat shape.
        """
        report: list[dict[str, object]] = []
        for alias, volk in self.volks.items():
            current_thread = getattr(volk, "current_thread", None)
            entry: dict[str, object] = {
                "alias": alias,
                "role": getattr(volk, "role", ""),
                "status": volk.status.value,
                "summary": volk.summary,
                "current_thread": current_thread,
                "busy": current_thread is not None,
            }
            if self._health_probe is not None:
                try:
                    health = self._health_probe.get_volk_health(alias)
                    entry["healthy"] = (
                        health.is_healthy if health is not None else False
                    )
                    entry["heartbeat_age_s"] = (
                        health.heartbeat_age_seconds if health is not None else None
                    )
                    entry["last_heartbeat"] = (
                        health.last_heartbeat.isoformat()
                        if health is not None and health.last_heartbeat is not None
                        else None
                    )
                except Exception:
                    # REQ-001-017: a probe failure for one volk must never
                    # break the roster report for the rest.
                    logging.getLogger(__name__).warning(
                        f"Health probe failed for volk {alias!r}; omitting "
                        "heartbeat health from its status_report row",
                        exc_info=True,
                    )
            report.append(entry)
        return report

    @traces(spec="000", story="US-006", reqs=["REQ-000-051", "REQ-000-052"])
    def review_volk(self, alias: str) -> str:
        """Get a volk's output for human review.

        Args:
            alias: The volk alias to review.

        Returns:
            Display string of the volk's output, or error message.
        """
        if alias not in self.volks:
            return f"Unknown volk: {alias}"
        # self.volks holds both BaseAgent volks and plugin adapters; only the
        # former expose get_output_display, so guard rather than assume.
        display = getattr(self.volks[alias], "get_output_display", None)
        if not callable(display):
            return f"Volk {alias} has no reviewable output."
        return str(display())

    @traces(spec="001", reqs=["REQ-001-US-002-033", "REQ-001-US-002-039"])
    def get_last_human_message(
        self, thread_id: str | None = None, consume: bool = False
    ) -> CXPPacket | None:
        """Get the most recent message addressed to the human.

        Args:
            thread_id: When given, scan the inbox in reverse and return the
                newest packet whose ``thread_id`` matches; None if no packet
                matches. When omitted, behavior is unchanged: the last
                packet in the inbox (or None if empty).
            consume: When True and a packet matches, remove it from
                ``human_inbox`` (D-H3, claim-check pattern,
                plans/kernel-dispatch-reply-hardening.md §3.2) so a later
                round on the same thread can never re-match — and re-emit as
                its own terminal — a reply an earlier round already claimed.
                Default False keeps every other caller (peeks) byte-identical
                (REQ-000-039/040).

        Returns:
            The matching CXPPacket, or None.
        """
        # The whole scan+match+remove runs under _human_inbox_lock, shared
        # with every append site: deque iteration raises RuntimeError on a
        # concurrent append, and deque.remove raises IndexError when the
        # deque mutates during its Python-__eq__ scan, so an unlocked
        # "snapshot" is not a fix — it is the crash site. Under the lock the
        # claim-check is atomic: a matched packet is still present when
        # removed, and two rounds can never claim the same reply.
        with self._human_inbox_lock:
            if thread_id is None:
                match = self.human_inbox[-1] if self.human_inbox else None
            else:
                match = None
                for packet in reversed(self.human_inbox):
                    if packet.thread_id == thread_id:
                        match = packet
                        break
            if match is not None and consume:
                self.human_inbox.remove(match)
        return match

    @traces(spec="003", story="FR-024", reqs=["REQ-003-FR24-003"])
    def resume(self, thread_id: str) -> str | None:
        """Resume a previously-checkpointed pipeline thread.

        LLM stages are not replayable, so resume reuses the outputs already
        checkpointed for completed stages and re-dispatches from the stage that
        follows the last completed one — injecting the prior outputs as the
        escalation payload so the downstream volk receives the real upstream
        work instead of re-running it.

        Args:
            thread_id: The thread to resume.

        Returns:
            The thread id when a resume was dispatched, or None when there is
            nothing to resume (no checkpointed stages).
        """
        state = self.checkpoint.resume_state(thread_id)
        if not state:
            print(f"  [Resume] No checkpoint for thread {thread_id}", flush=True)
            return None

        self.thread_ids.add(thread_id)
        last_stage, last_output = self._latest_checkpoint(state)
        next_target = self._resume_next_target(last_stage)

        print(
            f"  [Resume] thread={thread_id} reusing {len(state)} stage(s); "
            f"last={last_stage} → {next_target}",
            flush=True,
        )
        self.events.emit_intent(thread_id, f"resume after {last_stage}")

        if next_target is None or next_target not in self.volks:
            # Nothing left to run downstream — the pipeline was already complete.
            return thread_id

        payload = dict(last_output)
        payload["resumed_from"] = last_stage
        packet = CXPPacket(
            from_agent=last_stage,
            to_agent=next_target,
            thread_id=thread_id,
            subject=payload.get("subject", "implement"),
            payload=payload,
        )
        self._route_escalation(packet)
        return thread_id

    # Choreographed stage order used only to pick the next stage on resume.
    _RESUME_ORDER = ["U:Cx", "U:BA", "U:PM", "U:Ar", "U:Cd", "U:Qa", "U:Cp"]

    @traces(spec="003", story="FR-024", reqs=["REQ-003-FR24-003"])
    def _latest_checkpoint(self, state: dict[str, dict]) -> tuple[str, dict]:
        """Return the (stage, output) of the furthest-along completed stage."""
        ordered = [s for s in self._RESUME_ORDER if s in state]
        last_stage = ordered[-1] if ordered else next(iter(state))
        return last_stage, state[last_stage]

    @traces(spec="003", story="FR-024", reqs=["REQ-003-FR24-003"])
    def _resume_next_target(self, last_stage: str) -> str | None:
        """Return the stage that follows *last_stage* in the pipeline order."""
        if last_stage not in self._RESUME_ORDER:
            return None
        idx = self._RESUME_ORDER.index(last_stage)
        if idx + 1 >= len(self._RESUME_ORDER):
            return None
        return self._RESUME_ORDER[idx + 1]