orchestrata_ultimate.kernel.pipeline_escalation

Reference › orchestrata_ultimate › kernel › pipeline_escalation

The mixin that moves a packet from volk to volk. When a volk's reply is addressed to another volk, PipelineEscalationMixin carries it on, hop by hop, until a reply goes to the human or no further target exists. pipeline mixes it into PipelineRunner, and it calls the session, knowledge-graph and audit mixins through a typed host protocol (_EscalationHost). It serves SPEC-001 story US-005 (REQ-001-US-005-014, -015, -017).

_route_escalation is the loop. It iterates instead of recursing, limited by max_escalation_depth, and for each packet it does the following, in order.

• _hop_cycle_detected counts the sender-recipient pair and stops when a pair repeats more than max_hop_repeats times.
• _apply_capability_routing rewrites to_agent when the payload names needs_capability, using resolve_capability, so a volk can ask for a capability instead of an alias; without the key, or without a provider, the fixed target stays.
• _announce_escalation emits the escalation event, logs the transition, collects artifacts and decisions, and writes an audit record.
• _apply_escalation_side_effects asserts any prolog_facts from the payload into the Prolog gate, and syncs the project directory to the graph when QA sends a compliance check.
• _maybe_inject_context adds graph context, then _route_one_hop calls the postman's route_sync, writing outbound and inbound message signals around it and an audit record of the routing decision. A volk exception shuts the runner down and ends the loop.
• _next_escalation_target passes a human-bound reply to _human_handler, recording an open question for a clarification, and continues only when the reply names another known volk.

approve handles the human's approval of a thread. It finds the active volk, asks the Prolog gate may_complete, logs the policy query and, on success, calls _dispatch_approval, which builds the next packet with _build_escalation for the volk's escalation_target (the concierge escalates to U:PM by default) or marks the work resolved when there is no target. The packet subject is review_and_test for QA and implement otherwise.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

PipelineEscalationMixin

Bases: _MixinBase · Decorators: traces(spec='001', story='US-005', reqs=['REQ-001-US-005-014', 'REQ-001-US-005-015']) · Line 105

Escalation routing helpers for PipelineRunner.

Source

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

# mypy: ignore-errors
"""PipelineRunner escalation routing mixin — the PACKET-level mesh.

Extracted from pipeline.py to keep PipelineRunner under the
300-line class limit. Contains the iterative escalation loop
and escalation packet builder.

BOUNDARY (Mohamed's ruling, 2026-08-09). This project has TWO capability
meshes, deliberately, because they operate at different granularities on
different entry points:

* THIS module routes at the PACKET level — ``_apply_capability_routing``
  rewrites ``packet.to_agent``, and ``_route_one_hop`` /
  ``_next_escalation_target`` / ``_hop_cycle_detected`` carry a packet through
  multiple hops. It serves the gRPC path: ``agents/kernel_handler.py`` and the
  kernel CLIs.
* ``orchestrator.pipeline_escalation`` routes at the STAGE level —
  ``_build_capability_stage`` inserts a ``StageSpec`` into the remaining
  sequence and ``_classify_interrupt`` caps the insertions. It serves the
  gateway/UI path (``gateway/server.py`` -> ``PipelineKernelClient``) and
  ``session pipeline approve|respond|reject``.

They are complements, not duplicates, and neither absorbs the other. Do NOT
add a ``StageSpec`` builder here — ``tests/core/test_capability_mesh.py``
fails if either side grows the other's idiom, because at that point the two
really would be duplicates and the decision to keep both would have been
reversed by accident.

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

import logging
from collections import deque
from collections.abc import Coroutine
from typing import TYPE_CHECKING, Any, Protocol

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

if TYPE_CHECKING:
    from orchestrata_ultimate.governance.prolog_gate import PrologGate
    from orchestrata_ultimate.kernel.audit.logger import AuditLogger
    from orchestrata_ultimate.kernel.pipeline_events import PipelineEventBus
    from orchestrata_ultimate.postman.postman import Postman

    class _EscalationHost(Protocol):
        """Host contract ``PipelineEscalationMixin`` expects from ``PipelineRunner``.

        ``PipelineRunner`` composes this mixin alongside ``PipelineSessionMixin``
        (``_session_log``, ``_session_log_transition``, ``_collect_artifacts``,
        ``_collect_decisions``, ``_human_handler``, ``shutdown``),
        ``PipelineKGMixin`` (``_maybe_inject_context``, ``_sync_to_graph``), and
        ``PipelineAuditMixin`` (``audit``, ``_audit_fire`` — SPEC-001 US-008
        audit trail), plus its own ``__init__`` attributes (``postman``,
        ``prolog``, ``events``, ``thread_ids``). Pyright type-checks each mixin
        against its own bare ``self``, so it can't see across the composition
        without this contract (the same "mixin self-type" trick as
        ``agents/volks/qa/_host.py:_QaHost``). Real only at ``TYPE_CHECKING``
        time — at runtime ``PipelineEscalationMixin`` still inherits plain
        ``object``; the composed ``PipelineRunner`` MRO supplies every member
        below via the sibling mixins listed above it in ``pipeline.py``.
        """

        postman: Postman
        prolog: PrologGate
        events: PipelineEventBus
        thread_ids: set[str]
        audit: AuditLogger | None

        def _session_log(self, desc: str, details: dict | None = None) -> None: ...
        def _session_log_transition(self, packet: CXPPacket) -> None: ...
        def _collect_artifacts(self, packet: CXPPacket) -> None: ...
        def _collect_decisions(self, packet: CXPPacket) -> None: ...
        def _human_handler(self, packet: CXPPacket) -> CXPPacket | None: ...
        def shutdown(self, reason: str = "Manual shutdown") -> None: ...
        def _maybe_inject_context(self, packet: CXPPacket) -> CXPPacket: ...
        def _sync_to_graph(self, project_dir: str) -> None: ...
        def _audit_fire(self, coro: Coroutine[Any, Any, Any]) -> None: ...

    _MixinBase = _EscalationHost
else:
    _MixinBase = object

_logger = logging.getLogger(__name__)

# Capability → built-in provider alias. The fallback map when a volk does not
# expose its own ``capabilities``; keeps capability-routed hops working for the
# nine built-in delivery volks. A plugin/registered volk that declares
# ``capabilities`` is discovered dynamically and wins over this map.
# Re-exported from core.capability_mesh, which owns the one copy. The table
# and the resolver were duplicated byte-for-byte here and in
# orchestrator.pipeline_escalation; two copies of a routing table is how the
# two meshes start disagreeing about who provides "testing". The name is kept
# so existing importers of this module's private constant keep working.
_CAPABILITY_PROVIDERS: dict[str, str] = CAPABILITY_PROVIDERS


@traces(spec="001", story="US-005", reqs=["REQ-001-US-005-014", "REQ-001-US-005-015"])
class PipelineEscalationMixin(_MixinBase):
    """Escalation routing helpers for PipelineRunner.

    Requires the following attributes on the host class (see ``_EscalationHost``
    above for the full typed contract):
    - postman, prolog, events, volks, human_inbox, context_injector
    - _session_log, _session_log_transition (from PipelineSessionMixin)
    - _maybe_inject_context, _sync_to_graph (from PipelineKGMixin)
    - _write_message_signal (from PipelineKGMixin) for Tier-H message signals
    - audit, _audit_fire (from PipelineAuditMixin) for SPEC-001 US-008 audit
      trail (REQ-008-001 message routing, REQ-008-002 policy queries,
      REQ-008-003 escalation) — read via getattr, this mixin stays decoupled
      from the audit mixin's presence.
    """

    volks: dict[str, Any]
    human_inbox: deque[CXPPacket]

    # Loop guards for ``_route_escalation``. Class-level so a host (or a test
    # that must exercise the limit branch against the real graph, where every
    # hop costs ~5 serial transactional graph writes) can lower them per
    # instance; the production values below are the contract.
    max_escalation_depth: int = 200
    max_hop_repeats: int = 99

    def _audit_or_none(self):
        """``self.audit`` when audit is on and the fire-and-forget helper is
        present, else ``None`` — the guard every call site checks before
        constructing an audit coroutine (avoids building one just to close
        it unused). Read via getattr: this mixin stays decoupled from
        PipelineAuditMixin's presence on the composed host class."""
        audit = getattr(self, "audit", None)
        if audit is None or not callable(getattr(self, "_audit_fire", None)):
            return None
        return audit

    def _fire_message_routing_audit(
        self, packet: CXPPacket, thread_id: str, *, route_decision: str
    ) -> None:
        """REQ-008-001 message-routing audit trail for one ``_route_one_hop``
        dispatch (best-effort, never blocks routing). ``log_message_routing``
        always records ``AuditOutcome.SUCCESS`` (a fixed property of the
        SPEC-001 US-008 ``AuditLogger`` API, not touched here) — success vs.
        crash is instead distinguished in the free-text ``route_decision``.
        """
        audit = self._audit_or_none()
        if audit is None:
            return
        self._audit_fire(
            audit.log_message_routing(
                session_id=thread_id,
                volk_id=packet.from_agent,
                message_id=getattr(packet, "message_id", "") or "",
                from_volk=packet.from_agent,
                to_volk=packet.to_agent,
                route_decision=route_decision,
            )
        )

    @traces(
        spec="001", story="US-005", reqs=["REQ-001-US-005-014", "REQ-001-US-005-015"]
    )
    def _route_escalation(self, packet: CXPPacket) -> None:
        """Route escalation packets iteratively.

        Avoids RecursionError by iterating rather than recursing.
        Includes cycle detection (visited-set) and a depth limit to
        prevent infinite loops.

        Args:
            packet: The initial escalation CXP packet to route.
        """
        max_depth = self.max_escalation_depth
        hop_counts: dict[tuple, int] = {}
        current: CXPPacket | None = packet
        depth = 0

        while current and depth < max_depth:
            if self._hop_cycle_detected(current, hop_counts):
                break

            depth += 1
            current = self._apply_capability_routing(current)
            self._announce_escalation(current)
            self._apply_escalation_side_effects(current)
            current = self._maybe_inject_context(current)

            response, crashed = self._route_one_hop(current)
            if crashed:
                return

            current = self._next_escalation_target(response)
            if current is None:
                break
        else:
            if depth >= max_depth:
                print(
                    f"  [LIMIT] Depth limit ({max_depth}) reached — breaking",
                    flush=True,
                )
                self._session_log(f"DEPTH LIMIT: {max_depth} hops exhausted")

    @traces(spec="001", story="US-001", reqs=["REQ-001-012"])
    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.

        The built-in ``_CAPABILITY_PROVIDERS`` map is authoritative for the core
        capabilities (so core routing is stable); 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 falls back to
        the fixed target).
        """
        return _resolve_capability(self.volks, capability)

    def _apply_capability_routing(self, packet: CXPPacket) -> CXPPacket:
        """Route a packet by the capability it requests, not a hardcoded alias.

        When a volk sets ``payload['needs_capability']``, the kernel resolves it
        to whichever volk provides that capability and rewrites ``to_agent`` —
        the additive volk mesh. Absent/unknown capability leaves the packet's
        existing ``to_agent`` untouched, so the fixed choreography is preserved.
        """
        cap = (packet.payload or {}).get("needs_capability")
        if not cap:
            return packet
        target = self.resolve_capability(cap)
        if target and target != packet.to_agent:
            print(
                f"  [Mesh] '{cap}' → {target} (capability-routed, "
                f"was {packet.to_agent})",
                flush=True,
            )
            packet.to_agent = target
        return packet

    def _hop_cycle_detected(
        self, packet: CXPPacket, hop_counts: dict[tuple, int]
    ) -> bool:
        """Record a hop and report whether it has repeated past the limit."""
        max_hop_repeats = self.max_hop_repeats
        hop = (packet.from_agent, packet.to_agent)
        hop_counts[hop] = hop_counts.get(hop, 0) + 1
        if hop_counts[hop] > max_hop_repeats:
            print(
                f"  [LOOP] Cycle: {packet.from_agent} → {packet.to_agent} "
                f"(seen {hop_counts[hop]} times) — breaking",
                flush=True,
            )
            self._session_log(f"LOOP: {packet.from_agent} → {packet.to_agent}")
            return True
        return False

    def _announce_escalation(self, packet: CXPPacket) -> None:
        """Emit and log the escalation event for a single hop."""
        target = packet.to_agent
        print(f"  [Escalation] {packet.from_agent} → {target}")
        print(f"  [Postman] Escalating to {target}")
        self.events.emit_escalation(
            packet.thread_id,
            packet.from_agent,
            target,
            packet.subject,
        )
        self._session_log(f"{packet.from_agent} → {target}: {packet.subject}")
        self._session_log_transition(packet)
        # Capture any artifacts this hop carries (queryable + disk + checkpoint).
        self._collect_artifacts(packet)
        # Capture any decisions this hop carries ((:Volk)-[:MADE]->(:Decision)).
        self._collect_decisions(packet)
        # REQ-008-003: escalation audit trail (best-effort, never blocks routing).
        audit = self._audit_or_none()
        if audit is not None:
            self._audit_fire(
                audit.log_escalation(
                    session_id=packet.thread_id,
                    volk_id=packet.from_agent,
                    issue=packet.subject,
                    severity="info",
                    escalation_path=[packet.from_agent, target],
                    resolved=False,
                )
            )

    def _apply_escalation_side_effects(self, packet: CXPPacket) -> None:
        """Assert any prolog facts and sync the graph on a compliance check."""
        if packet.payload.get("prolog_facts"):
            for fact in packet.payload["prolog_facts"]:
                self.prolog.assert_fact(fact)
                print(f"  [Prolog] Asserted: {fact}")

        if packet.from_agent == "U:Qa" and packet.subject == "compliance_check":
            project_dir = packet.payload.get("project_dir", "")
            if project_dir and __import__("os").path.isdir(project_dir):
                self._sync_to_graph(project_dir)

    def _route_one_hop(self, packet: CXPPacket) -> tuple[CXPPacket | None, bool]:
        """Route one packet, handling a volk crash.

        Returns ``(response, crashed)``. On crash the runner is shut down and
        ``crashed`` is True (caller should stop routing).
        """
        target = packet.to_agent
        thread_id = packet.thread_id

        # Tier-H signal: record the outbound escalation message.
        write_message = getattr(self, "_write_message_signal", None)
        if callable(write_message):
            try:
                write_message(
                    packet.payload,
                    sender=packet.from_agent,
                    recipient=packet.to_agent,
                    subject=packet.subject,
                    thread_id=thread_id,
                    direction="out",
                )
            except Exception as exc:
                _logger.debug("Outbound message signal write failed: %s", exc)

        try:
            response = self.postman.route_sync(packet)
        except Exception as exc:
            print(f"  [CRASH] {target}.process() failed: {exc}", flush=True)
            self._session_log(f"CRASH in {target}: {exc}")
            self._fire_message_routing_audit(
                packet, thread_id, route_decision=f"crashed: {exc}"
            )
            self.shutdown(reason=f"Volk {target} crashed: {exc}")
            return None, True

        self._fire_message_routing_audit(packet, thread_id, route_decision="delivered")

        # Tier-H signal: record the inbound response message.
        if callable(write_message) and response is not None:
            try:
                write_message(
                    response.payload if isinstance(response.payload, dict) else {},
                    sender=response.from_agent,
                    recipient=response.to_agent,
                    subject=response.subject,
                    thread_id=thread_id,
                    direction="in",
                )
            except Exception as exc:
                _logger.debug("Inbound message signal write failed: %s", exc)

        return response, False

    def _next_escalation_target(self, response: CXPPacket | None) -> CXPPacket | None:
        """Resolve the next packet to route, or None to stop.

        Routes HUMAN-bound responses through ``_human_handler`` (so terminal
        outcomes like ``qa_complete``/``compliance_complete``/``compliance_failed``
        emit their ``pipeline.complete`` event and end the session) and continues
        only when the response targets another known volk.
        """
        if not response:
            return None
        if response.to_agent == "HUMAN":
            # F8 fix: a packet with subject "clarification" IS a genuine
            # open question (the volk needs human input before it can
            # complete) -- see ConciergeVolk, kernel/README.md and
            # pipeline_cli.py's own "== 'clarification'" check. Marking it
            # here is what makes PrologGate.may_complete's no_open_questions
            # gate actually able to DENY (kernel.pipeline_escalation.approve
            # used to self-assert task_has_output right at the check site
            # instead, so the completion gate could never fail on a real
            # unanswered question). respond() retracts the matching mark
            # when the human actually answers.
            if response.subject == "clarification":
                prolog = getattr(self, "prolog", None)
                if prolog is not None:
                    try:
                        prolog.assert_open_question(
                            response.from_agent, f"task_{response.thread_id}"
                        )
                    except Exception as exc:
                        _logger.debug("assert_open_question failed: %s", exc)
            # Delegate to _human_handler (PipelineSessionMixin): it appends to
            # the inbox AND fires terminal-stage side effects (complete events,
            # session end). Routing the response here instead of a bare append
            # is what makes a compliance violation a first-class FAILED outcome.
            self._human_handler(response)
            print(
                f"  [{response.from_agent}] → HUMAN (subject={response.subject})",
                flush=True,
            )
            return None
        if response.to_agent in self.volks:
            return response
        return None

    @traces(
        spec="000",
        story="US-006",
        reqs=[
            "REQ-000-053",
            "REQ-000-054",
            "REQ-000-057",
            "REQ-000-058",
        ],
    )
    def approve(self, thread_id: str) -> bool:
        """Approve work on a thread, routing escalation if needed.

        F8 fix: this used to assert_task_output() for the active volk
        immediately before checking may_complete() — a self-fulfilling
        gate (it could only ever fail on the open-questions leg, never on
        the acceptance-criteria/output-verified legs, since it had just
        made those true itself). task_has_output is now asserted where a
        volk's output is actually produced
        (PipelineSessionMixin._collect_artifacts, driven by real artifact
        payloads), so may_complete here reflects genuine pipeline state —
        a volk with no collected artifacts for this thread is genuinely
        denied.

        Args:
            thread_id: The thread to approve.

        Returns:
            True if approval succeeded, False otherwise.
        """
        if thread_id not in self.thread_ids:
            print(f"  [Error] Thread {thread_id} not found")
            return False

        active_alias, active_volk = self._find_active_volk(thread_id)
        if active_volk is None:
            print(f"  [Error] No active volk on thread {thread_id}")
            return False
        # _find_active_volk always returns the alias/volk pair together (both
        # None or both set) — this documents that invariant for pyright.
        assert active_alias is not None

        task_id = f"task_{thread_id}"
        allowed, gates = self.prolog.may_complete(active_alias, task_id)

        print(f"\n  [Prolog] may_complete({active_alias}, {task_id}):")
        for gate in gates:
            print(f"    {gate}")

        # REQ-008-002: policy-query audit trail (best-effort, never blocks
        # the completion gate).
        audit = self._audit_or_none()
        if audit is not None:
            self._audit_fire(
                audit.log_policy_query(
                    session_id=thread_id,
                    volk_id=active_alias,
                    query=f"may_complete({active_alias}, {task_id})",
                    result=str(allowed),
                    cached=False,
                )
            )

        if not allowed:
            print("  → FALSE — volk must continue")
            return False

        print("  → TRUE")
        self._dispatch_approval(active_alias, active_volk, thread_id)
        return True

    @traces(spec="000", story="US-006", reqs=["REQ-000-053"])
    def _find_active_volk(self, thread_id: str) -> tuple[str | None, Any | None]:
        """Find the deepest IN_PROGRESS volk on a thread.

        Args:
            thread_id: Thread to search.

        Returns:
            Tuple of (alias, volk) or (None, None) if not found.
        """
        active_alias, active_volk = None, None
        fallback_alias, fallback_volk = None, None
        for alias, volk in self.volks.items():
            if volk.current_thread != thread_id:
                continue
            if volk.status == WorkStatus.IN_PROGRESS:
                active_alias, active_volk = alias, volk
            elif fallback_volk is None:
                fallback_alias, fallback_volk = alias, volk
        if active_volk is None and fallback_volk is not None:
            return fallback_alias, fallback_volk
        return active_alias, active_volk

    @traces(spec="000", story="US-006", reqs=["REQ-000-053", "REQ-000-054"])
    def _dispatch_approval(self, alias: str, volk: Any, thread_id: str) -> None:
        """Route escalation or mark work complete after approval.

        Args:
            alias: The approving volk's alias.
            volk: The approving volk instance.
            thread_id: The active thread.
        """
        if hasattr(volk, "escalation_target") and volk.escalation_target:
            target = volk.escalation_target
        elif alias == "U:Cx":
            target = "U:PM"
        else:
            target = None

        if target and target in self.volks:
            escalation = self._build_escalation(alias, target, thread_id, volk)
            volk.status = WorkStatus.RESOLVED
            volk.summary = f"APPROVED — escalating to {target}"
            self._route_escalation(escalation)
        else:
            volk.status = WorkStatus.RESOLVED
            volk.summary = "APPROVED — work complete"
            print("  ✓ Work approved and marked RESOLVED")

    @traces(spec="001", story="US-005", reqs=["REQ-001-US-005-017"])
    def _build_escalation(
        self,
        from_alias: str,
        target: str,
        thread_id: str,
        volk: Any,
    ) -> CXPPacket:
        """Build an escalation CXPPacket from one volk to another.

        Constructs the correct payload structure depending on the
        from/target volk pair.

        Args:
            from_alias: The alias of the escalating volk.
            target: The alias of the destination volk.
            thread_id: The active thread identifier.
            volk: The escalating volk instance.

        Returns:
            A CXPPacket ready to route.
        """
        if target == "U:Qa":
            return CXPPacket(
                from_agent=from_alias,
                to_agent=target,
                thread_id=thread_id,
                subject="review_and_test",
                payload={
                    "files_created": getattr(volk, "files_created", []),
                    "task_summary": getattr(volk, "task_summary", ""),
                    "project_dir": getattr(volk, "project_dir", ""),
                    "structured_requirements": getattr(
                        volk, "structured_requirements", []
                    ),
                    "user_stories": getattr(volk, "user_stories", []),
                },
            )
        if target == "U:Cd" and from_alias == "U:Ar":
            return CXPPacket(
                from_agent=from_alias,
                to_agent=target,
                thread_id=thread_id,
                subject="implement",
                payload={
                    "architecture": getattr(volk, "design_docs", {}).get(
                        "architecture", ""
                    ),
                    "requirements": getattr(volk, "design_docs", {}).get(
                        "requirements", ""
                    ),
                    "file_tree": getattr(volk, "design_docs", {}).get("file_tree", ""),
                    "api_spec": getattr(volk, "design_docs", {}).get("api_spec", ""),
                    "project_dir": getattr(volk, "project_dir", ""),
                    "task_summary": getattr(volk, "task_summary", ""),
                    "structured_requirements": getattr(
                        volk, "structured_requirements", []
                    ),
                    "user_stories": getattr(volk, "user_stories", []),
                },
            )
        return CXPPacket(
            from_agent=from_alias,
            to_agent=target,
            thread_id=thread_id,
            subject="implement",
            payload={
                "requirements": getattr(volk, "requirements", ""),
                "task_summary": getattr(volk, "task_summary", ""),
                "prolog_facts": getattr(volk, "prolog_facts", []),
            },
        )