code/verified_claim_protection.py

#!/usr/bin/env python3
"""Independent CPU audit for the two already-verified benchmark claims.

This reads the pinned release archives directly, rather than trusting the
stored results.json.  It performs a second exact pass: JSONL record counts,
required-field coverage, canonical record fingerprints, numbered Lean-file
coverage, and the six-baseline solved/submission maxima.
"""

from __future__ import annotations

import argparse
import hashlib
import io
import json
import tarfile
from collections import defaultdict
from decimal import Decimal, ROUND_HALF_UP
from pathlib import PurePosixPath


BENCHMARKS = {
    "formal_math500": ("FormalMath500", 387),
    "minif2f_solving": ("MiniF2FSolving", 375),
    "putnam_solving": ("PutnamBenchSolving", 324),
}
REQUIRED = {
    "conclusions",
    "formal_answer",
    "formal_answer_type",
    "header",
    "hypotheses",
    "independent_variables",
    "informal_answer",
    "informal_problem",
    "metainfo",
}


def members(tar: tarfile.TarFile):
    for member in tar:
        if member.isfile():
            yield member


def read_member(tar: tarfile.TarFile, name: str) -> bytes:
    for member in members(tar):
        if member.name.endswith(name):
            handle = tar.extractfile(member)
            if handle is None:
                break
            return handle.read()
    raise FileNotFoundError(name)


def jsonl(data: bytes) -> list[dict]:
    return [json.loads(line) for line in data.decode("utf-8").splitlines() if line.strip()]


def fingerprint(record: dict) -> str:
    payload = {key: record[key] for key in sorted(REQUIRED)}
    encoded = json.dumps(payload, sort_keys=True, ensure_ascii=False).encode("utf-8")
    return hashlib.sha256(encoded).hexdigest()


def current_pass(path) -> dict:
    counts = {}
    schema_ok = 0
    unique_by_benchmark = {}
    lean_files = {}
    lean_sha256 = {}
    with tarfile.open(path, "r:gz") as tar:
        for key, (lean_dir, expected) in BENCHMARKS.items():
            records = jsonl(read_member(tar, f"/data/{key}.jsonl"))
            counts[key] = len(records)
            if len(records) != expected:
                raise AssertionError((key, len(records), expected))
            fps = [fingerprint(record) for record in records]
            schema_ok += sum(REQUIRED.issubset(record) for record in records)
            unique_by_benchmark[key] = len(set(fps))
            names = {
                PurePosixPath(member.name).name
                for member in members(tar)
                if f"/FormalProblemSolving/{lean_dir}/" in member.name
                and member.name.endswith(".lean")
            }
            expected_names = {f"{i}.lean" for i in range(1, expected + 1)}
            if names != expected_names:
                raise AssertionError((key, len(names), len(expected_names)))
            lean_files[key] = len(names)
            digests = []
            for i in range(1, expected + 1):
                raw = read_member(
                    tar,
                    f"/FormalProblemSolving/{lean_dir}/{i}.lean",
                )
                digests.append(hashlib.sha256(raw).hexdigest())
            lean_sha256[key] = {
                "files": len(digests),
                "unique_bytes": len(set(digests)),
                "first": digests[0],
                "last": digests[-1],
            }
    return {
        "counts": counts,
        "records_total": sum(counts.values()),
        "records_with_required_schema": schema_ok,
        "unique_records": unique_by_benchmark,
        "numbered_lean_files": lean_files,
        "lean_sha256": lean_sha256,
    }


def solved(row: dict) -> bool:
    return bool(row.get("submission")) and row.get("eq_proof") is not None


def initial_pass(path) -> dict:
    method_rows = defaultdict(lambda: {"records": 0, "solved": 0, "submitted": 0})
    with tarfile.open(path, "r:gz") as tar:
        for member in members(tar):
            name = PurePosixPath(member.name)
            if name.name not in {
                "solving.formal_math500.jsonl",
                "solving.minif2f_solving.jsonl",
                "solving.putnam_solving.jsonl",
            }:
                continue
            handle = tar.extractfile(member)
            if handle is None:
                raise AssertionError(member.name)
            benchmark = name.name.removeprefix("solving.").removesuffix(".jsonl")
            method = str(name.parent.relative_to(PurePosixPath(member.name).parts[0] + "/baseline_results"))
            # The archive root is not semantically relevant; locate the path
            # after baseline_results so the result is stable across tarballs.
            parts = PurePosixPath(member.name).parts
            start = parts.index("baseline_results") + 1
            method = "/".join(parts[start:-1])
            rows = jsonl(handle.read())
            key = (benchmark, method)
            method_rows[key]["records"] += len(rows)
            method_rows[key]["solved"] += sum(solved(row) for row in rows)
            method_rows[key]["submitted"] += sum(bool(row.get("submission")) for row in rows)

    if len(method_rows) != 18:
        raise AssertionError(f"expected 18 methods, got {len(method_rows)}")
    maxima = {}
    submission_maxima = defaultdict(int)
    for (benchmark, method), row in sorted(method_rows.items()):
        if benchmark not in maxima or row["solved"] > maxima[benchmark]["solved"]:
            maxima[benchmark] = {"method": method, "solved": row["solved"]}
        submission_maxima[benchmark] = max(submission_maxima[benchmark], row["submitted"])
    denominators = {key: value[1] for key, value in BENCHMARKS.items()}
    rates = {}
    for benchmark, row in maxima.items():
        fraction = Decimal(row["solved"]) / Decimal(denominators[benchmark]) * Decimal(100)
        rates[benchmark] = str(fraction.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP))
    return {
        "method_count": len(method_rows),
        "records_present": sum(row["records"] for row in method_rows.values()),
        "slots": 6 * sum(denominators.values()),
        "maxima": maxima,
        "rates_percent_decimal": rates,
        "submission_only_maxima": dict(submission_maxima),
    }


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--current", default="source/official-current.tar.gz")
    parser.add_argument("--initial", default="source/official-initial.tar.gz")
    args = parser.parse_args()
    current = current_pass(args.current)
    initial = initial_pass(args.initial)
    if current["records_total"] != 1086 or current["records_with_required_schema"] != 1086:
        raise AssertionError(current)
    if sum(current["unique_records"].values()) != 1086:
        raise AssertionError(current)
    expected = {
        "formal_math500": 23.77,
        "minif2f_solving": 27.47,
        "putnam_solving": 0.31,
    }
    if {key: Decimal(value) for key, value in initial["rates_percent_decimal"].items()} != {
        key: Decimal(str(value)) for key, value in expected.items()
    }:
        raise AssertionError(initial)
    if initial["records_present"] + 737 != initial["slots"]:
        raise AssertionError(initial)
    print(json.dumps({"claim5_fresh": current, "claim6_fresh": initial}, sort_keys=True))


if __name__ == "__main__":
    main()