Spaces:
Paused
Paused
| """Generate the determinism infrastructure verification artifact. | |
| Usage: | |
| python scripts/generate_determinism_artifact.py | |
| This script runs the determinism-related tests and produces | |
| ``research/generated/verification/determinism_infrastructure.json`` | |
| which the Gate B status builder uses instead of file-existence checks. | |
| The artifact contains: | |
| - status: "verified" if all tests pass | |
| - tested_tree_digest: SHA-256 of the current source tree | |
| - proofs: per-criterion boolean proof values | |
| - timestamp: ISO-8601 timestamp | |
| - schema_version: 1 | |
| """ | |
| from __future__ import annotations | |
| import json | |
| import subprocess | |
| import sys | |
| import time | |
| from pathlib import Path | |
| REPO_ROOT = Path(__file__).resolve().parent.parent | |
| ARTIFACT_DIR = REPO_ROOT / "research" / "generated" / "verification" | |
| ARTIFACT_PATH = ARTIFACT_DIR / "determinism_infrastructure.json" | |
| # Ensure src/ is importable | |
| sys.path.insert(0, str(REPO_ROOT)) | |
| def current_git_head() -> str | None: | |
| """Return the current git HEAD commit hash, or None.""" | |
| try: | |
| result = subprocess.run( | |
| ["git", "rev-parse", "HEAD"], | |
| capture_output=True, | |
| text=True, | |
| timeout=10, | |
| cwd=str(REPO_ROOT), | |
| ) | |
| return result.stdout.strip() if result.returncode == 0 else None | |
| except Exception: | |
| return None | |
| # Tests that verify determinism infrastructure | |
| DETERMINISM_TESTS = [ | |
| "tests/test_rng_persistence.py", | |
| "tests/test_iteration_determinism.py", | |
| "tests/test_structural_determinism.py", | |
| "tests/test_canonical_state.py", | |
| "tests/test_checkpoint_v4.py", | |
| "tests/test_engine_restore.py", | |
| "tests/test_production_restore.py", | |
| "tests/test_experiment_validity.py", | |
| ] | |
| def compute_tree_digest() -> str | None: | |
| """Compute SHA-256 digest using the canonical implementation.""" | |
| from src.dashboard.verification import compute_source_tree_digest | |
| return compute_source_tree_digest(REPO_ROOT) | |
| def compute_scope_tree_digest() -> str | None: | |
| """Compute the digest for the determinism evidence boundary.""" | |
| from src.dashboard.verification import compute_scope_digest | |
| return compute_scope_digest(REPO_ROOT, "restore_determinism") | |
| def run_tests() -> dict[str, bool]: | |
| """Run each determinism test file and return pass/fail per file.""" | |
| results: dict[str, bool] = {} | |
| for test_file in DETERMINISM_TESTS: | |
| test_path = REPO_ROOT / test_file | |
| if not test_path.exists(): | |
| print(f" SKIP {test_file} (not found)") | |
| results[test_file] = True # Skip missing files gracefully | |
| continue | |
| print(f" RUN {test_file} ...", end=" ") | |
| sys.stdout.flush() | |
| result = subprocess.run( | |
| [sys.executable, "-m", "pytest", str(test_path), "-q", "--tb=short"], | |
| capture_output=True, | |
| text=True, | |
| timeout=120, | |
| ) | |
| passed = result.returncode == 0 | |
| results[test_file] = passed | |
| if passed: | |
| print("PASSED") | |
| else: | |
| print("FAILED") | |
| # Print first few lines of failure output | |
| for line in result.stdout.split("\n")[-5:]: | |
| if line.strip(): | |
| print(f" {line.strip()}") | |
| for line in result.stderr.split("\n")[-5:]: | |
| if line.strip(): | |
| print(f" {line.strip()}") | |
| return results | |
| def main() -> int: | |
| print("=" * 60) | |
| print("Determinism Infrastructure Artifact Generator") | |
| print("=" * 60) | |
| print() | |
| # Compute tree digest | |
| print("Computing source tree digest...") | |
| digest = compute_tree_digest() | |
| if digest is None: | |
| print("ERROR: Failed to compute tree digest") | |
| return 1 | |
| print(f" tested_tree_digest = {digest}") | |
| scope_digest = compute_scope_tree_digest() | |
| if scope_digest is None: | |
| print("ERROR: Failed to compute restore_determinism scope digest") | |
| return 1 | |
| print(f" scope_digest = {scope_digest}") | |
| print() | |
| # Run tests | |
| print(f"Running {len(DETERMINISM_TESTS)} determinism test files...") | |
| print() | |
| test_results = run_tests() | |
| print() | |
| # Build proof map | |
| all_passed = all(test_results.values()) | |
| proofs = { | |
| "rng_state_persistence": test_results.get( | |
| "tests/test_rng_persistence.py", False | |
| ), | |
| "explicit_iteration_order": test_results.get( | |
| "tests/test_iteration_determinism.py", False | |
| ), | |
| "canonical_state_digest": test_results.get( | |
| "tests/test_canonical_state.py", False | |
| ), | |
| "structural_determinism": test_results.get( | |
| "tests/test_structural_determinism.py", False | |
| ), | |
| "checkpoint_v4_roundtrip": test_results.get( | |
| "tests/test_checkpoint_v4.py", False | |
| ), | |
| "engine_state_roundtrip": ( | |
| test_results.get("tests/test_engine_restore.py", False) | |
| and test_results.get("tests/test_production_restore.py", False) | |
| ), | |
| "experiment_validity": test_results.get( | |
| "tests/test_experiment_validity.py", False | |
| ), | |
| } | |
| # Get git HEAD for provenance (not freshness authority) | |
| head = current_git_head() | |
| # Build artifact | |
| artifact = { | |
| "schema_version": 1, | |
| "status": "verified" if all_passed else "failed", | |
| "test_run_head": head, | |
| "tested_commit": head, | |
| "tested_tree_digest": digest, | |
| "scope": "restore_determinism", | |
| "scope_digest": scope_digest, | |
| "timestamp": time.strftime("%Y-%m-%dT%H:%M:%S%z"), | |
| "python": f"{sys.version_info.major}.{sys.version_info.minor}.{sys.version_info.micro}", | |
| "test_files": DETERMINISM_TESTS, | |
| "test_results": { | |
| k: "passed" if v else "failed" for k, v in test_results.items() | |
| }, | |
| "proofs": proofs, | |
| } | |
| # Write artifact | |
| ARTIFACT_DIR.mkdir(parents=True, exist_ok=True) | |
| ARTIFACT_PATH.write_text( | |
| json.dumps(artifact, indent=2, ensure_ascii=False), | |
| encoding="utf-8", | |
| ) | |
| print(f"Artifact written to: {ARTIFACT_PATH}") | |
| print(f" status = {artifact['status']}") | |
| print(f" proofs = {sum(1 for v in proofs.values() if v)}/{len(proofs)} passed") | |
| if all_passed: | |
| print("\n✓ All determinism infrastructure tests pass.") | |
| return 0 | |
| else: | |
| print("\n✗ Some tests failed. Review output above.") | |
| return 1 | |
| if __name__ == "__main__": | |
| sys.exit(main()) | |