#!/usr/bin/env python3 """Run the HBF independent-reproduction package and emit a signed-ready record. The program verifies software, numerical benchmarks, routes and claim guards. It cannot verify the operator's identity or institutional independence and it never promotes model maturity automatically. Those remain review/sign-off steps under ``platform/science/validation-evidence.schema.json``. """ from __future__ import annotations import argparse import hashlib import json import os import platform import shutil import subprocess import sys import time from datetime import datetime, timezone from pathlib import Path ROOT = Path(__file__).resolve().parents[1] SCHEMA = "hbf-external-reproduction-run-1.0" COMMANDS = [ {"id": "reactivity-kernel", "argv": ["python3", "packages/reactivity/test_reactivity_kernel.py"]}, {"id": "cross-runtime-reactivity", "argv": ["python3", "tests/cross-language/compare.py"]}, {"id": "phase2-covariance", "argv": ["python3", "tests/test_phase2_covariance.py"], "quick": True}, {"id": "phase3-equilibrium-slowing", "argv": ["python3", "tests/test_phase3_physics.py"]}, {"id": "phase3-completion", "argv": ["python3", "tests/test_phase3_complete.py"], "quick": True}, {"id": "browser-core", "argv": ["node", "tests/browser_core.test.mjs"], "quick": True}, {"id": "phase4-readiness", "argv": ["python3", "tests/test_phase4_readiness.py"], "quick": True}, {"id": "science-foundation", "argv": ["node", "platform/science/tests/science_foundation.test.mjs"]}, {"id": "uncertainty-kernels", "argv": ["node", "platform/science/tests/kernel_uncertainty.test.mjs"]}, {"id": "run-contract", "argv": ["node", "packages/run-contract/test.mjs"]}, {"id": "ui-kit", "argv": ["node", "packages/kit-tests.mjs"]}, {"id": "virtual-lab-suite", "argv": ["sh", "tests/run-all.sh"], "cwd": "virtual-lab"}, {"id": "scientific-release-gate", "argv": ["python3", "tools/scientific_release_check.py"], "quick": True}, ] HASH_TARGETS = [ "content/sources/sikora-weller-2016/cross-section.csv", "content/sources/sikora-weller-2016/derived-covariance.json", "packages/reactivity/maxwellian.py", "packages/reactivity/covariance.py", "scientific-core/equilibrium/grad_shafranov.py", "scientific-core/kinetics/alpha_slowing.py", "scientific-core/kinetics/isotropic_distributions.py", "scientific-core/radiation/radiation_models.py", "scientific-core/stability/ideal_mhd_interface.py", "scientific-core/browser-kernels.mjs", "external-validation/run_validation.py", "external-validation/run-result.schema.json", "external-validation/PREREGISTRATION-TEMPLATE.md", "external-validation/SIGNOFF-TEMPLATE.json.example", "external-validation/REPORTING.md", "tests/test_phase4_readiness.py", "platform/scientific-status.json", ] def sha256(path: Path) -> str: h = hashlib.sha256() with path.open("rb") as f: for chunk in iter(lambda: f.read(1024 * 1024), b""): h.update(chunk) return h.hexdigest() def source_digest() -> str: """Digest relative paths and bytes, excluding self-referential/generated files.""" # Exclude VCS, dependency and build directories. Without these, a developer # simply running `git init`, `npm install` or a packaging step mutates the # digest and breaks the release-consistency gate even though no shipped # source file changed. These directories are never part of the release. excluded_names = { "__pycache__", ".pytest_cache", ".DS_Store", ".git", ".hg", ".svn", "node_modules", ".venv", "venv", "dist", "build", ".mypy_cache", ".ruff_cache", ".tox", "harness", } excluded_suffixes = {".pyc", ".pyo", ".zip"} records = [] for path in sorted(ROOT.rglob("*")): if not path.is_file() or any(p in excluded_names for p in path.parts): continue if path.suffix.lower() in excluded_suffixes: continue rel = path.relative_to(ROOT).as_posix() if rel in { "external-validation/validation-run.json", "platform/release-manifest.json", "platform/scientific-status/release-check.json", } or rel.startswith("api/.data/"): continue records.append(rel + "\0" + sha256(path)) return hashlib.sha256("\n".join(records).encode("utf-8")).hexdigest() def version(argv): exe = shutil.which(argv[0]) if not exe: return {"available": False, "command": argv} try: out = subprocess.run(argv, cwd=ROOT, text=True, capture_output=True, timeout=20, check=False) text = (out.stdout or out.stderr).strip().splitlines() return {"available": out.returncode == 0, "command": argv, "value": text[0] if text else "", "executable": exe} except Exception as exc: return {"available": False, "command": argv, "error": repr(exc)} def environment(): try: import numpy numpy_version = numpy.__version__ except Exception: numpy_version = None try: import scipy scipy_version = scipy.__version__ except Exception: scipy_version = None return { "platform": platform.platform(), "machine": platform.machine(), "python": sys.version.replace("\n", " "), "python_executable": sys.executable, "numpy": numpy_version, "scipy_optional": scipy_version, "node": version(["node", "--version"]), "timezone": time.tzname[0] if time.tzname else None, } def run_command(item, timeout_s=300): cwd = ROOT / item.get("cwd", ".") started = time.perf_counter() try: proc = subprocess.run(item["argv"], cwd=cwd, text=True, capture_output=True, timeout=timeout_s, check=False, env={**os.environ, "PYTHONDONTWRITEBYTECODE": "1"}) duration = time.perf_counter() - started combined = (proc.stdout or "") + ("\n" + proc.stderr if proc.stderr else "") if len(combined) > 40000: combined = "[output truncated to final 40000 characters]\n" + combined[-40000:] return {"id": item["id"], "argv": item["argv"], "cwd": item.get("cwd", "."), "return_code": proc.returncode, "passed": proc.returncode == 0, "duration_s": round(duration, 3), "output": combined.strip()} except subprocess.TimeoutExpired as exc: return {"id": item["id"], "argv": item["argv"], "cwd": item.get("cwd", "."), "return_code": None, "passed": False, "duration_s": round(time.perf_counter()-started, 3), "output": f"TIMEOUT after {timeout_s}s: {exc}"} def describe(quick=False): selected = [c for c in COMMANDS if (c.get("quick") or not quick)] return { "schema_version": SCHEMA, "mode": "quick" if quick else "full", "root_requirements": {"python": ">=3.10", "node": ">=18", "numpy": "required", "scipy": "optional"}, "commands": [{k: v for k, v in c.items() if k != "quick"} for c in selected], "hash_targets": HASH_TARGETS, "independence_note": "Identity and institutional independence are not verified by software.", "source_digest_scope": "all shipped files except the release manifest, generated release-check/run records, caches, archives, and api/.data", } def main(): parser = argparse.ArgumentParser() parser.add_argument("--quick", action="store_true", help="run new phase gates plus the release gate") parser.add_argument("--describe", action="store_true", help="print the command manifest without executing") parser.add_argument("--output", help="write the machine-readable run record to PATH") parser.add_argument("--operator", default="", help="self-reported operator; does not prove identity") parser.add_argument("--institution", default="", help="self-reported institution; does not prove independence") parser.add_argument("--timeout", type=int, default=300) args = parser.parse_args() if args.describe: print(json.dumps(describe(args.quick), indent=2)) return 0 selected = [c for c in COMMANDS if (c.get("quick") or not args.quick)] started_at = datetime.now(timezone.utc).replace(microsecond=0).isoformat() results = [] for item in selected: print(f"RUN {item['id']} {' '.join(item['argv'])}", flush=True) result = run_command(item, timeout_s=args.timeout) results.append(result) print(("PASS " if result["passed"] else "FAIL ") + item["id"] + f" ({result['duration_s']:.3f}s)", flush=True) hashes = {} hash_errors = [] for rel in HASH_TARGETS: path = ROOT / rel if path.is_file(): hashes[rel] = sha256(path) else: hash_errors.append(rel) passed = sum(r["passed"] for r in results) failed = len(results) - passed record = { "schema_version": SCHEMA, "run_id": "reproduction-" + started_at.replace(":", "").replace("+00:00", "Z"), "generated_at": datetime.now(timezone.utc).replace(microsecond=0).isoformat(), "mode": "quick" if args.quick else "full", "self_reported_operator": args.operator or None, "self_reported_institution": args.institution or None, "identity_verified": False, "institutional_independence_verified": False, "automatic_maturity_promotion": False, "environment": environment(), "source_tree_sha256": source_digest(), "source_tree_digest_scope": ( "all shipped files except the release manifest, generated release-check/run " "records, caches, archives, and api/.data"), "artifact_sha256": hashes, "hash_errors": hash_errors, "commands": results, "summary": {"passed": passed, "failed": failed, "total": len(results), "all_hash_targets_present": not hash_errors}, "claim_boundary": ( "A passing run demonstrates reproducibility of the supplied software " "and benchmarks on this environment. It does not establish experimental " "validation, institutional independence, reactor performance, gain or net energy."), "next_step": ( "An independent reviewer must submit a separately signed evidence artifact " "matching platform/science/validation-evidence.schema.json."), } if args.output: out = Path(args.output) if not out.is_absolute(): out = ROOT / out out.parent.mkdir(parents=True, exist_ok=True) out.write_text(json.dumps(record, indent=2, ensure_ascii=False) + "\n", encoding="utf-8") print("WROTE", out) print(json.dumps(record["summary"], indent=2)) return 1 if failed or hash_errors else 0 if __name__ == "__main__": raise SystemExit(main())