feat(perception): enforce six-layer freshness through receipt

config/perception/k1-perception-ddrnet39-rfdetr-tgs-prototype-v1.json

@@ -36,6 +36,37 @@
     "MKL_NUM_THREADS": 1,
     "reason": "bounded-single-thread-numeric-libraries-before-whole-graph-qualification"
   },
+  "operating_envelope": {
+    "status": "bounded-latency-reference-not-full-qualification",
+    "checker": "k1link.perception.worker_operating_envelope.operating_envelope_failures",
+    "reference_conditions": {
+      "envelope_id": "worker006-4090-610.47-stock-clock-reference/v1",
+      "gpu_name": "NVIDIA GeForce RTX 4090",
+      "driver_version": "610.47",
+      "cpu_limit_millicores": 8000,
+      "memory_limit_mib": 8192,
+      "minimum_sm_clock_mhz": 2610,
+      "minimum_memory_clock_mhz": 10251,
+      "maximum_snapshot_age_ms": 1000
+    },
+    "check_phase": "post-warmup-before-timed-window-plus-runtime-monitoring",
+    "container_may_set_host_clocks": false,
+    "persistent_host_clock_policy_installed": false,
+    "readiness_is_realtime_qualification": false,
+    "unknown_ownership_or_telemetry_is_pass": false,
+    "failed_envelope_allows_labelled_experiment": true
+  },
+  "output_freshness_contract": {
+    "schema_version": "missioncore.perception-scene-freshness/v1",
+    "required_layers": ["segmentation", "objects", "geometry", "motion", "costmap", "policy"],
+    "retain_original_layer_identity": true,
+    "propagate_required_input_age": true,
+    "recheck_at_publication_receipt_and_use": true,
+    "clock_uncertainty_adds_to_age": true,
+    "unavailable_or_stale_required_layer": "no-go-advisory-only",
+    "history_can_independently_authorize_terrain": false,
+    "transport_streamstart_and_lease_fencing_required_separately": true
+  },
   "input_contract": {
     "schema_version": "missioncore.perception-stream-start/v1",
     "channels": ["camera", "point-cloud", "pose"],

experiments/perception/worker/streaming_profile_stage1/pilot_freshness.py

@@ -0,0 +1,139 @@
+"""Pilot adapter for the shared six-layer ABI; source arrival clock, local receipt.
+
+Not a live clock synchronizer or lease authority. Domain history/unknown motion
+stays in existing payloads. An unavailable layer may retain diagnostic fields in
+the scene, but makes no valid payload/terrain-permission claim in the ABI.
+"""
+
+import hashlib
+import json
+
+from k1link.perception.realtime_contract import LayerEvidence
+from k1link.perception.realtime_scene import DEPENDENCIES, SceneFreshness, derive_layer
+
+CLOCK_DOMAIN = "pilot-original-host-arrival"
+
+
+def payload_digest(scene, layer):
+    if layer == "segmentation":
+        return scene["segmentation_sha256"]
+    fields = {
+        "objects": ("proposals",),
+        "geometry": ("observations", "surface_state", "range_estimator"),
+        "motion": ("tracks", "threats"),
+        "costmap": ("costmap_states", "costmap_material"),
+        "policy": ("policy_actions", "policy_counts"),
+    }[layer]
+    value = {key: scene[key] for key in fields}
+    return hashlib.sha256(
+        json.dumps(value, sort_keys=True, allow_nan=False, separators=(",", ":")).encode()
+    ).hexdigest()
+
+
+def build_freshness(scene, bundle, ddr_layer, epoch_id):
+    stamp, sequence = bundle["time_ns"], bundle["sequence"]
+    segmentation = LayerEvidence(
+        "segmentation",
+        epoch_id,
+        ddr_layer["source_sequence"],
+        ddr_layer["source_host_monotonic_ns"],
+        "current" if ddr_layer["state"] == "current" else "held",
+        payload_digest(scene, "segmentation"),
+        ddr_layer["source_host_monotonic_ns"],
+    )
+    objects = LayerEvidence(
+        "objects", epoch_id, sequence, stamp, "current", payload_digest(scene, "objects"), stamp
+    )
+    layers = {"segmentation": segmentation, "objects": objects}
+    for name in ("geometry", "motion", "costmap", "policy"):
+        oldest = stamp
+        if name == "geometry":
+            if not bundle["available"] or scene["surface_state"] != "valid":
+                layers[name] = LayerEvidence(name, epoch_id, None, None, "unavailable", None)
+                continue
+            lineage = bundle["lineage"]
+            oldest = min(
+                lineage["pose_host_monotonic_ns"],
+                *(item["host_monotonic_ns"] for item in lineage["point_increments"]),
+            )
+        if name == "costmap" and bundle["available"]:
+            # A rolling window can retain old occupied/unknown cells to prohibit
+            # terrain. Permissive cells must keep their own observation age.
+            oldest = scene["tgs_counts"]["oldest_permissive_cell_source_ns"]
+        layers[name] = derive_layer(
+            name,
+            epoch_id=epoch_id,
+            source_sequence=sequence,
+            source_time_ns=stamp,
+            payload_sha256=payload_digest(scene, name),
+            inputs=tuple(layers[key] for key in DEPENDENCIES[name]),
+            oldest_required_input_time_ns=oldest,
+        )
+    return SceneFreshness(epoch_id, CLOCK_DOMAIN, sequence, stamp, tuple(layers.values()))
+
+
+def assess(freshness, *, bundle, now_ns):
+    # Both sides of this pilot use the SAME monotonic clock. Map it back to
+    # original source arrival time; zero mapping uncertainty is local-only and
+    # is not a claim of hardware camera/LiDAR synchronization or network quality.
+    return freshness.assess(
+        epoch_id=freshness.epoch_id,
+        clock_domain_id=CLOCK_DOMAIN,
+        observed_source_time_ns=bundle["time_ns"] + now_ns - bundle["due_ns"],
+        clock_uncertainty_ms=0,
+        maximum_clock_uncertainty_ms=5,
+        maximum_layer_age_ms=250,
+    )
+
+
+def suppress_policy(scene):
+    scene["policy_actions"] = [2] * len(scene["policy_actions"])
+    scene["policy_counts"] = {
+        "ALLOW_candidate": 0,
+        "HIGH_COST": 0,
+        "NO_GO": len(scene["policy_actions"]),
+    }
+
+
+def prepare_publication(scene, bundle, ddr_layer, *, epoch_id, now_ns):
+    freshness = build_freshness(scene, bundle, ddr_layer, epoch_id)
+    checked = assess(freshness, bundle=bundle, now_ns=now_ns)
+    if not checked.fresh_complete:
+        suppress_policy(scene)
+        # The envelope hashes the actual guarded policy, not the discarded one.
+        freshness = build_freshness(scene, bundle, ddr_layer, epoch_id)
+        checked = assess(freshness, bundle=bundle, now_ns=now_ns)
+    scene["freshness"] = freshness.to_dict()
+    scene["freshness_at_publication"] = checked.to_dict()
+    scene["stale_at_publication"] = any(item.state == "stale" for item in checked.layers)
+    scene["oldest_required_input_age_ms"] = max(
+        (age for age in checked.age_upper_bound_ms if age is not None), default=None
+    )
+
+
+def validate_receipt(scene, bundle, *, epoch_id):
+    freshness = SceneFreshness.from_dict(scene["freshness"])
+    if (
+        freshness.epoch_id,
+        freshness.clock_domain_id,
+        freshness.source_sequence,
+        freshness.source_time_ns,
+    ) != (epoch_id, CLOCK_DOMAIN, bundle["sequence"], bundle["time_ns"]):
+        raise ValueError("collector freshness identity mismatch")
+    if scene["commands_enabled"] is not False or scene["actuation_allowed"] is not False:
+        raise ValueError("pilot result attempted control authority")
+    for item in freshness.layers:
+        if (
+            item.payload_sha256 is not None
+            and payload_digest(scene, item.layer) != item.payload_sha256
+        ):
+            raise ValueError("collector layer payload digest mismatch")
+    return freshness
+
+
+def receipt_view(scene, checked):
+    """A consumer view; never mutate the hashed published evidence."""
+    view = dict(scene)
+    if not checked.fresh_complete:
+        suppress_policy(view)
+    return view

experiments/perception/worker/streaming_profile_stage1/pilot_graph.py

@@ -444,7 +444,7 @@ class JointGraph:
         for index in diagnostic_cells:
             matching = np.flatnonzero(good & (cell_ids == index))
             diagnostics[str(index)] = {
-                "votes": dict(zip(self.material_names, votes[index].tolist())),
+                "votes": dict(zip(self.material_names, votes[index].tolist(), strict=True)),
                 "projected_ground_points": len(matching),
                 "points_truncated": len(matching) > 128,
                 "points": [
@@ -467,6 +467,12 @@ class JointGraph:
                 if index is not None:
                     cells[index] = 2
         actions = self.action_lut[material, cells]
+        permissive_stamps = last_seen[actions != 2]
+        if np.any(permissive_stamps < 0):
+            raise ValueError("permissive TGS cell lacks source observation")
+        oldest_permissive = (
+            int(permissive_stamps.min()) if len(permissive_stamps) else bundle["time_ns"]
+        )
         return (
             cells,
             actions,
@@ -478,6 +484,7 @@ class JointGraph:
                 "occupied": int(np.count_nonzero(states == 2)),
                 "rejected": int(np.count_nonzero(states == 3)),
                 "stale_cells": int(np.count_nonzero(stale)),
+                "oldest_permissive_cell_source_ns": oldest_permissive,
                 "state_counts": dict(Counter(int(x) for x in cells)),
                 **({"diagnostic_cells": diagnostics} if diagnostics else {}),
             },

experiments/perception/worker/streaming_profile_stage1/run_joint_pilot.py

@@ -24,6 +24,7 @@ from datetime import UTC, datetime
 from pathlib import Path
 
 import numpy as np
+from pilot_freshness import assess, prepare_publication, receipt_view, validate_receipt
 from pilot_ipc import receive, send
 from pilot_queue import Mailbox
 from pilot_scheduler import GpuStage
@@ -482,44 +483,11 @@ def run(args):
                     proposals=proposals,
                     detector_ms=(gpu_done - ddr_done) / 1e6,
                 )
-                # Late evidence remains inspectable but cannot authorize terrain.
+                freshness_started = time.monotonic_ns()
-                policy_age_ms = (time.monotonic_ns() - bundle["due_ns"]) / 1e6
+                prepare_publication(
-                sensor_source_age_ms = max(
+                    scene, bundle, ddr_layer, epoch_id=args.run_id,
-                    bundle["lineage"]["pose_age_ms"] or 0,
+                    now_ns=freshness_started,
-                    bundle["lineage"]["oldest_point_age_ms"] or 0,
                 )
-                scene["layer_freshness"] = {
-                    "segmentation": {
-                        **ddr_layer,
-                        "age_at_policy_ms": policy_age_ms + ddr_layer["source_age_ms"],
-                    },
-                    "detection": {
-                        "state": "current",
-                        "source_sequence": bundle["sequence"],
-                        "source_host_monotonic_ns": bundle["time_ns"],
-                        "source_age_ms": 0.0,
-                        "age_at_policy_ms": policy_age_ms,
-                    },
-                    "geometry": {
-                        "state": "current" if bundle["available"] else "unavailable",
-                        "source_sequence": bundle["sequence"],
-                        "source_host_monotonic_ns": bundle["time_ns"],
-                        "source_age_ms": sensor_source_age_ms,
-                        "age_at_policy_ms": policy_age_ms + sensor_source_age_ms,
-                    },
-                }
-                scene["oldest_required_input_age_ms"] = policy_age_ms + max(
-                    sensor_source_age_ms,
-                    ddr_layer["source_age_ms"],
-                )
-                scene["stale_at_publication"] = scene["oldest_required_input_age_ms"] > 250
-                if scene["stale_at_publication"]:
-                    scene["policy_actions"] = [2] * len(scene["policy_actions"])
-                    scene["policy_counts"] = {
-                        "ALLOW_candidate": 0,
-                        "HIGH_COST": 0,
-                        "NO_GO": len(scene["policy_actions"]),
-                    }
                 scene.update(
                     sequence=bundle["sequence"],
                     lineage=bundle["lineage"],
@@ -534,6 +502,10 @@ def run(args):
                 received = json.loads(encoded)
                 if received["sequence"] != bundle["sequence"]:
                     raise ValueError("collector identity mismatch")
+                freshness = validate_receipt(received, bundle, epoch_id=args.run_id)
+                checked_at = time.monotonic_ns()
+                checked = assess(freshness, bundle=bundle, now_ns=checked_at)
+                view = receipt_view(received, checked)
                 finished = time.monotonic_ns()
                 timing = {
                     **scene["timing_ms"],
@@ -549,6 +521,7 @@ def run(args):
                     "cpu_tail_to_receiver_ms": (finished - cpu_started) / 1e6,
                     "compute_to_receiver_ms": (finished - begin) / 1e6,
                     "source_due_to_receiver_ms": (finished - bundle["due_ns"]) / 1e6,
+                    "freshness_encode_receive_ms": (finished - freshness_started) / 1e6,
                 }
                 result = {
                     "sequence": bundle["sequence"],
@@ -565,7 +538,9 @@ def run(args):
                         x["metric_geometry"] is not None for x in scene["observations"]
                     ),
                     "tgs_counts": scene["tgs_counts"],
-                    "policy_counts": scene["policy_counts"],
+                    "policy_counts": view["policy_counts"],
+                    "freshness_at_receipt": checked.to_dict(),
+                    "policy_counts_at_publication": scene["policy_counts"],
                     "scene_bytes": len(encoded),
                     "scene_sha256": hashlib.sha256(encoded).hexdigest(),
                 }
@@ -711,8 +686,8 @@ def run(args):
         and max(s.get("cgroup_memory_mib", 99999) for s in samples) <= 8192,
         "all_modalities_fresh": len(results) == args.frames
         and all(r["available"] for r in results),
-        "layer_source_age_250ms": bool(results)
+        "all_six_layers_fresh_at_receipt": bool(results)
-        and all(r["ddrnet_source_age_ms"] <= 250 for r in results),
+        and all(r["freshness_at_receipt"]["fresh_complete"] for r in results),
         "network_end_to_end_qualified": False,
     }
     report["profile_realtime_qualified"] = False

src/k1link/perception/realtime_contract.py

@@ -181,6 +181,9 @@ class LayerEvidence:
     source_time_ns: int | None
     state: str
     payload_sha256: str | None
+    # Old callers omit this field. New layered results require it explicitly:
+    # recomputing a derived layer cannot refresh the age of its required inputs.
+    oldest_required_input_time_ns: int | None = None
 
     def __post_init__(self) -> None:
         if self.layer not in REQUIRED_LAYERS:
@@ -191,13 +194,22 @@ class LayerEvidence:
         if self.state == "unavailable":
             if any(
                 x is not None
-                for x in (self.source_sequence, self.source_time_ns, self.payload_sha256)
+                for x in (
+                    self.source_sequence,
+                    self.source_time_ns,
+                    self.payload_sha256,
+                    self.oldest_required_input_time_ns,
+                )
             ):
                 raise RealtimeContractError("unavailable evidence cannot claim a payload")
         else:
             _integer(self.source_sequence, "source_sequence")
             _integer(self.source_time_ns, "source_time_ns")
             _digest(self.payload_sha256, "payload_sha256")
+            if self.oldest_required_input_time_ns is not None:
+                _integer(self.oldest_required_input_time_ns, "oldest_required_input_time_ns")
+                if self.oldest_required_input_time_ns > cast(int, self.source_time_ns):
+                    raise RealtimeContractError("required input is newer than its layer")
 
 
 def validate_scene_layers(
@@ -223,7 +235,10 @@ def validate_scene_layers(
             raise RealtimeContractError("layer belongs to another stream epoch")
         if layer.source_time_ns is None:
             continue
-        age_ns = source_time_ns - layer.source_time_ns
+        if layer.source_time_ns > source_time_ns:
+            raise RealtimeContractError("future observations are not causal input")
+        stamp = layer.oldest_required_input_time_ns
+        age_ns = source_time_ns - (layer.source_time_ns if stamp is None else stamp)
         if age_ns < 0:
             raise RealtimeContractError("future observations are not causal input")
         if age_ns > maximum_layer_age_ms * 1_000_000 and layer.state in ("current", "held"):

src/k1link/perception/realtime_scene.py

@@ -0,0 +1,264 @@
+"""Bounded six-layer freshness ABI; no I/O, models or control authority.
+
+This wraps existing domain payloads. It is not the transport/session admission
+protocol: a transport must also validate StreamStart identities and lease fences.
+All times here are in ONE mapped source timeline, never raw remote worker clocks.
+"""
+
+from __future__ import annotations
+
+from dataclasses import asdict, dataclass, replace
+from typing import Final
+
+from .realtime_contract import (
+    REQUIRED_LAYERS,
+    LayerEvidence,
+    RealtimeContractError,
+    _identifier,
+    _integer,
+    _number,
+)
+
+SCENE_FRESHNESS_SCHEMA: Final = "missioncore.perception-scene-freshness/v1"
+_INT64_FIELDS: Final = ("source_sequence", "source_time_ns", "oldest_required_input_time_ns")
+
+
+def _wire_integer(value: int) -> str:
+    _integer(value, "wire integer")
+    if value > 2**63 - 1:
+        raise RealtimeContractError("wire integer exceeds int64")
+    return str(value)
+
+
+def _read_wire_integer(value: object) -> int:
+    if (
+        not isinstance(value, str)
+        or not value.isascii()
+        or not value.isdecimal()
+        or len(value) > 19
+        or (len(value) > 1 and value.startswith("0"))
+    ):
+        raise RealtimeContractError("int64 must be a canonical decimal string")
+    parsed = int(value)
+    _wire_integer(parsed)
+    return parsed
+
+
+def _layer_document(item: LayerEvidence) -> dict[str, object]:
+    value = asdict(item)
+    for name in _INT64_FIELDS:
+        number = value[name]
+        value[name] = None if number is None else _wire_integer(number)
+    return value
+
+
+# Geometry includes detector association; history may add prohibitions, not
+# independently authorize terrain. This is the full profile's dependency graph.
+DEPENDENCIES: Final = {
+    "segmentation": (),
+    "objects": (),
+    "geometry": ("objects",),
+    "motion": ("geometry",),
+    "costmap": ("segmentation", "geometry", "motion"),
+    "policy": ("costmap", "objects", "motion"),
+}
+
+
+def derive_layer(
+    layer: str,
+    *,
+    epoch_id: str,
+    source_sequence: int,
+    source_time_ns: int,
+    payload_sha256: str,
+    inputs: tuple[LayerEvidence, ...],
+    oldest_required_input_time_ns: int,
+) -> LayerEvidence:
+    """Propagate missing/stale inputs and their original ages, not execution time."""
+    if layer not in DEPENDENCIES or tuple(x.layer for x in inputs) != DEPENDENCIES[layer]:
+        raise RealtimeContractError("derived layer dependencies changed")
+    if any(x.epoch_id != epoch_id for x in inputs):
+        raise RealtimeContractError("derived input belongs to another epoch")
+    if any(x.state == "unavailable" for x in inputs):
+        return LayerEvidence(layer, epoch_id, None, None, "unavailable", None)
+    stamps = [oldest_required_input_time_ns]
+    for item in inputs:
+        if item.oldest_required_input_time_ns is None:
+            raise RealtimeContractError("required input age is missing")
+        stamps.append(item.oldest_required_input_time_ns)
+    return LayerEvidence(
+        layer,
+        epoch_id,
+        source_sequence,
+        source_time_ns,
+        "stale" if any(x.state == "stale" for x in inputs) else "current",
+        payload_sha256,
+        min(stamps),
+    )
+
+
+@dataclass(frozen=True, slots=True)
+class FreshnessAssessment:
+    checked_at_source_time_ns: int
+    clock_uncertainty_ms: float
+    layers: tuple[LayerEvidence, ...]
+    age_upper_bound_ms: tuple[float | None, ...]
+    failures: tuple[str, ...]
+
+    @property
+    def fresh_complete(self) -> bool:
+        return not self.failures
+
+    def to_dict(self) -> dict[str, object]:
+        return {
+            "checked_at_source_time_ns": _wire_integer(self.checked_at_source_time_ns),
+            "clock_uncertainty_ms": self.clock_uncertainty_ms,
+            "layers": {
+                item.layer: {**_layer_document(item), "age_upper_bound_ms": age}
+                for item, age in zip(self.layers, self.age_upper_bound_ms, strict=True)
+            },
+            "fresh_complete": self.fresh_complete,
+            "failures": list(self.failures),
+            "commands_enabled": False,
+            "actuation_allowed": False,
+        }
+
+
+@dataclass(frozen=True, slots=True)
+class SceneFreshness:
+    epoch_id: str
+    clock_domain_id: str
+    source_sequence: int
+    source_time_ns: int
+    layers: tuple[LayerEvidence, ...]
+
+    def __post_init__(self) -> None:
+        _identifier(self.epoch_id, "epoch_id")
+        _identifier(self.clock_domain_id, "clock_domain_id")
+        _integer(self.source_sequence, "source_sequence")
+        _integer(self.source_time_ns, "source_time_ns")
+        if (
+            not isinstance(self.layers, tuple)
+            or tuple(x.layer for x in self.layers) != REQUIRED_LAYERS
+        ):
+            raise RealtimeContractError("six layers required in canonical order")
+        indexed = {item.layer: item for item in self.layers}
+        for item in self.layers:
+            if item.epoch_id != self.epoch_id:
+                raise RealtimeContractError("layer belongs to another epoch")
+            if item.state == "unavailable":
+                continue
+            if (
+                item.source_sequence is None
+                or item.source_time_ns is None
+                or item.oldest_required_input_time_ns is None
+            ):
+                raise RealtimeContractError("layer input lineage missing")
+            if (
+                item.source_sequence > self.source_sequence
+                or item.source_time_ns > self.source_time_ns
+            ):
+                raise RealtimeContractError("future layer evidence")
+            if item.state == "current" and (
+                item.source_sequence != self.source_sequence
+                or item.source_time_ns != self.source_time_ns
+            ):
+                raise RealtimeContractError("retained evidence must be held, not current")
+            if item.state == "held" and item.source_sequence == self.source_sequence:
+                raise RealtimeContractError("held layer must retain an earlier observation")
+            for name in DEPENDENCIES[item.layer]:
+                parent = indexed[name]
+                if parent.oldest_required_input_time_ns is None:
+                    raise RealtimeContractError("available layer depends on unavailable input")
+                if item.oldest_required_input_time_ns > parent.oldest_required_input_time_ns:
+                    raise RealtimeContractError("derived layer refreshed an older input")
+                if parent.state == "stale" and item.state != "stale":
+                    raise RealtimeContractError("derived layer hid stale input")
+
+    def to_dict(self) -> dict[str, object]:
+        return {
+            "schema_version": SCENE_FRESHNESS_SCHEMA,
+            "epoch_id": self.epoch_id,
+            "clock_domain_id": self.clock_domain_id,
+            "source_sequence": _wire_integer(self.source_sequence),
+            "source_time_ns": _wire_integer(self.source_time_ns),
+            "layers": [_layer_document(item) for item in self.layers],
+        }
+
+    @classmethod
+    def from_dict(cls, value: object) -> SceneFreshness:
+        if not isinstance(value, dict) or set(value) != set(cls.__dataclass_fields__) | {
+            "schema_version"
+        }:
+            raise RealtimeContractError("scene freshness fields changed")
+        if value["schema_version"] != SCENE_FRESHNESS_SCHEMA:
+            raise RealtimeContractError("scene freshness schema changed")
+        raw = value["layers"]
+        if not isinstance(raw, list) or len(raw) != len(REQUIRED_LAYERS):
+            raise RealtimeContractError("six layers required")
+        if any(
+            not isinstance(x, dict) or set(x) != set(LayerEvidence.__dataclass_fields__)
+            for x in raw
+        ):
+            raise RealtimeContractError("layer fields changed")
+        layers = []
+        for item in raw:
+            parsed = dict(item)
+            for name in _INT64_FIELDS:
+                parsed[name] = None if item[name] is None else _read_wire_integer(item[name])
+            layers.append(LayerEvidence(**parsed))
+        return cls(
+            value["epoch_id"],
+            value["clock_domain_id"],
+            _read_wire_integer(value["source_sequence"]),
+            _read_wire_integer(value["source_time_ns"]),
+            tuple(layers),
+        )
+
+    def assess(
+        self,
+        *,
+        epoch_id: str,
+        clock_domain_id: str,
+        observed_source_time_ns: int,
+        clock_uncertainty_ms: float,
+        maximum_clock_uncertainty_ms: float,
+        maximum_layer_age_ms: float,
+    ) -> FreshnessAssessment:
+        """Re-evaluate at publication, receipt AND use. Uncertainty adds to age.
+
+        Unknown/unmapped clocks must not call this with a guessed zero offset.
+        The source adapter owns the mapping; excessive known uncertainty yields
+        degraded output, never a complete fresh-scene claim.
+        """
+        if (epoch_id, clock_domain_id) != (self.epoch_id, self.clock_domain_id):
+            raise RealtimeContractError("epoch or mapped clock domain mismatch")
+        _integer(observed_source_time_ns, "observed_source_time_ns")
+        _number(clock_uncertainty_ms, "clock_uncertainty_ms")
+        _number(maximum_clock_uncertainty_ms, "maximum_clock_uncertainty_ms", minimum=0.001)
+        _number(maximum_layer_age_ms, "maximum_layer_age_ms", minimum=0.001)
+        if observed_source_time_ns < self.source_time_ns:
+            raise RealtimeContractError("receipt precedes source observation")
+        clock_failed = clock_uncertainty_ms > maximum_clock_uncertainty_ms
+        failures = ["clock-uncertainty"] if clock_failed else []
+        evaluated, ages = [], []
+        for item in self.layers:
+            stamp = item.oldest_required_input_time_ns
+            age = (
+                None
+                if stamp is None
+                else (observed_source_time_ns - stamp) / 1e6 + clock_uncertainty_ms
+            )
+            if age is not None and (age > maximum_layer_age_ms or clock_failed):
+                item = replace(item, state="stale")
+            if item.state in ("stale", "unavailable"):
+                failures.append(f"{item.layer}-{item.state}")
+            evaluated.append(item)
+            ages.append(age)
+        return FreshnessAssessment(
+            observed_source_time_ns,
+            clock_uncertainty_ms,
+            tuple(evaluated),
+            tuple(ages),
+            tuple(failures),
+        )

src/k1link/perception/worker_operating_envelope.py

@@ -0,0 +1,153 @@
+"""Read-only post-warmup readiness checks, not GPU management or qualification.
+
+An envelope is preregistered for a measured hardware/software context. A failed
+check does not prohibit an explicitly labelled overload experiment. Passing is
+only a prerequisite: it cannot certify clocks throughout a run, latency, model
+quality, standalone packaging or network delivery. No host setters live here.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from .realtime_contract import (
+    RealtimeContractError,
+    StreamStart,
+    _digest,
+    _identifier,
+    _integer,
+)
+
+
+@dataclass(frozen=True, slots=True)
+class WorkerOperatingEnvelope:
+    envelope_id: str
+    gpu_name: str
+    driver_version: str
+    # Runtime versions and all model/CPU-thread settings belong to the pinned
+    # image/effective config in StreamStart; resource limits are explicit here.
+    cpu_limit_millicores: int
+    memory_limit_mib: int
+    minimum_sm_clock_mhz: int
+    minimum_memory_clock_mhz: int
+    maximum_snapshot_age_ms: int = 1000
+
+    def __post_init__(self) -> None:
+        _identifier(self.envelope_id, "envelope_id")
+        for field in ("gpu_name", "driver_version"):
+            value = getattr(self, field)
+            if not isinstance(value, str) or not value.strip() or len(value) > 160:
+                raise RealtimeContractError(f"{field} is missing or unbounded")
+        for field in (
+            "cpu_limit_millicores",
+            "memory_limit_mib",
+            "minimum_sm_clock_mhz",
+            "minimum_memory_clock_mhz",
+            "maximum_snapshot_age_ms",
+        ):
+            _integer(getattr(self, field), field, minimum=1)
+
+
+@dataclass(frozen=True, slots=True)
+class WorkerSnapshot:
+    worker_id: str
+    clock_domain_id: str
+    observed_monotonic_ns: int
+    gpu_name: str | None
+    driver_version: str | None
+    image_sha256: str | None
+    effective_config_sha256: str | None
+    cpu_limit_millicores: int | None
+    memory_limit_mib: int | None
+    sm_clock_mhz: int | None
+    memory_clock_mhz: int | None
+    # Authoritative controller lease, NOT inferred from low GPU utilization.
+    gpu_owner_run_id: str | None
+    lease_generation: int | None
+    competing_gpu_clients: tuple[str, ...] | None
+    warmup_complete: bool | None
+
+    def __post_init__(self) -> None:
+        _identifier(self.worker_id, "worker_id")
+        _identifier(self.clock_domain_id, "clock_domain_id")
+        _integer(self.observed_monotonic_ns, "observed_monotonic_ns")
+        for field in ("image_sha256", "effective_config_sha256"):
+            value = getattr(self, field)
+            if value is not None:
+                _digest(value, field)
+        for field in ("gpu_name", "driver_version"):
+            value = getattr(self, field)
+            if value is not None and (
+                not isinstance(value, str) or not value.strip() or len(value) > 160
+            ):
+                raise RealtimeContractError(f"{field} is invalid")
+        for field in (
+            "cpu_limit_millicores",
+            "memory_limit_mib",
+            "sm_clock_mhz",
+            "memory_clock_mhz",
+        ):
+            value = getattr(self, field)
+            if value is not None:
+                # Docker 0 means unlimited; never confuse it with unknown=None.
+                _integer(value, field)
+        if self.gpu_owner_run_id is not None:
+            _identifier(self.gpu_owner_run_id, "gpu_owner_run_id")
+        if self.lease_generation is not None:
+            _integer(self.lease_generation, "lease_generation", minimum=1)
+        if self.competing_gpu_clients is not None:
+            if (
+                not isinstance(self.competing_gpu_clients, tuple)
+                or len(self.competing_gpu_clients) > 64
+            ):
+                raise RealtimeContractError("GPU client inventory is unbounded")
+            for client in self.competing_gpu_clients:
+                _identifier(client, "competing_gpu_client")
+        if self.warmup_complete is not None and type(self.warmup_complete) is not bool:
+            raise RealtimeContractError("warmup_complete must be boolean or unknown")
+
+
+def operating_envelope_failures(
+    expected: WorkerOperatingEnvelope,
+    start: StreamStart,
+    observed: WorkerSnapshot,
+    *,
+    now_monotonic_ns: int,
+    clock_domain_id: str,
+) -> tuple[str, ...]:
+    """Check facts collected in the Worker's clock domain by a trusted controller.
+
+    No resource expansion, frequency lock, container stop, lease acquisition or
+    host mutation. Snapshot refresh and ongoing telemetry remain caller-owned.
+    Changing hardware/resource conditions requires a new measured envelope, not
+    pretending that an old qualification applies to all compatible machines.
+    """
+    _integer(now_monotonic_ns, "now_monotonic_ns")
+    if observed.clock_domain_id != clock_domain_id:
+        raise RealtimeContractError("worker snapshot clock domain mismatch")
+    age = now_monotonic_ns - observed.observed_monotonic_ns
+    if age < 0:
+        raise RealtimeContractError("worker snapshot is from the future")
+    failures = []
+    if age > expected.maximum_snapshot_age_ms * 1_000_000:
+        failures.append("worker-snapshot-expired")
+    for field in ("worker_id", "image_sha256", "effective_config_sha256"):
+        if getattr(observed, field) != getattr(start, field):
+            failures.append(f"{field}-mismatch-or-unknown")
+    for field in ("gpu_name", "driver_version", "cpu_limit_millicores", "memory_limit_mib"):
+        if getattr(observed, field) != getattr(expected, field):
+            failures.append(f"{field}-outside-envelope-or-unknown")
+    for field in ("sm_clock_mhz", "memory_clock_mhz"):
+        value = getattr(observed, field)
+        if value is None or value < getattr(expected, "minimum_" + field):
+            failures.append(f"{field}-below-envelope-or-unknown")
+    if (observed.gpu_owner_run_id, observed.lease_generation) != (
+        start.run_id,
+        start.lease_generation,
+    ):
+        failures.append("exclusive-worker-lease-unproved")
+    if observed.competing_gpu_clients != ():
+        failures.append("competing-gpu-clients-or-inventory-unknown")
+    if observed.warmup_complete is not True:
+        failures.append("warmup-not-complete")
+    return tuple(failures)

tests/test_perception_realtime_scene.py

@@ -0,0 +1,298 @@
+"""Small synthetic scene/consumer tests; no models, network or source archives."""
+
+import copy
+import importlib
+import json
+from dataclasses import replace
+from pathlib import Path
+
+import pytest
+
+from k1link.perception.realtime_contract import (
+    REQUIRED_LAYERS,
+    LayerEvidence,
+    RealtimeContractError,
+)
+from k1link.perception.realtime_scene import DEPENDENCIES, SceneFreshness, derive_layer
+
+NOW = 1_000_000_000
+
+
+def scene(*, held_ms=0, missing=False):
+    layers = {
+        "segmentation": LayerEvidence(
+            "segmentation",
+            "epoch",
+            1 if held_ms else 2,
+            NOW - held_ms * 1_000_000,
+            "held" if held_ms else "current",
+            "a" * 64,
+            NOW - held_ms * 1_000_000,
+        ),
+        "objects": LayerEvidence("objects", "epoch", 2, NOW, "current", "b" * 64, NOW),
+    }
+    for name in ("geometry", "motion", "costmap", "policy"):
+        if name == "geometry" and missing:
+            layers[name] = LayerEvidence(name, "epoch", None, None, "unavailable", None)
+        else:
+            layers[name] = derive_layer(
+                name,
+                epoch_id="epoch",
+                source_sequence=2,
+                source_time_ns=NOW,
+                payload_sha256="c" * 64,
+                inputs=tuple(layers[x] for x in DEPENDENCIES[name]),
+                oldest_required_input_time_ns=NOW,
+            )
+    return SceneFreshness("epoch", "mapped-source", 2, NOW, tuple(layers.values()))
+
+
+def assess(value, *, elapsed_ms=0, **kwargs):
+    arguments = dict(
+        epoch_id="epoch",
+        clock_domain_id="mapped-source",
+        observed_source_time_ns=NOW + elapsed_ms * 1_000_000,
+        clock_uncertainty_ms=0,
+        maximum_clock_uncertainty_ms=5,
+        maximum_layer_age_ms=250,
+    )
+    arguments.update(kwargs)
+    return value.assess(**arguments)
+
+
+def test_six_layer_round_trip_is_bounded_and_needs_no_eof():
+    value = scene()
+    wire = json.dumps(value.to_dict())
+    assert len(wire.encode()) < 8192
+    assert SceneFreshness.from_dict(json.loads(wire)) == value
+    checked = assess(value, elapsed_ms=125)
+    assert checked.fresh_complete
+    assert checked.age_upper_bound_ms == (125,) * 6
+    assert checked.to_dict()["actuation_allowed"] is False
+
+
+def test_json_projection_keeps_int64_exact_above_javascript_safe_integer():
+    value = scene()
+    large = 9_007_199_254_740_993
+    value = replace(
+        value,
+        source_time_ns=large,
+        layers=tuple(
+            replace(x, source_time_ns=large, oldest_required_input_time_ns=large)
+            for x in value.layers
+        ),
+    )
+    wire = value.to_dict()
+    assert wire["source_time_ns"] == str(large)
+    assert SceneFreshness.from_dict(json.loads(json.dumps(wire))) == value
+
+
+@pytest.mark.parametrize("number", [1_000_000_000, "01", "-1", "1.0", str(2**63)])
+def test_wire_rejects_lossy_or_noncanonical_int64(number):
+    wire = scene().to_dict()
+    wire["source_time_ns"] = number
+    with pytest.raises(RealtimeContractError):
+        SceneFreshness.from_dict(wire)
+
+
+def test_held_segmenter_keeps_original_identity_through_costmap_and_policy():
+    value = scene(held_ms=150)
+    assert value.layers[0].source_sequence == 1
+    assert value.layers[-1].source_sequence == 2
+    assert value.layers[-1].oldest_required_input_time_ns == NOW - 150_000_000
+    assert assess(value, elapsed_ms=100).fresh_complete
+    expired = assess(value, elapsed_ms=101)
+    assert expired.failures == ("segmentation-stale", "costmap-stale", "policy-stale")
+    assert expired.layers[-1].source_time_ns == NOW  # recomputation is not a new input
+
+
+def test_receipt_latency_cannot_inherit_fresh_publication():
+    value = scene()
+    assert assess(value, elapsed_ms=249).fresh_complete
+    assert not assess(value, elapsed_ms=251).fresh_complete
+    assert all(x.state == "stale" for x in assess(value, elapsed_ms=251).layers)
+
+
+def test_missing_geometry_propagates_without_fabricating_zero_distance():
+    value = scene(missing=True)
+    checked = assess(value)
+    assert not checked.fresh_complete
+    assert [x.layer for x in checked.layers if x.state == "unavailable"] == list(
+        REQUIRED_LAYERS[2:]
+    )
+    assert all(x.payload_sha256 is None for x in checked.layers[2:])
+    assert checked.age_upper_bound_ms[2:] == (None,) * 4
+
+
+@pytest.mark.parametrize(
+    "field,value",
+    [
+        ("epoch_id", "old-epoch"),
+        ("clock_domain_id", "raw-worker-clock"),
+        ("observed_source_time_ns", NOW - 1),
+        ("clock_uncertainty_ms", None),
+        ("clock_uncertainty_ms", float("nan")),
+        ("clock_uncertainty_ms", True),
+    ],
+)
+def test_wrong_or_unmapped_clock_rejected(field, value):
+    with pytest.raises(RealtimeContractError):
+        assess(scene(), **{field: value})
+
+
+def test_known_clock_uncertainty_counts_against_budget():
+    assert assess(scene(), elapsed_ms=245, clock_uncertainty_ms=5).fresh_complete
+    assert not assess(scene(), elapsed_ms=246, clock_uncertainty_ms=5).fresh_complete
+    assert "clock-uncertainty" in assess(scene(), clock_uncertainty_ms=6).failures
+
+
+@pytest.mark.parametrize(
+    "change",
+    [
+        {"source_sequence": 3},
+        {"source_time_ns": NOW + 1},
+        {"epoch_id": "old"},
+        {"source_sequence": 1},
+        {"state": "held"},
+        {"oldest_required_input_time_ns": None},
+    ],
+)
+def test_false_currentness_or_future_lineage_rejected(change):
+    value = scene()
+    layers = (replace(value.layers[0], **change), *value.layers[1:])
+    with pytest.raises(RealtimeContractError):
+        replace(value, layers=layers)
+
+
+def test_derived_layer_cannot_erase_old_or_missing_input():
+    value = scene(held_ms=150)
+    with pytest.raises(RealtimeContractError, match="refreshed"):
+        replace(
+            value,
+            layers=(
+                *value.layers[:-1],
+                replace(value.layers[-1], oldest_required_input_time_ns=NOW),
+            ),
+        )
+    missing = scene(missing=True)
+    with pytest.raises(RealtimeContractError, match="unavailable"):
+        replace(missing, layers=(*missing.layers[:-1], scene().layers[-1]))
+
+
+def test_stale_input_cannot_be_relabelled_fresh():
+    value = scene()
+    with pytest.raises(RealtimeContractError, match="stale"):
+        replace(value, layers=(replace(value.layers[0], state="stale"), *value.layers[1:]))
+
+
+@pytest.mark.parametrize(
+    "mutation",
+    [
+        lambda v: v.update(command="drive"),
+        lambda v: v.update(schema_version="unknown"),
+        lambda v: v["layers"].pop(),
+        lambda v: v["layers"].append(v["layers"][0]),
+        lambda v: v["layers"][0].update(age_ms=0),
+        lambda v: v["layers"].__setitem__(1, v["layers"][0]),
+    ],
+)
+def test_wire_schema_is_closed(mutation):
+    value = scene().to_dict()
+    mutation(value)
+    with pytest.raises(RealtimeContractError):
+        SceneFreshness.from_dict(value)
+
+
+@pytest.fixture
+def pilot(monkeypatch):
+    monkeypatch.syspath_prepend(
+        str(
+            Path(__file__).resolve().parents[1]
+            / "experiments/perception/worker/streaming_profile_stage1"
+        )
+    )
+    return importlib.import_module("pilot_freshness")
+
+
+def pilot_input():
+    payload = dict(
+        segmentation_sha256="a" * 64,
+        proposals=[],
+        observations=[],
+        tracks=[],
+        threats=[],
+        surface_state="valid",
+        range_estimator={"detected": "median-camera-z"},
+        costmap_states=[1, 2],
+        costmap_material=[1, 0],
+        policy_actions=[0, 2],
+        policy_counts={"ALLOW_candidate": 1, "HIGH_COST": 0, "NO_GO": 1},
+        tgs_counts={"oldest_permissive_cell_source_ns": NOW - 100_000_000},
+        commands_enabled=False,
+        actuation_allowed=False,
+    )
+    bundle = dict(
+        time_ns=NOW,
+        sequence=2,
+        due_ns=5_000_000_000,
+        available=True,
+        lineage={
+            "pose_host_monotonic_ns": NOW - 10_000_000,
+            "point_increments": [{"host_monotonic_ns": NOW - 20_000_000}],
+        },
+    )
+    ddr = dict(state="current", source_sequence=2, source_host_monotonic_ns=NOW)
+    return payload, bundle, ddr
+
+
+def test_pilot_receiver_checks_real_payloads_and_ages_after_serialization(pilot):
+    payload, bundle, ddr = pilot_input()
+    pilot.prepare_publication(
+        payload, bundle, ddr, epoch_id="pilot", now_ns=bundle["due_ns"] + 50_000_000
+    )
+    assert payload["freshness_at_publication"]["fresh_complete"]
+    assert payload["policy_actions"] == [0, 2]
+    received = json.loads(json.dumps(payload))
+    freshness = pilot.validate_receipt(received, bundle, epoch_id="pilot")
+    # The costmap's last observed permissive cell ages beyond 250 ms in transit.
+    checked = pilot.assess(freshness, bundle=bundle, now_ns=bundle["due_ns"] + 151_000_000)
+    view = pilot.receipt_view(received, checked)
+    assert checked.failures == ("costmap-stale", "policy-stale")
+    assert view["policy_actions"] == [2, 2]
+    assert received["policy_actions"] == [0, 2]  # published evidence remains immutable
+    received["costmap_material"][0] = 2
+    with pytest.raises(ValueError, match="digest"):
+        pilot.validate_receipt(received, bundle, epoch_id="pilot")
+
+
+def test_empty_detection_is_valid_but_missing_lidar_is_not(pilot):
+    payload, bundle, ddr = pilot_input()
+    bundle["available"] = False
+    pilot.prepare_publication(payload, bundle, ddr, epoch_id="pilot", now_ns=bundle["due_ns"])
+    fresh = pilot.validate_receipt(payload, bundle, epoch_id="pilot")
+    assert fresh.layers[1].state == "current"
+    assert all(x.state == "unavailable" for x in fresh.layers[2:])
+    assert payload["policy_actions"] == [2, 2]
+    assert payload["observations"] == []
+
+
+def test_failed_online_surface_is_not_fresh_geometry_even_with_sensor_pair(pilot):
+    payload, bundle, ddr = pilot_input()
+    payload["surface_state"] = "fit-failed"
+    pilot.prepare_publication(payload, bundle, ddr, epoch_id="pilot", now_ns=bundle["due_ns"])
+    fresh = pilot.validate_receipt(payload, bundle, epoch_id="pilot")
+    assert all(x.state == "unavailable" for x in fresh.layers[2:])
+    assert payload["policy_actions"] == [2, 2]
+
+
+def test_pilot_rejects_wrong_run_sequence_and_control_authority(pilot):
+    payload, bundle, ddr = pilot_input()
+    pilot.prepare_publication(payload, bundle, ddr, epoch_id="pilot", now_ns=bundle["due_ns"])
+    with pytest.raises(ValueError, match="identity"):
+        pilot.validate_receipt(payload, bundle, epoch_id="old-pilot")
+    with pytest.raises(ValueError, match="identity"):
+        pilot.validate_receipt(payload, {**bundle, "sequence": 3}, epoch_id="pilot")
+    changed = copy.deepcopy(payload)
+    changed["commands_enabled"] = True
+    with pytest.raises(ValueError, match="authority"):
+        pilot.validate_receipt(changed, bundle, epoch_id="pilot")

tests/test_worker_operating_envelope.py

@@ -0,0 +1,142 @@
+"""Synthetic readiness checks; passing must never mutate a Worker or grant control."""
+
+import json
+from dataclasses import replace
+from pathlib import Path
+
+import pytest
+
+from k1link.perception.realtime_contract import RealtimeContractError, StreamStart
+from k1link.perception.worker_operating_envelope import (
+    WorkerOperatingEnvelope,
+    WorkerSnapshot,
+    operating_envelope_failures,
+)
+
+
+def start():
+    return StreamStart(
+        "run-1",
+        "source-1",
+        "worker-006",
+        "epoch-1",
+        1,
+        "a" * 64,
+        "b" * 64,
+        "c" * 64,
+        "d" * 64,
+        "source-clock",
+        "recorded-source-paced",
+    )
+
+
+def envelope():
+    return WorkerOperatingEnvelope(
+        "synthetic-4090-fixed/v1", "RTX 4090", "610.47", 8000, 8192, 2610, 10251
+    )
+
+
+def snapshot():
+    return WorkerSnapshot(
+        "worker-006",
+        "worker-monotonic",
+        1_000_000_000,
+        "RTX 4090",
+        "610.47",
+        "b" * 64,
+        "c" * 64,
+        8000,
+        8192,
+        2610,
+        10251,
+        "run-1",
+        1,
+        (),
+        True,
+    )
+
+
+def check(value, **kwargs):
+    options = {"now_monotonic_ns": 1_100_000_000, "clock_domain_id": "worker-monotonic"}
+    options.update(kwargs)
+    return operating_envelope_failures(envelope(), start(), value, **options)
+
+
+def test_exact_measured_conditions_pass_only_readiness():
+    value = snapshot()
+    assert check(value) == ()
+    assert value == snapshot()
+    assert "qualified" not in value.__dataclass_fields__
+
+
+def test_candidate_preregisters_conditions_without_installing_host_policy():
+    path = (
+        Path(__file__).resolve().parents[1]
+        / "config/perception/k1-perception-ddrnet39-rfdetr-tgs-prototype-v1.json"
+    )
+    profile = json.loads(path.read_text())
+    value = profile["operating_envelope"]
+    parsed = WorkerOperatingEnvelope(**value["reference_conditions"])
+    assert parsed.minimum_memory_clock_mhz == 10251
+    assert not value["container_may_set_host_clocks"]
+    assert not value["readiness_is_realtime_qualification"]
+    assert value["failed_envelope_allows_labelled_experiment"]
+    assert profile["packaging"]["image_sha256"] is None
+
+
+@pytest.mark.parametrize(
+    "field,value,reason",
+    [
+        ("worker_id", "worker-007", "worker_id-mismatch"),
+        ("gpu_name", "RTX 5090", "gpu_name-outside"),
+        ("driver_version", "new-driver", "driver_version-outside"),
+        ("image_sha256", "e" * 64, "image_sha256-mismatch"),
+        ("effective_config_sha256", "e" * 64, "effective_config_sha256-mismatch"),
+        ("sm_clock_mhz", 450, "sm_clock_mhz-below"),
+        ("memory_clock_mhz", 405, "memory_clock_mhz-below"),
+        ("memory_clock_mhz", None, "memory_clock_mhz-below"),
+        ("cpu_limit_millicores", 4000, "cpu_limit_millicores-outside"),
+        ("memory_limit_mib", 0, "memory_limit_mib-outside"),
+        ("cpu_limit_millicores", None, "cpu_limit_millicores-outside"),
+        ("gpu_owner_run_id", None, "exclusive-worker-lease"),
+        ("gpu_owner_run_id", "old-run", "exclusive-worker-lease"),
+        ("lease_generation", 2, "exclusive-worker-lease"),
+        ("competing_gpu_clients", ("other-model",), "competing-gpu-clients"),
+        ("competing_gpu_clients", None, "competing-gpu-clients"),
+        ("warmup_complete", None, "warmup-not-complete"),
+        ("warmup_complete", False, "warmup-not-complete"),
+    ],
+)
+def test_auto_clocks_wrong_identity_unknown_or_busy_never_pass(field, value, reason):
+    assert any(x.startswith(reason) for x in check(replace(snapshot(), **{field: value})))
+
+
+def test_low_utilization_does_not_prove_ownership_and_idle_does_not_prove_warmup():
+    failures = check(replace(snapshot(), gpu_owner_run_id=None, warmup_complete=False))
+    assert failures == ("exclusive-worker-lease-unproved", "warmup-not-complete")
+
+
+def test_snapshot_age_and_clock_domain_cannot_be_ignored():
+    assert check(snapshot(), now_monotonic_ns=2_000_000_000) == ()
+    assert check(snapshot(), now_monotonic_ns=2_000_000_001) == ("worker-snapshot-expired",)
+    with pytest.raises(RealtimeContractError, match="clock domain"):
+        check(snapshot(), clock_domain_id="mac-monotonic")
+    with pytest.raises(RealtimeContractError, match="future"):
+        check(snapshot(), now_monotonic_ns=999_999_999)
+
+
+@pytest.mark.parametrize(
+    "change",
+    [
+        {"sm_clock_mhz": True},
+        {"memory_limit_mib": -1},
+        {"warmup_complete": 1},
+        {"image_sha256": "latest"},
+        {"lease_generation": 0},
+        {"competing_gpu_clients": ["unbounded-list"]},
+        {"competing_gpu_clients": ("client",) * 65},
+    ],
+)
+def test_invalid_facts_fail_closed(change):
+    with pytest.raises(RealtimeContractError):
+        replace(snapshot(), **change)