release_tools/verify_certs.py

#!/usr/bin/env python3
"""verify_certs.py — re-verify a certificate bundle through the local Lean judge.

Run this against ANY certs.json or manifest.json BEFORE posting a result. One
bad certificate invalidates the whole result on the eval space, so this is the
guard: it re-runs every claimed certificate through the official judge
(judge/verify.py) exactly as the eval space will, prints a PASS/FAIL line per
problem, and exits nonzero if anything is rejected.

Self-contained: needs only the judge repo + Python stdlib (no cebench, no pysat).

Usage:
    source <repo>/.env.judge          # exports LEAN_BIN / LAKE_BIN
    python3 verify_certs.py <repo> <bundle.json> [--jobs N] [--ids a,b,c]

Accepted input shapes (auto-detected):
    * certs map:   {"hard3_0001": {"verdict": "true", "code": "..."}, ...}
    * eq2 manifest: {"results": [{"id": ..., "verdict": ..., "code": ...,
                                  "solved": true}, ...], ...}
      (only rows with solved=true / a non-null code are checked)

Exit codes: 0 = all accepted; 1 = at least one rejected / error; 2 = usage.
"""
from __future__ import annotations

import concurrent.futures as cf
import json
import os
import sys
import time
from pathlib import Path

DEFAULT_PROOF_POLICY = {"allowed_axioms": ["propext", "Quot.sound", "Classical.choice"]}
CANONICAL_SETS = ("normal", "hard1", "hard2", "hard3")

# Match the eval space's size caps (pipeline/config.json), NOT judge/verify.py's
# stricter 50KB default — so this gate's verdicts equal what the eval will do.
# A gate stricter than the eval would false-reject valid 50-100KB TRUE certs.
# setdefault so a caller can still override via env.
os.environ.setdefault("MAX_CODE_LENGTH", "100000")       # TRUE proof code cap (eval)
os.environ.setdefault("MAX_FALSE_CERT_BYTES", "20000")   # FALSE cert cap (eval)


def load_canonical(repo: Path) -> dict:
    problems = {}
    for name in CANONICAL_SETS:
        f = repo / "examples" / "problems" / f"{name}.jsonl"
        if not f.exists():
            sys.exit(f"FATAL: canonical problem file missing: {f}")
        for line in f.read_text().splitlines():
            line = line.strip()
            if line:
                row = json.loads(line)
                problems[row["id"]] = row
    return problems


def extract_certs(bundle: dict) -> dict:
    """Return {id: {'verdict':..., 'code':...}} from either accepted shape."""
    if isinstance(bundle, dict) and "results" in bundle and isinstance(bundle["results"], list):
        out = {}
        for r in bundle["results"]:
            if r.get("solved") and r.get("code"):
                out[r["id"]] = {"verdict": r.get("verdict"), "code": r["code"]}
        return out
    # plain certs map
    out = {}
    for pid, cert in bundle.items():
        if isinstance(cert, dict) and "code" in cert:
            out[pid] = {"verdict": cert.get("verdict"), "code": cert["code"]}
    return out


def main() -> None:
    args = sys.argv[1:]
    if len(args) < 2:
        sys.exit(__doc__)
    repo = Path(args[0]).resolve()
    bundle_path = Path(args[1])
    jobs = 1
    only_ids = None
    i = 2
    while i < len(args):
        if args[i] == "--jobs":
            jobs = int(args[i + 1]); i += 2
        elif args[i] == "--ids":
            only_ids = set(args[i + 1].split(",")); i += 2
        else:
            sys.exit(f"unknown arg: {args[i]}\n{__doc__}")

    if not os.environ.get("LEAN_BIN") or not os.environ.get("LAKE_BIN"):
        sys.exit("FATAL: LEAN_BIN / LAKE_BIN not set — run `source <repo>/.env.judge` first.")

    sys.path.insert(0, str(repo))
    from judge.verify import verify_answer  # noqa: PLC0415

    problems = load_canonical(repo)
    bundle = json.loads(bundle_path.read_text())
    certs = extract_certs(bundle)
    if only_ids:
        certs = {k: v for k, v in certs.items() if k in only_ids}
    if not certs:
        sys.exit("FATAL: no certificates found in bundle (expected certs map or manifest with solved rows).")

    unknown = [pid for pid in certs if pid not in problems]
    if unknown:
        print(f"FAIL: {len(unknown)} non-canonical/unknown problem ids: {unknown[:5]}", file=sys.stderr)

    def check(pid: str):
        cert = certs[pid]
        base = problems.get(pid)
        if base is None:
            return pid, "unknown_id", "not a canonical problem", 0.0
        problem = {**base, "proof_policy": base.get("proof_policy") or DEFAULT_PROOF_POLICY}
        answer = json.dumps({"verdict": cert["verdict"], "code": cert["code"]})
        t0 = time.time()
        try:
            res = verify_answer(problem, answer)
            status = res.get("status", "?")
            msg = (res.get("message") or "")[:200]
        except Exception as e:  # noqa: BLE001
            status, msg = "error", f"{type(e).__name__}: {e}"
        return pid, status, msg, time.time() - t0

    ids = sorted(certs)
    results = []
    print(f"re-verifying {len(ids)} certificate(s) through the judge (jobs={jobs})...\n", flush=True)
    if jobs > 1:
        with cf.ThreadPoolExecutor(max_workers=jobs) as ex:
            for pid, status, msg, dt in ex.map(check, ids):
                results.append((pid, status, msg, dt))
                mark = "PASS" if status == "accepted" else "FAIL"
                print(f"  [{mark}] {pid}: {status} ({dt:.1f}s)"
                      + (f"  {msg}" if mark == "FAIL" else ""), flush=True)
    else:
        for pid in ids:
            pid, status, msg, dt = check(pid)
            results.append((pid, status, msg, dt))
            mark = "PASS" if status == "accepted" else "FAIL"
            print(f"  [{mark}] {pid}: {status} ({dt:.1f}s)"
                  + (f"  {msg}" if mark == "FAIL" else ""), flush=True)

    accepted = [r for r in results if r[1] == "accepted"]
    failed = [r for r in results if r[1] != "accepted"]
    print(f"\n=== {len(accepted)}/{len(results)} accepted; {len(failed)} rejected ===")
    if failed or unknown:
        print("REJECTED:", [r[0] for r in failed] + unknown)
        sys.exit(1)
    print("ALL CERTIFICATES ACCEPTED — safe to post.")
    sys.exit(0)


if __name__ == "__main__":
    main()