#!/usr/bin/env python3 """Freeze, run and analyze V28 (stream TTT under cross-dataset shift, Replica, GPU 4 + CPU). --stage freeze writes the contracts and the preregistration (CPU, CUDA hidden); --stage run on GPU 4 through cv_gpu_run.sh: Core B V8's sealed evaluation of every seed, an independent recomputation of every metric, the carrier-internal analysis, then (CPU) V27's 3- and 7-frame baselines and hybrids on this round's sealed predictions, the constant predictor, the check against REPLICA-TRANSFER-V1's rows, and the README. Run from the repository root under cv_host_run.py. Replica data stays on this host (research only). """ from __future__ import annotations import argparse import datetime import importlib.util import json import os import subprocess import sys from pathlib import Path import numpy as np import torch from PIL import Image from mcss.dynamic.types import hash_scene_state from mcss.mechanism_pilot.direct_capacity_statistics import describe, paired from mcss.mechanism_pilot.geometry_carrier_experiment import read, verify_data from mcss.mechanism_pilot.rgbd_stream_evaluation_v3 import preseal_depth_audit from mcss.mechanism_pilot.rgbd_stream_ttt_context_v6 import cohort_records from mcss.mechanism_pilot.rgbd_stream_ttt_v4 import ( RAYS_PER_FRAME, STREAM_LENGTH, load_frozen_carrier, stream_frame_ids, ) from mcss.mechanism_pilot.rgbd_stream_ttt_v7 import LEARNING_RATE from mcss.mechanism_pilot.rgbd_stream_ttt_v8 import evaluate_v8 from mcss.mechanism_pilot.small_training import sha, write_json from mcss.mechanism_pilot.statistics import measurement_metrics EXPERIMENT = "EXP-3D-RGBD-DYNAMIC-TTT-REPLICA-V28" COREB_V4 = Path("outputs/EXP-3D-RGBD-DYNAMIC-TTT-V4") COREB_V8 = Path("outputs/EXP-3D-RGBD-DYNAMIC-TTT-SOURCE-V8") V17 = Path("outputs/EXP-3D-RGBD-LONG-TRAIN-V17") REPLICA = Path("outputs/EXP-3D-RGBD-REPLICA-DATA-V1") TRANSFER = Path("outputs/EXP-3D-RGBD-REPLICA-TRANSFER-V1") CONCURRENT = ( "EXP-3D-RGBD-DYNAMIC-TTT-CARRIER-V10", "EXP-3D-RGBD-DYNAMIC-TTT-ONLINE-V11", "EXP-3D-RGBD-FAR-FOCUS-V26", ) PLAN = "PLAN_BEFORE_V28.md" PLAN_SHA256 = "8b154c923015e71bcdc8d35b7ab10cd48effe4cb9c2b9eb3f61bf84141c41174" CARRIER_STEP = 24000 COHORT = "REPLICA" EXPECTED_SPLIT = {COHORT: "REPLICA"} POLICY_SPECS = { "NO_STREAM": (False, None, 0), "OFF": (True, None, 0), "OFF_CTX_TTT_100": (True, "context", 100), "TTT_100": (True, "stream", 100), "TTT_1000": (True, "stream", 1000), } CONTROLS = {"TTT_100_WRONG_SCENE": "TTT_100"} SPECS = dict(POLICY_SPECS) GPU = "4" LAUNCHER = Path.home() / "cv_tools/cv_host_run.py" THREADS = { "OMP_NUM_THREADS": "1", "OPENBLAS_NUM_THREADS": "1", "MKL_NUM_THREADS": "1", "ATEN_CPU_CAPABILITY": "avx2", "MKL_CBWR": "AVX2", "DNNL_MAX_CPU_ISA": "AVX2", "NVIDIA_TF32_OVERRIDE": "0", } CONTRASTS = { "WRITE_UNDER_SHIFT": ("OFF", "TTT_100"), "STREAM_DOSE_1000": ("TTT_100", "TTT_1000"), "WRITE_1000_UNDER_SHIFT": ("OFF", "TTT_1000"), "CONTEXT_TTT_WITH_STREAM_100": ("OFF", "OFF_CTX_TTT_100"), "TTT_100_SPECIFICITY": ("TTT_100_WRONG_SCENE", "TTT_100"), "STREAM_CACHE_EFFECT": ("NO_STREAM", "OFF"), } HYBRID_CONTRASTS = { "HYBRID_VS_MODEL_FREE": ("REPROJ7_HARMONIC", "HFILL8_7_TTT_100"), "AHFILL8_VS_MODEL_FREE": ("REPROJ7_HARMONIC", "AHFILL8_7_TTT_100"), "HYBRID_NORM_VS_MODEL_FREE": ("REPROJ7_HARMONIC", "HFILL8_7N_TTT_100"), "TTT_GAIN_IN_HYBRID": ("HFILL8_7_OFF", "HFILL8_7_TTT_100"), "CARRIER_TTT_100_VS_MODEL_FREE": ("REPROJ7_HARMONIC", "CARRIER_TTT_100"), "CARRIER_TTT_100_VS_CONSTANT": ("CONSTANT", "CARRIER_TTT_100"), "MODEL_FREE_VS_CONSTANT": ("CONSTANT", "REPROJ7_HARMONIC"), "STREAM_FRAMES_FOR_MODEL_FREE": ("REPROJ3_HARMONIC", "REPROJ7_HARMONIC"), } DRAWS, BOOT_SEED = 10000, 20260928 METRICS = ("rgb_mse", "depth_absrel", "depth_rmse", "depth_delta1", "opacity", "coverage") KEYS = ("seed", "cohort", "scene_id", "role", "query_id", "policy") SCRIPTS = Path("scripts") def module(name, filename): spec = importlib.util.spec_from_file_location(name, SCRIPTS / filename) loaded = importlib.util.module_from_spec(spec) spec.loader.exec_module(loaded) return loaded def v3_analysis(): """Core B V3's scene aggregation and frozen V2 gate (as in Core B V4-V11 and V24-V27).""" return module("coreb_v3_analysis_v28", "analyze_rgbd_stream_write_v3.py") def hybrid_module(root): """V27's frozen baseline and hybrid code with this round as its only source.""" loaded = module("v27_hybrid_for_v28", "analyze_rgbd_hybrid_confirm_v27.py") loaded.ROUNDS = {COHORT: Path(root)} loaded.EXPECTED_SPLIT = dict(EXPECTED_SPLIT) loaded.V17 = V17 return loaded def resolve(path, base): path = Path(path) return path if path.is_absolute() else Path(base) / path def load_cohorts(root): item = read(Path(root) / "cohort_lock.json")[COHORT] if sha(Path(item["manifest"])) != item["manifest_sha256"]: raise PermissionError(f"{COHORT} manifest changed") return {COHORT: (read(Path(item["manifest"])), Path(item["manifest"]).parent)} def verify_lock_v28(root): lock = read(Path(root) / "preregistration.json") if lock["status"] != "FROZEN_BEFORE_FORMAL_TRAINING": raise PermissionError("Formal preregistration absent") for section in ("input_sha256", "source_sha256"): for name, digest in lock[section].items(): if sha(Path(name)) != digest: raise PermissionError(f"Frozen input/source changed: {name}") return read(Path(root) / "evaluation_contract.json") def seal_previous(root): files = {} for base in (Path("outputs"), Path("docs/experiments")): for path in sorted(base.rglob("*")): if path.is_file() and not {EXPERIMENT, *CONCURRENT} & set(path.parts): files[str(path)] = {"sha256": sha(path), "bytes": path.stat().st_size} write_json( root / "audit/previous_experiment_seal.json", { "created_utc": datetime.datetime.now(datetime.UTC).isoformat(), "excluded_concurrent": list(CONCURRENT), "files": files, }, ) return len(files) def replica_manifest(): """The prepared Replica manifest, its preparation integrity PASS.""" if read(REPLICA / "preparation_integrity.json")["status"] != "PASS": raise PermissionError("Replica preparation must PASS") path = (REPLICA / "manifest_replica.json").resolve() return {"manifest": str(path), "manifest_sha256": sha(path)} def freeze(root): root = Path(root).resolve() if root.name != EXPERIMENT or (root / "preregistration.json").exists(): raise PermissionError("Exact, unfrozen V28 directory required") if not (root / "PROTOCOL.md").is_file() or sha(root / PLAN) != PLAN_SHA256: raise PermissionError("The V28 plan and protocol must be in place") for done in (COREB_V4, COREB_V8): if read(done / "integrity.json")["status"] != "PASS": raise PermissionError(f"{done.name} must be finalized") if read(COREB_V4 / "selection.json")["settings"]["learning_rate"] != LEARNING_RATE: raise PermissionError("TTT must use the learning rate Core B V4 selected on TRAIN") if sha(LAUNCHER) != sha(Path("../_archives_cv/cv_host_run.py")): raise PermissionError("The path-mapping launcher must equal the migrated one") carriers = read(COREB_V4 / "carrier_lock.json") for seed, item in carriers["carriers"].items(): if sha(Path(item["path"])) != item["sha256"] or item["step"] != CARRIER_STEP: raise PermissionError(f"V17 step-{CARRIER_STEP} carrier changed: {seed}") load_frozen_carrier(item["path"], "cpu") write_json(root / "carrier_lock.json", carriers) write_json(root / "cohort_lock.json", {COHORT: replica_manifest()}) entries = cohort_records(load_cohorts(root), EXPECTED_SPLIT) episodes = {} for sid, (_, record, _, _) in sorted(entries.items()): for role in ("A", "B"): ids = stream_frame_ids(record, role) if ids is None: raise PermissionError(f"{sid}/{role} lacks a full stream") episodes[f"{sid}/{role}"] = list(ids) write_json( root / "stream_roles.json", { "rule": "Core B V1-V11 frame-id rule", "stream_length": STREAM_LENGTH, "episodes": episodes, }, ) eval_contract = read(V17 / "training_contract.json")["eval"] config = { "experiment": EXPERIMENT, "seeds": sorted(int(s) for s in carriers["carriers"]), "policies": {name: list(spec) for name, spec in SPECS.items()}, "controls": CONTROLS, "learning_rate": LEARNING_RATE, "rays_per_supervised_frame": RAYS_PER_FRAME, "stream_length": STREAM_LENGTH, "cohort": {"name": COHORT, "scenes": len(entries), "expected_split": "REPLICA"}, "baselines": "V27's frozen 3-/7-frame reprojection + harmonic code on this round's seals", "reprojection_fill_constant_m": eval_contract["reprojection_fill_constant_m"], "near_px": eval_contract["near_px"], "transfer_rows": { "path": str((TRANSFER / "raw/query_rows.json").resolve()), "sha256": sha(TRANSFER / "raw/query_rows.json"), }, "num_threads": 1, "thread_environment": THREADS, "cpu_affinity": list(range(56, 64)), "device": "cuda", "gpu": GPU, "host_launcher": {"path": str(LAUNCHER), "sha256": sha(LAUNCHER)}, } write_json(root / "evaluation_contract.json", config) write_json( root / "decision_rules.json", { "aggregation": "per seed: scene mean of role means of query AbsRel; seeds equally " "weighted (model-free baselines take the same value in every seed)", "bootstrap": {"unit": "scene", "draws": DRAWS, "seed": BOOT_SEED}, "gate": "frozen V2 primary-contrast checks (as Core B V2-V11 and V24-V27)", "P1_WRITE_UNDER_SHIFT_STATUS": "gate on AbsRel(OFF) - AbsRel(TTT_100)", "P2_HYBRID_VS_MODEL_FREE_STATUS": "gate on AbsRel(REPROJ7_HARMONIC) - " "AbsRel(HFILL8_7(TTT_100))", "interpretation": { "A": "both SUPPORTED", "B": "only P1 SUPPORTED", "C": "only P2 SUPPORTED", "D": "neither SUPPORTED", }, "descriptive": [*CONTRASTS, *HYBRID_CONTRASTS], "final_holdout": "never opened", }, ) sealed = seal_previous(root) data_hashes = {} for _, record, _, base in entries.values(): for frame in record["frames"]: for key in ("rgb", "depth"): path = resolve(frame[key], base) data_hashes[str(path)] = sha(path) write_json(root / "audit/data_hashes.json", data_hashes) names = ( "carrier_lock.json", "cohort_lock.json", "stream_roles.json", "evaluation_contract.json", "decision_rules.json", "PROTOCOL.md", PLAN, "audit/data_hashes.json", "audit/previous_experiment_seal.json", ) sources = sorted(Path("src/mcss").rglob("*.py")) sources += [ SCRIPTS / "run_rgbd_stream_ttt_replica_v28.py", SCRIPTS / "analyze_rgbd_hybrid_confirm_v27.py", SCRIPTS / "analyze_rgbd_stream_write_v3.py", SCRIPTS / "run_rgbd_geometric_baselines.py", SCRIPTS / "run_rgbd_inpaint_baselines.py", LAUNCHER, ] write_json( root / "preregistration.json", { "status": "FROZEN_BEFORE_FORMAL_TRAINING", "experiment": EXPERIMENT, "input_sha256": {str((root / n).resolve()): sha(root / n) for n in names}, "source_sha256": {str(p.resolve()): sha(p) for p in sources}, "data_sha256": data_hashes, "formal_training_started": False, }, ) print("FROZEN_V28", len(entries), "scenes", len(episodes), "episodes", sealed, "files sealed") def gate_contrasts(rows, contrasts, seeds): v3 = v3_analysis() out, per_seed = {}, {str(s): {} for s in seeds} for name, (a, b) in contrasts.items(): left, right = v3.scene_values(rows, a), v3.scene_values(rows, b) stats = paired(left, right, draws=DRAWS, seed=BOOT_SEED) status, _ = v3.gate(stats, True) out[name] = { "mean": stats["mean"], "ci95": stats["ci95"], "positive_tied_negative": [stats["positive"], stats["tied"], stats["negative"]], "gate_descriptive": status, } for seed in seeds: per_seed[str(seed)][name] = float( np.mean(list(v3.scene_values(rows, a, seed=seed).values())) ) - float(np.mean(list(v3.scene_values(rows, b, seed=seed).values()))) absolute = { p: describe(v3.scene_values(rows, p), draws=DRAWS, seed=BOOT_SEED)["mean"] for p in sorted({r["policy"] for r in rows}) } return {"absolute": absolute, "contrasts": out, "per_seed_means": per_seed} def branch(p1, p2): first, second = p1 == "SUPPORTED", p2 == "SUPPORTED" return {(True, True): "A", (True, False): "B", (False, True): "C", (False, False): "D"}[ first, second ] def recompute(root, cohorts, rows): """Every metric from the saved prediction and GT; NORM rows; the seals.""" entries = cohort_records(cohorts, EXPECTED_SPLIT) worst, norm_rows = 0.0, [] for row in rows: path = Path(root) / "raw/eval_ttt" / f"seed_{row['seed']}" / row["prediction_path"] if sha(path) != row["prediction_file_sha256"]: raise AssertionError("Prediction changed") pred = dict(np.load(path)) _, record, _, base = entries[row["scene_id"]] frame = next(f for f in record["frames"] if f["frame_id"] == row["query_id"]) with Image.open(resolve(frame["rgb"], base)) as image: rgb = np.array(image.convert("RGB"), dtype=np.float32) / 255 gt = np.load(resolve(frame["depth"], base)).squeeze() metrics = measurement_metrics(pred["rgb"], rgb, pred["depth"], gt, pred["opacity"]) for key in METRICS: worst = max(worst, abs(metrics[key] - row[key])) depth = pred["depth"] / np.maximum(pred["opacity"], 1e-3) norm = measurement_metrics(pred["rgb"], rgb, depth, gt, pred["opacity"]) norm_rows.append({**{k: row[k] for k in KEYS}, "depth_absrel": norm["depth_absrel"]}) if worst > 1e-6: raise AssertionError(f"Metric mismatch {worst}") records = {sid: item[1] for sid, item in entries.items()} for seed in {r["seed"] for r in rows}: out = Path(root) / "raw/eval_ttt" / f"seed_{seed}" audit = read(out / "integrity.json") if sha(out / "states.pt") != audit["states_file_sha256"]: raise AssertionError("Sealed states changed") states = torch.load(out / "states.pt", map_location="cpu", weights_only=False) index = read(out / "state_hashes.json") if any(hash_scene_state(s) != index[k]["state_hash"] for k, s in states.items()): raise AssertionError("State hash mismatch") access = read(out / "GT_access.json") marker = next( i for i, e in enumerate(access) if e.get("event") == "ALL_STREAM_STATES_SEALED" ) preseal_depth_audit(access[:marker], records) return worst, norm_rows def hybrid_part(root, config, cohorts, seeds): """V27's baselines and hybrids on this round's seals, plus the constant predictor.""" v27 = hybrid_module(root) cohorts27 = v27.load_cohorts() hybrid_root = Path(root) / "hybrid" seal = v27.compute_baselines(hybrid_root, config, cohorts27) rows27, shares, worst, _ = v27.score(cohorts27, seal) entries = cohort_records(cohorts, EXPECTED_SPLIT) constant = config["reprojection_fill_constant_m"] extra, transfer = [], {} for item in read(config["transfer_rows"]["path"]): if item["method"] in ("REPROJ_HARMONIC", "CONSTANT"): transfer[item["method"], item["scene_id"], item["role"], item["query_id"]] = item[ "absrel_ALL" ] worst_transfer, compared = 0.0, 0 for row in rows27["ALL"]: if row["policy"] != "REPROJ3_HARMONIC" or row["seed"] != seeds[0]: continue key = ("REPROJ_HARMONIC", row["scene_id"], row["role"], row["query_id"]) if key in transfer: worst_transfer = max(worst_transfer, abs(transfer[key] - row["depth_absrel"])) compared += 1 for sid, (_, record, _, base) in entries.items(): for role in ("A", "B"): for qid in record["roles"]["primary_query"]: frame = next(f for f in record["frames"] if f["frame_id"] == qid) gt = np.load(resolve(frame["depth"], base)).squeeze() valid = np.isfinite(gt) & (gt > 0) value = float(np.mean(np.abs(constant - gt[valid]) / gt[valid])) key = ("CONSTANT", sid, role, qid) if key in transfer and seeds: worst_transfer = max(worst_transfer, abs(transfer[key] - value)) compared += 1 for seed in seeds: extra.append( { "seed": seed, "cohort": COHORT, "scene_id": sid, "role": role, "query_id": qid, "policy": "CONSTANT", "depth_absrel": value, } ) all_rows = rows27["ALL"] + extra write_json(Path(root) / "raw/hybrid_all_rows.json", all_rows) write_json(Path(root) / "raw/hybrid_far7_rows.json", rows27["FAR7"]) far7 = gate_contrasts( rows27["FAR7"], {k: v for k, v in HYBRID_CONTRASTS.items() if "CONSTANT" not in v}, seeds, ) return { "all": gate_contrasts(all_rows, HYBRID_CONTRASTS, seeds), "far7": far7, "near7_share": float(np.mean([s["near7_share"] for s in shares])), "near3_share": float(np.mean([s["near3_share"] for s in shares])), "carrier_absrel_max_diff": worst, "transfer_check": {"compared": compared, "max_abs_difference": worst_transfer}, } def stat(value): lo, hi = value["ci95"] pos, tie, neg = value["positive_tied_negative"] return f"{value['mean']:+.4f} [{lo:+.4f}, {hi:+.4f}]，场景 {pos}/{tie}/{neg}" def report(root, result): main, hyb = result["carrier"], result["hybrid"] c, h = main["contrasts"], hyb["all"]["contrasts"] lines = [ f"# {EXPERIMENT}", "", "跨数据集（Replica，8 个场景）的流帧写入，并与只用同样 7 帧的无模型基线比较。" "carrier 是 V17 L 第 24000 步，只在 Hypersim 上训练过。" "计划见 [PLAN_BEFORE_V28.md](PLAN_BEFORE_V28.md)。" "状态构造与渲染在 GPU（4 号卡，关闭 TF32）上，基线与计分在 CPU 上。" "Replica 只允许研究用途，数据不进入仓库。", "", f"- **P1 WRITE_UNDER_SHIFT（OFF − TTT_100）：** {stat(c['WRITE_UNDER_SHIFT'])}，" f"{result['statuses']['P1_WRITE_UNDER_SHIFT_STATUS']}", "- **P2 HYBRID_VS_MODEL_FREE（REPROJ7_HARMONIC − HFILL8_7(TTT_100)）：** " f"{stat(h['HYBRID_VS_MODEL_FREE'])}，{result['statuses']['P2_HYBRID_VS_MODEL_FREE_STATUS']}", f"- **分支 {result['INTERPRETATION_BRANCH']}**", f"- **像素比例：** 3 帧 NEAR {100 * hyb['near3_share']:.1f}%，" f"7 帧 NEAR {100 * hyb['near7_share']:.1f}%。", "", "| carrier 内部的对比（正值：后者更好） | 全部像素 | 归一化读出 |", "|---|---|---|", ] for name in CONTRASTS: lines.append( f"| {name} | {stat(c[name])} | {stat(result['normalized']['contrasts'][name])} |" ) lines += ["", "| 方法 | AbsRel |", "|---|---|"] for name, value in sorted({**main["absolute"], **hyb["all"]["absolute"]}.items()): lines.append(f"| {name} | {value:.4f} |") lines += ["", "| 与无模型基线的对比 | 全部像素 | FAR7 |", "|---|---|---|"] for name in HYBRID_CONTRASTS: far = hyb["far7"]["contrasts"].get(name) lines.append(f"| {name} | {stat(h[name])} | {stat(far) if far else '—'} |") check = hyb["transfer_check"] lines += [ "", "- **每个 seed（P1）：** " + "；".join( f"{k}: {v['WRITE_UNDER_SHIFT']:+.4f}" for k, v in main["per_seed_means"].items() ), f"- **核对：** 3 帧调和插值与常数预测的 AbsRel 与 REPLICA-TRANSFER-V1 的结果行比较 " f"{check['compared']} 个，最大差 {check['max_abs_difference']:.1e}；" f"carrier 的 AbsRel 与本轮结果行最大差 {hyb['carrier_absrel_max_diff']:.1e}；" f"指标重新计算的最大误差 {result['max_metric_error']:.1e}。", "", "说明：这是 Replica 的第二次使用（第一次是 REPLICA-TRANSFER-V1 的静态迁移）；" "只有 8 个场景，功效有限。受保护的 Hypersim final holdout 未打开。", ] (Path(root) / "README.md").write_text("\n".join(lines) + "\n") def run(root): root = Path(root).resolve() config = verify_lock_v28(root) verify_data(root) if read(root / "tests.json")["status"] != "PASS": raise PermissionError("Full suite must PASS before the run") if not os.sched_getaffinity(0) <= set(config["cpu_affinity"]): raise PermissionError("Run must be pinned to the frozen CPU cores (taskset)") if os.environ.get("CUDA_VISIBLE_DEVICES") != config["gpu"]: raise PermissionError("Run must hold the frozen GPU through cv_gpu_run.sh") if sha(LAUNCHER) != config["host_launcher"]["sha256"] or not os.environ.get("CV_ROOT"): raise PermissionError("The frozen launcher and CV_ROOT are required") for key, value in config["thread_environment"].items(): if os.environ.get(key) != value: raise PermissionError(f"Thread environment {key} must be {value}") if sha(Path(config["transfer_rows"]["path"])) != config["transfer_rows"]["sha256"]: raise PermissionError("REPLICA-TRANSFER-V1 rows changed") torch.set_num_threads(1) lock = read(root / "carrier_lock.json")["carriers"] carriers = {int(s): load_frozen_carrier(item["path"], "cuda")[0] for s, item in lock.items()} cohorts = load_cohorts(root) rows = [] for seed, carrier in sorted(carriers.items()): out = root / "raw/eval_ttt" / f"seed_{seed}" seed_rows = evaluate_v8( carrier, cohorts, EXPECTED_SPLIT, out, seed, "cuda", SPECS, CONTROLS ) rows.extend(seed_rows) print("EVALUATED", seed, len(seed_rows), flush=True) write_json(root / "raw/eval_ttt_query_results.json", rows) worst, norm_rows = recompute(root, cohorts, rows) seeds = sorted(carriers) carrier_part = gate_contrasts(rows, CONTRASTS, seeds) normalized = gate_contrasts(norm_rows, CONTRASTS, seeds) hybrid = hybrid_part(root, config, cohorts, seeds) statuses = { "P1_WRITE_UNDER_SHIFT_STATUS": carrier_part["contrasts"]["WRITE_UNDER_SHIFT"][ "gate_descriptive" ], "P2_HYBRID_VS_MODEL_FREE_STATUS": hybrid["all"]["contrasts"]["HYBRID_VS_MODEL_FREE"][ "gate_descriptive" ], } result = { "statuses": statuses, "INTERPRETATION_BRANCH": branch(*statuses.values()), "carrier": carrier_part, "normalized": normalized, "hybrid": hybrid, "max_metric_error": worst, "max_rgb_clamp_excess": max(r["rgb_clamp_max_excess"] for r in rows), } write_json(root / "ttt_replica_results.json", result) write_json(root / "raw/norm_rows.json", norm_rows) old = read(root / "audit/previous_experiment_seal.json")["files"] changed = [p for p, info in old.items() if sha(Path(p)) != info["sha256"]] if changed: raise AssertionError(f"Old experiment files changed: {changed[:3]}") write_json( root / "integrity.json", { "status": "PASS", "metrics_recomputed": len(rows), "max_metric_error": worst, "old_files_unchanged": len(old), "settings_reselected": False, "slow_carrier_updated": False, "final_holdout_touched": False, "commit": subprocess.check_output(["git", "rev-parse", "HEAD"], text=True).strip(), "finished_utc": datetime.datetime.now(datetime.UTC).isoformat(), }, ) report(root, result) print("V28_COMPLETE", result["INTERPRETATION_BRANCH"], json.dumps(statuses), flush=True) if __name__ == "__main__": parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--root", required=True) parser.add_argument("--stage", choices=("freeze", "run"), required=True) args = parser.parse_args() if args.stage == "freeze": freeze(args.root) else: run(args.root) sys.exit(0)