orchestrata_ultimate.orchestrator.verification_gate

Reference › orchestrata_ultimate › orchestrator › verification_gate

The adapter that plugs the verification subsystem (see the verification module) into the stage loop of pipeline_runner, which is its only importer. PipelineVerificationGate composes collaborators from the verification package (session manager, engine, coordinator, escalation handler, Redis manager) and injects real defaults for each; the only fallible part, the Redis client, degrades to a disconnected manager.

VerificationGateConfig.from_env reads VERIFY_ENABLED (default on), VERIFY_ENFORCE (default off, so the default posture is audit-only) and VERIFY_MAX_ATTEMPTS (default 5, the escalation threshold). The gate has three moments in a run. open_session derives constraints from the intent and work type through a ConstraintSource (the default WorkTypeConstraintSource uses the verification intent broker adapter), opens a session, starts the workflow and persists it. after_stage maps the finished stage's capability to a verification checkpoint with checkpoint_for, builds a context from the stage artifacts with an ArtifactContextBuilder (which checks that code parses and folds in QA payload signals), and verifies that checkpoint. finalize runs the full verification over all artifacts and returns a VerificationVerdict with failed constraint ids, reasons and a recommendation; only with enforcement on does it call the escalation handler for a human. Every step is wrapped so a verification failure is logged and returns None, never raising into the pipeline.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

VerificationGateConfig (dataclass)

Decorators: traces(spec='035', story='US-008', reqs=['REQ-035-US-008-002', 'REQ-035-US-008-004', 'REQ-035-US-008-005']), dataclass(frozen=True, slots=True) · Line 100

Env-driven configuration for :class:PipelineVerificationGate.

VerificationVerdict (dataclass)

Decorators: traces(spec='035', story='US-005', reqs=['REQ-035-US-003-005', 'REQ-035-US-008-003', 'REQ-035-US-009-004']), dataclass · Line 138

The gate's final-checkpoint verdict for one pipeline run.

ConstraintSource (protocol)

Bases: Protocol · Line 178

Strategy for deriving verification constraints from a pipeline run's intent + work type (REQ-035-US-009-002/003).

WorkTypeConstraintSource

Decorators: traces(spec='035', story='US-009', reqs=['REQ-035-US-009-002', 'REQ-035-US-009-003', 'REQ-035-US-009-010']) · Line 231

Default :class:ConstraintSource: derives a per-WorkType constraint set (MVP — PipelineRunner.run receives intent as a bare string, no structured requirements in production yet) and…

StageContextBuilder (protocol)

Bases: Protocol · Line 275

Strategy building the ConstraintEvaluator context dict from a stage's (or the run's) collected artifacts (REQ-035-US-003-001).

ArtifactContextBuilder

Decorators: traces(spec='035', story='US-003', reqs=['REQ-035-US-003-001', 'REQ-035-US-003-005', 'REQ-035-US-003-014']) · Line 295

Default :class:StageContextBuilder.

PipelineVerificationGate

Decorators: traces(spec='035', story='US-004', reqs=['REQ-035-US-004-002', 'REQ-035-US-004-011', 'REQ-035-US-008-001', 'REQ-035-US-… · Line 435

Composition-only adapter wiring the verification subsystem into PipelineRunner's stage loop.

Source

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

"""PipelineVerificationGate — production wiring for the verification subsystem.

SPEC-035 names five checkpoint milestones (core implementation, unit tests,
integration tests, documentation, final verification) verified concurrently
with execution, with results flowing back to the Orchestrator and escalation
to a human after ``max_attempts`` failures (US-004/US-008/US-009). Before this
module, ``verification/`` (51 modules) had zero production construction
sites — fully built and tested, but unreachable from any live pipeline run.

This module is a composition-only adapter: it wires the EXISTING verification
subsystem (never regenerated) into ``orchestrator.pipeline_runner.
PipelineRunner``'s stage loop, the one place in production that already
realizes the spec's checkpoint sequence 1:1 (``Capability.IMPLEMENTATION``
== ``CORE_IMPLEMENTATION``, ``TESTING`` == ``UNIT_TESTS``, ``TRACEABILITY``
== ``INTEGRATION_TESTS``, ``DOCUMENTATION`` == ``DOCUMENTATION``, pipeline
completion == ``FINAL_VERIFICATION``).

Construction mirrors the existing P0.1 ``PrologGate`` pattern used by
``PipelineRunner.__init__`` (see pipeline_runner.py): every collaborator is
DI-injectable with a real default, and the one genuinely fallible piece of
construction (a real Redis client) degrades to a disconnected
``VerificationRedisManager`` on failure rather than crashing the gate.

Rollout posture (dispatcher ``policy_audit_mode`` precedent): AUDIT-ONLY by
default (``VerificationGateConfig.enforce=False``) — verdicts are computed
and attached to every ``PipelineResult`` but never fail or escalate a run
until ``VERIFY_ENFORCE=1`` is set. Enforcement, when enabled, reuses
``PipelineRunner``'s existing D1-addendum-2 escalation-as-interrupt seam
(``_save_escalation`` / ``StageStatus.ESCALATED``) — no new transport.

Dormant slice (deliberately, this contract — see decision memo in
``plans/contracts/spec-007b-wiring.contract.json``): the LLM tri-agent layer
(``agents/*``, ``models/*``), proofs, edge cases, and the NL requirements
capture layer stay unwired. ``after_stage``/``finalize`` are the exact seams
where a future ``VerificationAgent.analyze_via_llm`` plugs in
(``VerificationCoordinator.build_verification_result`` already accepts an
``analysis`` payload for this).
"""

from __future__ import annotations

import ast
import logging
import os
from collections.abc import Mapping
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Any, Protocol

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.tracing.tracer import Tracer
from orchestrata_ultimate.orchestrator.artifact_store import (
    ArtifactType,
    PipelineArtifact,
)
from orchestrata_ultimate.orchestrator.pipeline_router import (
    Capability,
    StageSpec,
    WorkType,
)
from orchestrata_ultimate.verification.escalation_handler import EscalationHandler
from orchestrata_ultimate.verification.integrations.intent_broker import (
    IntentBrokerAdapter,
)
from orchestrata_ultimate.verification.integrations.redis_manager import (
    VerificationRedisManager,
)
from orchestrata_ultimate.verification.orchestration.coordinator import (
    VerificationCoordinator,
)
from orchestrata_ultimate.verification.review.checks_safety import check_security
from orchestrata_ultimate.verification.session.verification_session import (
    VerificationSessionManager,
)
from orchestrata_ultimate.verification.types import (
    CheckpointName,
    Constraint,
    SessionStatus,
)
from orchestrata_ultimate.verification.verification.engine import VerificationEngine

if TYPE_CHECKING:
    from orchestrata_ultimate.orchestrator.pipeline_runner import PipelineResult
    from orchestrata_ultimate.verification.types import CheckpointResult

logger = logging.getLogger(__name__)

_TRUTHY = {"1", "true", "yes", "on"}
_FALSY = {"0", "false", "no", "off"}


# ── VerificationGateConfig ───────────────────────────────────────────────


@traces(
    spec="035",
    story="US-008",
    reqs=["REQ-035-US-008-002", "REQ-035-US-008-004", "REQ-035-US-008-005"],
)
@dataclass(frozen=True, slots=True)
class VerificationGateConfig:
    """Env-driven configuration for :class:`PipelineVerificationGate`.

    Default posture is AUDIT-ONLY (``enforce=False``) — the staged rollout
    precedent set by ``kernel.dispatcher``'s ``policy_audit_mode``.
    """

    enabled: bool = True
    enforce: bool = False
    max_attempts: int = 5

    @traces(spec="035", story="US-008", reqs=["REQ-035-US-008-004"])
    @classmethod
    def from_env(cls, env: Mapping[str, str]) -> VerificationGateConfig:
        """Build config from an environment mapping (read at call time).

        ``VERIFY_ENABLED`` (default true), ``VERIFY_ENFORCE`` (default
        FALSE — audit-only), ``VERIFY_MAX_ATTEMPTS`` (default 5, the
        spec's own escalation threshold).
        """
        enabled = env.get("VERIFY_ENABLED", "true").strip().lower() not in _FALSY
        enforce = env.get("VERIFY_ENFORCE", "0").strip().lower() in _TRUTHY
        try:
            max_attempts = int(env.get("VERIFY_MAX_ATTEMPTS", "5"))
        except ValueError:
            max_attempts = 5
        return cls(enabled=enabled, enforce=enforce, max_attempts=max_attempts)


# ── VerificationVerdict ───────────────────────────────────────────────────


@traces(
    spec="035",
    story="US-005",
    reqs=["REQ-035-US-003-005", "REQ-035-US-008-003", "REQ-035-US-009-004"],
)
@dataclass
class VerificationVerdict:
    """The gate's final-checkpoint verdict for one pipeline run.

    Shape mirrors the spec's CXP Verification Result payload.
    """

    session_id: str
    decision: str  # "SATISFIED" | "NOT_MET"
    checkpoint_results: dict[str, CheckpointResult] = field(default_factory=dict)
    failed_constraints: list[str] = field(default_factory=list)
    failure_reasons: dict[str, str] = field(default_factory=dict)
    attempt_count: int = 0
    recommendation: str = "CONTINUE_ITERATING"  # | "COMPLETE" | "ESCALATE"

    @traces(spec="035", story="US-009", reqs=["REQ-035-US-009-004"])
    def to_dict(self) -> dict[str, Any]:
        return {
            "session_id": self.session_id,
            "decision": self.decision,
            "checkpoint_results": {
                name: {
                    "checkpoint": cp.checkpoint.value,
                    "passed": cp.passed,
                    "total_constraints": cp.total_constraints,
                    "satisfied_count": cp.satisfied_count,
                    "pass_rate": cp.pass_rate,
                    "details": cp.details,
                }
                for name, cp in self.checkpoint_results.items()
            },
            "failed_constraints": self.failed_constraints,
            "failure_reasons": self.failure_reasons,
            "attempt_count": self.attempt_count,
            "recommendation": self.recommendation,
        }


# ── ConstraintSource strategy ────────────────────────────────────────────


class ConstraintSource(Protocol):
    """Strategy for deriving verification constraints from a pipeline run's
    intent + work type (REQ-035-US-009-002/003)."""

    def constraints_for(self, intent: str, work_type: str) -> list[Constraint]: ...


# Every predicate below MUST appear in some CHECKPOINT_CRITERIA value —
# WorkTypeConstraintSource is a total mapping using ONLY predicates the
# reused checkpoint verifier already knows how to route.
_WORKTYPE_PREDICATES: dict[WorkType, list[str]] = {
    WorkType.FEATURE: [
        "file_exists",
        "code_compiles",
        "test_coverage_percentage",
        "linting_passes",
        "documentation_complete",
        "secrets_not_in_code",
    ],
    WorkType.REFACTOR: [
        "file_exists",
        "code_compiles",
        "test_coverage_percentage",
        "linting_passes",
    ],
    WorkType.TEST: ["test_coverage_percentage", "linting_passes"],
    WorkType.DOCUMENTATION: ["documentation_complete"],
    WorkType.ANALYSIS: ["linting_passes"],
    WorkType.FIX: ["file_exists", "code_compiles", "test_coverage_percentage"],
    WorkType.REVIEW: ["linting_passes"],
    WorkType.SPEC: ["documentation_complete"],
    WorkType.DEPENDENCY: ["dependencies_defined", "code_compiles"],
    WorkType.PERFORMANCE: ["response_time_max_ms"],
}

_PREDICATE_CATEGORY: dict[str, str] = {
    "test_coverage_percentage": "quality",
    "linting_passes": "quality",
    "documentation_complete": "quality",
    "secrets_not_in_code": "security",
    "no_sql_injection_vulnerabilities": "security",
    "response_time_max_ms": "performance",
    "file_exists": "functional",
    "code_compiles": "functional",
    "dependencies_defined": "functional",
}


@traces(
    spec="035",
    story="US-009",
    reqs=["REQ-035-US-009-002", "REQ-035-US-009-003", "REQ-035-US-009-010"],
)
class WorkTypeConstraintSource:
    """Default :class:`ConstraintSource`: derives a per-``WorkType``
    constraint set (MVP — ``PipelineRunner.run`` receives intent as a bare
    string, no structured requirements in production yet) and delegates
    predicate -> ``Constraint`` conversion to the existing
    :class:`IntentBrokerAdapter` (REUSE, never re-implemented).
    """

    def __init__(
        self,
        broker: IntentBrokerAdapter | None = None,
        tracer: Tracer | None = None,
    ) -> None:
        self._broker = broker or IntentBrokerAdapter()
        self._tracer = tracer or Tracer.default()

    @traces(
        spec="035",
        story="US-009",
        reqs=["REQ-035-US-009-002", "REQ-035-US-009-003", "REQ-035-US-009-010"],
    )
    def constraints_for(self, intent: str, work_type: str) -> list[Constraint]:
        with self._tracer.span("WorkTypeConstraintSource.constraints_for"):
            try:
                wt = WorkType(work_type)
            except ValueError:
                wt = WorkType.FEATURE
            predicates = _WORKTYPE_PREDICATES.get(wt, ["file_exists"])
            requirements = [
                {
                    "predicate": predicate,
                    "category": _PREDICATE_CATEGORY.get(predicate, "functional"),
                    "expected": True,
                    "description": f"{wt.value}: {predicate} (from intent: {intent[:80]!r})",
                }
                for predicate in predicates
            ]
            intent_dict = {"action": str(work_type), "requirements": requirements}
            return self._broker.intent_to_constraints(intent_dict)


# ── StageContextBuilder strategy ─────────────────────────────────────────


class StageContextBuilder(Protocol):
    """Strategy building the ``ConstraintEvaluator`` context dict from a
    stage's (or the run's) collected artifacts (REQ-035-US-003-001)."""

    def build(
        self,
        stage_alias: str,
        artifacts: list[PipelineArtifact],
        result: PipelineResult,
    ) -> dict[str, Any]: ...


_QA_PAYLOAD_PREDICATES = ("test_coverage_percentage", "linting_passes")


@traces(
    spec="035",
    story="US-003",
    reqs=["REQ-035-US-003-001", "REQ-035-US-003-005", "REQ-035-US-003-014"],
)
class ArtifactContextBuilder:
    """Default :class:`StageContextBuilder`.

    ``ConstraintEvaluator.evaluate`` reads ``context[constraint.predicate]``
    as the ACTUAL value. This builder computes predicate actuals from what
    the pipeline genuinely produced: ``file_exists``/``code_compiles`` from
    CODE artifacts (``ast.parse`` the content), ``secrets_not_in_code`` via
    the existing ``checks_safety.check_security`` heuristic, and
    ``documentation_complete`` from DOCUMENTATION artifacts.
    ``test_coverage_percentage``/``linting_passes`` are read straight off a
    stage's output payload when a volk already reports them under those
    keys. Any signal genuinely absent is OMITTED (never fabricated) — the
    evaluator then reports ERROR "No value provided", which is honest,
    never a fake pass.
    """

    _CODE_TYPES = (ArtifactType.SOURCE_CODE, ArtifactType.TEST_CODE)

    def __init__(self, tracer: Tracer | None = None) -> None:
        self._tracer = tracer or Tracer.default()

    @traces(spec="035", story="US-003", reqs=["REQ-035-US-003-001"])
    def build(
        self,
        stage_alias: str,
        artifacts: list[PipelineArtifact],
        result: PipelineResult,
    ) -> dict[str, Any]:
        with self._tracer.span("ArtifactContextBuilder.build"):
            context: dict[str, Any] = {}
            file_exists = False
            code_compiles: bool | None = None
            secrets_clean: bool | None = None
            doc_complete = False

            for artifact in artifacts:
                if artifact.artifact_type in self._CODE_TYPES:
                    file_exists = True
                    ok, tree = self._parses(artifact.content)
                    code_compiles = (
                        ok if code_compiles is None else (code_compiles and ok)
                    )
                    if tree is not None:
                        clean = self._secrets_clean(artifact, tree)
                        if clean is not None:
                            secrets_clean = (
                                clean
                                if secrets_clean is None
                                else (secrets_clean and clean)
                            )
                elif artifact.artifact_type == ArtifactType.DOC_OUTPUT:
                    doc_complete = bool(artifact.content.strip())

            if file_exists:
                context["file_exists"] = True
            if code_compiles is not None:
                context["code_compiles"] = code_compiles
            if secrets_clean is not None:
                context["secrets_not_in_code"] = secrets_clean
            if doc_complete:
                context["documentation_complete"] = True

            self._fold_qa_payload_signals(result, context)
            return context

    @staticmethod
    @traces(spec="035", story="US-003", reqs=["REQ-035-US-003-001"])
    def _parses(content: str) -> tuple[bool, ast.Module | None]:
        try:
            tree = ast.parse(content)
        except SyntaxError:
            return False, None
        return True, tree

    @staticmethod
    @traces(spec="035", story="US-003", reqs=["REQ-035-US-003-001"])
    def _secrets_clean(artifact: PipelineArtifact, tree: ast.Module) -> bool | None:
        try:
            lines = artifact.content.splitlines()
            issues = check_security(artifact.stage_alias, tree, lines)
        except Exception as exc:
            logger.debug(
                "secrets heuristic failed for %s: %s", artifact.stage_alias, exc
            )
            return None
        return not issues

    @staticmethod
    @traces(spec="035", story="US-003", reqs=["REQ-035-US-003-001"])
    def _fold_qa_payload_signals(
        result: PipelineResult, context: dict[str, Any]
    ) -> None:
        """Read ``test_coverage_percentage``/``linting_passes`` straight off
        any stage's output payload when a volk already reports them there —
        forward-looking, never fabricated when absent."""
        for stage_result in result.stages.values():
            packet = stage_result.output_packet
            payload = packet.payload if packet is not None else None
            if not isinstance(payload, dict):
                continue
            for predicate in _QA_PAYLOAD_PREDICATES:
                if predicate not in context and predicate in payload:
                    context[predicate] = payload[predicate]


# ── PipelineVerificationGate ──────────────────────────────────────────────

# Total mapping over pipeline capabilities: which checkpoint (if any) a
# stage's capability realizes. INTAKE/ANALYSIS/SCOPING/ARCHITECTURE/
# UI_DESIGN are pre-solution stages with no checkpoint of their own.
# FINAL_VERIFICATION belongs to finalize(), never a stage.
_CAPABILITY_CHECKPOINTS: dict[str, CheckpointName] = {
    Capability.IMPLEMENTATION.value: CheckpointName.CORE_IMPLEMENTATION,
    Capability.PARALLEL.value: CheckpointName.CORE_IMPLEMENTATION,
    Capability.TESTING.value: CheckpointName.UNIT_TESTS,
    Capability.TRACEABILITY.value: CheckpointName.INTEGRATION_TESTS,
    Capability.DOCUMENTATION.value: CheckpointName.DOCUMENTATION,
}

# Synthetic stage_alias finalize() passes to the context builder — there is
# no real StageSpec at pipeline completion, only the accumulated result.
_FINAL_STAGE_ALIAS = "__final__"


@traces(
    spec="035",
    story="US-004",
    reqs=[
        "REQ-035-US-004-002",
        "REQ-035-US-004-011",
        "REQ-035-US-008-001",
        "REQ-035-US-008-006",
        "REQ-035-US-008-011",
        "REQ-035-US-008-014",
        "REQ-035-US-009-001",
        "REQ-035-US-009-004",
        "REQ-035-US-009-005",
        "REQ-035-US-009-012",
    ],
)
class PipelineVerificationGate:
    """Composition-only adapter wiring the verification subsystem into
    ``PipelineRunner``'s stage loop.

    Every collaborator is DI-injected with a real default (compose, never
    mock). Construction is non-fatal: the only genuinely fallible piece —
    a real Redis client for persistence — degrades to a disconnected
    ``VerificationRedisManager`` on failure (the P0.1 ``PrologGate``
    pattern: risky I/O is guarded, not the whole object graph).
    """

    def __init__(
        self,
        config: VerificationGateConfig | None = None,
        sessions: VerificationSessionManager | None = None,
        engine: VerificationEngine | None = None,
        coordinator: VerificationCoordinator | None = None,
        escalation: EscalationHandler | None = None,
        constraint_source: ConstraintSource | None = None,
        context_builder: StageContextBuilder | None = None,
        redis_manager: VerificationRedisManager | None = None,
        tracer: Tracer | None = None,
        env: Mapping[str, str] | None = None,
    ) -> None:
        self._tracer = tracer or Tracer.default()
        self._config = config or VerificationGateConfig.from_env(
            os.environ if env is None else env
        )
        self._sessions = sessions or VerificationSessionManager()
        self._engine = engine or VerificationEngine()
        self._coordinator = coordinator or VerificationCoordinator(
            max_attempts=self._config.max_attempts
        )
        self._escalation = escalation or EscalationHandler(
            max_attempts=self._config.max_attempts
        )
        self._constraint_source = constraint_source or WorkTypeConstraintSource(
            tracer=self._tracer
        )
        self._context_builder = context_builder or ArtifactContextBuilder(
            tracer=self._tracer
        )
        self._checkpoint_history: dict[str, dict[str, CheckpointResult]] = {}
        if redis_manager is not None:
            self._redis = redis_manager
        else:
            self._redis = VerificationRedisManager(redis=self._connect_redis(env))

    @staticmethod
    def _connect_redis(env: Mapping[str, str] | None) -> Any:
        """Best-effort real Redis client — non-fatal (P0.1 pattern): a
        misconfigured/unreachable Redis degrades persistence to a no-op
        rather than crashing gate construction."""
        try:
            from orchestrata_ultimate.core.redis_factory import create_sync_redis

            return create_sync_redis(env=env)
        except Exception as exc:
            logger.warning(
                "VerificationRedisManager: Redis unavailable (%s) — "
                "session persistence disabled",
                exc,
            )
            return None

    @property
    def enabled(self) -> bool:
        return self._config.enabled

    @property
    def enforce(self) -> bool:
        return self._config.enabled and self._config.enforce

    @traces(
        spec="035",
        story="US-004",
        reqs=["REQ-035-US-004-001", "REQ-035-US-004-003"],
    )
    def checkpoint_for(self, capability: str) -> CheckpointName | None:
        """Total mapping over pipeline capabilities -> checkpoint name (or
        ``None`` for pre-solution stages that verify nothing)."""
        return _CAPABILITY_CHECKPOINTS.get(str(capability))

    @traces(
        spec="035",
        story="US-009",
        reqs=["REQ-035-US-009-001", "REQ-035-US-001-010"],
    )
    def open_session(self, intent: str, work_type: str, thread_id: str) -> str | None:
        """Open a verification session for a pipeline run.

        ``constraint_source.constraints_for`` derives the constraint set,
        ``sessions.create_session`` opens the session, and
        ``coordinator.start_workflow`` starts the CXP-visible workflow.
        Never raises into the pipeline — returns ``None`` (verification
        skipped for this run) on any failure.
        """
        if not self._config.enabled:
            return None
        with self._tracer.span("PipelineVerificationGate.open_session"):
            try:
                constraints = self._constraint_source.constraints_for(
                    intent, str(work_type)
                )
                session = self._sessions.create_session(
                    constraints, max_attempts=self._config.max_attempts
                )
                self._coordinator.start_workflow(session)
                self._checkpoint_history[session.id] = {}
                self._persist(session.id)
                return session.id
            except Exception as exc:
                logger.warning(
                    "open_session failed for thread %s (%s) — verification "
                    "disabled for this run",
                    thread_id,
                    exc,
                )
                return None

    @traces(
        spec="035",
        story="US-004",
        reqs=["REQ-035-US-004-002", "REQ-035-US-004-011", "REQ-035-US-004-004"],
    )
    def after_stage(
        self,
        session_id: str,
        stage_spec: StageSpec,
        artifacts: list[PipelineArtifact],
        result: PipelineResult,
    ) -> CheckpointResult | None:
        """Verify the checkpoint (if any) a just-completed stage realizes.

        ``VerificationEngine.verify_checkpoint`` verifies whatever
        ``session.current_checkpoint`` already is (it takes no checkpoint
        argument) — this method drives that value to ``checkpoint_for``'s
        capability-mapped checkpoint FIRST, so the checkpoint actually
        verified always matches the stage that just ran, deterministically,
        rather than depending on the engine's own auto-advance-on-pass
        sequencing having stayed in lockstep with the router's stage order.

        Never raises into the stage loop; audit-only recording unless
        ``config.enforce`` (enforcement itself is the caller's -- i.e.
        ``PipelineRunner``'s -- responsibility via ``finalize``'s verdict).
        """
        if not self._config.enabled or not session_id:
            return None
        checkpoint = self.checkpoint_for(stage_spec.capability)
        if checkpoint is None:
            return None
        session = self._sessions.get_session(session_id)
        if session is None:
            return None
        with self._tracer.span("PipelineVerificationGate.after_stage"):
            try:
                session.current_checkpoint = checkpoint
                context = self._context_builder.build(
                    stage_spec.alias, artifacts, result
                )
                cp_result = self._engine.verify_checkpoint(session, context)
            except Exception as exc:
                logger.warning(
                    "after_stage verification failed for %s (%s)",
                    stage_spec.alias,
                    exc,
                )
                return None
            self._checkpoint_history.setdefault(session_id, {})[
                cp_result.checkpoint.value
            ] = cp_result
            # Move the CXP-visible workflow off constraint_evaluation the
            # first time a checkpoint is actually verified. Guarded on the
            # current step because this method runs once per STAGE while the
            # workflow has one checkpoint_verification PHASE -- an unguarded
            # advance would march the machine off the end of its own step list
            # on a multi-stage run. Best-effort: a workflow-bookkeeping
            # failure must not fail the stage.
            try:
                workflow = self._coordinator.get_workflow(session_id)
                if workflow and workflow.get("current_step") == "constraint_evaluation":
                    self._coordinator.advance_step(session_id)
            except Exception as exc:
                logger.warning(
                    "workflow advance failed for session %s (%s)", session_id, exc
                )
            self._persist(session_id)
            return cp_result

    @traces(
        spec="035",
        story="US-008",
        reqs=[
            "REQ-035-US-008-001",
            "REQ-035-US-008-003",
            "REQ-035-US-008-006",
            "REQ-035-US-009-004",
        ],
    )
    def finalize(
        self, session_id: str, result: PipelineResult
    ) -> VerificationVerdict | None:
        """Run final verification for a completed pipeline run.

        ``engine.verify`` increments the session's attempt count on
        NOT_MET and sets ``ESCALATED`` once ``max_attempts`` is exhausted.
        The verdict's ``recommendation`` honestly reflects that state
        regardless of enforcement posture; only when ``config.enforce`` is
        set does this method actually invoke
        ``escalation.escalate_to_human`` (the runner then routes the
        ESCALATE recommendation through its own ``_save_escalation`` seam).
        """
        if not self._config.enabled or not session_id:
            return None
        session = self._sessions.get_session(session_id)
        if session is None:
            return None
        with self._tracer.span("PipelineVerificationGate.finalize"):
            try:
                context = self._context_builder.build(
                    _FINAL_STAGE_ALIAS, result.artifacts, result
                )
                results = self._engine.verify(session, context)
                feedback = self._engine.generate_feedback(results)
            except Exception as exc:
                logger.warning(
                    "finalize verification failed for session %s (%s)",
                    session_id,
                    exc,
                )
                return None

            failed_ids = [r.constraint_id for r in results if not r.is_satisfied]
            failure_reasons = {
                r.constraint_id: r.details for r in results if not r.is_satisfied
            }

            escalate = (
                session.status == SessionStatus.ESCALATED
                or self._escalation.should_escalate(session)
            )
            if escalate and self._config.enforce:
                try:
                    self._escalation.escalate_to_human(session)
                    self._coordinator.build_escalation_message(session)
                except Exception as exc:
                    logger.warning(
                        "escalation handling failed for session %s (%s)",
                        session_id,
                        exc,
                    )

            try:
                self._coordinator.build_verification_result(
                    results,
                    analysis={
                        "summary": feedback.summary,
                        "pass_rate": feedback.pass_rate,
                    },
                )
            except Exception as exc:
                logger.debug(
                    "CXP verification-result build failed (non-fatal): %s", exc
                )

            decision = (
                "SATISFIED" if session.status == SessionStatus.COMPLETED else "NOT_MET"
            )
            if escalate:
                recommendation = "ESCALATE"
            elif decision == "SATISFIED":
                recommendation = "COMPLETE"
            else:
                recommendation = "CONTINUE_ITERATING"

            verdict = VerificationVerdict(
                session_id=session_id,
                decision=decision,
                checkpoint_results=dict(self._checkpoint_history.get(session_id, {})),
                failed_constraints=failed_ids,
                failure_reasons=failure_reasons,
                attempt_count=session.attempt_count,
                recommendation=recommendation,
            )
            # The run is over, so the CXP-visible workflow is too: drive it
            # through its remaining phases (proof_generation, llm_analysis,
            # final_decision -- which this method has just performed via
            # engine.verify, generate_feedback and the verdict above) and drop
            # it. Until this landed, `status` only ever became "completed"
            # inside advance_step and nothing called it, so every workflow
            # stayed "started" forever, `is_complete` was False by
            # construction, and `_workflows` grew one entry per verification
            # session with no eviction. Best-effort: workflow bookkeeping must
            # not swallow a verdict the pipeline is waiting on.
            try:
                self._coordinator.complete_workflow(session_id)
                self._coordinator.forget_workflow(session_id)
            except Exception as exc:
                logger.warning(
                    "workflow completion failed for session %s (%s)",
                    session_id,
                    exc,
                )
            self._persist(session_id)
            return verdict

    @traces(
        spec="035",
        story="US-001",
        reqs=["REQ-035-US-001-010", "REQ-035-US-009-006"],
    )
    def _persist(self, session_id: str) -> None:
        """Best-effort persistence of session state to Redis."""
        if not self._redis.available:
            return
        session = self._sessions.get_session(session_id)
        if session is None:
            return
        try:
            self._redis.save_state(f"session:{session_id}", session.to_dict())
        except Exception as exc:
            logger.debug("verification session persist failed (non-fatal): %s", exc)