"""Detection-quality gate — the hard feasibility constraint (Phase 4 task 8 of the retired openrouter-sast-harness plan). Aggregate recall and false-positive rate over the benchmark corpus must clear the project goal: recall >= 90% AND false-positive rate < 10%. This gate is computed **independently of the project score** so a score improvement can never unblock a gate-breaching change — the recall/FP constraint is never folded into the score reward. ``merge_gate`` composes this hard constraint with the (advisory-first) score gate by AND, so the score can only *add* a block, never remove one. Run as a module to evaluate the corpus and exit non-zero on a breach:: python -m openultrasast.gate """ from __future__ import annotations import sys from collections.abc import Mapping from dataclasses import dataclass from pathlib import Path from .benchmark import BenchmarkRun, evaluate_benchmark, load_benchmark_manifest, resolve_benchmark_source from .findings import quick_scan_findings from .mapping import analyze_entry_points, attach_reachability_hints from .preprocess import preprocess_repository from .rank import rank_targets # The in-scope corpus: language -> benchmark manifest names (smoke + real repo). # Single source of truth shared by the CI gate and the detection-benchmark test. LANGUAGE_MANIFESTS: dict[str, list[str]] = { "c_cpp": ["c-cpp-smoke", "cpp-damn-vulnerable"], "python": ["python-web-smoke", "python-vulnerable"], "javascript": ["javascript-node-web-smoke", "javascript-vulnerable"], "java": ["java-web-smoke", "java-spring-boot-vulnerable"], } RECALL_FLOOR = 0.90 FALSE_POSITIVE_CEILING = 0.10 DEFAULT_MANIFEST_DIR = Path("benchmarks/manifests") @dataclass(frozen=True) class DetectionMetrics: """Aggregate detection counts and the two rates derived from them.""" expected: int matched: int actual: int false_positives: int @property def recall(self) -> float: return self.matched / self.expected if self.expected else 1.0 @property def false_positive_rate(self) -> float: return self.false_positives / self.actual if self.actual else 0.0 def __add__(self, other: DetectionMetrics) -> DetectionMetrics: return DetectionMetrics( expected=self.expected + other.expected, matched=self.matched + other.matched, actual=self.actual + other.actual, false_positives=self.false_positives + other.false_positives, ) @dataclass(frozen=True) class DetectionGateVerdict: """Score-independent verdict: pass only if recall and FP both clear the goal.""" passed: bool overall: DetectionMetrics per_language: dict[str, DetectionMetrics] recall_floor: float fp_ceiling: float reasons: list[str] def _quick_scan(target: Path): # type: ignore[no-untyped-def] _, targets = preprocess_repository(target) targets = attach_reachability_hints(targets, analyze_entry_points(target, targets)) rankings = rank_targets(targets) return quick_scan_findings(target, targets, rankings) def language_metrics(manifest_names: list[str], manifest_dir: Path = DEFAULT_MANIFEST_DIR) -> DetectionMetrics: """Aggregate detection metrics across one language's manifests.""" total = DetectionMetrics(0, 0, 0, 0) for name in manifest_names: manifest_path = manifest_dir / f"{name}.toml" manifest = load_benchmark_manifest(manifest_path) target = resolve_benchmark_source(manifest_path, manifest) findings = _quick_scan(target) run = BenchmarkRun(benchmark_run_id="gate", root=Path("/tmp/gate"), manifest=manifest) result = evaluate_benchmark(run=run, mode="quick", findings=findings, scan_id=None, scan_run_dir=None) m = result.metrics total = total + DetectionMetrics( expected=m.expected_findings_total, matched=m.matched_findings_total, actual=m.actual_findings_total, false_positives=m.false_positive_findings_total, ) return total def evaluate_corpus( corpus: Mapping[str, list[str]] = LANGUAGE_MANIFESTS, manifest_dir: Path = DEFAULT_MANIFEST_DIR ) -> dict[str, DetectionMetrics]: """Compute per-language detection metrics across the whole corpus.""" return {language: language_metrics(names, manifest_dir) for language, names in sorted(corpus.items())} def detection_gate( per_language: Mapping[str, DetectionMetrics], *, recall_floor: float = RECALL_FLOOR, fp_ceiling: float = FALSE_POSITIVE_CEILING, ) -> DetectionGateVerdict: """Evaluate the hard recall/FP gate. No score input — it cannot be unblocked by score.""" overall = DetectionMetrics(0, 0, 0, 0) for metrics in per_language.values(): overall = overall + metrics reasons: list[str] = [] for language, metrics in sorted(per_language.items()): if metrics.recall < recall_floor: reasons.append(f"{language} recall {metrics.recall:.2%} below {recall_floor:.0%} ({metrics.matched}/{metrics.expected})") if metrics.false_positive_rate >= fp_ceiling: reasons.append( f"{language} FP rate {metrics.false_positive_rate:.2%} at or above {fp_ceiling:.0%} " f"({metrics.false_positives}/{metrics.actual})" ) if overall.recall < recall_floor: reasons.append(f"overall recall {overall.recall:.2%} below {recall_floor:.0%} ({overall.matched}/{overall.expected})") if overall.false_positive_rate >= fp_ceiling: reasons.append( f"overall FP rate {overall.false_positive_rate:.2%} at or above {fp_ceiling:.0%} ({overall.false_positives}/{overall.actual})" ) return DetectionGateVerdict( passed=not reasons, overall=overall, per_language=dict(sorted(per_language.items())), recall_floor=recall_floor, fp_ceiling=fp_ceiling, reasons=reasons, ) def merge_gate(detection: DetectionGateVerdict, score_gate: Mapping[str, object] | None = None) -> dict[str, object]: """Compose the hard detection gate with the (advisory-first) score gate by AND. The merge blocks if detection fails OR a *blocking* score gate fails. A passing score gate (however high the score) can never flip a failed detection gate to pass — the recall/FP constraint is hard and never folded into the score reward. """ score_passed = True if score_gate is None else bool(score_gate.get("passed", True)) return { "passed": detection.passed and score_passed, "detection_passed": detection.passed, "score_passed": score_passed, "reasons": list(detection.reasons), } def within_tolerance( baseline: DetectionMetrics, candidate: DetectionMetrics, *, recall_tolerance: float = 0.05, fp_tolerance: float = 0.05, ) -> bool: """True if ``candidate`` recall/FP stay within tolerance of ``baseline``. Used to confirm a model-backed stage does not drift detection beyond the configured band of the zero-dependency baseline. Recall may not fall more than ``recall_tolerance`` below baseline; FP may not rise more than ``fp_tolerance`` above it. """ recall_ok = candidate.recall >= baseline.recall - recall_tolerance fp_ok = candidate.false_positive_rate <= baseline.false_positive_rate + fp_tolerance return recall_ok and fp_ok def main() -> int: verdict = detection_gate(evaluate_corpus()) o = verdict.overall print( f"detection gate: recall {o.recall:.2%} ({o.matched}/{o.expected}), FP {o.false_positive_rate:.2%} ({o.false_positives}/{o.actual})" ) for language, m in verdict.per_language.items(): print( f" {language:<11} recall {m.recall:>6.2%} ({m.matched}/{m.expected}) " f"FP {m.false_positive_rate:>6.2%} ({m.false_positives}/{m.actual})" ) if verdict.passed: print("detection gate: PASS") return 0 print("detection gate: FAIL", file=sys.stderr) for reason in verdict.reasons: print(f" - {reason}", file=sys.stderr) return 1 if __name__ == "__main__": raise SystemExit(main())