src/mcss/mechanism_pilot/contracts.py
"""Fail-closed pilot boundaries; these interfaces do not establish carrier qualification.
The evaluator ledger controls its loader API, not arbitrary filesystem access. Callers must
keep query providers exclusively in the evaluator. No query provider belongs in the runtime.
"""
from collections.abc import Mapping
from copy import deepcopy
import torch
from torch import nn
from mcss.dynamic.types import SealedScene, clone_scene_state, hash_scene_state, hash_value
from mcss.evaluation.sealed_queries import _validate_sealed
from mcss.geometry import generate_rays
def validate_scene_splits(records):
"""Validate audited physical identities, including repeated scans of the same site."""
assignments, scans = {}, set()
counts = {name: set() for name in ("train", "dev", "test")}
for row in records:
scene, physical, split = (row.get(k) for k in ("scene_id", "physical_scene_id", "split"))
if not scene or not physical or row.get("physical_scene_identity_status") != "VERIFIED":
raise ValueError("Verified physical scene identity is required")
if split not in counts or scene in scans:
raise ValueError("Invalid split or duplicate scan identity")
if physical in assignments and assignments[physical] != split:
raise ValueError("Physical scene leakage across splits")
scans.add(scene)
assignments[physical] = split
counts[split].add(physical)
return {split: len(ids) for split, ids in counts.items()}
class EvaluationLedger:
"""Require the complete predetermined candidate set before any query loader runs."""
def __init__(self, expected_candidates):
ids = tuple(expected_candidates)
if (
not ids
or len(set(ids)) != len(ids)
or any(not isinstance(x, str) or not x for x in ids)
):
raise ValueError("Expected candidate IDs must be nonempty and unique")
self.expected_candidates = frozenset(ids)
self._sealed = {}
self._fingerprints = {}
self.events = []
def seal(self, candidate_id, sealed):
if candidate_id not in self.expected_candidates or candidate_id in self._sealed:
raise ValueError("Unexpected or already sealed candidate")
if not isinstance(sealed, SealedScene):
raise TypeError("SealedScene required")
_validate_sealed(sealed)
if hash_scene_state(sealed.scene_state) != sealed.state_hash:
raise ValueError("Candidate state hash mismatch")
self._sealed[candidate_id] = sealed
self._fingerprints[candidate_id] = hash_value(sealed)
self.events.append(
{"event": "seal", "candidate_id": candidate_id, "state_hash": sealed.state_hash}
)
def assert_ready(self):
if set(self._sealed) != self.expected_candidates:
raise PermissionError(
"All expected candidate states must be sealed before query access"
)
for candidate, sealed in self._sealed.items():
if hash_value(sealed) != self._fingerprints[candidate]:
raise PermissionError("Sealed candidate or provenance mutated")
def _read(self, query_id, loader, kind):
self.assert_ready()
if not isinstance(query_id, str) or not query_id:
raise ValueError("Query identity required")
self.events.append({"event": kind, "query_id": query_id, "status": "requested"})
value = loader()
self.assert_ready()
self.events[-1]["status"] = "completed"
return value
def read_query_camera(self, query_id, loader):
return self._read(query_id, loader, "query_camera")
def read_query_ground_truth(self, query_id, loader):
return self._read(query_id, loader, "query_ground_truth")
POLICY_KEYS = frozenset(
{
"arrived_rgb",
"arrived_camera",
"rgb_residual",
"opacity",
"coverage",
"state_summary",
"fast_summary",
"history_summary",
}
)
_FORBIDDEN = ("query", "depth", "future", "reward", "oracle", "ground_truth", "target", "label")
def validate_policy_inputs(inputs):
"""Reject unknown top-level channels and forbidden nested fields; never silently drop them.
This is a structural boundary. The caller must bind arrived inputs to declared frame IDs;
semantic relabeling of forbidden tensors cannot be detected from tensor values.
"""
if not isinstance(inputs, Mapping) or set(inputs) - POLICY_KEYS:
raise ValueError("Policy accepts only declared arrived-observation and summary channels")
def inspect(value):
if isinstance(value, Mapping):
for key, item in value.items():
if not isinstance(key, str) or any(word in key.lower() for word in _FORBIDDEN):
raise ValueError("Forbidden policy information")
inspect(item)
elif isinstance(value, (list, tuple)):
for item in value:
inspect(item)
elif not isinstance(value, (torch.Tensor, int, float, str, bool, type(None))):
# Cameras are explicitly admitted only for arrived_camera below.
raise TypeError("Opaque policy payload is not permitted")
from mcss.types import Cameras
for key, value in inputs.items():
if key == "arrived_camera" and type(value) is Cameras:
continue
inspect(value)
return inputs
class ResidualReadoutControl(nn.Module):
"""Query-conditioned RGB/depth residual; detached private state, ordinary optimizer API.
Train with model.train(), optimizer.zero_grad(), supervised_loss(model(state,cameras),
train_labels).backward(), optimizer.step(). Only train-split labels are admissible.
eval() does NOT imply trained or qualified. Training/evaluation provenance is external.
Existing FreeDecoderControl replaces outputs rather than adding residuals, so is not reused.
"""
def __init__(self, feature_dim, renderer):
super().__init__()
if list(renderer.parameters()):
raise ValueError("Residual baseline requires parameter-free fixed renderer")
self.renderer = renderer
self.head = nn.Sequential(nn.Conv2d(feature_dim + 6, 16, 1), nn.SiLU(), nn.Conv2d(16, 4, 1))
nn.init.zeros_(self.head[-1].weight)
nn.init.zeros_(self.head[-1].bias)
self.training_status = "UNTRAINED_UNEVALUATED"
def forward(self, state, cameras):
state_before, camera_before = hash_scene_state(state), hash_value(cameras)
private_state, private_camera = clone_scene_state(state), deepcopy(cameras)
with torch.no_grad():
base = self.renderer(private_state, private_camera, {"rgb", "depth", "visibility"})
features = self.renderer.render_features(private_state, private_camera)
origins, directions = generate_rays(private_camera)
rays = torch.cat((origins, directions), dim=-1).permute(0, 1, 4, 2, 3)
combined = torch.cat((features, rays), dim=2)
batch, views, channels, height, width = combined.shape
residual = self.head(combined.reshape(batch * views, channels, height, width))
residual = residual.reshape(batch, views, 4, height, width)
result = dict(base)
result["rgb"] = (base["rgb"] + residual[:, :, :3]).clamp(0, 1)
result["depth"] = (base["depth"] + residual[:, :, 3:4]).clamp_min(0)
if hash_scene_state(state) != state_before or hash_value(cameras) != camera_before:
raise RuntimeError("Readout modified shared inputs")
return result
def cross_scene_readout(
renderer, replacement_state, query_cameras, source_scene_id, target_scene_id
):
"""Return renderer output dict from replacement state and numerically unchanged cameras.
No registration/recentering is fitted. Input states must already use the preregistered
coordinate convention. Independent identity verification belongs in the split audit.
"""
if not source_scene_id or not target_scene_id or source_scene_id == target_scene_id:
raise ValueError("Replacement requires different scenes")
state_before, camera_before = hash_scene_state(replacement_state), hash_value(query_cameras)
state, cameras = clone_scene_state(replacement_state), deepcopy(query_cameras)
output = renderer(state, cameras, {"rgb", "depth", "visibility"})
if (
hash_value(cameras) != camera_before
or hash_value(query_cameras) != camera_before
or hash_scene_state(replacement_state) != state_before
):
raise RuntimeError("Cross-scene readout changed query cameras or shared state")
return output