model.py

# Adapted from the open-source implementation by LalaSkye (Apache 2.0)
# Original repository: https://github.com/LalaSkye/no-direct-bind
# Modifications: Adapted for the CAGE 8-tier governance architecture,
# extended with Gap 1/2/3/4 sub-proofs and a concurrency-interleaving
# sub-proof, and integrated with CAGE state machine phases
# (PENDING → CHECKING → SEAL_ISSUED → EXECUTED/DENIED).
#
# Original copyright notice preserved per Apache 2.0 Section 4(b):
# Copyright (c) LalaSkye contributors
# Licensed under the Apache License, Version 2.0
#
# ---------------------------------------------------------------------------
# Model Scope & Distributed Extensions (peer review Fix C)
# ---------------------------------------------------------------------------
# This model covers SINGLE-REQUEST concurrency within the governance pipeline.
# It proves the No-Direct-Bind invariant holds for all interleavings of the
# concurrent CBF/OPA tier evaluations within one request.
#
# State counts (C1-sub audit remediation — added NARROW/PAUSE states):
#   - Gated sequential model: ~57 reachable states (was 21 before NARROW/PAUSE)
#   - Concurrent CBF/OPA model: ~66 reachable states (was 24 before NARROW/PAUSE)
#   - Ungated (direct-bind) model: 19 reachable states (unchanged)
# The increase is due to:
#   - soft_threshold_exceeded flag (enables NARROW terminal state)
#   - transient_block flag (enables PAUSE terminal state)
#   - Non-deterministic branching in tier PASS transitions
#
# For MULTI-AGENT cross-Redis contention (distributed locking, split-brain
# scenarios, cross-shard coordination), see: proof/distributed_cbf_model.py
#
# The actuator seal check (routing_seal.verify_seal()) is a verified
# precondition in the routing_seal module. See: src/gateway/governance/routing_seal.py
# The seal verification is not modeled here because it is a distinct trust
# boundary — the actuator independently verifies the seal was issued by the
# governance pipeline, providing defense-in-depth.
# ---------------------------------------------------------------------------

"""
CAGE No-Direct-Bind Proof — Exhaustive State-Space Enumerator
==============================================================

Theorem (No-Direct-Bind):
    In any run of the CAGE gated architecture, the system reaches an EXECUTED
    state only via a transition guarded by resolvedAllow = TRUE.
    Equivalently: there is no reachable state in which an effect has occurred
    while authority was unresolved.

    Formally (TLA+ safety invariant):
        NoDirectBind == (phase = "EXECUTED") => (resolvedAllow = TRUE)

This file:
  1. Defines the CAGE 8-tier governance state machine.
  2. Enumerates every reachable state via BFS.
  3. Asserts the invariant holds in ALL reachable states.
  4. Defines an ungated (direct-bind) variant and proves it VIOLATES the
     invariant, producing an explicit counterexample — confirming the gate is
     load-bearing, not decorative.
  5. Proves the invariant is insensitive to the CBF/OPA evaluation order,
     which the runtime executes concurrently via ``asyncio.gather()``.

Usage (no dependencies beyond the Python standard library):
    python proof/model.py

Expected output:
    [gated]   No-Direct-Bind holds over all N reachable states: True
    [ungated] direct-bind shortcut produces a violation: True

The model is intentionally small and enumerable.  It captures the essential
structure of the CAGE pipeline:
  - 8 governance tiers, each of which can PASS or FAIL
  - A routing seal that is issued only after all tiers pass
  - A downstream actuator that verifies the seal before executing

Scope of the model
------------------
The tuple models ``SymbolicGovernor._run_checks()`` — Tiers 1 through 7,
with Tier 3 split into its concurrent ``cbf`` and ``opa`` components.
Tier 0.5 (FTRA) is deliberately NOT in the tuple: it is a *pre-execution*
gate that runs at the LangGraph graph level, before ``_run_checks()`` is
invoked, and operates on a whole ``ExecutionPlan`` rather than a single tool
call.  Its verdict (CLEAR | HITL_REQUIRED | BLOCKED) is recorded separately
in the LangGraph state.  See ``src/gateway/governance/ftra/node_factory.py``
and ``src/governed_financial_advisor/graph/graph.py``.

Tier 7 (FRIA) caveat
--------------------
The ``fria`` tier is included in the tuple because ``enforce_fria_boundary()``
runs as a distinct step inside ``_run_checks()`` after the causal gatekeeper.
However, the *adaptive* FRIA enforcement path (lines 603-660 of
``symbolic_governor.py``) is gated on
``CAGE_NORMATIVE_PROVIDER != "static"``.  That environment variable is not
set in any committed deployment manifest, so the adaptive path is unreachable
in the reference deployment.  The unconditional portion (OTel attribute
stamping, lines 662-693) always runs.  The proof models the tier as present
and capable of blocking (FAIL) — which is the conservative, safe-side
assumption: if the tier can block, the invariant must hold even when it does.
What the proof does NOT claim is that the adaptive FRIA logic is exercised in
the reference deployment; that depends on ``CAGE_NORMATIVE_PROVIDER`` being
set to a non-static value.

Gaps closed by this proof:
  Gap 1: Proves the ungated (no-seal) architecture violates the invariant,
         establishing that the seal-issuance gate is load-bearing
  Gap 2: Proves govern() path satisfies the invariant when seal is issued
  Gap 3: Proves CBF_FAIL_OPEN shortcut removes the CBF tier from the gate
  Gap 4: Proves DoWhy-absent shortcut removes the causal tier from the gate
  Concurrency: Proves the invariant is order-independent across the
         concurrently-evaluated CBF and OPA tiers
"""

from __future__ import annotations

from collections.abc import Iterator
from dataclasses import dataclass

# ---------------------------------------------------------------------------
# State definition
# ---------------------------------------------------------------------------

# Governance tiers in execution order, mirroring the sequence of checks in
# ``SymbolicGovernor._run_checks()``.  Each tier can be PENDING, PASS or FAIL.
#
# Tier numbering follows the paper (§4.2).  Tier 3 is split into its two
# concurrently-evaluated components (``cbf`` and ``opa``) because each can
# independently block the action; see ``concurrent_tier_transitions()`` for
# the proof that their evaluation order does not affect the invariant.
# FTRA (Tier 0.5) added to close proof/implementation divergence (ARCH-1).
TIERS = (
    "ftra",  # Tier 0.5: FTRA action classification & reachability analysis
    "stpa",  # Tier 1:  STAMP/STPA unsafe control action check
    "confidence",  # Tier 2:  agent confidence threshold
    "cbf",  # Tier 3a: Control Barrier Function (Redis cash barrier)
    "opa",  # Tier 3b: OPA Rego policy evaluation
    "fiscal",  # Tier 4:  fiscal limit pre-reservation
    "consensus",  # Tier 5:  multi-agent consensus gate
    "causal",  # Tier 6:  DoWhy causal gatekeeper
    "fria",  # Tier 7:  FRIA normative boundary enforcement
)

# The subset of tiers that the runtime evaluates concurrently
# (``asyncio.gather()`` in symbolic_governor.py).  Used by
# ``concurrent_tier_transitions()`` to explore both interleavings.
CONCURRENT_TIERS = frozenset({"cbf", "opa"})

# Execution phases — mirrors the TLA+ state machine.
# Updated to include NARROW and PAUSE states per C1-sub audit remediation:
#   - NARROW: All tiers pass but action parameters exceed soft thresholds;
#             seal issued on clamped params, resolvedAllow=TRUE (ALLOW variant)
#   - PAUSE:  Transient condition (rate limiting, circuit breaker) detected;
#             no seal issued, retryable without action modification
PHASES = ("PENDING", "CHECKING", "SEAL_ISSUED", "EXECUTED", "DENIED", "NARROW", "PAUSE")


@dataclass(frozen=True)
class State:
    """A single point in the CAGE governance state space.

    Attributes:
        phase:          Current execution phase (PENDING, CHECKING, SEAL_ISSUED,
                        EXECUTED, DENIED, NARROW, PAUSE).
        tier_results:   Tuple of (tier_name, result) pairs where result is
                        "PASS", "FAIL", or "PENDING".
        seal_present:   True if a valid routing seal has been issued.
        resolved_allow: True if all tiers have passed AND a seal is present.
                        This is the ``resolvedAllow`` variable in the TLA+ spec.
                        NARROW states have resolved_allow=TRUE (they are ALLOW variants).
                        PAUSE states have resolved_allow=FALSE (retryable, not allowed).
        soft_threshold_exceeded: True if action parameters exceeded soft thresholds.
                        Only relevant for NARROW state transitions.
        transient_block: True if a transient condition (rate limit, circuit breaker)
                        caused the PAUSE. Only relevant for PAUSE state transitions.
        seal_consumed:  (Peer Review Fix - Gap 2 alignment) True if the seal has been
                        consumed by the actuator. Seals are single-use: once consumed,
                        the same seal cannot authorize another EXECUTED transition.
                        This models the routing_seal.consume_seal() atomic operation.
        seal_expired:   (Peer Review Fix - Gap 2 alignment) True if the seal TTL has
                        elapsed before consumption. Expired seals cannot authorize
                        EXECUTED transitions; the action must re-enter CHECKING.
    """

    phase: str
    tier_results: tuple[tuple[str, str], ...]
    seal_present: bool
    resolved_allow: bool
    soft_threshold_exceeded: bool = False
    transient_block: bool = False
    seal_consumed: bool = False
    seal_expired: bool = False

    def tier_result(self, tier: str) -> str:
        return dict(self.tier_results).get(tier, "PENDING")

    def all_tiers_passed(self) -> bool:
        return all(r == "PASS" for _, r in self.tier_results)

    def any_tier_failed(self) -> bool:
        return any(r == "FAIL" for _, r in self.tier_results)

    def is_allow_variant(self) -> bool:
        """True if this state represents an ALLOW decision (SEAL_ISSUED, EXECUTED, NARROW)."""
        return self.phase in ("SEAL_ISSUED", "EXECUTED", "NARROW")

    def is_terminal(self) -> bool:
        """True if this state has no successors (EXECUTED, DENIED, NARROW, PAUSE)."""
        return self.phase in ("EXECUTED", "DENIED", "NARROW", "PAUSE")


def initial_state() -> State:
    """The single initial state: all tiers pending, no seal, phase=PENDING."""
    return State(
        phase="PENDING",
        tier_results=tuple((t, "PENDING") for t in TIERS),
        seal_present=False,
        resolved_allow=False,
        soft_threshold_exceeded=False,
        transient_block=False,
        seal_consumed=False,
        seal_expired=False,
    )


# ---------------------------------------------------------------------------
# Gated transition function (the correct CAGE architecture)
# ---------------------------------------------------------------------------


def gated_transitions(state: State) -> Iterator[State]:
    """Generate all successor states from *state* under the gated architecture.

    Rules (mirror the CAGE symbolic_governor._run_checks() pipeline):
      - From PENDING: begin checking (move to CHECKING).
      - From CHECKING: each tier can PASS or FAIL in sequence.
        - If any tier FAILs → move to DENIED immediately (fail-closed).
        - If all tiers PASS:
          - If soft_threshold_exceeded → move to NARROW (seal on clamped params)
          - If transient_block → move to PAUSE (no seal, retryable)
          - Otherwise → issue seal → move to SEAL_ISSUED
      - From SEAL_ISSUED: actuator verifies seal.
        - Seal valid → move to EXECUTED (resolvedAllow=TRUE).
        - Seal invalid/expired → move to DENIED.
      - EXECUTED, DENIED, NARROW, and PAUSE are terminal.
        - NARROW has seal_present=TRUE, resolved_allow=TRUE (ALLOW variant)
        - PAUSE has seal_present=FALSE, resolved_allow=FALSE (retryable)

    The key invariant: EXECUTED is only reachable from SEAL_ISSUED, and
    SEAL_ISSUED is only reachable when all tiers have passed.
    NARROW also requires all tiers to pass (it's an ALLOW variant).
    PAUSE does NOT lead to EXECUTED without re-entering CHECKING.
    """
    if state.phase == "PENDING":
        # Begin the governance pipeline
        yield State(
            phase="CHECKING",
            tier_results=state.tier_results,
            seal_present=False,
            resolved_allow=False,
            soft_threshold_exceeded=False,
            transient_block=False,
        )

    elif state.phase == "CHECKING":
        # Find the first PENDING tier
        results = dict(state.tier_results)
        pending_tiers = [t for t in TIERS if results[t] == "PENDING"]

        if not pending_tiers:
            # All tiers resolved
            if state.any_tier_failed():
                yield State(
                    phase="DENIED",
                    tier_results=state.tier_results,
                    seal_present=False,
                    resolved_allow=False,
                    soft_threshold_exceeded=False,
                    transient_block=False,
                )
            elif state.transient_block:
                # Transient condition detected (rate limit, circuit breaker)
                # PAUSE: no seal issued, retryable without action modification
                yield State(
                    phase="PAUSE",
                    tier_results=state.tier_results,
                    seal_present=False,
                    resolved_allow=False,  # PAUSE is NOT an ALLOW
                    soft_threshold_exceeded=False,
                    transient_block=True,
                )
            elif state.soft_threshold_exceeded:
                # Soft threshold exceeded → NARROW with clamped params
                # NARROW is an ALLOW variant: seal on clamped params
                yield State(
                    phase="NARROW",
                    tier_results=state.tier_results,
                    seal_present=True,  # seal issued on clamped params
                    resolved_allow=True,  # NARROW is an ALLOW variant
                    soft_threshold_exceeded=True,
                    transient_block=False,
                )
            else:
                # All passed, no soft threshold or transient block
                # Normal path: issue routing seal
                yield State(
                    phase="SEAL_ISSUED",
                    tier_results=state.tier_results,
                    seal_present=True,
                    resolved_allow=True,  # resolvedAllow = TRUE only here
                    soft_threshold_exceeded=False,
                    transient_block=False,
                )
        else:
            # Advance the next pending tier: it can PASS or FAIL
            # Additionally, when a tier passes, we non-deterministically
            # model the possibility of soft_threshold_exceeded or transient_block
            next_tier = pending_tiers[0]
            for outcome in ("PASS", "FAIL"):
                new_results = dict(state.tier_results)
                new_results[next_tier] = outcome
                new_tier_results = tuple((t, new_results[t]) for t in TIERS)

                if outcome == "FAIL":
                    # Fail-closed: any failure → DENIED immediately
                    yield State(
                        phase="DENIED",
                        tier_results=new_tier_results,
                        seal_present=False,
                        resolved_allow=False,
                        soft_threshold_exceeded=False,
                        transient_block=False,
                    )
                else:
                    # PASS: continue checking
                    # Model without any threshold/transient condition
                    yield State(
                        phase="CHECKING",
                        tier_results=new_tier_results,
                        seal_present=False,
                        resolved_allow=False,
                        soft_threshold_exceeded=state.soft_threshold_exceeded,
                        transient_block=state.transient_block,
                    )
                    # Model with soft_threshold_exceeded condition
                    # (only if not already set and no transient_block)
                    if not state.soft_threshold_exceeded and not state.transient_block:
                        yield State(
                            phase="CHECKING",
                            tier_results=new_tier_results,
                            seal_present=False,
                            resolved_allow=False,
                            soft_threshold_exceeded=True,
                            transient_block=False,
                        )
                        # Model with transient_block condition
                        yield State(
                            phase="CHECKING",
                            tier_results=new_tier_results,
                            seal_present=False,
                            resolved_allow=False,
                            soft_threshold_exceeded=False,
                            transient_block=True,
                        )

    elif state.phase == "SEAL_ISSUED":
        # ======================================================================
        # Peer Review Fix: Gap 2 alignment — seal single-use consumption model
        # ======================================================================
        # The actuator verifies the seal before executing. Seals are SINGLE-USE:
        # once consumed (seal_consumed=True), the same seal cannot authorize
        # another EXECUTED transition. This models routing_seal.consume_seal()
        # which atomically validates + marks the seal as consumed in Redis.
        #
        # Transitions from SEAL_ISSUED:
        #   1. Seal valid AND not consumed AND not expired → consume → EXECUTED
        #   2. Seal already consumed (replay attack) → DENIED (no re-execution)
        #   3. Seal expired (TTL elapsed) → DENIED (must re-enter governance)
        #   4. Seal invalid (HMAC mismatch, tampered) → DENIED
        # ======================================================================

        if not state.seal_consumed and not state.seal_expired:
            # Path 1: Seal valid, not consumed, not expired → consume and execute
            yield State(
                phase="EXECUTED",
                tier_results=state.tier_results,
                seal_present=True,
                resolved_allow=True,
                soft_threshold_exceeded=False,
                transient_block=False,
                seal_consumed=True,  # Mark as consumed (single-use)
                seal_expired=False,
            )
            # Path 3: Seal expires before consumption (non-deterministic TTL race)
            yield State(
                phase="DENIED",
                tier_results=state.tier_results,
                seal_present=False,
                resolved_allow=False,
                soft_threshold_exceeded=False,
                transient_block=False,
                seal_consumed=False,
                seal_expired=True,  # TTL elapsed
            )

        if state.seal_consumed:
            # Path 2: Seal already consumed → replay attack blocked → DENIED
            # This models the atomic Redis check in consume_seal() that prevents
            # the same seal from being used twice.
            yield State(
                phase="DENIED",
                tier_results=state.tier_results,
                seal_present=False,
                resolved_allow=False,
                soft_threshold_exceeded=False,
                transient_block=False,
                seal_consumed=True,
                seal_expired=False,
            )

        # Path 4: Seal invalid (HMAC mismatch) — always possible
        yield State(
            phase="DENIED",
            tier_results=state.tier_results,
            seal_present=False,
            resolved_allow=False,
            soft_threshold_exceeded=False,
            transient_block=False,
            seal_consumed=False,
            seal_expired=False,
        )

    # EXECUTED, DENIED, NARROW, and PAUSE are terminal — no successors


# ---------------------------------------------------------------------------
# Ungated (direct-bind) transition function — the violation variant
# ---------------------------------------------------------------------------


def ungated_transitions(state: State) -> Iterator[State]:
    """Generate successor states under the UNGATED (direct-bind) architecture.

    The direct-bind shortcut: execution proceeds without waiting for a resolved
    seal.  This models three concrete CAGE gaps:
      - Gap 2: govern() path that raises GovernanceError but issues no seal,
               and a caller that catches the exception and executes anyway.
      - Gap 3: CBF_FAIL_OPEN=true — CBF tier is skipped entirely.
      - Gap 4: DoWhy ImportError silently removes the causal tier.

    In this variant, EXECUTED is reachable from CHECKING directly (bypassing
    SEAL_ISSUED), which violates the No-Direct-Bind invariant.
    """
    if state.phase == "PENDING":
        yield State(
            phase="CHECKING",
            tier_results=state.tier_results,
            seal_present=False,
            resolved_allow=False,
            soft_threshold_exceeded=False,
            transient_block=False,
        )

    elif state.phase == "CHECKING":
        results = dict(state.tier_results)
        pending_tiers = [t for t in TIERS if results[t] == "PENDING"]

        if not pending_tiers:
            if state.any_tier_failed():
                yield State(
                    phase="DENIED",
                    tier_results=state.tier_results,
                    seal_present=False,
                    resolved_allow=False,
                    soft_threshold_exceeded=False,
                    transient_block=False,
                )
            else:
                # Direct-bind shortcut: skip SEAL_ISSUED, go straight to EXECUTED
                # without issuing or verifying a seal.
                # resolvedAllow remains FALSE — this is the violation.
                yield State(
                    phase="EXECUTED",
                    tier_results=state.tier_results,
                    seal_present=False,
                    resolved_allow=False,  # ← authority never resolved
                    soft_threshold_exceeded=False,
                    transient_block=False,
                )
        else:
            next_tier = pending_tiers[0]
            for outcome in ("PASS", "FAIL"):
                new_results = dict(state.tier_results)
                new_results[next_tier] = outcome
                new_tier_results = tuple((t, new_results[t]) for t in TIERS)

                if outcome == "FAIL":
                    yield State(
                        phase="DENIED",
                        tier_results=new_tier_results,
                        seal_present=False,
                        resolved_allow=False,
                        soft_threshold_exceeded=False,
                        transient_block=False,
                    )
                else:
                    yield State(
                        phase="CHECKING",
                        tier_results=new_tier_results,
                        seal_present=False,
                        resolved_allow=False,
                        soft_threshold_exceeded=False,
                        transient_block=False,
                    )

    elif state.phase == "SEAL_ISSUED":
        # Ungated variant never reaches SEAL_ISSUED, but handle for completeness
        yield State(
            phase="EXECUTED",
            tier_results=state.tier_results,
            seal_present=True,
            resolved_allow=True,
            soft_threshold_exceeded=False,
            transient_block=False,
        )


# ---------------------------------------------------------------------------
# Ungated NARROW transition function — negative control for NARROW state
# ---------------------------------------------------------------------------


def ungated_narrow_transitions(state: State) -> Iterator[State]:
    """Generate successor states where NARROW can bypass seal verification.

    This is a negative control variant that models a hypothetical bug where
    NARROW states allow execution without proper seal verification.

    The violation: NARROW → EXECUTED without seal verification, which should
    produce a counterexample demonstrating the seal gate is load-bearing
    for NARROW states as well.
    """
    if state.phase == "PENDING":
        yield State(
            phase="CHECKING",
            tier_results=state.tier_results,
            seal_present=False,
            resolved_allow=False,
            soft_threshold_exceeded=False,
            transient_block=False,
        )

    elif state.phase == "CHECKING":
        results = dict(state.tier_results)
        pending_tiers = [t for t in TIERS if results[t] == "PENDING"]

        if not pending_tiers:
            if state.any_tier_failed():
                yield State(
                    phase="DENIED",
                    tier_results=state.tier_results,
                    seal_present=False,
                    resolved_allow=False,
                    soft_threshold_exceeded=False,
                    transient_block=False,
                )
            elif state.soft_threshold_exceeded:
                # Bug: NARROW without seal, then allow EXECUTED transition
                yield State(
                    phase="NARROW",
                    tier_results=state.tier_results,
                    seal_present=False,  # ← No seal issued!
                    resolved_allow=False,  # ← Authority not resolved!
                    soft_threshold_exceeded=True,
                    transient_block=False,
                )
            else:
                # Normal SEAL_ISSUED path (for comparison)
                yield State(
                    phase="SEAL_ISSUED",
                    tier_results=state.tier_results,
                    seal_present=True,
                    resolved_allow=True,
                    soft_threshold_exceeded=False,
                    transient_block=False,
                )
        else:
            next_tier = pending_tiers[0]
            for outcome in ("PASS", "FAIL"):
                new_results = dict(state.tier_results)
                new_results[next_tier] = outcome
                new_tier_results = tuple((t, new_results[t]) for t in TIERS)

                if outcome == "FAIL":
                    yield State(
                        phase="DENIED",
                        tier_results=new_tier_results,
                        seal_present=False,
                        resolved_allow=False,
                        soft_threshold_exceeded=False,
                        transient_block=False,
                    )
                else:
                    # Continue checking, with possibility of soft_threshold
                    yield State(
                        phase="CHECKING",
                        tier_results=new_tier_results,
                        seal_present=False,
                        resolved_allow=False,
                        soft_threshold_exceeded=False,
                        transient_block=False,
                    )
                    # Also model soft_threshold path
                    yield State(
                        phase="CHECKING",
                        tier_results=new_tier_results,
                        seal_present=False,
                        resolved_allow=False,
                        soft_threshold_exceeded=True,
                        transient_block=False,
                    )

    elif state.phase == "NARROW":
        # Bug: NARROW can transition to EXECUTED without seal verification
        # This should produce a counterexample
        yield State(
            phase="EXECUTED",
            tier_results=state.tier_results,
            seal_present=False,  # ← No seal!
            resolved_allow=False,  # ← Authority not resolved! VIOLATION
            soft_threshold_exceeded=True,
            transient_block=False,
        )

    elif state.phase == "SEAL_ISSUED":
        yield State(
            phase="EXECUTED",
            tier_results=state.tier_results,
            seal_present=True,
            resolved_allow=True,
            soft_threshold_exceeded=False,
            transient_block=False,
        )


# ---------------------------------------------------------------------------
# Concurrent transition function — models the CBF/OPA parallel gate
# ---------------------------------------------------------------------------


def concurrent_tier_transitions(state: State) -> Iterator[State]:
    """Gated transitions in which the concurrent tiers may resolve in any order.

    ``gated_transitions()`` advances tiers strictly in ``TIERS`` order, which
    is an *under-approximation* of the runtime: ``SymbolicGovernor._run_checks()``
    dispatches the CBF and OPA checks together via ``asyncio.gather()``, so
    either may resolve first.

    This variant relaxes the ordering constraint for every tier in
    ``CONCURRENT_TIERS``: whenever the next pending tier belongs to that set,
    *any* pending concurrent tier may advance.  Every interleaving of the
    concurrent tiers — including the partial-failure states in which one has
    resolved and the other has not — therefore appears in the reachable set.

    Because this strictly adds transitions (and thus states) relative to
    ``gated_transitions()``, proving the invariant here is a strictly stronger
    result: the seal gate holds under every interleaving, not merely under the
    canonical order.

    Updated for NARROW/PAUSE states: the concurrent transitions preserve the
    soft_threshold_exceeded and transient_block flags, allowing NARROW and
    PAUSE terminal states to be reached via any CBF/OPA interleaving.
    """
    if state.phase != "CHECKING":
        yield from gated_transitions(state)
        return

    results = dict(state.tier_results)
    pending_tiers = [t for t in TIERS if results[t] == "PENDING"]

    if not pending_tiers:
        # Terminal resolution is order-independent — delegate.
        yield from gated_transitions(state)
        return

    next_tier = pending_tiers[0]
    if next_tier not in CONCURRENT_TIERS:
        yield from gated_transitions(state)
        return

    # The pipeline has reached the concurrent gate: any pending concurrent
    # tier may resolve next, in either order.
    advanceable = [t for t in pending_tiers if t in CONCURRENT_TIERS]
    for tier in advanceable:
        for outcome in ("PASS", "FAIL"):
            new_results = dict(state.tier_results)
            new_results[tier] = outcome
            new_tier_results = tuple((t, new_results[t]) for t in TIERS)

            if outcome == "FAIL":
                # Fail-closed: either concurrent check failing denies the action.
                yield State(
                    phase="DENIED",
                    tier_results=new_tier_results,
                    seal_present=False,
                    resolved_allow=False,
                    soft_threshold_exceeded=False,
                    transient_block=False,
                )
            else:
                # PASS: continue checking, preserving threshold/transient flags
                yield State(
                    phase="CHECKING",
                    tier_results=new_tier_results,
                    seal_present=False,
                    resolved_allow=False,
                    soft_threshold_exceeded=state.soft_threshold_exceeded,
                    transient_block=state.transient_block,
                )
                # Also model with soft_threshold_exceeded condition
                if not state.soft_threshold_exceeded and not state.transient_block:
                    yield State(
                        phase="CHECKING",
                        tier_results=new_tier_results,
                        seal_present=False,
                        resolved_allow=False,
                        soft_threshold_exceeded=True,
                        transient_block=False,
                    )
                    # Model with transient_block condition
                    yield State(
                        phase="CHECKING",
                        tier_results=new_tier_results,
                        seal_present=False,
                        resolved_allow=False,
                        soft_threshold_exceeded=False,
                        transient_block=True,
                    )


# ---------------------------------------------------------------------------
# BFS reachable-state enumerator
# ---------------------------------------------------------------------------


def enumerate_reachable(
    transition_fn,
    start: State | None = None,
) -> set[State]:
    """BFS over the state space.  Returns the set of all reachable states."""
    if start is None:
        start = initial_state()

    visited: set[State] = set()
    frontier: list[State] = [start]

    while frontier:
        state = frontier.pop()
        if state in visited:
            continue
        visited.add(state)
        for successor in transition_fn(state):
            if successor not in visited:
                frontier.append(successor)

    return visited


# ---------------------------------------------------------------------------
# Invariant checker
# ---------------------------------------------------------------------------


def check_no_direct_bind(states: set[State]) -> tuple[bool, State | None]:
    """Check the No-Direct-Bind invariant over all states.

    Invariant: (phase = "EXECUTED") => (resolvedAllow = TRUE)

    Returns:
        (holds, counterexample) — holds=True if the invariant holds everywhere,
        counterexample is the first violating state (or None if holds=True).
    """
    for state in states:
        if state.phase == "EXECUTED" and not state.resolved_allow:
            return False, state
    return True, None


# ---------------------------------------------------------------------------
# Main — run both proofs
# ---------------------------------------------------------------------------


def main() -> None:
    print("=" * 70)
    print("CAGE No-Direct-Bind Exhaustive State-Space Proof")
    print("=" * 70)
    print()

    # ── Gated architecture (correct CAGE) ────────────────────────────────────
    gated_states = enumerate_reachable(gated_transitions)
    gated_holds, gated_cex = check_no_direct_bind(gated_states)

    print(f"[gated]   Reachable states: {len(gated_states)}")
    print(
        f"[gated]   No-Direct-Bind holds over all {len(gated_states)} reachable states: {gated_holds}"
    )

    if not gated_holds:
        print(f"[gated]   ❌ COUNTEREXAMPLE FOUND: {gated_cex}")
    else:
        # Show the single EXECUTED state to confirm it has resolvedAllow=TRUE
        executed = [s for s in gated_states if s.phase == "EXECUTED"]
        print(f"[gated]   EXECUTED states: {len(executed)}")
        for s in executed:
            print(
                f"[gated]     → resolvedAllow={s.resolved_allow}  seal_present={s.seal_present}"
            )

    print()

    # ── Ungated architecture (direct-bind shortcut) ───────────────────────────
    ungated_states = enumerate_reachable(ungated_transitions)
    ungated_holds, ungated_cex = check_no_direct_bind(ungated_states)

    print(f"[ungated] Reachable states: {len(ungated_states)}")
    print(f"[ungated] No-Direct-Bind holds: {ungated_holds}")

    if ungated_cex is not None:
        print("[ungated] direct-bind shortcut produces a violation: True")
        print("[ungated] Counterexample state:")
        print(f"[ungated]   phase         = {ungated_cex.phase}")
        print(f"[ungated]   resolvedAllow = {ungated_cex.resolved_allow}")
        print(f"[ungated]   seal_present  = {ungated_cex.seal_present}")
        tier_summary = dict(ungated_cex.tier_results)
        print(f"[ungated]   tier_results  = {tier_summary}")
    else:
        print("[ungated] direct-bind shortcut produces a violation: False")
        print(
            "[ungated] ⚠️  WARNING: ungated variant did not produce a violation — "
            "check transition function."
        )

    print()

    # ── Concurrency sub-proof: CBF/OPA interleaving ───────────────────────────
    # gated_transitions() resolves tiers in a fixed order, which under-
    # approximates the runtime (asyncio.gather dispatches CBF and OPA
    # together).  concurrent_tier_transitions() explores both interleavings,
    # producing a strict superset of the gated reachable states.
    concurrent_states = enumerate_reachable(concurrent_tier_transitions)
    concurrent_holds, concurrent_cex = check_no_direct_bind(concurrent_states)
    concurrent_executed = [s for s in concurrent_states if s.phase == "EXECUTED"]

    print("Concurrency sub-proof (CBF ∥ OPA interleaving):")
    print(
        f"  Reachable states: {len(concurrent_states)} "
        f"(gated: {len(gated_states)}) — superset={gated_states <= concurrent_states}"
    )
    print(f"  No-Direct-Bind holds under every interleaving: {concurrent_holds}")
    print(
        f"  EXECUTED states: {len(concurrent_executed)} "
        f"(all with resolvedAllow=TRUE: "
        f"{all(s.resolved_allow for s in concurrent_executed)})"
    )
    if not concurrent_holds:
        print(f"  ❌ COUNTEREXAMPLE FOUND: {concurrent_cex}")
    print()

    # ── Gap-specific sub-proofs ───────────────────────────────────────────────
    print("Gap-specific sub-proofs:")
    print()

    # Gap 1: ungated architecture — the seal gate removed structurally.
    # Re-uses the ungated enumeration above; reported here so the four
    # gap-specific sub-proofs are enumerated contiguously (Gaps 1-4).
    print(
        f"  Gap 1 (no routing seal on approval): reachable states="
        f"{len(ungated_states)}, invariant holds={ungated_holds}"
    )
    print(
        "  → Violation confirmed: EXECUTED with resolvedAllow=False, seal_present=False"
    )
    print("  → Removing the seal-issuance step structurally (not merely skipping one")
    print("    tier's evaluation) yields a genuine, minimal counterexample — the")
    print("    seal gate in gated_transitions() is load-bearing.")
    print()

    # Gap 3: CBF_FAIL_OPEN — model as CBF tier always PASS regardless of check
    def cbf_fail_open_transitions(state: State) -> Iterator[State]:
        """CBF tier is skipped (always PASS) — models CBF_FAIL_OPEN=true."""
        if state.phase == "CHECKING":
            results = dict(state.tier_results)
            pending_tiers = [t for t in TIERS if results[t] == "PENDING"]
            if pending_tiers and pending_tiers[0] == "cbf":
                # Skip CBF: mark as PASS without any check
                new_results = dict(state.tier_results)
                new_results["cbf"] = "PASS"
                new_tier_results = tuple((t, new_results[t]) for t in TIERS)
                yield State(
                    phase="CHECKING",
                    tier_results=new_tier_results,
                    seal_present=False,
                    resolved_allow=False,
                    soft_threshold_exceeded=state.soft_threshold_exceeded,
                    transient_block=state.transient_block,
                )
                return
        yield from gated_transitions(state)

    cbf_states = enumerate_reachable(cbf_fail_open_transitions)
    # CBF_FAIL_OPEN alone doesn't violate the invariant structurally (seal is
    # still issued after all remaining tiers pass), but it removes a mandatory
    # tier from the gate.  The proof here confirms the seal path is preserved
    # even with CBF skipped — the Gap 3 fix (RuntimeError at startup) prevents
    # this configuration from being reachable in production at all.
    cbf_holds, _cbf_cex = check_no_direct_bind(cbf_states)
    print(
        f"  Gap 3 (CBF_FAIL_OPEN): reachable states={len(cbf_states)}, "
        f"invariant holds={cbf_holds}"
    )
    print("  → Structural invariant preserved, but CBF tier is absent from gate.")
    print("  → Production startup RuntimeError prevents this configuration.")
    print()

    # Gap 4: DoWhy absent — causal tier always PASS (silently skipped)
    def dowhy_absent_transitions(state: State) -> Iterator[State]:
        """Causal tier is silently skipped — models DoWhy ImportError."""
        if state.phase == "CHECKING":
            results = dict(state.tier_results)
            pending_tiers = [t for t in TIERS if results[t] == "PENDING"]
            if pending_tiers and pending_tiers[0] == "causal":
                # Skip causal: mark as PASS without any check
                new_results = dict(state.tier_results)
                new_results["causal"] = "PASS"
                new_tier_results = tuple((t, new_results[t]) for t in TIERS)
                yield State(
                    phase="CHECKING",
                    tier_results=new_tier_results,
                    seal_present=False,
                    resolved_allow=False,
                    soft_threshold_exceeded=state.soft_threshold_exceeded,
                    transient_block=state.transient_block,
                )
                return
        yield from gated_transitions(state)

    dowhy_states = enumerate_reachable(dowhy_absent_transitions)
    dowhy_holds, _dowhy_cex = check_no_direct_bind(dowhy_states)
    print(
        f"  Gap 4 (DoWhy absent): reachable states={len(dowhy_states)}, "
        f"invariant holds={dowhy_holds}"
    )
    print("  → Structural invariant preserved, but causal tier is absent from gate.")
    print("  → Production startup RuntimeError prevents this configuration.")
    print()

    # Gap 2: govern() without seal — models the old path where govern() returned
    # None and callers could proceed without seal verification.
    #
    # Peer Review Fix: Gap 2 alignment — clarify actuator gate semantics.
    # The actuator gate is the LAST defense: it verifies the seal before executing.
    # Without seal issuance, the actuator has no artifact to verify, and the action
    # could proceed without governance approval. This function models that violation.
    def no_seal_govern_transitions(state: State) -> Iterator[State]:
        """govern() issues no seal — models the pre-fix direct-bind path.

        Actuator Gate Semantics (Gap 2 alignment):
        -------------------------------------------
        The actuator is the component that EXECUTES the governed action (e.g., the
        trade execution endpoint). Its sole responsibility in the seal-based
        architecture is:
            1. Receive the seal from the caller
            2. Call routing_seal.consume_seal() to atomically validate + consume
            3. Only proceed with execution if consume_seal() returns True

        Without seal issuance (this function), the actuator receives no seal, and
        if it proceeds anyway, it creates a direct-bind violation: EXECUTED state
        is reached without resolvedAllow=TRUE.

        This models the pre-fix architecture where govern() returned None on
        approval (no seal), and the actuator had no artifact to verify.
        """
        if state.phase == "CHECKING":
            results = dict(state.tier_results)
            pending_tiers = [t for t in TIERS if results[t] == "PENDING"]
            if not pending_tiers and not state.any_tier_failed():
                # All tiers passed but no seal issued — direct bind to EXECUTED
                # This is a VIOLATION: actuator cannot verify governance approval
                yield State(
                    phase="EXECUTED",
                    tier_results=state.tier_results,
                    seal_present=False,
                    resolved_allow=False,  # ← VIOLATION: No seal → no resolution
                    soft_threshold_exceeded=False,
                    transient_block=False,
                    seal_consumed=False,
                    seal_expired=False,
                )
                return
        yield from ungated_transitions(state)

    no_seal_states = enumerate_reachable(no_seal_govern_transitions)
    no_seal_holds, no_seal_cex = check_no_direct_bind(no_seal_states)
    print(
        f"  Gap 2 (govern() no seal): reachable states={len(no_seal_states)}, "
        f"invariant holds={no_seal_holds}"
    )
    if no_seal_cex:
        print("  → Violation confirmed: EXECUTED with resolvedAllow=False")
        print("  → Fix: govern() now issues seal; callers verify before executing.")
    print()

    # ── NARROW/PAUSE state-space sub-proofs ───────────────────────────────────
    print("NARROW/PAUSE state-space sub-proofs (C1-sub audit remediation):")
    print()

    # Count NARROW and PAUSE states in the gated model
    narrow_states = [s for s in gated_states if s.phase == "NARROW"]
    pause_states = [s for s in gated_states if s.phase == "PAUSE"]
    print(f"  NARROW states: {len(narrow_states)}")
    for s in narrow_states:
        print(
            f"    → resolvedAllow={s.resolved_allow}  "
            f"seal_present={s.seal_present}  "
            f"soft_threshold_exceeded={s.soft_threshold_exceeded}"
        )
    print(f"  PAUSE states: {len(pause_states)}")
    for s in pause_states:
        print(
            f"    → resolvedAllow={s.resolved_allow}  "
            f"seal_present={s.seal_present}  "
            f"transient_block={s.transient_block}"
        )
    print()

    # Verify NARROW states satisfy NoDirectBind (they are ALLOW variants)
    narrow_valid = all(s.resolved_allow and s.seal_present for s in narrow_states)
    print(
        f"  NARROW states have resolvedAllow=TRUE and seal_present=TRUE: {narrow_valid}"
    )

    # Verify PAUSE states do NOT have seal_present=TRUE (they are retryable, not ALLOW)
    pause_no_seal = all(
        not s.seal_present and not s.resolved_allow for s in pause_states
    )
    print(
        f"  PAUSE states have seal_present=FALSE and resolvedAllow=FALSE: {pause_no_seal}"
    )
    print()

    # ── Ungated NARROW negative control ───────────────────────────────────────
    print("Ungated NARROW negative control (C1-sub):")
    ungated_narrow_states = enumerate_reachable(ungated_narrow_transitions)
    ungated_narrow_holds, ungated_narrow_cex = check_no_direct_bind(
        ungated_narrow_states
    )
    print(f"  Reachable states: {len(ungated_narrow_states)}")
    print(f"  No-Direct-Bind holds: {ungated_narrow_holds}")
    if ungated_narrow_cex is not None:
        print("  → Violation confirmed: EXECUTED via NARROW without seal verification")
        print(f"     phase={ungated_narrow_cex.phase}")
        print(f"     resolvedAllow={ungated_narrow_cex.resolved_allow}")
        print(f"     seal_present={ungated_narrow_cex.seal_present}")
        print(
            f"     soft_threshold_exceeded={ungated_narrow_cex.soft_threshold_exceeded}"
        )
    print()

    # ── Final assertions ──────────────────────────────────────────────────────
    print("=" * 70)
    assert gated_holds, "PROOF FAILED: gated architecture violates No-Direct-Bind!"
    assert not ungated_holds, (
        "PROOF FAILED: ungated variant should violate No-Direct-Bind!"
    )
    assert not no_seal_holds, (
        "PROOF FAILED: no-seal govern() should violate No-Direct-Bind!"
    )
    assert concurrent_holds, (
        "PROOF FAILED: No-Direct-Bind does not hold under CBF/OPA interleaving!"
    )
    assert gated_states <= concurrent_states, (
        "PROOF FAILED: the concurrent model must over-approximate the sequential one!"
    )
    # NARROW/PAUSE assertions
    assert narrow_valid, (
        "PROOF FAILED: NARROW states must have resolvedAllow=TRUE and seal_present=TRUE!"
    )
    assert pause_no_seal, (
        "PROOF FAILED: PAUSE states must have seal_present=FALSE and resolvedAllow=FALSE!"
    )
    assert not ungated_narrow_holds, (
        "PROOF FAILED: ungated NARROW variant should violate No-Direct-Bind!"
    )

    print("✅ All assertions passed.")
    print()
    print("PROVED:")
    print("  1. The gated CAGE architecture satisfies No-Direct-Bind over the")
    print(f"     entire reachable state space ({len(gated_states)} states).")
    print("  2. The ungated (direct-bind) variant provably violates the invariant.")
    print("  3. The pre-fix govern() path (no seal) provably violates the invariant.")
    print("  4. The invariant is order-independent across the concurrently-")
    print(f"     evaluated CBF and OPA tiers ({len(concurrent_states)} states,")
    print("     a strict superset of the sequential model).")
    print(f"  5. NARROW states ({len(narrow_states)}) are ALLOW variants with")
    print("     resolvedAllow=TRUE and seal_present=TRUE (seal on clamped params).")
    print(f"  6. PAUSE states ({len(pause_states)}) are retryable with")
    print(
        "     resolvedAllow=FALSE and seal_present=FALSE (no execution without re-check)."
    )
    print("  7. The ungated NARROW variant produces a counterexample, confirming")
    print("     the seal gate is load-bearing for NARROW decisions as well.")
    print()
    print("PLAUSIBLE (not proved here):")
    print("  That this model generalises to the full production CAGE stack.")
    print()
    print("NOT CLAIMED:")
    print("  That this is a security product or hardens any specific deployment.")


if __name__ == "__main__":
    main()