#!/usr/bin/env python3 """ bench.py — recall/precision over the security corpus (design spec §6). Modes: --inventory load every corpus canonical.json, print the coverage ledger as JSON. Reports blind_spots (classes with no MEASURED recall) — no-false-coverage-claim. --findings FILE [--require ID...] FILE = {cell_id: [finding_str, ...]} produced by the orchestrator. For each cell, recall = canonical flagged? (right-reason match: code-id symbol -> contiguous; phrase symbol -> all distinctive tokens in one finding; hand-judge-only canonicals PUNT). --require makes named cells mandatory: a missing canonical exits non-zero (the regression gate). bench does NOT call the LLM. It consumes orchestrator output. The orchestrator runs the bands (k>=3). """ import argparse, glob, importlib.util, json, os, subprocess, sys ROOT = os.path.dirname(os.path.abspath(__file__)) def _load(name, path): spec = importlib.util.spec_from_file_location(name, path) m = importlib.util.module_from_spec(spec) spec.loader.exec_module(m) return m # the right-reason match authority — unified into prevent/verdict.py (ARCHITECTURE §4) so the credit/punt # decision stops drifting between here and orchestrator/detect.py. bench consumes it; never re-implement it. verdict = _load("sg_verdict", os.path.join(ROOT, "prevent", "verdict.py")) CORPUS = os.path.join(ROOT, "domains", "security", "corpus") # band-3 deterministic detectors keep their cells WITH the detector (captured-audit / config fixtures, NOT # git-fix-pinned) so the corpus/ contract stays purely git-fix-pinned LLM ground truth. bench is the single # coverage ledger, so it globs both locations. DETECTOR_CELLS = os.path.join(ROOT, "domains", "security", "detectors", "*", "cells", "canonical.json") def load_cells(): cells = [] for cj in sorted(glob.glob(os.path.join(CORPUS, "*", "canonical.json")) + glob.glob(DETECTOR_CELLS)): with open(cj) as f: c = json.load(f) c["_dir"] = os.path.dirname(cj) cells.append(c) return cells def is_flagged(canonical, findings): """Right-reason auto-credit for a POSITIVE (canonical) expectation. Compat shim: the authority now lives in prevent/verdict.py (unified, was scattered here + orchestrator/detect.py). True iff Verdict credits. Hand-judge `why` and not-flagged both yield non-CREDIT (PUNT) -> False, identical to the original binary.""" exp = verdict.Expectation(symbol=canonical.get("canonical_symbol") or "", why=canonical.get("why") or "", polarity=verdict.POSITIVE) return verdict.match(findings, exp).outcome == verdict.CREDIT def cmd_inventory(cells): # DRIFT NOTICE: ledger.py (ARCHITECTURE §5) is the DERIVED authority for coverage/blind-spots/rate-capability. # This CURATED inventory is a human-readable convenience that MUST agree with `python3 ledger.py`; on any # disagreement, ledger is the oracle (§5 "a parallel hand-edited registry drifts"). Reconcile here, not there. # CURATED, not computed-from-corpus: S4/S6 are pilot-measured, so a naive "class with no corpus cell = blind" # would wrongly re-flag them. blind = classes with NO measured recall of any kind. S11 left the list 2026-06-17 # (band-3 deps+headers BUILT, RED/GREEN, tests/test_s11_band3.py). S9 gained a band-3 oracle detector cell # 2026-06-18 (detectors/oracle/cells, cross-file CATCH RED/GREEN, tests/test_oracle_xfile.py). blind = sorted(set()) # NONE: S1–S13 all measured (S2/S3 n=3; S1/S5/S7/S8/S10 n=1; S4/S6/S12/S13 pilot; S9 oracle; S11 band-3) out = { "cells": [{"id": c["id"], "class": c["class"], "band": c["band"], "shape": c.get("shape")} for c in cells], "n_cells": len(cells), "blind_spots": blind, "band3_detectors": ["deps", "headers"], # blind_spots=[] means EVERY class has a detector with a RED/GREEN cell — NOT every class is recall-validated # at rate. Read it as coverage-exists, never as 11/11-measured-at-rate. "rate_validated": ["S2", "S3"], # n>=3 cells → recall is a RATE "point_estimate": ["S1", "S5", "S7", "S8", "S10"], # n=1 LLM cells "pilot": ["S4", "S6", "S12", "S13"], "oracle": ["S9"], "deterministic_n1": ["S11"], # band-3 deps + headers: deterministic but each is a single cell "note": "blind_spots=[] = every class has a detector+cell, NOT every class rate-validated (n>=3). Recall is " "a RATE only for S2/S3; the rest are n=1 / pilot / oracle. S11 is DETERMINISTIC via band-3 detectors " "(deps n=1 real-CVE cell, headers n=1 synthetic config cell; no LLM) — see tests/test_s11_band3.py.", } print(json.dumps(out, indent=2)) return 0 def cmd_findings(cells, findings_path, required): with open(findings_path) as f: findings_by_cell = json.load(f) by_id = {c["id"]: c for c in cells} rows, missing_required = [], [] for cell_id, findings in findings_by_cell.items(): c = by_id.get(cell_id) if not c: rows.append({"id": cell_id, "status": "UNKNOWN_CELL"}); continue flagged = is_flagged(c, findings) rows.append({"id": cell_id, "recall": int(flagged), "class": c["class"]}) if cell_id in required and not flagged: missing_required.append(cell_id) print(json.dumps({"rows": rows, "missing_required": missing_required}, indent=2)) if missing_required: print(f"REGRESSION: required canonical(s) not flagged: {missing_required}", file=sys.stderr) return 1 return 0 def cmd_mapper(repo_root, cells): """Surface-recall = enumerated/actual on a REAL tree vs the INDEPENDENT grep denominator (NOT the canonical sample — that is near-tautological). canonical-`file` routing is a SECONDARY confirmation, GATED on the source repo being checked out (the zync trio is not -> report the gate, never a false 100%).""" import importlib.util def _load(name, path): spec = importlib.util.spec_from_file_location(name, path); m = importlib.util.module_from_spec(spec) spec.loader.exec_module(m); return m mapper = _load("sg_mapper", os.path.join(ROOT, "orchestrator", "mapper.py")) conv = _load("sg_conv", os.path.join(ROOT, "orchestrator", "conventions.py")) table_kinds = {c["kind"] for c in conv.CONVENTIONS} act = mapper.actual_surface(repo_root) enr = {e.path for e in mapper.enumerate_surface(repo_root)} ap_set = {p for p, _ in act} hit = ap_set & enr by_kind = {} for p, k in act: d = by_kind.setdefault(k, {"actual": 0, "enumerated": 0}) d["actual"] += 1 d["enumerated"] += 1 if p in enr else 0 not_enum_kinds = sorted({k for p, k in act if p not in enr and k not in table_kinds}) canon_files = {c.get("file") for c in cells if c.get("file")} present = {f for f in canon_files if os.path.isfile(os.path.join(repo_root, f))} if not present: canon_conf = "gated: no canonical source file present under repo_root (zync trio not checked out)" else: enr_rel = {os.path.relpath(p, repo_root) for p in enr} canon_conf = {"present": len(present), "enumerated": len(present & enr_rel), "missed": sorted(present - enr_rel)} file_routed = mapper.file_routed_recall(repo_root) out = { "kind_coverage_by_volume": round(len(hit) / max(1, len(ap_set)), 3), "actual": len(ap_set), "enumerated": len(hit), "by_kind": by_kind, "not_enumerated_kinds": not_enum_kinds, "within_kind_recall": file_routed, "canonical_confirmation": canon_conf, "note": "kind_coverage_by_volume = enumerated/actual: share of detected entry points whose KIND the mapper " "enumerates. File-routed kinds (http-file-route, edge-function) are counted by LOCATION " "(FILE_ROUTE_RULES) on BOTH sides; call-registered kinds by CONTENT signal. Per-enumerated-kind " "recall is 1.0 BY CONSTRUCTION; this only shows no-row kinds as misses. within_kind_recall (file-routed " "only) is now LOCATION-based and 1.0 by construction; retired_signal_recall shows how lossy the old " "content signal was. canonical_confirmation is SECONDARY and gated on the source repo; never read a " "gated value as recall.", } print(json.dumps(out, indent=2)) return 0 def _shipping_detector(cell_dir, detectors): """A band-3 conformance cell lives INSIDE its detector's dir -> that detector is the cell's INTENDED catcher (same containment rule ledger.py grades deterministic by). None for git-fix-pinned corpus cells.""" for d in detectors: dd = d.get("_dir") if dd and (cell_dir == dd or cell_dir.startswith(dd + os.sep)): return d return None def _cell_vuln_path(cell_dir, file_field): """On-disk vuln fixture for a cell. file_field is repo-source-relative for corpus cells (no local copy of that path) but detector-dir-relative for band-3 cells (a real local fixture). Prefer the named local fixture, else any vuln.* in the cell dir.""" if file_field: cand = os.path.join(cell_dir, os.path.basename(file_field)) if os.path.isfile(cand): return cand for g in sorted(glob.glob(os.path.join(cell_dir, "vuln.*"))): return g return None def cmd_routing(cells, detectors_root): """Class-routing recall — the narrowing shipping gate (ARCHITECTURE §10.7). For EVERY cell assert its INTENDED detector is SELECTED for the cell's vuln file. INTENDED = the band-3 detector whose dir contains the cell, else the non-baseline LLM specialist(s) covering the class. baseline (the always-on FLOOR) is EXCLUDED from the assertion ON PURPOSE: it catches the Narrowable corpus cells (S4/S5/S6/S7), so a baseline-inclusive 'someone caught it' is trivially green and measures the floor, not routing (the ledger false-green failure mode). Discriminator the gate enforces: disable a specialist -> its class MUST go MISS. A class with NO intended detector is a routing BLIND SPOT (no-false-coverage), reported distinctly, never a silent pass. Repo-scope deterministic detectors (deps) fire on manifest CHANGE, not on the vuln code file -> REPO_SCOPE, excluded from the per-file rate. LLM routing via detect.select_detectors (kind+signal); per-file deterministic via registry.applicable_per_file (the PREVENT path).""" detect = _load("sg_detect", os.path.join(ROOT, "orchestrator", "detect.py")) mapper = _load("sg_mapper", os.path.join(ROOT, "orchestrator", "mapper.py")) registry = _load("sg_registry", os.path.join(ROOT, "prevent", "registry.py")) detectors, _ = registry.load(detectors_root) rows, misses, blind, repo_scope, xfail = [], [], [], [], [] for c in cells: cls, cell_dir = c.get("class"), c["_dir"] ship = _shipping_detector(cell_dir, detectors) if ship is not None: intended = [ship] # band-3 conformance cell -> its own detector else: # corpus cell -> the non-baseline LLM specialist(s) of the class intended = [d for d in detectors if d.get("id") != "baseline" and cls in d.get("covers", [])] if not intended: blind.append({"id": c["id"], "class": cls}) rows.append({"id": c["id"], "class": cls, "status": "BLIND_SPOT", "note": "no specialist/deterministic detector for class — baseline-floor only"}) continue intended_ids = {d.get("id") for d in intended} # route path: corpus cell -> real source path (file-routing); band-3 -> on-disk fixture rel path (globs). vuln_local = _cell_vuln_path(cell_dir, c.get("file")) if ship is not None and vuln_local: route_path = os.path.relpath(vuln_local, ROOT) else: route_path = c.get("file") or "vuln.ts" body = "" if vuln_local: body = open(vuln_local, encoding="utf-8", errors="replace").read() kind = mapper.kind_of(route_path, body) selected = set() # LLM routing (DETECT): select_detectors reads target_path for the signal regex -> point at the fixture. for d in (detect.select_detectors(vuln_local or route_path, kind, detectors) or []): selected.add(d.get("id")) # per-file deterministic routing (PREVENT): scope_globs on the route path, across triggers. for trig in ("pre-commit", "pre-edit"): for d in registry.applicable_per_file(detectors, route_path, trig): selected.add(d.get("id")) hit = sorted(intended_ids & selected) # repo-scope deterministic (deps): change-triggered on manifests, NOT per-file routed -> own category. all_repo = all(d.get("scope") == "repo" for d in intended) if hit: status = "ROUTED" elif all_repo: status = "REPO_SCOPE"; repo_scope.append({"id": c["id"], "class": cls, "intended": sorted(intended_ids)}) elif c.get("routing_xfail"): # a KNOWN, accepted routing gap (e.g. dead detector) — excluded from rate + non-fatal (no-false-coverage: # reported, never silently passed). Remove the cell's routing_xfail field to make it fatal again. status = "XFAIL"; xfail.append({"id": c["id"], "class": cls, "reason": c["routing_xfail"]}) else: status = "MISS"; misses.append({"id": c["id"], "class": cls, "intended": sorted(intended_ids)}) rows.append({"id": c["id"], "class": cls, "kind": kind, "intended": sorted(intended_ids), "selected_intended": hit, "status": status}) scored = [r for r in rows if r["status"] in ("ROUTED", "MISS")] recall = round(sum(r["status"] == "ROUTED" for r in scored) / max(1, len(scored)), 3) out = { "routing_recall": recall, "n_scored": len(scored), "misses": misses, "blind_spots": blind, "repo_scope": repo_scope, "xfail": xfail, "rows": rows, "note": "routing_recall = ROUTED/(ROUTED+MISS) over cells whose class HAS a per-file intended detector. " "XFAIL (cell-flagged routing_xfail — a KNOWN accepted gap e.g. dead detector) is excluded + non-fatal. " "BLIND_SPOT (class with NO specialist/deterministic detector — baseline-floor only) and REPO_SCOPE " "(deps: change-triggered on manifests, not per-file) are EXCLUDED from the rate and reported " "separately (no-false-coverage). baseline is EXCLUDED from the assertion: disable a specialist -> " "its class MUST go MISS, else the gate measures the floor not routing.", } print(json.dumps(out, indent=2)) return 1 if misses else 0 def _resolve_exec(solution, root): """Resolve a SolutionAdapter exec argv to absolute script paths under `root`.""" exec_cmd = list(solution.get("exec") or []) if not exec_cmd: return None out = list(exec_cmd) for i, arg in enumerate(out[1:], 1): if arg.startswith("-"): continue out[i] = arg if os.path.isabs(arg) else os.path.join(root, arg) script = out[1] if len(out) > 1 else None if not (script and os.path.isfile(script)): return None return out def _run_solution(exec_cmd, finding, file_content): """Run a SolutionAdapter solve.py; return parsed JSON (None when the adapter abstains).""" payload = json.dumps({"finding": finding, "file_content": file_content}) p = subprocess.run(exec_cmd, input=payload, capture_output=True, text=True) if p.returncode != 0: raise RuntimeError(p.stderr.strip() or f"solve exited {p.returncode}") s = (p.stdout or "").strip() if not s or s == "null": return None return json.loads(s) def _is_specific_located(result, expect_contains): """True iff the adapter emitted a concrete located-suggestion (not generic advice / abstain).""" if not result or result.get("rung") != "located-suggestion" or result.get("patch") is not None: return False sug = result.get("suggestion") or "" return bool(expect_contains and expect_contains in sug) def _solution_cell_artifacts(solution): cdir = os.path.join(solution.get("_dir", ""), "cells") if not os.path.isdir(cdir): return [] out = [] for path in sorted(glob.glob(os.path.join(cdir, "*.json"))): with open(path, encoding="utf-8") as f: art = json.load(f) art["_artifact"] = os.path.basename(path) out.append(art) return out def _corpus_fixture(cell_id, name): path = os.path.join(CORPUS, cell_id, name) if not os.path.isfile(path): return None with open(path, encoding="utf-8", errors="replace") as f: return f.read() def _nonmatching_class(solution, ledger_mod, fallback="S1"): """A class token guaranteed not to match this adapter (consume-only negative probe).""" covered = ledger_mod.solution_classes(solution) if fallback not in covered: return fallback for alt in ("S1", "S2", "S3", "S5", "S6", "S7", "S8", "S9", "S10", "S11", "S12", "S13"): if alt not in covered: return alt return "S0" def cmd_solutions(cells, solutions_root): """Solution-adapter conformance — the behavioral resolution gate (mirrors cmd_routing). For EVERY SolutionAdapter from registry.load(..., kind='solution'), run its solve.py over ITS OWN cells/ artifacts and assert: vuln → SPECIFIC located-suggestion (concrete replacement named; generic/abstain = MISS), safe → ABSTAIN (a suggestion = MISS), non-matching input → nothing. A corpus class that HAS a cell but NO SolutionAdapter is a SOLUTION_BLIND_SPOT — reported distinctly (no-false-coverage), excluded from the conformance rate, non-fatal. Any adapter conformance failure = MISS = sys.exit(1).""" registry = _load("sg_registry", os.path.join(ROOT, "prevent", "registry.py")) ledger = _load("sg_ledger", os.path.join(ROOT, "ledger.py")) solutions, skipped = registry.load(solutions_root, kind="solution") rows, misses, blind = [], [], [] classes_with_cells = sorted({c.get("class") for c in cells if c.get("class")}) covered_classes = set() for s in solutions: covered_classes |= ledger.solution_classes(s) for cls in classes_with_cells: if cls not in covered_classes: blind.append({"class": cls, "note": "corpus class has cell(s) but no SolutionAdapter — resolution none"}) rows.append({"class": cls, "status": "SOLUTION_BLIND_SPOT", "note": "no SolutionAdapter for class — excluded from conformance rate"}) for s in solutions: sid, exec_cmd = s.get("id"), _resolve_exec(s, solutions_root) artifacts = _solution_cell_artifacts(s) if not exec_cmd: misses.append({"adapter": sid, "reason": "exec script missing or not a file"}) rows.append({"adapter": sid, "status": "MISS", "note": "exec script missing"}) continue if not artifacts: misses.append({"adapter": sid, "reason": "no cells/ conformance artifact"}) rows.append({"adapter": sid, "status": "MISS", "note": "declares rung but ships no cells/ artifact"}) continue for art in artifacts: cell_id = art.get("cell") or art.get("id") or art.get("_artifact", "?") finding = art.get("finding") or {} expect = art.get("expect_suggestion_contains") or "" vuln = _corpus_fixture(cell_id, "vuln.ts") safe = _corpus_fixture(cell_id, "safe.ts") probes = [] if vuln is not None: probes.append(("vuln", finding, vuln, lambda r: _is_specific_located(r, expect), "vuln must emit located-suggestion naming concrete replacement")) if safe is not None: probes.append(("safe", finding, safe, lambda r: r is None, "safe must abstain (no suggestion on fixed file)")) nm = dict(finding) nm["class"] = _nonmatching_class(s, ledger) if vuln is not None: probes.append(("nonmatching", nm, vuln, lambda r: r is None, "non-matching class must abstain (consume, never re-detect)")) for probe, fin, content, ok, note in probes: row = {"adapter": sid, "cell": cell_id, "probe": probe, "class": art.get("class") or finding.get("class")} try: result = _run_solution(exec_cmd, fin, content) except RuntimeError as e: status, detail = "MISS", str(e) else: status, detail = ("CONFORMANT", None) if ok(result) else ("MISS", note) if status == "MISS": misses.append({"adapter": sid, "cell": cell_id, "probe": probe, "detail": detail or note}) row.update({"status": status, "detail": detail}) rows.append(row) scored = [r for r in rows if r["status"] in ("CONFORMANT", "MISS")] recall = round(sum(r["status"] == "CONFORMANT" for r in scored) / max(1, len(scored)), 3) out = { "solution_conformance": recall, "n_scored": len(scored), "misses": misses, "solution_blind_spots": blind, "rows": rows, "skipped_solutions": [{"path": p, "reason": r} for p, r in skipped], "note": "solution_conformance = CONFORMANT/(CONFORMANT+MISS) over every SolutionAdapter probe " "(vuln→specific located-suggestion, safe→abstain, non-matching→nothing). " "SOLUTION_BLIND_SPOT (class with corpus cell(s) but no SolutionAdapter) is EXCLUDED from the rate " "and reported separately (no-false-coverage). Any adapter/probe MISS is fatal (exit 1).", } print(json.dumps(out, indent=2)) return 1 if misses else 0 def main(): ap = argparse.ArgumentParser() ap.add_argument("--inventory", action="store_true") ap.add_argument("--findings") ap.add_argument("--require", action="append", default=[]) ap.add_argument("--mapper", help="REPO_ROOT: score enumerated/actual surface-recall on a real tree") ap.add_argument("--routing", action="store_true", help="class-routing recall: assert each cell's INTENDED detector is selected (narrowing gate, §10.7)") ap.add_argument("--solutions", action="store_true", help="solution-adapter conformance: run each SolutionAdapter over its cells/ (resolution gate)") ap.add_argument("--detectors-root", default=ROOT, help="detector manifests root (default: repo root)") ap.add_argument("--solutions-root", default=ROOT, help="SolutionAdapter manifests root (default: repo root)") a = ap.parse_args() cells = load_cells() if a.inventory: sys.exit(cmd_inventory(cells)) if a.findings: sys.exit(cmd_findings(cells, a.findings, set(a.require))) if a.mapper: sys.exit(cmd_mapper(a.mapper, cells)) if a.routing: sys.exit(cmd_routing(cells, a.detectors_root)) if a.solutions: sys.exit(cmd_solutions(cells, a.solutions_root)) ap.error("need --inventory, --findings, --mapper, --routing, or --solutions") if __name__ == "__main__": main()