test(perception): verify reuse of existing raw live ingress

tests/test_perception_streaming_ingress.py

@@ -8,7 +8,7 @@ from dataclasses import replace
 
 import pytest
 
-from k1link.compute.live_perception import LiveIngressEvent
+from k1link.compute.live_perception import LiveIngressEvent, LivePerceptionIngress
 from k1link.perception import streaming_wire as wire
 from k1link.perception.graph_contracts import GraphState
 from k1link.perception.realtime_contract import StreamStart
@@ -311,3 +311,28 @@ def test_sender_timeout_closes_transport_instead_of_building_backlog():
     finally:
         left.close()
         right.close()
+
+
+def test_existing_raw_first_ingress_events_feed_new_writer_without_wire_v1_changes(tmp_path):
+    source = LivePerceptionIngress()
+    source.begin_session("capture-1")
+    source.open_consumer("stream-adapter")
+    assert source.take_next("stream-adapter", timeout=0).modality == "control"
+    assert source.publish(
+        modality="lidar",
+        source_id="existing-raw-topic",
+        source_sequence=12,
+        captured_at_epoch_ns=1799999999123456789,
+        received_monotonic_ns=9007199254740993,
+        payload=b"unchanged raw payload",
+    )
+    raw = source.take_next("stream-adapter", timeout=0)
+    legacy = raw.wire_bytes()
+    with harness(tmp_path) as (sock, receiver, _, received, _):
+        sender = StreamingSender(sock, start(), "capture-1", 1, lambda: None)
+        sender.send(raw)
+        sender.end()
+        assert receiver.join() and receiver.terminal == "end"
+        assert received == [raw] and raw.wire_bytes() == legacy
+    source.close_consumer("stream-adapter")
+    source.close()