feat(perception): seal M4.8S reference graph replay

tests/test_perception_graph.py

@@ -4,7 +4,7 @@ import json
 import subprocess
 import sys
 import threading
-from collections.abc import Iterator
+from collections.abc import Callable, Iterator
 from dataclasses import replace
 from pathlib import Path
 from queue import Queue
@@ -39,6 +39,7 @@ from k1link.perception.contracts import (
     TimestampBundle,
 )
 from k1link.perception.graph import (
+    DeliveredFrame,
     GraphExecutionError,
     GraphRunMode,
     GraphRunResult,
@@ -460,6 +461,8 @@ def _graph_v2(
     capacity: int = 8,
     run_mode: GraphRunMode = GraphRunMode.LOSSLESS_REPLAY,
     terminal_timeout_ns: int = 500_000_000,
+    delivery_observer: Callable[[DeliveredFrame, int], None] | None = None,
+    delivery_evidence_observer: Callable[..., None] | None = None,
 ) -> ReferencePerceptionGraphV2:
     return ReferencePerceptionGraphV2(
         config=_config_v2(capacity, terminal_timeout_ns),
@@ -471,6 +474,8 @@ def _graph_v2(
         rolling=_Rolling(),
         threat=_Threat(),
         run_mode=run_mode,
+        delivery_observer=delivery_observer,
+        delivery_evidence_observer=delivery_evidence_observer,
         clock_ns=lambda: 10_000,
     )
 
@@ -527,6 +532,50 @@ def test_reference_graph_v2_publishes_current_and_retained_occupancy() -> None:
     }
 
 
+def test_reference_graph_v2_observes_final_delivery_completion_age() -> None:
+    observed: list[tuple[int, int, int]] = []
+
+    def observe(delivery: DeliveredFrame, completion_age_ns: int) -> None:
+        observed.append(
+            (
+                delivery.sequence,
+                delivery.obstacle_map.output_age_ns,
+                completion_age_ns,
+            )
+        )
+
+    result = _graph_v2(
+        _Source((_packet(0, source_age_ns=123),)),
+        delivery_observer=observe,
+    ).run()
+
+    assert result.state is GraphState.STOPPED
+    assert observed == [(0, 123, 123)]
+
+
+def test_reference_graph_v2_observes_exact_delivery_evidence_inputs() -> None:
+    observed: list[tuple[int, int, tuple[str, ...], frozenset[str], int]] = []
+
+    def observe(delivery, packet, proposals, associated_ids, completion_age_ns) -> None:
+        observed.append(
+            (
+                delivery.sequence,
+                packet.envelope.sequence,
+                tuple(proposal.proposal_id for proposal in proposals),
+                associated_ids,
+                completion_age_ns,
+            )
+        )
+
+    result = _graph_v2(
+        _Source((_packet(0, source_age_ns=123),)),
+        delivery_evidence_observer=observe,
+    ).run()
+
+    assert result.state is GraphState.STOPPED
+    assert observed == [(0, 0, ("proposal-0",), frozenset({"proposal-0"}), 123)]
+
+
 def test_reference_graph_v2_lossless_mode_applies_bounded_backpressure() -> None:
     release = Event()
 
@@ -784,6 +833,32 @@ def test_decoded_recorded_source_attaches_images_without_detector_logic(tmp_path
     assert int(packets[1].image_payload[0, 0, 0]) == 7
 
 
+def test_decoded_recorded_source_primes_decoder_before_source_clock() -> None:
+    events: list[str] = []
+
+    class Source(RecordedRavnoves00Source):
+        def __init__(self) -> None:
+            pass
+
+        def packets(self, stop_event: Event) -> Iterator[SourcePacket]:
+            events.append("source-clock-started")
+            if not stop_event.is_set():
+                yield _packet(0)
+
+    class Decoder:
+        def frames(self, stop_event: Event) -> Iterator[np.ndarray]:
+            events.append("decoder-primed")
+            if not stop_event.is_set():
+                yield np.zeros((600, 800, 3), dtype=np.uint8)
+
+    packets = list(
+        DecodedRecordedSource(source=Source(), decoder=Decoder()).packets(Event())
+    )
+
+    assert len(packets) == 1
+    assert events == ["decoder-primed", "source-clock-started"]
+
+
 def test_camera_only_path_never_invents_metric_occupancy_or_free_space() -> None:
     result = _graph(_Source((_packet(0, lidar=False),))).run()
     delivery = result.deliveries[0]