orchestrata_ultimate.orchestrator.pipeline_runner

Reference › orchestrata_ultimate › orchestrator › pipeline_runner

PipelineRunner is where accepted work is actually carried out: it takes a human intent and a WorkType, asks pipeline_router for the stage sequence, and runs each volk stage by routing a CXP packet through Postman, so every message is persisted, deduplicated and visible in the CXP feed. It is the largest module in the package and the hub of the pipeline flow described in the orchestrator package narrative. It is driven by pipeline_kernel for the gateway, by session.cli.pipeline, the cheetahclaws integration and several scripts, and is extended by pipeline_escalation and verification_gate.

The class mixes in PipelineKGMixin (knowledge-graph context at stage boundaries and best-effort sync of the intent, artifacts and decisions when a run ends) and EscalationInterruptMixin. Its constructor builds a router, an ArtifactStore (durable under <project>/pipeline_output/artifacts when a project directory is given, in memory otherwise), a volk registry, a Postman instance, a checkpoint store that keeps small state in Redis and large blobs on disk (None when Redis is unavailable, which degrades resume to nothing to resume), a Prolog governance gate and a verification gate. The retry backoff schedule can be overridden with the environment variable PIPELINE_RETRY_BACKOFF_S.

Main flow:

• run generates a thread id, asks the governance gate whether the human may submit an intent, routes the work type, registers the stage volks as Postman handlers, binds a memory engine, opens a verification session and calls _execute_stages inside a pipeline trace. It then attaches the final verification verdict and syncs the run to the graph.
• _execute_stages loops over the stages, checks governance before each, calls _run_stage, collects artifacts, and asks the escalation mixin whether the response means to continue, insert a provider stage for a requested capability, or suspend the run for a human.
• _run_stage builds the packet (the first stage from the intent text and optional project directory, later stages chained from the previous output), routes it with retry and exponential backoff, treats import and syntax errors as non-retryable, saves a checkpoint after each success, and collects artifacts by ArtifactType.
• resume rebuilds the completed stages from checkpoints and continues from the first incomplete one.

Results are StageResult and PipelineResult (with successful()), and StageStatus records each state. A governance block, a failed stage or a failed run marks the result failed rather than raising.

At a glance

Class diagram

Dependency diagram

Truncated at 25 nodes: showing 16 of 17 imports and 8 of 8 importers.

API

Constants

Classes

StageStatus (enum)

Bases: StrEnum · Line 84

StageResult (dataclass)

Decorators: dataclass · Line 97

PipelineResult (dataclass)

Decorators: dataclass · Line 109

PipelineRunner

Bases: PipelineKGMixin, EscalationInterruptMixin · Decorators: traces(spec='003', story='US-004', reqs=['REQ-003-US-004-020']) · Line 175

Runs an agent pipeline through Postman routing.

Functions

Source

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

"""PipelineRunner — executes dynamic agent stages through Postman routing.

Uses PipelineRouter to map work type → agent sequence, then routes
each stage through Postman for Redis persistence, deterministic IDs,
deduplication, CXP auditing, and message logging. All stages end with
Compliance (U:Cp) for traceability.

Communication model:
    PipelineRunner → Postman.route_sync(packet) → handler → agent.process()
                                                      │
                                                      ├── Redis persistence
                                                      ├── Deterministic message IDs
                                                      ├── Deduplication
                                                      ├── CXPAuditor logging
                                                      ├── Thread audit trail
                                                      └── Message log (CXPFeed)

W3.1 absorption (plans/decisions-2026-07-06.md D1-addendum): this engine
(the live gateway/session/TUI path) absorbs the kernel engine's checkpoint
persistence and KG-sync capabilities ahead of the kernel engine's eventual
retirement — checkpoint/KG-sync are cross-package imports from
``orchestrata_ultimate.kernel`` (kernel/orchestrator are layer-4 peers, no
cycle: kernel pipeline modules import nothing from orchestrator).

D1-addendum-2 (plans/design-escalation-absorption-2026-07-06.md): escalation
also absorbs here, reframed from the kernel's free-hop packet routing into a
checkpoint-backed INTERRUPT on this engine's fixed sequence — a stage that
signals it needs human input SUSPENDS the run (``StageStatus.ESCALATED``);
``EscalationInterruptMixin.approve``/``reject``/``respond`` (wired to the
session CLI's ``pipeline approve``/``reject``/``respond``) resolve it. See
``orchestrator.pipeline_escalation`` for the full mechanism; the kernel
engine's own mixin is untouched and keeps serving ``pipeline_cli``'s REPL
until that engine's own retirement.
"""

from __future__ import annotations

import asyncio
import logging
import os
import uuid
from collections.abc import Mapping
from dataclasses import dataclass, field
from datetime import UTC, datetime
from enum import StrEnum
from pathlib import Path
from typing import TYPE_CHECKING

from orchestrata_ultimate.agents.volk_registry import (
    VolkCapability,
    VolkDefinition,
    VolkRegistry,
)
from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.packet import CXPPacket, WorkStatus
from orchestrata_ultimate.governance.prolog_gate import PrologGate
from orchestrata_ultimate.kernel.pipeline_kg import PipelineKGMixin
from orchestrata_ultimate.orchestrator.artifact_store import (
    ArtifactStore,
    ArtifactType,
    PipelineArtifact,
)
from orchestrata_ultimate.orchestrator.pipeline_escalation import (
    EscalationInterruptMixin,
)
from orchestrata_ultimate.orchestrator.pipeline_router import (
    PipelineRouter,
    StageSpec,
    WorkType,
)
from orchestrata_ultimate.orchestrator.verification_gate import (
    PipelineVerificationGate,
    VerificationGateConfig,
)

if TYPE_CHECKING:
    from orchestrata_ultimate.kernel.pipeline_checkpoint import PipelineCheckpoint
    from orchestrata_ultimate.postman.postman import Postman
    from orchestrata_ultimate.session.controller._core import SessionController

# ── PipelineResult ────────────────────────────────────────────────────


class StageStatus(StrEnum):
    PENDING = "pending"
    IN_PROGRESS = "in_progress"
    RESOLVED = "resolved"
    FAILED = "failed"
    TIMED_OUT = "timed_out"
    SKIPPED = "skipped"
    # D1-addendum-2 (escalation-as-interrupt): a stage asked for human input
    # and the run suspended — see EscalationInterruptMixin.
    ESCALATED = "escalated"


@dataclass
class StageResult:
    stage_alias: str
    status: StageStatus = StageStatus.PENDING
    output_packet: CXPPacket | None = None
    artifacts: list[PipelineArtifact] = field(default_factory=list)
    duration_ms: float = 0.0
    error: str | None = None
    retries: int = 0
    retry_backoff_s: float = 0.0


@dataclass
class PipelineResult:
    thread_id: str
    work_type: str
    intent: str
    stages: dict[str, StageResult] = field(default_factory=dict)
    artifacts: list[PipelineArtifact] = field(default_factory=list)
    started_at: datetime = field(default_factory=lambda: datetime.now(UTC))
    completed_at: datetime | None = None
    overall_status: StageStatus = StageStatus.PENDING
    # SPEC-035 wiring: the verification gate's final-checkpoint verdict
    # (VerificationVerdict.to_dict()), or None when verification is
    # disabled/unavailable for this run. AUDIT-ONLY by default — this
    # field is populated for observability regardless of enforcement.
    verification: dict | None = None

    @traces(spec="003", story="FR-024", reqs=["REQ-003-FR24-001"])
    def successful(self) -> bool:
        return self.overall_status == StageStatus.RESOLVED


# ── PipelineRunner ────────────────────────────────────────────────────

# First-stage subjects per capability (when there's no previous agent)
_CAPABILITY_INTAKE_SUBJECT: dict[str, str] = {
    "intake": "intent",
    "testing": "review_and_test",
    "analysis": "analyze_requirements",
    # These must match CAPABILITY_SUBJECTS in pipeline_router.py — they are
    # the same handoff vocabulary seen from the first stage instead of a
    # mid-pipeline edge, and they carried the same four wrong strings
    # ("design_architecture", "code_implementation", "scope_stories",
    # "design_ux"), which no volk answers. Corrected together on 2026-08-29;
    # test_every_routed_subject_is_accepted_by_its_volk covers both tables.
    "architecture": "scope_validated",
    "implementation": "implement",
    "documentation": "generate_docs",
    "traceability": "compliance_check",
    "scoping": "scope_review",
    "ui_design": "ux_design",
    "parallel": "implement",
}

_logger = logging.getLogger(__name__)

# Retry configuration
MAX_RETRIES = 3
RETRY_BACKOFF_SECONDS = [5.0, 10.0, 20.0]


def _resolve_retry_backoff(env: Mapping[str, str]) -> list[float]:
    """Per-instance retry-backoff schedule, overridable via ``env``.

    Reads ``PIPELINE_RETRY_BACKOFF_S`` (comma-separated seconds) from the
    SAME injected ``env`` mapping the constructor already threads through
    (see ``PipelineRunner.__init__``'s ``env`` param) so tests can shrink
    the real ``asyncio.sleep(backoff)`` wall-clock without mutating process
    ``os.environ`` or monkeypatching. Defaults to the unchanged production
    schedule when unset.
    """
    raw = env.get("PIPELINE_RETRY_BACKOFF_S")
    if not raw:
        return RETRY_BACKOFF_SECONDS
    return [float(v) for v in raw.split(",") if v.strip()]


@traces(spec="003", story="US-004", reqs=["REQ-003-US-004-020"])
class PipelineRunner(PipelineKGMixin, EscalationInterruptMixin):
    """Runs an agent pipeline through Postman routing.

    Uses PipelineRouter to determine which agents to run and in what order.
    Routes every message through Postman for persistence, auditing,
    and CXPFeed visibility. All stages end with Compliance (U:Cp).

    Mixes in ``PipelineKGMixin`` (W3.1 absorption) for Neo4j knowledge-graph
    provenance — context injection at stage boundaries and best-effort
    intent/artifact/decision sync at run completion — and
    ``EscalationInterruptMixin`` (W3.1 absorption, D1-addendum-2) for the
    checkpoint-backed escalation interrupt: ``approve``/``reject``/
    ``respond`` resolve a suspended (ESCALATED) run.
    """

    def __init__(
        self,
        artifact_store: ArtifactStore | None = None,
        session_controller: SessionController | None = None,
        env: Mapping[str, str] | None = None,
        project_dir: str | None = None,
        verification_gate: PipelineVerificationGate | None = None,
        volk_registry: VolkRegistry | None = None,
    ):
        # Target project root, propagated into the FIRST stage's payload
        # under the established "project_dir" contract key (the QA and
        # compliance volks already honor it on chained stages). Without it
        # those volks fell back to CWD-relative "src"/"tests": the
        # compliance trace scan silently analyzed whatever directory the
        # process happened to run in instead of the pipeline's target
        # project (measured: a review run inside this repo spent ~35s
        # scanning all 122K LOC of the WRONG tree).
        self._project_dir = project_dir
        self.router = PipelineRouter()
        # Durable by default when the target project is known: the store's
        # JSONL layer (``<project>/pipeline_output/artifacts``) is what makes
        # a run's work product retrievable after the process exits. Without a
        # project_dir there is no sanctioned base dir, so the store stays
        # in-memory (bare test constructions must not cross-contaminate via
        # jsonl replay).
        self.artifact_store = (
            artifact_store
            if artifact_store is not None
            else ArtifactStore(
                output_dir=(
                    Path(project_dir) / "pipeline_output" / "artifacts"
                    if project_dir
                    else None
                )
            )
        )
        # Orchestrator↔volk heartbeat awareness: a registry that tracks the
        # pipeline's live volks. Injected for tests; built lazily otherwise.
        self.volk_registry = volk_registry or VolkRegistry()
        # PipelineKGMixin.sync_artifacts_to_kg reads self.artifacts (the
        # kernel's attribute name) — alias so both names see the same store.
        self.artifacts = self.artifact_store
        self._postman = self._create_postman()
        self._session_ctrl = session_controller
        self._session_log: list[str] = []
        # Volk roster (alias -> instantiated agent), populated as stages are
        # registered — lets PipelineKGMixin.sync_artifacts_to_kg attribute an
        # artifact to a known volk (GENERATED_BY edge), mirroring the
        # kernel's self.volks.
        self.volks: dict[str, object] = {}
        # Neo4j context injector (W3.1 KG absorption) — non-fatal; degrades
        # to injection-disabled when the graph is unavailable.
        self._init_context_injector()
        # Per-stage output checkpoint store enabling a real resume() (backlog
        # #38, W3.1 absorption): small state in Redis, large blobs to disk
        # (kernel.pipeline_checkpoint.PipelineCheckpoint, zero kernel-engine
        # coupling). Best-effort: None when Redis is unavailable, degrading
        # resume() to "nothing to resume" rather than crashing.
        # ``env`` is exposed so tests can exercise misconfiguration without
        # mutating the process environment.
        self._env = os.environ if env is None else env
        self._retry_backoff_s = _resolve_retry_backoff(self._env)
        self.checkpoint = self._create_checkpoint(self._env)
        # P0.1: Governance enforcement — PrologGate validates agent actions
        # against PAM++ policy rules before execution. Non-fatal: degrades
        # to permissive mode when Prolog rules fail to load.
        self.prolog = PrologGate()
        self.prolog.load_rules()
        # Feed the gate its own facts. `may_execute(alias, capability)` in
        # _execute_stages denies any agent the rule base has no volk_exists/1
        # fact for (the permissive catch-all was deleted 2026-08-10), and
        # _register_handlers registers every stage volk here — so this is the
        # wire that turns "registered with the runner" into "known to
        # governance". Set after load_rules so the facts land in a consulted
        # rule base, and applied to an INJECTED registry too (set_governance
        # backfills whatever is already in it).
        self.volk_registry.set_governance(self.prolog)
        # SPEC-035: verification-subsystem checkpoint gate — same non-fatal
        # construction pattern as PrologGate above. AUDIT-ONLY by default
        # (VERIFY_ENFORCE=1 opts into failing/escalating a run); a
        # construction failure degrades to a disabled gate rather than
        # crashing the runner.
        try:
            self.verification_gate = verification_gate or PipelineVerificationGate(
                env=self._env
            )
        except Exception as exc:
            _logger.warning(
                "PipelineVerificationGate construction failed (%s) — "
                "verification disabled for this runner",
                exc,
            )
            self.verification_gate = PipelineVerificationGate(
                config=VerificationGateConfig(
                    enabled=False, enforce=False, max_attempts=5
                )
            )

    def _create_postman(self) -> Postman:
        """Create Postman instance with Redis persistence."""
        from orchestrata_ultimate.postman.postman import Postman

        return Postman(config={"enable_cxp_audit": True})

    @traces(spec="003", story="FR-024", reqs=["REQ-003-FR24-002", "REQ-003-FR24-003"])
    def _create_checkpoint(
        self, env: Mapping[str, str] | None = None
    ) -> PipelineCheckpoint | None:
        """Best-effort real checkpoint store construction.

        Mirrors ``session/cli/pipeline.py``'s ``_get_checkpoint_store()`` — a
        real sync Redis client backing ``kernel.pipeline_checkpoint.
        PipelineCheckpoint`` (cross-package import; kernel/orchestrator are
        layer-4 peers, see plans/decisions-2026-07-06.md D1-addendum).
        Returns None on any failure (misconfigured Redis env vars, a
        transient connection issue) so the pipeline still runs without
        persistence rather than crash — the same non-fatal pattern as
        ``_init_context_injector``.

        Args:
            env: Optional environment mapping (defaults to ``os.environ``).
                Exposed so tests can exercise misconfiguration without
                mutating the process environment.
        """
        try:
            from orchestrata_ultimate.core.redis_factory import create_sync_redis
            from orchestrata_ultimate.kernel.pipeline_checkpoint import (
                PipelineCheckpoint,
            )

            return PipelineCheckpoint(redis_client=create_sync_redis(env=env))
        except Exception as exc:
            _logger.warning("Checkpoint store unavailable (%s) — resume disabled", exc)
            return None

    @traces(spec="000", reqs=["REQ-000-058"])
    @traces(spec="003", story="US-004", reqs=["REQ-003-US-004-021"])
    async def run(
        self,
        intent: str,
        work_type: WorkType | str = "feature",
        thread_id: str | None = None,
        has_ui: bool = False,
    ) -> PipelineResult:
        """Execute pipeline for a work type through Postman.

        Agents are registered as Postman handlers. Every message is
        routed through Postman.route_sync() for Redis persistence,
        deterministic IDs, deduplication, and CXP auditing.

        Args:
            intent: Human intent text.
            work_type: WorkType enum or string.
            thread_id: Optional thread ID.
            has_ui: Whether UI keywords were detected.

        Returns:
            PipelineResult with stage results and artifacts.
        """
        thread_id = thread_id or f"pipeline_{uuid.uuid4().hex[:8]}"

        result = PipelineResult(
            thread_id=thread_id,
            work_type=str(work_type),
            intent=intent,
        )

        # P0.1: Governance check before intent dispatch
        allowed, reason = self.prolog.may_execute("HUMAN", "submit_intent")
        if not allowed:
            _logger.warning("Governance blocked intent submission: %s", reason)
            result.overall_status = StageStatus.FAILED
            result.completed_at = datetime.now(UTC)
            return result

        stages = self.router.route(work_type, has_ui=has_ui)

        # Register all agents as Postman handlers
        self._register_handlers(stages)

        from orchestrata_ultimate.core.logging import new_correlation_id
        from orchestrata_ultimate.core.tracing.context import pipeline_trace

        new_correlation_id()
        # Bind a memory engine for this run so volks recall prior context on the
        # in-process (VizInt/UI) path too — the kernel path already does this.
        memory_bound = await self._bind_memory_engine(thread_id)
        # SPEC-035: open a verification session before stage execution so
        # after_stage() has a session to verify checkpoints against. Never
        # raises — a failure here just leaves verification off for this run.
        try:
            session_id = self.verification_gate.open_session(
                intent, str(work_type), thread_id
            )
        except Exception as exc:
            _logger.debug("verification open_session failed (non-fatal): %s", exc)
            session_id = None
        try:
            with pipeline_trace(thread_id) as tracer:
                if tracer is not None:
                    tracer.trace_event(
                        "pipeline_start",
                        {
                            "work_type": str(work_type),
                            "intent": intent[:200],
                            "stages": [s.alias for s in stages],
                        },
                    )

                # D1-addendum-2: _execute_stages owns interrupt handling
                # (escalate-to-human / needs_capability insertion) uniformly
                # for run(), resume(), and EscalationInterruptMixin's
                # approve/reject/respond re-entries — see pipeline_escalation.py.
                await self._execute_stages(
                    result,
                    stages,
                    0,
                    None,
                    thread_id,
                    intent,
                    work_type,
                    has_ui,
                    session_id,
                )

                # SPEC-035: final-checkpoint verdict, attached regardless of
                # enforcement posture; only enforce mode can flip
                # overall_status (see _apply_verification_verdict).
                self._apply_verification_verdict(
                    result, session_id, stages, thread_id, intent, work_type, has_ui
                )

                if tracer is not None:
                    tracer.trace_event(
                        "pipeline_complete",
                        {
                            "overall_status": result.overall_status.value,
                            "stages_completed": len(result.stages),
                            "total_artifacts": len(result.artifacts),
                        },
                    )

                # SPEC-012 REQ-012-051/055: end-of-iteration PM notification
                # (triggers the US-005 iteration summary). Advisory: a
                # missing U:PM handler is a silent no-op.
                await self._notify_pm_iteration_end(
                    result, thread_id, project_dir=self._project_dir
                )

            result.completed_at = datetime.now(UTC)
            self._sync_run_to_kg(result, intent)
            return result
        finally:
            if memory_bound:
                from orchestrata_ultimate.memory.memory_context import unregister_memory

                unregister_memory(thread_id)

    @traces(spec="003", story="FR-024", reqs=["REQ-003-FR24-003"])
    async def resume(
        self,
        thread_id: str,
        work_type: WorkType | str | None = None,
        has_ui: bool = False,
    ) -> PipelineResult:
        """Resume a checkpointed pipeline from its first incomplete stage.

        Real resume (backlog #38, W3.1 absorption): reconstructs completed
        stages from the checkpoint store and derives the FULL stage order
        from ``PipelineRouter.route(work_type, has_ui=has_ui)`` — the router
        is the single source of truth for stage sequencing, replacing the
        kernel's hardcoded ``_RESUME_ORDER``. Stages already checkpointed
        (a contiguous prefix of the router's order, in normal operation —
        see kernel's docstring for why: LLM stages are not replayable and
        are only ever checkpointed in that same order) are reused verbatim
        rather than re-invoked; execution re-enters ``_run_stage`` from the
        first stage in router order with no checkpointed output, chaining
        the last checkpointed stage's output in as its input payload —
        exactly like a fresh ``run()`` chains stage-to-stage.

        Args:
            thread_id: Thread to resume. Must have at least one checkpointed
                stage (see ``PipelineCheckpoint.resume_state``).
            work_type: Work type used to derive the stage order — should
                match the original run's work type so the right stages are
                considered "remaining" (defaults to "feature", the broadest
                sequence, when the caller doesn't know the original).
            has_ui: Whether the UX stage should be part of the derived order.

        Returns:
            A PipelineResult like ``run()``'s. Checkpointed stages appear as
            synthetic RESOLVED StageResults (no artifacts re-extracted — they
            were already captured and KG-synced by the original run);
            ``overall_status`` is FAILED with empty stages when nothing at
            all is checkpointed for ``thread_id``.
        """
        if work_type is None:
            stored = (
                self.checkpoint.stored_work_type(thread_id) if self.checkpoint else None
            )
            work_type = stored if stored is not None else "feature"

        state = self.checkpoint.resume_state(thread_id) if self.checkpoint else {}
        result = PipelineResult(
            thread_id=thread_id, work_type=str(work_type), intent="resume"
        )
        if not state:
            _logger.info("Resume: no checkpoint for thread %s", thread_id)
            result.overall_status = StageStatus.FAILED
            return result

        stages = self.router.route(work_type, has_ui=has_ui)

        current_packet: CXPPacket | None = None
        resume_index = 0
        for stage_spec in stages:
            checkpointed = state.get(stage_spec.alias)
            if checkpointed is None:
                break
            packet = CXPPacket(
                from_agent=stage_spec.alias,
                to_agent="HUMAN",
                thread_id=thread_id,
                subject="response",
                payload=dict(checkpointed),
                work_status=WorkStatus.RESOLVED,
            )
            result.stages[stage_spec.alias] = StageResult(
                stage_alias=stage_spec.alias,
                status=StageStatus.RESOLVED,
                output_packet=packet,
            )
            current_packet = packet
            resume_index += 1

        remaining = stages[resume_index:]
        self._register_handlers(remaining)

        # P0.1: Governance check before resuming stage execution
        allowed, reason = self.prolog.may_execute("HUMAN", "submit_intent")
        if not allowed:
            _logger.warning(
                "Governance blocked resume of thread %s: %s", thread_id, reason
            )
            result.overall_status = StageStatus.FAILED
            result.completed_at = datetime.now(UTC)
            return result

        # SPEC-035: open a verification session for the remaining stages —
        # never raises (see run()'s identical, non-fatal open_session call).
        try:
            session_id = self.verification_gate.open_session(
                "resume", str(work_type), thread_id
            )
        except Exception as exc:
            _logger.debug("verification open_session failed (non-fatal): %s", exc)
            session_id = None

        # D1-addendum-2: shared with run() — a later stage may itself
        # escalate or request a capability insertion; see pipeline_escalation.py.
        await self._execute_stages(
            result,
            stages,
            resume_index,
            current_packet,
            thread_id,
            "resume",
            work_type,
            has_ui,
            session_id,
        )

        self._apply_verification_verdict(
            result, session_id, stages, thread_id, "resume", work_type, has_ui
        )

        # SPEC-012 REQ-012-051/055: resume is a run completion too —
        # advisory end-of-iteration PM notification.
        await self._notify_pm_iteration_end(
            result, thread_id, project_dir=self._project_dir
        )

        result.completed_at = datetime.now(UTC)
        self._sync_run_to_kg(result, "resume")
        return result

    @traces(spec="000", story="US-006", reqs=["REQ-000-058"])
    async def _execute_stages(
        self,
        result: PipelineResult,
        stages: list[StageSpec],
        start_index: int,
        current_packet: CXPPacket | None,
        thread_id: str,
        intent: str,
        work_type: WorkType | str,
        has_ui: bool,
        session_id: str | None = None,
    ) -> None:
        """Execute ``stages[start_index:]`` into ``result``, honoring the
        escalation interrupt model (D1-addendum-2,
        plans/design-escalation-absorption-2026-07-06.md).

        Shared tail engine for ``run()``, ``resume()``, and
        ``EscalationInterruptMixin``'s ``reject``/``respond`` re-entries —
        every entry point into stage execution goes through here so
        escalation/capability-insertion is caught uniformly no matter how
        the sequence was reached. ``stages`` may grow in place: a resolved
        stage's ``needs_capability`` request that isn't already the router's
        own next stage gets its provider volk inserted right after the
        requesting stage (``EscalationInterruptMixin._classify_interrupt``/
        ``_build_capability_stage``) — a plain ``list.insert`` ahead of the
        loop index, picked up on the next iteration since list iteration is
        index-based.

        Mutates ``result.stages``/``result.artifacts``/``result.
        overall_status`` in place; does not set ``result.completed_at`` or
        sync to the KG — callers do that (``run()``/``resume()``) once the
        whole result, not just this tail, is final.

        Args:
            session_id: SPEC-035 verification session id from
                ``self.verification_gate.open_session`` (``run()``/
                ``resume()``), or ``None`` to skip per-stage checkpoint
                verification — the default for callers that don't open one
                (``EscalationInterruptMixin``'s ``reject``/``respond``
                re-entries; verification for those re-entries is a phase-2
                concern, see the SPEC-035 wiring contract).
        """
        capability_insertions: dict[str, int] = {}
        idx = start_index
        while idx < len(stages):
            stage_spec = stages[idx]

            # P0.1: Governance check before each stage dispatch
            allowed, reason = self.prolog.may_execute(
                stage_spec.alias, stage_spec.capability
            )
            if not allowed:
                _logger.warning(
                    "Governance blocked stage %s for %s: %s",
                    stage_spec.capability,
                    stage_spec.alias,
                    reason,
                )
                stage_result = StageResult(
                    stage_alias=stage_spec.alias,
                    status=StageStatus.FAILED,
                    error=f"Blocked by governance: {reason}",
                )
                result.stages[stage_spec.alias] = stage_result
                result.overall_status = StageStatus.FAILED
                return

            stage_result = await self._run_stage(
                stage_spec, current_packet, thread_id, intent, work_type
            )
            result.stages[stage_spec.alias] = stage_result
            result.artifacts.extend(stage_result.artifacts)

            if stage_result.status == StageStatus.FAILED:
                result.overall_status = StageStatus.FAILED
                return

            # SPEC-035: verify this stage's checkpoint (if its capability
            # maps to one) against the artifacts it just produced. Never
            # raises into the stage loop — audit-only recording unless the
            # gate is configured to enforce (enforcement itself happens at
            # finalize(), see _apply_verification_verdict).
            if session_id:
                try:
                    self.verification_gate.after_stage(
                        session_id, stage_spec, stage_result.artifacts, result
                    )
                except Exception as exc:
                    _logger.debug(
                        "verification after_stage failed for %s (non-fatal): %s",
                        stage_spec.alias,
                        exc,
                    )

            decision = self._classify_interrupt(
                stage_result.output_packet, stages, idx, capability_insertions
            )
            if decision.kind == "escalate":
                self._save_escalation(
                    thread_id,
                    stage_spec,
                    work_type,
                    has_ui,
                    intent,
                    stage_result.output_packet,
                    decision.reason,
                )
                result.overall_status = StageStatus.ESCALATED
                return
            if (
                decision.kind == "insert"
                and decision.provider_alias
                and decision.capability
            ):
                new_stage = self._build_capability_stage(
                    decision.capability, decision.provider_alias, stage_spec
                )
                stages.insert(idx + 1, new_stage)
                self._register_handlers([new_stage])

            current_packet = stage_result.output_packet
            idx += 1

        if result.overall_status not in (StageStatus.FAILED, StageStatus.ESCALATED):
            result.overall_status = StageStatus.RESOLVED

    @traces(
        spec="035",
        story="US-004",
        reqs=["REQ-035-US-004-002", "REQ-035-US-008-020"],
    )
    def _apply_verification_verdict(
        self,
        result: PipelineResult,
        session_id: str | None,
        stages: list[StageSpec],
        thread_id: str,
        intent: str,
        work_type: WorkType | str,
        has_ui: bool,
    ) -> None:
        """Finalize the SPEC-035 verification session for a completed run
        and attach its verdict to ``result.verification``.

        AUDIT-ONLY by default: the verdict is computed and recorded but
        never changes ``result.overall_status``. Only when
        ``self.verification_gate.enforce`` is set AND the verdict recommends
        ``ESCALATE`` AND the run hasn't already FAILED/ESCALATED on its own
        does this route through the EXISTING escalation-as-interrupt seam
        (``_save_escalation`` / ``StageStatus.ESCALATED``, D1-addendum-2) —
        no new escalation transport. Never raises: a verification failure
        never crashes an already-computed pipeline result.
        """
        if not session_id:
            return
        try:
            verdict = self.verification_gate.finalize(session_id, result)
        except Exception as exc:
            _logger.debug("verification finalize failed (non-fatal): %s", exc)
            verdict = None

        result.verification = verdict.to_dict() if verdict else None

        if (
            verdict is not None
            and verdict.recommendation == "ESCALATE"
            and getattr(self.verification_gate, "enforce", False)
            and result.overall_status not in (StageStatus.FAILED, StageStatus.ESCALATED)
        ):
            last_stage = stages[-1] if stages else None
            if last_stage is not None:
                self._save_escalation(
                    thread_id,
                    last_stage,
                    work_type,
                    has_ui,
                    intent,
                    None,
                    "verification exhausted max attempts",
                )
                result.overall_status = StageStatus.ESCALATED

    @traces(
        spec="008",
        story="US-001",
        reqs=["REQ-008-006", "REQ-008-002", "REQ-008-023"],
    )
    def _sync_run_to_kg(self, result: PipelineResult, intent: str) -> None:
        """Best-effort provenance sync to the knowledge graph at run completion.

        Batches the intent/run node, every artifact's GENERATED_BY edge, and
        any per-stage decisions in one place (``PipelineKGMixin.
        sync_intent_to_kg`` / ``sync_artifacts_to_kg`` / ``sync_decisions_to_kg``).
        The kernel writes the same data per-stage as packets flow
        (``pipeline_session.py``'s ``_collect_artifacts``/``_collect_decisions``);
        this dynamic-router engine runs the whole pipeline inline in
        ``run()``/``resume()``, so one sync at completion is equivalent and
        simpler. Never raises — a KG outage must never fail a pipeline run.

        Args:
            result: The completed (or failed) PipelineResult.
            intent: The human intent text this run was started from.
        """
        try:
            self.sync_intent_to_kg(result.thread_id, intent)
            self.sync_artifacts_to_kg(result.artifacts)
            for alias, stage_result in result.stages.items():
                packet = stage_result.output_packet
                if packet is None or not isinstance(packet.payload, dict):
                    continue
                decisions = packet.payload.get("decisions")
                if decisions:
                    self.sync_decisions_to_kg(alias, decisions)
        except Exception as exc:
            _logger.debug("KG run sync failed (non-fatal): %s", exc)

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

        try:
            from orchestrata_ultimate.memory.memory_context import register_memory
            from orchestrata_ultimate.memory.memory_engine import MemoryEngine

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

    def _register_handlers(self, stages: list[StageSpec]) -> None:
        """Register agent process() methods as Postman handlers.

        Matches the kernel's pattern: self.postman.register(alias, volk.process).
        Each stage instantiates its agent once and registers process() as
        the handler so Postman.route_sync() can dispatch to it. Also records
        the instantiated agent in ``self.volks`` (see ``__init__``) so KG
        artifact-provenance sync can attribute artifacts to a known volk.

        Registers each stage volk in the orchestrator's VolkRegistry so the
        runner (and the volks themselves, via tools) have real-time heartbeat
        and status awareness.
        """
        for stage in stages:
            alias = stage.alias
            handler_already_registered = alias in self._postman._handlers
            if not handler_already_registered:
                agent_class = self.router.get_agent_class(alias)
                agent = agent_class()
                self._postman.register(alias, agent.process)
                self.volks[alias] = agent
                if hasattr(agent, "set_volk_registry"):
                    agent.set_volk_registry(self.volk_registry)
                if hasattr(agent, "set_runtime_writer"):
                    agent.set_runtime_writer(getattr(self, "_runtime_writer", None))
                if hasattr(agent, "set_context_injector"):
                    agent.set_context_injector(getattr(self, "context_injector", None))

            # Orchestrator↔volk heartbeat awareness: register the volk even if
            # a custom handler was pre-registered, so the runner always tracks
            # its live volks.
            capability_name = (
                stage.capability.value
                if hasattr(stage.capability, "value")
                else str(stage.capability)
            )
            if not self.volk_registry.is_registered(alias):
                self.volk_registry.register_volk(
                    VolkDefinition(
                        alias=alias,
                        name=alias,
                        role=capability_name,
                        capabilities=[VolkCapability(name=capability_name)],
                        protocols=[],
                    )
                )

    @traces(spec="003", story="US-004", reqs=["REQ-003-US-004-022"])
    async def _run_stage(
        self,
        stage_spec: StageSpec,
        input_packet: CXPPacket | None,
        thread_id: str,
        intent: str,
        work_type: WorkType | str | None = None,
    ) -> StageResult:
        """Route a message through Postman with retry, backoff, and checkpointing.

        Production error handling:
        - MAX_RETRIES=3 with exponential backoff (5s/10s/20s)
        - Non-retryable errors (import, syntax) fail immediately
        - All retries logged through the stdlib ``logging`` (core.logging)
        - Partial artifacts captured on failure
        - Stage timing includes retry backoff duration
        - Checkpoint saved to Redis after each successful stage

        Args:
            stage_spec: Stage specification from router.
            input_packet: Output from previous stage (None for first).
            thread_id: Pipeline thread ID.
            intent: Original user intent.

        Returns:
            StageResult with output packet, artifacts, and timing.
        """
        started = datetime.now(UTC)
        result = StageResult(stage_alias=stage_spec.alias)

        from orchestrata_ultimate.core.tracing.context import get_tracer

        tracer = get_tracer(thread_id)

        # Build the outbound packet once — retries re-route the same packet.
        if input_packet is None:
            # First stage — create from intent
            subject = _CAPABILITY_INTAKE_SUBJECT.get(
                stage_spec.capability.value, "intent"
            )
            first_payload: dict[str, object] = {
                "text": intent,
                "requirements": intent,
            }
            if self._project_dir:
                first_payload["project_dir"] = self._project_dir
            packet = CXPPacket(
                from_agent="HUMAN",
                to_agent=stage_spec.alias,
                thread_id=thread_id,
                subject=subject,
                payload=first_payload,
                work_status=WorkStatus.PENDING,
            )
        else:
            # Subsequent stage — chain from previous output
            packet = CXPPacket(
                from_agent=input_packet.to_agent,
                to_agent=stage_spec.alias,
                thread_id=thread_id,
                subject=stage_spec.input_subject or "process",
                payload=input_packet.payload if input_packet else {},
                work_status=WorkStatus.PENDING,
            )

        # W3.1 KG absorption: inject graph context into the outbound payload
        # before dispatch — the payload-level adapter shared with the
        # kernel's CXP-packet escalation path (PipelineKGMixin.
        # inject_for_subject / _maybe_inject_context). When the assembler is
        # wired, memory/artifacts/mission/task-type context are folded into
        # the same graph_context block under a single token budget.
        # Best-effort/no-op when the subject doesn't warrant injection or
        # the graph is unavailable.
        packet.payload = self.inject_for_subject(
            packet.payload,
            packet.subject,
            thread_id=thread_id,
            artifact_store=self.artifact_store,
        )

        # Tier-H signal: record the outbound request message.
        self._write_message_signal(
            packet.payload,
            sender=packet.from_agent,
            recipient=packet.to_agent,
            subject=packet.subject,
            thread_id=thread_id,
            direction="out",
        )

        if tracer is not None:
            tracer.trace_event(
                "stage_start",
                {"volk": stage_spec.alias, "subject": packet.subject},
            )

        # Route through Postman with bounded retry + exponential backoff. Only
        # transient failures (see ``_is_retryable``) are retried; structural
        # errors (import/syntax/missing-file) and a missing handler fail fast.
        for attempt in range(MAX_RETRIES + 1):
            if attempt > 0:
                # Postman.route_sync() deduplicates by a deterministic ID
                # derived from (from_agent, to_agent, subject, thread_id,
                # timestamp) — see cxp.deterministic_ids. The failed first
                # attempt already marks that ID as seen (dedup runs before
                # dispatch), so re-routing the *same* packet object on retry
                # would be silently swallowed as a duplicate and never reach
                # the handler again (response=None, misdiagnosed as a missing
                # handler instead of the real transient error). Refreshing the
                # timestamp gives each retry a fresh ID so it actually
                # redelivers.
                packet.timestamp = datetime.now(UTC)
            try:
                response = await asyncio.to_thread(self._postman.route_sync, packet)
            except Exception as e:
                if _is_retryable(e) and attempt < MAX_RETRIES:
                    backoff = self._retry_backoff_s[
                        min(attempt, len(self._retry_backoff_s) - 1)
                    ]
                    result.retries = attempt + 1
                    result.retry_backoff_s += backoff
                    _logger.warning(
                        "Stage %s attempt %d failed (%s: %s) — retrying in %.0fs",
                        stage_spec.alias,
                        attempt + 1,
                        type(e).__name__,
                        e,
                        backoff,
                    )
                    if tracer is not None:
                        tracer.trace_event(
                            "stage_retry",
                            {
                                "volk": stage_spec.alias,
                                "attempt": attempt + 1,
                                "error": str(e)[:200],
                                "backoff_s": backoff,
                            },
                        )
                    await asyncio.sleep(backoff)
                    continue

                # Non-retryable error, or retries exhausted — fail the stage.
                import traceback as _tb

                result.status = StageStatus.FAILED
                result.error = f"{type(e).__name__}: {e}"
                if tracer is not None:
                    tracer.trace_event(
                        "stage_error",
                        {
                            "volk": stage_spec.alias,
                            "error": str(e)[:500],
                            "traceback": _tb.format_exc(),
                        },
                    )
                break

            if response is None:
                # Missing handler / queued — structural, retrying will not help.
                # Log loudly (dogfood F18): a volk whose process() doesn't
                # handle the routed subject returns None and, without this
                # line, the stage failed with ZERO trace anywhere.
                result.status = StageStatus.FAILED
                result.error = (
                    f"Agent {stage_spec.alias} returned no response for "
                    f"subject {packet.subject!r} (unhandled subject, handler "
                    f"missing, or queued)"
                )
                _logger.warning("stage %s FAILED: %s", stage_spec.alias, result.error)
                break

            result.status = StageStatus.RESOLVED
            result.output_packet = response

            # Tier-H signal: record the inbound response message.
            response_payload = (
                response.payload
                if response and isinstance(response.payload, dict)
                else {}
            )
            self._write_message_signal(
                response_payload,
                sender=packet.to_agent,
                recipient=packet.from_agent,
                subject=response.subject if response else packet.subject,
                thread_id=thread_id,
                direction="in",
            )

            # Orchestrator↔volk heartbeat awareness: record liveness for the
            # stage volk after a successful dispatch/response.
            try:
                self.volk_registry.record_heartbeat(stage_spec.alias)
            except Exception as e:
                _logger.debug(
                    "Heartbeat recording failed for %s: %s", stage_spec.alias, e
                )

            self._save_stage_checkpoint(
                stage_spec, packet, response, thread_id, work_type=str(work_type)
            )
            result.artifacts.extend(
                self._collect_stage_artifacts(stage_spec, response, thread_id)
            )
            break

        result.duration_ms = (datetime.now(UTC) - started).total_seconds() * 1000
        if tracer is not None:
            tracer.trace_event(
                "stage_complete",
                {
                    "volk": stage_spec.alias,
                    "status": result.status.value,
                    "duration_ms": round(result.duration_ms, 2),
                    "artifact_count": len(result.artifacts),
                },
            )
        self._session_log.append(
            f"Pipeline: {stage_spec.alias} — {result.status.value} "
            f"({len(result.artifacts)} artifacts, {result.duration_ms:.0f}ms"
            f"{f', retries={result.retries}' if result.retries > 0 else ''})"
        )
        await self._notify_stage_pm(stage_spec, result, thread_id)

        return result

    @traces(spec="012", story="US-012-Progress", reqs=["REQ-012-051"])
    async def _notify_stage_pm(
        self,
        stage_spec: StageSpec,
        result: StageResult,
        thread_id: str,
    ) -> None:
        """Advisory U:PM notification for one completed stage.

        Delivers a ``progress_update`` CXP packet to a registered U:PM
        handler (SPEC-012 REQ-012-051). Advisory-only by contract
        (SPEC-012 NFR-002): without a registered U:PM handler, or on any
        routing failure, this is a silent no-op.
        """
        await self._send_pm_progress(
            thread_id,
            {
                "stage": stage_spec.alias,
                "status": result.status.value,
                "duration_ms": round(result.duration_ms, 2),
                "artifact_count": len(result.artifacts),
            },
        )

    @traces(spec="012", story="US-012-Progress", reqs=["REQ-012-051"])
    async def _notify_pm_iteration_end(
        self,
        result: PipelineResult,
        thread_id: str,
        project_dir: str | None = None,
    ) -> None:
        """Advisory U:PM end-of-run notification (SPEC-012 REQ-012-051).

        The ``iteration_complete`` payload triggers the US-005 iteration
        summary on the PM side. Advisory-only by contract
        (SPEC-012 NFR-002): without a registered U:PM handler, or on any
        routing failure, this is a silent no-op.
        """
        payload: dict[str, object] = {"iteration_complete": True}
        if project_dir:
            payload["project_dir"] = project_dir
        await self._send_pm_progress(thread_id, payload)

    async def _send_pm_progress(
        self, thread_id: str, payload: dict[str, object]
    ) -> None:
        """Route one progress payload to a registered U:PM handler.

        Transport helper: the public notifiers above carry the traces;
        this never raises.
        """
        try:
            handlers = getattr(self._postman, "_handlers", {})
            if "U:PM" not in handlers:
                return
            packet = CXPPacket(
                from_agent="U:PIPELINE",
                to_agent="U:PM",
                thread_id=thread_id,
                subject="progress_update",
                payload=payload,
                work_status=WorkStatus.RESOLVED,
            )
            await asyncio.to_thread(self._postman.route_sync, packet)
        except Exception as exc:
            _logger.debug("U:PM progress notify skipped (advisory): %s", exc)

    @traces(spec="003", story="FR-024", reqs=["REQ-003-FR24-003"])
    def _save_stage_checkpoint(
        self,
        stage_spec: StageSpec,
        request_packet: CXPPacket,
        response: CXPPacket,
        thread_id: str,
        work_type: str | None = None,
    ) -> None:
        """Persist a resolved stage's output for a later real ``resume()``
        (backlog #38, W3.1 absorption).

        Shape matches ``kernel.pipeline_checkpoint.PipelineCheckpoint.
        save_stage`` exactly — ``session/cli/pipeline.py``'s ``pipeline
        status``/``pipeline resume`` read the identical Redis hash via
        ``completed_stages()``/``resume_state()``. Traces-union with the
        kernel's equivalent per-stage write (``pipeline_session.py``'s
        ``_checkpoint_stage``, REQ-001-960) over the same ``save_stage()``
        call (REQ-001-954). Best-effort — a checkpoint failure never fails
        an already-resolved stage.

        Args:
            stage_spec: The resolved stage's specification.
            request_packet: The packet routed TO the stage (carries the
                subject the stage was invoked under).
            response: The stage's output CXPPacket.
            thread_id: Pipeline thread id.
        """
        if self.checkpoint is None:
            return
        output = response.payload if isinstance(response.payload, dict) else {}
        try:
            self.checkpoint.save_stage(
                thread_id=thread_id,
                stage=stage_spec.alias,
                subject=request_packet.subject,
                output=output,
                work_type=work_type,
            )
        except Exception as exc:
            _logger.debug(
                "Checkpoint save failed for stage %s: %s", stage_spec.alias, exc
            )

    @traces(spec="001", story="US-006", reqs=["REQ-001-US-006-018"])
    def _collect_stage_artifacts(
        self, stage_spec: StageSpec, response: CXPPacket, thread_id: str
    ) -> list[PipelineArtifact]:
        """Extract and store artifacts from a resolved stage's response payload.

        Traces-union with ``kernel.pipeline_session.PipelineSessionMixin.
        _collect_artifacts`` (REQ-001-US-006-018) — the orchestrator engine's own
        per-stage artifact capture (W3.1 absorption; kernel's ArtifactStore
        itself is NOT moved, per plans/decisions-2026-07-06.md D1-addendum).

        Args:
            stage_spec: The resolved stage's specification (artifact_types).
            response: The stage's output CXPPacket.
            thread_id: Pipeline thread id, attributed on each artifact.

        Returns:
            The artifacts extracted (already added to ``self.artifact_store``).
        """
        collected: list[PipelineArtifact] = []
        for at_str in stage_spec.artifact_types:
            try:
                at = ArtifactType(at_str)
            except ValueError:
                continue
            content = self._extract_content(at, response.payload)
            if content:
                artifact = PipelineArtifact(
                    stage_alias=stage_spec.alias,
                    artifact_type=at,
                    content=str(content)[:10000],
                    thread_id=thread_id,
                )
                collected.append(artifact)
                self.artifact_store.add(artifact)
        return collected

    def _extract_content(self, at: ArtifactType, payload: dict) -> str | None:
        """Extract artifact content from agent output payload."""
        mapping = {
            ArtifactType.REQUIREMENTS: payload.get("requirements"),
            ArtifactType.PROLOG_FACTS: "\n".join(payload.get("prolog_facts", [])),
            ArtifactType.TASK_SUMMARY: payload.get("task_summary"),
            ArtifactType.USER_STORIES: payload.get("user_stories"),
            ArtifactType.REQUIREMENTS_ANALYSIS: payload.get("requirements_analysis"),
            ArtifactType.PROJECT_META: str(
                {
                    "project_type": payload.get("project_type", ""),
                    "technology": payload.get("technology", {}),
                }
            ),
            ArtifactType.SCOPE_ESTIMATE: str(payload.get("estimate", "")),
            ArtifactType.WARNINGS: str(payload.get("warnings", [])),
            ArtifactType.ARCHITECTURE_DESIGN: payload.get("architecture"),
            ArtifactType.FILE_TREE: payload.get("file_tree"),
            ArtifactType.API_SPEC: payload.get("api_spec"),
            ArtifactType.CLASS_DIAGRAM: payload.get("class_diagram"),
            ArtifactType.SOURCE_CODE: str(payload.get("files_created", [])),
            ArtifactType.IMPLEMENTATION_PLAN: str(
                payload.get("implementation_plan", "")
            ),
            ArtifactType.TEST_CODE: str(payload.get("test_files", [])),
            # qa/volk.py emits the LLM review under ``review`` (set only when
            # issues were found; a clean pass carries no review key) --
            # ``review_report`` was a phantom key nothing ever emitted, so
            # QA_REPORT extraction was always None and the review work
            # product was discarded. On a clean pass, compose the test
            # verdict + output so a QA stage always leaves a retrievable
            # report.
            ArtifactType.QA_REPORT: (
                str(payload["review"])
                if payload.get("review") is not None
                else (
                    "tests {}\n{}".format(
                        "PASS" if payload.get("test_passed") else "FAIL",
                        payload.get("test_output", ""),
                    ).strip()
                    if "test_passed" in payload
                    else None
                )
            ),
            ArtifactType.RTM: str(payload.get("rtm", "")),
            ArtifactType.COMPLIANCE_REPORT: payload.get("compliance_report"),
            # Dogfood run-3f F14: prefer the full authored document body
            # (``doc_content``, now persisted + carried through the payload
            # by DocumentorVolk) over the one-line ``doc_summary`` -- the
            # summary alone was never a real deliverable.
            ArtifactType.DOC_OUTPUT: payload.get("doc_content")
            or payload.get("doc_summary"),
            ArtifactType.UX_DESIGN: payload.get("ux_design"),
        }
        content = mapping.get(at)
        return content if content else None

    def get_session_log(self) -> list[str]:
        return list(self._session_log)

    @traces(spec="003", story="FR-024", reqs=["REQ-003-FR24-001"])
    def get_message_log(self) -> list[str]:
        """Get string representations of all routed messages for CXPFeed."""
        return self._postman.get_log() if self._postman else []

    def _log_stage_to_session(
        self,
        stage_alias: str,
        status: str,
        artifact_count: int,
        duration_ms: float,
        error: str | None = None,
    ) -> None:
        """Log pipeline stage completion to session controller."""
        if not self._session_ctrl:
            return
        try:
            desc = f"Pipeline: {stage_alias} — {status} ({artifact_count} artifacts, {duration_ms:.0f}ms)"
            details = [
                f"Stage: {stage_alias}",
                f"Status: {status}",
                f"Artifacts: {artifact_count}",
            ]
            if error:
                details.append(f"Error: {error[:200]}")
            self._session_ctrl.log_work(desc=desc, details=details)
        except Exception:
            pass  # Session logging is best-effort

    def _set_handoff(self, result: PipelineResult) -> None:
        """Set pipeline result as session handoff data for next session."""
        if not self._session_ctrl:
            return
        try:
            self._session_ctrl.set_handoff(
                decisions=[
                    f"Pipeline completed: {result.work_type} — {result.overall_status.value}"
                ],
                active_work=[
                    f"Pipeline {result.thread_id}: {result.work_type} — {result.overall_status.value}"
                ],
                key_files=list(result.stages.keys()) if result.stages else [],
                cmds=[f"session pipeline status {result.thread_id}"],
                alerts=[
                    f"Pipeline {result.thread_id}: {'success' if result.successful() else 'FAILED'} — {len(result.artifacts)} artifacts"
                ],
            )
        except Exception:
            pass


def _is_retryable(error: Exception) -> bool:
    """Determine if an error is retryable.

    Non-retryable: import errors, syntax errors, missing files.
    Retryable: timeouts, connection errors, transient infrastructure failures.
    """
    # Structural errors are checked first: FileNotFoundError is an OSError
    # subclass, so it must be excluded before the broad OSError retryable sweep
    # below (otherwise a genuinely missing file would be retried pointlessly).
    if isinstance(
        error, (ImportError, ModuleNotFoundError, SyntaxError, FileNotFoundError)
    ):
        return False
    retryable_types = (
        TimeoutError,
        asyncio.TimeoutError,
        ConnectionError,
        ConnectionRefusedError,
        ConnectionResetError,
        OSError,  # Transient I/O
    )
    if isinstance(error, retryable_types):
        return True
    # Default: retry unknown errors (conservative — rather retry than silently fail)
    return True