#!/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()