| |
| """Elaborated-type hash of formal statements (plan v3 Phase 0a item 5, first bullet). |
| |
| Why: token/alpha hashes (p2_common.statement_sha / alpha_sha) are blind to hypothesis order, notation |
| (`x ∈ Finset.range n` vs `x < n`-style rewrites that elaborate to the same Expr), implicit coercion spelling, and |
| `open` differences. The elaborated type of `theorem _align_ref <sig> := by sorry` in the pinned Lean env is |
| not. Expr uses de Bruijn indices for bound variables, so binder names never reach the hash. |
| |
| What is hashed (all decided in Lean, `LEAN_DUMP`, then canonicalised and hashed in pure Python): |
| * the statement type is split with `forallTelescope` into binders and a conclusion; |
| * instance-implicit binders are DROPPED, and so are instance-implicit ARGUMENTS of every application |
| (`getFunInfo` paramInfo), so `@HAdd.hAdd ℝ ℝ ℝ inst a b` is `HAdd.hAdd ℝ ℝ ℝ a b`; |
| * data binders (non-Prop) keep their order (later types may depend on earlier ones) and are named by that |
| ordinal `v0, v1, ...`; Prop binders (hypotheses) are serialised with the variables as `v_k`, and the |
| hypothesis list is SORTED, so reordering hypotheses does not change the hash; |
| * universe levels are dropped; binder names/infos inside terms are dropped; `mdata` is stripped; numerals |
| keep their type argument (`OfNat.ofNat ℝ 2` differs from `OfNat.ofNat ℕ 2`). |
| Known blind spots (by design, documented): permuting the order of data variables that are interchangeable is not |
| canonicalised; hypotheses that mention other hypotheses (proof-dependent types) are serialised with `?h`. |
| |
| # unit-testable pure parts: parse_dump, canonical_form, elab_sha, build_lean_prelude |
| # <=50 statements validation (2 Lean workers max): |
| .venv/bin/python scripts/p2/supply/elab_hash.py --in .tmp/p2/problems/minif2f.jsonl --out .tmp/p2/supply/elab_minif2f.jsonl --limit 50 --workers 2 |
| # FULL RUN (GPU-free window only; ~0.5-3 s/statement/worker after header warm-up; resumable by --out): |
| .venv/bin/python scripts/p2/supply/elab_hash.py --in <set>.clean.jsonl --out <set>.elab.jsonl --workers 2 |
| """ |
|
|
| from __future__ import annotations |
|
|
| import argparse |
| import hashlib |
| import importlib.util |
| import json |
| import os |
| import re |
| import sys |
| import time |
| from pathlib import Path |
| from typing import Any, Iterable, Optional |
|
|
| REPO = Path(__file__).resolve().parents[3] |
| sys.path.insert(0, str(REPO)) |
|
|
| MARK = "ALIGN_ELAB_HASH " |
| NS = "AlignElabHash" |
| REF = "_align_elab_ref" |
|
|
| |
| LEAN_DUMP = r''' |
| namespace AlignElabHash |
| open Lean Meta |
| |
| partial def ser (nm : FVarId → Option String) (e : Expr) : MetaM String := do |
| match e with |
| | .bvar i => return s!"#{i}" |
| | .fvar id => return (nm id).getD "?fv" |
| | .mvar _ => return "?m" |
| | .sort l => return s!"Sort({toString l})" |
| | .const n _ => return n.toString |
| | .lit (.natVal n) => return toString n |
| | .lit (.strVal s) => return s!"\"{s}\"" |
| | .mdata _ b => ser nm b |
| | .proj s i b => return s!"{s}.{i}({← ser nm b})" |
| | .lam _ t b _ => return s!"(fun {← ser nm t} => {← ser nm b})" |
| | .forallE _ t b _ => return s!"(all {← ser nm t} , {← ser nm b})" |
| | .letE _ t v b _ => return s!"(let {← ser nm t} := {← ser nm v} in {← ser nm b})" |
| | .app .. => |
| let f := e.getAppFn |
| let args := e.getAppArgs |
| let mut keep : Array Bool := args.map fun _ => true |
| try |
| let info ← getFunInfoNArgs f args.size |
| for i in [0:args.size] do |
| if i < info.paramInfo.size && info.paramInfo[i]!.isInstImplicit then |
| keep := keep.set! i false |
| catch _ => pure () |
| let mut out := #[← ser nm f] |
| for i in [0:args.size] do |
| if keep[i]! then out := out.push (← ser nm args[i]!) |
| return "(" ++ " ".intercalate out.toList ++ ")" |
| |
| def dump (n : Name) : Elab.Command.CommandElabM Unit := do |
| let some ci := (← getEnv).find? n | throwError "no such declaration {n}" |
| let js ← Elab.Command.liftTermElabM do |
| forallTelescope ci.type fun xs concl => do |
| let mut names : List (FVarId × String) := [] |
| let mut vars : Array String := #[] |
| let mut hyps : Array String := #[] |
| let mut k := 0 |
| for x in xs do |
| let d ← x.fvarId!.getDecl |
| if d.binderInfo.isInstImplicit then continue |
| let nmf := fun (id : FVarId) => (names.find? (fun p => p.1 == id)).map (·.2) |
| let ty ← ser nmf d.type |
| if (← isProp d.type) then |
| hyps := hyps.push ty |
| else |
| vars := vars.push ty |
| names := (x.fvarId!, s!"v{k}") :: names |
| k := k + 1 |
| let nmf := fun (id : FVarId) => (names.find? (fun p => p.1 == id)).map (·.2) |
| let c ← ser nmf concl |
| let hs := hyps.toList.map fun h => Json.str h |
| return Json.compress (Json.mkObj [("vars", Json.arr (vars.toList.map Json.str).toArray), ("hyps", Json.arr hs.toArray), ("concl", Json.str c)]) |
| logInfo m!"ALIGN_ELAB_HASH {js}" |
| |
| end AlignElabHash |
| ''' |
|
|
|
|
| |
|
|
|
|
| def build_lean_prelude() -> str: |
| return LEAN_DUMP |
|
|
|
|
| def build_ref_command(sig_text: str) -> str: |
| return f"theorem {REF} {sig_text} := by\n sorry" |
|
|
|
|
| DUMP_COMMAND = f"#eval {NS}.dump `{REF}" |
|
|
|
|
| def parse_dump(messages: Iterable[dict]) -> dict: |
| """Extract the JSON the Lean dump logged. Raises ValueError with the Lean error otherwise.""" |
| errs = [] |
| for m in messages: |
| if not isinstance(m, dict): |
| continue |
| data = str(m.get("data", "")) |
| if m.get("severity") == "error": |
| errs.append(data) |
| elif MARK in data: |
| return json.loads(data.split(MARK, 1)[1].strip()) |
| raise ValueError("no dump in output" + (": " + errs[0][:200] if errs else "")) |
|
|
|
|
| def canonical_form(dump: dict) -> str: |
| """Hypotheses sorted (multiset order-invariance); data binders and conclusion kept in order.""" |
| if not {"vars", "hyps", "concl"} <= set(dump): |
| raise ValueError(f"dump lacks fields: {sorted(dump)}") |
| return json.dumps({"vars": list(dump["vars"]), "hyps": sorted(dump["hyps"]), "concl": dump["concl"]}, |
| ensure_ascii=False, separators=(",", ":"), sort_keys=True) |
|
|
|
|
| def elab_sha(dump: dict) -> str: |
| return hashlib.sha256(canonical_form(dump).encode("utf-8")).hexdigest() |
|
|
|
|
| |
|
|
|
|
| def _load_common(): |
| if "p2_common" in sys.modules: |
| return sys.modules["p2_common"] |
| spec = importlib.util.spec_from_file_location("p2_common", REPO / "scripts" / "p2" / "p2_common.py") |
| mod = importlib.util.module_from_spec(spec) |
| sys.modules["p2_common"] = mod |
| spec.loader.exec_module(mod) |
| return mod |
|
|
|
|
| def hash_statement(worker, statement: str, timeout_s: float = 120.0) -> dict: |
| """Elaborate one statement in the pinned env and return {elab_sha, dump} or {error}.""" |
| from align.verify.lean.repl_pool import ReplError, ReplProtocolError |
| from align.verify.lean.strict import parse_statement |
|
|
| ref = parse_statement(statement) |
| if not ref.ok: |
| return {"error": f"unsupported_statement: {ref.problem}"} |
| try: |
| env = worker.header_env(ref.header_text, LEAN_DUMP) |
| except ReplProtocolError as e: |
| return {"error": "header_error: " + " ".join(str(e).split())[:200]} |
| except ReplError as e: |
| return {"error": e.code if hasattr(e, "code") else type(e).__name__} |
| try: |
| r1 = worker.command(build_ref_command(ref.sig_text), env, timeout_s) |
| errs = [m for m in r1.get("messages", []) if isinstance(m, dict) and m.get("severity") == "error"] |
| if errs or "env" not in r1: |
| return {"error": "elab_error: " + " ".join(str(errs[0].get("data", r1) if errs else r1).split())[:200]} |
| r2 = worker.command(DUMP_COMMAND, r1["env"], timeout_s) |
| dump = parse_dump(r2.get("messages", [])) |
| except ReplError as e: |
| return {"error": e.code if hasattr(e, "code") else type(e).__name__} |
| except ValueError as e: |
| return {"error": f"dump_error: {e}"} |
| return {"elab_sha": elab_sha(dump), "dump": dump} |
|
|
|
|
| def main(argv=None) -> int: |
| ap = argparse.ArgumentParser(description=__doc__.split("\n")[0]) |
| ap.add_argument("--in", dest="inp", required=True) |
| ap.add_argument("--out", required=True, help="JSONL {problem_id, elab_sha | error}; existing ids are skipped (resume)") |
| ap.add_argument("--limit", type=int, default=0) |
| ap.add_argument("--workers", type=int, default=2) |
| ap.add_argument("--timeout", type=float, default=120.0) |
| a = ap.parse_args(argv) |
| if not 1 <= a.workers <= 2: |
| raise SystemExit("--workers must be 1 or 2 (>2 needs the heavy-job lock; do the full run in a GPU-free window)") |
| C = _load_common() |
| from align.utils import guard |
| from align.verify.lean import ReplPool, find_toolchain |
| from align.verify.lean.repl_pool import ReplLimits |
|
|
| guard.preflight_host_check(strict=True) |
| done = set() |
| if os.path.exists(a.out): |
| done = {r["problem_id"] for r in C.iter_jsonl(a.out)} |
| rows = [] |
| for r in C.iter_jsonl(a.inp): |
| if r["problem_id"] not in done: |
| rows.append(r) |
| if a.limit and len(rows) >= a.limit: |
| break |
| tc = find_toolchain() |
| pool = ReplPool.from_toolchain(tc, n_workers=a.workers, limits=ReplLimits(), preflight=True, |
| scratch_dir=str(Path(os.environ.get("TMPDIR", str(REPO / ".tmp"))) / "p2_elab_pool")) |
| t0 = time.time() |
| ok = 0 |
| with C.AppendJsonl(a.out) as out: |
| for i in range(0, len(rows), 16): |
| chunk = rows[i : i + 16] |
| res = pool.map(lambda w, r: hash_statement(w, r["formal_statement"], a.timeout), chunk) |
| for r, x in zip(chunk, res): |
| if isinstance(x, BaseException): |
| raise x |
| out.append({"problem_id": r["problem_id"], **({"elab_sha": x["elab_sha"]} if "elab_sha" in x else {"error": x["error"]})}) |
| ok += "elab_sha" in x |
| pool.close() |
| print(json.dumps({"statements": len(rows), "hashed": ok, "errors": len(rows) - ok, "seconds": round(time.time() - t0, 1)})) |
| return 0 |
|
|
|
|
| if __name__ == "__main__": |
| raise SystemExit(main()) |
|
|
