spec(pre-push-safety-net): pre-register evaluation protocol v2 and its runner

Committed before any v2 case is scanned. It has two changes from v1, both chosen by
the maintainer from lessons of the v1 run:
- declared sites: a finding at a `path::function` the advisory names is credited,
  as well as one on the fix's changed lines;
- two runs of the vulnerable and fixed pins: detection needs the finding in both
  runs, a fixed-side alert in either counts, and the difference between runs is
  reported as stability.

evaluate.py takes --population. v2 scans six pins per case and scores with
matches_v2 / score_v2. v1's scoring is unchanged.

Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
Claude-Session: https://claude.ai/code/session_01EmM1j46qgKMh3zqD1QMwEo

benchmarks/independent/evaluate.py

@@ -8,9 +8,12 @@ question is an instrument failure and is retried once.
 `score` applies protocol-v1.md exactly -- matching, detection, fixed-side and benign false alerts, unanswered --
 and draws the precision sample (every third pooled vulnerable-pin finding, or all of them if 15 or fewer).
 
+Protocol v2 (protocol-v2.md, population-v2.toml) scans the vulnerable and fixed pins twice each and credits a
+finding at a declared site as well; `--population population-v2.toml` selects it for both commands.
+
 Usage:
-    python benchmarks/independent/evaluate.py run --source <frozen analyzer tree> [--only ID,...]
-    python benchmarks/independent/evaluate.py score
+    python benchmarks/independent/evaluate.py run --source <frozen analyzer tree> [--only ID,...] [--population F]
+    python benchmarks/independent/evaluate.py score [--population F]
 """
 
 from __future__ import annotations
@@ -37,7 +40,28 @@ def cases() -> list[dict]:
     return tomllib.loads(POPULATION.read_text())["case"]
 
 
+def version() -> str:
+    return POPULATION.stem.removeprefix("population-")
+
+
+def use(population: Path) -> None:
+    """Point the module at a population file and its own results directory."""
+    global POPULATION, RESULTS
+    POPULATION = population if population.is_absolute() else HERE / population.name
+    RESULTS = Path.home() / "ousast-results" / f"independent-{version()}"
+
+
 def pins(case: dict) -> dict[str, str]:
+    if version() != "v1":
+        # Protocol v2: the vulnerable and fixed pins twice each, the benign pair once.
+        return {
+            "vulnerable_a": case["vulnerable"],
+            "vulnerable_b": case["vulnerable"],
+            "fixed_a": case["fixed"],
+            "fixed_b": case["fixed"],
+            "benign_base": case["benign"]["base"],
+            "benign_tip": case["benign"]["tip"],
+        }
     return {
         "vulnerable": case["vulnerable"],
         "fixed": case["fixed"],
@@ -207,15 +231,109 @@ def score(args: argparse.Namespace) -> int:
     return 0
 
 
+def matches_v2(finding: dict, case: dict, ranges: dict[str, list[tuple[int, int]]]) -> bool:
+    """protocol-v2.md: v1's rule with the record's declared site functions, or a declared `path::function`."""
+    path, _, rest = finding["site"].partition(":")
+    line_text, _, function = rest.partition(":")
+    if finding.get("family") != case["family"]:
+        return False
+    sites = set(case.get("sites", []))
+    if f"{path}::{function}" in sites:
+        return True
+    if path not in ranges:
+        return False
+    if function in {site.partition("::")[2] for site in sites}:
+        return True
+    line = int(line_text) if line_text.isdigit() else -1
+    return any(first - LINE_WINDOW <= line <= last + LINE_WINDOW for first, last in ranges[path])
+
+
+def stability(one: list[dict], two: list[dict]) -> tuple[int, int]:
+    """(findings in exactly one run, findings in either)."""
+    a, b = {f["site"] for f in one}, {f["site"] for f in two}
+    return len(a ^ b), len(a | b)
+
+
+def score_v2() -> int:
+    per_case, pool, unstable_pool = [], [], 0
+    changing, total_questions, total_completed = [0, 0], 0, 0
+    for case in cases():
+        path = lambda label: RESULTS / "scans" / f"{case['id']}--{label}.json"  # noqa: E731
+        scans = {label: json.loads(path(label).read_text()) for label in pins(case) if path(label).is_file()}
+        if len(scans) < 6:
+            per_case.append({"id": case["id"], "status": "incomplete", "scans": sorted(scans)})
+            continue
+        old, new = hunks(case, "old"), hunks(case, "new")
+        va, vb = scans["vulnerable_a"].get("findings", []), scans["vulnerable_b"].get("findings", [])
+        fa, fb = scans["fixed_a"].get("findings", []), scans["fixed_b"].get("findings", [])
+        hit_a = {f["site"] for f in va if matches_v2(f, case, old)}
+        hit_b = {f["site"] for f in vb if matches_v2(f, case, old)}
+        fixed_alerts = sorted({f["site"] for f in fa + fb if matches_v2(f, case, new)})
+        base_regions = {tuple(f["site"].split(":")[0::2]) for f in scans["benign_base"].get("findings", [])}
+        benign_alerts = [f["site"] for f in scans["benign_tip"].get("findings", []) if tuple(f["site"].split(":")[0::2]) not in base_regions]
+        site_files = {site.partition("::")[0] for site in case.get("sites", [])}
+        covered = set(old) | site_files
+        answered = any(r.split(":")[0] in covered for label in ("vulnerable_a", "vulnerable_b") for r in scans[label].get("completed_regions", []))
+        diff_v, union_v = stability(va, vb)
+        diff_f, union_f = stability(fa, fb)
+        changing[0] += diff_v + diff_f
+        changing[1] += union_v + union_f
+        for scan in scans.values():
+            total_questions += int(scan.get("questions") or 0)
+            total_completed += int(scan.get("completed") or 0)
+        detected = sorted(hit_a & hit_b)
+        per_case.append(
+            {
+                "id": case["id"], "family": case["family"], "language": case["language"],
+                "detected": bool(detected), "detected_sites": detected,
+                "unstable": bool(hit_a ^ hit_b) and not detected, "unstable_sites": sorted(hit_a ^ hit_b),
+                "unanswered": not (hit_a | hit_b) and not answered,
+                "instrument_errors": {label: s["instrument_error"] for label, s in scans.items() if "instrument_error" in s},
+                "fixed_side_alerts": fixed_alerts, "benign_alerts": benign_alerts,
+                "stability": {"vulnerable": [diff_v, union_v], "fixed": [diff_f, union_f]},
+                "completion": {label: [s.get("completed"), s.get("questions")] for label, s in scans.items()},
+                "seconds": {label: s.get("seconds") for label, s in scans.items()},
+            }
+        )  # fmt: skip
+        stable = {f["site"] for f in vb}
+        pool += [{"case": case["id"], **f} for f in va if f["site"] in stable]
+        unstable_pool += len({f["site"] for f in va} ^ stable)
+    pool.sort(key=lambda f: (f["case"], f["site"]))
+    sample = pool if len(pool) <= 15 else pool[0::3]
+    scored = [c for c in per_case if c.get("status") != "incomplete"]
+    detected = sum(c["detected"] for c in scored)
+    result = {
+        "protocol": "protocol-v2.md",
+        "cases_scored": len(scored), "cases_incomplete": [c["id"] for c in per_case if c.get("status") == "incomplete"],
+        "recall": [detected, len(scored)], "recall_wilson95": wilson(detected, len(scored)),
+        "unstable_detections": [c["id"] for c in scored if c["unstable"]],
+        "fixed_side_alerts": sum(len(c["fixed_side_alerts"]) for c in scored),
+        "benign_alerts": sum(len(c["benign_alerts"]) for c in scored),
+        "completion": [total_completed, total_questions],
+        "stability": changing,
+        "pooled_stable_vulnerable_findings": len(pool), "unstable_vulnerable_findings": unstable_pool,
+        "precision_sample_size": len(sample),
+        "cases": per_case, "precision_sample": sample,
+    }  # fmt: skip
+    (RESULTS / "score.json").write_text(json.dumps(result, indent=2) + "\n")
+    print(json.dumps({k: v for k, v in result.items() if k not in ("cases", "precision_sample")}, indent=1))
+    return 0
+
+
 def main() -> int:
     parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
     sub = parser.add_subparsers(dest="command", required=True)
     runner = sub.add_parser("run")
     runner.add_argument("--source", type=Path, required=True, help="a frozen export of the analyzer (freeze_source.sh)")
     runner.add_argument("--only", default="")
-    sub.add_parser("score")
+    runner.add_argument("--population", type=Path, default=POPULATION)
+    scorer = sub.add_parser("score")
+    scorer.add_argument("--population", type=Path, default=POPULATION)
     args = parser.parse_args()
-    return run(args) if args.command == "run" else score(args)
+    use(args.population)
+    if args.command == "run":
+        return run(args)
+    return score(args) if version() == "v1" else score_v2()
 
 
 if __name__ == "__main__":