diff --git a/agent-team/agent_team/graph.py b/agent-team/agent_team/graph.py index efb8d19..d176f8e 100644 --- a/agent-team/agent_team/graph.py +++ b/agent-team/agent_team/graph.py @@ -89,10 +89,16 @@ __all__ = [ "CLARIFY", "DEFAULT_CLARIFY_DEADLINE", "DISPATCH_NODE", + "GATE_APPROVE_ROUTE", + "GATE_REVISE_ROUTE", + "GATE_TERMINAL_ROUTE", "INTAKE", + "MAX_PLAN_GATE_VISITS", "P1_PHASE_SEQUENCE", "PARKED_ROUTE", "PLAN", + "PLAN_DECISION_KIND", + "PLAN_GATE", "REVIEW", "VERIFY_NODE", "build_checkpoint_serde", @@ -102,9 +108,11 @@ __all__ = [ "get_pipeline_state", "intake_node", "pending_question", + "plan_gate_node", "plan_node", "plan_phase", "resume_task", + "route_after_plan_gate", "start_task", "thread_config", ] @@ -125,6 +133,44 @@ REVIEW = "review" BUILD_ROUTE = "build" PARKED_ROUTE = "parked" +# Plan-review human-decision gate (Phase B2a). PLAN_GATE is the graph vertex the +# review-cap dead-end (the route the review loop returns as PARKED_ROUTE/ESCALATE) +# is repointed at when the gate is wired: instead of terminally parking, the task +# suspends on a resumable ``interrupt()`` so the owner can decide. The gate +# mirrors the clarifier's interrupt/resume contract exactly (same payload shape + +# the same stable question_id/turn helpers), so the existing pending_question(...) +# extractor and the turn-guarded ResumeWorker drive it uniformly. The route ids +# the gate's conditional-edge function returns are kept distinct from the vertex +# id so a route key never collides with a node name. +PLAN_GATE = "plan_gate" +GATE_APPROVE_ROUTE = "gate_approve" +GATE_REVISE_ROUTE = "gate_revise" +GATE_TERMINAL_ROUTE = "gate_terminal" + +# The interrupt-payload discriminator that tells the responder/ledger this is a +# plan-decision gate (vs the clarifier's question-set). The clarifier payload has +# no ``kind`` (legacy = "clarify"); the gate stamps this so a mixed-state ledger +# can tell the two apart. +PLAN_DECISION_KIND = "plan_decision" + +# Combined ceiling on plan-gate visits (Phase B2a, bounded termination). A human +# ``request_changes`` re-enters plan<->review, which can hit the review cap and +# gate AGAIN. Each gate visit increments ``plan_gate_visits``; once it reaches +# this constant the gate stops offering request_changes and the task goes +# terminal PARKED with a "revision ceiling reached" marker. This is what proves +# the human-in-the-loop revision cycle terminates: the only non-terminal gate +# decision (request_changes) strictly consumes one of a finite number of visits. +MAX_PLAN_GATE_VISITS = 3 + +# Disjoint namespace base for the gate's stable ``turn`` derivation. The +# clarifier numbers its turns 0..N from ``qa_history`` length; the gate numbers +# its turns from a high base offset by ``plan_gate_visits`` so a gate turn can +# never collide with a clarifier turn for the same thread (the turn guard in +# ResumeWorker matches a resume to the open interrupt by ``turn``). Only one +# interrupt is ever open per thread, but keeping the spaces disjoint makes the +# stable-question_id derivation unambiguous across the task's whole life. +_PLAN_GATE_TURN_BASE = 1_000_000 + # P3 (build -> verify subgraph) vertex ids. These are the GRAPH VERTEX names the # opt-in P3 subgraph hangs off the review loop's "build" route; they are kept # distinct from the route-id constants above (BUILD_ROUTE / PARKED_ROUTE) and @@ -287,6 +333,201 @@ def plan_node(state: PipelineState) -> PipelineState: ) +def _plan_gate_turn(visits: int) -> int: + """Return the stable gate ``turn`` for the ``visits``-th gate visit (B2a). + + Offset into a high, disjoint namespace (:data:`_PLAN_GATE_TURN_BASE`) so a + gate turn can never collide with a clarifier turn (``qa_history`` length) for + the same thread. Monotonic in ``visits`` so each successive gate suspend has + its own stable ``(thread_id, turn)`` identity (and thus its own question_id). + """ + return _PLAN_GATE_TURN_BASE + visits + + +def plan_gate_node(state: PipelineState) -> PipelineState: + """PLAN-GATE stage: the resumable human decision at the review-cap dead-end. + + Replaces the terminal PARKED escalation. When the plan<->review loop cannot + converge (the review loop returned the cap/ESCALATE route) — or a salvaged + partial plan is presented — this node suspends with ``interrupt()`` exactly + like the clarifier so the owner can decide. The interrupt payload mirrors the + clarifier's §3.3.1 shape (``thread_id``, ``question_id``, ``turn``, + ``transport``, ``deadline``, ``slack_thread_ts``) so the existing + :func:`pending_question` extractor and the turn-guarded + :class:`~agent_team.resume_worker.ResumeWorker` drive it with no special + casing, PLUS: + + * ``kind`` = :data:`PLAN_DECISION_KIND` — the discriminator that tells the + responder/ledger this is a plan-decision (not a clarifier question-set); + * ``plan`` / ``findings`` — the review context the owner decides over (the + current plan and the latest review verdict's findings). + + **Bounded termination.** Before suspending, the node bumps + ``plan_gate_visits``. If the ceiling (:data:`MAX_PLAN_GATE_VISITS`) is already + exhausted it does **not** interrupt: it returns terminal PARKED with a + "revision ceiling reached" marker. So every gate suspend strictly consumes + one of a finite number of visits and the human loop always terminates. + + On resume, ``interrupt()`` returns whatever was passed to + ``Command(resume=...)`` — the decision contract + ``{"decision": "approve"|"request_changes"|"abandon", "notes": str}``. The + node consumes it and writes the routing state (see + :func:`route_after_plan_gate`): + + * ``approve`` -> the same terminal "approved plan" state an auto-approved + plan reaches (phase ``BUILD``, status ``ACTIVE``); + * ``request_changes`` -> phase ``PLAN``, status ``ACTIVE`` (loop back), with + the human ``notes`` folded into ``review_verdicts`` as a synthetic + REQUEST_CHANGES verdict so the planner's ``_format_review_feedback`` reads + it on the re-plan; + * ``abandon`` (or any unrecognized decision) -> terminal ``FAILED``. + """ + prior_visits = int(state.get("plan_gate_visits", 0) or 0) + + # Ceiling guard FIRST: an exhausted budget means the gate must not offer + # another request_changes loop. Park terminally rather than suspend. + if prior_visits >= MAX_PLAN_GATE_VISITS: + return PipelineState( + status=TaskStatus.PARKED.value, + current_phase=_phase_value(Phase.PARKED), + failure_reason=( + "plan-gate revision ceiling reached " + f"({prior_visits}/{MAX_PLAN_GATE_VISITS} gate visits)" + ), + updated_at=_utc_now_iso(), + ) + + visits = prior_visits + 1 + thread_id = state.get("thread_id", "") + transport = state.get("transport", "") + slack_thread_ts = state.get("slack_thread_ts", "") + turn = _plan_gate_turn(visits) + question_id = _question_id_for(thread_id, turn) + deadline = (datetime.now(timezone.utc) + DEFAULT_CLARIFY_DEADLINE).isoformat() + + decision = interrupt( + { + "thread_id": thread_id, + "question_id": question_id, + "turn": turn, + "kind": PLAN_DECISION_KIND, + "transport": transport, + "deadline": deadline, + "slack_thread_ts": slack_thread_ts, + "plan": state.get("plan"), + "findings": _latest_review_findings(state), + } + ) + + return _apply_plan_decision(state, decision, visits=visits) + + +def _latest_review_findings(state: PipelineState) -> str: + """Return the latest review verdict's findings text (gate presentation).""" + verdicts = state.get("review_verdicts") or [] + if not verdicts: + return "" + last = verdicts[-1] + if isinstance(last, dict): + return str(last.get("findings") or last.get("notes") or "") + return str(last) + + +def _apply_plan_decision( + state: PipelineState, decision: Any, *, visits: int +) -> PipelineState: + """Consume the owner's resume decision and return the routing state (B2a). + + ``decision`` is the value passed to ``Command(resume=...)`` — the contract + ``{"decision": ..., "notes": ...}``. A bare string is tolerated as the + decision verb. The default for an unrecognized/empty decision is the SAFE + direction: terminal ``FAILED`` (never an accidental approve). + """ + verb, notes = _parse_decision(decision) + now = _utc_now_iso() + + if verb == "approve": + # Settle exactly as an auto-APPROVED plan does today (review_node's + # APPROVE branch: phase BUILD, status ACTIVE) so the terminus matches. + return PipelineState( + status=TaskStatus.ACTIVE.value, + current_phase=_phase_value(Phase.BUILD), + plan_gate_visits=visits, + updated_at=now, + ) + + if verb == "request_changes": + # Loop back to the planner. Fold the human notes into review_verdicts as + # a synthetic REQUEST_CHANGES verdict whose ``findings`` carry the notes, + # so the planner's _format_review_feedback surfaces them on the re-plan. + verdicts = list(state.get("review_verdicts") or []) + verdicts.append( + { + "verdict": "request_changes", + "outcome": "loop_back", + "findings": notes, + "reviewer": "human_plan_gate", + "created_at": now, + } + ) + return PipelineState( + status=TaskStatus.ACTIVE.value, + current_phase=_phase_value(Phase.PLAN), + review_verdicts=verdicts, + plan_gate_visits=visits, + updated_at=now, + ) + + # abandon (or any unrecognized decision) -> terminal FAILED. + return PipelineState( + status=TaskStatus.FAILED.value, + current_phase=_phase_value(Phase.PARKED), + plan_gate_visits=visits, + failure_reason=f"plan abandoned at human gate: {notes}".rstrip(": "), + updated_at=now, + ) + + +def _parse_decision(decision: Any) -> tuple[str, str]: + """Normalize a resume decision into ``(verb, notes)`` (B2a). + + Accepts the ``{"decision": ..., "notes": ...}`` mapping or a bare string. The + verb is lowercased/stripped and mapped to one of ``approve`` / + ``request_changes`` / ``abandon``; anything unrecognized maps to ``abandon`` + (the SAFE terminal direction — never an accidental approve). + """ + raw_decision: Any = "" + notes = "" + if isinstance(decision, dict): + raw_decision = decision.get("decision", "") + notes = str(decision.get("notes", "") or "") + else: + raw_decision = decision + verb = str(raw_decision or "").strip().lower() + if verb in {"approve", "request_changes", "abandon"}: + return verb, notes + return "abandon", notes + + +def route_after_plan_gate(state: PipelineState) -> str: + """LangGraph conditional-edge after the plan gate (B2a). + + Reads the routing state :func:`plan_gate_node` wrote on resume (or on the + ceiling-reached terminal park) and maps it to a route id: + + * status ACTIVE + phase BUILD -> :data:`GATE_APPROVE_ROUTE` (END/approved); + * status ACTIVE + phase PLAN -> :data:`GATE_REVISE_ROUTE` (loop to planner); + * anything else (FAILED, or PARKED ceiling) -> :data:`GATE_TERMINAL_ROUTE`. + """ + status = state.get("status") + phase = state.get("current_phase") + if status == TaskStatus.ACTIVE.value and phase == _phase_value(Phase.PLAN): + return GATE_REVISE_ROUTE + if status == TaskStatus.ACTIVE.value and phase == _phase_value(Phase.BUILD): + return GATE_APPROVE_ROUTE + return GATE_TERMINAL_ROUTE + + def _author_questions(state: PipelineState) -> list[str]: """Deterministic stand-in for the Claude clarifier's question authoring. @@ -389,6 +630,7 @@ def build_graph( ] | None = None, dispatch_node: Callable[[PipelineState], Any] | None = None, + plan_gate: bool = False, ) -> CompiledStateGraph: """Assemble + compile the P1 pipeline ``StateGraph`` (§3.3, §7.1). @@ -489,6 +731,13 @@ def build_graph( "subgraph." ) + if plan_gate and review_node is None: + raise ValueError( + "build_graph: plan_gate requires review_node — the gate is the " + "resumable replacement for the review loop's terminal PARKED route, " + "so there is no review-cap dead-end to repoint without a review loop." + ) + builder: StateGraph = StateGraph(PipelineState) builder.add_node(INTAKE, _instrument(INTAKE, intake_node, transition_recorder)) builder.add_node(CLARIFY, _instrument(CLARIFY, clarify, transition_recorder)) @@ -506,12 +755,35 @@ def build_graph( builder.add_node(REVIEW, _instrument(REVIEW, review_node, transition_recorder)) builder.add_edge(PLAN, REVIEW) + # The review loop's PARKED/ESCALATE route normally terminates at END. When + # the plan gate is wired it is REPOINTED at the PLAN_GATE vertex instead: + # the cap dead-end suspends on a resumable human decision rather than + # terminally parking. The gate's own conditional edges then route + # approve -> END, request_changes -> PLAN (loop back), terminal -> END. + if plan_gate: + parked_target = PLAN_GATE + builder.add_node( + PLAN_GATE, + _instrument(PLAN_GATE, plan_gate_node, transition_recorder), + ) + builder.add_conditional_edges( + PLAN_GATE, + route_after_plan_gate, + { + GATE_APPROVE_ROUTE: END, + GATE_REVISE_ROUTE: PLAN, + GATE_TERMINAL_ROUTE: END, + }, + ) + else: + parked_target = END + if build_verify is None: # P2: the review's "build" route is the approved-plan terminus. builder.add_conditional_edges( REVIEW, route_review, - {BUILD_ROUTE: END, PLAN: PLAN, PARKED_ROUTE: END}, + {BUILD_ROUTE: END, PLAN: PLAN, PARKED_ROUTE: parked_target}, ) else: # P3 (opt-in): repoint the review's "build" route at the BUILD node, @@ -530,7 +802,7 @@ def build_graph( builder.add_conditional_edges( REVIEW, route_review, - {BUILD_ROUTE: BUILD_NODE, PLAN: PLAN, PARKED_ROUTE: END}, + {BUILD_ROUTE: BUILD_NODE, PLAN: PLAN, PARKED_ROUTE: parked_target}, ) # Linear order is BUILD -> DISPATCH -> VERIFY so DISPATCH triggers CI # and captures ``state["run_id"]`` BEFORE VERIFY reads it (design §4 diff --git a/agent-team/agent_team/task_model.py b/agent-team/agent_team/task_model.py index 5391f95..a40810f 100644 --- a/agent-team/agent_team/task_model.py +++ b/agent-team/agent_team/task_model.py @@ -114,6 +114,15 @@ class TaskRecord: # parks once it reaches VerifierConfig.max_build_loops so a perpetually- # failing task can never loop BUILD->DISPATCH->VERIFY forever (LOGIC-RACE-01). build_loops: int = 0 + # Count of plan-review human-decision gate visits already consumed for THIS + # task (Phase B2a). The graph routes a review-cap dead-end to the plan gate, + # which interrupts for an owner approve / request_changes / abandon decision. + # A human ``request_changes`` re-enters plan<->review, which can hit the cap + # and gate AGAIN; this counter is the combined ceiling on human-driven gate + # loops (mirrors PipelineState.plan_gate_visits) so the loop always + # terminates: once it reaches ``MAX_PLAN_GATE_VISITS`` the gate stops + # offering request_changes and the task goes terminal PARKED. + plan_gate_visits: int = 0 transport: str = "" created_at: str | None = None updated_at: str | None = None @@ -162,6 +171,12 @@ class PipelineState(TypedDict, total=False): # budget is real (LOGIC-RACE-01: it was previously read from the shared # wiring-time config and never advanced). build_loops: int + # Plan-review human-decision gate visits consumed for THIS task (mirrors + # TaskRecord.plan_gate_visits; Phase B2a). The combined ceiling on + # human-driven plan<->review loops: once it reaches MAX_PLAN_GATE_VISITS the + # gate stops offering request_changes and the task goes terminal PARKED, so + # the human-in-the-loop revision cycle can never spin forever. + plan_gate_visits: int transport: str created_at: str | None updated_at: str | None @@ -198,6 +213,7 @@ def task_from_dict(data: dict[str, Any]) -> TaskRecord: ci_correlation_tag=data.get("ci_correlation_tag"), dispatched_at=data.get("dispatched_at"), build_loops=data.get("build_loops", 0), + plan_gate_visits=data.get("plan_gate_visits", 0), transport=data.get("transport", ""), created_at=data.get("created_at"), updated_at=data.get("updated_at"), diff --git a/agent-team/tests/test_graph.py b/agent-team/tests/test_graph.py index afb8a15..52fa8ae 100644 --- a/agent-team/tests/test_graph.py +++ b/agent-team/tests/test_graph.py @@ -501,6 +501,213 @@ def test_p2_graph_loops_then_escalates_on_persistent_changes( assert len(final["review_verdicts"]) == 3 # looped to the cap, then escalated +# --- Plan-review human decision gate (Phase B2a). --------------------------- + + +def _plan_gate_graph(review_text: str): + """Compile a P2 graph with the plan gate wired (plan_gate=True). + + A reviewer that never approves drives the plan<->review loop to the review + cap, which — with the gate wired — suspends on the resumable PLAN_GATE + interrupt instead of terminally parking. + """ + from agent_team.nodes import review_loop + + review_loop.set_review_invoker(lambda prompt, **kw: review_text) + return build_graph( + checkpointer=_Saver(), + live_plan_node=_p2_plan_stub, + review_node=review_loop.bind_review_node(), + route_review=review_loop.route_after_review, + plan_gate=True, + ) + + +def _drive_to_plan_gate(graph): + """Start a task and resume the clarifier so it lands on the plan gate. + + Returns ``(thread_id, gate_payload)`` where ``gate_payload`` is the pending + plan-decision interrupt payload. + """ + from agent_team.graph import pending_question + + thread_id, _ = start_task(graph, transport="slack", slack_thread_ts="ROOT.1") + resume_task(graph, thread_id=thread_id, answer="scope is X") + payload = pending_question(graph, thread_id=thread_id) + return thread_id, payload + + +def test_plan_gate_requires_review_node() -> None: + with pytest.raises(ValueError, match="plan_gate requires review_node"): + build_graph(plan_gate=True) + + +def test_review_cap_interrupts_at_plan_gate(restore_review_invoker) -> None: + # Driving to the review-cap dead-end suspends on the plan gate (a pending + # plan_decision interrupt with the plan + findings) rather than parking. + from agent_team.graph import PLAN_DECISION_KIND + + graph = _plan_gate_graph("VERDICT: REQUEST CHANGES\nstill not ready") + thread_id, payload = _drive_to_plan_gate(graph) + + assert payload is not None + assert payload["kind"] == PLAN_DECISION_KIND + # Mirrors the clarifier contract: thread_id / question_id / turn / transport / + # deadline / slack_thread_ts all present (so pending_question + ResumeWorker + # drive it uniformly). + assert payload["thread_id"] == thread_id + assert payload["question_id"] + assert isinstance(payload["turn"], int) + assert payload["transport"] == "slack" + assert payload["deadline"] + assert payload["slack_thread_ts"] == "ROOT.1" + # Plus the review context the owner decides over. + assert payload["plan"] is not None + assert "still not ready" in payload["findings"] + + # It did NOT terminally park: the task is suspended on the gate interrupt + # (resumable), not finished. (The status channel still reads the review + # node's carried-over "parked" until the gate's resume overwrites it; the + # load-bearing signal is the live pending interrupt.) + from agent_team.graph import pending_question + + assert pending_question(graph, thread_id=thread_id) is not None + snapshot = graph.get_state(thread_config(thread_id)) + assert snapshot.next # graph is suspended, not at a terminal END + + +def test_plan_gate_approve_settles_as_approved_plan(restore_review_invoker) -> None: + # Command(resume={"decision":"approve"}) -> the same terminal "approved plan" + # state an auto-approved plan reaches today (phase BUILD, status ACTIVE). + graph = _plan_gate_graph("VERDICT: REQUEST CHANGES\nnope") + thread_id, _ = _drive_to_plan_gate(graph) + final = resume_task(graph, thread_id=thread_id, answer={"decision": "approve"}) + + assert final["current_phase"] == Phase.BUILD.value + assert final["status"] == TaskStatus.ACTIVE.value + + +def test_plan_gate_request_changes_loops_back_with_notes( + restore_review_invoker, +) -> None: + # Command(resume={"decision":"request_changes","notes":"do X"}) loops back to + # the planner; the notes must reach _format_review_feedback / the planner. + from agent_team.nodes.planner import _format_review_feedback + + captured: dict[str, object] = {} + + def capturing_plan(state): + captured["feedback"] = _format_review_feedback( + list(state.get("review_verdicts") or []) + ) + # After observing the folded-in notes, APPROVE on the re-plan so the + # graph settles (the review invoker is bound per-test below). + return _p2_plan_stub(state) + + from agent_team.nodes import review_loop + + # First review round REQUEST CHANGES (to reach the gate); after the human + # request_changes loops back, the next review APPROVES so the task settles. + texts = iter( + [ + "VERDICT: REQUEST CHANGES\nnot ready", + "VERDICT: APPROVE\nnow good", + ] + ) + last = "VERDICT: APPROVE\nnow good" + + def invoker(prompt, **kw): + nonlocal last + try: + last = next(texts) + except StopIteration: + pass + return last + + review_loop.set_review_invoker(invoker) + graph = build_graph( + checkpointer=_Saver(), + live_plan_node=capturing_plan, + review_node=review_loop.bind_review_node({"max_review_rounds": 1}), + route_review=review_loop.route_after_review, + plan_gate=True, + ) + + thread_id, _ = start_task(graph, transport="slack") + resume_task(graph, thread_id=thread_id, answer="scope is X") + final = resume_task( + graph, + thread_id=thread_id, + answer={"decision": "request_changes", "notes": "do X"}, + ) + + # The human notes reached the planner's review-feedback formatter on re-plan. + assert "do X" in str(captured.get("feedback", "")) + # And the loop re-entered plan -> review and settled (not stuck at the gate). + assert final["current_phase"] == Phase.BUILD.value + + +def test_plan_gate_abandon_fails(restore_review_invoker) -> None: + # Command(resume={"decision":"abandon"}) -> terminal FAILED. + graph = _plan_gate_graph("VERDICT: REQUEST CHANGES\nnope") + thread_id, _ = _drive_to_plan_gate(graph) + final = resume_task(graph, thread_id=thread_id, answer={"decision": "abandon"}) + + assert final["status"] == TaskStatus.FAILED.value + + +def test_plan_gate_unrecognized_decision_fails_safe(restore_review_invoker) -> None: + # An unrecognized decision must NOT accidentally approve — it fails terminal. + graph = _plan_gate_graph("VERDICT: REQUEST CHANGES\nnope") + thread_id, _ = _drive_to_plan_gate(graph) + final = resume_task(graph, thread_id=thread_id, answer={"decision": "huh?"}) + + assert final["status"] == TaskStatus.FAILED.value + + +def test_plan_gate_request_changes_terminates_at_ceiling( + restore_review_invoker, +) -> None: + # Repeated request_changes resumes must eventually hit the gate ceiling and + # go terminal PARKED ("ceiling reached") — NOT loop unbounded. The reviewer + # NEVER approves, so every gate visit is request_changes until the ceiling. + from agent_team.graph import MAX_PLAN_GATE_VISITS, pending_question + + graph = _plan_gate_graph("VERDICT: REQUEST CHANGES\nstill not ready") + thread_id, _ = _drive_to_plan_gate(graph) + + # Each request_changes consumes exactly one gate visit; bound the loop well + # above the ceiling to prove it terminates on its own, not by our cap. + for _ in range(MAX_PLAN_GATE_VISITS + 5): + if pending_question(graph, thread_id=thread_id) is None: + break + resume_task( + graph, + thread_id=thread_id, + answer={"decision": "request_changes", "notes": "again"}, + ) + + state = get_pipeline_state(graph, thread_id=thread_id) + assert pending_question(graph, thread_id=thread_id) is None + assert state["status"] == TaskStatus.PARKED.value + assert "ceiling" in (state.get("failure_reason") or "") + assert state.get("plan_gate_visits") == MAX_PLAN_GATE_VISITS + + +def test_auto_approved_plan_does_not_hit_gate(restore_review_invoker) -> None: + # No regression: an auto-APPROVED plan settles WITHOUT visiting the gate even + # when the gate is wired (gate only catches the review-cap dead-end). + from agent_team.graph import pending_question + + graph = _plan_gate_graph("VERDICT: APPROVE\nlooks solid") + thread_id, _ = start_task(graph, transport="slack") + final = resume_task(graph, thread_id=thread_id, answer="scope is X") + + assert final["current_phase"] == Phase.BUILD.value + assert pending_question(graph, thread_id=thread_id) is None + assert final.get("plan_gate_visits", 0) == 0 + + # --- P3 build -> verify subgraph wiring (opt-in). ---------------------------