orchestrata_ultimate.orchestrator.pipeline_escalation

Reference › orchestrata_ultimate › orchestrator › pipeline_escalation

The escalation-as-interrupt behaviour of the pipeline, kept as a mixin that pipeline_runner inherits. The host class must supply the router, the volk roster, the checkpoint store, Postman and the stage-running methods. It also uses core.capability_mesh, kernel.pipeline_checkpoint and the stage specs.

After each resolved stage, _classify_interrupt examines the response and returns an InterruptDecision of kind none, escalate or insert. A response payload with needs_capability is resolved through resolve_capability to a provider volk. If the router's own next stage already provides it, nothing changes. An unknown capability escalates. Otherwise _build_capability_stage inserts the provider stage right after the requesting one, with a guard that allows one insertion per capability and a bounded number per run, beyond which the run escalates. A response addressed to HUMAN that is still in progress also escalates, because the volk asked for human input.

On escalation _save_escalation writes an EscalationState (thread, stage, work type, intent, payload, reason) to the checkpoint store, best effort. A human then resolves it through three public methods, each returning a PipelineResult: approve consumes the marker and resumes from the next stage; reject takes a reason and re-enters the stage; respond passes the human's answer text back into the stage (_collect_prior_human_answers carries earlier answers). With no pending escalation they return a result saying so. The docstrings state that no escalation is ever approved automatically.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

InterruptDecision (dataclass)

Decorators: dataclass · Line 127

What a resolved stage's response means for sequence control.

EscalationState (dataclass)

Decorators: traces(spec='000', story='US-006', reqs=['REQ-000-058']), dataclass · Line 138

Persisted state for one suspended (ESCALATED) pipeline run.

EscalationInterruptMixin

Line 156

Escalation-as-interrupt helpers for PipelineRunner.

Functions

Source

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

"""PipelineRunner escalation INTERRUPT mixin (W3.1 absorption, final piece).

Ports ``kernel.pipeline_escalation.PipelineEscalationMixin``'s approval
workflow and capability-mesh resolver from the kernel's free-hop packet
routing shape onto the orchestrator's fixed-``PipelineRouter``-sequence
shape, per the ruling in plans/decisions-2026-07-06.md D1-addendum-2 and
the full design in plans/design-escalation-absorption-2026-07-06.md:

Escalation is an INTERRUPT on the fixed sequence, not a routing mode:
    - ESCALATE: a stage addresses HUMAN while still ``WorkStatus.IN_PROGRESS``
      (blocked / needs review — the same convention real ported volks already
      use: see ``agents/volks/concierge/volk.py``'s clarification loop and
      ``agents/volks/codedirector/volk.py``'s ``implementation_ready`` gate,
      both IN_PROGRESS-to-HUMAN; contrast with a normal hand-off, which is
      PENDING, and a truly finished run's last stage, which is RESOLVED —
      see ``agents/volks/compliance/volk.py``). The run SUSPENDS: the
      escalation is checkpointed and ``PipelineResult.overall_status``
      becomes ``StageStatus.ESCALATED``.
    - APPROVE: marks the escalation resolved and calls the existing
      ``PipelineRunner.resume()`` — the fixed sequence continues from the
      NEXT router stage (the escalated stage's own output was already
      checkpointed RESOLVED by ``_run_stage``/``_save_stage_checkpoint``).
    - REJECT/RESPOND: re-enter the SAME stage with the human's feedback/text
      folded into its payload (REQ-000-056) instead of advancing.
    - ``needs_capability``: resolved via the ported ``_CAPABILITY_PROVIDERS``
      mesh (``resolve_capability``) and, when the resolved provider is not
      already the router's own next stage, INSERTED into the remaining
      sequence right after the requesting stage — a bounded, router-visible
      amendment, not a free hop. Depth-capped and cycle-guarded (see
      ``_classify_interrupt``); an unresolvable capability escalates to
      human cleanly instead of silently continuing.

No auto-approval anywhere (REQ-000-058): the only way an ESCALATED run
continues is an explicit ``approve``/``reject``/``respond`` call — nothing in
this module calls those on its own.

BOUNDARY (Mohamed's ruling, 2026-08-09). This project has TWO capability
meshes, deliberately:

* THIS module routes at the STAGE level — ``_build_capability_stage`` inserts
  a ``StageSpec`` into the remaining sequence, ``_classify_interrupt`` caps
  the insertions. It serves the gateway/UI path (``gateway/server.py`` ->
  ``PipelineKernelClient``) and ``session pipeline approve|respond|reject``.
* ``kernel.pipeline_escalation`` routes at the PACKET level — it rewrites
  ``packet.to_agent`` and hops, for the gRPC path.

They are complements, not duplicates, and neither absorbs the other. Do NOT
add a ``to_agent`` rewrite here — ``tests/core/test_capability_mesh.py`` fails
if either side grows the other's idiom.

The RESOLUTION step they share lives once, in ``core.capability_mesh``.
"""

from __future__ import annotations

import dataclasses
import logging
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Any, Literal

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.capability_mesh import (
    CAPABILITY_PROVIDERS,
)
from orchestrata_ultimate.core.capability_mesh import (
    resolve_capability as _resolve_capability,
)
from orchestrata_ultimate.core.packet import CXPPacket, WorkStatus
from orchestrata_ultimate.orchestrator.pipeline_router import Capability, StageSpec

if TYPE_CHECKING:
    from orchestrata_ultimate.kernel.pipeline_checkpoint import PipelineCheckpoint
    from orchestrata_ultimate.orchestrator.pipeline_router import PipelineRouter
    from orchestrata_ultimate.orchestrator.pipeline_runner import PipelineResult
    from orchestrata_ultimate.postman.postman import Postman

_logger = logging.getLogger(__name__)

# Re-exported from core.capability_mesh, which owns the one copy. This used to
# be a verbatim port of kernel.pipeline_escalation's private table; the name is
# kept so existing importers keep working.
_CAPABILITY_PROVIDERS: dict[str, str] = CAPABILITY_PROVIDERS

# Best-effort mapping from the mesh's free-form capability strings (above) to
# the router's Capability enum, used only to shape the synthetic StageSpec a
# capability insertion builds (artifact_types lookup) — falls back to the
# requesting stage's own capability when a mesh string has no enum
# counterpart (e.g. "clarification", "planning", "coding").
_CAPABILITY_TO_STAGE_CAPABILITY: dict[str, Capability] = {
    "requirements": Capability.INTAKE,
    "clarification": Capability.INTAKE,
    "analysis": Capability.ANALYSIS,
    "planning": Capability.SCOPING,
    "scope": Capability.SCOPING,
    "design": Capability.ARCHITECTURE,
    "architecture": Capability.ARCHITECTURE,
    "ux": Capability.UI_DESIGN,
    "implementation": Capability.IMPLEMENTATION,
    "coding": Capability.IMPLEMENTATION,
    "testing": Capability.TESTING,
    "review": Capability.TESTING,
    "compliance": Capability.TRACEABILITY,
    "traceability": Capability.TRACEABILITY,
}

# Per-run bound on capability-triggered stage insertions — mirrors the
# kernel mixin's depth-limit pattern (``_route_escalation``'s
# ``max_depth``/``_hop_cycle_detected``), scaled down from packet-hop
# granularity (200 hops) to whole-STAGE granularity: inserting the same
# whole pipeline stage more than once is essentially always a cycle at this
# level, so a capability may trigger at most one insertion per run, and the
# run-wide total is capped defensively too.
MAX_CAPABILITY_INSERTIONS_PER_RUN = 5

# Reserved checkpoint "stage" key holding the current escalation state for a
# thread — never collides with a real volk alias (all real aliases are
# "U:Xx"/"J:Xxx" shaped). Piggybacks on the existing PipelineCheckpoint
# save_stage/get_stage API (no changes to kernel.pipeline_checkpoint needed):
# resume()'s own stage loop only reads keys matching router stage aliases,
# so this sentinel entry is invisible to it.
_ESCALATION_STAGE_KEY = "__escalation__"

InterruptKind = Literal["none", "escalate", "insert"]


@dataclass
class InterruptDecision:
    """What a resolved stage's response means for sequence control."""

    kind: InterruptKind
    reason: str | None = None
    provider_alias: str | None = None
    capability: str | None = None


@traces(spec="000", story="US-006", reqs=["REQ-000-058"])
@dataclass
class EscalationState:
    """Persisted state for one suspended (ESCALATED) pipeline run.

    Stored via the existing ``PipelineCheckpoint.save_stage``/``get_stage``
    under :data:`_ESCALATION_STAGE_KEY` — plain JSON-serialisable fields only.
    """

    thread_id: str
    stage: str
    work_type: str
    has_ui: bool
    intent: str
    subject: str
    payload: dict[str, Any] = field(default_factory=dict)
    reason: str | None = None
    status: str = "pending"


class EscalationInterruptMixin:
    """Escalation-as-interrupt helpers for PipelineRunner.

    Requires the following attributes/methods on the host class (all present
    on ``orchestrator.pipeline_runner.PipelineRunner``): ``router``,
    ``volks``, ``checkpoint``, ``_postman``, ``_register_handlers``,
    ``_run_stage``, ``_execute_stages``, ``resume``.
    """

    router: PipelineRouter
    volks: dict[str, Any]
    checkpoint: PipelineCheckpoint | None
    _postman: Postman
    # Typed as Any (not Callable[...]) deliberately: PipelineRunner defines
    # these as real bound methods with their own (incompatible-looking to
    # pyright's override check, since a Callable member type omits `self`)
    # signatures — Any avoids a spurious reportIncompatibleMethodOverride
    # while still documenting the expected host surface (see class docstring).
    _register_handlers: Any
    _execute_stages: Any
    resume: Any

    @traces(spec="005", story="US-008", reqs=["REQ-005-FR10-001", "REQ-005-AC11-001"])
    def resolve_capability(self, capability: str) -> str | None:
        """Return the alias of a registered, dispatchable volk that provides
        ``capability`` — the capability-routing resolver behind the volk mesh.

        Ported from ``kernel.pipeline_escalation.PipelineEscalationMixin.
        resolve_capability`` unchanged: the built-in ``_CAPABILITY_PROVIDERS``
        map is authoritative for the core capabilities; a registered volk that
        declares its own ``capabilities`` list provides any *additional*
        capability so plugin volks extend the mesh without hijacking core
        hops. Returns None when no provider is registered in ``self.volks``
        (caller escalates to human instead of silently continuing).
        """
        return _resolve_capability(self.volks, capability)

    def _classify_interrupt(
        self,
        response: CXPPacket | None,
        stages: list[StageSpec],
        idx: int,
        capability_insertions: dict[str, int],
    ) -> InterruptDecision:
        """Decide what a resolved stage's response means for the sequence.

        Args:
            response: The resolved stage's output packet.
            stages: The full (possibly already-amended) stage list.
            idx: The just-resolved stage's index in ``stages``.
            capability_insertions: Per-run counter of insertions already made
                per capability string — mutated in place (cycle guard).

        Returns:
            "none" — happy path, continue with the fixed sequence unchanged.
            "escalate" — the run must suspend (see ``reason``).
            "insert" — insert ``provider_alias`` (providing ``capability``)
                right after ``idx`` in ``stages``.
        """
        if response is None or not isinstance(response.payload, dict):
            return InterruptDecision("none")

        cap = response.payload.get("needs_capability")
        if cap:
            target = self.resolve_capability(cap)
            already_next = (
                target is not None
                and idx + 1 < len(stages)
                and stages[idx + 1].alias == target
            )
            if already_next:
                # The router's own next stage already provides this
                # capability — nothing to amend, preserves the happy path
                # for the common case (e.g. Architect's needs_capability=
                # "implementation" ahead of the router's own U:Cd stage).
                return InterruptDecision("none")
            if target is None:
                return InterruptDecision(
                    "escalate", reason=f"unknown capability: {cap}"
                )
            count = capability_insertions.get(cap, 0)
            total = sum(capability_insertions.values())
            if count >= 1 or total >= MAX_CAPABILITY_INSERTIONS_PER_RUN:
                return InterruptDecision(
                    "escalate",
                    reason=(
                        f"capability '{cap}' insertion limit reached "
                        "(cycle/depth guard)"
                    ),
                )
            capability_insertions[cap] = count + 1
            return InterruptDecision("insert", provider_alias=target, capability=cap)

        if (
            response.to_agent == "HUMAN"
            and response.work_status == WorkStatus.IN_PROGRESS
        ):
            return InterruptDecision("escalate", reason="volk requested human input")

        return InterruptDecision("none")

    def _build_capability_stage(
        self, capability: str, provider_alias: str, origin: StageSpec
    ) -> StageSpec:
        """Build the synthetic StageSpec inserted for a resolved capability
        request, bookkeeping-only (the run doesn't consult router tables for
        an inserted stage): reuses ``origin``'s output subject as the input
        subject (the requesting stage described that as the CXP subject for
        whatever comes next) and the router's own per-capability artifact
        types when the mesh string maps to a real ``Capability``.
        """
        cap_enum = _CAPABILITY_TO_STAGE_CAPABILITY.get(capability, origin.capability)
        cap_artifacts = getattr(self.router, "_cap_artifacts", {})
        return StageSpec(
            alias=provider_alias,
            capability=cap_enum,
            input_subject=origin.output_subject or "implement",
            artifact_types=list(cap_artifacts.get(cap_enum, [])),
        )

    def _save_escalation(
        self,
        thread_id: str,
        stage_spec: StageSpec,
        work_type: object,
        has_ui: bool,
        intent: str,
        response: CXPPacket | None,
        reason: str | None,
    ) -> None:
        """Checkpoint the ESCALATED state carrying the escalation payload.

        Best-effort like every other checkpoint write in this engine
        (``PipelineRunner._save_stage_checkpoint``) — a checkpoint failure
        must never crash an already-computed escalation.
        """
        if self.checkpoint is None:
            return
        payload = (
            dict(response.payload)
            if response is not None and isinstance(response.payload, dict)
            else {}
        )
        state = EscalationState(
            thread_id=thread_id,
            stage=stage_spec.alias,
            work_type=str(work_type),
            has_ui=has_ui,
            intent=intent,
            subject=response.subject if response is not None else "",
            payload=payload,
            reason=reason,
            status="pending",
        )
        try:
            self.checkpoint.save_stage(
                thread_id=thread_id,
                stage=_ESCALATION_STAGE_KEY,
                subject="escalated",
                output=dataclasses.asdict(state),
            )
        except Exception as exc:
            _logger.debug(
                "Escalation checkpoint save failed for thread %s: %s", thread_id, exc
            )

    @traces(spec="000", story="US-006", reqs=["REQ-000-058"])
    def _load_escalation(self, thread_id: str) -> EscalationState | None:
        """Return the pending escalation for ``thread_id``, or None when
        there is none (never checkpointed, or already consumed by a prior
        approve/reject/respond — REQ-000-058, no re-approving a settled
        escalation)."""
        if self.checkpoint is None:
            return None
        raw = self.checkpoint.get_stage(thread_id, _ESCALATION_STAGE_KEY)
        if not raw or raw.get("status") != "pending":
            return None
        try:
            return EscalationState(**raw)
        except TypeError:
            return None

    @traces(spec="000", story="US-006", reqs=["REQ-000-058"])
    def _resolve_escalation(
        self, thread_id: str, state: EscalationState, new_status: str
    ) -> None:
        """Consume the pending escalation record by overwriting its status
        (best-effort, mirrors ``_save_escalation``)."""
        if self.checkpoint is None:
            return
        updated = dataclasses.replace(state, status=new_status)
        try:
            self.checkpoint.save_stage(
                thread_id=thread_id,
                stage=_ESCALATION_STAGE_KEY,
                subject=new_status,
                output=dataclasses.asdict(updated),
            )
        except Exception as exc:
            _logger.debug(
                "Escalation checkpoint update failed for thread %s: %s",
                thread_id,
                exc,
            )

    @traces(spec="000", story="US-004", reqs=["REQ-000-039"])
    def _collect_prior_human_answers(self, thread_id: str) -> list[str]:
        """Prior human answers on this thread, oldest-first — mirrors
        ``kernel.pipeline.PipelineRunner.respond``'s own history replay so a
        re-entered stage sees the same conversation context."""
        try:
            packets = self._postman.get_thread_packets(thread_id)
        except Exception:
            return []
        return [
            p.payload.get("text", "")
            for p in packets
            if p.from_agent == "HUMAN" and p.subject == "response"
        ]

    @traces(spec="000", story="US-004", reqs=["REQ-000-039", "REQ-000-056"])
    async def _reenter_escalated_stage(
        self,
        state: EscalationState,
        feedback_payload: dict[str, Any],
        subject_override: str | None,
    ) -> PipelineResult:
        """Re-run the escalated stage with ``feedback_payload`` as its input,
        then continue the remaining fixed sequence — the shared engine
        behind ``reject``/``respond``.

        Args:
            state: The (already-consumed) escalation being resolved.
            feedback_payload: The payload to send to the escalated stage —
                the human's feedback/response folded into the stage's own
                last payload.
            subject_override: When given, the CXP subject to re-enter the
                stage under (``respond`` uses "response", matching the
                kernel's own respond() convention); ``reject`` leaves this
                None so the stage's normal ``input_subject`` is reused.
        """
        from orchestrata_ultimate.orchestrator.pipeline_runner import (
            PipelineResult,
            StageStatus,
        )

        result = PipelineResult(
            thread_id=state.thread_id, work_type=state.work_type, intent=state.intent
        )
        stages = self.router.route(state.work_type, has_ui=state.has_ui)
        idx = next((i for i, s in enumerate(stages) if s.alias == state.stage), None)
        if idx is None:
            _logger.warning(
                "Escalated stage %s no longer in %s route order for thread %s",
                state.stage,
                state.work_type,
                state.thread_id,
            )
            result.overall_status = StageStatus.FAILED
            return result

        stage_spec = stages[idx]
        if subject_override:
            stage_spec = dataclasses.replace(stage_spec, input_subject=subject_override)
            stages[idx] = stage_spec

        self._register_handlers([stage_spec])
        # NOTE: ``_run_stage``'s "subsequent stage" packet builder derives the
        # REAL dispatched packet's from_agent from THIS packet's to_agent
        # (``from_agent=input_packet.to_agent``) — exactly like it does for
        # every chained stage in run()/resume() (``current_packet`` is the
        # PREVIOUS stage's own response, whose ``to_agent`` is conventionally
        # "HUMAN"). Setting to_agent="HUMAN" here makes the real re-entry
        # packet's from_agent "HUMAN" too, matching the kernel's own
        # respond()/reject() convention and what _collect_prior_human_answers
        # filters on.
        input_packet = CXPPacket(
            from_agent="HUMAN",
            to_agent="HUMAN",
            thread_id=state.thread_id,
            subject=subject_override or "response",
            payload=feedback_payload,
        )

        await self._execute_stages(
            result,
            stages,
            idx,
            input_packet,
            state.thread_id,
            state.intent,
            state.work_type,
            state.has_ui,
        )
        return result

    @traces(
        spec="000",
        story="US-006",
        reqs=["REQ-000-053", "REQ-000-054", "REQ-000-057", "REQ-000-058"],
    )
    async def approve(self, thread_id: str) -> PipelineResult:
        """Approve the pending escalation on ``thread_id``.

        The escalated stage's own output is already checkpointed RESOLVED
        (``_run_stage``/``_save_stage_checkpoint`` ran before the interrupt
        was classified) — approving just consumes the escalation marker and
        resumes the fixed sequence from the NEXT router stage via the
        existing ``resume()`` (REQ-001-961). No auto-approval anywhere
        (REQ-000-058): this is the only path that can advance an ESCALATED
        run past the point it suspended.
        """
        state = self._load_escalation(thread_id)
        if state is None:
            return _no_pending_escalation_result(thread_id)
        self._resolve_escalation(thread_id, state, "approved")
        result: PipelineResult = await self.resume(
            thread_id, work_type=state.work_type, has_ui=state.has_ui
        )
        return result

    @traces(
        spec="000", story="US-006", reqs=["REQ-000-055", "REQ-000-056", "REQ-000-057"]
    )
    async def reject(self, thread_id: str, reason: str) -> PipelineResult:
        """Reject the pending escalation on ``thread_id``, re-entering the
        SAME stage with ``reason`` folded into its payload (REQ-000-056:
        reject keeps the volk working with feedback, rather than advancing).
        """
        state = self._load_escalation(thread_id)
        if state is None:
            return _no_pending_escalation_result(thread_id)
        self._resolve_escalation(thread_id, state, "rejected")
        feedback_payload = {**state.payload, "feedback": reason, "rejected": True}
        return await self._reenter_escalated_stage(
            state, feedback_payload, subject_override=None
        )

    @traces(
        spec="000", story="US-004", reqs=["REQ-000-039", "REQ-000-040", "REQ-000-041"]
    )
    async def respond(self, thread_id: str, text: str) -> PipelineResult:
        """Deliver a human response to the escalated stage mid-conversation
        (e.g. Concierge's clarification loop) — re-enters the SAME stage
        under the "response" CXP subject with ``text``/``prior_answers``,
        exactly like the kernel's own ``respond()``. If the stage is still
        not done (asks another question), a fresh escalation is checkpointed
        by the same interrupt classification that handled the first one —
        callers keep responding until the stage finally resolves.
        """
        state = self._load_escalation(thread_id)
        if state is None:
            return _no_pending_escalation_result(thread_id)
        prior_answers = self._collect_prior_human_answers(thread_id)
        self._resolve_escalation(thread_id, state, "responded")
        feedback_payload = {
            **state.payload,
            "text": text,
            "prior_answers": prior_answers,
        }
        return await self._reenter_escalated_stage(
            state, feedback_payload, subject_override="response"
        )


@traces(spec="000", story="US-006", reqs=["REQ-000-058"])
def _no_pending_escalation_result(thread_id: str) -> PipelineResult:
    """FAILED PipelineResult shape returned by approve/reject/respond when
    there is no pending escalation for ``thread_id`` (never escalated, or
    already settled by a prior call — REQ-000-058)."""
    from orchestrata_ultimate.orchestrator.pipeline_runner import (
        PipelineResult,
        StageStatus,
    )

    result = PipelineResult(thread_id=thread_id, work_type="unknown", intent="")
    result.overall_status = StageStatus.FAILED
    return result