From ca987035389fd734df2d41dc2f036d50e5622656 Mon Sep 17 00:00:00 2001 From: claude-bot <10+claude-bot@noreply.localhost> Date: Thu, 22 Oct 2026 05:52:28 +0000 Subject: [PATCH] Pilot prep: simplex schema, projection, alignment, lint, stats, pilot scripts, protocol v1 Co-Authored-By: Claude Sonnet 5.5 Claude-Session: https://claude.ai/code/session_01FYqwNMyPUHeTbFWg83GQq6 --- .gitignore | 6 + docs/experiments/PROTOCOL.md | 85 ++++++++ experiments/pilot/README.md | 15 ++ experiments/pilot/make_sheet.py | 176 ++++++++++++++++ experiments/pilot/run_pilot.py | 110 ++++++++++ experiments/pilot/score.py | 93 +++++++++ leanbench/align.py | 359 ++++++++++++++++++++++++++++++++ leanbench/evalstats.py | 93 +++++++++ leanbench/lint.py | 48 +++++ leanbench/project.py | 260 +++++++++++++++++++++++ leanbench/schema.py | 135 ++++++++++++ leanbench/srcsimplex.py | 165 +++++++++++++++ tests/test_align.py | 148 +++++++++++++ tests/test_evalstats.py | 44 ++++ tests/test_lint.py | 34 +++ tests/test_pilot_scripts.py | 38 ++++ tests/test_project.py | 94 +++++++++ tests/test_schema.py | 49 +++++ tests/test_srcsimplex.py | 72 +++++++ 19 files changed, 2024 insertions(+) create mode 100644 docs/experiments/PROTOCOL.md create mode 100644 experiments/pilot/README.md create mode 100644 experiments/pilot/make_sheet.py create mode 100644 experiments/pilot/run_pilot.py create mode 100644 experiments/pilot/score.py create mode 100644 leanbench/align.py create mode 100644 leanbench/evalstats.py create mode 100644 leanbench/lint.py create mode 100644 leanbench/project.py create mode 100644 leanbench/schema.py create mode 100644 leanbench/srcsimplex.py create mode 100644 tests/test_align.py create mode 100644 tests/test_evalstats.py create mode 100644 tests/test_lint.py create mode 100644 tests/test_pilot_scripts.py create mode 100644 tests/test_project.py create mode 100644 tests/test_schema.py create mode 100644 tests/test_srcsimplex.py diff --git a/.gitignore b/.gitignore index 33e3fe9..96f7889 100644 --- a/.gitignore +++ b/.gitignore @@ -37,3 +37,9 @@ htmlcov/ .DS_Store .idea/ .vscode/ + +# Local experiment data: passages from copyrighted books and per-run outputs +runs/ +**/private/ +*.labels.json +.leanbench-cache/ diff --git a/docs/experiments/PROTOCOL.md b/docs/experiments/PROTOCOL.md new file mode 100644 index 0000000..d6964ce --- /dev/null +++ b/docs/experiments/PROTOCOL.md @@ -0,0 +1,85 @@ +# Experiment protocol v1 (for approval) + +Status: **awaiting approval. No experiment has been run and no API spend has occurred.** +Written 2026-10-22. Companion to `docs/research-plan.md` (what and why), `docs/TODO.md` (task list), `docs/acl-2027-bar.md` (bar). Target: ACL 2027 via ARR, deadline 4 Jan 2027 AoE. + +## 1. Decision requested + +Approve **Phase 0 (pilot)** now. It costs a few yuan of API calls, about 4 hours of one person's labeling time, and about 2 days elapsed. Its only job is to tell us whether the paper's central idea (role alignment) has signal before we spend six weeks and a large token budget on it. Phases 1-3 are described so you can see where this leads, but each needs its own go from you after the previous gate. + +| Phase | What | Cost (API / people) | Elapsed | Gate to continue | +|---|---|---|---|---| +| 0 Pilot | 65 existing items: detectors vs. blind human labels | ~200 calls (< 1M tokens) / 1 labeler, ~4 h | 2 days | Section 3 rule | +| 1 Calibration | 6-8 books ingested, 150-page detection gold, labeling guideline, kappa pilot on 50 items | ~10-20M tokens / 2 annotators, ~15 h each | 2 weeks | kappa >= 0.6 | +| 2 Main | E4 (4+ models), E5 (600 labeled), E6 (filter comparison), E7, E8 | ~135M tokens + judges / 2 annotators, ~60 h total | 3-4 weeks | go/no-go on 8 Dec | +| 3 Extras | E9 retrieval/proving, E10-E11, robustness | open, bounded by time left | to 24 Dec | tables frozen 24 Dec | + +Token and CPU figures are the plan's rough estimates (`research-plan.md` 7.9); they are recomputed from the Phase 0 numbers before Phase 1. + +## 2. Hypotheses tested in Phase 0 + +- **H1 (gap).** Compiling overstates fidelity: a large share of formalizations, compiled or not, do not say what the passage says. +- **H3 (role alignment).** The deterministic alignment of the passage's roles against the Lean signature's roles flags unfaithful items better than the alternatives. +- Phase 0 does *not* test H2, H4, H5 (guided formalization, retrieval, proving). They need Lean runs and more data. + +## 3. Phase 0 design + +**Items.** The 65 items in `runs/p85-110/items.jsonl` (pages 85-110 of one algebra text; 37 statements, 28 problems). Only 10 of them compiled in Lean. So the pilot judges *fidelity detection over all items*; it cannot yet say anything solid about "compiles but wrong" (n <= 10). That subset is reported with its interval but does not decide anything. + +**Labels (ground truth).** One Lean-literate labeler (you, or someone you name) uses `labelsheet.html`: passage and Lean code side by side, shuffled, no compile status, no scores, no model output. Verdicts: faithful / minor drift / wrong / not formalizable, plus error types (missing or extra hypothesis, wrong type, quantifier, wrong conclusion, degenerate, hallucinated API). Labels are finished and saved before any detector output is opened. One labeler means no kappa in Phase 0; that is a stated limitation, fixed in Phase 1. + +**Detectors, all run on the same 65 items:** + +| Detector | Sees | Cost | +|---|---|---| +| compile-only | Lean verdict stored in the items file | 0 | +| static lint | Lean code (projected simplex) | 0 | +| single LLM judge | passage + Lean code, one call | 65 calls | +| **role alignment** | passage -> source simplex (LLM, never sees Lean); Lean -> simplex (deterministic); deterministic alignment | <= 130 calls | + +Model: `qwen-max` via DashScope (the only key available). Temperature 0. Prompts are versioned and hashed in the run metadata (`srcsimplex-v1`, `judge-v1`). The judge and the source extractor are the *same model family*, which is a known confound and is listed as a limitation; Phase 2 adds a second family. + +**Metrics.** Positive class = "wrong". For each detector: recall, precision, false-flag rate on faithful items, F1, Wilson 95% intervals; role-alignment score AUC. Secondary: positive = wrong or minor drift. Per-error-type recall for role alignment. Reported alongside: share of items degenerate in the projection (closed statements with no binders project to one role; 15 of 65 here, and several of those are also wrong formalizations, e.g. `IsGroup`, a name that does not exist). + +**Decision rule (fixed now, before any label exists).** + +- **GO**: role recall on wrong items >= 0.5, false-flag rate on faithful items <= 0.2, role recall >= judge recall + 0.10, and role false-flag rate no higher than the judge's. +- **PARTIAL**: role recall >= 0.4 with false-flag rate <= 0.3, *or* role alignment flags at least 3 wrong items the judge misses (complementary signal). +- **NO-GO**: otherwise. Then the paper is re-scoped to evaluation + taxonomy + resource (Findings-level), and I tell you before spending anything else. + +With about 65 items every interval is wide (a recall of 0.5 on 20 wrong items has a 95% interval of roughly 0.30-0.70). The pilot can reject an idea with no signal; it cannot confirm effect size. I will say so in the report and will not present a GO as a result for the paper. + +**What I will not do.** Tune thresholds (`0.3`, `0.2`), the concept table, or the prompts after seeing labels or detector output on these 65 items; if any change is needed it is recorded as a new version and rerun on a fresh sample in Phase 1. No cherry-picking of items. Every run writes its model id, prompt hashes, call counts and token usage next to its results. + +## 4. Phase 1-3 outline (not requested yet) + +- **Phase 1.** Pick and license-check 6-8 books (>= 3 open); ingest to >= 1,500 items; sample 150 pages (detection gold), 600 items (audited core), 150 passages (gold source simplices); annotation guideline v1 with 30 worked examples; two annotators label 50 items; **gate kappa >= 0.6**. Also E0 (projection vs. manual parse on >= 100 signatures, gate >= 95%). +- **Phase 2.** E4 gap across >= 4 general models + 1 specialised formalizer (paired McNemar, bootstrap CIs); E5 labels for the 600-item core (kappa on a 200-item overlap, third adjudicator); E6 filter comparison against single judge, two-judge, BEq+, static checkers, CriticLean/FormalAlign, trajectory length, with shuffled-span and same-family-judge controls; E7 slot-constrained vs. free-form; E8 repair ablation. **Go/no-go on 8 Dec**: role alignment must beat a single judge on at least two error types at equal false-reject rate to stay a headline contribution. +- **Phase 3.** E9 schema-guided retrieval/proving (only if schedule holds), E10 contamination, E11 cost/time; tables frozen 24 Dec; writing from 1 Dec; submit by 3 Jan. + +## 5. Infrastructure (ready) + +- Runner `ubuntu24.04-x64-lean4.35`, Lean 4.35.0, Mathlib v4.35.0 cache, image `lean-bench:latest` kept on host; CI green on both repos. +- Code on branch `exp/pilot-prep`: schema, projection, source-side extraction, alignment, lint, stats; labeling sheet, run and score scripts; 140 tests, ruff clean. Offline smoke of run -> score passes with a stub model. +- Pilot data and labels stay local (`runs/`, `**/private/` are gitignored). The copyrighted book and text derived from it are never committed or released; the public release will use open-licensed books and hashes/spans only. + +## 6. Risks specific to the near term + +| Risk | Effect | Mitigation | +|---|---|---| +| Single model family for extractor and judge | Judge and role alignment errors correlate; bias favors neither fairly | Disclosed; second family in Phase 2 | +| One labeler | No agreement estimate; label noise unknown | Phase 1 two-annotator kappa; pilot reported as signal-only | +| Closed statements project to one role | Role alignment has little to compare | Counted and reported; degenerate flag treated as its own detector output | +| 65 items from one chapter span | Narrow topics, small n | Pilot only gates a decision; Phase 1 widens | +| Competing 2026 papers (Beyond Compilation, ShadowBench, Lean-GAP, ATLAS) | Novelty narrows | Position as role-level structural fidelity; weekly literature check | +| 11 weeks to deadline | Slippage in annotation | Fixed gates; fallback to evaluation + resource paper | + +## 7. What I need from you + +1. **Approve Phase 0**, or tell me what to change. +2. **A labeler** for the 65 items (about 4 hours; Lean-literate). If it is you, say so and I will hand you `labelsheet.html`. I will not label them myself, because I produced the code being judged and the detectors share my model's blind spots. +3. For Phase 1 (not yet): a second annotator, a second model key (a different family from Qwen, for the judge control), and your choice of books once I have the license shortlist. + +## 8. After approval + +I run, in order: build the sheet; (you label); dry-run to confirm counts; run detectors (<= 400 calls hard cap, resumable); score; report the decision with the numbers, the failures, and the items where detectors disagree with you. Then I stop and wait for your go on Phase 1. diff --git a/experiments/pilot/README.md b/experiments/pilot/README.md new file mode 100644 index 0000000..8d7d62c --- /dev/null +++ b/experiments/pilot/README.md @@ -0,0 +1,15 @@ +# Pilot (Phase 0) + +Question: on the 65 existing items, does role alignment (passage simplex vs. Lean simplex) find +unfaithful formalizations better than a single LLM judge, compile-only, or static lint? +Full protocol and decision rule: `docs/experiments/PROTOCOL.md`. + + python experiments/pilot/make_sheet.py --items runs/p85-110/items.jsonl --out runs/pilot/private + # a person opens runs/pilot/private/labelsheet.html, labels all items, saves labels.json + python experiments/pilot/run_pilot.py --items runs/p85-110/items.jsonl --out runs/pilot/results.jsonl --dry-run + python experiments/pilot/run_pilot.py --items runs/p85-110/items.jsonl --out runs/pilot/results.jsonl + python experiments/pilot/score.py --results runs/pilot/results.jsonl \ + --labels runs/pilot/private/labels.json --key runs/pilot/private/key.json + +`runs/` and `**/private/` are gitignored: the passages come from a copyrighted book and must not be +committed. Labeling must finish BEFORE `score.py` is run; do not look at `results.jsonl` while labeling. diff --git a/experiments/pilot/make_sheet.py b/experiments/pilot/make_sheet.py new file mode 100644 index 0000000..7424c5e --- /dev/null +++ b/experiments/pilot/make_sheet.py @@ -0,0 +1,176 @@ +"""Build the blind labeling sheet for the pilot. + + python experiments/pilot/make_sheet.py --items runs/p85-110/items.jsonl \ + --out runs/pilot/private --seed 7 + +Writes, into a directory that must stay local (it contains passages from a +copyrighted book; ``runs/`` and ``**/private/`` are gitignored): + + labelsheet.html open in a browser; label every item; press "Download labels.json" + key.json blind id -> item id (keep it; the scorer needs it) + +The sheet shows the passage and the Lean code. It deliberately hides whether Lean +accepted the code, any model output, and any score, so the labels are independent of +everything being evaluated. Order is shuffled with --seed. +""" +from __future__ import annotations + +import argparse +import html +import json +import random +from pathlib import Path + +FIDELITY = [ + ("faithful", "Faithful", "Says what the passage says (notation and encoding choices aside)."), + ("minor_drift", "Minor drift", + "Right idea, but a small difference a careful reader would object to."), + ("wrong", "Wrong", "States a different or false/vacuous claim, or drops/adds a " + "hypothesis that changes the meaning."), + ("not_formalizable", "Not formalizable / no claim", + "The passage is not a precise statement, or the extraction is garbled."), +] +ERRORS = [ + ("missing_hypothesis", "Missing hypothesis"), + ("extra_hypothesis", "Extra hypothesis"), + ("wrong_type", "Wrong type / structure (e.g. ℕ for ℤ, Group for CommGroup)"), + ("quantifier", "Quantifier error (∀/∃, order)"), + ("wrong_conclusion", "Wrong conclusion"), + ("degenerate", "Vacuous / trivial / degenerate"), + ("hallucinated_api", "Names that do not exist in Mathlib"), + ("other", "Other (say in notes)"), +] + +PAGE = """ + +lean-bench pilot labeling sheet +
+

lean-bench pilot: label {n} items

+

For each item: if this Lean code compiled, would it state what the passage +states? Judge the meaning of what is written, not whether Lean accepts it (that is hidden on +purpose). Notation and harmless encoding choices are fine. Tick the error types that apply +when the verdict is not "Faithful". Your answers are saved in this browser as you go; +press Download labels.json when finished and put the file next to key.json.

+
Verdicts
    {legend}
+
+{cards} +
+""" + + +def esc(text: str) -> str: + return html.escape(text or "", quote=True) + + +def card(blind: str, item: dict) -> str: + verdicts = "".join( + f'' + for key, name, _ in FIDELITY + ) + errors = "".join( + f'' + for key, name in ERRORS + ) + label = esc(item.get("label") or "") + return ( + f'
' + f'{blind} · page {item.get("start_page")} · ' + f'{esc(item.get("kind", ""))} · {label}' + f'
Passage
' + f'
{esc(item.get("source_text", ""))}
' + f'
Lean
{esc(item.get("lean_code", ""))}
' + f'
Verdict{verdicts}
' + f'
Errors (if not faithful){errors}
' + f'
' + ) + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + parser.add_argument("--items", required=True, type=Path) + parser.add_argument("--out", required=True, type=Path) + parser.add_argument("--seed", type=int, default=7) + args = parser.parse_args() + + items = [json.loads(line) for line in args.items.read_text(encoding="utf-8").splitlines() + if line.strip()] + order = list(range(len(items))) + random.Random(args.seed).shuffle(order) + blind_ids = [f"P{n:02d}" for n in range(1, len(items) + 1)] + key = {blind_ids[pos]: items[idx]["id"] for pos, idx in enumerate(order)} + + legend = "".join(f"
  • {esc(n)}: {esc(d)}
    • " for _, n, d in FIDELITY) + cards = "\n".join(card(blind_ids[pos], items[idx]) for pos, idx in enumerate(order)) + args.out.mkdir(parents=True, exist_ok=True) + (args.out / "labelsheet.html").write_text( + PAGE.format(n=len(items), legend=legend, cards=cards, ids=json.dumps(blind_ids)), + encoding="utf-8", + ) + (args.out / "key.json").write_text( + json.dumps({"seed": args.seed, "items_file": args.items.name, "key": key}, indent=2), + encoding="utf-8", + ) + print(f"wrote {args.out/'labelsheet.html'} ({len(items)} items) and key.json") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/experiments/pilot/run_pilot.py b/experiments/pilot/run_pilot.py new file mode 100644 index 0000000..f30693a --- /dev/null +++ b/experiments/pilot/run_pilot.py @@ -0,0 +1,110 @@ +"""Run the pilot detectors over the pilot items. Resumable; one JSON line per item. + + python experiments/pilot/run_pilot.py --items runs/p85-110/items.jsonl \ + --out runs/pilot/results.jsonl --dry-run # no API calls: shows the plan + python experiments/pilot/run_pilot.py --items ... --out ... --max-calls 400 + +Detectors (all see the passage and/or the Lean code, never the human labels): + lint static patterns on the projected simplex (no LLM) + role source-side simplex (LLM sees only the passage) aligned to the Lean simplex + judge one LLM call that sees passage + Lean and says faithful / not + compile the stored Lean verdict (``verified``), read from the items file + +Hard stops: --max-calls bounds API calls; a rerun skips items already in --out. +""" +from __future__ import annotations + +import argparse +import hashlib +import json +import sys +from pathlib import Path + +from leanbench import srcsimplex +from leanbench.align import align +from leanbench.config import Settings +from leanbench.lint import lint +from leanbench.llm import OpenAICompatibleClient +from leanbench.project import project + +JUDGE_PROMPT_VERSION = "judge-v1" +JUDGE_SYSTEM = ( + "You check a Lean 4 (Mathlib) formalization against a textbook passage. Decide whether the " + "Lean statement says what the passage says: same objects, same hypotheses, same conclusion, " + "same quantifiers. Ignore whether it compiles. Reply with JSON only: " + '{"faithful": true|false, "confidence": 0..1, "reason": "one sentence"}' +) + + +def judge(client, passage: str, lean: str) -> dict: + reply = client.complete_json([ + {"role": "system", "content": JUDGE_SYSTEM}, + {"role": "user", "content": f"Passage:\n{passage}\n\nLean:\n{lean}"}, + ]) + if not isinstance(reply, dict) or "faithful" not in reply: + raise ValueError(f"unusable judge reply: {str(reply)[:200]}") + return {"faithful": bool(reply["faithful"]), "confidence": reply.get("confidence"), + "reason": str(reply.get("reason", ""))[:300]} + + +def run_item(client, item: dict) -> dict: + code = item.get("lean_code") or "" + row: dict = {"id": item["id"], "compiled": bool(item.get("verified")), "errors": {}} + proj = project(code) + row["projection"] = proj.as_dict() + row["lint"] = lint(proj) + try: + src = srcsimplex.extract_source_simplex(client, item.get("source_text", ""), + item.get("label", "")) + al = align(src, proj) + row["source"] = [r.__dict__ for r in src.roles] + row["role"] = al.as_dict() + except Exception as error: # a failed extraction is data, not a crash + row["errors"]["role"] = str(error)[:300] + try: + row["judge"] = judge(client, item.get("source_text", ""), code) + except Exception as error: + row["errors"]["judge"] = str(error)[:300] + return row + + +def main() -> int: + ap = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + ap.add_argument("--items", required=True, type=Path) + ap.add_argument("--out", required=True, type=Path) + ap.add_argument("--max-calls", type=int, default=400) + ap.add_argument("--dry-run", action="store_true") + args = ap.parse_args() + + items = [json.loads(x) for x in args.items.read_text(encoding="utf-8").splitlines() if x.strip()] + done = set() + if args.out.exists(): + done = {json.loads(x)["id"] for x in args.out.read_text(encoding="utf-8").splitlines() if x.strip()} + todo = [i for i in items if i["id"] not in done] + settings = Settings.from_env() + est = len(todo) * 3 # extract (<=2 tries) + judge + print(f"items={len(items)} done={len(done)} todo={len(todo)} model={settings.llm.model} " + f"est_calls<={est} (extract x2 + judge) cap={args.max_calls}") + print("prompts:", srcsimplex.PROMPT_VERSION, JUDGE_PROMPT_VERSION, + hashlib.sha256((srcsimplex.SYSTEM_PROMPT + JUDGE_SYSTEM).encode()).hexdigest()[:12]) + if args.dry_run: + return 0 + + client = OpenAICompatibleClient.from_settings(settings.llm) + args.out.parent.mkdir(parents=True, exist_ok=True) + with args.out.open("a", encoding="utf-8") as fh: + for item in todo: + if client.usage.calls >= args.max_calls: + print("call cap reached; rerun to resume", file=sys.stderr) + break + fh.write(json.dumps(run_item(client, item), ensure_ascii=False) + "\n") + fh.flush() + meta = {"model": settings.llm.model, "usage": client.usage.as_dict(), + "prompts": [srcsimplex.PROMPT_VERSION, JUDGE_PROMPT_VERSION]} + (args.out.parent / "run_meta.json").write_text(json.dumps(meta, indent=2), encoding="utf-8") + print(json.dumps(meta)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/experiments/pilot/score.py b/experiments/pilot/score.py new file mode 100644 index 0000000..8ee534c --- /dev/null +++ b/experiments/pilot/score.py @@ -0,0 +1,93 @@ +"""Score the pilot: detectors vs. the blind human labels, with the pre-registered decision. + + python experiments/pilot/score.py --results runs/pilot/results.jsonl \ + --labels runs/pilot/private/labels.json --key runs/pilot/private/key.json + +Positive class = unfaithful. Primary: verdict == "wrong", over ALL labeled items (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 json +from pathlib import Path + +from leanbench.evalstats import auc, confusion + + +def load(args): + results = {json.loads(x)["id"]: json.loads(x) for x in args.results.read_text(encoding="utf-8").splitlines() if x.strip()} + labels = json.loads(args.labels.read_text(encoding="utf-8"))["labels"] + key = json.loads(args.key.read_text(encoding="utf-8"))["key"] + rows = [] + for blind, item_id in key.items(): + lab = labels.get(blind) or {} + if item_id in results and lab.get("verdict"): + rows.append({**results[item_id], "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["role"]["auc_score"] = auc([-(r.get("role", {}).get("score", 1.0)) for r in rows], + [positive(r) for r in rows]) + return out + + +def decide(sub): + role, judge = sub["role"], sub["judge"] + if role["tp"] + role["fn"] == 0: + return "INSUFFICIENT: no wrong items in the labels" + 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) + 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) + args = ap.parse_args() + rows = [r for r in load(args) if r["verdict"] != "not_formalizable"] + compiled = [r for r in rows if r["compiled"]] + summary = { + "n_labeled": len(rows), "n_compiled": len(compiled), + "verdicts": {v: sum(r["verdict"] == v for r in rows) for v in + ("faithful", "minor_drift", "wrong")}, + "compiled_but_wrong": sum(r["verdict"] == "wrong" for r in compiled), + "primary_all": report(rows, 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["decision"] = decide(summary["primary_all"]) + print(json.dumps(summary, indent=2, default=str)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/leanbench/align.py b/leanbench/align.py new file mode 100644 index 0000000..c186911 --- /dev/null +++ b/leanbench/align.py @@ -0,0 +1,359 @@ +"""Compare the source-side simplex with the Lean-side one, role by role. + +The result says *which* parameter disagrees -- a hypothesis the Lean dropped, one it +invented, a number system that changed, an ``∃`` turned into a ``∀`` -- which a +single "faithful: yes/no" from a judge cannot. Matching is deterministic and cheap: +shared symbols (``G``, ``n``) and shared concepts (``group``, ``finite``, ``prime``) +through a small synonym table. An LLM pairing (``llm_pairing``) can replace the +matching step, so the two can be compared as an ablation. +""" +from __future__ import annotations + +import re +from dataclasses import dataclass, field +from typing import Any + +from .llm import ChatClient +from .project import Projection +from .schema import KIND_FACT, KIND_GOAL, KIND_OBJECT, KIND_OWN, Role +from .srcsimplex import SourceRole, SourceSimplex + +DEFAULT_THRESHOLD = 0.3 +GOAL_THRESHOLD = 0.2 + +ERROR_TYPES = ( + "missing_hypothesis", "missing_object", "extra_hypothesis", "extra_object", + "type_conflict", "quantifier", "goal_mismatch", "degenerate", +) +# extra_object is reported but does not flag an item: Lean often introduces helpers. +FLAG_ERRORS = frozenset(ERROR_TYPES) - {"extra_object"} + +CONCEPTS: dict[str, set[str]] = { + "group": {"group", "grp"}, "abelian": {"abelian", "comm", "commutative"}, + "finite": {"finite", "fintype", "card"}, "infinite": {"infinite"}, + "ring": {"ring"}, "field": {"field"}, "domain": {"domain"}, + "integer": {"integer", "int", "ℤ"}, "natural": {"natural", "nat", "ℕ"}, + "real": {"real", "ℝ"}, "complex": {"complex", "ℂ"}, "rational": {"rational", "ℚ"}, + "prime": {"prime"}, "coprime": {"coprime", "relatively"}, + "divides": {"divides", "divisor", "dvd", "∣", "divisible", "divisibility"}, + "subgroup": {"subgroup"}, "ideal": {"ideal"}, "normal": {"normal"}, + "homomorphism": {"homomorphism", "hom", "morphism"}, + "isomorphism": {"isomorphism", "equiv", "≃", "≅", "isomorphic"}, + "set": {"set", "subset"}, "function": {"function", "map"}, + "even": {"even"}, "odd": {"odd"}, "cyclic": {"cyclic"}, "order": {"order", "orderof"}, + "gcd": {"gcd"}, "lcm": {"lcm"}, "positive": {"positive", "pos"}, + "nonzero": {"nonzero"}, "unit": {"unit", "units"}, "matrix": {"matrix"}, + "vector": {"vector"}, "space": {"space"}, "continuous": {"continuous"}, + "sequence": {"sequence", "seq"}, "polynomial": {"polynomial"}, + "kernel": {"kernel", "ker"}, "image": {"image", "range"}, + "injective": {"injective"}, "surjective": {"surjective"}, "bijective": {"bijective"}, +} +_TOKEN_TO_CONCEPT = {tok: concept for concept, toks in CONCEPTS.items() for tok in toks} +NUMBER_SYSTEMS = {"integer", "natural", "real", "complex", "rational"} + +_TOKEN = re.compile(r"[A-Z]+(?![a-z])|[A-Z]?[a-z]+|[ℕℤℚℝℂ∣≃≅]") +_IDENT = re.compile(r"[A-Za-z_][A-Za-z0-9_']*") + + +def tokens(text: str) -> set[str]: + out = set() + for raw in _TOKEN.findall(text): + word = raw.lower() + out.add(word) + if word.endswith("s") and len(word) > 3: + out.add(word[:-1]) + return out + + +def concepts(text_or_words: str | list[str]) -> set[str]: + words = tokens(text_or_words) if isinstance(text_or_words, str) else { + w for item in text_or_words for w in tokens(item)} + return {_TOKEN_TO_CONCEPT[w] for w in words if w in _TOKEN_TO_CONCEPT} + + +@dataclass +class Unit: + """One thing on the Lean side a source role can be matched to.""" + + key: str + kind: str # object | fact | instance + role: Role | None + owner: str = "" + symbols: set[str] = field(default_factory=set) + concept_set: set[str] = field(default_factory=set) + + +@dataclass +class Match: + source: str + lean: str + score: float + how: str + + def as_dict(self) -> dict[str, Any]: + return {"source": self.source, "lean": self.lean, "score": round(self.score, 3), + "how": self.how} + + +@dataclass +class Conflict: + kind: str + detail: str + source: str = "" + lean: str = "" + + def as_dict(self) -> dict[str, str]: + return {"kind": self.kind, "detail": self.detail, "source": self.source, + "lean": self.lean} + + +@dataclass +class Alignment: + matches: list[Match] = field(default_factory=list) + conflicts: list[Conflict] = field(default_factory=list) + score: float = 0.0 + flag_errors: frozenset[str] = FLAG_ERRORS + + @property + def errors(self) -> list[str]: + seen: list[str] = [] + for conflict in self.conflicts: + if conflict.kind not in seen: + seen.append(conflict.kind) + return seen + + @property + def flagged(self) -> bool: + return any(c.kind in self.flag_errors for c in self.conflicts) + + def as_dict(self) -> dict[str, Any]: + return { + "score": round(self.score, 3), "flagged": self.flagged, "errors": self.errors, + "matches": [m.as_dict() for m in self.matches], + "conflicts": [c.as_dict() for c in self.conflicts], + } + + +def lean_units(projection: Projection) -> list[Unit]: + units: list[Unit] = [] + for role in projection.roles: + if role.kind == KIND_OBJECT: + classes = [str(c) for c in role.attrs.get("instances", [])] + units.append(Unit( + key=role.name, kind="object", role=role, symbols={role.name}, + concept_set=concepts(role.type_text) | concepts(" ".join(classes)))) + for text in classes: + units.append(Unit( + key=f"inst:{role.name}:{text}", kind="instance", role=None, + owner=role.name, symbols={role.name}, concept_set=concepts(text))) + elif role.kind == KIND_FACT: + units.append(Unit( + key=role.name, kind="fact", role=role, + symbols=set(_IDENT.findall(role.type_text)) | {role.name}, + concept_set=concepts(role.type_text) | _relation_concepts(role.type_text))) + return units + + +def _relation_concepts(text: str) -> set[str]: + found: set[str] = set() + if re.search(r"0\s*<|>\s*0", text): + found.add("positive") + if re.search(r"≠\s*0|0\s*≠", text): + found.add("nonzero") + if "∣" in text: + found.add("divides") + return found + + +def _score(source: SourceRole, unit: Unit) -> float: + symbol = 1.0 if (set(source.symbols) & unit.symbols) else 0.0 + wanted = concepts(source.keywords) | concepts(source.description) + if wanted and unit.concept_set: + concept = len(wanted & unit.concept_set) / min(len(wanted), len(unit.concept_set)) + else: + concept = 0.0 + return 0.6 * symbol + 0.4 * concept + + +def _compatible(source: SourceRole, unit: Unit) -> bool: + if source.kind == "object": + return unit.kind == "object" + if source.kind == "fact": + return unit.kind in ("fact", "instance") + return False + + +def _pair_deterministic( + source: SourceSimplex, units: list[Unit], threshold: float +) -> dict[str, tuple[Unit, float, str]]: + candidates = [] + for role in source.roles: + for unit in units: + if _compatible(role, unit): + score = _score(role, unit) + if score >= threshold: + candidates.append((score, role.alias, unit.key, role, unit)) + candidates.sort(key=lambda c: (-c[0], c[1], c[2])) + taken_sources: set[str] = set() + taken_units: set[str] = set() + pairs: dict[str, tuple[Unit, float, str]] = {} + for score, alias, key, _role, unit in candidates: + if alias in taken_sources: + continue + if key in taken_units and unit.kind != "instance": + continue + taken_sources.add(alias) + taken_units.add(key) + pairs[alias] = (unit, score, "symbol+concept") + return pairs + + +def _existential_names(goal_text: str) -> set[str]: + names: set[str] = set() + for match in re.finditer(r"∃([^,]*),", goal_text): + names.update(_IDENT.findall(match.group(1))) + return names + + +def align( + source: SourceSimplex, + projection: Projection, + *, + threshold: float = DEFAULT_THRESHOLD, + pairing: dict[str, str] | None = None, + flag_errors: frozenset[str] = FLAG_ERRORS, +) -> Alignment: + out = Alignment(flag_errors=flag_errors) + if projection.degenerate: + out.conflicts.append(Conflict("degenerate", "; ".join(projection.notes) or "no roles")) + units = lean_units(projection) + by_key = {u.key: u for u in units} + + if pairing is not None: + pairs = {a: (by_key[k], 1.0, "llm") for a, k in pairing.items() if k in by_key} + else: + pairs = _pair_deterministic(source, units, threshold) + + goal_role = next((r for r in projection.roles if r.kind in (KIND_GOAL, KIND_OWN)), None) + goal_text = goal_role.type_text if goal_role else "" + exist_names = _existential_names(goal_text) + + used_units = {u.key for u, _s, _h in pairs.values()} + matched_objects: dict[str, Unit] = {} + for role in source.roles: + if role.kind == "goal": + continue + hit = pairs.get(role.alias) + if hit: + unit, score, how = hit + out.matches.append(Match(role.alias, unit.key, score, how)) + if role.kind == "object": + matched_objects[role.alias] = unit + _check_object(role, unit, exist_names, out) + continue + symbols = set(role.symbols) + if role.kind == "object": + if symbols & exist_names and role.introduced == "exists": + out.matches.append(Match(role.alias, "∃-in-goal", 1.0, "existential")) + elif symbols & exist_names: + out.conflicts.append(Conflict( + "quantifier", f"{sorted(symbols & exist_names)} is ∃-bound in Lean " + f"but the source says '{role.introduced}'", role.alias)) + else: + out.conflicts.append(Conflict("missing_object", role.description, role.alias)) + else: + if _absorbed(role, units): + out.matches.append(Match(role.alias, "type", 0.5, "absorbed")) + else: + out.conflicts.append(Conflict("missing_hypothesis", role.description, role.alias)) + + for unit in units: + if unit.kind == "fact" and unit.key not in used_units: + out.conflicts.append(Conflict("extra_hypothesis", unit.role.type_text, "", unit.key)) + elif unit.kind == "object" and unit.key not in used_units: + out.conflicts.append(Conflict( + "extra_object", f"{unit.key} : {unit.role.type_text}", "", unit.key)) + + src_goal = source.goal + goal_ok = True + if src_goal is not None and goal_role is not None: + goal_symbols = set(_IDENT.findall(goal_text)) + sym = 1.0 if set(src_goal.symbols) & goal_symbols else 0.0 + wanted = concepts(src_goal.keywords) | concepts(src_goal.description) + have = concepts(goal_text) | _relation_concepts(goal_text) + con = len(wanted & have) / min(len(wanted), len(have)) if wanted and have else 0.0 + informative = bool(src_goal.symbols or wanted) + if informative and 0.5 * sym + 0.5 * con < GOAL_THRESHOLD: + goal_ok = False + out.conflicts.append(Conflict( + "goal_mismatch", f"source: {src_goal.description} | lean: {goal_text[:120]}", + src_goal.alias, goal_role.name)) + lean_exists = goal_text.count("∃") > 0 + if source.goal_has_exists != lean_exists: + out.conflicts.append(Conflict( + "quantifier", + f"source goal ∃={source.goal_has_exists}, lean goal ∃={lean_exists}", + src_goal.alias, goal_role.name)) + if goal_ok and src_goal is not None and goal_role is not None: + out.matches.append(Match(src_goal.alias, goal_role.name, 1.0, "goal")) + + n_source = len(source.roles) + n_lean = len([u for u in units if u.kind != "instance"]) + (1 if goal_role else 0) + out.score = 2 * len(out.matches) / (n_source + n_lean) if (n_source + n_lean) else 0.0 + return out + + +def _check_object(role: SourceRole, unit: Unit, exist_names: set[str], out: Alignment) -> None: + wanted = (concepts(role.keywords) | concepts(role.description)) & NUMBER_SYSTEMS + have = unit.concept_set & NUMBER_SYSTEMS + if wanted and have and wanted.isdisjoint(have): + out.conflicts.append(Conflict( + "type_conflict", f"source {sorted(wanted)} vs lean {sorted(have)}", + role.alias, unit.key)) + bound_exists = unit.key in exist_names + if role.introduced == "exists" and not bound_exists: + out.conflicts.append(Conflict( + "quantifier", f"source asserts {unit.key} exists, lean takes it as given", + role.alias, unit.key)) + + +def _absorbed(role: SourceRole, units: list[Unit]) -> bool: + wanted = concepts(role.keywords) | concepts(role.description) + if not wanted: + return False + for unit in units: + if unit.kind == "object" and wanted & unit.concept_set: + if not role.symbols or set(role.symbols) & unit.symbols: + return True + return False + + +PAIRING_PROMPT = """\ +You pair the logical parameters of a math passage with the binders of a formal +statement. For each SOURCE role, give the key of the LEAN item that expresses the +same parameter, or null if none does. Pair by meaning, not by spelling. A LEAN item can +be used once, except instance items. Reply with JSON: {"pairs": {"": +""}}.""" + + +def llm_pairing( + client: ChatClient, source: SourceSimplex, projection: Projection +) -> dict[str, str]: + """Ablation: let a model do the matching step (it sees roles, not the passage).""" + units = lean_units(projection) + listing = { + "source": [r.as_dict() for r in source.roles if r.kind != "goal"], + "lean": [ + {"key": u.key, "kind": u.kind, + "text": (u.role.type_text if u.role else u.key.split(":", 2)[-1])} + for u in units + ], + } + import json + reply = client.complete_json([ # type: ignore[attr-defined] + {"role": "system", "content": PAIRING_PROMPT}, + {"role": "user", "content": json.dumps(listing, ensure_ascii=False)}, + ]) + pairs = reply.get("pairs", {}) if isinstance(reply, dict) else {} + valid = {u.key for u in units} + return {a: k for a, k in pairs.items() if isinstance(k, str) and k in valid} diff --git a/leanbench/evalstats.py b/leanbench/evalstats.py new file mode 100644 index 0000000..f67ac1b --- /dev/null +++ b/leanbench/evalstats.py @@ -0,0 +1,93 @@ +"""Small, dependency-free statistics for evaluating a detector against human labels.""" +from __future__ import annotations + +import math +from collections.abc import Sequence +from dataclasses import dataclass + + +def wilson(successes: int, total: int, z: float = 1.96) -> tuple[float, float]: + """Wilson score interval for a proportion; (0, 1) when there is no data.""" + if total <= 0: + return 0.0, 1.0 + p = successes / total + denom = 1 + z * z / total + centre = (p + z * z / (2 * total)) / denom + half = z * math.sqrt(p * (1 - p) / total + z * z / (4 * total * total)) / denom + return max(0.0, centre - half), min(1.0, centre + half) + + +@dataclass +class Confusion: + """``positive`` means 'the item is unfaithful' and a detector 'flags' it.""" + + tp: int = 0 + fp: int = 0 + fn: int = 0 + tn: int = 0 + + @property + def n(self) -> int: + return self.tp + self.fp + self.fn + self.tn + + @property + def recall(self) -> float: + return self.tp / (self.tp + self.fn) if (self.tp + self.fn) else float("nan") + + @property + def precision(self) -> float: + return self.tp / (self.tp + self.fp) if (self.tp + self.fp) else float("nan") + + @property + def false_flag_rate(self) -> float: + return self.fp / (self.fp + self.tn) if (self.fp + self.tn) else float("nan") + + @property + def f1(self) -> float: + p, r = self.precision, self.recall + return 2 * p * r / (p + r) if (p == p and r == r and (p + r) > 0) else float("nan") + + def as_dict(self) -> dict[str, float | int]: + return { + "tp": self.tp, "fp": self.fp, "fn": self.fn, "tn": self.tn, "n": self.n, + "recall": self.recall, "precision": self.precision, + "false_flag_rate": self.false_flag_rate, "f1": self.f1, + "recall_ci": wilson(self.tp, self.tp + self.fn), + "false_flag_ci": wilson(self.fp, self.fp + self.tn), + } + + +def confusion(flagged: Sequence[bool], unfaithful: Sequence[bool]) -> Confusion: + if len(flagged) != len(unfaithful): + raise ValueError("flagged and unfaithful must have the same length") + c = Confusion() + for f, u in zip(flagged, unfaithful, strict=True): + if f and u: + c.tp += 1 + elif f and not u: + c.fp += 1 + elif u: + c.fn += 1 + else: + c.tn += 1 + return c + + +def auc(scores: Sequence[float], unfaithful: Sequence[bool]) -> float: + """ROC-AUC of 'higher score means unfaithful' (rank-sum form, ties count half).""" + pos = [s for s, u in zip(scores, unfaithful, strict=True) if u] + neg = [s for s, u in zip(scores, unfaithful, strict=True) if not u] + if not pos or not neg: + return float("nan") + wins = sum((p > n) + 0.5 * (p == n) for p in pos for n in neg) + return wins / (len(pos) * len(neg)) + + +def cohen_kappa(a: Sequence[str], b: Sequence[str]) -> float: + if len(a) != len(b) or not a: + raise ValueError("need two label lists of equal, non-zero length") + n = len(a) + observed = sum(x == y for x, y in zip(a, b, strict=True)) / n + labels = set(a) | set(b) + expected = sum((a.count(k) / n) * (b.count(k) / n) for k in labels) + return 1.0 if expected == 1 else (observed - expected) / (1 - expected) diff --git a/leanbench/lint.py b/leanbench/lint.py new file mode 100644 index 0000000..9ad844a --- /dev/null +++ b/leanbench/lint.py @@ -0,0 +1,48 @@ +"""Cheap static checks for formalizations that compile and are probably wrong. + +These are a baseline for the fidelity filter, not a replacement for one: each looks +at the projected roles and the text of the declaration only, with no model and no +Lean. They catch encoding hazards (natural-number subtraction, integer division), +binders the statement never uses, and goals that say nothing. +""" +from __future__ import annotations + +import re + +from .project import Projection +from .schema import KIND_FACT, KIND_GOAL, KIND_OBJECT + +_NAT = re.compile(r"\b(ℕ|Nat)\b") +_OTHER_NUMBERS = re.compile(r"[ℤℚℝℂ]|\b(Int|Rat|Real|Complex)\b") +_SUBTRACTION = re.compile(r"[\w)\]]\s*-\s*[\w(]") +_DIVISION = re.compile(r"[\w)\]]\s*/\s*[\w(]") +_IDENT = re.compile(r"[A-Za-z_][A-Za-z0-9_']*") + + +def lint(projection: Projection) -> list[str]: + findings: list[str] = [] + if projection.degenerate: + findings.append("degenerate") + texts = [r.type_text for r in projection.roles if r.kind in (KIND_FACT, KIND_GOAL)] + body = " ".join(texts) + declared = " ".join(r.type_text for r in projection.roles if r.kind == KIND_OBJECT) + + for role in projection.roles: + if role.kind != KIND_OBJECT or role.attrs.get("binder_kind") != "explicit": + continue + if role.type_text.strip().startswith(("Type", "Sort")): + continue # a carrier: used by the instances that constrain it + used = set(_IDENT.findall(body)) | set(_IDENT.findall(declared.replace(role.type_text, ""))) + if role.name != "_" and role.name not in used: + findings.append(f"unused_binder:{role.name}") + + nat_only = _NAT.search(declared + " " + body) and not _OTHER_NUMBERS.search(declared + body) + if nat_only and _SUBTRACTION.search(body): + findings.append("nat_subtraction") + if (_NAT.search(declared) or re.search(r"\b(ℤ|Int)\b", declared)) and _DIVISION.search(body): + findings.append("integer_division") + + goal = next((r for r in projection.roles if r.kind == KIND_GOAL), None) + if goal is not None and goal.type_text.strip() in ("", "True", "False", "1 = 1", "0 = 0"): + findings.append("trivial_goal") + return findings diff --git a/leanbench/project.py b/leanbench/project.py new file mode 100644 index 0000000..f147b2e --- /dev/null +++ b/leanbench/project.py @@ -0,0 +1,260 @@ +"""Read a Lean declaration's parameters as roles. + +``project(code)`` is deterministic and never calls Lean or a model. It splits the +signature into + +* **objects** explicit/implicit binders whose type is data or a sort: ``(G : Type*)``, + ``(n : ℕ)``; instance binders ``[Group G]`` attach to the object they + constrain (``attrs["instances"]``); +* **facts** binders whose type is a proposition: ``(h : a ∣ b)``, and, when the goal + is an implication, its premises; +* **goal** the conclusion of a theorem, or **own** for what a definition is. + +Leading ``∀`` binders of the goal are unpacked into objects, so ``(n : ℕ) : n = n`` and +``∀ n : ℕ, n = n`` project to the same roles. Whether a binder is a fact is decided by +a syntactic heuristic (relation symbols, known predicate heads, the ``h…`` naming +convention) and every role records how it was decided in ``attrs["classified_by"]``, +so the heuristic's accuracy can be measured against a manual parse. +""" +from __future__ import annotations + +import re +from dataclasses import dataclass, field +from typing import Any + +from .schema import ( + KIND_FACT, + KIND_GOAL, + KIND_OBJECT, + KIND_OWN, + Role, + Simplex, +) +from .typeindex import _DECL, _IDENT, PROOF_KEYWORDS, _split_top_level, _strip_comments + +OPENERS = {"(": ")", "{": "}", "[": "]", "⦃": "⦄", "⟨": "⟩"} +CLOSERS = set(OPENERS.values()) +BINDER_KIND = {"(": "explicit", "{": "implicit", "⦃": "strict_implicit", "[": "instance"} + +_SORT = re.compile(r"^(Type|Sort)(\b|\*|_)") +_RELATION = re.compile(r"[=≠<>≤≥∣∈∉⊆⊂⊇⊃≡∧∨¬↔∀∃≃≅≈]|\bTrue\b|\bFalse\b") +_PROP_HEAD = re.compile( + r"^(Is[A-Z]\w*|Nat\.Prime|Prime|Nat\.Coprime|Int\.Coprime|Coprime|Function\.\w+|" + r"Injective|Surjective|Bijective|Continuous\w*|Differentiable\w*|Monotone\w*|Antitone\w*|" + r"Finite|Set\.Finite|Set\.Infinite|Infinite|Odd|Even|Irreducible|Squarefree|" + r"Nonempty|Set\.Nonempty|Nat\.Even|Nat\.Odd)\b" +) +_HYPOTHESIS_NAME = re.compile(r"^h[\w'₀-₉]*$") + + +def _split_depth0(text: str, separator: str) -> list[str]: + """Split on ``separator`` outside every bracket pair.""" + parts: list[str] = [] + depth = 0 + start = 0 + for index, char in enumerate(text): + if char in OPENERS: + depth += 1 + elif char in CLOSERS: + depth = max(0, depth - 1) + elif depth == 0 and text.startswith(separator, index): + parts.append(text[start:index]) + start = index + len(separator) + parts.append(text[start:]) + return parts + + +def _first_depth0(text: str, needle: str) -> int: + depth = 0 + for index, char in enumerate(text): + if char in OPENERS: + depth += 1 + elif char in CLOSERS: + depth = max(0, depth - 1) + elif depth == 0 and text.startswith(needle, index): + if needle == ":" and text.startswith(":=", index): + continue + return index + return -1 + + +def classify(type_text: str, names: list[str], objects: set[str]) -> tuple[str, str]: + """(kind, how) for a binder with these names and this type.""" + text = type_text.strip() + if not text: + return KIND_OBJECT, "untyped" + if _SORT.match(text): + return KIND_OBJECT, "sort" + if _RELATION.search(text): + return KIND_FACT, "relation" + if _PROP_HEAD.match(text): + return KIND_FACT, "predicate" + if names and all(_HYPOTHESIS_NAME.match(n) for n in names): + tokens = {t for t in _IDENT.findall(text) if t[:1].islower()} + if tokens & objects: + return KIND_FACT, "name" + return KIND_OBJECT, "default" + + +@dataclass +class Projection: + name: str + keyword: str + roles: list[Role] = field(default_factory=list) + simplex: Simplex | None = None + notes: list[str] = field(default_factory=list) + + @property + def degenerate(self) -> bool: + return self.simplex is None + + def by_kind(self, kind: str) -> list[Role]: + return [r for r in self.roles if r.kind == kind] + + def as_dict(self) -> dict[str, Any]: + return { + "name": self.name, + "keyword": self.keyword, + "degenerate": self.degenerate, + "roles": [r.as_dict() for r in self.roles], + "notes": self.notes, + } + + +def _cut_signature(rest: str) -> str: + depth = 0 + for index, char in enumerate(rest): + if char in OPENERS: + depth += 1 + elif char in CLOSERS: + depth = max(0, depth - 1) + elif depth == 0: + if rest.startswith(":=", index) or rest.startswith("\nwhere", index): + return rest[:index] + if rest.startswith(" where", index) and index > 0: + return rest[:index] + return rest + + +def _roles_from_group(opener: str, inner: str, objects: set[str], origin: str) -> list[Role]: + binder_kind = BINDER_KIND.get(opener, "explicit") + head, _, tail = inner.partition(":") + names = head.split() + type_text = tail.strip() if ":" in inner else "" + if not names or any(not _IDENT.fullmatch(n) and n != "_" for n in names): + return [] + kind, how = classify(type_text, names, objects) + roles = [] + for name in names: + attrs = {"binder_kind": binder_kind, "classified_by": how, "origin": origin} + roles.append(Role(name, kind, type_text=type_text, attrs=attrs)) + if kind == KIND_OBJECT: + objects.add(name) + return roles + + +def _attach_instance(inner: str, roles: dict[str, Role], notes: list[str]) -> None: + text = inner.split(":", 1)[1].strip() if ":" in inner else inner.strip() + owner = None + for token in _IDENT.findall(text): + if token in roles and roles[token].kind == KIND_OBJECT: + owner = roles[token] + if owner is None: + notes.append(f"instance [{text}] constrains no named object") + return + owner.attrs.setdefault("instances", []).append(text) + + +def _unpack_forall(goal: str, objects: set[str]) -> tuple[list[Role], str]: + """Lift leading ``∀`` binders of the goal into object/fact roles.""" + lifted: list[Role] = [] + text = goal.strip() + while text.startswith("∀"): + body = text[1:].lstrip() + comma = _first_depth0(body, ",") + if comma < 0: + break + binders, text = body[:comma].strip(), body[comma + 1 :].strip() + groups = _split_top_level(binders) + if groups and binders.startswith(tuple(OPENERS)): + for opener, inner in groups: + lifted.extend(_roles_from_group(opener, inner, objects, "forall_in_goal")) + continue + bound = re.match(r"^([\w'₀-₉ ]+?)\s*(∈|∉|<|≤|>|≥|≠)\s*(.+)$", binders) + if bound and ":" not in binders: + for name in bound.group(1).split(): + lifted.append(Role(name, KIND_OBJECT, attrs={ + "binder_kind": "explicit", "classified_by": "bounded", "origin": "forall_in_goal"})) + objects.add(name) + lifted.append(Role( + f"h_{name}", KIND_FACT, type_text=f"{name} {bound.group(2)} {bound.group(3)}", + attrs={"binder_kind": "explicit", "classified_by": "bounded", + "origin": "forall_in_goal"})) + continue + lifted.extend(_roles_from_group("(", binders, objects, "forall_in_goal")) + return lifted, text + + +def project(code: str) -> Projection: + clean = _strip_comments(code) + match = _DECL.search(clean) + if not match: + return Projection("", "", notes=["no declaration found"]) + keyword, name = match.group(1), match.group(2) + result = Projection(name, keyword) + signature = _cut_signature(clean[match.end() :]) + + colon = _first_depth0(signature, ":") + binder_text = signature if colon < 0 else signature[:colon] + goal_text = "" if colon < 0 else signature[colon + 1 :].strip() + + objects: set[str] = set() + named: dict[str, Role] = {} + instances: list[str] = [] + for opener, inner in _split_top_level(binder_text): + if opener == "[": + instances.append(inner) + continue + if opener == "⟨": + continue + for role in _roles_from_group(opener, inner, objects, "binder"): + result.roles.append(role) + named[role.name] = role + for inner in instances: + _attach_instance(inner, named, result.notes) + + if keyword in PROOF_KEYWORDS: + lifted, goal_text = _unpack_forall(goal_text, objects) + for role in lifted: + result.roles.append(role) + named[role.name] = role + parts = _split_depth0(goal_text, "→") if goal_text else [] + if len(parts) >= 2: + for index, premise in enumerate(parts[:-1], start=1): + result.roles.append(Role( + f"premise{index}", KIND_FACT, type_text=premise.strip(), + attrs={"binder_kind": "goal_implication", "classified_by": "implication", + "origin": "goal"})) + goal_text = parts[-1].strip() + result.roles.append(Role( + "goal", KIND_GOAL, type_text=goal_text, + attrs={ + "exists_count": goal_text.count("∃"), + "forall_count": goal_text.count("∀"), + "is_iff": "↔" in goal_text, + })) + else: + result.roles.append(Role( + "own", KIND_OWN, type_text=goal_text or keyword, attrs={"keyword": keyword})) + + seen: dict[str, int] = {} + aliased: dict[str, Role] = {} + for role in result.roles: + count = seen.get(role.name, 0) + 1 + seen[role.name] = count + aliased[role.name if count == 1 else f"{role.name}_{count}"] = role + if len(aliased) >= 2: + result.simplex = Simplex(name, aliased) + else: + result.notes.append(f"degenerate: only {len(aliased)} role(s)") + return result diff --git a/leanbench/schema.py b/leanbench/schema.py new file mode 100644 index 0000000..b59f667 --- /dev/null +++ b/leanbench/schema.py @@ -0,0 +1,135 @@ +"""Roles and simplices: the parameters of a declaration as a small typed schema. + +A Lean declaration is a bundle of roles: the objects it is about (``(G : Type*)``), +the facts it assumes (``(h : a ∣ b)``), and the goal it asserts or the thing it +defines. A ``Simplex`` is such a bundle with at least two roles. The same schema can +be read off the prose of a book, so the two can be compared role by role. + +Design notes (from the archived ``schema-simplex`` prototype): + +* a role may be *constrained* to a closed set of allowed values (``choices``); +* a simplex is itself a role, so simplices nest (a lemma used by another is a role + of it), which is how a dependency structure appears; +* validation lives in ``__init__``. The prototype put it in a dataclass + ``__post_init__`` next to a hand-written ``__init__``, where it never ran. +""" +from __future__ import annotations + +from collections.abc import Sequence +from typing import Any + +KIND_OBJECT = "object" # something the statement is about: a group G, an integer n +KIND_FACT = "fact" # something assumed: n > 0, G is finite +KIND_GOAL = "goal" # what is asserted, or asked to be shown +KIND_OWN = "own" # what a definition is +KIND_SIMPLEX = "simplex" +KINDS = (KIND_OBJECT, KIND_FACT, KIND_GOAL, KIND_OWN, KIND_SIMPLEX) + +MIN_ROLES = 2 + + +class Role: + """A named slot with a kind, a description and optionally a closed value set.""" + + def __init__( + self, + name: str, + kind: str = KIND_OBJECT, + *, + type_text: str = "", + context: str = "", + choices: Sequence[Any] | None = None, + attrs: dict[str, Any] | None = None, + ) -> None: + if not name: + raise ValueError("a role needs a name") + if kind not in KINDS: + raise ValueError(f"unknown role kind {kind!r}; expected one of {KINDS}") + self.name = name + self.kind = kind + self.type_text = type_text + self.context = context + self.choices: list[Any] | None = list(choices) if choices is not None else None + self.attrs: dict[str, Any] = dict(attrs or {}) + + def is_constrained(self) -> bool: + return self.choices is not None + + def validate_value(self, value: Any) -> bool: + return self.choices is None or value in self.choices + + def check_value(self, value: Any) -> None: + if not self.validate_value(value): + raise ValueError( + f"{value!r} is not allowed for role {self.name!r}; allowed: {self.choices}" + ) + + def as_dict(self) -> dict[str, Any]: + out: dict[str, Any] = {"name": self.name, "kind": self.kind} + if self.type_text: + out["type"] = self.type_text + if self.context: + out["context"] = self.context + if self.choices is not None: + out["choices"] = self.choices + if self.attrs: + out["attrs"] = self.attrs + return out + + def __repr__(self) -> str: + return f"" + + +class Simplex(Role): + """A reusable bundle of at least two roles, keyed by alias. Itself a role.""" + + def __init__(self, name: str, roles: dict[str, Role], *, context: str = "") -> None: + super().__init__(name, KIND_SIMPLEX, context=context) + if len(roles) < MIN_ROLES: + raise ValueError( + f"simplex {name!r} needs at least {MIN_ROLES} roles, got {len(roles)}" + ) + self.roles = dict(roles) + + def by_kind(self, kind: str) -> list[Role]: + return [role for role in self.roles.values() if role.kind == kind] + + def walk(self) -> list[Role]: + """Every non-simplex role, descending into nested simplices.""" + out: list[Role] = [] + for role in self.roles.values(): + if isinstance(role, Simplex): + out.extend(role.walk()) + else: + out.append(role) + return out + + def as_dict(self) -> dict[str, Any]: + return { + "name": self.name, + "roles": {alias: role.as_dict() for alias, role in self.roles.items()}, + } + + +class State: + """A role together with a value; unresolved until it has one.""" + + def __init__(self, role: Role, value: Any = None) -> None: + self.role = role + self.value = value + + def is_resolved(self) -> bool: + return self.value is not None + + +class Fact(State): + """Known at the start.""" + + def __init__(self, role: Role, value: Any) -> None: + if value is None: + raise ValueError("a Fact must carry a value") + super().__init__(role, value) + + +class Goal(State): + """To be resolved.""" diff --git a/leanbench/srcsimplex.py b/leanbench/srcsimplex.py new file mode 100644 index 0000000..820f464 --- /dev/null +++ b/leanbench/srcsimplex.py @@ -0,0 +1,165 @@ +"""The source-side simplex: what the *prose* gives, assumes and claims. + +One model call reads only the book passage -- never the Lean -- and lists its roles. +Keeping the Lean out of view is the point: the schema must come from the source, so +that comparing it with the Lean-side schema is a check and not an echo. +""" +from __future__ import annotations + +import re +from dataclasses import dataclass, field +from typing import Any + +from .llm import ChatClient, parse_json + +PROMPT_VERSION = "srcsimplex-v1" + +INTRODUCED = ("given", "forall", "exists") + +SYSTEM_PROMPT = """\ +You read ONE passage from a mathematics book and describe its logical parameters. +Do not write Lean. Do not solve anything. Describe only what the passage itself says. + +Return a JSON object: +{ + "roles": [ + {"kind": "object", "symbols": ["G"], "description": "finite group", + "keywords": ["finite", "group"], "introduced": "given"}, + {"kind": "fact", "symbols": ["n"], "description": "n is positive", + "keywords": ["positive"], "introduced": "given"}, + {"kind": "goal", "symbols": [], "description": "every element's order divides |G|", + "keywords": ["order", "divides"], "introduced": "given"} + ], + "goal_has_exists": false +} + +Rules: +- "object": a mathematical object the passage is about (a group G, an integer n, a + function f). One entry per object, even if several share a description. +- "fact": a stated assumption or condition on those objects ("n > 0", "G is + finite", "a and b are coprime"). Include conditions hidden in prose ("let G be a + finite group" gives an object G AND a fact that G is finite). +- "goal": exactly one entry: the claim being asserted, the property being defined, or + for an exercise the thing to be shown. For a definition, the goal is what is defined. +- "symbols": the letters/names used in the text, as written. Empty if none. +- "keywords": 1-5 lowercase words naming the concepts involved (e.g. "integer", + "prime", "subgroup", "homomorphism", "finite"). Words from the text, not synonyms. +- "introduced" (objects only): "given" if the passage fixes it ("let n be..."), + "forall" if it ranges over all of them ("for every n"), "exists" if the passage asserts + one exists ("there is an n such that"). Use "given" when unsure. +- "goal_has_exists": true only if the claim itself asserts that something exists. +- At least two roles in total, and exactly one goal. +Reply with the JSON object only.""" + + +@dataclass +class SourceRole: + kind: str + symbols: list[str] = field(default_factory=list) + description: str = "" + keywords: list[str] = field(default_factory=list) + introduced: str = "given" + alias: str = "" + + def as_dict(self) -> dict[str, Any]: + return { + "alias": self.alias, "kind": self.kind, "symbols": self.symbols, + "description": self.description, "keywords": self.keywords, + "introduced": self.introduced, + } + + +@dataclass +class SourceSimplex: + roles: list[SourceRole] + goal_has_exists: bool = False + notes: list[str] = field(default_factory=list) + + def by_kind(self, kind: str) -> list[SourceRole]: + return [r for r in self.roles if r.kind == kind] + + @property + def goal(self) -> SourceRole | None: + goals = self.by_kind("goal") + return goals[0] if goals else None + + def as_dict(self) -> dict[str, Any]: + return { + "roles": [r.as_dict() for r in self.roles], + "goal_has_exists": self.goal_has_exists, + "notes": self.notes, + } + + +def _strings(value: Any) -> list[str]: + if isinstance(value, str): + value = [value] + if not isinstance(value, list): + return [] + return [str(v).strip() for v in value if str(v).strip()] + + +def parse_source_simplex(payload: Any) -> SourceSimplex: + """Validate a model reply; raises ``ValueError`` when it is not a usable schema.""" + if not isinstance(payload, dict) or not isinstance(payload.get("roles"), list): + raise ValueError("reply has no 'roles' list") + roles: list[SourceRole] = [] + counters: dict[str, int] = {} + notes: list[str] = [] + for item in payload["roles"]: + if not isinstance(item, dict): + continue + kind = str(item.get("kind", "")).strip().lower() + if kind not in ("object", "fact", "goal"): + notes.append(f"dropped role with kind {kind!r}") + continue + introduced = str(item.get("introduced", "given")).strip().lower() + counters[kind] = counters.get(kind, 0) + 1 + roles.append(SourceRole( + kind=kind, + symbols=_strings(item.get("symbols")), + description=str(item.get("description", "")).strip(), + keywords=[k.lower() for k in _strings(item.get("keywords"))], + introduced=introduced if introduced in INTRODUCED else "given", + alias=f"{kind}{counters[kind]}", + )) + goals = [r for r in roles if r.kind == "goal"] + if len(goals) != 1: + raise ValueError(f"expected exactly one goal role, got {len(goals)}") + if len(roles) < 2: + raise ValueError("a simplex needs at least two roles") + return SourceSimplex(roles, bool(payload.get("goal_has_exists")), notes) + + +def passage_prompt(text: str, label: str = "") -> list[dict[str, str]]: + header = f"Label: {label}\n\n" if label else "" + body = re.sub(r"[ \t]+", " ", text).strip() + return [ + {"role": "system", "content": SYSTEM_PROMPT}, + {"role": "user", "content": f"{header}Passage:\n<<>>\n{body}\n<<>>"}, + ] + + +def extract_source_simplex( + client: ChatClient, text: str, label: str = "", *, attempts: int = 2 +) -> SourceSimplex: + """Ask the model for the passage's roles; retry once if the reply is unusable.""" + messages = passage_prompt(text, label) + last: ValueError | None = None + for _ in range(attempts): + payload = client.complete_json(messages) # type: ignore[attr-defined] + try: + return parse_source_simplex(payload) + except ValueError as error: + last = error + messages = messages + [ + {"role": "assistant", "content": str(payload)[:3000]}, + {"role": "user", "content": f"Not usable: {error}. Reply with the JSON only."}, + ] + raise ValueError(f"no usable source simplex: {last}") + + +__all__ = [ + "PROMPT_VERSION", "SYSTEM_PROMPT", "SourceRole", "SourceSimplex", + "extract_source_simplex", "parse_json", "parse_source_simplex", "passage_prompt", +] diff --git a/tests/test_align.py b/tests/test_align.py new file mode 100644 index 0000000..d6288e7 --- /dev/null +++ b/tests/test_align.py @@ -0,0 +1,148 @@ +import json + +from leanbench.align import align, llm_pairing +from leanbench.llm import ScriptedClient +from leanbench.project import project +from leanbench.srcsimplex import SourceRole, SourceSimplex + + +def src(*roles, exists=False): + out = [] + counts: dict[str, int] = {} + for kind, symbols, desc, keywords, *rest in roles: + counts[kind] = counts.get(kind, 0) + 1 + out.append(SourceRole( + kind, symbols, desc, keywords, rest[0] if rest else "given", f"{kind}{counts[kind]}")) + return SourceSimplex(out, exists) + + +GROUP = project( + "theorem t (G : Type*) [Group G] [Fintype G] (g : G) : g ^ Fintype.card G = 1 := by sorry") +GROUP_SRC = src( + ("object", ["G"], "finite group", ["finite", "group"]), + ("object", ["g"], "element of G", []), + ("fact", ["G"], "G is finite", ["finite"]), + ("goal", ["g", "G"], "g raised to the size of G is the identity", ["order"]), +) + + +def test_faithful_statement_is_not_flagged(): + result = align(GROUP_SRC, GROUP) + assert not result.flagged, result.as_dict() + assert result.score > 0.8 + assert {m.source for m in result.matches} >= {"object1", "object2", "fact1", "goal1"} + + +def test_missing_hypothesis_is_named(): + lean = project("theorem t (n : ℕ) (a : ℤ) : a ^ n = a ^ n := by rfl") + source = src( + ("object", ["n"], "positive integer", ["integer"]), + ("object", ["a"], "integer", ["integer"]), + ("fact", ["a", "n"], "a and n are coprime", ["coprime"]), + ("goal", ["a", "n"], "a to the n", []), + ) + result = align(source, lean) + assert "missing_hypothesis" in result.errors + assert result.flagged + + +def test_extra_hypothesis_is_named(): + lean = project("theorem t (n : ℤ) (h : 0 < n) : n = n := by rfl") + source = src( + ("object", ["n"], "integer", ["integer"]), + ("goal", ["n"], "n equals itself", []), + ) + result = align(source, lean) + assert "extra_hypothesis" in result.errors and result.flagged + + +def test_changed_number_system_is_a_type_conflict(): + lean = project("theorem t (n : ℕ) : n = n := by rfl") + source = src( + ("object", ["n"], "an integer", ["integer"]), + ("goal", ["n"], "n equals itself", []), + ) + result = align(source, lean) + assert [c.kind for c in result.conflicts if c.kind == "type_conflict"] == ["type_conflict"] + + +def test_exists_turned_into_given_is_a_quantifier_error(): + lean = project("theorem t (m : ℕ) : m = m := by rfl") + source = src( + ("object", ["m"], "a natural number", ["natural"], "exists"), + ("goal", ["m"], "there is such an m", [], "given"), + ) + result = align(source, lean) + assert "quantifier" in result.errors + + +def test_existential_in_goal_is_a_clean_match(): + lean = project("theorem t (n : ℕ) : ∃ m : ℕ, m > n := by sorry") + source = src( + ("object", ["n"], "natural number", ["natural"]), + ("object", ["m"], "a larger natural number", ["natural"], "exists"), + ("goal", ["m", "n"], "there is a larger number", []), + exists=True, + ) + result = align(source, lean) + assert "quantifier" not in result.errors + assert "missing_object" not in result.errors + + +def test_goal_exists_mismatch(): + lean = project("theorem t (n : ℕ) (a : ℕ) : n = a := by sorry") + source = src( + ("object", ["n"], "natural number", ["natural"]), + ("object", ["a"], "natural number", ["natural"]), + ("goal", ["n", "a"], "there exist", []), + exists=True, + ) + assert "quantifier" in align(source, lean).errors + + +def test_wrong_goal_is_flagged(): + lean = project("theorem t (G : Type*) [Group G] (g : G) : g * g = g := by sorry") + source = src( + ("object", ["G"], "group", ["group"]), + ("object", ["g"], "element", []), + ("goal", ["x"], "the subgroup is normal", ["normal", "subgroup"]), + ) + assert "goal_mismatch" in align(source, lean).errors + + +def test_degenerate_lean_is_reported(): + lean = project("theorem t : 2 + 2 = 4 := by norm_num") + source = src(("object", ["n"], "integer", ["integer"]), ("goal", ["n"], "n", [])) + assert "degenerate" in align(source, lean).errors + + +def test_extra_object_is_reported_but_not_flagged_alone(): + lean = project("theorem t (n : ℕ) (k : ℕ) : n = n := by rfl") + source = src(("object", ["n"], "natural", ["natural"]), ("goal", ["n"], "n = n", [])) + result = align(source, lean) + assert "extra_object" in result.errors and not result.flagged + + +def test_pairing_overrides_the_deterministic_matcher(): + lean = project("theorem t (x : ℤ) (hx : 0 < x) : x = x := by rfl") + source = src( + ("object", ["z"], "integer", ["integer"]), + ("fact", [], "it is positive", []), + ("goal", [], "trivial", []), + ) + pairing = {"object1": "x", "fact1": "hx"} + result = align(source, lean, pairing=pairing) + assert {(m.source, m.lean) for m in result.matches} >= {("object1", "x"), ("fact1", "hx")} + assert "missing_hypothesis" not in result.errors + + +def test_llm_pairing_keeps_only_real_keys(): + client = ScriptedClient([json.dumps({"pairs": {"object1": "x", "fact1": "nonsense"}})]) + lean = project("theorem t (x : ℤ) (hx : 0 < x) : x = x := by rfl") + source = src(("object", ["z"], "integer", []), ("fact", [], "positive", []), + ("goal", [], "", [])) + assert llm_pairing(client, source, lean) == {"object1": "x"} + + +def test_alignment_dict_is_json_serialisable(): + json.dumps(align(GROUP_SRC, GROUP).as_dict()) diff --git a/tests/test_evalstats.py b/tests/test_evalstats.py new file mode 100644 index 0000000..3478315 --- /dev/null +++ b/tests/test_evalstats.py @@ -0,0 +1,44 @@ +import math + +import pytest + +from leanbench.evalstats import auc, cohen_kappa, confusion, wilson + + +def test_wilson_matches_known_values(): + lo, hi = wilson(5, 10) + assert math.isclose(lo, 0.2366, abs_tol=1e-3) and math.isclose(hi, 0.7634, abs_tol=1e-3) + assert wilson(0, 0)[0] == 0.0 and wilson(0, 0)[1] == 1.0 + assert wilson(10, 10)[1] == 1.0 + + +def test_confusion_counts_and_rates(): + c = confusion([True, True, False, False, True], [True, False, True, False, True]) + assert (c.tp, c.fp, c.fn, c.tn) == (2, 1, 1, 1) + assert math.isclose(c.recall, 2 / 3) and math.isclose(c.precision, 2 / 3) + assert math.isclose(c.false_flag_rate, 0.5) + assert c.as_dict()["n"] == 5 + + +def test_confusion_length_mismatch(): + with pytest.raises(ValueError): + confusion([True], [True, False]) + + +def test_empty_classes_give_nan_not_a_crash(): + c = confusion([False, False], [False, False]) + assert math.isnan(c.recall) and math.isnan(c.precision) + + +def test_auc(): + assert auc([0.9, 0.8, 0.2, 0.1], [True, True, False, False]) == 1.0 + assert auc([0.1, 0.2, 0.8, 0.9], [True, True, False, False]) == 0.0 + assert auc([0.5, 0.5], [True, False]) == 0.5 + assert math.isnan(auc([0.5], [True])) + + +def test_kappa(): + assert cohen_kappa(["a", "b", "a"], ["a", "b", "a"]) == 1.0 + assert math.isclose(cohen_kappa(["a", "a", "b", "b"], ["a", "b", "a", "b"]), 0.0) + with pytest.raises(ValueError): + cohen_kappa([], []) diff --git a/tests/test_lint.py b/tests/test_lint.py new file mode 100644 index 0000000..d5702be --- /dev/null +++ b/tests/test_lint.py @@ -0,0 +1,34 @@ +from leanbench.lint import lint +from leanbench.project import project + + +def found(code): + return lint(project(code)) + + +def test_clean_statement_has_no_findings(): + code = "theorem t (G : Type*) [Group G] (g : G) : g * 1 = g := by sorry" + assert found(code) == [] + + +def test_unused_binder(): + assert "unused_binder:k" in found("theorem t (n k : ℕ) : n = n := by rfl") + + +def test_carrier_type_is_not_flagged_as_unused(): + assert not any(f.startswith("unused") for f in found( + "theorem t (G : Type*) [Group G] (g : G) : g = g := by rfl")) + + +def test_natural_subtraction_hazard(): + assert "nat_subtraction" in found("theorem t (a b : ℕ) : a - b + b = a := by sorry") + assert "nat_subtraction" not in found("theorem t (a b : ℤ) : a - b + b = a := by ring") + + +def test_integer_division_hazard(): + assert "integer_division" in found("theorem t (a : ℕ) : a / 2 * 2 = a := by sorry") + + +def test_trivial_goal_and_degenerate(): + assert "trivial_goal" in found("theorem t (n : ℕ) : True := by trivial") + assert "degenerate" in found("theorem t : 2 + 2 = 4 := by norm_num") diff --git a/tests/test_pilot_scripts.py b/tests/test_pilot_scripts.py new file mode 100644 index 0000000..58fcd48 --- /dev/null +++ b/tests/test_pilot_scripts.py @@ -0,0 +1,38 @@ +"""Offline smoke test of the pilot scripts with a stub model (no network).""" +import importlib.util +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] / "experiments" / "pilot" + + +def load(name): + spec = importlib.util.spec_from_file_location(name, ROOT / f"{name}.py") + mod = importlib.util.module_from_spec(spec) + spec.loader.exec_module(mod) + return mod + + +class Stub: + def complete_json(self, messages, **kw): + if "roles" in messages[0]["content"] or "passage" in messages[0]["content"].lower() and "faithful" not in messages[0]["content"]: + return {"roles": [ + {"kind": "object", "symbols": ["G"], "description": "finite group", "keywords": ["group", "finite"]}, + {"kind": "goal", "symbols": [], "description": "g^n = e", "keywords": ["identity"]}, + ]} + return {"faithful": True, "confidence": 0.9, "reason": "ok"} + + +ITEM = {"id": "x1", "source_text": "Let G be a finite group. Then g^n = e.", "label": "t", + "lean_code": "theorem t (G : Type*) [Group G] [Fintype G] (g : G) : g ^ Fintype.card G = 1 := sorry", + "verified": True} + + +def test_run_item_and_score(tmp_path): + rp, sc = load("run_pilot"), load("score") + row = rp.run_item(Stub(), ITEM) + assert row["id"] == "x1" and row["compiled"] and "judge" in row + assert set(sc.flags(row)) == {"compile_only", "lint", "role", "judge"} + rows = [dict(row, verdict=v) for v in ("wrong", "faithful", "faithful", "wrong")] + out = sc.report(rows, lambda r: r["verdict"] == "wrong") + assert out["judge"]["n"] == 4 + assert sc.decide(out) in {"GO", "PARTIAL", "NO-GO"} diff --git a/tests/test_project.py b/tests/test_project.py new file mode 100644 index 0000000..f74286a --- /dev/null +++ b/tests/test_project.py @@ -0,0 +1,94 @@ +from leanbench.project import classify, project +from leanbench.schema import KIND_FACT, KIND_GOAL, KIND_OBJECT, KIND_OWN + + +def roles(code): + return {r.name: r for r in project(code).roles} + + +def test_objects_instances_and_goal(): + p = project( + "theorem t (G : Type*) [Group G] [Fintype G] (g : G) : g ^ Fintype.card G = 1 := by sorry" + ) + got = {r.name: r for r in p.roles} + assert list(got) == ["G", "g", "goal"] + assert got["G"].kind == KIND_OBJECT + assert got["G"].attrs["instances"] == ["Group G", "Fintype G"] + assert got["goal"].kind == KIND_GOAL + assert not p.degenerate + assert p.simplex is not None and len(p.simplex.roles) == 3 + + +def test_hypothesis_binders_become_facts(): + got = roles("theorem t (n : ℕ) (a : ℤ) (h : Nat.Coprime a n) : a ≡ 1 [ZMOD n] := by sorry") + assert got["h"].kind == KIND_FACT + assert got["h"].attrs["classified_by"] == "predicate" + assert got["n"].kind == KIND_OBJECT + + +def test_relation_symbols_mark_a_fact(): + got = roles("theorem t (a b : ℤ) (hab : a ∣ b) (hpos : 0 < a) : a ≤ b := by sorry") + assert got["hab"].kind == KIND_FACT and got["hpos"].kind == KIND_FACT + assert got["a"].kind == KIND_OBJECT + + +def test_function_type_is_data_not_a_fact(): + kind, how = classify("ℕ → ℕ", ["f"], set()) + assert kind == KIND_OBJECT and how == "default" + assert classify("Type*", ["G"], set()) == (KIND_OBJECT, "sort") + + +def test_leading_forall_is_lifted_so_both_spellings_agree(): + a = project("theorem t (n : ℕ) : n = n := by rfl") + b = project("theorem t : ∀ n : ℕ, n = n := by rfl") + assert [(r.name, r.kind) for r in a.roles] == [(r.name, r.kind) for r in b.roles] + assert {r.attrs.get("origin") for r in b.roles if r.name == "n"} == {"forall_in_goal"} + + +def test_goal_implication_splits_into_premises(): + got = roles("theorem t : ∀ n m : ℕ, n < m → m < 10 → n < 10 := by omega") + assert [n for n in got if n.startswith("premise")] == ["premise1", "premise2"] + assert got["goal"].type_text == "n < 10" + + +def test_bounded_forall_adds_an_object_and_a_fact(): + got = roles("theorem t : ∀ x ∈ S, f x = 0 := by sorry") + assert got["x"].kind == KIND_OBJECT and got["h_x"].kind == KIND_FACT + + +def test_definition_has_an_own_role(): + p = project("def divides (a b : Int) : Prop := ∃ k, b = a * k") + assert p.by_kind(KIND_OWN)[0].type_text == "Prop" + assert [r.name for r in p.roles] == ["a", "b", "own"] + + +def test_structure_stops_at_where_and_ignores_extends(): + p = project("structure Ring (R : Type*) extends AddCommGroup R, Mul R where\n x : R") + assert [r.name for r in p.roles] == ["R", "own"] + + +def test_bare_statement_is_degenerate_not_an_error(): + p = project("theorem t : 2 + 2 = 4 := by norm_num") + assert p.degenerate and p.simplex is None + assert [r.name for r in p.roles] == ["goal"] + + +def test_no_declaration(): + p = project("-- just a comment") + assert p.degenerate and p.notes == ["no declaration found"] + + +def test_instance_without_target_is_noted(): + p = project("theorem t (n : ℕ) [Fact (1 < 2)] : n = n := by rfl") + assert any("constrains no named object" in n for n in p.notes) + + +def test_duplicate_names_are_aliased(): + p = project("theorem t (h : 0 < 1) (h : 1 < 2) : True := by trivial") + assert p.simplex is not None + assert "h" in p.simplex.roles and "h_2" in p.simplex.roles + + +def test_comments_and_proof_body_are_ignored(): + code = "/- doc -/\ntheorem t (n : ℕ) : n = n := by\n -- (x : T) here is noise\n rfl" + assert [r.name for r in project(code).roles] == ["n", "goal"] diff --git a/tests/test_schema.py b/tests/test_schema.py new file mode 100644 index 0000000..c64ad0f --- /dev/null +++ b/tests/test_schema.py @@ -0,0 +1,49 @@ +import pytest + +from leanbench.schema import KIND_FACT, KIND_OBJECT, Fact, Goal, Role, Simplex + + +def test_simplex_needs_two_roles(): + only = Role("n", KIND_OBJECT) + with pytest.raises(ValueError, match="at least 2"): + Simplex("solo", {"n": only}) + two = Simplex("pair", {"n": only, "h": Role("h", KIND_FACT)}) + assert [r.name for r in two.by_kind(KIND_FACT)] == ["h"] + + +def test_validation_runs_in_init_not_a_dead_post_init(): + # The archived prototype defined __post_init__ beside a hand-written __init__, + # so its minimum-role check never executed. Guard against that regression. + with pytest.raises(ValueError): + Simplex("empty", {}) + + +def test_constrained_role_rejects_values_outside_choices(): + role = Role("parity", KIND_OBJECT, choices=["even", "odd"]) + assert role.is_constrained() + assert role.validate_value("even") + with pytest.raises(ValueError, match="not allowed"): + role.check_value("prime") + assert Role("free", KIND_OBJECT).validate_value(object()) + + +def test_simplices_nest_and_walk_flattens(): + inner = Simplex("inner", {"a": Role("a"), "b": Role("b")}) + outer = Simplex("outer", {"inner": inner, "c": Role("c")}) + assert sorted(r.name for r in outer.walk()) == ["a", "b", "c"] + assert outer.as_dict()["roles"]["inner"]["roles"]["a"]["name"] == "a" + + +def test_unknown_kind_and_empty_name_are_rejected(): + with pytest.raises(ValueError): + Role("x", "banana") + with pytest.raises(ValueError): + Role("") + + +def test_states(): + role = Role("n") + assert not Goal(role).is_resolved() + assert Fact(role, 3).is_resolved() + with pytest.raises(ValueError): + Fact(role, None) diff --git a/tests/test_srcsimplex.py b/tests/test_srcsimplex.py new file mode 100644 index 0000000..3dcd553 --- /dev/null +++ b/tests/test_srcsimplex.py @@ -0,0 +1,72 @@ +import json + +import pytest + +from leanbench.llm import ScriptedClient +from leanbench.srcsimplex import ( + SYSTEM_PROMPT, + extract_source_simplex, + parse_source_simplex, + passage_prompt, +) + +GOOD = { + "roles": [ + {"kind": "object", "symbols": ["G"], "description": "finite group", + "keywords": ["Finite", "group"], "introduced": "given"}, + {"kind": "goal", "symbols": [], "description": "g^|G| = e", "keywords": []}, + ], + "goal_has_exists": False, +} + + +def test_parse_lowercases_keywords_and_numbers_aliases(): + s = parse_source_simplex(GOOD) + assert s.roles[0].keywords == ["finite", "group"] + assert [r.alias for r in s.roles] == ["object1", "goal1"] + assert s.goal is not None and s.goal.kind == "goal" + + +@pytest.mark.parametrize("payload", [ + {"roles": []}, + {"roles": [{"kind": "goal"}]}, + {"roles": [{"kind": "object"}, {"kind": "object"}]}, + {"roles": [{"kind": "goal"}, {"kind": "goal"}]}, + ["not", "a", "dict"], +]) +def test_unusable_replies_are_rejected(payload): + with pytest.raises(ValueError): + parse_source_simplex(payload) + + +def test_unknown_kinds_are_dropped_with_a_note(): + payload = {"roles": GOOD["roles"] + [{"kind": "banana"}]} + s = parse_source_simplex(payload) + assert len(s.roles) == 2 and s.notes + + +def test_bad_introduced_defaults_to_given(): + payload = {"roles": [ + {"kind": "object", "introduced": "sometimes"}, {"kind": "goal"}]} + assert parse_source_simplex(payload).roles[0].introduced == "given" + + +def test_extract_retries_once_then_succeeds(): + client = ScriptedClient([json.dumps({"roles": []}), json.dumps(GOOD)]) + s = extract_source_simplex(client, "Let G be a finite group.", "Prop 1") + assert s.goal is not None + assert len(client.calls) == 2 + + +def test_extract_gives_up_loudly(): + client = ScriptedClient([json.dumps({"roles": []})] * 2) + with pytest.raises(ValueError, match="no usable source simplex"): + extract_source_simplex(client, "x") + + +def test_prompt_never_shows_lean(): + messages = passage_prompt("Let G be a group.\n", "Theorem 2") + assert "Lean" in SYSTEM_PROMPT # only to forbid writing it + assert "Do not write Lean" in SYSTEM_PROMPT + assert "Label: Theorem 2" in messages[1]["content"] + assert "Let G be a group." in messages[1]["content"]