#!/usr/bin/env python3 """Package frozen Round 07 outputs without recomputing any result gate.""" from __future__ import annotations import argparse import copy import hashlib import json from pathlib import Path from typing import Any PROTOCOL_ID = "llm-atlas-k3-attnres-local-path-v1" SEEDS = (2026073001, 2026073002, 2026073003) def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser() parser.add_argument("--raw-dir", type=Path, required=True) parser.add_argument("--manifest", type=Path, required=True) parser.add_argument("--aggregate", type=Path, required=True) parser.add_argument("--reproduction-output", type=Path, required=True) parser.add_argument("--compact-output", type=Path, required=True) return parser.parse_args() 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 canonical_sha256(value: Any) -> str: payload = json.dumps( value, ensure_ascii=False, sort_keys=True, separators=(",", ":") ).encode() return hashlib.sha256(payload).hexdigest() def load_canonical(path: Path) -> dict[str, Any]: value = json.loads(path.read_text()) expected = value["canonical_sha256_without_self"] payload = { key: item for key, item in value.items() if key != "canonical_sha256_without_self" } if canonical_sha256(payload) != expected: raise RuntimeError(f"canonical hash mismatch: {path}") return value def write_canonical(path: Path, value: dict[str, Any]) -> None: value["canonical_sha256_without_self"] = canonical_sha256(value) path.parent.mkdir(parents=True, exist_ok=True) temporary = path.with_suffix(path.suffix + ".tmp") temporary.write_text( json.dumps(value, ensure_ascii=False, indent=2, sort_keys=True) + "\n" ) temporary.replace(path) def replay_payload(value: dict[str, Any]) -> dict[str, Any]: cleaned = copy.deepcopy(value) for key in ("run_kind", "timing", "canonical_sha256_without_self"): cleaned.pop(key) return cleaned def artifact_hashes(repo_root: Path) -> dict[str, str]: paths = { "runner": "experiments/k3/attnres_local_path/train.py", "analyzer": "experiments/k3/attnres_local_path/analyze.py", "packager": "experiments/k3/attnres_local_path/package.py", "manifest": "experiments/k3/attnres_local_path/manifest.json", "protocol": "research/K3_ATTNRES_LOCAL_PATH_PROTOCOL.md", "scoping": "research/K3_ATTNRES_LOCAL_PATH_SCOPING.md", "preresult_grok_review": ( "research/K3_ATTNRES_LOCAL_PATH_GROK_REVIEW.md" ), } return { name: file_sha256(repo_root / path) for name, path in paths.items() } def main() -> None: args = parse_args() repo_root = Path(__file__).resolve().parents[3] manifest = json.loads(args.manifest.read_text()) aggregate = load_canonical(args.aggregate) if ( manifest["protocol_id"] != PROTOCOL_ID or aggregate["protocol_id"] != PROTOCOL_ID ): raise RuntimeError("protocol mismatch") formal = {} raw_files = {} for seed in SEEDS: name = f"formal-seed-{seed}.json" path = args.raw_dir / name run = load_canonical(path) if ( run["run_kind"] != "formal" or run["seed"] != seed or not run["round06_equivalence"]["passed"] ): raise RuntimeError(f"invalid formal run: {name}") formal[seed] = run raw_files[name] = { "file_sha256": file_sha256(path), "canonical_sha256": run["canonical_sha256_without_self"], } replay_name = f"replay-seed-{SEEDS[0]}.json" replay_path = args.raw_dir / replay_name replay = load_canonical(replay_path) if replay["run_kind"] != "replay" or replay["seed"] != SEEDS[0]: raise RuntimeError("invalid replay") raw_files[replay_name] = { "file_sha256": file_sha256(replay_path), "canonical_sha256": replay["canonical_sha256_without_self"], } compare_payload = replay_payload(formal[SEEDS[0]]) replay_exact = compare_payload == replay_payload(replay) if not replay_exact or not aggregate["replay"][ "formal_seed1_exact_excluding_run_kind_and_timing" ]: raise RuntimeError("replay exactness failed") replay_gate = { "passed": True, "excluded_fields": [ "run_kind", "timing", "canonical_sha256_without_self", ], "frozen_compare_sha256": canonical_sha256(compare_payload), } reproduction = { "schema_version": 1, "protocol_id": PROTOCOL_ID, "raw_files": raw_files, "replay_gate": replay_gate, "artifacts": artifact_hashes(repo_root), "aggregate": { "file_sha256": file_sha256(args.aggregate), "canonical_sha256": aggregate[ "canonical_sha256_without_self" ], }, "post_result_grok_review": { "session": "019fb19d-94a3-7231-9a63-3a1ef33a9892", "role": "read-only adversarial implementation audit; not an evidence source", "blocking_errors": 0, "localization_status_confirmed": True, }, } write_canonical(args.reproduction_output, reproduction) final_spectra = [] for seed in SEEDS: final = formal[seed]["diagnostics"][-1]["local_matrix"] final_spectra.append( { "seed": seed, "modes": { mode: { "normalized": final[mode]["positions"][ "post_mlp_state" ]["reductions"]["element_rms"]["statistics"][ "normalized" ], "spike_contrast": final[mode]["positions"][ "post_mlp_state" ]["reductions"]["element_rms"]["statistics"][ "spike_contrast" ], "peak_normalized": final[mode]["positions"][ "post_mlp_state" ]["reductions"]["element_rms"]["statistics"][ "peak_normalized" ], "peak_layer": final[mode]["positions"][ "post_mlp_state" ]["reductions"]["element_rms"]["statistics"][ "peak_layer" ], "uniform_count": final[mode]["selector"][ "uniform_count" ], } for mode in manifest["matrix_modes"] }, } ) compact = { "schema_version": 1, "protocol_id": PROTOCOL_ID, "study": { "identity": manifest["study_identity"], "seeds": list(SEEDS), "steps": manifest["training"]["steps"], "formal_target_bytes": ( len(SEEDS) * manifest["training"]["target_bytes_per_cell"] ), "total_target_bytes_with_replay": ( (len(SEEDS) + 1) * manifest["training"]["target_bytes_per_cell"] ), "modes": manifest["matrix_modes"], "spike_layers": manifest["primary_object"][ "spike_layers_one_based" ], "metrics": manifest["primary_object"]["metrics"], }, "thresholds": manifest["thresholds"], "formulas": manifest["formulas"], "formal_cells": aggregate["formal_cells"], "scores": aggregate["scores"], "means": aggregate["means"], "gates": aggregate["gates"], "replay": { **aggregate["replay"], "frozen_compare_sha256": replay_gate[ "frozen_compare_sha256" ], }, "final_spectra": final_spectra, "limitations": aggregate["limitations"], "hashes": { "aggregate_file_sha256": file_sha256(args.aggregate), "aggregate_canonical_sha256": aggregate[ "canonical_sha256_without_self" ], "reproduction_file_sha256": file_sha256( args.reproduction_output ), "reproduction_canonical_sha256": reproduction[ "canonical_sha256_without_self" ], "manifest_file_sha256": file_sha256(args.manifest), }, } write_canonical(args.compact_output, compact) print( json.dumps( { "reproduction": str(args.reproduction_output), "compact": str(args.compact_output), "raw_files": len(raw_files), "replay_exact": replay_exact, "localization": aggregate["gates"]["localization"][ "status" ], "compact_canonical_sha256": compact[ "canonical_sha256_without_self" ], }, ensure_ascii=False, indent=2, ) ) if __name__ == "__main__": main()