feat(TASK-029): review-gauntlet fault-only lane re-run loop

Give the review gauntlet a deterministic retry loop for lanes that FAULT
(PIPELINE-FAULT marker, missing verdict sentinel, or a nested graph's own
degraded-run wording) without ever re-running a lane that reached a real
verdict.

- graph.yaml: the four lane maps route to a new `retry_gate` script node
  (static `next: verdict_gate`; emits `_next: build_items` to retry).
  initial_state gains lane_attempts / kept_results / retry_lanes /
  review_incomplete / gauntlet_started_at. settings.timeout 24000 -> 69600
  (3 passes x 2 attempts x 11400 + 1200 margin).
- scripts/retry_gate.py (new): stashes every lane report that ran
  (completed and faulted alike) and restores reports a re-run pass
  overwrote with []; classifies with verdict_gate.py's sentinel regexes
  verbatim; counts attempts; retries while attempts < 3 and inside a
  45600 s wall-clock budget; otherwise records the lane in
  review_incomplete and falls through. Crash guard appends to
  signals_error and never retries.
- scripts/signals.py: stamps gauntlet_started_at before the git work so a
  failed diff still starts the clock.
- scripts/build_items.py: retry pass emits items only for retry_lanes
  (with attempt numbers), adds a `retried:` note to lanes_summary, resets
  retry_lanes; falsy-guarded stamp fallback; crash guard appends to
  signals_error and preserves lanes_summary.
- Tests: hermetic python3-guarded script tests for retry_gate /
  build_items / signals, graph wiring + initial_state pins, sentinel and
  LANE_KEYS cross-pins, retry-budget envelope vs settings.timeout, a
  first-pass golden payload, and an engine-level executor test proving a
  script `_next` back-edge re-fires a static fan-out and an empty map
  re-run clears its collect_into.

Plan: PLAN-review-retry section A items 1-3 and 5. Verdict/report surface
(section A items 4 and 6) is TASK-032.

assets/agents/review-gauntlet/graph.yaml

@@ -41,13 +41,14 @@ variables:
 settings:
   max_loop_iterations: 10
   log_state_snapshots: true
-  # max over agent lanes of (max_attempts × lane timeout) = 2 × 11400 = 22800
+  # A faulted lane can be re-run by retry_gate for up to 3 passes, and each
-  # (agent-node retries each get a fresh per-attempt budget,
+  # pass's agent node retries up to max_attempts = 2 with a fresh per-attempt
-  # src/graph/agent.rs retry_transient/bounded_attempt) + 1200 margin for
+  # budget (src/graph/agent.rs retry_transient/bounded_attempt):
-  # the script stages (signals/select/build/gate) = 24000; the parse and
+  # 3 passes × 2 attempts × 11400 (max lane timeout) = 68400, + 1200 margin
-  # select_lanes llm nodes carry no `timeout:` and are bounded by
+  # for the script stages (signals/select/build/retry gate/verdict gate)
-  # max_iterations.
+  # = 69600; the parse and select_lanes llm nodes carry no `timeout:` and
-  timeout: 24000
+  # are bounded by max_iterations.
+  timeout: 69600
 
 initial_state:
   diff_spec: worktree
@@ -84,6 +85,11 @@ initial_state:
   adversary_results: []
   security_results: []
   probe_results: []
+  lane_attempts: {}
+  kept_results: {}
+  retry_lanes: []
+  review_incomplete: []
+  gauntlet_started_at: 0
   lanes_summary: ''
   gauntlet_verdict: ''
   gauntlet_report: ''
@@ -251,7 +257,7 @@ nodes:
   build_items:
     id: build_items
     type: script
-    description: Deterministic lane builder — hard rules are the floor, LLM flags may only add, forced lanes override exactly. Emits one 0-or-1-item list per lane; an empty list skips that lane's map.
+    description: Deterministic lane builder — hard rules are the floor, LLM flags may only add, forced lanes override exactly. Emits one 0-or-1-item list per lane; an empty list skips that lane's map. On a re-run pass (retry_lanes non-empty) it skips selection and emits items only for those lanes, each carrying its attempt number.
     script: scripts/build_items.py
     timeout: 30
     next: [map_code_review, map_adversary, map_security, map_probe]
@@ -265,7 +271,7 @@ nodes:
     output_key: lane_out
     collect_into: code_review_results
     max_concurrency: 1
-    next: verdict_gate
+    next: retry_gate
 
   map_adversary:
     id: map_adversary
@@ -276,7 +282,7 @@ nodes:
     output_key: lane_out
     collect_into: adversary_results
     max_concurrency: 1
-    next: verdict_gate
+    next: retry_gate
 
   map_security:
     id: map_security
@@ -287,7 +293,7 @@ nodes:
     output_key: lane_out
     collect_into: security_results
     max_concurrency: 1
-    next: verdict_gate
+    next: retry_gate
 
   map_probe:
     id: map_probe
@@ -298,7 +304,7 @@ nodes:
     output_key: lane_out
     collect_into: probe_results
     max_concurrency: 1
-    next: verdict_gate
+    next: retry_gate
 
   run_code_review:
     id: run_code_review
@@ -412,6 +418,14 @@ nodes:
     script: scripts/lane_fault.py
     timeout: 30
 
+  retry_gate:
+    id: retry_gate
+    type: script
+    description: Deterministic fault-only re-run gate. Stashes every lane report that ran this pass, restores reports a re-run pass overwrote with [] (a lane not being re-run maps over nothing), classifies each lane that ran as completed / faulted / skipped, counts attempts, and routes back into build_items for faulted lanes that still have attempts left inside the wall-clock budget; otherwise it records them in review_incomplete and falls through to verdict_gate. A real verdict is never re-run, and a gate crash never retries — it records a PIPELINE-FAULT and falls through.
+    script: scripts/retry_gate.py
+    timeout: 30
+    next: verdict_gate
+
   verdict_gate:
     id: verdict_gate
     type: script

assets/agents/review-gauntlet/scripts/build_items.py

@@ -9,6 +9,10 @@ skip mechanism — no dynamic routing anywhere.
 The LLM's add_* flags can only ADD lanes. A flag that is false never
 removes a lane the hard rules selected.
 
+A re-run pass (retry_gate emitted `retry_lanes`) skips selection: only the
+named lanes get an item, stamped with their attempt number, and every other
+lane maps over nothing — retry_gate restores their earlier reports.
+
 Fail-closed: a builder crash emits all four lists empty plus a
 PIPELINE-FAULT in signals_error — verdict_gate blocks on it and the crash
 never kills the graph.
@@ -16,6 +20,7 @@ never kills the graph.
 
 import json
 import os
+import time
 
 # Keep in sync with default_lanes.ALIASES — a caller-forced lane must
 # canonicalize the same way on the degraded path as it does here.
@@ -35,6 +40,18 @@ ALIASES = {
     "usage-pattern-testing": "probe",
 }
 
+# Keep in sync with retry_gate.LANES
+LANE_KEYS = {
+    "code-review": "code_review_items",
+    "adversary": "adversary_items",
+    "security": "security_items",
+    "probe": "probe_items",
+}
+
+# Filled by main() so the crash guard can append to an upstream fault
+# instead of overwriting it, and keep the first pass's lanes_summary.
+crash_context = {"signals_error": "", "lanes_summary": ""}
+
 
 def load_state():
     if path := os.environ.get("GRAPH_STATE_FILE"):
@@ -45,11 +62,38 @@ def load_state():
 
 def main():
     state = load_state()
+    prior = state.get("signals_error")
+    crash_context["signals_error"] = prior if isinstance(prior, str) else ""
+    # retry_gate's wall-clock budget is measured from this stamp; signals.py
+    # is the normal stamper, this is the defensive fallback.
+    stamp = {} if state.get("gauntlet_started_at") else {"gauntlet_started_at": time.time()}
 
     def text(key):
         v = state.get(key)
         return v.strip() if isinstance(v, str) else ""
 
+    crash_context["lanes_summary"] = text("lanes_summary")
+    retry_lanes = [lane for lane in (state.get("retry_lanes") or []) if lane in LANE_KEYS]
+    if retry_lanes:
+        attempts = state.get("lane_attempts") or {}
+        retried = [(lane, int(attempts.get(lane) or 0) + 1) for lane in retry_lanes]
+        note = "- retried: [" + ", ".join(f"{lane} (attempt {n})" for lane, n in retried) + "]"
+        summary = text("lanes_summary")
+        print(
+            json.dumps(
+                {
+                    **{key: [] for key in LANE_KEYS.values()},
+                    **{
+                        LANE_KEYS[lane]: [{"lane": lane, "attempt": n}] for lane, n in retried
+                    },
+                    "lanes_summary": f"{summary}\n{note}" if summary else note,
+                    "retry_lanes": [],
+                    **stamp,
+                }
+            )
+        )
+        return
+
     forced_raw = state.get("forced_lanes") or []
     forced, unknown = [], []
     for lane in forced_raw:
@@ -126,6 +170,7 @@ def main():
                 "security_items": [{"lane": "security"}] if "security" in lanes else [],
                 "probe_items": [{"lane": "probe"}] if "probe" in lanes else [],
                 "lanes_summary": "\n".join(f"- {r}" for r in reasons),
+                **stamp,
             }
         )
     )
@@ -134,6 +179,9 @@ def main():
 try:
     main()
 except Exception as e:  # noqa: BLE001 — a builder crash must never kill the graph
+    fault = f"PIPELINE-FAULT: lane builder crashed — {e}; no lanes were run"
+    if crash_context["signals_error"]:
+        fault = f"{crash_context['signals_error']}; {fault}"
     print(
         json.dumps(
             {
@@ -141,8 +189,8 @@ except Exception as e:  # noqa: BLE001 — a builder crash must never kill the g
                 "adversary_items": [],
                 "security_items": [],
                 "probe_items": [],
-                "lanes_summary": "",
+                "lanes_summary": crash_context["lanes_summary"],
-                "signals_error": f"PIPELINE-FAULT: lane builder crashed — {e}; no lanes were run",
+                "signals_error": fault,
             }
         )
     )

assets/agents/review-gauntlet/scripts/retry_gate.py

@@ -0,0 +1,159 @@
+#!/usr/bin/env python3
+"""Deterministic fault-only re-run gate for the review gauntlet.
+
+Sits between the four lane maps and the verdict gate. Every lane map runs
+on every pass — a lane that is not being re-run maps over [] — so a re-run
+pass overwrites the untouched lanes' collected results with []. The gate's
+hard rules:
+
+  - every lane that ran this pass is stashed in kept_results, completed
+    and faulted alike, and a lane whose map ran nothing this pass gets
+    its stashed report restored — the verdict gate always sees each
+    lane's LAST report;
+  - a lane is FAULTED only when its report is missing, non-string, the
+    lane_fault.py marker, sentinel-less, or carries the nested graph's own
+    degraded-run wording; a real verdict (DIVERGES, FAIL, INCONCLUSIVE,
+    NEEDS-HUMAN, …) is never re-run — the review's judgment stands;
+  - a faulted lane is re-run only while it has attempts left AND the
+    gauntlet is inside its wall-clock budget; otherwise it is recorded in
+    review_incomplete and its fault flows to the verdict gate as-is;
+  - the gate itself must never retry by accident: any internal error
+    records a PIPELINE-FAULT in signals_error and falls through to the
+    verdict gate, which blocks on it.
+"""
+
+import json
+import os
+import re
+import time
+
+MAX_RETRY_ELAPSED_SECS = 45600
+MAX_ATTEMPTS = 3
+
+# lane name -> (items key, results key)
+LANES = {
+    "code-review": ("code_review_items", "code_review_results"),
+    "adversary": ("adversary_items", "adversary_results"),
+    "security": ("security_items", "security_results"),
+    "probe": ("probe_items", "probe_results"),
+}
+
+# Must stay identical to verdict_gate.py's sentinel regexes: a lane this
+# gate calls completed must be one the verdict gate can parse, or a drift
+# would re-run lanes the verdict gate accepts (or accept lanes it cannot
+# read). A cross-pin test compares the literals in both files.
+SENTINELS = {
+    "code-review": r"Verdict:\**\s*\**\s*(MERGE-READY|NEEDS-HUMAN)",
+    "adversary": r"ADVERSARIAL_REVIEW:\s*(CONFORMS|DIVERGES)",
+    "security": r"SECURITY_REVIEW:\s*(PASS|FAIL)",
+    "probe": r"USAGE_PROBE:\s*(PASS|FAIL|INCONCLUSIVE)",
+}
+
+# Filled by main() so the crash guard can append to an upstream fault
+# instead of overwriting it, and report the lanes it had already declined.
+crash_context = {"signals_error": "", "review_incomplete": []}
+
+
+def load_state():
+    if path := os.environ.get("GRAPH_STATE_FILE"):
+        with open(path) as f:
+            return json.load(f)
+    return json.loads(os.environ.get("GRAPH_STATE", "{}"))
+
+
+def nested_degraded(lane, report):
+    # The nested graphs complete with a sentinel even when a stage inside
+    # them died; their own degraded-run wording marks a partial review,
+    # which is not a completed one. Same anchoring as verdict_gate.py.
+    if lane == "code-review":
+        verdict_line = re.search(r"^\*\*Verdict: .*$", report, re.MULTILINE)
+        return bool(
+            verdict_line
+            and re.search(
+                r"^\*\*Verdict: NEEDS-HUMAN\*\* — .*"
+                r"(pipeline fault\(s\) recorded|PIPELINE-FAULT:|report rendering error)",
+                verdict_line.group(0),
+            )
+        )
+    if lane == "adversary":
+        return bool(
+            re.search(
+                r"^Criteria: .*degraded run: pipeline fault recorded",
+                report,
+                re.MULTILINE,
+            )
+        )
+    return False
+
+
+def is_faulted(lane, report):
+    if not isinstance(report, str):
+        return True
+    if report.strip().startswith("PIPELINE-FAULT:"):
+        return True
+    if not re.search(SENTINELS[lane], report):
+        return True
+    return nested_degraded(lane, report)
+
+
+def main():
+    state = load_state()
+    prior = state.get("signals_error")
+    crash_context["signals_error"] = prior if isinstance(prior, str) else ""
+    incomplete = list(state.get("review_incomplete") or [])
+    crash_context["review_incomplete"] = incomplete
+    attempts = dict(state.get("lane_attempts") or {})
+    kept = dict(state.get("kept_results") or {})
+    out = {}
+    faulted = []
+
+    for lane, (items_key, results_key) in LANES.items():
+        results = state.get(results_key) or []
+        if not state.get(items_key):
+            if not results and lane in kept:
+                out[results_key] = [kept[lane]]
+            continue
+        report = results[0] if results else None
+        if report is not None:
+            kept[lane] = report
+        if is_faulted(lane, report):
+            faulted.append(lane)
+        attempts[lane] = int(attempts.get(lane) or 0) + 1
+
+    # A missing/zero stamp reads as an ancient start and declines every
+    # retry — the safe direction when the clock's origin is unknown.
+    started = state.get("gauntlet_started_at") or 0
+    within_budget = (time.time() - float(started)) <= MAX_RETRY_ELAPSED_SECS
+    retry = [lane for lane in faulted if within_budget and attempts[lane] < MAX_ATTEMPTS]
+    declined = [lane for lane in faulted if lane not in retry]
+    incomplete = sorted(set(incomplete) | set(declined))
+    crash_context["review_incomplete"] = incomplete
+
+    out.update(
+        {
+            "lane_attempts": attempts,
+            "kept_results": kept,
+            "retry_lanes": retry,
+            "review_incomplete": incomplete,
+        }
+    )
+    if retry:
+        out["_next"] = "build_items"
+    print(json.dumps(out))
+
+
+try:
+    main()
+except Exception as e:  # noqa: BLE001 — a gate crash must fall through, never retry
+    fault = f"PIPELINE-FAULT: retry gate crashed — {e}"
+    if crash_context["signals_error"]:
+        fault = f"{crash_context['signals_error']}; {fault}"
+    print(
+        json.dumps(
+            {
+                "retry_lanes": [],
+                "review_incomplete": crash_context["review_incomplete"],
+                "signals_error": fault,
+            }
+        )
+    )

assets/agents/review-gauntlet/scripts/signals.py

@@ -12,6 +12,7 @@ import json
 import os
 import re
 import subprocess
+import time
 
 def load_state():
     if path := os.environ.get("GRAPH_STATE_FILE"):
@@ -37,6 +38,10 @@ def git(*args):
 
 
 out = {"project_dir": proj}
+# retry_gate measures its wall-clock budget from this stamp; it is set
+# before the git work so a failed diff still starts the clock.
+if not state.get("gauntlet_started_at"):
+    out["gauntlet_started_at"] = time.time()
 try:
     if spec in ("", "worktree"):
         names = git("diff", "--name-only", "HEAD")

src/config/request_context.rs

@@ -11663,6 +11663,631 @@ mod tests {
         );
     }
 
+    fn now_secs() -> f64 {
+        SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .unwrap()
+            .as_secs_f64()
+    }
+
+    /// State for a single gauntlet lane whose map ran one item and collected
+    /// `report`, stamped with a fresh start so the retry budget is open.
+    fn gauntlet_lane_ran(lane: &str, report: &str) -> serde_json::Value {
+        let key = lane.replace('-', "_");
+        let mut state = json!({ "gauntlet_started_at": now_secs() });
+        state[format!("{key}_items").as_str()] = json!([{ "lane": lane }]);
+        state[format!("{key}_results").as_str()] = json!([report]);
+        state
+    }
+
+    const GAUNTLET_MERGE_READY: &str = "# Review\n\n**Verdict: MERGE-READY**\n";
+    const GAUNTLET_ADVERSARY_FAULT: &str =
+        "PIPELINE-FAULT: adversary lane failed after retries — Agent node failed: boom";
+
+    #[test]
+    fn gauntlet_retry_gate_stashes_and_restores_reports() {
+        if !cmd_available("python3") {
+            eprintln!("skipping: python3 not available");
+            return;
+        }
+        let first = run_gauntlet_script(
+            "retry_gate.py",
+            &json!({
+                "code_review_items": [{"lane": "code-review"}],
+                "code_review_results": [GAUNTLET_MERGE_READY],
+                "adversary_items": [{"lane": "adversary"}],
+                "adversary_results": [GAUNTLET_ADVERSARY_FAULT],
+                "gauntlet_started_at": now_secs(),
+            }),
+        );
+        assert_eq!(
+            first["kept_results"],
+            json!({"code-review": GAUNTLET_MERGE_READY, "adversary": GAUNTLET_ADVERSARY_FAULT}),
+            "completed and faulted lanes must both be stashed verbatim: {first}"
+        );
+        assert_eq!(first["retry_lanes"], json!(["adversary"]), "{first}");
+        assert_eq!(first["_next"], "build_items", "{first}");
+
+        // The re-run pass: code-review's map ran nothing and overwrote its
+        // collected results with []; adversary ran again and completed.
+        let second = run_gauntlet_script(
+            "retry_gate.py",
+            &json!({
+                "code_review_items": [],
+                "code_review_results": [],
+                "adversary_items": [{"lane": "adversary", "attempt": 2}],
+                "adversary_results": ["ADVERSARIAL_REVIEW: CONFORMS\nCriteria: all verified"],
+                "kept_results": first["kept_results"],
+                "lane_attempts": first["lane_attempts"],
+                "gauntlet_started_at": now_secs(),
+            }),
+        );
+        assert_eq!(
+            second["code_review_results"],
+            json!([GAUNTLET_MERGE_READY]),
+            "the overwritten code-review report must be restored from the stash: {second}"
+        );
+        assert_eq!(second["retry_lanes"], json!([]), "{second}");
+        assert!(
+            second.get("_next").is_none(),
+            "a pass with nothing to retry must fall through statically: {second}"
+        );
+        assert_eq!(
+            second["kept_results"]["adversary"],
+            "ADVERSARIAL_REVIEW: CONFORMS\nCriteria: all verified",
+            "the stash must carry the lane's latest report: {second}"
+        );
+        assert_eq!(second["lane_attempts"]["adversary"], 2, "{second}");
+
+        // A pass where nothing ran: both lanes come back from the stash,
+        // the faulted adversary report included — it is restored, not
+        // re-classified for another retry.
+        let restored = run_gauntlet_script(
+            "retry_gate.py",
+            &json!({
+                "code_review_items": [],
+                "code_review_results": [],
+                "adversary_items": [],
+                "adversary_results": [],
+                "security_items": [],
+                "security_results": [],
+                "probe_items": [],
+                "probe_results": [],
+                "kept_results": first["kept_results"],
+                "lane_attempts": first["lane_attempts"],
+                "gauntlet_started_at": now_secs(),
+            }),
+        );
+        assert_eq!(
+            restored["code_review_results"],
+            json!([GAUNTLET_MERGE_READY]),
+            "{restored}"
+        );
+        assert_eq!(
+            restored["adversary_results"],
+            json!([GAUNTLET_ADVERSARY_FAULT]),
+            "a faulted stash is restored as-is for the verdict gate: {restored}"
+        );
+        assert_eq!(restored["retry_lanes"], json!([]), "{restored}");
+        assert!(
+            restored.get("_next").is_none(),
+            "restoring a stash must never trigger a retry: {restored}"
+        );
+        assert_eq!(
+            restored["lane_attempts"], first["lane_attempts"],
+            "a lane that did not run must not be charged an attempt: {restored}"
+        );
+    }
+
+    #[test]
+    fn gauntlet_retry_gate_classifies_lane_reports() {
+        if !cmd_available("python3") {
+            eprintln!("skipping: python3 not available");
+            return;
+        }
+        let completed = run_gauntlet_script(
+            "retry_gate.py",
+            &gauntlet_lane_ran("code-review", GAUNTLET_MERGE_READY),
+        );
+        assert_eq!(
+            completed["retry_lanes"],
+            json!([]),
+            "a report carrying its sentinel is completed, never re-run: {completed}"
+        );
+        assert!(completed.get("_next").is_none(), "{completed}");
+
+        let faulted = [
+            ("adversary", GAUNTLET_ADVERSARY_FAULT),
+            ("security", "some text"),
+            (
+                "code-review",
+                "# Review\n\n**Verdict: NEEDS-HUMAN** — 1 pipeline fault(s) recorded; see Human attention required\n\n## Findings\n",
+            ),
+            (
+                "adversary",
+                "ADVERSARIAL_REVIEW: DIVERGES\nCriteria: none verified — degraded run: pipeline fault recorded (fail-closed).\nComplaints:\n- none",
+            ),
+        ];
+        for (lane, report) in faulted {
+            let out = run_gauntlet_script("retry_gate.py", &gauntlet_lane_ran(lane, report));
+            assert_eq!(
+                out["retry_lanes"],
+                json!([lane]),
+                "{lane} report {report:?} must classify as faulted: {out}"
+            );
+            assert_eq!(out["_next"], "build_items", "{lane}: {out}");
+        }
+
+        // A lane whose map ran nothing (items []) is skipped: no attempt, no
+        // retry, no incompleteness — and nothing to restore without a stash.
+        let mut state = gauntlet_lane_ran("adversary", GAUNTLET_ADVERSARY_FAULT);
+        state["probe_items"] = json!([]);
+        state["probe_results"] = json!([]);
+        let out = run_gauntlet_script("retry_gate.py", &state);
+        assert_eq!(out["retry_lanes"], json!(["adversary"]), "{out}");
+        assert_eq!(out["review_incomplete"], json!([]), "{out}");
+        assert!(
+            out["lane_attempts"].get("probe").is_none(),
+            "a skipped lane must not count an attempt: {out}"
+        );
+        assert!(out.get("probe_results").is_none(), "{out}");
+    }
+
+    #[test]
+    fn gauntlet_retry_gate_counts_attempts_per_lane_that_ran() {
+        if !cmd_available("python3") {
+            eprintln!("skipping: python3 not available");
+            return;
+        }
+        let out = run_gauntlet_script(
+            "retry_gate.py",
+            &json!({
+                "code_review_items": [{"lane": "code-review"}],
+                "code_review_results": [GAUNTLET_MERGE_READY],
+                "adversary_items": [{"lane": "adversary"}],
+                "adversary_results": [GAUNTLET_ADVERSARY_FAULT],
+                "security_items": [],
+                "probe_items": [],
+                "gauntlet_started_at": now_secs(),
+            }),
+        );
+        assert_eq!(
+            out["lane_attempts"],
+            json!({"code-review": 1, "adversary": 1}),
+            "only lanes that ran count an attempt: {out}"
+        );
+
+        let mut state = gauntlet_lane_ran("adversary", GAUNTLET_ADVERSARY_FAULT);
+        state["lane_attempts"] = json!({"adversary": 1});
+        let out = run_gauntlet_script("retry_gate.py", &state);
+        assert_eq!(out["lane_attempts"]["adversary"], 2, "{out}");
+    }
+
+    #[test]
+    fn gauntlet_retry_gate_retries_until_attempts_are_exhausted() {
+        if !cmd_available("python3") {
+            eprintln!("skipping: python3 not available");
+            return;
+        }
+        for prior in [0, 1] {
+            let mut state = gauntlet_lane_ran("adversary", GAUNTLET_ADVERSARY_FAULT);
+            state["lane_attempts"] = json!({"adversary": prior});
+            let out = run_gauntlet_script("retry_gate.py", &state);
+            assert_eq!(
+                out["_next"],
+                "build_items",
+                "attempt {} must route back into the builder: {out}",
+                prior + 1
+            );
+            assert_eq!(out["retry_lanes"], json!(["adversary"]), "{out}");
+            assert_eq!(
+                out["kept_results"]["adversary"], GAUNTLET_ADVERSARY_FAULT,
+                "{out}"
+            );
+            assert_eq!(out["review_incomplete"], json!([]), "{out}");
+        }
+
+        let mut state = gauntlet_lane_ran("adversary", GAUNTLET_ADVERSARY_FAULT);
+        state["lane_attempts"] = json!({"adversary": 2});
+        let out = run_gauntlet_script("retry_gate.py", &state);
+        assert!(
+            out.get("_next").is_none(),
+            "the third attempt must fall through to the verdict gate: {out}"
+        );
+        assert_eq!(out["lane_attempts"]["adversary"], 3, "{out}");
+        assert_eq!(out["retry_lanes"], json!([]), "{out}");
+        assert_eq!(out["review_incomplete"], json!(["adversary"]), "{out}");
+    }
+
+    #[test]
+    fn gauntlet_retry_gate_declines_outside_wall_clock_budget() {
+        if !cmd_available("python3") {
+            eprintln!("skipping: python3 not available");
+            return;
+        }
+        let mut state = gauntlet_lane_ran("adversary", GAUNTLET_ADVERSARY_FAULT);
+        state["gauntlet_started_at"] = json!(now_secs() - 100000.0);
+        let out = run_gauntlet_script("retry_gate.py", &state);
+        assert!(
+            out.get("_next").is_none(),
+            "a first-attempt fault past the budget must not retry: {out}"
+        );
+        assert_eq!(out["retry_lanes"], json!([]), "{out}");
+        assert_eq!(out["review_incomplete"], json!(["adversary"]), "{out}");
+    }
+
+    #[test]
+    fn gauntlet_retry_gate_crash_falls_through_without_retry() {
+        if !cmd_available("python3") {
+            eprintln!("skipping: python3 not available");
+            return;
+        }
+        // dict(42) raises inside main after the upstream fault was read.
+        let out = run_gauntlet_script(
+            "retry_gate.py",
+            &json!({
+                "lane_attempts": 42,
+                "adversary_items": [{"lane": "adversary"}],
+                "adversary_results": ["x"],
+                "signals_error": "PIPELINE-FAULT: earlier fault",
+            }),
+        );
+        assert!(
+            out["signals_error"]
+                .as_str()
+                .unwrap()
+                .starts_with("PIPELINE-FAULT: earlier fault; PIPELINE-FAULT: retry gate crashed"),
+            "the crash must append to the upstream fault, not overwrite it: {out}"
+        );
+        assert_eq!(out["retry_lanes"], json!([]), "{out}");
+        assert!(
+            out["review_incomplete"].is_array(),
+            "review_incomplete must stay a list on the crash path: {out}"
+        );
+        assert!(
+            out.get("_next").is_none(),
+            "a crashed gate must never retry: {out}"
+        );
+    }
+
+    /// retry_gate.py decides "completed" with the same sentinel regexes
+    /// verdict_gate.py parses; a drift would re-run lanes the verdict gate
+    /// accepts, or accept lanes it cannot read.
+    #[test]
+    fn gauntlet_retry_gate_sentinels_match_verdict_gate() {
+        let scripts = std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
+            .join("assets/agents/review-gauntlet/scripts");
+        let retry = read_to_string(scripts.join("retry_gate.py")).unwrap();
+        let verdict = read_to_string(scripts.join("verdict_gate.py")).unwrap();
+        for sentinel in [
+            r"Verdict:\**\s*\**\s*(MERGE-READY|NEEDS-HUMAN)",
+            r"ADVERSARIAL_REVIEW:\s*(CONFORMS|DIVERGES)",
+            r"SECURITY_REVIEW:\s*(PASS|FAIL)",
+            r"USAGE_PROBE:\s*(PASS|FAIL|INCONCLUSIVE)",
+        ] {
+            assert!(
+                retry.contains(sentinel),
+                "retry_gate.py must carry the sentinel regex {sentinel}"
+            );
+            assert!(
+                verdict.contains(sentinel),
+                "verdict_gate.py must carry the sentinel regex {sentinel}"
+            );
+        }
+        for fragment in [
+            r"^\*\*Verdict: NEEDS-HUMAN\*\* — .*",
+            r"(pipeline fault\(s\) recorded|PIPELINE-FAULT:|report rendering error)",
+        ] {
+            assert!(
+                retry.contains(fragment),
+                "retry_gate.py must carry the nested-degraded fragment {fragment}"
+            );
+            assert!(
+                verdict.contains(fragment),
+                "verdict_gate.py must carry the nested-degraded fragment {fragment}"
+            );
+        }
+        let adversary_verdict = read_to_string(
+            std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
+                .join("assets/agents/adversary/scripts/verdict.py"),
+        )
+        .unwrap();
+        let degraded = "degraded run: pipeline fault recorded";
+        assert!(
+            retry.contains(degraded),
+            "retry_gate.py must match the adversary's degraded-run wording {degraded:?}"
+        );
+        assert!(
+            adversary_verdict.contains(degraded),
+            "assets/agents/adversary/scripts/verdict.py must still emit {degraded:?}"
+        );
+    }
+
+    /// build_items.py's LANE_KEYS and retry_gate.py's LANES name the same
+    /// four lanes and item keys; a lane present in one but not the other
+    /// would be built but never stashed, or stashed but never rebuilt.
+    #[test]
+    fn gauntlet_build_items_lane_keys_match_retry_gate_lanes() {
+        let scripts = std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
+            .join("assets/agents/review-gauntlet/scripts");
+        let build = read_to_string(scripts.join("build_items.py")).unwrap();
+        let retry = read_to_string(scripts.join("retry_gate.py")).unwrap();
+        assert!(
+            build.contains("Keep in sync with retry_gate.LANES"),
+            "build_items.py LANE_KEYS must carry its sync note"
+        );
+        for lane in ["code-review", "adversary", "security", "probe"] {
+            let name = format!("\"{lane}\"");
+            let items = format!("{}_items", lane.replace('-', "_"));
+            for (file, source) in [("build_items.py", &build), ("retry_gate.py", &retry)] {
+                assert!(source.contains(&name), "{file} must name the lane {name}");
+                assert!(
+                    source.contains(&items),
+                    "{file} must carry the items key {items}"
+                );
+            }
+        }
+    }
+
+    #[test]
+    fn gauntlet_build_items_retry_pass_rebuilds_only_retried_lanes() {
+        if !cmd_available("python3") {
+            eprintln!("skipping: python3 not available");
+            return;
+        }
+        let out = run_gauntlet_script(
+            "build_items.py",
+            &json!({
+                "retry_lanes": ["adversary"],
+                "lane_attempts": {"adversary": 1, "code-review": 1},
+                "lanes_summary": "- code-review: always on",
+                "forced_lanes": ["code-review", "adversary"],
+                "gauntlet_started_at": 123.0,
+            }),
+        );
+        assert_eq!(
+            out["adversary_items"],
+            json!([{"lane": "adversary", "attempt": 2}]),
+            "{out}"
+        );
+        for key in ["code_review_items", "security_items", "probe_items"] {
+            assert_eq!(
+                out[key],
+                json!([]),
+                "{key} must map over nothing on a re-run pass, even when forced: {out}"
+            );
+        }
+        assert_eq!(out["retry_lanes"], json!([]), "{out}");
+        let summary = out["lanes_summary"].as_str().unwrap();
+        assert!(
+            summary.contains("retried: [adversary (attempt 2)]")
+                && summary.contains("code-review: always on"),
+            "the retry note must append to the first pass's summary: {summary}"
+        );
+        assert!(
+            out.get("gauntlet_started_at").is_none(),
+            "an existing start stamp must be left alone: {out}"
+        );
+    }
+
+    #[test]
+    fn gauntlet_build_items_stamps_start_time_once() {
+        if !cmd_available("python3") {
+            eprintln!("skipping: python3 not available");
+            return;
+        }
+        let out = run_gauntlet_script("build_items.py", &json!({"forced_lanes": ["code-review"]}));
+        assert!(
+            out["gauntlet_started_at"].as_f64().is_some_and(|t| t > 0.0),
+            "a first pass without a stamp must start the clock: {out}"
+        );
+        let out = run_gauntlet_script(
+            "build_items.py",
+            &json!({"forced_lanes": ["code-review"], "gauntlet_started_at": 5.0}),
+        );
+        assert!(
+            out.get("gauntlet_started_at").is_none(),
+            "an existing start stamp must be left alone: {out}"
+        );
+    }
+
+    /// With the clock already stamped (the normal graph path — signals.py
+    /// runs first), a first pass must serialize exactly the five-key payload
+    /// it always has: same keys, same order, same values, nothing extra.
+    #[test]
+    fn gauntlet_build_items_first_pass_payload_is_unchanged_when_stamped() {
+        if !cmd_available("python3") {
+            eprintln!("skipping: python3 not available");
+            return;
+        }
+        let path = std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
+            .join("assets/agents/review-gauntlet/scripts/build_items.py");
+        let state =
+            json!({"forced_lanes": ["code-review", "adversary"], "gauntlet_started_at": 5.0});
+        let out = std::process::Command::new("python3")
+            .arg(&path)
+            .env("GRAPH_STATE", state.to_string())
+            .env_remove("GRAPH_STATE_FILE")
+            .output()
+            .expect("failed to invoke python3 build_items.py");
+        assert!(
+            out.status.success(),
+            "build_items.py exited nonzero: {}",
+            String::from_utf8_lossy(&out.stderr)
+        );
+        let stdout = String::from_utf8_lossy(&out.stdout);
+        assert_eq!(
+            stdout.trim_end(),
+            r#"{"code_review_items": [{"lane": "code-review"}], "adversary_items": [{"lane": "adversary"}], "security_items": [], "probe_items": [], "lanes_summary": "- forced lanes honored exactly: ['adversary', 'code-review']"}"#,
+            "the stamped first-pass payload must be byte-identical to the pre-loop form"
+        );
+    }
+
+    #[test]
+    fn gauntlet_build_items_crash_appends_to_prior_fault() {
+        if !cmd_available("python3") {
+            eprintln!("skipping: python3 not available");
+            return;
+        }
+        let out = run_gauntlet_script(
+            "build_items.py",
+            &json!({
+                "forced_lanes": 42,
+                "signals_error": "PIPELINE-FAULT: earlier",
+                "lanes_summary": "- code-review: always on",
+            }),
+        );
+        assert!(
+            out["signals_error"]
+                .as_str()
+                .unwrap()
+                .starts_with("PIPELINE-FAULT: earlier; PIPELINE-FAULT: lane builder crashed"),
+            "the crash must append to the upstream fault, not overwrite it: {out}"
+        );
+        assert_eq!(
+            out["lanes_summary"], "- code-review: always on",
+            "a crash must not wipe the first pass's lanes_summary: {out}"
+        );
+    }
+
+    #[test]
+    fn gauntlet_signals_stamps_start_time_before_git_runs() {
+        if !cmd_available("python3") {
+            eprintln!("skipping: python3 not available");
+            return;
+        }
+        let dir = utils::temp_file("-gauntlet-signals-", "");
+        create_dir_all(&dir).unwrap();
+        let project_dir = dir.to_string_lossy().to_string();
+        let out = run_gauntlet_script(
+            "signals.py",
+            &json!({"project_dir": project_dir, "diff_spec": "worktree"}),
+        );
+        let stamped = run_gauntlet_script(
+            "signals.py",
+            &json!({"project_dir": project_dir, "diff_spec": "worktree", "gauntlet_started_at": 77.0}),
+        );
+        let _ = remove_dir_all(&dir);
+        assert!(
+            out["signals_error"]
+                .as_str()
+                .is_some_and(|e| e.contains("diff signals could not be computed")),
+            "a non-repo project_dir must degrade the signals: {out}"
+        );
+        assert!(
+            out["gauntlet_started_at"].as_f64().is_some_and(|t| t > 0.0),
+            "the start stamp must be set even when git fails: {out}"
+        );
+        assert!(
+            stamped.get("gauntlet_started_at").is_none(),
+            "an existing start stamp must be left alone: {stamped}"
+        );
+    }
+
+    #[test]
+    fn gauntlet_retry_loop_wiring() {
+        use crate::graph::NodeType;
+        let graph = load_bundled_graph("review-gauntlet");
+        let maps = [
+            "map_code_review",
+            "map_adversary",
+            "map_security",
+            "map_probe",
+        ];
+        for id in maps {
+            let node = graph
+                .get_node(id)
+                .unwrap_or_else(|| panic!("review-gauntlet must have a {id} node"));
+            assert_eq!(
+                node.next.as_ref().map(|n| n.as_slice()),
+                Some(&["retry_gate".to_string()][..]),
+                "{id} must route into retry_gate: {:?}",
+                node.next
+            );
+        }
+        let gate = graph
+            .get_node("retry_gate")
+            .expect("review-gauntlet must have a retry_gate node");
+        let NodeType::Script(s) = &gate.node_type else {
+            panic!("retry_gate must be a script node");
+        };
+        assert_eq!(s.script, "scripts/retry_gate.py");
+        assert_eq!(
+            gate.next.as_ref().map(|n| n.as_slice()),
+            Some(&["verdict_gate".to_string()][..]),
+            "retry_gate must fall through statically to verdict_gate: {:?}",
+            gate.next
+        );
+        let build = graph
+            .get_node("build_items")
+            .expect("review-gauntlet must have a build_items node");
+        let fan_out: Vec<String> = maps.iter().map(|m| m.to_string()).collect();
+        assert_eq!(
+            build.next.as_ref().map(|n| n.as_slice()),
+            Some(fan_out.as_slice()),
+            "build_items must still fan out to the four maps: {:?}",
+            build.next
+        );
+        for (key, expected) in [
+            ("lane_attempts", json!({})),
+            ("kept_results", json!({})),
+            ("retry_lanes", json!([])),
+            ("review_incomplete", json!([])),
+            ("gauntlet_started_at", json!(0)),
+        ] {
+            assert_eq!(
+                graph.initial_state.get(key),
+                Some(&expected),
+                "initial_state.{key} must seed the retry loop"
+            );
+        }
+        assert_eq!(graph.settings.timeout, Some(69600));
+        let script = std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
+            .join("assets/agents/review-gauntlet/scripts/retry_gate.py");
+        assert!(script.exists(), "{} must exist", script.display());
+    }
+
+    /// MAX_RETRY_ELAPSED_SECS in retry_gate.py and settings.timeout in
+    /// graph.yaml live in different files. A re-run admitted at the very end
+    /// of the retry budget can still burn a full lane envelope (max_attempts
+    /// × the slowest lane) plus the script stages, and a graph timeout kills
+    /// from outside — no verdict would be rendered.
+    #[test]
+    fn gauntlet_retry_budget_fits_inside_graph_timeout() {
+        use crate::graph::NodeType;
+        const OTHER_STAGES_MARGIN_SECS: u64 = 1200;
+        let script = std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
+            .join("assets/agents/review-gauntlet/scripts/retry_gate.py");
+        let source = read_to_string(&script).unwrap();
+        let line = source
+            .lines()
+            .find(|l| l.starts_with("MAX_RETRY_ELAPSED_SECS = "))
+            .expect("retry_gate.py must define MAX_RETRY_ELAPSED_SECS");
+        let max_retry: u64 = line
+            .split_once(" = ")
+            .and_then(|(_, v)| v.trim().parse().ok())
+            .unwrap_or_else(|| panic!("MAX_RETRY_ELAPSED_SECS no longer parses: {line}"));
+
+        let graph = load_bundled_graph("review-gauntlet");
+        let max_lane_envelope = graph
+            .nodes
+            .values()
+            .filter_map(|node| match &node.node_type {
+                NodeType::Agent(a) => a.timeout.map(|t| u64::from(a.max_attempts) * t),
+                _ => None,
+            })
+            .max()
+            .expect("review-gauntlet must have agent lanes with timeouts");
+        let graph_timeout = graph
+            .settings
+            .timeout
+            .expect("review-gauntlet must set settings.timeout");
+        assert!(
+            max_retry + max_lane_envelope + OTHER_STAGES_MARGIN_SECS <= graph_timeout,
+            "assets/agents/review-gauntlet/scripts/retry_gate.py MAX_RETRY_ELAPSED_SECS ({max_retry}) + the largest lane envelope, max_attempts × timeout ({max_lane_envelope}) + {OTHER_STAGES_MARGIN_SECS}s margin must fit inside assets/agents/review-gauntlet/graph.yaml settings.timeout ({graph_timeout})"
+        );
+    }
+
     // The value run_checks executes with a shell must never be something an
     // LLM lifted out of the prompt — the prompt also carries pasted plan/diff
     // text from the repo under review.

src/graph/executor.rs

@@ -1325,6 +1325,117 @@ nodes:
         );
     }
 
+    /// A script `_next` back-edge into a node with a static fan-out must
+    /// re-fire every fan-out target on the next pass, and a map that
+    /// re-runs over zero items must overwrite its `collect_into` with `[]`
+    /// rather than leave the previous pass's results in place.
+    #[tokio::test]
+    async fn script_next_back_edge_refires_static_fan_out_and_empty_map_rerun_clears_collect_into()
+    {
+        if !cmd_available("python3") {
+            eprintln!("skipping: python3 not available");
+            return;
+        }
+        let ws = TestWorkspace::new();
+        ws.write_script(
+            "build.py",
+            r#"#!/usr/bin/env python3
+import os, json
+state = json.loads(os.environ.get("GRAPH_STATE", "{}"))
+if state["pass"] == 0:
+    print(json.dumps({"a_items": [1], "b_items": [1]}))
+else:
+    print(json.dumps({"a_items": [], "b_items": [2]}))
+"#,
+        );
+        ws.write_script(
+            "worker.py",
+            r#"#!/usr/bin/env python3
+import os, json
+state = json.loads(os.environ.get("GRAPH_STATE", "{}"))
+print(json.dumps({"output": f"report-{state['item']}"}))
+"#,
+        );
+        ws.write_script(
+            "gate.py",
+            r#"#!/usr/bin/env python3
+import os, json
+state = json.loads(os.environ.get("GRAPH_STATE", "{}"))
+if state["pass"] == 0:
+    print(json.dumps({"pass": 1, "_next": "build"}))
+else:
+    print(json.dumps({}))
+"#,
+        );
+
+        let yaml = r#"
+name: fan_out_rerun_test
+start: build
+initial_state:
+  pass: 0
+  a_items: []
+  b_items: []
+  a_results: []
+  b_results: []
+nodes:
+  build:
+    type: script
+    script: build.py
+    next: [map_a, map_b]
+  map_a:
+    type: map
+    over: "{{a_items}}"
+    as: item
+    branch: worker
+    collect_into: a_results
+    next: gate
+  map_b:
+    type: map
+    over: "{{b_items}}"
+    as: item
+    branch: worker
+    collect_into: b_results
+    next: gate
+  worker:
+    type: script
+    script: worker.py
+    state_updates: {}
+    timeout: 60
+  gate:
+    type: script
+    script: gate.py
+    next: done
+  done:
+    type: end
+    output: "{{a_results}}|{{b_results}}"
+"#;
+        let graph: Graph = serde_yaml::from_str(yaml).unwrap();
+        let mut ctx = make_ctx();
+        let abort = create_abort_signal();
+        let result = GraphExecutor::new(graph, &ws.dir)
+            .execute(&mut ctx, abort)
+            .await
+            .unwrap_or_else(|e| panic!("executor failed: {e:#}"));
+
+        let (a, b) = result
+            .split_once('|')
+            .unwrap_or_else(|| panic!("expected `a|b`, got: {result}"));
+        let a: Value =
+            serde_json::from_str(a).unwrap_or_else(|_| panic!("a_results not JSON: {result}"));
+        let b: Value =
+            serde_json::from_str(b).unwrap_or_else(|_| panic!("b_results not JSON: {result}"));
+        assert_eq!(
+            b,
+            serde_json::json!(["report-2"]),
+            "the back-edge into build must re-fire both fan-out targets: {result}"
+        );
+        assert_eq!(
+            a,
+            serde_json::json!([]),
+            "map_a must re-fire on the back-edge pass and clear its collect_into over zero items: {result}"
+        );
+    }
+
     #[tokio::test]
     async fn parallel_branch_error_aborts_super_step() {
         if !cmd_available("bash") {