Spaces:
Running
Running
File size: 4,979 Bytes
ca74b23 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 | from __future__ import annotations
import hashlib
import json
import os
import platform
import subprocess
import sys
import time
from datetime import UTC, datetime
from pathlib import Path
from td_repro.claim1 import verify_claim1
from td_repro.claim5 import verify_claim5
from td_repro.figures import make_figures
from td_repro.theorem_audit import verify_theorem_audits
ROOT = Path(__file__).resolve().parents[2]
FIXED_COMMAND = "uv sync --frozen && uv run --no-sync python -m td_repro"
def file_sha256(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as handle:
for block in iter(lambda: handle.read(1024 * 1024), b""):
digest.update(block)
return digest.hexdigest()
def git_sha() -> str:
return subprocess.check_output(
["git", "rev-parse", "HEAD"], cwd=ROOT, text=True
).strip()
def cpu_allocation() -> dict:
affinity = None
if hasattr(os, "sched_getaffinity"):
affinity = len(os.sched_getaffinity(0))
model = "unknown"
cpuinfo = Path("/proc/cpuinfo")
if cpuinfo.exists():
for line in cpuinfo.read_text().splitlines():
if line.lower().startswith("model name"):
model = line.split(":", 1)[1].strip()
break
return {
"estimated_cores_required": 16,
"selected_backend": "hf",
"selected_flavor": "cpu-upgrade",
"logical_cpus": os.cpu_count(),
"affinity_cpus": affinity,
"cpu_model": model,
"gpu_used": False,
}
def write_json(path: Path, data: dict) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(data, indent=2, sort_keys=True) + "\n")
def main() -> int:
started = time.perf_counter()
claim1 = verify_claim1()
theorem_claims = verify_theorem_audits()
claim5 = verify_claim5()
visible_verifier = subprocess.run(
[sys.executable, str(ROOT / "space" / "code" / "current_verifier.py")],
cwd=ROOT,
text=True,
capture_output=True,
check=False,
)
artifact_commands = {
"artifact_validation": [sys.executable, "-m", "td_repro.artifact_validation"],
"marimo_check": [sys.executable, "-m", "marimo", "check", "notebooks/td_reproduction.py"],
}
artifact_validators = {}
for name, command in artifact_commands.items():
result = subprocess.run(command, cwd=ROOT, text=True, capture_output=True, check=False)
artifact_validators[name] = {
"command": " ".join(command),
"returncode": result.returncode,
"stdout": result.stdout.strip(),
"stderr": result.stderr.strip(),
}
runtime = time.perf_counter() - started
evidence = {
"paper": "arXiv:2603.02577",
"run_started_utc": datetime.now(UTC).isoformat(),
"git_sha": git_sha(),
"fixed_command": FIXED_COMMAND,
"python": platform.python_version(),
"platform": platform.platform(),
"uv_lock_sha256": file_sha256(ROOT / "uv.lock"),
"compute": cpu_allocation(),
"deterministic_seeds": claim1["deterministic_seeds"],
"claims": [claim1, *theorem_claims, claim5],
"evaluator_visible_verifier": {
"command": "python space/code/current_verifier.py",
"returncode": visible_verifier.returncode,
"stdout": visible_verifier.stdout.strip(),
"stderr": visible_verifier.stderr.strip(),
},
"artifact_validators": artifact_validators,
"all_claims_passed": claim1["passed"] and all(claim["passed"] for claim in theorem_claims) and claim5["passed"] and visible_verifier.returncode == 0 and all(item["returncode"] == 0 for item in artifact_validators.values()),
"total_runtime_seconds": runtime,
}
evidence["report_figures_svg_base64"] = make_figures(evidence["claims"])
artifact_dir = ROOT / ".openresearch" / "artifacts" / "claim5"
write_json(artifact_dir / "raw" / "results.json", evidence)
write_json(artifact_dir / "independent_checker_output.json", claim5["independent_checker"])
write_json(artifact_dir / "negative_control_output.json", claim5["negative_control"])
for claim in evidence["claims"]:
claim_dir = ROOT / ".openresearch" / "artifacts" / claim["claim_id"]
write_json(claim_dir / "raw" / "results.json", claim)
if "independent_checker" in claim:
write_json(claim_dir / "independent_checker_output.json", claim["independent_checker"])
if "proof_replay" in claim:
write_json(claim_dir / "proof_replay.json", claim["proof_replay"])
if "negative_control" in claim:
write_json(claim_dir / "negative_control_output.json", claim["negative_control"])
print("OPENRESEARCH_EVIDENCE_BEGIN")
print(json.dumps(evidence, indent=2, sort_keys=True))
print("OPENRESEARCH_EVIDENCE_END")
return 0 if evidence["all_claims_passed"] else 1
|