"""Validate and replay a current-chat assistant review without calling any model. Private source schemas and reasons stay in data/. Public summaries contain only categories, counts, numeric scores, hashes and an explicit non-independent provenance. """ from __future__ import annotations import argparse import hashlib import html import json import math from collections import Counter from pathlib import Path from leanbench.align import DEFAULT_THRESHOLD, GOAL_THRESHOLD, align from leanbench.evalstats import auc, confusion, wilson from leanbench.lint import lint from leanbench.project import project from leanbench.srcsimplex import parse_source_simplex VERDICTS = {"faithful", "minor_drift", "wrong", "not_formalizable"} ERRORS = {"missing_hypothesis", "extra_hypothesis", "wrong_type", "quantifier", "wrong_conclusion", "degenerate", "hallucinated_api", "other"} def digest(path: Path) -> str: return hashlib.sha256(path.read_bytes()).hexdigest() def finite(value): if isinstance(value, dict): return {k: finite(v) for k, v in value.items()} if isinstance(value, (list, tuple)): return [finite(v) for v in value] if isinstance(value, float) and not math.isfinite(value): return None return value def unique(rows, field): indexed = {row[field]: row for row in rows} if len(indexed) != len(rows): raise ValueError(f"Duplicate {field}") return indexed def evaluate(items, key, assessment): """Keep model-authored assessments separate from deterministic detector outputs.""" by_id = unique(items, "id") records = unique(assessment["records"], "blind_id") if (not key or len(set(key.values())) != len(key) or set(records) != set(key) or set(by_id) != set(key.values())): raise ValueError("Items, key and assessments must cover the same cohort exactly once") output = [] for blind, item_id in key.items(): item, record = by_id[item_id], records[blind] if record["id"] != item_id or not item.get("source_text", "").strip(): raise ValueError(f"Missing source text or mismatched item for {blind}") judge = record["judge"] if (judge.get("verdict") not in VERDICTS or type(judge.get("faithful")) is not bool or judge["faithful"] != (judge["verdict"] == "faithful") or not isinstance(judge.get("reason"), str) or not judge["reason"].strip() or not isinstance(judge.get("errors"), list) or not set(judge["errors"]) <= ERRORS): raise ValueError(f"Invalid authored judgment for {blind}") source = parse_source_simplex(record["source"]) projection = project(item["lean_code"]) alignment = align(source, projection) output.append({"id": item_id, "blind_id": blind, "source": source.as_dict(), "judge": judge, "projection": projection.as_dict(), "role": alignment.as_dict(), "lint": lint(projection), "compiled": bool(item.get("verified")), "errors": {}}) return output def public_items(rows, labels): if set(labels) != {row["blind_id"] for row in rows}: raise ValueError("Reference labels must cover the entire cohort") out = [] for row in rows: reference = labels[row["blind_id"]]["verdict"] if reference not in VERDICTS: raise ValueError("Invalid reference verdict") out.append({"blind_id": row["blind_id"], "reference_verdict": reference, "assistant_verdict": row["judge"]["verdict"], "assistant_error_types": row["judge"]["errors"], "verdict_agrees": reference == row["judge"]["verdict"], "compiled": row["compiled"], "lint_flagged": bool(row["lint"]), "role_flagged": row["role"]["flagged"], "role_score": row["role"]["score"], "role_conflict_types": row["role"]["errors"], "source_role_count": len(row["source"]["roles"]), "lean_role_count": len(row["projection"]["roles"]), "degenerate_projection": row["projection"]["degenerate"]}) return out def agreement(rows, labels, secondary=False): cohort = [r for r in rows if labels[r["blind_id"]]["verdict"] != "not_formalizable" and (secondary or labels[r["blind_id"]]["verdict"] != "minor_drift")] positives = {"wrong", "minor_drift"} if secondary else {"wrong"} target = [labels[r["blind_id"]]["verdict"] in positives for r in cohort] out = {} for name, flags in { "role": [r["role"]["flagged"] for r in cohort], "assistant_judge": [not r["judge"]["faithful"] for r in cohort], "lint": [bool(r["lint"]) for r in cohort], "compile_only": [not r["compiled"] for r in cohort], }.items(): counts = confusion(flags, target) out[name] = {**counts.as_dict(), "precision_ci": wilson(counts.tp, counts.tp + counts.fp)} out["role"]["auc"] = auc([-r["role"]["score"] for r in cohort], target) return {"n": len(cohort), "metrics_are_non_independent_reference_agreement": True, "detectors": out} def write_review(path, rows, items): by_id = {r["id"]: r for r in items} sections = [] for row in rows: item = by_id[row["id"]] sections.append( f'

{row["blind_id"]}

' f'

Passage

{html.escape(item["source_text"])}
' f'

Lean

{html.escape(item["lean_code"])}
' f'

Fresh assistant judgment: {row["judge"]["verdict"]}

' f'

{html.escape(row["judge"]["reason"])}

' f'

Source roles

{html.escape(json.dumps(row["source"], indent=2, ensure_ascii=False))}
' f'

Deterministic alignment

{html.escape(json.dumps(row["role"], indent=2, ensure_ascii=False))}
' ) path.write_text('' 'P0 GPT-assisted diagnostic' '

P0: 65-item GPT-assisted diagnostic

' '

Private review. Same-assistant, non-blind diagnosis; no independent accuracy or gate decision. ' 'Passages and model-derived descriptions must remain outside Git.

' + ''.join(sections) + '', encoding="utf-8") def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--items", type=Path, required=True) parser.add_argument("--key", type=Path, required=True) parser.add_argument("--assessments", type=Path, required=True) parser.add_argument("--labels", type=Path, required=True) parser.add_argument("--private-out", type=Path, required=True) parser.add_argument("--summary", type=Path, required=True) parser.add_argument("--public-items", type=Path, required=True) args = parser.parse_args() if "data" not in args.private_out.parts: parser.error("Private output must be under the gitignored data/ directory") items = [json.loads(x) for x in args.items.read_text().splitlines() if x.strip()] key = json.loads(args.key.read_text())["key"] assessment = json.loads(args.assessments.read_text()) provenance = assessment["provenance"] if (provenance["items_sha256"] != digest(args.items) or provenance["key_sha256"] != digest(args.key)): parser.error("Authored review is bound to different inputs or blind key") rows = evaluate(items, key, assessment) labels = json.loads(args.labels.read_text())["labels"] public = public_items(rows, labels) summary = { "experiment": "P0-GPT-diagnostic", "status": "complete", "n": len(rows), "provenance": provenance, "decision": None, "decision_withheld_reason": "Same-assistant, non-blind diagnostic; not the preregistered Qwen pilot or independent ground-truth validation.", "hashes": {"items": digest(args.items), "key": digest(args.key), "assessments": digest(args.assessments), "reference_labels": digest(args.labels), "replay_script": digest(Path(__file__)), **{f"leanbench/{name}.py": digest(Path(__file__).resolve().parents[2] / "leanbench" / f"{name}.py") for name in ("align", "project", "srcsimplex", "lint")}}, "fixed_thresholds": {"matching": DEFAULT_THRESHOLD, "goal": GOAL_THRESHOLD}, "verdict_counts": dict(Counter(r["judge"]["verdict"] for r in rows)), "role_flagged": sum(r["role"]["flagged"] for r in rows), "lint_flagged": sum(bool(r["lint"]) for r in rows), "compiled": sum(r["compiled"] for r in rows), "degenerate_projection": sum(r["projection"]["degenerate"] for r in rows), "exact_verdict_agreement": sum(r["verdict_agrees"] for r in public), "changed_verdicts": [r["blind_id"] for r in public if not r["verdict_agrees"]], "primary_reference_agreement": agreement(rows, labels), "secondary_reference_agreement": agreement(rows, labels, secondary=True), "role_flagged_by_new_verdict": {v: {"n": sum(r["judge"]["verdict"] == v for r in rows), "flagged": sum(r["judge"]["verdict"] == v and r["role"]["flagged"] for r in rows)} for v in sorted(VERDICTS)}, } args.private_out.mkdir(parents=True, exist_ok=True) (args.private_out / "results.jsonl").write_text(''.join(json.dumps(r, ensure_ascii=False) + '\n' for r in rows)) write_review(args.private_out / "review.html", rows, items) args.summary.parent.mkdir(parents=True, exist_ok=True) args.public_items.parent.mkdir(parents=True, exist_ok=True) args.summary.write_text(json.dumps(finite(summary), indent=2, allow_nan=False) + '\n') args.public_items.write_text(json.dumps(public, indent=2, allow_nan=False) + '\n') print(json.dumps({k: summary[k] for k in ("n", "status", "verdict_counts", "role_flagged", "exact_verdict_agreement", "decision")})) if __name__ == "__main__": main()