diff --git a/assets/agents/review-gauntlet/graph.yaml b/assets/agents/review-gauntlet/graph.yaml index 56a42a0..a606d62 100644 --- a/assets/agents/review-gauntlet/graph.yaml +++ b/assets/agents/review-gauntlet/graph.yaml @@ -235,7 +235,7 @@ nodes: default_lanes: id: default_lanes type: script - description: Deterministic lane-selection fallback (reached only via select_lanes' fallback) — forces code-review + adversary, plus probe when consumer_surface, when the additive LLM judgment dies. Selection degrades WIDER, never narrower; the degradation note flows into lanes_summary via build_items. + description: Deterministic lane-selection fallback (reached only via select_lanes' fallback) — when the additive LLM judgment dies, forces code-review + adversary, plus probe when consumer_surface, plus security on attack-surface signals (auth/deps/exec) or hardened posture, unioned with any caller-forced lanes. Selection degrades WIDER, never narrower; the degradation note flows into lanes_summary via build_items. script: scripts/default_lanes.py timeout: 30 next: build_items diff --git a/assets/agents/review-gauntlet/scripts/default_lanes.py b/assets/agents/review-gauntlet/scripts/default_lanes.py index d33bca6..4387b3a 100644 --- a/assets/agents/review-gauntlet/scripts/default_lanes.py +++ b/assets/agents/review-gauntlet/scripts/default_lanes.py @@ -3,10 +3,13 @@ Reached ONLY via select_lanes' `fallback:` route, when the additive LLM judgment dies after retries. Emits `forced_lanes` — code-review + adversary, -plus probe when the deterministic signals flagged a consumer-facing -surface — which build_items honors exactly, so selection degrades WIDER, -never narrower. The degradation note goes into `lanes_degraded`; build_items -folds it into lanes_summary so the final report names the degradation. +plus probe when the deterministic signals flagged a consumer-facing surface, +plus security when the attack-surface signals (auth/deps/exec) or a hardened +posture demand it, unioned with any lanes the caller forced (canonicalized +with the same aliases build_items uses) — which build_items honors exactly, +so selection degrades WIDER, never narrower. The degradation note goes into +`lanes_degraded`; build_items folds it into lanes_summary so the final +report names the degradation. """ import json @@ -14,6 +17,27 @@ import os MAX_DETAIL_CHARS = 500 +# Keep in sync with build_items.ALIASES — a caller-forced lane must +# canonicalize the same way here as it would on the non-degraded path. +ALIASES = { + "code-review": "code-review", + "code_review": "code-review", + "code-reviewer": "code-review", + "adversary": "adversary", + "adversarial": "adversary", + "adversarial-review": "adversary", + "security": "security", + "security-review": "security", + "security-reviewer": "security", + "probe": "probe", + "usage": "probe", + "usage-pattern": "probe", + "usage-pattern-testing": "probe", +} + +# Stable emission order keeps the fallback deterministic. +LANE_ORDER = ["code-review", "adversary", "security", "probe"] + def load_state(): if path := os.environ.get("GRAPH_STATE_FILE"): @@ -24,9 +48,24 @@ def load_state(): def main(): state = load_state() - forced = ["code-review", "adversary"] + lanes = {"code-review", "adversary"} if state.get("consumer_surface"): - forced.append("probe") + lanes.add("probe") + # The deterministic security signals are already in state when this + # fallback runs; a degraded selection must never drop the security lane + # build_items' hard rules would have selected. + if ( + state.get("touches_auth") + or state.get("touches_deps") + or state.get("touches_exec") + or state.get("security_posture") == "hardened" + ): + lanes.add("security") + # Caller-forced lanes survive degradation — union, never overwrite. + for lane in state.get("forced_lanes") or []: + if canon := ALIASES.get(str(lane).strip().lower()): + lanes.add(canon) + forced = [lane for lane in LANE_ORDER if lane in lanes] raw = state.get("select_lanes_failure") detail = raw.strip().replace("\n", " ") if isinstance(raw, str) else "" if len(detail) > MAX_DETAIL_CHARS: diff --git a/assets/agents/review-gauntlet/scripts/verdict_gate.py b/assets/agents/review-gauntlet/scripts/verdict_gate.py index 6664624..4d6fa39 100644 --- a/assets/agents/review-gauntlet/scripts/verdict_gate.py +++ b/assets/agents/review-gauntlet/scripts/verdict_gate.py @@ -5,9 +5,11 @@ Parses each configured lane's verdict SENTINEL with a regex — no LLM re-reads, no judgment. The gate's hard rules: - a missing sentinel is a LANE FAILURE and blocks — never a pass; - - a PIPELINE-FAULT lane result (the lane's agent died after retries) or a - PIPELINE-FAULT in signals_error (builder/parse fault) blocks — a - degraded pipeline is never a pass; + - a PIPELINE-FAULT lane result (the lane's agent died after retries; + detected only when the report IS the lane_fault.py marker, i.e. it + STARTS with it — a real review merely quoting the marker flows to the + normal rules) or a PIPELINE-FAULT in signals_error (builder/parse + fault) blocks — a degraded pipeline is never a pass; - any 🔴 finding in the code-review report blocks regardless of the lane's own verdict line; - NEEDS-HUMAN without 🔴 passes but is surfaced as attention required; @@ -49,6 +51,13 @@ def main(): ) return line.strip()[:200] + def is_lane_fault(report): + # A genuine lane fault is the ENTIRE report emitted by lane_fault.py, + # which always begins with the marker. Matching the marker anywhere + # would let a real review that merely QUOTES "PIPELINE-FAULT:" bypass + # 🔴 counting and sentinel parsing. + return report.strip().startswith("PIPELINE-FAULT:") + def lane_fault_blocker(lane, report): blockers.append( f"{lane}: PIPELINE-FAULT — the lane failed after retries and produced " @@ -60,7 +69,7 @@ def main(): cr = lane_report("code_review_results") if cr is None: record("code-review", "SKIPPED", "not selected") - elif "PIPELINE-FAULT:" in cr: + elif is_lane_fault(cr): lane_fault_blocker("code-review", cr) reports.append(("code-review", cr)) else: @@ -91,7 +100,7 @@ def main(): adv = lane_report("adversary_results") if adv is None: record("adversary", "SKIPPED", "not selected") - elif "PIPELINE-FAULT:" in adv: + elif is_lane_fault(adv): lane_fault_blocker("adversary", adv) reports.append(("adversary", adv)) else: @@ -114,7 +123,7 @@ def main(): sec = lane_report("security_results") if sec is None: record("security", "SKIPPED", "not selected") - elif "PIPELINE-FAULT:" in sec: + elif is_lane_fault(sec): lane_fault_blocker("security", sec) reports.append(("security", sec)) else: @@ -135,7 +144,7 @@ def main(): pb = lane_report("probe_results") if pb is None: record("probe", "SKIPPED", "not selected") - elif "PIPELINE-FAULT:" in pb: + elif is_lane_fault(pb): lane_fault_blocker("probe", pb) reports.append(("probe", pb)) else: diff --git a/src/config/request_context.rs b/src/config/request_context.rs index cdaa788..0980c5a 100644 --- a/src/config/request_context.rs +++ b/src/config/request_context.rs @@ -8869,6 +8869,132 @@ mod tests { ); } + #[test] + fn gauntlet_verdict_gate_reds_block_despite_merge_ready() { + if !cmd_available("python3") { + eprintln!("skipping: python3 not available"); + return; + } + let state = json!({ + "code_review_results": ["🔴 [correctness] broken invariant\n\n**Verdict: MERGE-READY**"], + "adversary_results": ["ADVERSARIAL_REVIEW: CONFORMS"], + "security_results": [], + "probe_results": [] + }); + let out = run_gauntlet_script("verdict_gate.py", &state); + assert_eq!(out["gauntlet_verdict"], "BLOCKED"); + let report = out["gauntlet_report"].as_str().unwrap(); + assert!( + report.contains("code-review: 1 🔴 CRITICAL finding(s)"), + "🔴 findings must block regardless of the verdict line: {report}" + ); + assert!( + report.contains("| code-review | BLOCKED | 1 🔴 finding(s) |"), + "the lane table must carry the 🔴 count: {report}" + ); + } + + #[test] + fn gauntlet_verdict_gate_needs_human_without_reds_passes_with_attention() { + if !cmd_available("python3") { + eprintln!("skipping: python3 not available"); + return; + } + let state = json!({ + "code_review_results": ["🟡 [convention] minor nit\n\n**Verdict: NEEDS-HUMAN**"], + "adversary_results": ["ADVERSARIAL_REVIEW: CONFORMS"], + "security_results": [], + "probe_results": [] + }); + let out = run_gauntlet_script("verdict_gate.py", &state); + assert_eq!( + out["gauntlet_verdict"], "PASS", + "NEEDS-HUMAN without 🔴 must pass: {out}" + ); + let report = out["gauntlet_report"].as_str().unwrap(); + assert!( + report.contains("## Human attention required"), + "the attention section must be surfaced: {report}" + ); + assert!( + report.contains("| code-review | GREEN (attention) | NEEDS-HUMAN, no 🔴 |"), + "the lane table must record the attention status: {report}" + ); + } + + #[test] + fn gauntlet_verdict_gate_probe_inconclusive_blocks_with_environment_note() { + if !cmd_available("python3") { + eprintln!("skipping: python3 not available"); + return; + } + let state = json!({ + "code_review_results": ["**Verdict: MERGE-READY**"], + "adversary_results": ["ADVERSARIAL_REVIEW: CONFORMS"], + "security_results": [], + "probe_results": ["USAGE_PROBE: INCONCLUSIVE — could not boot the stack"] + }); + let out = run_gauntlet_script("verdict_gate.py", &state); + assert_eq!(out["gauntlet_verdict"], "BLOCKED"); + let report = out["gauntlet_report"].as_str().unwrap(); + assert!( + report.contains("the ENVIRONMENT could not be established"), + "INCONCLUSIVE must carry the environment note: {report}" + ); + assert!( + report.contains("| probe | BLOCKED | INCONCLUSIVE (environment) |"), + "the lane table must record the environment block: {report}" + ); + } + + #[test] + fn gauntlet_verdict_gate_crash_fails_closed() { + if !cmd_available("python3") { + eprintln!("skipping: python3 not available"); + return; + } + // A truthy non-list lane result makes lane_report index into an int, + // raising inside main; the top-level guard must emit BLOCKED (exit 0 + // is asserted in the helper) — never a crash into a silent pass. + let out = run_gauntlet_script("verdict_gate.py", &json!({"code_review_results": 42})); + assert_eq!(out["gauntlet_verdict"], "BLOCKED"); + assert!( + out["gauntlet_report"] + .as_str() + .unwrap() + .contains("verdict gate error"), + "the internal error must be named in the report: {out}" + ); + } + + #[test] + fn gauntlet_verdict_gate_quoted_fault_marker_not_misrouted() { + if !cmd_available("python3") { + eprintln!("skipping: python3 not available"); + return; + } + // A real review that merely QUOTES the marker mid-text (e.g. a code + // review of the gauntlet itself) must flow to the normal verdict + // rules: the 🔴 count blocks it, NOT the lane-fault rule. + let state = json!({ + "code_review_results": ["🔴 [correctness] gate mishandles reports quoting 'PIPELINE-FAULT:' mid-text\n\n**Verdict: MERGE-READY**"], + "adversary_results": ["ADVERSARIAL_REVIEW: CONFORMS"], + "security_results": [], + "probe_results": [] + }); + let out = run_gauntlet_script("verdict_gate.py", &state); + assert_eq!(out["gauntlet_verdict"], "BLOCKED"); + let report = out["gauntlet_report"].as_str().unwrap(); + assert!( + report.contains("| code-review | BLOCKED | 1 🔴 finding(s) |"), + "the quoting report must be routed to the 🔴 rule: {report}" + ); + assert!( + !report.contains("code-review: PIPELINE-FAULT — the lane failed"), + "a quoting report must NOT be treated as a lane fault: {report}" + ); + } + #[test] fn gauntlet_build_items_crash_fails_closed() { if !cmd_available("python3") { @@ -9030,6 +9156,64 @@ mod tests { assert_eq!(out, again, "same input must produce identical output"); } + #[test] + fn gauntlet_default_lanes_preserves_caller_forced_lanes() { + if !cmd_available("python3") { + eprintln!("skipping: python3 not available"); + return; + } + // A caller who forced a lane must not lose it when selection + // degrades — the fallback unions, never overwrites — and forced + // names canonicalize via the same aliases build_items uses. + let out = run_gauntlet_script( + "default_lanes.py", + &json!({"forced_lanes": ["security-reviewer"]}), + ); + assert_eq!( + out["forced_lanes"], + json!(["code-review", "adversary", "security"]), + "a caller-forced lane must survive the fallback: {out}" + ); + + // Unknown forced names are ignored, not crashed on. + let bogus = run_gauntlet_script("default_lanes.py", &json!({"forced_lanes": ["bogus"]})); + assert_eq!( + bogus["forced_lanes"], + json!(["code-review", "adversary"]), + "unknown forced names must be ignored: {bogus}" + ); + + let again = run_gauntlet_script( + "default_lanes.py", + &json!({"forced_lanes": ["security-reviewer"]}), + ); + assert_eq!(out, again, "same input must produce identical output"); + } + + #[test] + fn gauntlet_default_lanes_security_signals_force_security_lane() { + if !cmd_available("python3") { + eprintln!("skipping: python3 not available"); + return; + } + // The deterministic security signals are already in state when the + // fallback runs; degradation must never drop the security lane + // build_items' hard rules would have selected. + for state in [ + json!({"touches_auth": true}), + json!({"touches_deps": true}), + json!({"touches_exec": true}), + json!({"security_posture": "hardened"}), + ] { + let out = run_gauntlet_script("default_lanes.py", &state); + assert_eq!( + out["forced_lanes"], + json!(["code-review", "adversary", "security"]), + "security signal {state} must force the security lane: {out}" + ); + } + } + #[test] fn gauntlet_lane_prompts_carry_passthroughs() { use crate::graph::{GraphParser, NodeType};