| |
| """P2 evaluation report: pass@k, Wilson / bootstrap CIs, paired McNemar, correction lift. |
| |
| `--config sota` runs the separately registered SOTA analysis instead (docs/PREREG_SOTA.md): co-primary class A and |
| class B on miniF2F-test, unbiased pass@32 from n = 64, paired sign-flip test, Holm over the two, secondaries, exploratory |
| outputs, protocol-tuple tags, and the rewritten-statement count. Every report that touches a test split is appended to |
| an append-only, hash-chained look log. The P2 primary below is untouched. |
| |
| .venv/bin/python scripts/p2/eval_report.py \\ |
| --problems minif2f.jsonl mobench.jsonl \\ |
| --run base:gens_base.jsonl:ver_base.jsonl --run sft:gens_sft.jsonl:ver_sft.jsonl \\ |
| --baseline base --candidate sft --out-dir REPORT_DIR |
| |
| `--run NAME:GEN[,GEN..]:VER[,VER..]`. Writes `report.json` and `report.md`. Every table row names |
| its problem set, its token cap and the verifier budget; nothing is pooled across caps. |
| |
| Pre-registered endpoints (docs/P2.md, written before any data): PRIMARY = miniF2F solved@8 at cap |
| 16384, baseline vs post-SFT, paired exact McNemar, alpha = 0.05. Everything else in the report is |
| labelled SECONDARY and is NOT multiplicity-corrected. |
| |
| What this script refuses to do (each is a number that would be about a different experiment): |
| * score a run with a missing verification, a verification with no generation, a stale verdict |
| (`code_sha256` mismatch), or a problem with a different number of round-0 samples than the |
| rest (unless --allow-ragged-n): a missing sample is not a failure of the model; |
| * report pass@k for k > n; |
| * compare two runs whose verifier pins / config / budget differ (--allow-verifier-mismatch |
| to override; the mismatch is then printed in the report header); |
| * compare two runs that do not cover exactly the same problems; |
| * compare two runs generated with different sampling settings or prompted differently (--allow-design-mismatch); |
| * label a comparison PRIMARY unless it is the registered test: miniF2F at cap 16384, k = 8, ALL 244 problems, baseline = the |
| registered model revision, temperature 1.0 / top_p 0.95 / no top_k (anything else is printed as secondary, with the reason). |
| """ |
|
|
| from __future__ import annotations |
|
|
| import argparse |
| import getpass |
| import hashlib |
| import importlib.util |
| import json |
| import os |
| import re |
| import socket |
| import sys |
| from collections import Counter, defaultdict |
| from datetime import datetime, timezone |
| from pathlib import Path |
| from typing import Optional |
|
|
| REPO = Path(__file__).resolve().parents[2] |
| sys.path.insert(0, str(REPO)) |
|
|
|
|
| def _load_common(): |
| spec = importlib.util.spec_from_file_location("p2_common", Path(__file__).with_name("p2_common.py")) |
| mod = importlib.util.module_from_spec(spec) |
| sys.modules["p2_common"] = mod |
| spec.loader.exec_module(mod) |
| return mod |
|
|
|
|
| C = sys.modules.get("p2_common") or _load_common() |
|
|
| from align.eval import paired as PD |
| from align.eval import proof_stats as S |
|
|
| |
| |
| |
| PRIMARY = {"source": "minif2f", "cap": 16384, "k": 8, "n_problems": 244, |
| "baseline_model_rev": "dfd02e6271a58375dfbf3ece0175277cf6b6a89a", |
| "sampling": {"temperature": 1.0, "top_p": 0.95, "top_k": [0, -1]}} |
| DEFAULT_KS = (1, 2, 4, 8, 16, 32) |
|
|
|
|
| DOC = REPO / "docs" / "P2.md" |
| _PREREG_BLOCK = re.compile(r"<!-- PREREG:BEGIN -->\n(.*?)<!-- PREREG:END -->", re.S) |
| _PREREG_RECORDED = re.compile(r"PREREG_SHA256:\s*`([0-9a-f]{64})`") |
|
|
|
|
| def prereg_status(doc: Path = DOC) -> dict: |
| """Recompute the hash of the pre-registered block of docs/P2.md and compare it with the |
| value recorded next to it. A report whose analysis plan was edited after registration says so |
| in its header instead of silently presenting the new plan as the registered one.""" |
| if not Path(doc).exists(): |
| return {"status": "missing", "doc": str(doc)} |
| text = Path(doc).read_text() |
| m = _PREREG_BLOCK.search(text) |
| rec = _PREREG_RECORDED.search(text) |
| if not m: |
| return {"status": "no_block", "doc": str(doc)} |
| h = hashlib.sha256(m.group(1).encode()).hexdigest() |
| if not rec: |
| return {"status": "unrecorded", "computed": h} |
| return {"status": "verified" if rec.group(1) == h else "MODIFIED_AFTER_REGISTRATION", "recorded": rec.group(1), "computed": h} |
|
|
|
|
| class Sample: |
| __slots__ = ("accepted", "reasons", "finish", "n_tok", "parent", "budget", "verify_s", "eff_max", "sampling", "model_rev", "prompt_sha", |
| "fidelity") |
|
|
| def __init__(self, accepted, reasons, finish, n_tok, parent, budget, verify_s, eff_max, sampling=None, model_rev=None, prompt_sha=None, |
| fidelity=None): |
| self.accepted, self.reasons, self.finish = accepted, reasons, finish |
| self.n_tok, self.parent, self.budget, self.verify_s = n_tok, parent, budget, verify_s |
| self.eff_max = eff_max |
| self.sampling = sampling |
| self.model_rev, self.prompt_sha = model_rev, prompt_sha |
| |
| |
| self.fidelity = fidelity |
|
|
|
|
| def _sampling_of(rec: dict) -> tuple: |
| """(temperature, top_p, top_k) exactly as recorded; a missing key stays None (unknown is not 'default').""" |
| sp = rec.get("sampling") or {} |
| return (sp.get("temperature"), sp.get("top_p"), sp.get("top_k")) |
|
|
|
|
| def verifier_signature(v: dict) -> str: |
| cfg = dict(v.get("config", {})) |
| return hashlib.sha256(json.dumps({"pins": v.get("pins"), "config": cfg}, sort_keys=True).encode()).hexdigest()[:16] |
|
|
|
|
| |
|
|
|
|
| def _fidelity(statements: Optional[dict], pid: str, completion_text: str) -> Optional[str]: |
| if statements is None or pid not in statements: |
| return None |
| shown = C.displayed_signature(completion_text) |
| if shown is None: |
| return "unreadable" |
| return "faithful" if shown == C.statement_key(statements[pid])[2] else "rewritten" |
|
|
|
|
| def load_run(gen_paths, ver_paths, statements: Optional[dict] = None) -> tuple[dict[tuple, Sample], dict]: |
| """`statements` ({problem_id: formal_statement}) switches on the statement-fidelity check of accepted samples (SOTA configuration).""" |
| ver: dict[tuple, dict] = {} |
| sigs: Counter = Counter() |
| for p in ver_paths: |
| for rec in C.iter_jsonl(p): |
| key = C.gen_key(rec) |
| if key in ver: |
| raise C.ContractError(f"{p}: duplicate verification key {key}") |
| ver[key] = rec |
| sigs[verifier_signature(rec["verifier"])] += 1 |
| samples: dict[tuple, Sample] = {} |
| for p in gen_paths: |
| with open(p, encoding="utf-8") as f: |
| for lineno, line in enumerate(f, 1): |
| if not line.strip(): |
| continue |
| rec = json.loads(line) |
| C.validate_generation(rec) |
| key = C.gen_key(rec) |
| if key in samples: |
| raise C.ContractError(f"{p}:{lineno}: duplicate generation key {key}") |
| v = ver.get(key) |
| if v is None: |
| raise C.ContractError(f"{p}:{lineno}: generation {key} has no verification record") |
| if v["code_sha256"] != C.code_sha(rec["lean_code"]): |
| raise C.ContractError(f"{p}:{lineno}: stale verification for {key} (code_sha256 mismatch)") |
| par = rec["parent"] |
| samples[key] = Sample(bool(v["accepted"]), list(v["reasons"]), rec["finish_reason"], len(rec["completion_ids"]), |
| None if par is None else (par["problem_id"], int(par["sample_idx"]), int(par["round"]), key[3]), |
| v["budget_s"], v["verify_s"], C.effective_max_tokens(rec), |
| _sampling_of(rec), rec["model_rev"], hashlib.sha256(json.dumps(rec["prompt_ids"]).encode()).hexdigest()[:16], |
| _fidelity(statements, rec["problem_id"], rec["completion_text"]) if v["accepted"] else None) |
| orphan = set(ver) - set(samples) |
| if orphan: |
| raise C.ContractError(f"{len(orphan)} verification records have no generation (first: {sorted(orphan)[0]})") |
| meta = {"verifier_signatures": dict(sigs), "n_samples": len(samples), |
| "budgets": sorted({s.budget for s in samples.values()}), |
| "verifier": next(iter(ver.values()))["verifier"] if ver else None} |
| return samples, meta |
|
|
|
|
| |
|
|
|
|
| def round0_by_problem(samples, problem_ids, cap: int, allow_ragged: bool) -> dict[str, list[bool]]: |
| """problem_id -> accepted flags of round-0 samples ordered by sample_idx (contiguous 0..n-1).""" |
| per: dict[str, dict[int, bool]] = defaultdict(dict) |
| for (pid, si, rd, c), s in samples.items(): |
| if rd == 0 and c == cap and pid in problem_ids: |
| per[pid][si] = s.accepted |
| missing = sorted(set(problem_ids) - set(per)) |
| if missing: |
| raise C.ContractError(f"incomplete run at cap {cap}: {len(missing)} problems have no round-0 sample (first: {missing[0]})") |
| out = {} |
| for pid, d in per.items(): |
| if sorted(d) != list(range(len(d))): |
| raise C.ContractError(f"{pid}: round-0 sample_idx not contiguous from 0 ({sorted(d)[:12]}...)") |
| out[pid] = [d[i] for i in range(len(d))] |
| ns = {len(v) for v in out.values()} |
| if len(ns) > 1 and not allow_ragged: |
| raise C.ContractError(f"problems have different numbers of round-0 samples at cap {cap}: {sorted(ns)} (--allow-ragged-n to override)") |
| return out |
|
|
|
|
| def run_metrics(samples, problem_ids, cap: int, ks, seed: int, n_boot: int, allow_ragged: bool = False) -> dict: |
| flags = round0_by_problem(samples, set(problem_ids), cap, allow_ragged) |
| pids = sorted(flags) |
| n_min = min(len(v) for v in flags.values()) |
| counts = [(len(flags[p]), sum(flags[p])) for p in pids] |
| out: dict = {"cap_tokens": cap, "n_problems": len(pids), "n_per_problem_min": n_min, |
| "n_per_problem_max": max(len(v) for v in flags.values()), "n_samples_round0": sum(n for n, _ in counts)} |
| tot_s, tot_c = out["n_samples_round0"], sum(c for _, c in counts) |
| per_problem_p1 = [c / n for n, c in counts] |
| lo, hi = S.bootstrap_mean_ci(per_problem_p1, seed, n_boot) |
| nlo, nhi = S.wilson_interval(tot_c, tot_s) |
| out["pass@1"] = {"value": tot_c / tot_s, "ci95_cluster_bootstrap": [lo, hi], |
| "ci95_wilson_pooled_samples_ANTICONSERVATIVE": [nlo, nhi]} |
| out["pass@k"] = {} |
| out["solved@k"] = {} |
| for k in ks: |
| if k > n_min: |
| out["pass@k"][str(k)] = None |
| out["solved@k"][str(k)] = None |
| continue |
| vals = [S.pass_at_k(n, c, k) for n, c in counts] |
| blo, bhi = S.bootstrap_mean_ci(vals, seed + k, n_boot) |
| solved = sum(1 for p in pids if S.solved_at_k(flags[p], k)) |
| wlo, whi = S.wilson_interval(solved, len(pids)) |
| out["pass@k"][str(k)] = {"value": sum(vals) / len(vals), "ci95_cluster_bootstrap": [blo, bhi]} |
| out["solved@k"][str(k)] = {"solved": solved, "n": len(pids), "fraction": solved / len(pids), "ci95_wilson": [wlo, whi]} |
| r0 = [s for (pid, si, rd, c), s in samples.items() if rd == 0 and c == cap and pid in set(pids)] |
| out["completion"] = { |
| "length_capped_fraction": sum(1 for s in r0 if s.finish == "length") / len(r0), |
| "mean_completion_tokens": sum(s.n_tok for s in r0) / len(r0), |
| "no_code_fraction": sum(1 for s in r0 if s.reasons == ["no_code"]) / len(r0), |
| "reason_histogram": dict(Counter(r for s in r0 if not s.accepted for r in s.reasons[:1]).most_common(12)), |
| "verify_s_mean": sum(s.verify_s for s in r0) / len(r0), |
| "context_clipped_fraction": sum(1 for s in r0 if s.eff_max < cap) / len(r0), |
| "mean_effective_max_tokens": sum(s.eff_max for s in r0) / len(r0), |
| } |
| out["design"] = {"model_revs": sorted({s.model_rev for s in r0}), "sampling": sorted({tuple(s.sampling or ()) for s in r0}, key=str)} |
| out["_flags"] = flags |
| return out |
|
|
|
|
| def primary_blockers(src: str, cap: int, k: int, base_metrics: dict) -> list[str]: |
| """Why a comparison of this shape may NOT carry the PRIMARY label ([] = it is the registered test).""" |
| if (src, cap, k) != (PRIMARY["source"], PRIMARY["cap"], PRIMARY["k"]): |
| return ["not the registered (set, cap, k)"] |
| why = [] |
| if base_metrics["n_problems"] != PRIMARY["n_problems"]: |
| why.append(f"{base_metrics['n_problems']} problems scored, the registered set has {PRIMARY['n_problems']}") |
| revs = base_metrics["design"]["model_revs"] |
| if revs != [PRIMARY["baseline_model_rev"]]: |
| why.append(f"baseline model_rev {revs} is not the registered {PRIMARY['baseline_model_rev']}") |
| want = PRIMARY["sampling"] |
| for t, p, kk in base_metrics["design"]["sampling"] or [()]: |
| if (t, p) != (want["temperature"], want["top_p"]) or kk not in want["top_k"]: |
| why.append(f"baseline sampling (temperature, top_p, top_k) = {(t, p, kk)} is not the registered {(want['temperature'], want['top_p'], want['top_k'])}") |
| return why |
|
|
|
|
| PRIMARY_ALPHA = 0.05 |
|
|
|
|
| def registered_decision(m: dict) -> str: |
| """The pre-registered decision rule, applied mechanically so the wording is not chosen after the data: |
| success = p < alpha AND the candidate solves more problems; p < alpha with the candidate lower is a |
| significant degradation; anything else is "no detectable improvement" (never "no effect").""" |
| sig = m["p_value_exact_two_sided"] < PRIMARY_ALPHA |
| if sig and m["solved_b"] > m["solved_a"]: |
| return "SUCCESS: significant improvement" |
| if sig and m["solved_b"] < m["solved_a"]: |
| return "significant DEGRADATION" |
| return "no detectable improvement" |
|
|
|
|
| def design_mismatches(base: dict[tuple, Sample], cand: dict[tuple, Sample]) -> list[str]: |
| """Differences between the two systems' GENERATION that no verifier signature can show: sampling |
| settings and the round-0 prompts. A paired test of two runs that were not asked the same question |
| with the same sampler measures the harness, not the model.""" |
| out = [] |
| sb = {tuple(s.sampling or ()) for s in base.values()} |
| sc = {tuple(s.sampling or ()) for s in cand.values()} |
| if len(sb) != 1 or len(sc) != 1 or sb != sc: |
| out.append(f"sampling settings differ or are mixed: baseline {sorted(sb, key=str)} vs candidate {sorted(sc, key=str)}") |
|
|
| common = sorted(k for k in set(base) & set(cand) if k[2] == 0) |
| bad = sorted({k[0] for k in common if base[k].prompt_sha != cand[k].prompt_sha}) |
| if bad: |
| out.append(f"{len(bad)} problems were prompted differently in the two runs at the same sample keys (first: {bad[0]})") |
| return out |
|
|
|
|
| def compare(base: dict, cand: dict, k: int, seed: int, n_boot: int) -> dict: |
| fb, fc = base["_flags"], cand["_flags"] |
| if set(fb) != set(fc): |
| raise C.ContractError("the two runs cover different problem sets") |
| if min(base["n_per_problem_min"], cand["n_per_problem_min"]) < k: |
| return {"k": k, "available": False, "why": f"needs >= {k} samples per problem in both runs"} |
| sa = {p: S.solved_at_k(fb[p], k) for p in fb} |
| sb = {p: S.solved_at_k(fc[p], k) for p in fb} |
| m = S.mcnemar_exact(sa, sb) |
| pids = sorted(fb) |
| va = [S.pass_at_k(len(fb[p]), sum(fb[p]), k) for p in pids] |
| vb = [S.pass_at_k(len(fc[p]), sum(fc[p]), k) for p in pids] |
| dlo, dhi = S.paired_bootstrap_diff_ci(va, vb, seed + 7 * k, n_boot) |
| return {"k": k, "available": True, "mcnemar": m.to_json(), "delta_solved": m.solved_b - m.solved_a, |
| "delta_solved_fraction": (m.solved_b - m.solved_a) / m.n_problems, |
| "delta_pass@k": sum(vb) / len(vb) - sum(va) / len(va), "delta_pass@k_ci95_paired_bootstrap": [dlo, dhi]} |
|
|
|
|
| def correction_lift(samples, problem_ids, cap: int) -> Optional[dict]: |
| """Per-round table: how many unsolved problems did correction rounds convert, at what cost. |
| |
| A problem is solved through round r when ANY sample of rounds <= r was accepted. The lift is |
| confounded with compute (corrections are extra samples): the table reports samples and |
| completion tokens spent per round so the comparison against `pass@(n0 + m)` can be made.""" |
| pids = set(problem_ids) |
| rounds = sorted({rd for (pid, si, rd, c) in samples if c == cap and pid in pids}) |
| if len(rounds) < 2: |
| return None |
| acc_by_round: dict[int, set[str]] = defaultdict(set) |
| gen_by_round: Counter = Counter() |
| tok_by_round: Counter = Counter() |
| acc_samples: Counter = Counter() |
| for (pid, si, rd, c), s in samples.items(): |
| if c != cap or pid not in pids: |
| continue |
| gen_by_round[rd] += 1 |
| tok_by_round[rd] += s.n_tok |
| if s.accepted: |
| acc_by_round[rd].add(pid) |
| acc_samples[rd] += 1 |
| if rd > 0: |
| par = s.parent |
| if par is None or par not in samples: |
| raise C.ContractError(f"correction sample {(pid, si, rd, c)} has no parent among the generations") |
| if samples[par].accepted: |
| raise C.ContractError(f"correction sample {(pid, si, rd, c)} corrects an ACCEPTED parent {par}") |
| n = len(pids) |
| solved = set(acc_by_round[0]) |
| rows = [{"round": 0, "samples": gen_by_round[0], "completion_tokens": tok_by_round[0], "accepted_samples": acc_samples[0], |
| "newly_solved_problems": len(solved), "solved_through_round": len(solved), "solved_fraction": len(solved) / n}] |
| unsolved_after0 = n - len(solved) |
| for r in rounds[1:]: |
| new = acc_by_round[r] - solved |
| solved |= acc_by_round[r] |
| lo, hi = S.wilson_interval(len(solved) - rows[0]["solved_through_round"], unsolved_after0) if unsolved_after0 else (0.0, 0.0) |
| rows.append({"round": r, "samples": gen_by_round[r], "completion_tokens": tok_by_round[r], "accepted_samples": acc_samples[r], |
| "sample_success_rate": acc_samples[r] / gen_by_round[r] if gen_by_round[r] else None, |
| "newly_solved_problems": len(new), "solved_through_round": len(solved), "solved_fraction": len(solved) / n, |
| "conversion_of_round0_unsolved": (len(solved) - rows[0]["solved_through_round"]) / unsolved_after0 if unsolved_after0 else None, |
| "conversion_ci95_wilson": [lo, hi]}) |
| return {"cap_tokens": cap, "n_problems": n, "rows": rows, |
| "note": "lift is extra-sample compute, not free: compare solved_through_round with pass@(n0 + corrections per problem) of round 0 when n0 allows"} |
|
|
|
|
| |
|
|
|
|
| def build_report(problems_by_source: dict[str, dict], runs: dict[str, tuple], baseline: Optional[str], candidate: Optional[str], |
| ks, seed: int, n_boot: int, allow_ragged: bool, allow_verifier_mismatch: bool, allow_design_mismatch: bool = False) -> dict: |
| rep: dict = {"preregistered_primary": dict(PRIMARY), "prereg": prereg_status(), "runs": {}, "comparisons": [], "correction_lift": {}} |
| metrics: dict[str, dict] = {} |
| for name, (samples, meta) in runs.items(): |
| caps = sorted({c for (_, _, _, c) in samples}) |
| rep["runs"][name] = {"meta": meta, "tables": {}} |
| metrics[name] = {} |
| for src, probs in problems_by_source.items(): |
| for cap in caps: |
| covered = {pid for (pid, _, _, c) in samples if c == cap and pid in probs} |
| if not covered: |
| continue |
| m = run_metrics(samples, set(probs), cap, ks, seed, n_boot, allow_ragged) |
| metrics[name][(src, cap)] = m |
| rep["runs"][name]["tables"][f"{src}@{cap}"] = {k: v for k, v in m.items() if not k.startswith("_")} |
| lift = correction_lift(samples, set(probs), cap) |
| if lift: |
| rep["correction_lift"][f"{name}/{src}@{cap}"] = lift |
| if baseline and candidate: |
| sig_b = {verifier_signature(runs[baseline][1]["verifier"])} if runs[baseline][1]["verifier"] else set() |
| sig_c = {verifier_signature(runs[candidate][1]["verifier"])} if runs[candidate][1]["verifier"] else set() |
| if sig_b != sig_c or runs[baseline][1]["budgets"] != runs[candidate][1]["budgets"]: |
| msg = f"verifier differs: baseline {sig_b} budgets {runs[baseline][1]['budgets']} vs candidate {sig_c} budgets {runs[candidate][1]['budgets']}" |
| if not allow_verifier_mismatch: |
| raise C.ContractError(msg + " (--allow-verifier-mismatch to compare anyway)") |
| rep["verifier_mismatch"] = msg |
| design = design_mismatches(runs[baseline][0], runs[candidate][0]) |
| if design: |
| if not allow_design_mismatch: |
| raise C.ContractError("; ".join(design) + " (--allow-design-mismatch to compare anyway)") |
| rep["design_mismatch"] = design |
| for (src, cap), mb in sorted(metrics[baseline].items()): |
| mc = metrics[candidate].get((src, cap)) |
| if mc is None: |
| continue |
| for k in ks: |
| cmpd = compare(mb, mc, k, seed, n_boot) |
| blockers = primary_blockers(src, cap, k, mb) |
| cmpd.update({"source": src, "cap_tokens": cap, "baseline": baseline, "candidate": candidate, |
| "endpoint": "PRIMARY" if not blockers else "secondary"}) |
| if not blockers and cmpd["available"]: |
| cmpd["decision"] = registered_decision(cmpd["mcnemar"]) |
| if blockers and blockers != ["not the registered (set, cap, k)"]: |
| cmpd["not_primary_because"] = blockers |
| rep["comparisons"].append(cmpd) |
| if not any(c["endpoint"] == "PRIMARY" and c["available"] for c in rep["comparisons"]): |
| rep["primary_missing"] = "the pre-registered primary comparison (miniF2F, cap 16384, k=8) could not be computed from these runs" |
| near = [c for c in rep["comparisons"] if c.get("not_primary_because")] |
| if near: |
| rep["primary_missing"] += "; the primary-shaped comparison below is NOT the registered test: " + "; ".join(near[0]["not_primary_because"]) |
| return rep |
|
|
|
|
| def render_md(rep: dict) -> str: |
| L = ["# P2 evaluation report", "", |
| f"Pre-registered primary endpoint: {PRIMARY['source']} solved@{PRIMARY['k']} at cap {PRIMARY['cap']}, paired exact McNemar, alpha 0.05. All other rows are secondary and not multiplicity-corrected.", ""] |
| pr = rep.get("prereg", {}) |
| L += [f"Pre-registration (docs/P2.md): {pr.get('status')}" + (f" (sha256 {pr.get('computed')})" if pr.get("computed") else ""), ""] |
| if pr.get("status") != "verified": |
| L += ["WARNING: the registered analysis plan could not be verified; treat every p-value below as exploratory.", ""] |
| if rep.get("verifier_mismatch"): |
| L += [f"WARNING: {rep['verifier_mismatch']}", ""] |
| if rep.get("design_mismatch"): |
| L += [f"WARNING: generation design differs between the runs: {'; '.join(rep['design_mismatch'])}", ""] |
| if rep.get("primary_missing"): |
| L += [f"WARNING: {rep['primary_missing']}", ""] |
| pct = lambda x: "n/a" if x is None else f"{100 * x:.1f}%" |
| for name, r in rep["runs"].items(): |
| L += [f"## Run `{name}`", "", f"verifier budgets (s): {r['meta']['budgets']}; samples: {r['meta']['n_samples']}", "", |
| "| set @ cap | problems | n/prob | pass@1 [95% cluster-bootstrap] | " + " | ".join(f"pass@{k}" for k in DEFAULT_KS) + " | length-capped | mean tok | mean effective max_tokens |", |
| "|" + "---|" * (7 + len(DEFAULT_KS))] |
| for tab, t in r["tables"].items(): |
| p1 = t["pass@1"] |
| cells = [] |
| for k in DEFAULT_KS: |
| v = t["pass@k"].get(str(k)) |
| cells.append("n/a" if v is None else f"{pct(v['value'])} ({t['solved@k'][str(k)]['solved']}/{t['n_problems']})") |
| L.append(f"| {tab} | {t['n_problems']} | {t['n_per_problem_min']} | {pct(p1['value'])} [{pct(p1['ci95_cluster_bootstrap'][0])}, {pct(p1['ci95_cluster_bootstrap'][1])}] | " |
| + " | ".join(cells) + f" | {pct(t['completion']['length_capped_fraction'])} | {t['completion']['mean_completion_tokens']:.0f} | {t['completion']['mean_effective_max_tokens']:.0f} |") |
| L.append("") |
| if rep["comparisons"]: |
| L += ["## Paired comparison (McNemar exact on solved@k)", "", |
| "| endpoint | set @ cap | k | solved base | solved cand | only base | only cand | discordant | p (exact, 2-sided) | delta pass@k [95% paired bootstrap] |", "|" + "---|" * 10] |
| for c in rep["comparisons"]: |
| if not c["available"]: |
| L.append(f"| {c['endpoint']} | {c['source']} @ {c['cap_tokens']} | {c['k']} | n/a | n/a | | | | | {c['why']} |") |
| continue |
| m = c["mcnemar"] |
| L.append(f"| {c['endpoint']} | {c['source']} @ {c['cap_tokens']} | {c['k']} | {m['solved_a']} | {m['solved_b']} | {m['only_a']} | {m['only_b']} | " |
| f"{m['n_discordant']} | {m['p_value_exact_two_sided']:.4f} | {100 * c['delta_pass@k']:+.1f} pp [{100 * c['delta_pass@k_ci95_paired_bootstrap'][0]:+.1f}, {100 * c['delta_pass@k_ci95_paired_bootstrap'][1]:+.1f}] |") |
| L.append("") |
| for c in rep["comparisons"]: |
| if c.get("decision"): |
| L += [f"**Pre-registered decision (primary, alpha {PRIMARY_ALPHA}): {c['decision']}.** A null result means no gain of roughly 6 pp or more was detectable, not that there is no effect (docs/P2.md section 1).", ""] |
| if rep["correction_lift"]: |
| L += ["## Correction-round lift", "", "| run / set @ cap | round | samples | completion tokens | accepted samples | newly solved problems | solved through round | solved % | conversion of round-0 unsolved [95% Wilson] |", "|" + "---|" * 9] |
| for name, t in rep["correction_lift"].items(): |
| for r in t["rows"]: |
| conv = r.get("conversion_of_round0_unsolved") |
| ci = r.get("conversion_ci95_wilson") |
| L.append(f"| {name} | {r['round']} | {r['samples']} | {r['completion_tokens']} | {r['accepted_samples']} | {r['newly_solved_problems']} | " |
| f"{r['solved_through_round']} | {pct(r['solved_fraction'])} | " + ("" if conv is None else f"{pct(conv)} [{pct(ci[0])}, {pct(ci[1])}]") + " |") |
| L += ["", "Lift costs extra samples: compare against round-0 pass@k at the same total samples per problem before calling it a gain.", ""] |
| return "\n".join(L) |
|
|
|
|
| |
| |
| |
| |
| |
| |
| |
|
|
| SOTA = { |
| "name": "sota-v1", |
| "alpha": 0.05, |
| "coprimary": {"source": "minif2f", "split": "test", "n_problems": 244, "cap": 40960, "n": 64, "k": 32}, |
| "class_b": {"max_repairs": 2, "ledger_tokens": 40960}, |
| "secondary": {"mcnemar_first_n": 32, "proofnet": {"source": "proofnet", "split": "test", "k": 32}}, |
| "verifier_budget_s": 300.0, |
| "n_flips": 100_000, |
| "exploratory_ks": (1, 8, 16, 32), |
| } |
|
|
| LOOK_LOG_DEFAULT = REPO / "docs" / "evidence" / "test_split_looks.jsonl" |
|
|
|
|
| def _prereg_hash_module(): |
| spec = importlib.util.spec_from_file_location("p2_prereg_hash", Path(__file__).with_name("prereg_hash.py")) |
| mod = importlib.util.module_from_spec(spec) |
| sys.modules["p2_prereg_hash"] = mod |
| spec.loader.exec_module(mod) |
| return mod |
|
|
|
|
| def prereg_overall_hash() -> Optional[str]: |
| """Current sha256 of (registration text + analysis code), as printed by scripts/p2/prereg_hash.py; None if an input is missing.""" |
| try: |
| return _prereg_hash_module().compute()["sha256"] |
| except FileNotFoundError: |
| return None |
|
|
|
|
| def verifier_hash(v: Optional[dict]) -> Optional[str]: |
| """Full sha256 of the verifier identity (pins + config, canonical JSON; `workers` is deliberately excluded, as in |
| `verifier_signature`). This is the value the frozen-verifier rule compares.""" |
| if not v: |
| return None |
| return hashlib.sha256(json.dumps({"pins": v.get("pins"), "config": dict(v.get("config", {}))}, sort_keys=True).encode()).hexdigest() |
|
|
|
|
| def class_a_counts(samples, pids, cap: int) -> tuple[dict, dict]: |
| """-> ({pid: (n, c)}, {pid: flags}) from round-0 samples at `cap` (ragged n is refused).""" |
| flags = round0_by_problem(samples, set(pids), cap, allow_ragged=False) |
| return {p: (len(f), sum(f)) for p, f in flags.items()}, flags |
|
|
|
|
| def class_b_counts(samples, pids, cap: int, max_repairs: int, ledger: int, allow_unrepaired: bool = False) -> Optional[dict]: |
| """Class B: {pid: (n_trajectories, c_solved)}; None if the run has no repair rounds at all. |
| |
| A trajectory is the chain round-0 sample -> its repair -> its repair (<= `max_repairs`). It is solved iff some element |
| is accepted AND the cumulative completion tokens up to and including that element are <= `ledger` (the 40k TOTAL). |
| A failed sample with ledger left but no repair record is a missing record, not a failure: refused unless `allow_unrepaired`.""" |
| pset = set(pids) |
| if not any(rd > 0 and c == cap and pid in pset for (pid, _, rd, c) in samples): |
| return None |
| child: dict[tuple, tuple] = {} |
| for key, s in samples.items(): |
| if key[2] > 0 and key[3] == cap and key[0] in pset: |
| if s.parent in child: |
| raise C.ContractError(f"two repairs claim the same parent {s.parent}") |
| child[s.parent] = key |
| out = {} |
| for pid in sorted(pset): |
| n = solved = 0 |
| for key in sorted(k for k in samples if k[0] == pid and k[2] == 0 and k[3] == cap): |
| n += 1 |
| cum, cur, ok = 0, key, False |
| for _ in range(max_repairs + 1): |
| s = samples[cur] |
| cum += s.n_tok |
| if s.accepted: |
| ok = cum <= ledger |
| break |
| if cum >= ledger or cur[2] >= max_repairs: |
| break |
| nxt = child.get(cur) |
| if nxt is None: |
| if not allow_unrepaired: |
| raise C.ContractError(f"{cur} failed with ledger left but has no repair record (--allow-unrepaired to count it as a failure)") |
| break |
| cur = nxt |
| solved += ok |
| out[pid] = (n, solved) |
| return out |
|
|
|
|
| def sota_contrast(cb: dict, cc: dict, k: int, seed: int, n_boot: int, n_flips: int) -> dict: |
| """Paired contrast of the unbiased pass@k estimand; `cb`/`cc` are {pid: (n, c)}. Sign-flip p + problem bootstrap CI.""" |
| if set(cb) != set(cc): |
| raise C.ContractError("the two runs cover different problem sets") |
| pids = sorted(cb) |
| if min(min(n for n, _ in cb.values()), min(n for n, _ in cc.values())) < k: |
| return {"available": False, "k": k, "why": f"needs >= {k} samples per problem in both runs"} |
| va = PD.pass_at_k_array([cb[p][0] for p in pids], [cb[p][1] for p in pids], k) |
| vb = PD.pass_at_k_array([cc[p][0] for p in pids], [cc[p][1] for p in pids], k) |
| d = vb - va |
| t = PD.sign_flip_test(d, n_flips=n_flips, seed=seed) |
| lo, hi = PD.paired_bootstrap_ci(d, seed + 11 * k, n_boot) |
| return {"available": True, "k": k, "n_problems": len(pids), "n_per_problem": sorted({cb[p][0] for p in pids} | {cc[p][0] for p in pids}), |
| "mean_baseline": float(va.mean()), "mean_candidate": float(vb.mean()), "delta": float(d.mean()), |
| "delta_ci95_paired_bootstrap": [lo, hi], "p_sign_flip": t["p"], "sign_flip_method": t["method"], "sign_flip_n": t["n_flips"], |
| "n_changed_problems": t["n_nonzero"], "null_quantiles_of_delta": t["null_quantiles"]} |
|
|
|
|
| def _fidelity_summary(samples, pids, cap: int) -> dict: |
| pset = set(pids) |
| acc = [(key, s) for key, s in samples.items() if key[3] == cap and key[0] in pset and s.accepted] |
| cnt = Counter(s.fidelity or "unchecked" for _, s in acc) |
| bad = [(k, s) for k, s in acc if s.fidelity in ("rewritten", "unreadable")] |
| return {"accepted_samples": len(acc), "faithful": cnt["faithful"], "rewritten": cnt["rewritten"], "unreadable": cnt["unreadable"], |
| "unchecked": cnt["unchecked"], |
| "rewritten_statement_count": cnt["rewritten"] + cnt["unreadable"], |
| "problems_with_rewritten_accept": len({k[0] for k, _ in bad}), |
| "in_repair_rounds": sum(1 for k, _ in bad if k[2] > 0)} |
|
|
|
|
| def _strict_class_a(samples, pids, cap: int) -> dict: |
| """Class-A (n, c) with rewritten/unreadable accepts counted as failures (exploratory fidelity sensitivity).""" |
| per: dict[str, dict[int, bool]] = defaultdict(dict) |
| for (pid, si, rd, c), s in samples.items(): |
| if rd == 0 and c == cap and pid in pids: |
| per[pid][si] = s.accepted and s.fidelity == "faithful" |
| return {p: (len(d), sum(d.values())) for p, d in per.items()} |
|
|
|
|
| def append_look_log(path, entry: dict) -> dict: |
| """Append ONE hash-chained line to the test-split look log (append-only: O_APPEND, flock, fsync; each line carries the |
| sha256 of the previous line's content, so a deletion or edit is detectable by `verify_look_log`).""" |
| import fcntl |
|
|
| path = Path(path) |
| path.parent.mkdir(parents=True, exist_ok=True) |
| with open(path, "a+b") as f: |
| fcntl.flock(f, fcntl.LOCK_EX) |
| f.seek(0) |
| lines = [l for l in f.read().splitlines() if l.strip()] |
| prev = json.loads(lines[-1])["entry_sha256"] if lines else "0" * 64 |
| rec = dict(entry, prev_sha256=prev) |
| rec["entry_sha256"] = hashlib.sha256(json.dumps(rec, sort_keys=True).encode()).hexdigest() |
| f.write((json.dumps(rec, sort_keys=True) + "\n").encode()) |
| f.flush() |
| os.fsync(f.fileno()) |
| return rec |
|
|
|
|
| def verify_look_log(path) -> dict: |
| """Recompute the chain. -> {ok, n_entries, error}.""" |
| prev, n = "0" * 64, 0 |
| for i, line in enumerate(Path(path).read_text().splitlines(), 1): |
| if not line.strip(): |
| continue |
| rec = json.loads(line) |
| got = rec.pop("entry_sha256", None) |
| if rec.get("prev_sha256") != prev: |
| return {"ok": False, "n_entries": n, "error": f"line {i}: chain broken (prev_sha256 does not match the previous entry)"} |
| if hashlib.sha256(json.dumps(rec, sort_keys=True).encode()).hexdigest() != got: |
| return {"ok": False, "n_entries": n, "error": f"line {i}: entry content does not match its hash"} |
| prev, n = got, n + 1 |
| return {"ok": True, "n_entries": n, "error": None} |
|
|
|
|
| def look_entry(config: str, splits: list, report_json: Path, report_md: Path, inputs: dict) -> dict: |
| try: |
| who = getpass.getuser() |
| except Exception: |
| who = os.environ.get("USER", "unknown") |
| return {"who": {"user": who, "host": socket.gethostname(), "pid": os.getpid(), "argv": list(sys.argv)}, |
| "when_utc": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%fZ"), |
| "what": {"config": config, "splits": sorted(splits), "report_json": str(report_json), "report_md": str(report_md), |
| "report_json_sha256": C.sha256_file(report_json), "report_md_sha256": C.sha256_file(report_md)}, |
| "inputs": inputs} |
|
|
|
|
| def build_sota_report(problems_by_source: dict, runs: dict, baseline: str, candidate: str, seed: int, n_boot: int, |
| frozen_verifier_hash: Optional[str] = None, registered_hash: Optional[str] = None, |
| allow_verifier_mismatch: bool = False, allow_design_mismatch: bool = False, allow_unrepaired: bool = False, |
| n_flips: Optional[int] = None) -> dict: |
| cfg, alpha = SOTA, SOTA["alpha"] |
| n_flips = n_flips or cfg["n_flips"] |
| cp, cb_cfg = cfg["coprimary"], cfg["class_b"] |
| sb, sc = runs[baseline][0], runs[candidate][0] |
| mb_, mc_ = runs[baseline][1], runs[candidate][1] |
| sig_b = {verifier_signature(mb_["verifier"])} if mb_["verifier"] else set() |
| sig_c = {verifier_signature(mc_["verifier"])} if mc_["verifier"] else set() |
| rep: dict = {"configuration": cfg["name"], "registered": {k: v for k, v in cfg.items()}, "baseline": baseline, "candidate": candidate, |
| "seed": seed} |
| if sig_b != sig_c or mb_["budgets"] != mc_["budgets"]: |
| msg = f"verifier differs: baseline {sig_b} budgets {mb_['budgets']} vs candidate {sig_c} budgets {mc_['budgets']}" |
| if not allow_verifier_mismatch: |
| raise C.ContractError(msg + " (--allow-verifier-mismatch to compare anyway)") |
| rep["verifier_mismatch"] = msg |
| design = design_mismatches(sb, sc) |
| if design: |
| if not allow_design_mismatch: |
| raise C.ContractError("; ".join(design) + " (--allow-design-mismatch to compare anyway)") |
| rep["design_mismatch"] = design |
|
|
| cur = prereg_overall_hash() |
| rep["prereg"] = {"current_overall_sha256": cur, "registered_overall_sha256": registered_hash, |
| "status": "unregistered" if not registered_hash else ("verified" if registered_hash == cur else "MODIFIED_AFTER_REGISTRATION")} |
|
|
| def subset(src, split): |
| return {pid: r for pid, r in problems_by_source.get(src, {}).items() if r.get("split") == split} |
|
|
| test = subset(cp["source"], cp["split"]) |
| pids = set(test) |
| if not pids: |
| raise C.ContractError(f"no {cp['source']}/{cp['split']} problems supplied: the co-primary set is empty") |
| prob_set_hash = hashlib.sha256(json.dumps(sorted((p, test[p]["statement_sha"]) for p in test)).encode()).hexdigest() |
|
|
| |
| blockers = [] |
| if len(pids) != cp["n_problems"]: |
| blockers.append(f"{len(pids)} {cp['source']}/{cp['split']} problems scored, the registered set has {cp['n_problems']}") |
| for name, (_, meta) in runs.items(): |
| if name not in (baseline, candidate): |
| continue |
| if len(meta["verifier_signatures"]) != 1: |
| blockers.append(f"run {name}: records were verified under {len(meta['verifier_signatures'])} different verifier configurations") |
| if meta["budgets"] != [cfg["verifier_budget_s"]]: |
| blockers.append(f"run {name}: verifier budgets {meta['budgets']} are not the registered [{cfg['verifier_budget_s']}]") |
| vh = verifier_hash(meta["verifier"]) |
| if frozen_verifier_hash is None: |
| blockers.append("no frozen verifier hash supplied (--frozen-verifier-hash): the frozen-verifier rule cannot be checked") |
| elif vh != frozen_verifier_hash: |
| blockers.append(f"run {name}: verifier hash {vh} is not the frozen {frozen_verifier_hash}") |
| blockers = list(dict.fromkeys(blockers)) |
| if rep["prereg"]["status"] != "verified": |
| blockers.append(f"pre-registration hash status: {rep['prereg']['status']}") |
| if rep.get("verifier_mismatch"): |
| blockers.append("verifier mismatch overridden") |
| if rep.get("design_mismatch"): |
| blockers.append("generation design mismatch overridden") |
| if allow_unrepaired: |
| blockers.append("--allow-unrepaired: failed samples without a repair record were counted as failures") |
|
|
| fid = {name: _fidelity_summary(runs[name][0], pids, cp["cap"]) for name in (baseline, candidate)} |
| if any(f["unchecked"] for f in fid.values()): |
| blockers.append("statement fidelity (rewritten-statement count) was not checked for every accepted sample") |
| tok = {name: sum(s.n_tok for (pid, _, _, c), s in runs[name][0].items() if pid in pids and c == cp["cap"]) for name in (baseline, candidate)} |
|
|
| def tag(klass: str, source: str, split: str, n_prob: int, n: int, k: int, ledger=None) -> dict: |
| return {"class": klass, "benchmark": f"{source}/{split}", "n_problems": n_prob, "problem_set_sha256": prob_set_hash if source == cp["source"] else None, |
| "n": n, "k": k, "cap_tokens": cp["cap"], "ledger_tokens": ledger, |
| "lean_mathlib_pins": (mb_["verifier"] or {}).get("pins"), "verifier_hash": verifier_hash(mb_["verifier"]), |
| "verifier_budget_s": mb_["budgets"], "verifier_config": (mb_["verifier"] or {}).get("config"), |
| "rewritten_statement_count": {"baseline": fid[baseline]["rewritten_statement_count"], "candidate": fid[candidate]["rewritten_statement_count"]}, |
| "completion_tokens_total": {"baseline": tok[baseline], "candidate": tok[candidate]}, "statement_version": "as supplied in the problems file (see problem_set_sha256)"} |
|
|
| |
| co: dict = {} |
| ca_b, fl_b = class_a_counts(sb, pids, cp["cap"]) |
| ca_c, fl_c = class_a_counts(sc, pids, cp["cap"]) |
| for nm, cn in (("baseline", ca_b), ("candidate", ca_c)): |
| ns = {n for n, _ in cn.values()} |
| if ns != {cp["n"]}: |
| blockers.append(f"{nm}: n per problem is {sorted(ns)}, the registered n is {cp['n']}") |
| A = sota_contrast(ca_b, ca_c, cp["k"], seed, n_boot, n_flips) |
| A["protocol"] = tag("A", cp["source"], cp["split"], len(pids), cp["n"], cp["k"]) |
| co["A"] = A |
| |
| bb = class_b_counts(sb, pids, cp["cap"], cb_cfg["max_repairs"], cb_cfg["ledger_tokens"], allow_unrepaired) |
| bc = class_b_counts(sc, pids, cp["cap"], cb_cfg["max_repairs"], cb_cfg["ledger_tokens"], allow_unrepaired) |
| if bb is None or bc is None: |
| B = {"available": False, "k": cp["k"], "why": "no Lean-feedback repair rounds in the " + ("baseline" if bb is None else "candidate") + " run"} |
| else: |
| B = sota_contrast(bb, bc, cp["k"], seed + 1, n_boot, n_flips) |
| if B["available"] and B["n_per_problem"] != [cp["n"]]: |
| blockers.append(f"class B: trajectories per problem {B['n_per_problem']}, registered {cp['n']}") |
| B["protocol"] = tag("B", cp["source"], cp["split"], len(pids), cp["n"], cp["k"], cb_cfg["ledger_tokens"]) |
| co["B"] = B |
| pA = A["p_sign_flip"] if A["available"] else 1.0 |
| pB = B["p_sign_flip"] if B["available"] else 1.0 |
| h = PD.holm([pA, pB], alpha) |
| for key, nm, adj, rej in zip(("A", "B"), ("A", "B"), h["adjusted"], h["reject"]): |
| e = co[key] |
| e["p_holm"], e["reject_holm"] = adj, bool(rej) |
| if e["available"]: |
| e["decision"] = ("SUCCESS: significant improvement" if rej and e["delta"] > 0 else |
| "significant DEGRADATION" if rej and e["delta"] < 0 else "no detectable improvement") |
| co["holm"] = {"alpha": alpha, "m": 2, "p_in": [pA, pB], **{k: v for k, v in h.items() if k != "alpha"}} |
| rep["coprimary"] = co |
| rep["confirmatory_blockers"] = blockers |
| rep["label"] = "CONFIRMATORY" if not blockers and A["available"] and B["available"] else "NOT CONFIRMATORY" |
| rep["fidelity"] = fid |
|
|
| |
| sec: dict = {} |
| n1 = cfg["secondary"]["mcnemar_first_n"] |
| if min(len(f) for f in fl_b.values()) >= n1 and min(len(f) for f in fl_c.values()) >= n1: |
| m = S.mcnemar_exact({p: S.solved_at_k(fl_b[p], n1) for p in fl_b}, {p: S.solved_at_k(fl_c[p], n1) for p in fl_c}) |
| sec["mcnemar_first_32"] = dict(m.to_json(), definition=f"solved@{n1} = any of the FIRST {n1} samples (by sample_idx), class A") |
| else: |
| sec["mcnemar_first_32"] = {"available": False, "why": f"needs >= {n1} samples per problem"} |
| sec["mean_pass@1"] = sota_contrast(ca_b, ca_c, 1, seed + 2, n_boot, n_flips) |
| pn = cfg["secondary"]["proofnet"] |
| pset = set(subset(pn["source"], pn["split"])) |
| if pset and all(any(k[0] == p and k[2] == 0 and k[3] == cp["cap"] for k in sb) for p in pset) and all(any(k[0] == p and k[2] == 0 and k[3] == cp["cap"] for k in sc) for p in pset): |
| pb_, _ = class_a_counts(sb, pset, cp["cap"]) |
| pc_, _ = class_a_counts(sc, pset, cp["cap"]) |
| sec["proofnet_class_A"] = sota_contrast(pb_, pc_, pn["k"], seed + 3, n_boot, n_flips) |
| sec["proofnet_class_A"]["protocol"] = tag("A", pn["source"], pn["split"], len(pset), min(n for n, _ in pb_.values()), pn["k"]) |
| else: |
| sec["proofnet_class_A"] = {"available": False, "why": "no complete proofnet/test class-A samples in both runs"} |
| rep["secondary"] = sec |
|
|
| |
| exp: dict = {"fidelity_strict_class_A": None, "other_sets": {}, "class_A_other_k": {}} |
| sa_b, sa_c = _strict_class_a(sb, pids, cp["cap"]), _strict_class_a(sc, pids, cp["cap"]) |
| exp["fidelity_strict_class_A"] = dict(sota_contrast(sa_b, sa_c, cp["k"], seed + 4, n_boot, n_flips), |
| definition="class A with rewritten/unreadable accepted samples counted as failures") |
| for k in cfg["exploratory_ks"]: |
| if k != cp["k"]: |
| exp["class_A_other_k"][str(k)] = sota_contrast(ca_b, ca_c, k, seed + 5 + k, n_boot, n_flips) |
| for src, probs in sorted(problems_by_source.items()): |
| for split in sorted({r.get("split") for r in probs.values()}): |
| if (src, split) in ((cp["source"], cp["split"]), (pn["source"], pn["split"])): |
| continue |
| ids = set(subset(src, split)) |
| try: |
| xb, _ = class_a_counts(sb, ids, cp["cap"]) |
| xc, _ = class_a_counts(sc, ids, cp["cap"]) |
| except C.ContractError as e: |
| exp["other_sets"][f"{src}/{split}"] = {"available": False, "why": str(e)} |
| continue |
| exp["other_sets"][f"{src}/{split}"] = {str(k): sota_contrast(xb, xc, k, seed + 9 + k, n_boot, n_flips) for k in cfg["exploratory_ks"]} |
| rep["exploratory"] = exp |
| return rep |
|
|
|
|
| def render_sota_md(rep: dict) -> str: |
| pct = lambda x: "n/a" if x is None else f"{100 * x:.2f}" |
| L = [f"# SOTA evaluation report ({rep['configuration']})", "", |
| f"**{rep['label']}.** Baseline `{rep['baseline']}`, candidate `{rep['candidate']}`. Registered in docs/PREREG_SOTA.md: co-primary A and B on miniF2F-test, " |
| "unbiased pass@32 from n = 64, paired sign-flip test, two-sided alpha 0.05, Holm over {A, B}.", ""] |
| pr = rep["prereg"] |
| L += [f"Pre-registration hash: {pr['status']} (current {pr['current_overall_sha256']}, registered {pr['registered_overall_sha256']}).", ""] |
| if rep["confirmatory_blockers"]: |
| L += ["WARNING: not confirmatory because:"] + [f"* {b}" for b in rep["confirmatory_blockers"]] + [""] |
| for w in ("verifier_mismatch", "design_mismatch"): |
| if rep.get(w): |
| L += [f"WARNING: {w}: {rep[w]}", ""] |
| L += ["## Co-primary (Holm over A and B)", "", |
| "| class | k | n/prob | pass@k base % | pass@k cand % | delta pp [95% problem bootstrap] | p (sign-flip) | p (Holm) | decision |", "|" + "---|" * 9] |
| for key in ("A", "B"): |
| e = rep["coprimary"][key] |
| if not e["available"]: |
| L.append(f"| {key} | {e['k']} | | | | | | | UNAVAILABLE: {e['why']} |") |
| continue |
| L.append(f"| {key} | {e['k']} | {e['n_per_problem']} | {pct(e['mean_baseline'])} | {pct(e['mean_candidate'])} | {pct(e['delta']).rjust(1)} " |
| f"[{pct(e['delta_ci95_paired_bootstrap'][0])}, {pct(e['delta_ci95_paired_bootstrap'][1])}] | {e['p_sign_flip']:.5f} ({e['sign_flip_method']}) | " |
| f"{e['p_holm']:.5f} | {e['decision']} |") |
| L += ["", "Rewritten-statement count (accepted samples whose displayed theorem differs from the reference statement; unreadable included):", ""] |
| for name, f in rep["fidelity"].items(): |
| L.append(f"* `{name}`: {f['rewritten_statement_count']} of {f['accepted_samples']} accepted samples " |
| f"(rewritten {f['rewritten']}, unreadable {f['unreadable']}, unchecked {f['unchecked']}; {f['problems_with_rewritten_accept']} problems; {f['in_repair_rounds']} in repair rounds)") |
| L += ["", "Protocol tuple (class A):", "", "```json", json.dumps(rep["coprimary"]["A"]["protocol"], indent=1, default=str), "```", "", |
| "## Secondary (not multiplicity-corrected)", ""] |
| m = rep["secondary"]["mcnemar_first_32"] |
| if m.get("n_problems"): |
| L.append(f"* McNemar exact, solved@32 of the first 32 samples, class A: solved {m['solved_a']} vs {m['solved_b']}, only-base {m['only_a']}, only-cand {m['only_b']}, p = {m['p_value_exact_two_sided']:.4f}") |
| else: |
| L.append(f"* McNemar first 32: unavailable ({m.get('why')})") |
| for key, nm in (("mean_pass@1", "mean pass@1 (class A)"), ("proofnet_class_A", "ProofNet-test class A pass@32")): |
| e = rep["secondary"][key] |
| L.append(f"* {nm}: " + (f"{pct(e['mean_baseline'])} -> {pct(e['mean_candidate'])}, delta {pct(e['delta'])} pp [{pct(e['delta_ci95_paired_bootstrap'][0])}, {pct(e['delta_ci95_paired_bootstrap'][1])}], p = {e['p_sign_flip']:.4f}" |
| if e.get("available") else f"unavailable ({e.get('why')})")) |
| L += ["", "## Exploratory (descriptive; no decision attached)", ""] |
| e = rep["exploratory"]["fidelity_strict_class_A"] |
| if e.get("available"): |
| L.append(f"* class A pass@32 with rewritten accepts counted as failures: {pct(e['mean_baseline'])} -> {pct(e['mean_candidate'])}, delta {pct(e['delta'])} pp, p = {e['p_sign_flip']:.4f}") |
| for k, e in rep["exploratory"]["class_A_other_k"].items(): |
| if e.get("available"): |
| L.append(f"* miniF2F-test class A pass@{k}: delta {pct(e['delta'])} pp [{pct(e['delta_ci95_paired_bootstrap'][0])}, {pct(e['delta_ci95_paired_bootstrap'][1])}], p = {e['p_sign_flip']:.4f}") |
| for name, ks in rep["exploratory"]["other_sets"].items(): |
| for k, e in (ks.items() if "available" not in ks else []): |
| if e.get("available"): |
| L.append(f"* {name} class A pass@{k}: delta {pct(e['delta'])} pp, p = {e['p_sign_flip']:.4f}") |
| if "available" in ks: |
| L.append(f"* {name}: unavailable ({ks['why']})") |
| return "\n".join(L) + "\n" |
|
|
|
|
| def main(argv=None) -> int: |
| ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) |
| ap.add_argument("--config", choices=("p2", "sota"), default="p2", |
| help="p2 = the registered P2 primary (default, unchanged); sota = the separately registered SOTA analysis (docs/PREREG_SOTA.md)") |
| ap.add_argument("--verify-prereg", action="store_true", help="print the pre-registration hash status and exit") |
| ap.add_argument("--problems", nargs="+") |
| ap.add_argument("--run", action="append", help="NAME:GEN[,GEN..]:VER[,VER..]") |
| ap.add_argument("--baseline") |
| ap.add_argument("--candidate") |
| ap.add_argument("--out-dir") |
| ap.add_argument("--ks", type=int, nargs="+", default=list(DEFAULT_KS)) |
| ap.add_argument("--seed", type=int, default=0) |
| ap.add_argument("--bootstrap", type=int, default=10_000) |
| ap.add_argument("--allow-ragged-n", action="store_true") |
| ap.add_argument("--allow-verifier-mismatch", action="store_true") |
| ap.add_argument("--allow-design-mismatch", action="store_true", help="compare runs whose sampling settings or round-0 prompts differ (recorded in the report)") |
| ap.add_argument("--frozen-verifier-hash", help="sota: sha256 of the frozen verifier (pins + config); a run verified under anything else is not confirmatory") |
| ap.add_argument("--registered-hash", help="sota: the overall hash from scripts/p2/prereg_hash.py that was time-stamped") |
| ap.add_argument("--n-flips", type=int, default=SOTA["n_flips"], help="sota: Monte Carlo sign flips when more than 20 problems differ") |
| ap.add_argument("--allow-unrepaired", action="store_true", help="sota: count failed samples with no repair record as class-B failures (recorded)") |
| ap.add_argument("--look-log", default=os.environ.get("P2_LOOK_LOG", str(LOOK_LOG_DEFAULT)), |
| help="append-only log of every test-split report (default docs/evidence/test_split_looks.jsonl, or $P2_LOOK_LOG)") |
| a = ap.parse_args(argv) |
| if a.verify_prereg: |
| st = prereg_status() |
| print(json.dumps(st)) |
| return 0 if st["status"] == "verified" else 1 |
| if not (a.problems and a.run and a.out_dir): |
| ap.error("--problems, --run and --out-dir are required") |
| problems_by_source: dict[str, dict] = defaultdict(dict) |
| for p in a.problems: |
| for pid, rec in C.load_problems(p).items(): |
| problems_by_source[rec["source"]][pid] = rec |
| statements = {pid: rec["formal_statement"] for probs in problems_by_source.values() for pid, rec in probs.items()} if a.config == "sota" else None |
| runs = {} |
| for spec in a.run: |
| name, gens, vers = spec.split(":") |
| if name in runs: |
| raise SystemExit(f"duplicate run name {name}") |
| runs[name] = load_run(gens.split(","), vers.split(","), statements) |
| if bool(a.baseline) != bool(a.candidate): |
| raise SystemExit("--baseline and --candidate go together") |
| for n in (a.baseline, a.candidate): |
| if n and n not in runs: |
| raise SystemExit(f"unknown run {n}") |
| if a.config == "sota": |
| if not (a.baseline and a.candidate): |
| raise SystemExit("--config sota needs --baseline and --candidate") |
| rep = build_sota_report(dict(problems_by_source), runs, a.baseline, a.candidate, a.seed, a.bootstrap, a.frozen_verifier_hash, a.registered_hash, |
| a.allow_verifier_mismatch, a.allow_design_mismatch, a.allow_unrepaired, a.n_flips) |
| md = render_sota_md(rep) |
| else: |
| rep = build_report(dict(problems_by_source), runs, a.baseline, a.candidate, tuple(a.ks), a.seed, a.bootstrap, a.allow_ragged_n, a.allow_verifier_mismatch, a.allow_design_mismatch) |
| md = render_md(rep) |
| out = Path(a.out_dir) |
| out.mkdir(parents=True, exist_ok=True) |
| (out / "report.json").write_text(json.dumps(rep, indent=1, default=str)) |
| (out / "report.md").write_text(md) |
| splits = sorted({f"{src}/{r.get('split')}" for src, probs in problems_by_source.items() for r in probs.values() if r.get("split") == "test"}) |
| if splits: |
| inputs = {"problems": {p: C.sha256_file(p) for p in a.problems}, |
| "runs": {n: {"gen": {g: C.sha256_file(g) for g in spec.split(":")[1].split(",")}, "ver": {v: C.sha256_file(v) for v in spec.split(":")[2].split(",")}} |
| for n, spec in ((sp.split(":")[0], sp) for sp in a.run)}} |
| append_look_log(a.look_log, look_entry(a.config, splits, out / "report.json", out / "report.md", inputs)) |
| print(md) |
| return 0 |
|
|
|
|
| if __name__ == "__main__": |
| raise SystemExit(main()) |
|
|
