#!/usr/bin/env python3 """recall_gate.py — the measured-recall RATE gate (item #5). Standalone DETERMINISTIC gate (NOT bench.py) over the measured-recall SoT `domains/security/recall/records.json`. It makes the project's load-bearing LLM-validation doctrine a machine check: "LLM bands are measured STATISTICALLY: k>=3 rolls, recall as a RATE. NEVER a single-run claim." (CLAUDE.md / spec §2,§6) and binds every stored rate to the EXACT detector prompt it was rolled against, so any prompt edit auto-invalidates the claim. That binding is the no-false-clean DUAL for measured recall: a frozen `catches/k` carrying only a date is an assumption wearing a measurement's clothes — once the prompt it measured no longer exists, the rate is no longer valid and the gate must say so, not certify it. (It also operationalizes the #14/#15 apply-gates: applying either candidate edits the prompt -> every bound record flips STALE -> this gate reds until the rolls are re-run.) HARD gates (exit 1): - MALFORMED — missing field, catches>k, catches<0, k<1, unresolvable detector/prompt - UNKNOWN_CELL — record names a cell absent from the corpus (no-false-coverage) - SINGLE_RUN — k < 3 (an under-rolled LLM claim is forbidden) - STALE — measured_against != sha256(current detector prompt): the rolls were against a prompt that no longer exists; rate invalid REPORT-ONLY (exit 0 unless --strict-floor): - BELOW_FLOOR — rate < floor on a record that is NOT waived. The project ships some cells on no-regression BELOW the bar floor (e.g. S2-role 3/9, a documented pre-existing flaky cell); a hard floor would wrongly red a deliberately-shipped cell, so the floor is advisory and per-cell waivable (a `waiver` field, mirroring the ratchet). - UNMEASURED — corpus cells with NO record (honestly unmeasured-at-rate; most cells are n=1 point estimates, not a failure) Derived output (--json): the RATE-VALIDATED set = every record that is well-formed, known-cell, k>=3 and hash-current (floor-independent — "statistically measured", which is what validation means). This is the machine replacement for the prose-curated `bench.py:72` `rate_validated:["S2","S3"]`. Reconciling it into bench.py / ledger.py is a DOWNSTREAM cursor task (those files are not edited here, per the work split); this gate derives the set independently. """ import argparse import hashlib import json import os import sys ROOT = os.path.dirname(os.path.abspath(__file__)) DEF_RECORDS = os.path.join(ROOT, "domains", "security", "recall", "records.json") DEF_CORPUS = os.path.join(ROOT, "domains", "security", "corpus") DEF_DETECTORS = os.path.join(ROOT, "domains", "security", "detectors") DEF_FLOOR = 0.66 # the documented 2/3 bar floor (2/3 = 0.667 passes; below = reported) REQUIRED = ("cell_id", "detector", "catches", "k", "measured_against", "metric") HARD = {"MALFORMED", "UNKNOWN_CELL", "SINGLE_RUN", "STALE"} # `metric` is REQUIRED, not cosmetic: the same cell can read 9/9 under per-roll-union and 1/9 under # max-single-entry (citation doc, S3-report-schedule). An UNLABELED rate is the soft false-coverage this # gate exists to prevent — a bare 1.000 next to CLAUDE.md's 86% (max-single) headline is a flat # contradiction. So a record with no `metric` is MALFORMED, and the label rides next to every rate. def sha256_file(path): h = hashlib.sha256() with open(path, "rb") as fh: for chunk in iter(lambda: fh.read(65536), b""): h.update(chunk) return h.hexdigest() def load_corpus(corpus_root): """cell_id -> {class} for every cell dir carrying a canonical.json.""" cells = {} if not os.path.isdir(corpus_root): return cells for name in os.listdir(corpus_root): cj = os.path.join(corpus_root, name, "canonical.json") if not os.path.isfile(cj): continue with open(cj, encoding="utf-8") as fh: c = json.load(fh) cells[c.get("id", name)] = {"class": c.get("class", "")} return cells def prompt_hash(detectors_root, detector, _cache): """sha256 of the detector's CURRENT prompt, or None if unresolvable.""" if detector in _cache: return _cache[detector] det_dir = os.path.join(detectors_root, detector) manifest = os.path.join(det_dir, "detector.json") digest = None if os.path.isfile(manifest): try: with open(manifest, encoding="utf-8") as fh: pf = json.load(fh).get("prompt") if pf: p = os.path.join(det_dir, pf) if os.path.isfile(p): digest = sha256_file(p) except (json.JSONDecodeError, OSError): digest = None _cache[detector] = digest return digest def evaluate_record(rec, corpus, detectors_root, floor, _cache): """One record -> verdict dict {cell_id, status, rate, k, detail, waived}.""" def v(status, detail, rate=None, k=None, waived=False): return {"cell_id": rec.get("cell_id", ""), "detector": rec.get("detector"), "status": status, "rate": rate, "k": k, "metric": rec.get("metric"), "detail": detail, "waived": waived} for f in REQUIRED: if f not in rec: return v("MALFORMED", f"missing required field '{f}'") cell_id, detector = rec["cell_id"], rec["detector"] catches, k = rec["catches"], rec["k"] if not isinstance(catches, int) or not isinstance(k, int): return v("MALFORMED", "catches and k must be integers") if k < 1 or catches < 0 or catches > k: return v("MALFORMED", f"catches/k out of range ({catches}/{k})") cur = prompt_hash(detectors_root, detector, _cache) if cur is None: return v("MALFORMED", f"detector '{detector}' prompt unresolvable", k=k) if cell_id not in corpus: return v("UNKNOWN_CELL", "record names a cell absent from the corpus", k=k) rate = catches / k if k < 3: return v("SINGLE_RUN", f"k={k} < 3 — under-rolled, not a rate", rate, k) if rec["measured_against"] != cur: return v("STALE", "measured_against != current prompt hash — re-roll required", rate, k) waiver = rec.get("waiver") or {} waived = bool(waiver.get("below_floor_ok")) if rate < floor: if waived: return v("OK_WAIVED", f"rate {rate:.3f} < floor {floor} but waived: " f"{waiver.get('reason', '')}", rate, k, waived=True) return v("BELOW_FLOOR", f"rate {rate:.3f} < floor {floor} (report-only)", rate, k) return v("OK", f"rate {rate:.3f} >= floor {floor}", rate, k) def run(records_path, corpus_root, detectors_root, floor, strict_floor): """-> (exit_code, result). Pure: no printing, so tests can assert on it.""" corpus = load_corpus(corpus_root) if not os.path.isfile(records_path): return 1, {"error": f"records file not found: {records_path}", "records": [], "rate_validated_cells": [], "rate_validated_classes": [], "below_floor": [], "unmeasured_cells": sorted(corpus)} with open(records_path, encoding="utf-8") as fh: data = json.load(fh) records = data.get("records", data) if isinstance(data, dict) else data cache = {} verdicts = [evaluate_record(r, corpus, detectors_root, floor, cache) for r in records] recorded = {v["cell_id"] for v in verdicts} hard = [v for v in verdicts if v["status"] in HARD] below = [v for v in verdicts if v["status"] == "BELOW_FLOOR"] validated = [v for v in verdicts if v["status"] not in HARD] # k>=3, hash-current, well-formed validated_classes = sorted({corpus.get(v["cell_id"], {}).get("class", "") for v in validated} - {""}) unmeasured = sorted(c for c in corpus if c not in recorded) # rich per-cell detail for downstream (the bench:72 reconcile needs the METRIC, not just the id — # a per-roll-union 1.000 and a max-single 0.11 are the SAME cell; reconciling values without the # metric label silently merges incompatible quantities). validated_detail = [{"cell_id": v["cell_id"], "rate": v["rate"], "k": v["k"], "metric": v["metric"], "status": v["status"], "waived": v["waived"]} for v in validated] exit_code = 1 if hard or (strict_floor and below) else 0 result = { "records": verdicts, "hard_failures": hard, "below_floor": below, "rate_validated_cells": sorted(v["cell_id"] for v in validated), "rate_validated": validated_detail, "rate_validated_classes": validated_classes, "unmeasured_cells": unmeasured, } return exit_code, result def main(argv=None): ap = argparse.ArgumentParser(description="Measured-recall RATE gate (#5) — k>=3 + prompt-hash binding.") ap.add_argument("--records", default=DEF_RECORDS) ap.add_argument("--corpus-root", default=DEF_CORPUS) ap.add_argument("--detectors-root", default=DEF_DETECTORS) ap.add_argument("--floor", type=float, default=DEF_FLOOR, help="advisory rate floor (default 0.66 = the 2/3 bar floor)") ap.add_argument("--strict-floor", action="store_true", help="make BELOW_FLOOR (unwaived) a HARD exit-1 failure too") ap.add_argument("--json", action="store_true", help="emit the derived rate-validated set as JSON") args = ap.parse_args(argv) code, res = run(args.records, args.corpus_root, args.detectors_root, args.floor, args.strict_floor) if args.json: print(json.dumps({ "rate_validated": res["rate_validated"], # carries the metric label per cell "rate_validated_classes": res["rate_validated_classes"], "below_floor": [v["cell_id"] for v in res["below_floor"]], "unmeasured_cells": res["unmeasured_cells"], }, indent=2)) return code if "error" in res: print(f"recall-gate: {res['error']}") return code print("── recall-rate gate (#5): k>=3 + prompt-hash binding ──") for v in res["records"]: mark = "FAIL" if v["status"] in HARD else ("warn" if v["status"] == "BELOW_FLOOR" else "ok") r = f"{v['rate']:.3f}" if v["rate"] is not None else " - " metric = v.get("metric") or "-" print(f" [{mark:>4}] {v['status']:<11} {v['cell_id']:<38} rate={r} k={v['k']} [{metric}] {v['detail']}") print(f"\n rate-validated cells : {len(res['rate_validated_cells'])} " f"(classes: {', '.join(res['rate_validated_classes']) or '—'})") if res["below_floor"]: print(f" below-floor (report) : {', '.join(v['cell_id'] for v in res['below_floor'])}") if res["unmeasured_cells"]: print(f" UNMEASURED-at-rate : {len(res['unmeasured_cells'])} cells " f"(no-false-coverage — n=1, not validated): {', '.join(res['unmeasured_cells'])}") if res["hard_failures"]: print("\nrecall-gate: HARD FAILURE — single-run / stale / malformed / unknown-cell record(s):") for v in res["hard_failures"]: print(f" - {v['cell_id']}: {v['status']} — {v['detail']}") print("Fix the record or RE-ROLL k>=3 against the current prompt.") else: print("\nrecall-gate: all recorded rates are k>=3 and bound to the current prompt.") return code if __name__ == "__main__": sys.exit(main())