feat(perception): expire costmap permissions per cell

experiments/perception/worker/streaming_profile_stage1/pilot_freshness.py

@@ -7,9 +7,17 @@ the scene, but makes no valid payload/terrain-permission claim in the ABI.
 
 import hashlib
 import json
+from dataclasses import replace
 
+from k1link.perception.costmap_freshness import CostmapCellEvidence
 from k1link.perception.realtime_contract import LayerEvidence
-from k1link.perception.realtime_scene import DEPENDENCIES, SceneFreshness, derive_layer
+from k1link.perception.realtime_scene import (
+    DEPENDENCIES,
+    SceneFreshness,
+    _read_wire_integer,
+    _wire_integer,
+    derive_layer,
+)
 
 CLOCK_DOMAIN = "pilot-original-host-arrival"
 
@@ -25,6 +33,9 @@ def payload_digest(scene, layer):
         "policy": ("policy_actions", "policy_counts"),
     }[layer]
     value = {key: scene[key] for key in fields}
+    if layer == "costmap" and "costmap_cell_evidence" in scene:
+        value["costmap_cell_evidence"] = scene["costmap_cell_evidence"]
+        value["costmap_freshness_mode"] = scene["costmap_freshness_mode"]
     return hashlib.sha256(
         json.dumps(value, sort_keys=True, allow_nan=False, separators=(",", ":")).encode()
     ).hexdigest()
@@ -59,7 +70,11 @@ def build_freshness(scene, bundle, ddr_layer, epoch_id):
         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"]
+            oldest = (
+                _read_wire_integer(scene["cell_assessment"]["oldest_permissive_source_time_ns"])
+                if scene.get("costmap_freshness_mode") == "per-cell"
+                else scene["tgs_counts"]["oldest_permissive_cell_source_ns"]
+            )
         layers[name] = derive_layer(
             name,
             epoch_id=epoch_id,
@@ -95,7 +110,41 @@ def suppress_policy(scene):
     }
 
 
-def prepare_publication(scene, bundle, ddr_layer, *, epoch_id, now_ns):
+def apply_cell_expiry(scene, *, source_time_ns, observed_source_time_ns, uncertainty_ms=0):
+    evidence = CostmapCellEvidence.from_dict(scene["costmap_cell_evidence"])
+    checked = evidence.assess(
+        states=tuple(scene["costmap_states"]),
+        actions=tuple(scene["policy_actions"]),
+        source_time_ns=source_time_ns,
+        observed_source_time_ns=observed_source_time_ns,
+        maximum_age_ms=250,
+        clock_uncertainty_ms=uncertainty_ms,
+    )
+    scene["costmap_states"] = list(checked.states)
+    scene["policy_actions"] = list(checked.actions)
+    scene["policy_counts"] = {
+        key: checked.actions.count(index)
+        for index, key in enumerate(("ALLOW_candidate", "HIGH_COST", "NO_GO"))
+    }
+    scene["cell_assessment"] = {
+        "checked_at_source_time_ns": _wire_integer(observed_source_time_ns),
+        "expired_ground_cells": checked.expired_cells,
+        "missing_ground_support_cells": checked.missing_support_cells,
+        "suppressed_permissions": checked.suppressed_permissions,
+        "oldest_permissive_source_time_ns": _wire_integer(checked.oldest_permissive_source_time_ns),
+    }
+
+
+def prepare_publication(scene, bundle, ddr_layer, *, epoch_id, now_ns, mode="whole-scene"):
+    if mode not in ("whole-scene", "per-cell"):
+        raise ValueError("unknown costmap freshness mode")
+    scene["costmap_freshness_mode"] = mode
+    if mode == "per-cell":
+        apply_cell_expiry(
+            scene,
+            source_time_ns=bundle["time_ns"],
+            observed_source_time_ns=bundle["time_ns"] + now_ns - bundle["due_ns"],
+        )
     freshness = build_freshness(scene, bundle, ddr_layer, epoch_id)
     checked = assess(freshness, bundle=bundle, now_ns=now_ns)
     if not checked.fresh_complete:
@@ -112,6 +161,11 @@ def prepare_publication(scene, bundle, ddr_layer, *, epoch_id, now_ns):
 
 
 def validate_receipt(scene, bundle, *, epoch_id):
+    mode = scene.get("costmap_freshness_mode", "whole-scene")
+    if mode not in ("whole-scene", "per-cell"):
+        raise ValueError("unknown costmap freshness mode")
+    if mode == "per-cell" or "costmap_cell_evidence" in scene:
+        CostmapCellEvidence.from_dict(scene["costmap_cell_evidence"])
     freshness = SceneFreshness.from_dict(scene["freshness"])
     if (
         freshness.epoch_id,
@@ -137,3 +191,52 @@ def receipt_view(scene, checked):
     if not checked.fresh_complete:
         suppress_policy(view)
     return view
+
+
+def assess_receipt(scene, freshness, *, bundle, now_ns):
+    """Derive a current consumer view without rewriting published evidence.
+
+    Cell expiry never upgrades a published action. Remaining permissions retain
+    their original support time. Layer identity/source timestamps do not change;
+    updated costmap/policy hashes belong to the derived view, not the wire bytes.
+    """
+    observed = bundle["time_ns"] + now_ns - bundle["due_ns"]
+    previous = scene.get("freshness_at_receipt") or scene["freshness_at_publication"]
+    if observed < _read_wire_integer(previous["checked_at_source_time_ns"]):
+        raise ValueError("consumer clock moved backwards")
+    view = dict(scene)
+    if scene.get("costmap_freshness_mode") == "per-cell":
+        apply_cell_expiry(
+            view,
+            source_time_ns=freshness.source_time_ns,
+            observed_source_time_ns=observed,
+        )
+        layers = {item.layer: item for item in freshness.layers[:4]}
+        for name in ("costmap", "policy"):
+            oldest = (
+                _read_wire_integer(view["cell_assessment"]["oldest_permissive_source_time_ns"])
+                if name == "costmap"
+                else freshness.source_time_ns
+            )
+            layers[name] = derive_layer(
+                name,
+                epoch_id=freshness.epoch_id,
+                source_sequence=freshness.source_sequence,
+                source_time_ns=freshness.source_time_ns,
+                payload_sha256=payload_digest(view, name),
+                inputs=tuple(layers[key] for key in DEPENDENCIES[name]),
+                oldest_required_input_time_ns=oldest,
+            )
+        freshness = replace(freshness, layers=tuple(layers.values()))
+    checked = assess(freshness, bundle=bundle, now_ns=now_ns)
+    view = receipt_view(view, checked)
+    # Global suppression also changes the policy payload. A derived view must
+    # remain internally verifiable, including on a second consumer boundary.
+    policy = freshness.layers[-1]
+    if policy.payload_sha256 is not None:
+        policy = replace(policy, payload_sha256=payload_digest(view, "policy"))
+        freshness = replace(freshness, layers=(*freshness.layers[:-1], policy))
+    checked = assess(freshness, bundle=bundle, now_ns=now_ns)
+    view["freshness"] = freshness.to_dict()
+    view["freshness_at_receipt"] = checked.to_dict()
+    return view, checked

experiments/perception/worker/streaming_profile_stage1/pilot_graph.py

@@ -34,6 +34,7 @@ from k1link.perception.contracts import (
     TemporalState,
     TimestampBundle,
 )
+from k1link.perception.costmap_freshness import CostmapCellEvidence
 from k1link.perception.detector import NativeRfDetrShadowDetectorProvider
 from k1link.perception.geometry import (
     GeometryFrame,
@@ -358,6 +359,7 @@ class JointGraph:
         timing["motion_rolling_threat_ms"] = (time.monotonic_ns() - begin) / 1e6
         begin = time.monotonic_ns()
         costmap, actions, material, tgs_counts = self.costmap(bundle, segmentation, scene_map)
+        cell_evidence = tgs_counts.pop("cell_evidence")
         timing["tgs_costmap_policy_ms"] = (time.monotonic_ns() - begin) / 1e6
         return {
             "timing_ms": timing,
@@ -370,6 +372,7 @@ class JointGraph:
             "tgs_counts": tgs_counts,
             "costmap_states": costmap.tolist(),
             "costmap_material": material.tolist(),
+            "costmap_cell_evidence": cell_evidence,
             "policy_actions": actions.tolist(),
             "policy_counts": {
                 key: int(np.count_nonzero(actions == value))
@@ -395,7 +398,10 @@ class JointGraph:
                 np.zeros(count, np.uint8),
                 np.full(count, 2, np.uint8),
                 material,
-                {"unavailable": True},
+                {
+                    "unavailable": True,
+                    "cell_evidence": CostmapCellEvidence((None,) * count).to_dict(),
+                },
             )
         position, quaternion = bundle["pose"]
         points = bundle["rolling_points"]
@@ -485,6 +491,9 @@ class JointGraph:
                 "rejected": int(np.count_nonzero(states == 3)),
                 "stale_cells": int(np.count_nonzero(stale)),
                 "oldest_permissive_cell_source_ns": oldest_permissive,
+                "cell_evidence": CostmapCellEvidence(
+                    tuple(None if stamp < 0 else int(stamp) for stamp in last_seen)
+                ).to_dict(),
                 "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,7 +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_freshness import assess_receipt, prepare_publication, validate_receipt
 from pilot_ipc import receive, send
 from pilot_queue import Mailbox
 from pilot_scheduler import GpuStage
@@ -263,6 +263,7 @@ def run(args):
         "telemetry_mode": args.telemetry_mode,
         "telemetry_device_state": args.telemetry_device_state,
         "triton_verbose": args.triton_verbose,
+        "costmap_freshness_mode": args.costmap_freshness,
         "ddrnet_min_source_interval_ms": args.ddrnet_min_source_interval_ms,
     }
     children, logs, results, samples = [], [], [], []
@@ -270,6 +271,7 @@ def run(args):
     mailbox = Mailbox(capacity=2)
     source_report = {}
     producer = monitor = graph = gpu_stage = ddr_backend = None
+    cpu_bundle = None
 
     def child(name, command, env=None):
         log = (output / (name + ".log")).open("wb")
@@ -410,9 +412,7 @@ def run(args):
                         tensor = np.frombuffer(raw, "<f4").reshape(1, 3, 512, 512)
                     else:
                         tensor = native_preprocess(bundle["image"])
-                        preprocess_result = {
-                            "preprocess_ms": (time.monotonic_ns() - begin) / 1e6
-                        }
+                        preprocess_result = {"preprocess_ms": (time.monotonic_ns() - begin) / 1e6}
                     preprocessed = time.monotonic_ns()
                     mask = ddr_backend.infer(tensor)
                     inferred = time.monotonic_ns()
@@ -436,9 +436,7 @@ def run(args):
                 ddr_result = {
                     "component_ms": 0.0,
                     "forward_ms": 0.0,
-                    "stages_ms": {
-                        key: 0.0 for key in ddr_cache["result"]["stages_ms"]
-                    },
+                    "stages_ms": {key: 0.0 for key in ddr_cache["result"]["stages_ms"]},
                 }
                 mask = ddr_cache["mask"]
             ddr_done = time.monotonic_ns()
@@ -470,10 +468,12 @@ def run(args):
                     if item is None:
                         break
                     bundle, computed = item
+                    cpu_bundle = bundle
                 else:
                     bundle = mailbox.take()
                     if bundle is None:
                         break
+                    cpu_bundle = bundle
                     computed = compute_gpu(bundle)
                 begin, ddr_done, gpu_done, ddr_result, mask, proposals, ddr_layer = computed
                 cpu_started = time.monotonic_ns()
@@ -483,10 +483,17 @@ def run(args):
                     proposals=proposals,
                     detector_ms=(gpu_done - ddr_done) / 1e6,
                 )
+                raw_costmap_sha256 = hashlib.sha256(
+                    json.dumps(scene["costmap_states"], separators=(",", ":")).encode()
+                ).hexdigest()
                 freshness_started = time.monotonic_ns()
                 prepare_publication(
-                    scene, bundle, ddr_layer, epoch_id=args.run_id,
+                    scene,
+                    bundle,
+                    ddr_layer,
+                    epoch_id=args.run_id,
                     now_ns=freshness_started,
+                    mode=args.costmap_freshness,
                 )
                 scene.update(
                     sequence=bundle["sequence"],
@@ -504,8 +511,9 @@ def run(args):
                     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)
+                view, checked = assess_receipt(
+                    received, freshness, bundle=bundle, now_ns=checked_at
+                )
                 finished = time.monotonic_ns()
                 timing = {
                     **scene["timing_ms"],
@@ -541,11 +549,17 @@ def run(args):
                     "policy_counts": view["policy_counts"],
                     "freshness_at_receipt": checked.to_dict(),
                     "policy_counts_at_publication": scene["policy_counts"],
+                    "cell_assessment_at_receipt": view.get("cell_assessment"),
+                    "raw_costmap_states_sha256": raw_costmap_sha256,
+                    "effective_costmap_states_sha256": hashlib.sha256(
+                        json.dumps(view["costmap_states"], separators=(",", ":")).encode()
+                    ).hexdigest(),
                     "scene_bytes": len(encoded),
                     "scene_sha256": hashlib.sha256(encoded).hexdigest(),
                 }
                 results.append(result)
                 mailbox.release(bundle)
+                cpu_bundle = None
                 sink.write(encoded + b"\n")
                 if len(results) == 1:
                     report["reads_at_first_result"] = {
@@ -560,16 +574,14 @@ def run(args):
                                 "completed": len(results),
                                 "last_sequence": bundle["sequence"],
                                 "last_age_ms": timing["source_due_to_receiver_ms"],
-                                "drops": len(mailbox.dropped),
+                                "drops": mailbox.dropped_count,
                             }
                         ),
                         flush=True,
                     )
         if mailbox.error:
             raise RuntimeError(mailbox.error)
-        report["triton_statistics_delta"] = stats_delta(
-            stats_before, read_stats(triton_models)
-        )
+        report["triton_statistics_delta"] = stats_delta(stats_before, read_stats(triton_models))
         report["execution_complete"] = True
     except Exception:
         report["execution_complete"] = False
@@ -595,6 +607,10 @@ def run(args):
         if monitor:
             monitor.join(timeout=2)
         report["gpu_stage_stopped"] = gpu_stage.close() if gpu_stage else True
+        mailbox.cancel()
+        if cpu_bundle is not None:
+            mailbox.release(cpu_bundle, discard_reason="processing-failed")
+        report["input_payloads_released"] = mailbox.quiescent
         if graph:
             graph.backend.close()
         if ddr_backend:
@@ -633,16 +649,14 @@ def run(args):
     report["accounting"] = {
         "released": camera_count,
         "completed": len(results),
-        "dropped": len(mailbox.dropped),
-        "unaccounted": camera_count - len(results) - len(mailbox.dropped),
+        "dropped": mailbox.dropped_count,
+        "unaccounted": camera_count - len(results) - mailbox.dropped_count,
     }
     ddrnet_executed = sum(r.get("ddrnet_state") == "current" for r in results)
     report["model_cadence"] = {
         "ddrnet_executed": ddrnet_executed,
         "ddrnet_reused": len(results) - ddrnet_executed,
-        "ddrnet_source_age_ms": distribution(
-            [r["ddrnet_source_age_ms"] for r in results]
-        ),
+        "ddrnet_source_age_ms": distribution([r["ddrnet_source_age_ms"] for r in results]),
         "detector_executed": len(results),
     }
     latency = report["distributions_ms"].get("source_due_to_receiver_ms")
@@ -669,7 +683,7 @@ def run(args):
     report["gates"] = {
         "execution": report["execution_complete"],
         "zero_drops_complete_accounting": camera_count == args.frames == len(results)
-        and not mailbox.dropped,
+        and mailbox.dropped_count == 0,
         "worker_p95_p99_125ms": bool(latency and latency["p95"] <= 125 and latency["p99"] <= 125),
         "source_release_lag_25ms": bool(
             source_report.get("release_lag_ms") and source_report["release_lag_ms"]["max"] <= 25
@@ -679,7 +693,8 @@ def run(args):
         "warmup_120s": report.get("warmup_ms", math.inf) <= 120000,
         "stop_5s": report["stop_ms"] <= 5000
         and report["children_stopped"]
-        and report["gpu_stage_stopped"],
+        and report["gpu_stage_stopped"]
+        and report["input_payloads_released"],
         "vram_22000MiB": bool(samples)
         and max(s.get("gpu_used_mib", 99999) for s in samples) <= 22000,
         "container_memory_8192MiB": bool(samples)
@@ -735,6 +750,9 @@ if __name__ == "__main__":
         "--ddrnet-preprocess", choices=("pinned-child", "native-numpy"), default="pinned-child"
     )
     parser.add_argument("--schedule", choices=("serial", "overlap-cpu"), default="serial")
+    parser.add_argument(
+        "--costmap-freshness", choices=("whole-scene", "per-cell"), default="per-cell"
+    )
     parser.add_argument("--telemetry-mode", choices=("nvml", "none"), default="nvml")
     parser.add_argument("--telemetry-device-state", action="store_true")
     parser.add_argument("--triton-verbose", action="store_true")

src/k1link/perception/costmap_freshness.py

@@ -0,0 +1,140 @@
+"""Expiry of individual TGS cells, independent of models and transport.
+
+Uses the existing TGS states 0=unobserved, 1=ground, 2=occupied, 3=rejected
+and advisory actions 0=allow-candidate, 1=high-cost, 2=no-go. Expiry can only
+remove permission. Retained occupied cells keep their prohibition. This is not
+vehicle clearance, spatial reprojection validation, or motor authority.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Final
+
+from .realtime_contract import RealtimeContractError, _integer, _number
+from .realtime_scene import _read_wire_integer, _wire_integer
+
+CELL_FRESHNESS_SCHEMA: Final = "missioncore.costmap-cell-freshness/v1"
+MAXIMUM_CELLS: Final = 8192
+
+
+@dataclass(frozen=True, slots=True)
+class CellAssessment:
+    states: tuple[int, ...]
+    actions: tuple[int, ...]
+    expired_cells: int
+    missing_support_cells: int
+    suppressed_permissions: int
+    oldest_permissive_source_time_ns: int
+
+
+@dataclass(frozen=True, slots=True)
+class CostmapCellEvidence:
+    """Index-aligned with one immutable grid/state/material payload in a scene.
+
+    The enclosing scene binds epoch/clock/sequence and hashes this descriptor
+    together with the grid outputs. No timestamp is manufactured for unseen
+    cells; no coordinate transformation or observation is generated here.
+    """
+
+    observed_source_time_ns: tuple[int | None, ...]
+
+    def __post_init__(self) -> None:
+        if (
+            not isinstance(self.observed_source_time_ns, tuple)
+            or not 1 <= len(self.observed_source_time_ns) <= MAXIMUM_CELLS
+        ):
+            raise RealtimeContractError("costmap cell count outside bound")
+        for stamp in self.observed_source_time_ns:
+            if stamp is not None:
+                _wire_integer(stamp)
+
+    def to_dict(self) -> dict[str, object]:
+        return {
+            "schema_version": CELL_FRESHNESS_SCHEMA,
+            "observed_source_time_ns": [
+                None if stamp is None else _wire_integer(stamp)
+                for stamp in self.observed_source_time_ns
+            ],
+        }
+
+    @classmethod
+    def from_dict(cls, value: object) -> CostmapCellEvidence:
+        if (
+            not isinstance(value, dict)
+            or set(value) != {"schema_version", "observed_source_time_ns"}
+            or value["schema_version"] != CELL_FRESHNESS_SCHEMA
+            or not isinstance(value["observed_source_time_ns"], list)
+            or not 1 <= len(value["observed_source_time_ns"]) <= MAXIMUM_CELLS
+        ):
+            raise RealtimeContractError("costmap cell evidence schema or bound changed")
+        return cls(
+            tuple(
+                None if stamp is None else _read_wire_integer(stamp)
+                for stamp in value["observed_source_time_ns"]
+            )
+        )
+
+    def assess(
+        self,
+        *,
+        states: tuple[int, ...],
+        actions: tuple[int, ...],
+        source_time_ns: int,
+        observed_source_time_ns: int,
+        maximum_age_ms: float,
+        clock_uncertainty_ms: float,
+    ) -> CellAssessment:
+        """Produce a new view; repeated aging can never restore permissions.
+
+        The same mapped source clock as SceneFreshness must be used. Mandatory
+        segmentation/pose/geometry freshness is still checked for the WHOLE scene
+        separately. Passing individual cells cannot override missing inputs.
+        """
+        _integer(source_time_ns, "source_time_ns")
+        _integer(observed_source_time_ns, "observed_source_time_ns")
+        _number(maximum_age_ms, "maximum_age_ms", minimum=0.001)
+        _number(clock_uncertainty_ms, "clock_uncertainty_ms")
+        if observed_source_time_ns < source_time_ns:
+            raise RealtimeContractError("costmap assessment precedes source observation")
+        if (
+            not isinstance(states, tuple)
+            or not isinstance(actions, tuple)
+            or len(states) != len(actions)
+            or len(states) != len(self.observed_source_time_ns)
+        ):
+            raise RealtimeContractError("costmap arrays must share one bounded grid")
+        output_states, output_actions, permissive_stamps = [], [], []
+        expired = missing = suppressed = 0
+        for state, action, stamp in zip(states, actions, self.observed_source_time_ns, strict=True):
+            if type(state) is not int or not 0 <= state <= 3:
+                raise RealtimeContractError("invalid TGS state")
+            if type(action) is not int or not 0 <= action <= 2:
+                raise RealtimeContractError("invalid advisory action")
+            if state != 1 and action != 2:
+                raise RealtimeContractError("non-ground cell attempted terrain permission")
+            if stamp is not None and stamp > source_time_ns:
+                raise RealtimeContractError("future cell observation")
+            too_old = stamp is not None and (
+                (observed_source_time_ns - stamp) / 1e6 + clock_uncertainty_ms > maximum_age_ms
+            )
+            # Existing occupied/unknown/rejected cells never become free. Only
+            # positive ground evidence requires a current supporting observation.
+            if state == 1 and (stamp is None or too_old):
+                expired += int(too_old)
+                missing += int(stamp is None)
+                suppressed += int(action != 2)
+                state, action = 3, 2
+            if action != 2:
+                assert stamp is not None  # validated positive ground support above
+                permissive_stamps.append(stamp)
+            output_states.append(state)
+            output_actions.append(action)
+        return CellAssessment(
+            tuple(output_states),
+            tuple(output_actions),
+            expired,
+            missing,
+            suppressed,
+            min(permissive_stamps, default=source_time_ns),
+        )

src/k1link/perception/realtime_scene.py

@@ -79,6 +79,13 @@ def derive_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")
+    _integer(source_sequence, "source_sequence")
+    _integer(source_time_ns, "source_time_ns")
+    _integer(oldest_required_input_time_ns, "oldest_required_input_time_ns")
+    if oldest_required_input_time_ns > source_time_ns or any(
+        x.source_time_ns is not None and x.source_time_ns > source_time_ns for x in inputs
+    ):
+        raise RealtimeContractError("derived layer references a future input")
     if any(x.state == "unavailable" for x in inputs):
         return LayerEvidence(layer, epoch_id, None, None, "unavailable", None)
     stamps = [oldest_required_input_time_ns]

tests/test_perception_costmap_freshness.py

@@ -0,0 +1,76 @@
+"""Bounded synthetic per-cell freshness tests, with no model/source dependency."""
+
+import json
+
+import pytest
+
+from k1link.perception.costmap_freshness import CostmapCellEvidence
+from k1link.perception.realtime_contract import RealtimeContractError
+
+NOW = 9_007_199_254_740_993
+
+
+def check(stamps, states, actions, *, elapsed_ms=0, uncertainty_ms=0):
+    return CostmapCellEvidence(tuple(stamps)).assess(
+        states=tuple(states),
+        actions=tuple(actions),
+        source_time_ns=NOW,
+        observed_source_time_ns=NOW + elapsed_ms * 1_000_000,
+        maximum_age_ms=250,
+        clock_uncertainty_ms=uncertainty_ms,
+    )
+
+
+def test_expiry_is_addressed_and_cannot_clear_occupied_or_unknown_cells():
+    stamps = (NOW - 200_000_000, NOW, NOW - 999_000_000, None, None)
+    result = check(stamps, [1, 1, 2, 0, 1], [0, 1, 2, 2, 0], elapsed_ms=51)
+    assert result.states == (3, 1, 2, 0, 3)
+    assert result.actions == (2, 1, 2, 2, 2)
+    assert result.expired_cells == 1 and result.missing_support_cells == 1
+    assert result.suppressed_permissions == 2
+    assert result.oldest_permissive_source_time_ns == NOW
+    # Even an earlier assessment cannot restore a suppressed permission.
+    repeated = check(stamps, result.states, result.actions)
+    assert repeated.states == result.states and repeated.actions == result.actions
+
+
+def test_ttl_boundary_includes_mapping_uncertainty_without_refreshing_support():
+    stamp = NOW - 200_000_000
+    assert check([stamp], [1], [0], elapsed_ms=45, uncertainty_ms=5).actions == (0,)
+    assert check([stamp], [1], [0], elapsed_ms=46, uncertainty_ms=5).actions == (2,)
+    assert check([stamp], [1], [0]).oldest_permissive_source_time_ns == stamp
+
+
+def test_cell_wire_is_exact_int64_and_has_a_bounded_shape():
+    evidence = CostmapCellEvidence((NOW, None))
+    assert evidence.to_dict()["observed_source_time_ns"] == [str(NOW), None]
+    assert CostmapCellEvidence.from_dict(json.loads(json.dumps(evidence.to_dict()))) == evidence
+    with pytest.raises(RealtimeContractError):
+        CostmapCellEvidence((NOW,) * 8193)
+
+
+@pytest.mark.parametrize("stamp", [NOW, "01", "-1", str(2**63), True, {}, "NaN"])
+def test_invalid_wire_timestamp_is_rejected(stamp):
+    value = CostmapCellEvidence((NOW,)).to_dict()
+    value["observed_source_time_ns"] = [stamp]
+    with pytest.raises(RealtimeContractError):
+        CostmapCellEvidence.from_dict(value)
+
+
+@pytest.mark.parametrize(
+    "stamps,states,actions,elapsed,uncertainty",
+    [
+        ([NOW + 1], [1], [0], 0, 0),
+        ([NOW], [1], [0], -1, 0),
+        ([NOW], [1], [0], 0, float("nan")),
+        ([NOW], [1], [0], 0, -1),
+        ([NOW], [True], [0], 0, 0),
+        ([NOW], [1], [True], 0, 0),
+        ([NOW], [2], [0], 0, 0),
+        ([None], [0], [1], 0, 0),
+        ([NOW], [], [], 0, 0),
+    ],
+)
+def test_invalid_evidence_never_grants_permission(stamps, states, actions, elapsed, uncertainty):
+    with pytest.raises(RealtimeContractError):
+        check(stamps, states, actions, elapsed_ms=elapsed, uncertainty_ms=uncertainty)

tests/test_perception_realtime_scene.py

@@ -8,6 +8,7 @@ from pathlib import Path
 
 import pytest
 
+from k1link.perception.costmap_freshness import CostmapCellEvidence
 from k1link.perception.realtime_contract import (
     REQUIRED_LAYERS,
     LayerEvidence,
@@ -296,3 +297,112 @@ def test_pilot_rejects_wrong_run_sequence_and_control_authority(pilot):
     changed["commands_enabled"] = True
     with pytest.raises(ValueError, match="authority"):
         pilot.validate_receipt(changed, bundle, epoch_id="pilot")
+
+
+@pytest.mark.parametrize("own_time", [NOW + 1, True])
+def test_derivation_cannot_hide_invalid_own_time_behind_an_older_parent(own_time):
+    value = scene(held_ms=100)
+    with pytest.raises(RealtimeContractError):
+        derive_layer(
+            "costmap",
+            epoch_id="epoch",
+            source_sequence=2,
+            source_time_ns=NOW,
+            payload_sha256="a" * 64,
+            inputs=tuple(value.layers[i] for i in (0, 2, 3)),
+            oldest_required_input_time_ns=own_time,
+        )
+
+
+def cell_input():
+    payload, bundle, ddr = pilot_input()
+    payload["costmap_states"] = [1, 1]
+    payload["costmap_material"] = [1, 1]
+    payload["policy_actions"] = [0, 0]
+    payload["policy_counts"] = {"ALLOW_candidate": 2, "HIGH_COST": 0, "NO_GO": 0}
+    payload["costmap_cell_evidence"] = CostmapCellEvidence((NOW - 200_000_000, NOW)).to_dict()
+    payload["tgs_counts"]["oldest_permissive_cell_source_ns"] = NOW - 200_000_000
+    return payload, bundle, ddr
+
+
+def test_consumer_expires_only_old_cell_and_rehashes_view_without_mutating_wire(pilot):
+    payload, bundle, ddr = cell_input()
+    pilot.prepare_publication(
+        payload,
+        bundle,
+        ddr,
+        epoch_id="pilot",
+        now_ns=bundle["due_ns"] + 10_000_000,
+        mode="per-cell",
+    )
+    original = copy.deepcopy(payload)
+    freshness = pilot.validate_receipt(payload, bundle, epoch_id="pilot")
+    view, checked = pilot.assess_receipt(
+        payload,
+        freshness,
+        bundle=bundle,
+        now_ns=bundle["due_ns"] + 60_000_000,
+    )
+    assert checked.fresh_complete
+    assert view["costmap_states"] == [3, 1] and view["policy_actions"] == [2, 0]
+    assert view["cell_assessment"]["expired_ground_cells"] == 1
+    assert checked.layers[4].oldest_required_input_time_ns == NOW - 20_000_000
+    assert payload == original
+    derived = pilot.validate_receipt(view, bundle, epoch_id="pilot")
+    assert derived.source_time_ns == freshness.source_time_ns
+    aged, expired = pilot.assess_receipt(
+        view,
+        derived,
+        bundle=bundle,
+        now_ns=bundle["due_ns"] + 260_000_000,
+    )
+    assert not expired.fresh_complete and aged["policy_actions"] == [2, 2]
+    pilot.validate_receipt(aged, bundle, epoch_id="pilot")
+    with pytest.raises(ValueError, match="backwards"):
+        pilot.assess_receipt(view, derived, bundle=bundle, now_ns=bundle["due_ns"])
+
+
+@pytest.mark.parametrize("missing,held_ms", [(True, 0), (False, 220)])
+def test_cell_freshness_cannot_override_missing_lidar_or_stale_segmentation(
+    pilot, missing, held_ms
+):
+    payload, bundle, ddr = cell_input()
+    bundle["available"] = not missing
+    if held_ms:
+        ddr.update(state="held", source_sequence=1, source_host_monotonic_ns=NOW - 220_000_000)
+    pilot.prepare_publication(
+        payload,
+        bundle,
+        ddr,
+        epoch_id="pilot",
+        now_ns=bundle["due_ns"],
+        mode="per-cell",
+    )
+    fresh = pilot.validate_receipt(payload, bundle, epoch_id="pilot")
+    view, checked = pilot.assess_receipt(
+        payload,
+        fresh,
+        bundle=bundle,
+        now_ns=bundle["due_ns"] + 60_000_000,
+    )
+    assert not checked.fresh_complete and view["policy_actions"] == [2, 2]
+    pilot.validate_receipt(view, bundle, epoch_id="pilot")
+
+
+def test_cell_timestamp_and_mode_are_bound_to_costmap_hash(pilot):
+    payload, bundle, ddr = cell_input()
+    pilot.prepare_publication(
+        payload,
+        bundle,
+        ddr,
+        epoch_id="pilot",
+        now_ns=bundle["due_ns"],
+        mode="per-cell",
+    )
+    changed = copy.deepcopy(payload)
+    changed["costmap_cell_evidence"]["observed_source_time_ns"][0] = str(NOW)
+    with pytest.raises(ValueError, match="digest"):
+        pilot.validate_receipt(changed, bundle, epoch_id="pilot")
+    payload["costmap_freshness_mode"] = "whole-scene"
+    with pytest.raises(ValueError, match="digest"):
+        pilot.validate_receipt(payload, bundle, epoch_id="pilot")