#!/usr/bin/env python3 """Replay the decisive Qwen30 Fixed-PD cells with Frontier state outputs.""" from __future__ import annotations import argparse import importlib.util import json import subprocess import sys import time from pathlib import Path from typing import Any HERE = Path(__file__).resolve().parent CONFIGS = { "tp2_mns64": "tp2", "tp4_mns32": "tp4", "tp4_mns64": "tp4", } def load_module(name: str, path: Path): spec = importlib.util.spec_from_file_location(name, path) if spec is None or spec.loader is None: raise ImportError(path) module = importlib.util.module_from_spec(spec) sys.modules[spec.name] = module spec.loader.exec_module(module) return module BASE = load_module( "qwen30_fixed_pd_state_base", HERE / "run_qwen235_fixed_pd_state_replay.py" ) Q30 = load_module( "qwen30_fixed_pd_surface", HERE / "run_frontier_qwen30_exact_trace_surface.py" ) def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser() parser.add_argument("--base-sim-root", type=Path, required=True) parser.add_argument("--output-root", type=Path, required=True) parser.add_argument("--frontier-source", type=Path, required=True) parser.add_argument("--python-deps", type=Path, required=True) parser.add_argument("--config", action="append", choices=CONFIGS) parser.add_argument("--timeout-seconds", type=float, default=1800) parser.add_argument("--resume", action="store_true") parser.add_argument("--op-trace", action="store_true") return parser.parse_args() def run_cell( *, config: str, base_sim_root: Path, output_root: Path, frontier_source: Path, python_deps: Path, timeout_seconds: float, resume: bool, op_trace: bool, ) -> dict[str, Any]: base_run = base_sim_root / "runs" / config / CONFIGS[config] base_command_path = base_run / "command.json" base_result_path = base_run / "result.json" if not base_command_path.is_file() or not base_result_path.is_file(): raise FileNotFoundError(f"frozen inputs missing under {base_run}") base_command = json.loads(base_command_path.read_text()) base_result = json.loads(base_result_path.read_text()) if base_result.get("status") != "completed": raise ValueError(f"frozen result is not completed: {base_result_path}") run_root = output_root / config result_path = run_root / "result.json" if resume and result_path.is_file(): previous = json.loads(result_path.read_text()) if ( previous.get("status") == "PASS" and previous.get("inputs", {}).get("base_command_sha256") == BASE.sha256_file(base_command_path) and previous.get("inputs", {}).get("base_result_sha256") == BASE.sha256_file(base_result_path) and bool(previous.get("op_trace_enabled")) == op_trace ): return previous if run_root.exists() and any(run_root.iterdir()): raise FileExistsError(f"refusing non-empty output: {run_root}") run_root.mkdir(parents=True, exist_ok=True) command = BASE.transform_command( base_command, metrics_root=run_root / "frontier_metrics", run_id=f"qwen30_fixed_pd_state_{config}", op_trace=op_trace, ) BASE.atomic_json(run_root / "command.json", command) inputs = { "base_run": str(base_run.resolve()), "base_command_sha256": BASE.sha256_file(base_command_path), "base_result_sha256": BASE.sha256_file(base_result_path), "base_request_metrics_sha256": base_result["request_metrics_sha256"], } manifest = { "schema": "qwen30-fixed-pd-state-replay-v1", "config": config, "inputs": inputs, "command_sha256": BASE.sha256_file(run_root / "command.json"), "frontier_git_head": subprocess.check_output( ["git", "-C", str(frontier_source), "rev-parse", "HEAD"], text=True ).strip(), "op_trace_enabled": op_trace, "controlled_changes": [ "metrics output directory", "metrics run id", "full Frontier stage/batch ledger enabled", "individual batch metrics enabled", ] + (["op-level tracing enabled"] if op_trace else []), } BASE.atomic_json(run_root / "run_manifest.json", manifest) started = time.monotonic() with (run_root / "stdout.log").open("w") as stdout, ( run_root / "stderr.log" ).open("w") as stderr: try: completed = subprocess.run( command, cwd=frontier_source, env=BASE.environment(frontier_source, python_deps), stdout=stdout, stderr=stderr, timeout=timeout_seconds, check=False, ) returncode = int(completed.returncode) except subprocess.TimeoutExpired: returncode = 124 elapsed_seconds = time.monotonic() - started if returncode != 0: failure = { "status": "STOP", "config": config, "returncode": returncode, "elapsed_seconds": elapsed_seconds, } BASE.atomic_json(run_root / "failure.json", failure) raise RuntimeError(f"state replay failed: {failure}") fallback_evidence = Q30.collective_fallback_evidence(run_root) if fallback_evidence: raise RuntimeError(f"collective-profile fallback detected: {fallback_evidence}") paths = BASE.find_state_metrics(run_root) request_metrics_sha256 = BASE.sha256_file(paths["requests"]) scorer_equivalent = request_metrics_sha256 == base_result["request_metrics_sha256"] if not scorer_equivalent: raise RuntimeError( f"observation changed scorer input for {config}: " f"{request_metrics_sha256} != {base_result['request_metrics_sha256']}" ) state = BASE.summarize_frontier( system_metrics_path=paths["system"], request_metrics_path=paths["requests"], batch_metrics_path=paths["batches"], ledger_path=paths["ledger"], ) BASE.atomic_json(run_root / "common-state.json", state) if op_trace: matches = sorted(run_root.glob("frontier_metrics/**/op_traces.jsonl")) if len(matches) != 1: raise ValueError(f"expected one op trace, found {len(matches)}") paths["op_trace"] = matches[0] result = { "schema": "qwen30-fixed-pd-state-replay-result-v1", "status": "PASS", "config": config, "elapsed_seconds": elapsed_seconds, "returncode": returncode, "inputs": inputs, "scorer_equivalence": { "request_metrics_byte_identical": scorer_equivalent, "request_metrics_sha256": request_metrics_sha256, "base_metrics": base_result["metrics"], }, "state_artifacts": { name: {"path": str(path.resolve()), "sha256": BASE.sha256_file(path)} for name, path in paths.items() }, "common_state": state, "collective_fallback_evidence": fallback_evidence, "op_trace_enabled": op_trace, } BASE.atomic_json(result_path, result) return result def main() -> None: args = parse_args() for name in ("base_sim_root", "output_root", "frontier_source", "python_deps"): setattr(args, name, getattr(args, name).resolve()) results = [] for config in tuple(args.config or CONFIGS): result = run_cell( config=config, base_sim_root=args.base_sim_root, output_root=args.output_root, frontier_source=args.frontier_source, python_deps=args.python_deps, timeout_seconds=args.timeout_seconds, resume=args.resume, op_trace=args.op_trace, ) results.append(result) print(json.dumps({"config": config, "status": result["status"]}), flush=True) BASE.atomic_json( args.output_root / "state_replay.json", { "schema": "qwen30-fixed-pd-state-replay-aggregate-v1", "status": "PASS", "results": results, }, ) if __name__ == "__main__": main()