150 lines
7.4 KiB
Python
150 lines
7.4 KiB
Python
"""Score the pilot against frozen blind labels using the pre-registered decision.
|
|
|
|
python experiments/P0-pilot/score.py --results experiments/P0-pilot/data/results.jsonl \
|
|
--labels experiments/P0-pilot/labels/labels.json --key experiments/P0-pilot/labels/key.json --summary experiments/P0-pilot/results/summary.json
|
|
|
|
Positive class = unfaithful. Primary: wrong versus faithful, excluding minor drift (role alignment and
|
|
the judge do not need Lean to accept the code; only 10 of the 65 pilot items compiled, too few to
|
|
decide on). Secondary: wrong or minor_drift; and the compiled-only subset (the paper's claim, reported
|
|
with its interval, not used to decide). "not_formalizable" items are excluded. Decision (fixed before
|
|
looking at labels):
|
|
|
|
GO role recall on wrong items >= 0.5 AND false-flag rate on faithful items <= 0.2
|
|
AND role recall >= judge recall + 0.10 at a false-flag rate no higher than the judge's
|
|
PARTIAL role recall >= 0.4 with false-flag <= 0.3, or role complements the judge
|
|
(flags >= 3 wrong items the judge misses)
|
|
NO-GO otherwise (re-scope to evaluation + taxonomy + resource)
|
|
With ~65 items every number carries a wide Wilson interval; the pilot checks for signal, not effect size.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import math
|
|
from pathlib import Path
|
|
|
|
from leanbench.evalstats import auc, confusion, wilson
|
|
|
|
|
|
def load(args):
|
|
result_rows = [json.loads(x) for x in args.results.read_text(encoding="utf-8").splitlines() if x.strip()]
|
|
results = {row["id"]: row for row in result_rows}
|
|
labels = json.loads(args.labels.read_text(encoding="utf-8"))["labels"]
|
|
key = json.loads(args.key.read_text(encoding="utf-8"))["key"]
|
|
if not key or len(set(key.values())) != len(key):
|
|
raise ValueError("The blind key must contain unique item IDs")
|
|
if len(results) != len(result_rows) or set(results) != set(key.values()):
|
|
raise ValueError("Results must cover every keyed item exactly once")
|
|
if set(labels) != set(key):
|
|
raise ValueError("Labels must cover every blind ID exactly once")
|
|
rows = []
|
|
for blind, item_id in key.items():
|
|
lab = labels.get(blind) or {}
|
|
if lab.get("verdict") not in {"faithful", "minor_drift", "wrong", "not_formalizable"}:
|
|
raise ValueError(f"Missing or invalid verdict for {blind}")
|
|
row = results[item_id]
|
|
if row.get("errors") or "judge" not in row or "role" not in row:
|
|
raise ValueError(f"Unresolved detector failure for {blind}; no decision can be issued")
|
|
rows.append({**row, "blind": blind, "verdict": lab["verdict"],
|
|
"human_errors": lab.get("errors", [])})
|
|
return rows
|
|
|
|
|
|
def flags(row):
|
|
return {
|
|
"compile_only": not row["compiled"],
|
|
"lint": bool(row.get("lint")),
|
|
"role": bool(row.get("role", {}).get("flagged")),
|
|
"judge": (not row["judge"]["faithful"]) if "judge" in row else False,
|
|
}
|
|
|
|
|
|
def report(rows, positive):
|
|
out = {}
|
|
for name in ("compile_only", "lint", "role", "judge"):
|
|
out[name] = confusion([flags(r)[name] for r in rows], [positive(r) for r in rows]).as_dict()
|
|
out[name]["precision_ci"] = wilson(out[name]["tp"], out[name]["tp"] + out[name]["fp"])
|
|
out["role"]["auc_score"] = auc([-(r.get("role", {}).get("score", 1.0)) for r in rows],
|
|
[positive(r) for r in rows])
|
|
out["role"]["complementary_wrong"] = sum(
|
|
positive(r) and flags(r)["role"] and not flags(r)["judge"] for r in rows
|
|
)
|
|
return out
|
|
|
|
|
|
def decide(sub):
|
|
role, judge = sub["role"], sub["judge"]
|
|
if role["tp"] + role["fn"] == 0 or role["fp"] + role["tn"] == 0:
|
|
return "INSUFFICIENT: both wrong and faithful items are required"
|
|
go = (role["recall"] >= 0.5 and role["false_flag_rate"] <= 0.2
|
|
and role["recall"] >= judge["recall"] + 0.10
|
|
and role["false_flag_rate"] <= judge["false_flag_rate"])
|
|
partial = ((role["recall"] >= 0.4 and role["false_flag_rate"] <= 0.3)
|
|
or role.get("complementary_wrong", 0) >= 3)
|
|
return "GO" if go else "PARTIAL" if partial else "NO-GO"
|
|
|
|
|
|
def main() -> int:
|
|
ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
|
|
ap.add_argument("--results", required=True, type=Path)
|
|
ap.add_argument("--labels", required=True, type=Path)
|
|
ap.add_argument("--key", required=True, type=Path)
|
|
ap.add_argument("--summary", type=Path, default=None,
|
|
help="also write the (text-free) summary here, e.g. results/summary.json")
|
|
args = ap.parse_args()
|
|
meta = json.loads(args.results.with_suffix(".meta.json").read_text(encoding="utf-8"))
|
|
if (meta.get("status") != "complete"
|
|
or meta.get("labels_sha256") != hashlib.sha256(args.labels.read_bytes()).hexdigest()
|
|
or meta.get("results_sha256") != hashlib.sha256(args.results.read_bytes()).hexdigest()):
|
|
ap.error("Scoring requires a complete run with unchanged frozen labels and results")
|
|
all_rows = load(args)
|
|
rows = [r for r in all_rows if r["verdict"] != "not_formalizable"]
|
|
primary = [r for r in rows if r["verdict"] != "minor_drift"]
|
|
compiled = [r for r in primary if r["compiled"]]
|
|
summary = {
|
|
"run_metadata": meta,
|
|
"n_labeled": len(all_rows), "n_evaluable": len(rows), "n_primary": len(primary),
|
|
"n_compiled": len(compiled),
|
|
"verdicts": {v: sum(r["verdict"] == v for r in rows) for v in
|
|
("faithful", "minor_drift", "wrong", "not_formalizable")},
|
|
"compiled_but_wrong": sum(r["verdict"] == "wrong" for r in compiled),
|
|
"primary_all": report(primary, lambda r: r["verdict"] == "wrong"),
|
|
"secondary_all": report(rows, lambda r: r["verdict"] in ("wrong", "minor_drift")),
|
|
"compiled_only": report(compiled, lambda r: r["verdict"] == "wrong"),
|
|
}
|
|
summary["verdicts"]["not_formalizable"] = len(all_rows) - len(rows)
|
|
summary["role_recall_by_error"] = {
|
|
error: {"flagged": sum(flags(r)["role"] for r in rows
|
|
if r["verdict"] == "wrong" and error in r["human_errors"]),
|
|
"total": sum(r["verdict"] == "wrong" and error in r["human_errors"] for r in rows)}
|
|
for error in sorted({e for r in rows for e in r["human_errors"]})
|
|
}
|
|
summary["degenerate_projection"] = {
|
|
"total": sum(bool(r.get("projection", {}).get("degenerate")) for r in all_rows),
|
|
"wrong": sum(r["verdict"] == "wrong" and bool(r.get("projection", {}).get("degenerate"))
|
|
for r in all_rows),
|
|
}
|
|
summary["disagreements"] = [
|
|
{"blind": r["blind"], "verdict": r["verdict"], "flags": flags(r)}
|
|
for r in primary if any(f != (r["verdict"] == "wrong") for f in flags(r).values())
|
|
]
|
|
summary["decision"] = decide(summary["primary_all"])
|
|
def finite_json(value):
|
|
if isinstance(value, dict):
|
|
return {k: finite_json(v) for k, v in value.items()}
|
|
if isinstance(value, (list, tuple)):
|
|
return [finite_json(v) for v in value]
|
|
return None if isinstance(value, float) and not math.isfinite(value) else value
|
|
|
|
text = json.dumps(finite_json(summary), indent=2, allow_nan=False)
|
|
if args.summary:
|
|
args.summary.parent.mkdir(parents=True, exist_ok=True)
|
|
args.summary.write_text(text + "\n", encoding="utf-8")
|
|
print(text)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|