#!/usr/bin/env python3 """Plan and execute the versioned v5.3 development evaluation. The default is a side-effect-free plan. Pass ``--execute`` to run a stage. Rank-loss selection is validation-only; a new test artifact is emitted only after a rank weight has been locked. """ from __future__ import annotations import argparse from dataclasses import asdict, dataclass import hashlib from itertools import zip_longest import json import math from pathlib import Path import re import subprocess import sys import time from typing import Iterable REPO_ROOT = Path(__file__).resolve().parents[1] if str(REPO_ROOT) not in sys.path: sys.path.insert(0, str(REPO_ROOT)) from evaluation.provenance import ( # noqa: E402 artifact_manifest, git_provenance, runtime_provenance, ) DEFAULT_SEEDS = [42, 43, 44, 45, 46] DEFAULT_RANK_WEIGHTS = [0.5, 1.0] LOCKED_MODEL = "meta-llama/Llama-3.2-1B-Instruct" LOCKED_MODEL_REVISION = "9213176726f574b556790deb65791e0c5aa438b6" LOCKED_OPERATING_FPR = 0.01 STAGES = ( "smoke", "rank-selection", "lattice", "controls", "baselines", "mva-pilot", "mva-full", "report", "core", "all", ) @dataclass(frozen=True) class PlannedCommand: command_id: str stage: str argv: list[str] log_path: str expected_artifacts: list[str] purpose: str def parse_args(argv=None): parser = argparse.ArgumentParser( description="Run the locked paper-ready v5.3 development evaluation." ) parser.add_argument( "--source-dataset", type=Path, required=True, help="Raw v5.3 generation/label artifact retained for provenance.", ) parser.add_argument( "--input", type=Path, required=True, help="TokenEvalRow export consumed by training/evaluation.", ) parser.add_argument("--split-summary", type=Path, required=True) parser.add_argument("--model", required=True) parser.add_argument("--model-revision", required=True) parser.add_argument("--output-root", type=Path, required=True) parser.add_argument( "--signal-cache-dir", type=Path, default=None, help="Existing validated LatticeSpike signal cache; defaults under output root.", ) parser.add_argument( "--baseline-feature-cache-dir", type=Path, default=None, help="Existing content-addressed baseline cache; defaults under output root.", ) parser.add_argument( "--require-baseline-feature-cache", action="store_true", help="fail rather than extracting any missing baseline feature", ) parser.add_argument( "--skip-signal-extraction", action="store_true", help="Require and validate the existing signal cache without loading the LLM.", ) parser.add_argument("--stage", choices=STAGES, default="smoke") parser.add_argument("--execute", action="store_true") parser.add_argument( "--resume", action="store_true", help="resume only commands whose atomic manifests and artifact hashes validate", ) parser.add_argument("--allow-dirty-git", action="store_true") parser.add_argument("--seeds", nargs="+", type=int, default=DEFAULT_SEEDS) parser.add_argument( "--rank-weights", nargs="+", type=float, default=DEFAULT_RANK_WEIGHTS ) parser.add_argument( "--locked-rank-weight", type=float, default=None, help="Required for lattice/control stages unless rank-selection runs first.", ) parser.add_argument("--operating-fpr", type=float, default=0.01) parser.add_argument("--bootstrap-draws", type=int, default=10_000) parser.add_argument( "--retrain-locked-winner", action="store_true", help="diagnostic compatibility mode; retrain instead of evaluating selected checkpoints", ) parser.add_argument( "--diagnostic-unlocked-contract", action="store_true", help="allow noncanonical inputs/seeds/weights/FPR and mark the run diagnostic", ) parser.add_argument("--mva-pilot-trials", type=int, default=5) parser.add_argument("--mva-full-trials", type=int, default=200) parser.add_argument("--mva-pilot-timeout", type=float, default=3600.0) parser.add_argument( "--confirm-full-mva-search", action="store_true", help="Required before executing the expensive 200-trial MVA stage.", ) return parser.parse_args(argv) def _weight_slug(weight: float) -> str: return f"{float(weight):g}".replace("-", "m").replace(".", "p") def _file_sha256(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for chunk in iter(lambda: handle.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def _locked_input_contract() -> list[dict]: path = REPO_ROOT / "docs" / "artifacts" / "v5_3_locked_inputs.json" payload = json.loads(path.read_text(encoding="utf-8")) return list(payload["artifacts"]) def validate_source_export_binding(source_path: Path, export_path: Path) -> dict: """Prove that the TokenEvalRow export is a lossless view of raw v5.3.""" def rows(path: Path): with path.open(encoding="utf-8") as file: for line_number, line in enumerate(file, start=1): if line.strip(): yield line_number, json.loads(line) label_map = {"G": 0, "H": 1, "U": 2} count = 0 for source_entry, export_entry in zip_longest(rows(source_path), rows(export_path)): if source_entry is None or export_entry is None: raise ValueError("raw v5.3 and TokenEvalRow export have different row counts") source_line, source = source_entry export_line, exported = export_entry metadata = exported.get("metadata") or {} checks = { "id": (source.get("id"), exported.get("id")), "prompt": (source.get("question"), exported.get("prompt")), "answer": (source.get("generated_answer"), exported.get("answer")), "tokens": (source.get("gen_tokens"), metadata.get("generated_tokens")), "token_ids": ( source.get("gen_token_ids"), metadata.get("generated_token_ids"), ), "labels": ( [label_map.get(value, value) for value in source.get("token_labels", [])], exported.get("labels"), ), "t_star_index": ( source.get("t_star_index"), metadata.get("t_star_index"), ), } for field, (raw_value, exported_value) in checks.items(): if raw_value != exported_value: raise ValueError( f"source/export {field} mismatch at raw line {source_line}, " f"export line {export_line}" ) count += 1 if count == 0: raise ValueError("raw v5.3 and TokenEvalRow export cannot be empty") return {"rows": count, "fields_verified": sorted(checks)} def _python() -> str: return sys.executable def _lattice_paths(root: Path, kind: str, seed: int, weight: float) -> dict[str, Path]: directory = root / kind / f"rank_{_weight_slug(weight)}" / f"seed_{seed}" return { "directory": directory, "predictions": directory / "predictions.jsonl", "summary": directory / "summary.json", "results": directory / "results.md", "manifest": directory / "signals_manifest.jsonl", "log": directory / "run.log", "checkpoint_dir": directory / "checkpoints", } def _lattice_command( args, *, kind: str, seed: int, weight: float, validation_only: bool, smoke: bool = False, lstm_volume: bool = False, evaluate_checkpoint: Path | None = None, ) -> PlannedCommand: paths = _lattice_paths(args.output_root, kind, seed, weight) argv = [ _python(), "scripts/train_latticespike_llm_eval.py", "--input", str(args.input), "--split-summary", str(args.split_summary), "--model", args.model, "--model-revision", args.model_revision, "--prompt-template", "tanazzah-llama-chat", "--signals-dir", str(args.signal_cache_dir or (args.output_root / "signal_cache")), "--manifest", str(paths["manifest"]), "--out-dir", str(paths["directory"] / "checkpoints"), "--predictions", str(paths["predictions"]), "--summary", str(paths["summary"]), "--results-md", str(paths["results"]), "--seed", str(seed), "--split-seed", str(seed), "--lam-anchor", "0", "--lam-rank", str(weight), "--tie-box-token-emb", "--threshold-policy", "validation_target_fpr", "--operating-fpr", str(args.operating_fpr), "--checkpoint-metric", "validation_auprc", ] if validation_only: argv.append("--validation-only") if evaluate_checkpoint is not None: argv.extend(["--evaluate-checkpoint", str(evaluate_checkpoint)]) if args.skip_signal_extraction: argv.append("--skip-extract") if lstm_volume: argv.extend(["--ablation", "lstm", "--lstm-use-volume"]) if smoke: argv.extend( [ "--epochs", "1", "--early-stopping-patience", "0", "--limit-train", "128", "--limit-validation", "128", "--limit-eval", "32", "--diagnostic-limit", "4", "--no-component-gradient-diagnostics", ] ) expected_predictions = paths["predictions"] expected_summary = paths["summary"] expected_results = paths["results"] checkpoint_dir = paths["checkpoint_dir"] if lstm_volume: expected_predictions = expected_predictions.with_name( expected_predictions.stem + "_lstm_vol" + expected_predictions.suffix ) expected_summary = expected_summary.with_name( expected_summary.stem + "_lstm_vol" + expected_summary.suffix ) expected_results = expected_results.with_name( expected_results.stem + "_lstm_vol" + expected_results.suffix ) checkpoint_dir = checkpoint_dir.with_name(checkpoint_dir.name + "_lstm_vol") expected = [ expected_summary, expected_predictions, expected_results, paths["manifest"], checkpoint_dir / "best.pt", checkpoint_dir / "selected.pt", checkpoint_dir / "last.pt", checkpoint_dir / "run_provenance.json", checkpoint_dir / "diagnostics.json", ] return PlannedCommand( command_id=f"{kind}-rank-{_weight_slug(weight)}-seed-{seed}", stage=kind, argv=argv, log_path=str(paths["log"]), expected_artifacts=[str(path) for path in expected], purpose=( "validation-only rank-weight selection" if validation_only else "locked held-out LatticeSpike evaluation" ), ) def _baseline_paths(root: Path, suite: str, variant: str, seed: int) -> dict[str, Path]: directory = root / "baselines" / suite / variant / f"seed_{seed}" return { "directory": directory, "predictions": directory / "predictions.jsonl", "responses": directory / "responses.jsonl", "summary": directory / "summary.json", "log": directory / "run.log", } def _baseline_command( args, *, suite: str, variant: str, seed: int, detectors: Iterable[str], mva_trials: int | None = None, mva_timeout: float | None = None, ) -> PlannedCommand: paths = _baseline_paths(args.output_root, suite, variant, seed) argv = [ _python(), "scripts/train_faithful_baselines_dgx.py", "--input", str(args.input), "--split-summary", str(args.split_summary), "--model", args.model, "--model-revision", args.model_revision, "--prompt-template", "tanazzah-llama-chat", "--suite", suite, "--seed", str(seed), "--output", str(paths["predictions"]), "--response-output", str(paths["responses"]), "--summary", str(paths["summary"]), "--save-dir", str(paths["directory"] / "models"), "--feature-cache-dir", str( args.baseline_feature_cache_dir or (args.output_root / "baseline_feature_cache") ), ] if args.require_baseline_feature_cache: argv.append("--require-feature-cache") for detector in detectors: argv.extend(["--detector", detector]) if variant in {"official_code_raw", "official_code_norm"}: argv.extend(["--mva-feature-variant", variant]) if mva_trials is not None: argv.extend(["--mva-optuna-trials", str(mva_trials)]) if mva_timeout is not None: argv.extend(["--mva-optuna-timeout", str(mva_timeout)]) return PlannedCommand( command_id=f"baseline-{suite}-{variant}-seed-{seed}", stage="baselines" if mva_trials is None else "mva", argv=argv, log_path=str(paths["log"]), expected_artifacts=[ str(paths["summary"]), str(paths["predictions"]), str(paths["responses"]), str(paths["directory"] / "models"), ], purpose=f"{suite} baseline evaluation ({variant})", ) def rank_selection_commands(args) -> list[PlannedCommand]: return [ _lattice_command( args, kind="rank_selection", seed=seed, weight=weight, validation_only=True, ) for weight in args.rank_weights for seed in args.seeds ] def locked_lattice_commands(args, weight: float) -> list[PlannedCommand]: return [ _lattice_command( args, kind="lattice", seed=seed, weight=weight, validation_only=False, evaluate_checkpoint=( None if args.retrain_locked_winner else _lattice_paths( args.output_root, "rank_selection", seed, weight )["checkpoint_dir"] / "selected.pt" ), ) for seed in args.seeds ] def control_commands(args, weight: float) -> list[PlannedCommand]: return [ _lattice_command( args, kind="lstm_volume_control", seed=seed, weight=weight, validation_only=False, lstm_volume=True, ) for seed in args.seeds ] def standard_baseline_commands(args) -> list[PlannedCommand]: commands = [] for seed in args.seeds: commands.append( _baseline_command( args, suite="causal_matched", variant="causal", seed=seed, detectors=["epr", "lookback"], ) ) commands.append( _baseline_command( args, suite="method_faithful", variant="native", seed=seed, detectors=["epr", "wepr", "lookback"], ) ) return commands def mva_commands(args, *, trials: int, pilot: bool) -> list[PlannedCommand]: seeds = args.seeds[:1] if pilot else args.seeds timeout = args.mva_pilot_timeout if pilot else None commands = [] for seed in seeds: for variant in ("official_code_raw", "official_code_norm"): commands.append( _baseline_command( args, suite="method_faithful", variant=variant, seed=seed, detectors=["multi_view_attention_transformer_crf"], mva_trials=trials, mva_timeout=timeout, ) ) commands.append( _baseline_command( args, suite="offline", variant="official_code_raw", seed=seed, detectors=["multi_view_attention_transformer_crf"], mva_trials=trials, mva_timeout=timeout, ) ) return commands def select_rank_weight(args) -> tuple[float, dict]: rows = [] for weight in args.rank_weights: values = [] for seed in args.seeds: path = _lattice_paths( args.output_root, "rank_selection", seed, weight )["summary"] payload = json.loads(path.read_text(encoding="utf-8")) validate_lattice_summary( args, path, seed=seed, weight=weight, validation_only=True, ) value = payload.get("validation_eval", {}).get("auprc") if value is None: raise ValueError(f"missing validation AUPRC: {path}") values.append(float(value)) rows.append( { "rank_weight": float(weight), "validation_auprc_by_seed": values, "validation_auprc_mean": sum(values) / len(values), } ) winner = max(rows, key=lambda row: (row["validation_auprc_mean"], -row["rank_weight"])) result = { "selection_split": "validation", "selection_metric": "mean_seed_validation_auprc", "tie_break": "smaller_rank_weight", "candidates": rows, "selected_rank_weight": winner["rank_weight"], } path = args.output_root / "locked_rank_weight.json" path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(result, indent=2, sort_keys=True) + "\n", encoding="utf-8") return float(winner["rank_weight"]), result def validate_lattice_summary( args, path: Path, *, seed: int, weight: float, validation_only: bool, ) -> dict: payload = json.loads(path.read_text(encoding="utf-8")) training = payload.get("training") or {} split = payload.get("split") or {} errors = [] if payload.get("scoring_protocol") != "latticespike_causal_token_v2": errors.append("scoring_protocol") if int(training.get("training_seed", -1)) != int(seed): errors.append("training_seed") if not math.isclose(float(training.get("lam_rank", float("nan"))), float(weight)): errors.append("lam_rank") if bool(training.get("validation_only")) != bool(validation_only): errors.append("validation_only") if validation_only and payload.get("eval") is not None: errors.append("test_eval_present") if training.get("threshold_policy") != "validation_target_fpr": errors.append("threshold_policy") if not math.isclose( float(training.get("operating_fpr", float("nan"))), float(args.operating_fpr), ): errors.append("operating_fpr") if split.get("input_sha256") != _file_sha256(args.input): errors.append("input_sha256") if split.get("split_sha256") != _file_sha256(args.split_summary): errors.append("split_sha256") if split.get("validation_split_source") != "bound_split": errors.append("validation_split_source") if errors: raise ValueError(f"invalid locked LatticeSpike summary {path}: {errors}") prediction_path = Path(payload["predictions"]) if validation_only and prediction_path.read_text(encoding="utf-8").strip(): raise ValueError(f"validation-only run wrote prediction rows: {prediction_path}") return payload def report_command(args, weight: float) -> PlannedCommand: canonical = _baseline_paths( args.output_root, "causal_matched", "causal", args.seeds[0] ) argv = [ _python(), "scripts/report_paper_ready_eval.py", "--primary-method", "lattice_spike_rsnn_anomaly", "--comparator-method", "lookback_lens", "--bootstrap-draws", str(args.bootstrap_draws), "--expected-seeds", *[str(seed) for seed in args.seeds], "--expected-model-revision", args.model_revision, "--expected-input-sha256", _file_sha256(args.input), "--expected-split-sha256", _file_sha256(args.split_summary), "--expected-git-commit", git_provenance(REPO_ROOT, require_clean=False)["commit"], "--expected-operating-fpr", str(args.operating_fpr), "--canonical-metadata-run", f"lookback_lens:{args.seeds[0]}:{canonical['predictions']}", "--output-json", str(args.output_root / "report" / "paper_ready_results.json"), "--output-md", str(args.output_root / "report" / "paper_ready_results.md"), ] for seed in args.seeds: lattice = _lattice_paths(args.output_root, "lattice", seed, weight) causal = _baseline_paths(args.output_root, "causal_matched", "causal", seed) argv.extend( [ "--run", f"lattice_spike_rsnn_anomaly:{seed}:{lattice['predictions']}", "--run", f"lookback_lens:{seed}:{causal['predictions']}", ] ) return PlannedCommand( command_id="paper-ready-report", stage="report", argv=argv, log_path=str(args.output_root / "report" / "run.log"), expected_artifacts=[ str(args.output_root / "report" / "paper_ready_results.json"), str(args.output_root / "report" / "paper_ready_results.md"), ], purpose="common-support paired development report", ) def _write_plan(args, commands: list[PlannedCommand]) -> Path: plan = { "schema_version": 1, "status": "v5.3 development evaluation; not untouched confirmatory evidence", "dataset": { "source": artifact_manifest([args.source_dataset])[0], "token_eval_input": artifact_manifest([args.input])[0], "grouped_split": artifact_manifest([args.split_summary])[0], }, "source_export_binding": args.source_export_binding, "model": args.model, "model_revision": args.model_revision, "seeds": args.seeds, "rank_weights": args.rank_weights, "locked_rank_weight": args.locked_rank_weight, "operating_fpr": args.operating_fpr, "signal_cache_dir": str( args.signal_cache_dir or (args.output_root / "signal_cache") ), "baseline_feature_cache_dir": str( args.baseline_feature_cache_dir or (args.output_root / "baseline_feature_cache") ), "skip_signal_extraction": bool(args.skip_signal_extraction), "require_baseline_feature_cache": bool( args.require_baseline_feature_cache ), "git": git_provenance(REPO_ROOT, require_clean=False), "commands": [asdict(command) for command in commands], } canonical = json.dumps(plan, sort_keys=True, separators=(",", ":")).encode() fingerprint = hashlib.sha256(canonical).hexdigest() plan["plan_sha256"] = fingerprint path = args.output_root / "plans" / f"{args.stage}_{fingerprint[:12]}.json" _atomic_write_json(path, plan) index_path = args.output_root / "run_plan.json" index = {"schema_version": 2, "plans": []} if index_path.is_file(): index = json.loads(index_path.read_text(encoding="utf-8")) entry = { "stage": args.stage, "plan_sha256": fingerprint, "path": str(path), } index["plans"] = [ row for row in index.get("plans", []) if row.get("plan_sha256") != fingerprint ] + [entry] _atomic_write_json(index_path, index) return path def _atomic_write_json(path: Path, payload: object) -> None: path.parent.mkdir(parents=True, exist_ok=True) temporary = path.with_name(path.name + ".tmp") temporary.write_text( json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8" ) temporary.replace(path) def _expand_artifact_paths(paths: Iterable[str | Path]) -> list[Path]: expanded = set() for raw_path in paths: path = Path(raw_path) if path.is_file(): expanded.add(path) elif path.is_dir(): expanded.update(candidate for candidate in path.rglob("*") if candidate.is_file()) return sorted(expanded, key=lambda path: str(path)) def _command_manifest_path(args, command: PlannedCommand) -> Path: return args.output_root / "command_manifests" / f"{command.command_id}.json" def _failed_command_manifest_path(args, command: PlannedCommand) -> Path: return ( args.output_root / "command_manifests" / "failures" / f"{command.command_id}_{time.time_ns()}.json" ) def _argv_value(argv: list[str], name: str) -> str: try: return argv[argv.index(name) + 1] except (ValueError, IndexError) as exc: raise ValueError(f"planned command is missing {name}") from exc def _validate_command_outputs(args, command: PlannedCommand) -> None: missing = [ path for path in command.expected_artifacts if not Path(path).exists() ] if missing: raise RuntimeError(f"{command.command_id} omitted artifacts: {missing}") if command.stage in {"smoke", "rank_selection", "lattice", "lstm_volume_control"}: seed = int(_argv_value(command.argv, "--seed")) weight = float(_argv_value(command.argv, "--lam-rank")) validation_only = "--validation-only" in command.argv payload = validate_lattice_summary( args, Path(command.expected_artifacts[0]), seed=seed, weight=weight, validation_only=validation_only, ) if not validation_only and payload.get("eval") is None: raise ValueError(f"locked test run omitted test metrics: {command.command_id}") if command.stage == "lattice" and not args.retrain_locked_winner: if not (payload.get("training") or {}).get("evaluate_checkpoint"): raise ValueError("locked lattice stage did not use selected checkpoint") def _validated_resume_manifest(args, command: PlannedCommand) -> dict | None: path = _command_manifest_path(args, command) if not path.is_file(): return None payload = json.loads(path.read_text(encoding="utf-8")) if payload.get("status") != "completed": return None if payload.get("command") != asdict(command): raise ValueError(f"resume command contract changed: {command.command_id}") current_commit = git_provenance(REPO_ROOT, require_clean=False)["commit"] if payload.get("git", {}).get("commit") != current_commit: raise ValueError(f"resume commit changed: {command.command_id}") for artifact in payload.get("artifacts", []): artifact_path = Path(artifact["path"]) if not artifact_path.is_file(): raise ValueError(f"resume artifact is missing: {artifact_path}") if artifact_path.stat().st_size != int(artifact["bytes"]): raise ValueError(f"resume artifact size changed: {artifact_path}") if _file_sha256(artifact_path) != artifact["sha256"]: raise ValueError(f"resume artifact hash changed: {artifact_path}") _validate_command_outputs(args, command) return payload def execute_commands( args, commands: list[PlannedCommand], *, plan_path: Path, append_timings: bool = True, ) -> list[dict]: timing_path = args.output_root / "timings.json" timings = [] if timing_path.is_file(): timings = json.loads(timing_path.read_text(encoding="utf-8")) timing_by_id = {row["command_id"]: row for row in timings} stage_timings = [] for index, command in enumerate(commands, start=1): if args.resume and _validated_resume_manifest(args, command) is not None: print(f"[{index}/{len(commands)}] {command.command_id} [validated resume]", flush=True) if command.command_id in timing_by_id: stage_timings.append(timing_by_id[command.command_id]) continue log_path = Path(command.log_path) log_path.parent.mkdir(parents=True, exist_ok=True) started = time.monotonic() print(f"[{index}/{len(commands)}] {command.command_id}", flush=True) with log_path.open("w", encoding="utf-8") as log: process = subprocess.run( command.argv, cwd=REPO_ROOT, stdout=log, stderr=subprocess.STDOUT, text=True, check=False, ) elapsed = time.monotonic() - started timing = { "command_id": command.command_id, "returncode": process.returncode, "wall_seconds": elapsed, "log": str(log_path), } timing_by_id[command.command_id] = timing stage_timings.append(timing) _atomic_write_json( timing_path, sorted(timing_by_id.values(), key=lambda row: row["command_id"]), ) if process.returncode: _atomic_write_json( _failed_command_manifest_path(args, command), { "schema_version": 2, "status": "failed", "command": asdict(command), "git": git_provenance(REPO_ROOT, require_clean=False), "plan": artifact_manifest([plan_path])[0], "timing": timing, "artifacts": artifact_manifest([log_path]), }, ) write_orchestrator_manifest(args) raise RuntimeError( f"{command.command_id} failed with exit {process.returncode}; see {log_path}" ) try: _validate_command_outputs(args, command) except Exception as exc: present = _expand_artifact_paths( [*command.expected_artifacts, log_path] ) _atomic_write_json( _failed_command_manifest_path(args, command), { "schema_version": 2, "status": "invalid_outputs", "error": str(exc), "command": asdict(command), "git": git_provenance(REPO_ROOT, require_clean=False), "plan": artifact_manifest([plan_path])[0], "timing": timing, "artifacts": artifact_manifest(present), }, ) write_orchestrator_manifest(args) raise artifacts = artifact_manifest( _expand_artifact_paths([*command.expected_artifacts, log_path]) ) _atomic_write_json( _command_manifest_path(args, command), { "schema_version": 2, "status": "completed", "command": asdict(command), "git": git_provenance(REPO_ROOT, require_clean=False), "plan": artifact_manifest([plan_path])[0], "timing": timing, "artifacts": artifacts, }, ) write_orchestrator_manifest(args) return stage_timings def completed_artifact_manifest( commands: list[PlannedCommand], extra_paths: Iterable[str | Path] = () ) -> list[dict]: paths = { Path(path) for command in commands for path in [*command.expected_artifacts, command.log_path] if Path(path).exists() } paths.update(Path(path) for path in extra_paths if Path(path).exists()) return artifact_manifest(_expand_artifact_paths(paths)) def write_orchestrator_manifest(args) -> Path: command_manifests = [] outputs = {} manifest_dir = args.output_root / "command_manifests" for path in sorted(manifest_dir.rglob("*.json")) if manifest_dir.is_dir() else []: payload = json.loads(path.read_text(encoding="utf-8")) command_manifests.append(artifact_manifest([path])[0]) for artifact in payload.get("artifacts", []): outputs[artifact["path"]] = artifact extra_paths = [ args.output_root / "run_plan.json", args.output_root / "timings.json", args.output_root / "locked_rank_weight.json", args.output_root / "mva_pilot_forecast.json", ] for artifact in artifact_manifest( [path for path in extra_paths if path.is_file()] ): outputs[artifact["path"]] = artifact plan_paths = sorted((args.output_root / "plans").glob("*.json")) manifest = { "schema_version": 2, "git": git_provenance(REPO_ROOT, require_clean=False), "dataset": { "source": artifact_manifest([args.source_dataset])[0], "token_eval_input": artifact_manifest([args.input])[0], "grouped_split": artifact_manifest([args.split_summary])[0], }, "source_export_binding": args.source_export_binding, "runtime": runtime_provenance( ["numpy", "scikit-learn", "torch", "transformers", "optuna"] ), "plans": artifact_manifest(plan_paths), "command_manifests": command_manifests, "outputs": sorted(outputs.values(), key=lambda row: row["path"]), } path = args.output_root / "orchestrator_manifest.json" _atomic_write_json(path, manifest) return path def write_mva_pilot_forecast(args, timings: list[dict]) -> Path | None: pilot = [row for row in timings if row["command_id"].startswith("baseline-")] pilot = [ row for row in pilot if "official_code_raw" in row["command_id"] or "official_code_norm" in row["command_id"] ] if not pilot: return None ratio = args.mva_full_trials / args.mva_pilot_trials rows = [] for row in pilot: rows.append( { "command_id": row["command_id"], "pilot_trials": args.mva_pilot_trials, "pilot_wall_seconds": row["wall_seconds"], "full_trials": args.mva_full_trials, "linear_upper_bound_wall_seconds": row["wall_seconds"] * ratio, "note": ( "Conservative linear extrapolation: feature extraction and " "cache warm-up are included in the pilot wall time." ), } ) path = args.output_root / "mva_pilot_forecast.json" path.write_text(json.dumps(rows, indent=2, sort_keys=True) + "\n", encoding="utf-8") return path def _validate_args(args) -> None: if not all( path.is_file() for path in (args.source_dataset, args.input, args.split_summary) ): raise FileNotFoundError( "--source-dataset, --input, and --split-summary must be existing files" ) args.source_export_binding = validate_source_export_binding( args.source_dataset, args.input ) if not re.fullmatch(r"[0-9a-fA-F]{40,64}", args.model_revision): raise ValueError("--model-revision must be an immutable commit hash") if not args.seeds or len(set(args.seeds)) != len(args.seeds): raise ValueError("--seeds must be non-empty and unique") if not args.rank_weights or any(weight < 0 for weight in args.rank_weights): raise ValueError("--rank-weights must be non-empty and nonnegative") if not 0 <= args.operating_fpr <= 1: raise ValueError("--operating-fpr must be in [0, 1]") if not args.diagnostic_unlocked_contract: supplied = (args.source_dataset, args.input, args.split_summary) locked = _locked_input_contract() for path, expected in zip(supplied, locked): if path.stat().st_size != int(expected["bytes"]): raise ValueError(f"locked input size mismatch: {path}") if _file_sha256(path) != expected["sha256"]: raise ValueError(f"locked input SHA-256 mismatch: {path}") if args.model != LOCKED_MODEL or args.model_revision != LOCKED_MODEL_REVISION: raise ValueError("model/revision differ from the locked v5.3 contract") if args.seeds != DEFAULT_SEEDS: raise ValueError("reportable v5.3 runs require seeds 42 43 44 45 46") if args.rank_weights != DEFAULT_RANK_WEIGHTS: raise ValueError("reportable v5.3 runs require rank weights 0.5 and 1.0") if not math.isclose(args.operating_fpr, LOCKED_OPERATING_FPR): raise ValueError("reportable v5.3 runs require operating FPR 0.01") if args.stage == "mva-full" and args.execute: raise ValueError( "mva-full is disabled until validation-only tune-once/locked-eval is implemented" ) if args.skip_signal_extraction: cache_dir = args.signal_cache_dir or (args.output_root / "signal_cache") if not cache_dir.is_dir(): raise FileNotFoundError( "--skip-signal-extraction requires an existing --signal-cache-dir" ) if args.require_baseline_feature_cache: cache_dir = args.baseline_feature_cache_dir or ( args.output_root / "baseline_feature_cache" ) if not cache_dir.is_dir(): raise FileNotFoundError( "--require-baseline-feature-cache requires an existing cache directory" ) if args.execute and args.output_root.exists() and any(args.output_root.iterdir()): if not args.resume: raise ValueError( "non-empty --output-root requires --resume; use a fresh protocol-v2 root" ) def main(argv=None) -> None: args = parse_args(argv) _validate_args(args) if args.execute: git_provenance(REPO_ROOT, require_clean=not args.allow_dirty_git) selected_weight = args.locked_rank_weight commands: list[PlannedCommand] = [] if args.stage == "smoke": weight = selected_weight if selected_weight is not None else args.rank_weights[0] commands = [ _lattice_command( args, kind="smoke", seed=args.seeds[0], weight=weight, validation_only=True, smoke=True, ) ] elif args.stage == "rank-selection": commands = rank_selection_commands(args) elif args.stage == "lattice": if selected_weight is None: raise ValueError("--locked-rank-weight is required for lattice") commands = locked_lattice_commands(args, selected_weight) elif args.stage == "controls": if selected_weight is None: raise ValueError("--locked-rank-weight is required for controls") commands = control_commands(args, selected_weight) elif args.stage == "baselines": commands = standard_baseline_commands(args) elif args.stage == "mva-pilot": commands = mva_commands(args, trials=args.mva_pilot_trials, pilot=True) elif args.stage == "mva-full": commands = mva_commands(args, trials=args.mva_full_trials, pilot=False) elif args.stage == "report": if selected_weight is None: raise ValueError("--locked-rank-weight is required for report") commands = [report_command(args, selected_weight)] elif args.stage in {"core", "all"}: commands = rank_selection_commands(args) if not args.execute: commands += standard_baseline_commands(args) if selected_weight is not None: commands += locked_lattice_commands(args, selected_weight) commands += [report_command(args, selected_weight)] if args.stage == "all": commands += mva_commands( args, trials=args.mva_pilot_trials, pilot=True ) if selected_weight is not None: commands += control_commands(args, selected_weight) plan_path = _write_plan(args, commands) print(plan_path) if not args.execute: return if args.stage in {"core", "all"}: execute_commands(args, commands, plan_path=plan_path) selected_weight, selection = select_rank_weight(args) args.locked_rank_weight = selected_weight print(json.dumps(selection, indent=2, sort_keys=True)) remaining = locked_lattice_commands(args, selected_weight) remaining += standard_baseline_commands(args) if args.stage == "all": remaining += control_commands(args, selected_weight) remaining += mva_commands( args, trials=args.mva_pilot_trials, pilot=True ) remaining += [report_command(args, selected_weight)] remaining_plan = _write_plan(args, commands + remaining) timings = execute_commands( args, remaining, plan_path=remaining_plan, append_timings=True ) if args.stage == "all": write_mva_pilot_forecast(args, timings) executed_commands = commands + remaining else: timings = execute_commands(args, commands, plan_path=plan_path) executed_commands = commands if args.stage == "rank-selection": selected_weight, selection = select_rank_weight(args) print(json.dumps(selection, indent=2, sort_keys=True)) if args.stage == "mva-pilot": write_mva_pilot_forecast(args, timings) manifest_path = write_orchestrator_manifest(args) print(manifest_path) if __name__ == "__main__": main()