scripts/audit_direct_capacity.py
"""Post-run direct-capacity audit from states, optimization logs and saved predictions.
No optimizer or renderer is invoked. Both phases must be complete before any audit
GT bytes are read, and protected holdout identities are permanently excluded.
"""
import argparse
import json
from collections import defaultdict
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.small_training import sha, write_json
from mcss.mechanism_pilot.statistics import measurement_metrics
METRICS = (
"rgb_mse",
"rgb_ssim",
"depth_absrel",
"depth_rmse",
"depth_delta1",
"opacity",
"coverage",
)
CHECKPOINT_HASH = "be7b8b6d2ef366cad245732b9ff227802db596da65082ac0e160f24541579e42"
def read(path):
return json.loads(Path(path).read_text())
def location(root, path):
p = Path(path)
return p if p.is_absolute() else root / p
def check_hashes(mapping):
for path, digest in mapping.items():
assert sha(Path(path)) == digest, f"Frozen file changed: {path}"
def run(root, checkpoint):
root = Path(root)
destination = root / "audit/postrun_validation.json"
if destination.exists():
raise FileExistsError(destination)
specs = read(root / "state_plan.json")["states"]
assert len(specs) == 425 and len({s["key"] for s in specs}) == 425
manifest = read(root / "scene_manifest.json")
records = {r["scene_id"]: r for r in manifest["scenes"]}
holdout = set(manifest["FINAL_HOLDOUT_PROHIBITED"])
assert len(records) == 17 and len(holdout) == 12 and not set(records) & holdout
assert all(r["cohort_role"] == "CAPACITY_DEV_EXPOSED" for r in records.values())
preregistration = read(root / "preregistration.json")
assert preregistration["config_sha256"] == sha(root / "config.json")
assert preregistration["state_plan_sha256"] == sha(root / "state_plan.json")
config = read(root / "config.json")
assert config["state_plan_sha256"] == sha(root / "state_plan.json")
assert config["scene_manifest_sha256"] == sha(root / "scene_manifest.json")
check_hashes(config["source_sha256"])
# All context and privileged-oracle optimization/evaluation phases must already exist.
for phase in ("context", "oracle"):
complete = read(root / f"{phase}_optimization_complete.json")
assert complete["status"] == "PASS" and complete["phase"] == phase
assert complete["states"] == sum(s["phase"] == phase for s in specs)
check_hashes(complete["input_hashes"])
integrity = read(root / "raw" / f"{phase}_evaluation_integrity.json")
assert integrity["status"] == "PASS" and not integrity["final_holdout_touched"]
assert not integrity["dynamic_ttt_run"] and not integrity["new_carrier_trained"]
assert integrity["all_states_immutable"] and integrity["all_frozen_inputs_unchanged"]
def optimizer_directory(spec):
path = location(root, spec["optimization_path"])
return path if path.is_dir() else path.parent
context_completion_times = {
spec["key"]: (optimizer_directory(spec) / "completion.json").stat().st_mtime_ns
for spec in specs
if spec["phase"] == "context"
}
context_eval_lock_time = (root / "raw/context_evaluation_lock.json").stat().st_mtime_ns
context_eval_integrity_time = (
(root / "raw/context_evaluation_integrity.json").stat().st_mtime_ns
)
oracle_lock_times = {
spec["key"]: (optimizer_directory(spec) / "lock.json").stat().st_mtime_ns
for spec in specs
if spec["phase"] == "oracle"
}
oracle_complete_time = (root / "oracle_optimization_complete.json").stat().st_mtime_ns
oracle_eval_lock_time = (root / "raw/oracle_evaluation_lock.json").stat().st_mtime_ns
assert all(t <= context_eval_lock_time for t in context_completion_times.values())
assert context_eval_integrity_time <= min(oracle_lock_times.values())
assert oracle_complete_time <= oracle_eval_lock_time
chronology = {
"unit": "filesystem st_mtime_ns",
"limitation": (
"Auxiliary chronology evidence only; "
"OS file timestamps are mutable and not tamper-proof"
),
"latest_context_state_completion": max(context_completion_times.values()),
"context_evaluation_lock": context_eval_lock_time,
"context_evaluation_integrity": context_eval_integrity_time,
"first_oracle_optimization_lock": min(oracle_lock_times.values()),
"oracle_optimization_complete": oracle_complete_time,
"oracle_evaluation_lock": oracle_eval_lock_time,
"context_completion_before_context_evaluation": True,
"context_evaluation_before_oracle_optimization": True,
"oracle_completion_before_oracle_evaluation": True,
}
baseline = read(root / "baseline_integrity.json")
assert baseline["status"] == "PASS" and baseline["checkpoint_sha256"] == CHECKPOINT_HASH
assert sha(Path(checkpoint)) == CHECKPOINT_HASH
state_hashes, optimization_locks, optimization_checks = {}, {}, []
for spec in specs:
sid, phase, track = spec["scene_id"], spec["phase"], spec["track"]
assert sid in records and sid not in holdout
directory = location(root, spec["optimization_path"])
if not directory.is_dir():
directory = directory.parent
completed, summary, locked = (
read(directory / name) for name in ("completion.json", "summary.json", "lock.json")
)
assert completed["spec"] == spec and completed["phase"] == phase
assert completed["plan_sha256"] == sha(root / "state_plan.json")
assert completed["state_file_sha256"] == sha(location(root, spec["state_path"]))
assert completed["source_hashes"] == config["source_sha256"]
assert locked["config"] == config["optimization"]
assert locked["grid"] == spec["grid"] and locked["renderer_samples"] == spec["samples"]
assert np.allclose(locked["bounds"], spec["bounds"], atol=1e-6, rtol=1e-6)
assert locked["renderer_kernel_unchanged"] and locked["carrier_loaded"] is False
record = records[sid]
query_ids = record["roles"]["primary_query"]
if phase == "context":
expected_ids = record["roles"]["context_a" if spec["role"] == "A" else "context_b"]
assert track in ("RGBD", "RGB_ONLY") and set(expected_ids).isdisjoint(query_ids)
else:
expected_ids = query_ids
assert track == "QUERY_ORACLE"
assert locked["supervision_frame_ids"] == expected_ids
assert summary["shared_supervision_frame_ids"] == expected_ids
assert locked["has_depth"] == summary["has_depth"] == (track != "RGB_ONLY")
accesses = read(directory / "access.json")
for event in accesses:
assert event["scene_id"] == sid and sid not in holdout
if phase == "context":
assert event["frame_id"] in expected_ids
assert event["purpose"] == f"CONTEXT_ONLY_{track}"
if track == "RGB_ONLY":
assert "depth" not in event["channels"]
else:
if event["frame_id"] in query_ids:
assert event["purpose"] == "DIAGNOSTIC_QUERY_SUPERVISED_ORACLE_NOT_CONTEXT_ONLY"
else:
assert event["frame_id"] == record["roles"]["context_a"][0]
assert (
event["purpose"] == "CONTEXT_ONLY_RGB_ONLY"
and "depth" not in event["channels"]
)
checks = summary["full_objective_checkpoints"]
expected_steps = [0] + list(
range(
config["optimization"]["checkpoint_interval"],
config["optimization"]["steps"] + 1,
config["optimization"]["checkpoint_interval"],
)
)
if expected_steps[-1] != config["optimization"]["steps"]:
expected_steps.append(config["optimization"]["steps"])
assert [c["step"] for c in checks] == expected_steps
assert all(np.isfinite(c["context_objective"]) for c in checks)
best = min(checks, key=lambda c: (c["context_objective"], c["step"]))
assert best["step"] == summary["selected_step"]
assert np.isclose(
best["context_objective"],
summary["selected"]["context_objective"],
atol=1e-6,
rtol=1e-5,
)
assert summary["selection_used_only_supervision_objective"] and summary["all_steps_finite"]
state = torch.load(
location(root, spec["state_path"]), map_location="cpu", weights_only=False
)
assert hash_scene_state(state) == summary["state_hash"]
assert state.features is None and state.density_logits.shape[-3:] == (spec["grid"],) * 3
state_hashes[spec["key"]] = summary["state_hash"]
optimization_locks[spec["key"]] = locked
optimization_checks.append(
{
"key": spec["key"],
"selected_step": best["step"],
"earliest_minimum_objective": best["context_objective"],
}
)
# Inspect the actual frozen plan/optimization locks, not only synthetic config contracts.
for sid in records:
for role in ("A", "B"):
for track in ("RGBD", "RGB_ONLY"):
group = [
s
for s in specs
if s["phase"] == "context"
and s["scene_id"] == sid
and s["role"] == role
and s["track"] == track
and s["bounds_mode"] == "CURRENT_BOUNDS"
and s["samples"] == 64
]
assert sorted(s["grid"] for s in group) == [8, 16, 32]
reference = optimization_locks[group[0]["key"]]
for spec in group[1:]:
locked = optimization_locks[spec["key"]]
for field in (
"config",
"seed",
"bounds",
"renderer_samples",
"supervision_frame_ids",
"has_depth",
"sampling",
):
assert locked[field] == reference[field]
assert (
records[spec["scene_id"]]["roles"]["primary_query"]
== records[sid]["roles"]["primary_query"]
)
diagnostic = [s for s in specs if s["scene_id"] == sid and s["samples"] == 128]
assert len(diagnostic) == 4
for spec in diagnostic:
assert spec["track"] == "RGBD" and spec["grid"] in (8, 16)
assert spec["bounds_mode"] == "CURRENT_BOUNDS" and spec["phase"] == "context"
reference_spec = next(
s
for s in specs
if s["scene_id"] == sid
and s["role"] == spec["role"]
and s["track"] == "RGBD"
and s["grid"] == spec["grid"]
and s["samples"] == 64
and s["bounds_mode"] == "CURRENT_BOUNDS"
and s["phase"] == "context"
)
for field in (
"config",
"seed",
"bounds",
"supervision_frame_ids",
"has_depth",
"sampling",
):
assert (
optimization_locks[spec["key"]][field]
== optimization_locks[reference_spec["key"]][field]
)
audit_access, truth_cache, phases = [], {}, {}
for phase in ("context", "oracle"):
phase_specs = [s for s in specs if s["phase"] == phase]
seal_events = read(root / "raw" / f"{phase}_seal_events.json")
n = len(phase_specs)
assert all(e["event"] == "seal" for e in seal_events[:n])
assert {e["state_id"] for e in seal_events[:n]} == {s["key"] for s in phase_specs}
assert all(
e["event"] == "query_camera_GT" and e["scene_id"] not in holdout
for e in seal_events[n:]
)
for event in seal_events[:n]:
assert event["state_hash"] == state_hashes[event["state_id"]]
lock = read(root / "raw" / f"{phase}_evaluation_lock.json")
assert lock["all_states_sealed_before_evaluation"] and not lock["final_holdout_touched"]
check_hashes(lock["input_sha256"])
check_hashes(lock["optimizer_source_sha256"])
check_hashes(lock["evaluator_source_sha256"])
access = read(root / "raw" / f"{phase}_GT_access.json")
assert all(e["scene_id"] in records and e["scene_id"] not in holdout for e in access)
assert all(
e["purpose"]
in (
"HASH_INPUT_AFTER_ALL_STATES_SEALED",
"POST_SEAL_CONTEXT_EVALUATION_NOT_OPTIMIZER",
"SEALED_QUERY_EVALUATION",
)
for e in access
)
rows = read(root / "raw" / f"{phase}_results.json")
direct_rows = defaultdict(list)
groups = defaultdict(list)
errors, exact = dict.fromkeys(METRICS, 0.0), dict.fromkeys(METRICS, 0)
for row in rows:
sid, fid = row["scene_id"], row["query_id"]
assert sid in records and sid not in holdout
assert fid in records[sid]["roles"]["primary_query"]
groups[row["key"], fid].append(row)
if row["method"] == "direct":
direct_rows[row["key"]].append(row)
assert row["used_state_hash"] == state_hashes[row["key"]]
if row["method"] == "wrong_scene":
assert row["donor_scene_id"] != sid and row["donor_scene_id"] in records
donor = next(
s
for s in phase_specs
if s["scene_id"] == row["donor_scene_id"]
and all(
s[k] == row[k] for k in ("role", "track", "grid", "samples", "bounds_mode")
)
)
assert row["used_state_hash"] == state_hashes[donor["key"]]
if (sid, fid) not in truth_cache:
frame = next(f for f in records[sid]["frames"] if f["frame_id"] == fid)
rgb_path, depth_path = (location(root, frame[k]) for k in ("rgb", "depth"))
assert not any(
identity in str(rgb_path) or identity in str(depth_path) for identity in holdout
)
with Image.open(rgb_path) as im:
rgb = np.asarray(im.convert("RGB"), dtype=np.float32) / 255
depth = np.load(depth_path).squeeze()
truth_cache[sid, fid] = rgb, depth
audit_access.append(
{
"scene_id": sid,
"query_id": fid,
"purpose": "POSTRUN_NPZ_RECOMPUTATION",
"rgb_path": str(rgb_path),
"depth_path": str(depth_path),
"final_holdout": False,
}
)
rgb, depth = truth_cache[sid, fid]
with np.load(location(root, row["prediction_path"])) as pred:
metrics = measurement_metrics(
pred["rgb"], rgb, pred["depth"], depth, pred["opacity"]
)
for metric in METRICS:
difference = abs(metrics[metric] - row[metric])
errors[metric] = max(errors[metric], difference)
exact[metric] += metrics[metric] == row[metric]
assert np.isclose(metrics[metric], row[metric], atol=1e-6, rtol=1e-5)
assert len({(r["key"], r["query_id"], r["method"]) for r in rows}) == len(rows)
expected_row_count = 0
for spec in phase_specs:
controls = (
phase == "context"
and spec["grid"] == 8
and spec["samples"] == 64
and spec["bounds_mode"] == "CURRENT_BOUNDS"
)
expected_methods = {"direct"} | (
{
"spatial_shuffle",
"density_only",
"color_only",
"zero_density_color",
"wrong_scene",
}
if controls
else set()
)
query_ids = records[spec["scene_id"]]["roles"]["primary_query"]
expected_row_count += len(query_ids) * len(expected_methods)
for fid in query_ids:
assert {r["method"] for r in groups[spec["key"], fid]} == expected_methods
assert len(rows) == expected_row_count
assert set(direct_rows) == {s["key"] for s in phase_specs}
for spec in phase_specs:
selected = direct_rows[spec["key"]]
assert sorted(r["query_id"] for r in selected) == sorted(
records[spec["scene_id"]]["roles"]["primary_query"]
)
assert len({r["used_state_hash"] for r in selected}) == 1
assert all(len({r["query_camera_hash"] for r in rr}) == 1 for rr in groups.values())
phases[phase] = {
"states": n,
"rows_recomputed": len(rows),
"metrics_max_absolute_error": errors,
"metric_exact_row_counts": exact,
"all_evaluator_queries_after_complete_seal": True,
"shared_state_across_queries": True,
"wrong_scene_preserves_camera": True,
}
check_hashes(config["source_sha256"])
assert sha(Path(checkpoint)) == CHECKPOINT_HASH
write_json(root / "audit/postrun_GT_access.json", audit_access)
report = {
"status": "PASS",
"optimized_state_count": len(specs),
"preregistration_config_and_state_plan_hashes_match": True,
"filesystem_chronology_auxiliary": chronology,
"completion_plan_state_source_hashes_match": True,
"all_selected_checkpoints_minimum_full_objective_earliest_tie": True,
"context_role_ids_only": True,
"RGB_ONLY_optimizer_never_reads_depth": True,
"query_oracle_explicitly_separate": True,
"actual_resolution_sweep_only_grid_changes": True,
"actual_samples_diagnostic_only_8_16_RGBD": True,
"phases": phases,
"optimization_checks": optimization_checks,
"baseline_checkpoint_unchanged": True,
"baseline_checkpoint_sha256": CHECKPOINT_HASH,
"final_holdout_touched": False,
"dynamic_ttt_run": False,
"new_carrier_trained": False,
"metric_recomputation_tolerance": {"atol": 1e-6, "rtol": 1e-5},
"statistics_reproduction": "SEPARATE_INDEPENDENT_ANALYSIS_AUDIT",
"audit_script_sha256": sha(Path(__file__)),
}
write_json(destination, report)
return {"status": "PASS", "optimized_state_count": len(specs), "phases": phases}
if __name__ == "__main__":
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--root", required=True)
parser.add_argument(
"--checkpoint",
default="outputs/EXP-3D-20260927-centered-training-1000a-600b-v1/training/phase_b_final.pt",
)
args = parser.parse_args()
print(json.dumps(run(args.root, args.checkpoint), indent=2))