diff --git a/experiments/perception/worker/streaming_profile_stage1/pilot_freshness.py b/experiments/perception/worker/streaming_profile_stage1/pilot_freshness.py index 4a68474..9edd85a 100644 --- a/experiments/perception/worker/streaming_profile_stage1/pilot_freshness.py +++ b/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 diff --git a/experiments/perception/worker/streaming_profile_stage1/pilot_graph.py b/experiments/perception/worker/streaming_profile_stage1/pilot_graph.py index d7b6243..bfacbb5 100644 --- a/experiments/perception/worker/streaming_profile_stage1/pilot_graph.py +++ b/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 {}), }, diff --git a/experiments/perception/worker/streaming_profile_stage1/run_joint_pilot.py b/experiments/perception/worker/streaming_profile_stage1/run_joint_pilot.py index f329a80..918623c 100644 --- a/experiments/perception/worker/streaming_profile_stage1/run_joint_pilot.py +++ b/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, " 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), + ) diff --git a/src/k1link/perception/realtime_scene.py b/src/k1link/perception/realtime_scene.py index c40f12d..5b0257f 100644 --- a/src/k1link/perception/realtime_scene.py +++ b/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] diff --git a/tests/test_perception_costmap_freshness.py b/tests/test_perception_costmap_freshness.py new file mode 100644 index 0000000..c0fb29a --- /dev/null +++ b/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) diff --git a/tests/test_perception_realtime_scene.py b/tests/test_perception_realtime_scene.py index 7e43b17..04048ec 100644 --- a/tests/test_perception_realtime_scene.py +++ b/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")