release_tools/gate_union.py

#!/usr/bin/env python3
"""gate_union.py — release gate for the team pool-union bundle.

Final check before a union result posts. Rigorous but fast: it does NOT blindly
re-verify all ~1156 certs — it leverages an already-certified baseline (the v2
manifest, which leaner independently verified 1127/1127) by byte-comparing the
union's overlap with it, and only sends the *net-new* certs (union minus
baseline) to the judge. Any byte mismatch in the overlap or any net-new reject
fails the gate.

Checks:
  1. structural: every id canonical + unique; solved_count == #certs == expected.
  2. coverage: union ⊇ baseline solved set (no regression).
  3. integrity: for ids in both, union cert code == baseline cert code (byte).
     A mismatch means the assembly corrupted a known-good cert — hard fail.
  4. net-new: judge-verify every id in union but not baseline. Any reject fails.

Usage:
    source <repo>/.env.judge
    python3 gate_union.py <repo> <union.json> <baseline_manifest.json> \
        [--expect N] [--jobs K] [--full]   # --full: judge-verify ALL, ignore baseline

Exit 0 = GREEN (safe to post); 1 = FAIL; 2 = usage.
"""
from __future__ import annotations

import json
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))
from verify_certs import extract_certs, load_canonical, DEFAULT_PROOF_POLICY  # noqa: E402
import concurrent.futures as cf  # noqa: E402
import os  # noqa: E402
import time  # noqa: E402

CANON = ("normal", "hard1", "hard2", "hard3")


def main() -> None:
    a = sys.argv[1:]
    if len(a) < 3:
        sys.exit(__doc__)
    repo = Path(a[0]).resolve()
    union = extract_certs(json.loads(Path(a[1]).read_text()))
    baseline = extract_certs(json.loads(Path(a[2]).read_text()))
    expect = None
    jobs = 6
    full = False
    i = 3
    while i < len(a):
        if a[i] == "--expect":
            expect = int(a[i + 1]); i += 2
        elif a[i] == "--jobs":
            jobs = int(a[i + 1]); i += 2
        elif a[i] == "--full":
            full = True; i += 1
        else:
            sys.exit(f"unknown arg {a[i]}")

    if not os.environ.get("LEAN_BIN"):
        sys.exit("FATAL: source <repo>/.env.judge first.")
    sys.path.insert(0, str(repo))  # so `import judge.verify` resolves
    problems = load_canonical(repo)
    fails = []

    # 1. structural
    ids = list(union)
    bad_id = [p for p in ids if p not in problems]
    if bad_id:
        fails.append(f"non-canonical ids: {bad_id[:5]}")
    if expect is not None and len(union) != expect:
        fails.append(f"count {len(union)} != expected {expect}")
    print(f"[1] structural: {len(union)} certs, {len(bad_id)} non-canonical, "
          f"expect={expect}")

    # 2. coverage
    missing = set(baseline) - set(union)
    if missing:
        fails.append(f"union missing {len(missing)} baseline ids: {sorted(missing)[:5]}")
    print(f"[2] coverage: union ⊇ baseline? {'yes' if not missing else 'NO ('+str(len(missing))+' missing)'}")

    # 3. integrity (byte-compare overlap) — skipped under --full
    overlap = [p for p in union if p in baseline]
    if not full:
        mism = [p for p in overlap if union[p]["code"] != baseline[p]["code"]
                or union[p]["verdict"] != baseline[p]["verdict"]]
        if mism:
            fails.append(f"{len(mism)} overlap certs differ from certified baseline: {mism[:5]}")
        print(f"[3] integrity: {len(overlap)} overlap certs, {len(mism)} byte-mismatches "
              f"(pre-certified baseline, no re-verify)")
        to_verify = [p for p in union if p not in baseline]
    else:
        to_verify = ids
        print("[3] integrity: --full -> re-verifying ALL certs, ignoring baseline")

    # 4. judge-verify the net-new (or all under --full)
    from judge.verify import verify_answer  # noqa: PLC0415

    def check(pid):
        c = union[pid]
        prob = {**problems[pid], "proof_policy": problems[pid].get("proof_policy") or DEFAULT_PROOF_POLICY}
        t0 = time.time()
        try:
            res = verify_answer(prob, json.dumps({"verdict": c["verdict"], "code": c["code"]}))
            return pid, res.get("status"), time.time() - t0
        except Exception as e:  # noqa: BLE001
            return pid, f"error:{type(e).__name__}", time.time() - t0

    print(f"[4] judge-verifying {len(to_verify)} {'ALL' if full else 'net-new'} certs (jobs={jobs})...")
    rej = []
    with cf.ThreadPoolExecutor(max_workers=jobs) as ex:
        for pid, status, dt in ex.map(check, sorted(to_verify)):
            mark = "PASS" if status == "accepted" else "FAIL"
            if mark == "FAIL":
                rej.append((pid, status))
            print(f"    [{mark}] {pid}: {status} ({dt:.1f}s)")
    if rej:
        fails.append(f"{len(rej)} net-new certs rejected: {rej[:5]}")

    print("\n" + "=" * 60)
    if fails:
        print("GATE: FAIL")
        for f in fails:
            print("  -", f)
        sys.exit(1)
    print(f"GATE: GREEN — {len(union)} certs, {len(overlap)} pre-certified + "
          f"{len(to_verify)} judge-verified, coverage ⊇ baseline. Safe to post.")
    sys.exit(0)


if __name__ == "__main__":
    main()