201 lines
10 KiB
Python
201 lines
10 KiB
Python
"""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'<section id="{html.escape(row["blind_id"])}"><h2>{row["blind_id"]}</h2>'
|
|
f'<h3>Passage</h3><pre>{html.escape(item["source_text"])}</pre>'
|
|
f'<h3>Lean</h3><pre>{html.escape(item["lean_code"])}</pre>'
|
|
f'<h3>Fresh assistant judgment: {row["judge"]["verdict"]}</h3>'
|
|
f'<p>{html.escape(row["judge"]["reason"])}</p>'
|
|
f'<h3>Source roles</h3><pre>{html.escape(json.dumps(row["source"], indent=2, ensure_ascii=False))}</pre>'
|
|
f'<h3>Deterministic alignment</h3><pre>{html.escape(json.dumps(row["role"], indent=2, ensure_ascii=False))}</pre></section>'
|
|
)
|
|
path.write_text('<!doctype html><html lang="en"><meta charset="utf-8">'
|
|
'<title>P0 GPT-assisted diagnostic</title><style>'
|
|
'body{font:16px/1.6 system-ui;max-width:1050px;margin:32px auto;padding:0 20px}'
|
|
'pre{white-space:pre-wrap;overflow-wrap:anywhere;background:#f4f5f6;padding:16px}'
|
|
'section{border-top:1px solid #ccc;margin:40px 0}</style>'
|
|
'<h1>P0: 65-item GPT-assisted diagnostic</h1>'
|
|
'<p>Private review. Same-assistant, non-blind diagnosis; no independent accuracy or gate decision. '
|
|
'Passages and model-derived descriptions must remain outside Git.</p>'
|
|
+ ''.join(sections) + '</html>', 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()
|