scripts/smoke_repro.py

#!/usr/bin/env python3
"""Run a real, verdict-isolated admission workload and seal its receipt."""

from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))

from repro_control.artifacts import finalize_attempt, verify_read_back
from repro_control.hashing import atomic_write_json
from repro_control.heartbeat import heartbeat, marker
from repro_control.runtime import assert_smoke_isolated, load_job_manifest, verify_science_spec
from repro_control.smoke_runtime import (
    fixture_hashes,
    orchestrate_family,
    run_c5_probe_child,
    run_family_child,
)


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--science-spec", type=Path)
    parser.add_argument("--job-manifest", type=Path)
    parser.add_argument("--family", choices=("sudoku", "maze-c5", "mnist"))
    parser.add_argument("--project-root", type=Path, default=Path(__file__).resolve().parents[1])
    parser.add_argument("--handoff", type=Path)
    parser.add_argument("--output-dir", type=Path)
    parser.add_argument("--steps-warmup", type=int, default=5)
    parser.add_argument("--steps-measured", type=int, default=20)
    parser.add_argument("--dry-run", action="store_true")
    parser.add_argument("--allow-cpu-preflight", action="store_true")
    parser.add_argument("--internal-family-child", choices=("sudoku", "maze-c5", "mnist"))
    parser.add_argument("--internal-c5-probe-child", action="store_true")
    parser.add_argument("--size", type=int)
    parser.add_argument("--batch-size", type=int)
    parser.add_argument("--timing", action="store_true")
    args = parser.parse_args()

    if args.internal_family_child:
        value = run_family_child(
            args.project_root,
            args.handoff,
            args.internal_family_child,
            warmup_steps=args.steps_warmup,
            measured_steps=args.steps_measured,
            allow_cpu=args.allow_cpu_preflight,
        )
        print(json.dumps(value))
        return 0
    if args.internal_c5_probe_child:
        value = run_c5_probe_child(
            args.handoff,
            size=args.size,
            batch_size=args.batch_size,
            timing=args.timing,
            allow_cpu=args.allow_cpu_preflight,
        )
        print(json.dumps(value))
        return 0

    if not all((args.science_spec, args.job_manifest, args.family, args.handoff)):
        parser.error("top-level smoke requires spec, manifest, family, and handoff")
    verify_science_spec(args.science_spec)
    manifest = load_job_manifest(args.job_manifest)
    assert_smoke_isolated(manifest)
    if (args.steps_warmup, args.steps_measured) != (5, 20):
        raise SystemExit("registered smoke requires 5 warm-up and 20 measured steps")
    if args.dry_run:
        print(
            json.dumps(
                {
                    "contract_verified": True,
                    "family": args.family,
                    "verdict_metrics_emitted": False,
                    "outcomes": {},
                }
            )
        )
        return 0
    if args.output_dir is None:
        raise SystemExit("--output-dir is required for execution")
    args.output_dir.mkdir(parents=True, exist_ok=False)
    marker("CPU_PREFLIGHT_READY" if args.allow_cpu_preflight else "GPU_READY", args.family)
    with heartbeat(f"GPU_SMOKE:{args.family}"):
        result = orchestrate_family(
            Path(__file__).resolve(),
            args.project_root,
            args.handoff,
            args.family,
            warmup_steps=args.steps_warmup,
            measured_steps=args.steps_measured,
            allow_cpu=args.allow_cpu_preflight,
        )
    receipt = {
        "format": 1,
        "logical_id": manifest["logical_id"],
        "attempt_id": manifest["attempt_id"],
        "family": args.family,
        "fixture_hashes": fixture_hashes(args.handoff, args.family),
        "warmup_steps": 5,
        "measured_steps": 20,
        "result": result,
        "verdict_metrics_emitted": False,
        "outcomes": {},
    }
    atomic_write_json(args.output_dir / "smoke-receipt.json", receipt)
    finalize_attempt(
        args.output_dir,
        logical_id=manifest["logical_id"],
        attempt_id=manifest["attempt_id"],
        expected_outputs=manifest["expected_outputs"],
    )
    verify_read_back(
        args.output_dir,
        logical_id=manifest["logical_id"],
        attempt_id=manifest["attempt_id"],
    )
    marker("DONE", f"{manifest['logical_id']} {manifest['attempt_id']}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())