tests/test_observability_supervision_evaluation.py
"""Tiny synthetic phase seals, controls, reference matching, and protected-data guards."""
import json
from pathlib import Path
import pytest
import torch
from test_direct_capacity_evaluation import setup as old_setup
from mcss.dynamic.types import hash_scene_state
from mcss.mechanism_pilot.direct_capacity_contracts import StateSealBarrier
from mcss.mechanism_pilot.observability_supervision_evaluation import (
baseline_comparison,
ensure_phase,
evaluate_supervision,
)
from mcss.mechanism_pilot.small_training import sha, write_json
def setup(root):
torch.set_num_threads(1)
previous = old_setup(root)
specs = []
for old in previous:
state = torch.load(root / old["state_path"], weights_only=False)
state.density_logits = (
state.density_logits.repeat_interleave(2, -1)
.repeat_interleave(2, -2)
.repeat_interleave(2, -3)
)
state.color = (
state.color.repeat_interleave(2, -1).repeat_interleave(2, -2).repeat_interleave(2, -3)
)
state.log_variance = torch.zeros_like(state.density_logits)
for variant in ("S0", "S1", "S2", "S3"):
key = f"{old['scene_id']}__{old['role']}__{variant}"
folder = root / "checkpoints" / key
folder.mkdir()
torch.save(state, folder / "state.pt")
spec = {
"key": key,
"scene_id": old["scene_id"],
"role": old["role"],
"variant": variant,
"grid": 16,
"samples": 64,
"bounds": state.bounds.reshape(2, 3).tolist(),
"state_path": str((folder / "state.pt").relative_to(root)),
"optimization_path": str(folder.relative_to(root)),
}
write_json(
folder / "summary.json",
{
"state_hash": hash_scene_state(state),
"variant": variant,
"supervision": "CONTEXT_ONLY_RGBD",
"selection": "FIXED_BUDGET",
"selected_step": 10000,
},
)
write_json(folder / "access.json", [])
specs.append(spec)
sources = {str(Path(__file__).resolve()): sha(Path(__file__))}
config = {
"source_sha256": sources,
"baseline_reproduction": {"mean_atol": 0.001, "per_row_atol": 0.01},
}
write_json(root / "config.json", config)
write_json(root / "state_plan.json", {"states": specs})
write_json(
root / "preregistration.json",
{
"locked_file_sha256": {
name: sha(root / name)
for name in ("config.json", "state_plan.json", "scene_manifest.json")
}
},
)
for spec in specs:
folder = root / spec["optimization_path"]
write_json(
folder / "completion.json",
{
"status": "PASS",
"key": spec["key"],
"plan_sha256": sha(root / "state_plan.json"),
"config_sha256": sha(root / "config.json"),
"source_sha256": sources,
"state_file_sha256": sha(folder / "state.pt"),
"state_hash": json.loads((folder / "summary.json").read_text())["state_hash"],
"summary_sha256": sha(folder / "summary.json"),
"access_sha256": sha(folder / "access.json"),
"query_selection": False,
},
)
for phase, count in [("baseline", 4), ("formal", 16)]:
write_json(
root / f"{phase}_optimization_complete.json",
{
"status": "PASS",
"phase": phase,
"states": count,
"input_sha256": {
str(root / name): sha(root / name)
for name in (
"config.json",
"state_plan.json",
"scene_manifest.json",
"preregistration.json",
)
},
},
)
write_json(root / "baseline_reproduction.json", {"status": "PASS"})
return specs, config
def test_full_phase_controls_seal_before_media_and_baseline_replay(tmp_path, monkeypatch):
specs, config = setup(tmp_path)
count = []
original = StateSealBarrier.seal
def observe(self, *args):
original(self, *args)
count.append(1)
def guarded_sha(path):
if str(path).endswith((".png", ".npy")):
assert len(count) == len(specs)
return sha(path)
monkeypatch.setattr(StateSealBarrier, "seal", observe)
monkeypatch.setattr(
"mcss.mechanism_pilot.observability_supervision_evaluation.sha", guarded_sha
)
rows = evaluate_supervision(tmp_path, "formal", "cpu")
assert len(rows) == 80
for spec in specs:
subset = [r for r in rows if r["key"] == spec["key"]]
for fid in (8, 9):
assert len({r["query_camera_hash"] for r in subset if r["query_id"] == fid}) == 1
direct = [r for r in subset if r["method"] == "direct"]
assert len({r["used_state_hash"] for r in direct}) == 1
for row in subset:
assert row["prediction_file_sha256"] == sha(tmp_path / row["prediction_path"])
if row["method"] == "zero":
assert row["opacity"] == 0
events = json.loads((tmp_path / "raw/formal_seal_events.json").read_text())
assert all(e["event"] == "seal" for e in events[:16])
assert all(e["event"] == "query_camera_GT" for e in events[16:])
contexts = json.loads((tmp_path / "raw/context_results.json").read_text())
query = [r for r in rows if r["variant"] == "S0" and r["method"] == "direct"]
context = [r for r in contexts if r["variant"] == "S0"]
write_json(tmp_path / "raw/historical_baseline_query.json", query)
write_json(tmp_path / "raw/historical_baseline_context.json", context)
assert baseline_comparison(tmp_path, query, context, config)["status"] == "PASS"
altered = [{**r, "depth_absrel": r["depth_absrel"] + 0.02} for r in query]
assert baseline_comparison(tmp_path, altered, context, config)["status"] == "FAIL"
with pytest.raises(PermissionError, match="Baseline reproduction"):
ensure_phase(tmp_path, "formal")
monkeypatch.setattr("mcss.mechanism_pilot.observability_supervision_evaluation.sha", sha)
baseline = evaluate_supervision(tmp_path, "baseline", "cpu")
assert len(baseline) == 8
assert json.loads((tmp_path / "baseline_reproduction.json").read_text())["status"] == "PASS"
def test_incomplete_and_holdout_rejected_before_media(tmp_path, monkeypatch):
specs, _ = setup(tmp_path)
def forbid(*args, **kwargs):
raise AssertionError("Media accessed before guards")
monkeypatch.setattr("PIL.Image.open", forbid)
(tmp_path / specs[-1]["state_path"]).unlink()
with pytest.raises(PermissionError, match="All states"):
evaluate_supervision(tmp_path, "formal", "cpu")
manifest = json.loads((tmp_path / "scene_manifest.json").read_text())
manifest["FINAL_HOLDOUT_PROHIBITED"].append(manifest["scenes"][0]["scene_id"])
write_json(tmp_path / "scene_manifest.json", manifest)
with pytest.raises(PermissionError, match="prohibited"):
evaluate_supervision(tmp_path, "baseline", "cpu")