451 lines
17 KiB
Python
451 lines
17 KiB
Python
"""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")
|