src/mcss/mechanism_pilot/observability_supervision_evaluation.py
"""Post-seal context-geometry supervision evaluation, no query-adaptive choices."""
from __future__ import annotations
import json
import time
from pathlib import Path
import numpy as np
import torch
from mcss.dynamic.types import clone_scene_state, hash_scene_state, hash_value
from mcss.geometry import transform_cameras
from mcss.measurements import FixedMeasurementRenderer
from mcss.mechanism_pilot.direct_capacity_contracts import (
CapacityEvaluator,
ContextOnlyLoader,
StateSealBarrier,
)
from mcss.mechanism_pilot.direct_capacity_evaluation import (
_coverage,
_metrics,
_resolve,
_summary_path,
)
from mcss.mechanism_pilot.small_training import sha, write_json
def read(path):
return json.loads(Path(path).read_text())
def ensure_phase(root, phase):
root = Path(root)
if phase not in ("baseline", "formal"):
raise ValueError("Unknown supervision phase")
manifest, plan, config = [
read(root / f"{name}.json") for name in ("scene_manifest", "state_plan", "config")
]
allowed = {r["scene_id"] for r in manifest["scenes"]}
holdout = set(manifest["FINAL_HOLDOUT_PROHIBITED"])
if len(allowed) != len(manifest["scenes"]) or allowed & holdout:
raise PermissionError("Duplicate or prohibited scene")
if any(r.get("cohort_role") != "CAPACITY_DEV_EXPOSED" for r in manifest["scenes"]):
raise PermissionError("Only exposed attribution scenes allowed")
all_specs = plan["states"]
expected = {
(sid, role, variant)
for sid in allowed
for role in ("A", "B")
for variant in ("S0", "S1", "S2", "S3")
}
actual = {(s["scene_id"], s["role"], s["variant"]) for s in all_specs}
if actual != expected or len(all_specs) != len(expected):
raise PermissionError("Complete matched four-variant plan required")
if len({s["key"] for s in all_specs}) != len(all_specs):
raise PermissionError("Duplicate state key")
specs = [s for s in all_specs if phase == "formal" or s["variant"] == "S0"]
marker_path = root / f"{phase}_optimization_complete.json"
if not marker_path.is_file():
raise PermissionError("Complete phase marker required before any media")
marker = read(marker_path)
if (
marker.get("status") != "PASS"
or marker.get("phase") != phase
or marker.get("states") != len(specs)
):
raise PermissionError("Phase marker does not match frozen plan")
if phase == "formal" and read(root / "baseline_reproduction.json").get("status") != "PASS":
raise PermissionError("Baseline reproduction must PASS before formal evaluation")
for path, digest in marker["input_sha256"].items():
if sha(Path(path)) != digest:
raise PermissionError("Frozen optimization input changed")
for name, digest in read(root / "preregistration.json")["locked_file_sha256"].items():
if sha(root / name) != digest:
raise PermissionError(f"Preregistered input changed: {name}")
for path, digest in config["source_sha256"].items():
if sha(Path(path)) != digest:
raise PermissionError(f"Frozen source changed: {path}")
paths = [
root / f"{n}.json" for n in ("scene_manifest", "state_plan", "config", "preregistration")
]
paths.append(marker_path)
for spec in specs:
if spec["grid"] != 16 or spec["samples"] != 64:
raise PermissionError("Only frozen 16-grid/64-sample states allowed")
folder = _resolve(root, spec["optimization_path"])
files = [folder / n for n in ("completion.json", "summary.json", "access.json")]
state_path = _resolve(root, spec["state_path"])
if not all(p.is_file() for p in files + [state_path]):
raise PermissionError("All states/completions required before media reads")
completion, summary = read(files[0]), read(files[1])
checks = {
"status": "PASS",
"key": spec["key"],
"plan_sha256": sha(root / "state_plan.json"),
"config_sha256": sha(root / "config.json"),
"source_sha256": config["source_sha256"],
"state_file_sha256": sha(state_path),
"summary_sha256": sha(files[1]),
"access_sha256": sha(files[2]),
"state_hash": summary["state_hash"],
"query_selection": False,
}
if any(completion.get(k) != v for k, v in checks.items()):
raise PermissionError("State completion differs from locked protocol")
if (
summary.get("selection") != "FIXED_BUDGET"
or summary.get("selected_step") != 10000
or summary.get("variant") != spec["variant"]
or summary.get("supervision") != "CONTEXT_ONLY_RGBD"
):
raise PermissionError("Only fixed-budget context RGBD states permitted")
paths += files + [state_path]
return manifest, plan, config, specs, paths
def diagnostic_state(state, method, seed=20260927):
"""Deterministic frozen controls; no renderer/state semantics changes."""
out = clone_scene_state(state)
if method == "zero":
out.density_logits.fill_(-1e6)
out.color.zero_()
elif method == "spatial_shuffle":
generator = torch.Generator(device="cpu").manual_seed(seed)
permutation = torch.randperm(16**3, generator=generator).to(out.color.device)
for name in ("density_logits", "color", "log_variance"):
value = getattr(out, name)
setattr(out, name, value.flatten(2)[:, :, permutation].reshape_as(value))
else:
raise ValueError("Unknown diagnostic state")
return out
def baseline_comparison(root, rows, context_rows, config):
comparisons = {}
passed = True
for name, current, id_key in (
("query", rows, "query_id"),
("context", context_rows, "frame_id"),
):
historical = read(root / "raw" / f"historical_baseline_{name}.json")
def keyed(values, id_key=id_key):
out = {(r["scene_id"], r["role"], r[id_key]): r for r in values}
if len(out) != len(values):
raise PermissionError("Duplicate baseline reference identity")
return out
a, b = keyed(current), keyed(historical)
if set(a) != set(b):
raise PermissionError("Baseline/reference frame identity mismatch")
differences = [abs(a[k]["depth_absrel"] - b[k]["depth_absrel"]) for k in a]
# Equal scene, then role, then view weighting; every matched scene has both roles.
def macro(values):
scenes = sorted({k[0] for k in values})
return float(
np.mean(
[
np.mean(
[
np.mean(
[
v["depth_absrel"]
for k, v in values.items()
if k[0] == sid and k[1] == role
]
)
for role in ("A", "B")
]
)
for sid in scenes
]
)
)
new_mean, old_mean = macro(a), macro(b)
thresholds = config["baseline_reproduction"]
ok = (
abs(new_mean - old_mean) <= thresholds["mean_atol"]
and max(differences) <= thresholds["per_row_atol"]
)
passed &= ok
metrics = (
"rgb_mse",
"rgb_ssim",
"depth_absrel",
"depth_rmse",
"depth_delta1",
"opacity",
"coverage",
)
comparisons[name] = {
"status": "PASS" if ok else "FAIL",
"rows": len(a),
"current_scene_mean_absrel": new_mean,
"historical_scene_mean_absrel": old_mean,
"mean_absolute_difference": abs(new_mean - old_mean),
"max_row_absrel_difference": max(differences),
"max_metric_absolute_differences": {
m: max(abs(a[k][m] - b[k][m]) for k in a) for m in metrics
},
"historical_raw_sha256": sha(root / "raw" / f"historical_baseline_{name}.json"),
}
report = {
"status": "PASS" if passed else "FAIL",
"comparisons": comparisons,
"thresholds": config["baseline_reproduction"],
"query_used_for_tuning": False,
"failure_blocks_new_supervision": True,
}
write_json(root / "baseline_reproduction.json", report)
return report
@torch.no_grad()
def evaluate_supervision(root, phase="baseline", device="cuda"):
root = Path(root)
manifest, plan, config, specs, input_paths = ensure_phase(root, phase)
manifest_path = root / "scene_manifest.json"
allowed = tuple(r["scene_id"] for r in manifest["scenes"])
holdout = tuple(manifest["FINAL_HOLDOUT_PROHIBITED"])
source_hashes = config["source_sha256"]
if phase == "baseline":
input_paths += [
root / "raw" / f"historical_baseline_{kind}.json" for kind in ("query", "context")
]
raw = root / "raw"
result_path = raw / ("baseline_results.json" if phase == "baseline" else "query_results.json")
prediction_dir = raw / f"{phase}_predictions"
if result_path.exists() or prediction_dir.exists():
raise FileExistsError("Preserve prior phase evaluation; refusing overwrite")
input_hashes = {str(p.resolve()): sha(p) for p in input_paths}
barrier = StateSealBarrier(s["key"] for s in specs)
records = {r["scene_id"]: r for r in manifest["scenes"]}
for spec in specs:
state = torch.load(
_resolve(root, spec["state_path"]), map_location="cpu", weights_only=False
)
summary = json.loads(_summary_path(root, spec).read_text())
if hash_scene_state(state) != summary["state_hash"]:
raise PermissionError("Optimized state differs from checkpoint selection summary")
if (
state.density_logits.shape[-3:] != (16,) * 3
or state.features is not None
or not torch.equal(
state.bounds.cpu(),
torch.tensor(spec["bounds"], dtype=state.bounds.dtype).reshape(1, 2, 3),
)
):
raise ValueError("Unexpected direct state grid/features")
barrier.seal(spec["key"], spec["scene_id"], state)
barrier.assert_ready()
access = []
for record in manifest["scenes"]:
for frame in record["frames"]:
for field in ("rgb", "depth"):
path = _resolve(manifest_path.parent, frame[field])
access.append(
{
"scene_id": record["scene_id"],
"frame_id": frame["frame_id"],
"purpose": "HASH_INPUT_AFTER_ALL_STATES_SEALED",
"channels": [field],
"path": str(path),
}
)
input_hashes[str(path.resolve())] = sha(path)
evaluator_source_hashes = {str(Path(__file__).resolve()): sha(Path(__file__))}
raw.mkdir(parents=True, exist_ok=True)
prediction_dir.mkdir()
write_json(
raw / f"{phase}_evaluation_lock.json",
{
"phase": phase,
"input_sha256": input_hashes,
"optimizer_source_sha256": source_hashes,
"evaluator_source_sha256": evaluator_source_hashes,
"planned_state_keys": [s["key"] for s in specs],
"all_states_sealed_before_evaluation": True,
"final_holdout_touched": False,
"new_carrier_trained": False,
"dynamic_ttt_run": False,
},
)
rows, context_rows = [], []
evaluators = {
sid: CapacityEvaluator(
record,
manifest["image_size"],
manifest_path.parent,
device,
barrier=barrier,
capacity_scene_ids=allowed,
holdout_scene_ids=holdout,
access_log=access,
)
for sid, record in records.items()
}
configs = {}
for spec in specs:
key = (spec["role"], spec["variant"])
configs.setdefault(key, {})[spec["scene_id"]] = spec
started = time.perf_counter()
for spec_index, spec in enumerate(specs):
sid, record = spec["scene_id"], records[spec["scene_id"]]
frozen = barrier.state(spec["key"])
state = frozen.to(device=device)
state_hash = hash_scene_state(state)
renderer = FixedMeasurementRenderer(n_samples=spec["samples"], ray_chunk_size=2048).to(
device
)
anchor_frame = next(
f for f in record["frames"] if f["frame_id"] == record["roles"]["context_a"][0]
)
anchor = torch.tensor(anchor_frame["c2w"], dtype=torch.float32, device=device)
inverse_anchor = torch.linalg.inv(anchor)
if True:
# Post-seal evaluator may measure depth even for RGB-only fitted states.
barrier.assert_ready()
first_event = len(access)
batch = ContextOnlyLoader(
record,
manifest["image_size"],
manifest_path.parent,
device,
track="RGBD",
capacity_scene_ids=allowed,
holdout_scene_ids=holdout,
access_log=access,
).context(spec["role"])
for event in access[first_event:]:
event["purpose"] = "POST_SEAL_CONTEXT_EVALUATION_NOT_OPTIMIZER"
event["trained_variant"] = spec["variant"]
camera = transform_cameras(batch.cameras, inverse_anchor)
for j, fid in enumerate(batch.frame_ids):
one_camera = camera.select_views(torch.tensor([j], device=device))
if str(device).startswith("cuda"):
torch.cuda.synchronize()
tick = time.perf_counter()
pred = renderer(state, one_camera, ("rgb", "depth", "visibility"))
if str(device).startswith("cuda"):
torch.cuda.synchronize()
seconds = time.perf_counter() - tick
context_rows.append(
{
**spec,
"frame_id": fid,
"method": "direct",
"seconds": seconds,
"camera_hash": hash_value(one_camera),
"used_state_hash": state_hash,
**_metrics(pred, batch.rgb[:, j : j + 1], batch.depth[:, j : j + 1]),
}
)
barrier.assert_ready()
controls = phase == "formal"
for fid in record["roles"]["primary_query"]:
batch = evaluators[sid].query(fid)
camera = transform_cameras(batch.cameras, inverse_anchor)
camera_hash = hash_value(camera)
reference = None
methods = ["direct", "wrong_scene"] if controls else ["direct"]
if phase == "formal" and spec["variant"] == "S3":
methods += ["spatial_shuffle", "zero"]
for method in methods:
donor_id = None
if method == "direct":
used_state = state
elif method == "wrong_scene":
config_key = (spec["role"], spec["variant"])
group = configs[config_key]
ids = sorted(group)
if len(ids) < 2 or set(ids) != set(allowed):
raise PermissionError(
"Wrong-scene control requires matched states for every scene"
)
donor_id = ids[(ids.index(sid) + 1) % len(ids)]
used_state = barrier.state(group[donor_id]["key"]).to(device=device)
elif method in ("spatial_shuffle", "zero"):
used_state = diagnostic_state(state, method)
else:
raise ValueError("Unknown predeclared control")
before = hash_scene_state(used_state)
if str(device).startswith("cuda"):
torch.cuda.synchronize()
tick = time.perf_counter()
pred = renderer(used_state, camera, ("rgb", "depth", "visibility"))
if str(device).startswith("cuda"):
torch.cuda.synchronize()
seconds = time.perf_counter() - tick
if method == "direct":
reference = {k: pred[k].clone() for k in ("rgb", "depth")}
row = {
**spec,
"query_id": fid,
"method": method,
"donor_scene_id": donor_id,
"query_camera_hash": camera_hash,
"used_state_hash": before,
"seconds": seconds,
"prediction_hash": hash_value(pred),
**_metrics(pred, batch.rgb, batch.depth),
**_coverage(used_state, camera, batch.depth),
"prediction_change_from_direct": {
k: {
"rms": float((pred[k] - reference[k]).square().mean().sqrt()),
"max_abs": float((pred[k] - reference[k]).abs().max()),
}
for k in ("rgb", "depth")
},
}
prediction_path = prediction_dir / f"{spec_index:04d}_{fid}_{method}.npz"
np.savez_compressed(
prediction_path,
rgb=pred["rgb"][0, 0].cpu().numpy(),
depth=pred["depth"][0, 0, 0].cpu().numpy(),
opacity=pred["visibility"][0, 0, 0].cpu().numpy(),
)
row["prediction_path"] = str(prediction_path.relative_to(root))
row["prediction_file_sha256"] = sha(prediction_path)
rows.append(row)
assert hash_scene_state(used_state) == before and hash_value(camera) == camera_hash
assert hash_scene_state(state) == state_hash
barrier.assert_ready()
barrier.assert_ready()
assert all(sha(Path(path)) == digest for path, digest in input_hashes.items())
assert all(sha(Path(p)) == h for p, h in source_hashes.items())
assert all(sha(Path(p)) == h for p, h in evaluator_source_hashes.items())
write_json(result_path, rows)
write_json(
raw / ("baseline_context_results.json" if phase == "baseline" else "context_results.json"),
context_rows,
)
write_json(raw / f"{phase}_GT_access.json", access)
write_json(raw / f"{phase}_seal_events.json", barrier.events)
write_json(
raw / f"{phase}_evaluation_integrity.json",
{
"status": "PASS",
"phase": phase,
"n_states": len(specs),
"n_query_rows": len(rows),
"n_context_rows": len(context_rows),
"all_states_immutable": True,
"all_frozen_inputs_unchanged": True,
"query_access_after_all_phase_states_sealed": True,
"context_RGBD_only_optimization": True,
"final_holdout_touched": False,
"dynamic_ttt_run": False,
"new_carrier_trained": False,
"seconds": time.perf_counter() - started,
},
)
if phase == "baseline":
baseline_comparison(root, rows, context_rows, config)
return rows