#!/usr/bin/env python3 """Seal prior artifacts and prepare exact frozen camera-only bounds, without media.""" from __future__ import annotations import argparse import json import shutil import time from pathlib import Path import torch from mcss.dynamic.types import hash_value from mcss.mechanism_pilot.direct_capacity_statistics import scene_values from mcss.mechanism_pilot.optimization_bounds_contracts import ( CURRENT_BOUNDS, FrozenTrainingPrior, context_camera_bundle, frozen_gt_free_bounds, ) from mcss.mechanism_pilot.small_training import sha, write_json OLD = Path("outputs/EXP-3D-DIRECT-STATE-CAPACITY-V1") OLD_DOCS = Path("docs/experiments/EXP-3D-DIRECT-STATE-CAPACITY-V1") CHECKPOINT = Path( "outputs/EXP-3D-20260927-centered-training-1000a-600b-v1/training/phase_b_final.pt" ) EXPECTED_CHECKPOINT = "be7b8b6d2ef366cad245732b9ff227802db596da65082ac0e160f24541579e42" METRICS = ( "depth_absrel", "depth_rmse", "depth_delta1", "rgb_mse", "rgb_ssim", "opacity", "coverage", ) def seal_previous(root): checks = [] for base, name in ((OLD, "artifact_manifest.json"), (OLD_DOCS, "report_manifest.json")): expected = json.loads((base / name).read_text()) for relative, digest in expected.items(): path = base / relative if sha(path) != digest: raise RuntimeError(f"Previous archive manifest mismatch: {path}") checks.append( { "manifest": str(base / name), "sha256": sha(base / name), "verified_files": len(expected), } ) source = json.loads((OLD / "source_manifest.json").read_text()) for name, digest in source.items(): if sha(Path(name)) != digest: raise RuntimeError(f"Previous frozen source changed: {name}") if sha(CHECKPOINT) != EXPECTED_CHECKPOINT: raise RuntimeError("Frozen carrier checkpoint changed") files = {} for base in (OLD, OLD_DOCS): for path in sorted(base.rglob("*")): if path.is_file(): files[str(path)] = { "sha256": sha(path), "bytes": path.stat().st_size, "mtime_ns": path.stat().st_mtime_ns, } for name, digest in source.items(): files[name] = { "sha256": digest, "bytes": Path(name).stat().st_size, "mtime_ns": Path(name).stat().st_mtime_ns, } report = { "status": "PASS", "created_unix": time.time(), "previous_manifests": checks, "previous_source_manifest_sha256": sha(OLD / "source_manifest.json"), "checkpoint_sha256": sha(CHECKPOINT), "files": files, "query_media_decoded": False, "holdout_media_read": False, "scope": ( "Only prior capacity output/doc artifacts and frozen source; no source datasets opened." ), } write_json(root / "audit/previous_experiment_seal.json", report) return report def prepare(root, docs): root, docs = Path(root), Path(docs) root.mkdir(parents=True, exist_ok=True) docs.mkdir(parents=True, exist_ok=True) (root / "audit").mkdir(exist_ok=True) seal = seal_previous(root) for name in ("scene_manifest.json", "train_depth_prior.json"): if (root / name).exists(): raise FileExistsError(f"Refusing to overwrite locked file: {root / name}") shutil.copy2(OLD / name, root / name) manifest = json.loads((root / "scene_manifest.json").read_text()) ids = [s["scene_id"] for s in manifest["scenes"]] if ( len(ids) != 17 or len(set(ids)) != 17 or set(ids) & set(manifest["FINAL_HOLDOUT_PROHIBITED"]) ): raise RuntimeError("Exposed scene roster invalid") raw_prior = json.loads((root / "train_depth_prior.json").read_text()) prior = FrozenTrainingPrior( raw_prior["near_m"], raw_prior["far_m"], sha(root / "train_depth_prior.json") ) previous = json.loads((OLD / "predeclared_geometry_bounds.json").read_text())["bounds"] bounds = {} for scene in manifest["scenes"]: for role in ("A", "B"): key = f"{scene['scene_id']}/{role}" bundle = context_camera_bundle(scene, role, manifest["image_size"]) alternative = frozen_gt_free_bounds(bundle, prior) old = torch.tensor(previous[key]["bounds"], dtype=torch.float64) if not torch.equal(old, alternative): raise RuntimeError(f"GT-free bounds no longer bit-exact: {key}") current = torch.tensor(CURRENT_BOUNDS, dtype=torch.float64) bounds[key] = { "scene_id": scene["scene_id"], "role": role, "context_ids": list(bundle.frame_ids), "camera_hash": hash_value(bundle.cameras), "anchor_hash": hash_value(bundle.anchor_c2w), "prior_hash": prior.source_sha256, "CURRENT_BOUNDS": current.tolist(), "FROZEN_GT_FREE_BOUNDS": alternative.tolist(), "current_bounds_hash": hash_value(current), "bounds_hash": hash_value(alternative), "previous_alternative_exact": True, } write_json( root / "bounds_contract.json", { "status": "FROZEN_EXACT_PREVIOUS_ALTERNATIVE", "inputs_allowed": [ "context_intrinsics", "context_extrinsics", "frozen_training_depth_prior", ], "inputs_forbidden": [ "context_depth_current_scene", "query_camera", "query_depth", "query_rgb", "carrier_prediction", "query_score", ], "rule": ( "all128x160 context unit rays at train q01/q99, anchor-camera coordinates, " "AABB5%per-side padding, minextent1m; float64 thenstateconsumerfloat32" ), "near_m": prior.near_m, "far_m": prior.far_m, "prior_hash": prior.source_sha256, "previous_bounds_artifact_sha256": sha(OLD / "predeclared_geometry_bounds.json"), "camera_hash_definition": "hash_value(tuple of float64 Cameras in fixed context order)", "bounds_hash_definition": "hash_value(float64[2,3] tensor)", "current_bounds": CURRENT_BOUNDS, "context_roles_unchanged": True, "source_sha256": sha(Path("src/mcss/mechanism_pilot/optimization_bounds_contracts.py")), }, ) write_json( root / "bounds_lock.json", { "status": "PASS", "n_scenes": 17, "n_contexts": 34, "scene_manifest_sha256": sha(root / "scene_manifest.json"), "prior_sha256": prior.source_sha256, "contract_sha256": sha(root / "bounds_contract.json"), "bounds": bounds, "current_scene_depth_used": False, "query_camera_or_GT_used": False, "all34_previous_alternatives_bit_exact": True, }, ) baseline = json.loads((OLD / "baseline_results.json").read_text()) shutil.copy2(OLD / "baseline_results.json", root / "baseline_results.json") old_results = json.loads((OLD / "direct_state_results.json").read_text()) comparisons = {} for name, methods in [("CARRIER", ("A", "B")), ("ANCHOR", ("anchor",)), ("PRIOR", ("prior",))]: rows = [dict(r, role=r["method"]) for r in baseline if r["method"] in methods] for metric in METRICS: values = scene_values(rows, metric) expected = old_results["frozen_carrier"][name]["metrics"][metric]["per_scene"] if values != expected: raise RuntimeError(f"Old raw baseline recomputation changed: {name}/{metric}") comparisons[f"{name}/{metric}"] = { "mean": sum(values.values()) / len(values), "per_scene": values, "exact": True, } # Reference previous direct 8/16/32 and oracle groups from numeric raw only. refs = {} context = json.loads((OLD / "raw/context_results.json").read_text()) oracle = json.loads((OLD / "raw/oracle_results.json").read_text()) for track, rows, block in [ ("RGBD", context, "context_only"), ("QUERY_ORACLE", oracle, "diagnostic_oracle"), ]: for grid in (8, 16, 32): selected = [ r for r in rows if r["track"] == track and r["grid"] == grid and r["samples"] == 64 and r["bounds_mode"] == "CURRENT_BOUNDS" and r["method"] == "direct" ] values = scene_values(selected, "depth_absrel") key = f"{track}|g{grid}|s64|CURRENT_BOUNDS|direct" if values != old_results[block][key]["metrics"]["depth_absrel"]["per_scene"]: raise RuntimeError(f"Old reference raw changed: {key}") refs[key] = { "per_scene": values, "mean": sum(values.values()) / len(values), "exact": True, } write_json( root / "baseline_reproduction.json", { "status": "PASS", "carrier_checkpoint_sha256": EXPECTED_CHECKPOINT, "baseline_rows": len(baseline), "baseline_metrics": comparisons, "previous_direct_references": refs, "source_raw_sha256": { str(OLD / n): sha(OLD / n) for n in [ "baseline_results.json", "raw/context_results.json", "raw/oracle_results.json", "direct_state_results.json", ] }, "scope": ( "Existing raw numeric recomputation only; no model, optimizer, " "query camera or media read." ), "FINAL_HOLDOUT_TOUCHED": False, }, ) return { "status": "PASS", "sealed_file_count": len(seal["files"]), "bounds_contexts": len(bounds), "scene_manifest": str(root / "scene_manifest.json"), "baseline_reproduction": "PASS_NUMERIC_RAW_EXACT", } if __name__ == "__main__": p = argparse.ArgumentParser(description=__doc__) p.add_argument("--root", required=True) p.add_argument("--docs", required=True) a = p.parse_args() print(json.dumps(prepare(a.root, a.docs), indent=2))