"""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.costmap_freshness import CostmapCellEvidence 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") @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"]) def test_cross_host_uncertainty_expires_per_cell_and_survives_publication(pilot): payload, bundle, ddr = cell_input() # Without uncertainty old ground would be 249 ms old; upper bound is253 ms. bundle["clock_observer"] = lambda _: (NOW + 49_000_000, 4_000_000) pilot.prepare_publication(payload, bundle, ddr, epoch_id="pilot", now_ns=123, mode="per-cell") assert payload["cell_assessment"]["expired_ground_cells"] == 1 assert payload["policy_actions"] == [2, 0] assert payload["freshness_at_publication"]["clock_uncertainty_ms"] == 4 fresh = pilot.validate_receipt(payload, bundle, epoch_id="pilot") # An independently measured later receipt can have a slightly smaller # midpoint while still overlapping the published interval. local_bundle = {k: v for k, v in bundle.items() if k != "clock_observer"} view, checked = pilot.assess_receipt( payload, fresh, bundle=local_bundle, now_ns=bundle["due_ns"] + 48_000_000 ) assert checked.clock_uncertainty_ms == 0 and view["policy_actions"] == [2, 0] with pytest.raises(ValueError, match="backwards"): pilot.assess_receipt( payload, fresh, bundle=local_bundle, now_ns=bundle["due_ns"] + 44_000_000 ) def test_one_clock_snapshot_per_publication_and_receipt_boundary(pilot): payload, bundle, ddr = cell_input() calls = [] def observe(now): assert now not in calls # A later refresh may no longer describe this instant. calls.append(now) return NOW + now, 4_000_000 bundle["clock_observer"] = observe pilot.prepare_publication( payload, bundle, ddr, epoch_id="pilot", now_ns=260_000_000, mode="per-cell" ) assert payload["policy_actions"] == [2, 2] # Exercises suppression/reassessment too. fresh = pilot.validate_receipt(payload, bundle, epoch_id="pilot") view, checked = pilot.assess_receipt(payload, fresh, bundle=bundle, now_ns=270_000_000) assert calls == [260_000_000, 270_000_000] assert not checked.fresh_complete and view["policy_actions"] == [2, 2] @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")