| |
| """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")) |
| |
| |
| 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() |
|
|
