diff --git a/assets/agents/review-gauntlet/graph.yaml b/assets/agents/review-gauntlet/graph.yaml index 0c08bb8..e2aeaf5 100644 --- a/assets/agents/review-gauntlet/graph.yaml +++ b/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 - # (agent-node retries each get a fresh per-attempt budget, - # src/graph/agent.rs retry_transient/bounded_attempt) + 1200 margin for - # the script stages (signals/select/build/gate) = 24000; the parse and - # select_lanes llm nodes carry no `timeout:` and are bounded by - # max_iterations. - timeout: 24000 + # A faulted lane can be re-run by retry_gate for up to 3 passes, and each + # pass's agent node retries up to max_attempts = 2 with a fresh per-attempt + # budget (src/graph/agent.rs retry_transient/bounded_attempt): + # 3 passes × 2 attempts × 11400 (max lane timeout) = 68400, + 1200 margin + # for the script stages (signals/select/build/retry gate/verdict gate) + # = 69600; the parse and select_lanes llm nodes carry no `timeout:` and + # 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 diff --git a/assets/agents/review-gauntlet/scripts/build_items.py b/assets/agents/review-gauntlet/scripts/build_items.py index f8198eb..0183c91 100644 --- a/assets/agents/review-gauntlet/scripts/build_items.py +++ b/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": "", - "signals_error": f"PIPELINE-FAULT: lane builder crashed — {e}; no lanes were run", + "lanes_summary": crash_context["lanes_summary"], + "signals_error": fault, } ) ) diff --git a/assets/agents/review-gauntlet/scripts/retry_gate.py b/assets/agents/review-gauntlet/scripts/retry_gate.py new file mode 100644 index 0000000..2831e47 --- /dev/null +++ b/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, + } + ) + ) diff --git a/assets/agents/review-gauntlet/scripts/signals.py b/assets/agents/review-gauntlet/scripts/signals.py index 6ccecb4..2e5e365 100644 --- a/assets/agents/review-gauntlet/scripts/signals.py +++ b/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") diff --git a/src/config/request_context.rs b/src/config/request_context.rs index ef594d5..7710ef5 100644 --- a/src/config/request_context.rs +++ b/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 = 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. diff --git a/src/graph/executor.rs b/src/graph/executor.rs index 2e1c4dc..71a87b5 100644 --- a/src/graph/executor.rs +++ b/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") {