feat(perception): connect binary ingress to the supervised full graph

experiments/perception/worker/streaming_profile_stage1/pilot_binary_bridge.py

@@ -0,0 +1,172 @@
+"""Binary input → existing full-graph bundles; source paths never enter this adapter."""
+
+import hashlib
+import socket
+import threading
+import time
+from collections import Counter
+
+import numpy as np
+
+from k1link.perception.streaming_ingress import StreamingIngress
+from k1link.perception.streaming_sensors import (
+    CausalSensorWindow,
+    increment_identity,
+    milliseconds,
+    normalized_sensor,
+)
+
+
+class BinaryGraphBridge:
+    def __init__(self, runtime, decoder, source, report):
+        self.runtime, self.decoder, self.source, self.report = runtime, decoder, source, report
+        self.source_zero, self.wall_zero = source.source_zero, source.wall_zero
+        self.window = CausalSensorWindow(runtime.mailbox)
+        self.accepted = []  # Bounded pilot diagnostics, not a runtime history queue.
+        self.binding_reasons = Counter()
+        self.preroll = 0
+        self.bgr_hashes = []
+        self.left = right = None
+        try:
+            self.left, right = socket.socketpair()
+            self.receiver = StreamingIngress(
+                right, runtime, "recorded-acquisition", 1, self.consume, self.notice
+            )
+            self.producer = threading.Thread(
+                target=source.run, args=(self.left,), daemon=True, name="binary-recorded-source"
+            )
+            runtime.track_thread(self.producer)
+        except BaseException:
+            # No threads have started. A failed registration/lease must not
+            # strand the preallocated cache or either end of the socket.
+            for connection in (self.left, right):
+                if connection is not None:
+                    connection.close()
+            self.window.close()
+            raise
+
+    def start(self):
+        self.receiver.start()
+        self.producer.start()
+
+    def notice(self, value):
+        # A gap is not a new codec epoch. This profile requires a new run/lease
+        # and fresh decoder state; never continue predictive decoding across it.
+        raise ValueError("binary full profile requires restart after an explicit source gap")
+
+    def consume(self, event):
+        self.runtime.check_current(self.runtime.start)
+        stamp = event.received_monotonic_ns
+        self.window.advance(stamp)
+        if event.modality == "camera-init":
+            if event.source_id != "sensor.camera.right":
+                raise ValueError("camera source mapping changed")
+            self.decoder.configure(event.payload)
+            return
+        if event.modality in ("lidar", "pose"):
+            expected = {
+                "lidar": "normalized-map-point-increments",
+                "pose": "normalized-map-from-lidar-pose",
+            }
+            if event.source_id != expected[event.modality]:
+                raise ValueError("normalized sensor mapping changed")
+            self.window.append(
+                normalized_sensor(event.modality, event.source_sequence, stamp, event.payload)
+            )
+            return
+        if event.modality != "camera-frame" or event.source_id != "sensor.camera.right":
+            raise ValueError("unsupported full-profile input")
+        # Freeze the causal cut BEFORE the decoder RPC. The synchronous ingress
+        # cannot consume future sensor messages while this callback is running.
+        binding = self.window.bind(stamp)
+        pose = self.window.pose
+        rolling = tuple(self.window.rolling)
+        fresh = binding.increments
+        count = sum(len(e.value[0]) for e in fresh)
+        rolling_count = sum(len(e.value[0]) for e in rolling)
+        size = 1440000 + count * 24 + rolling_count * 32 + 4096
+        reservation = self.runtime.mailbox.reserve_ingress(size)
+        raw = image = points = rolling_points = rolling_times = bundle = None
+        transferred = False
+        decode_started = time.monotonic_ns()
+        try:
+            raw = bytearray(1440000)
+            decoded = self.decoder.decode(event.payload, raw)
+            image = np.frombuffer(raw, "u1").reshape(600, 800, 3)
+            image.setflags(write=False)
+            if decoded["frame_index"] != event.source_sequence:
+                raise ValueError("camera and decoder frame identity diverged")
+            points = np.concatenate([e.value[0] for e in fresh]) if fresh else np.empty((0, 3))
+            rolling_points = (
+                np.concatenate([e.value[0] for e in rolling]) if rolling else np.empty((0, 3))
+            )
+            rolling_times = np.empty(rolling_count, np.int64)
+            offset = 0
+            for e in rolling:
+                length = len(e.value[0])
+                rolling_times[offset : offset + length] = e.time_ns
+                offset += length
+            due = self.wall_zero + stamp - self.source_zero
+            bundle = {
+                "sequence": event.source_sequence,
+                "time_ns": stamp,
+                "source_ns": stamp - self.source_zero,
+                "due_ns": due,
+                "utc_ns": event.captured_at_epoch_ns,
+                "image": image,
+                "points": points,
+                "rolling_points": rolling_points,
+                "rolling_times": rolling_times,
+                "pose": pose.value if pose else None,
+                "available": binding.available,
+                "binding_age_ms": milliseconds(binding.binding_age_ns),
+                "sensor_binding": binding.document(),
+                "lineage": {
+                    "camera_index_sequence": event.source_sequence + 1,
+                    "camera_host_monotonic_ns": stamp,
+                    "pose_sequence": pose.sequence if pose else None,
+                    "pose_host_monotonic_ns": pose.time_ns if pose else None,
+                    "point_increments": [increment_identity(e) for e in fresh],
+                    "pose_age_ms": milliseconds(binding.pose_age_ns),
+                    "newest_point_age_ms": milliseconds(binding.newest_point_age_ns),
+                    "oldest_point_age_ms": milliseconds(binding.oldest_point_age_ns),
+                },
+                "source_release_lag_ms": None,  # Recorded independently by the source, not guessed.
+                "decode_ms": decoded["decode_ms"],
+                "decode_rpc_ms": (time.monotonic_ns() - decode_started) / 1e6,
+                "enqueued_ns": time.monotonic_ns(),
+                "payload_bytes": size,
+            }
+            self.bgr_hashes.append(hashlib.sha256(image).hexdigest())
+            bundle["enqueued_ns"] = time.monotonic_ns()
+            transferred = self.runtime.admit_reserved(self.runtime.start, bundle, reservation)
+            self.accepted.append(event.source_sequence)
+            self.binding_reasons.update(binding.reasons)
+            self.preroll += sum(len(e.value[0]) for e in binding.history_only)
+            self.report["last_camera_due_ns"] = due
+            self.window.finish_camera(stamp)
+        finally:
+            raw = image = points = rolling_points = rolling_times = bundle = None
+            if not transferred:
+                reservation.release()
+
+    def close(self):
+        if self.producer.ident is not None:
+            self.producer.join(timeout=2)
+        else:
+            self.left.close()
+        if self.receiver.thread.ident is None:
+            self.receiver.connection.close()
+        joined = self.receiver.join(timeout=2) if self.receiver.thread.ident is not None else True
+        self.report.update(
+            binary_ingress=self.receiver.snapshot(),
+            decoder_bgr_sha256=self.bgr_hashes,
+            failed_camera_sequences=sorted(set(self.source.released) - set(self.accepted)),
+            sensor_binding_reasons=dict(self.binding_reasons),
+            preroll_history_only_points=self.preroll,
+        )
+        if joined and not self.producer.is_alive():
+            self.window.close()
+            self.decoder.close()
+            return True
+        return False

experiments/perception/worker/streaming_profile_stage1/pilot_binary_bridge_probe.py

@@ -0,0 +1,222 @@
+"""CPU-only decoded-bundle oracle before a full GPU graph run; bounded prefix."""
+
+import argparse
+import hashlib
+import json
+import os
+import subprocess
+import threading
+import time
+from datetime import UTC, datetime
+from pathlib import Path
+from types import SimpleNamespace
+
+from pilot_binary_bridge import BinaryGraphBridge
+from pilot_binary_source import RecordingSource
+
+from k1link.perception.realtime_contract import StreamStart
+from k1link.perception.streaming_decoder_client import StreamingDecoderClient
+from k1link.perception.streaming_lifecycle import StreamingLifecycle
+from k1link.perception.streaming_queue import StreamMailbox
+from k1link.perception.worker_lease import WorkerLeaseError
+
+
+def digest(bundle):
+    result = {
+        key: bundle[key]
+        for key in (
+            "sequence",
+            "time_ns",
+            "utc_ns",
+            "lineage",
+            "sensor_binding",
+            "available",
+            "binding_age_ms",
+        )
+    }
+    result["arrays"] = {
+        key: hashlib.sha256(bundle[key]).hexdigest()
+        for key in ("points", "rolling_points", "rolling_times")
+    }
+    result["pose"] = [a.tolist() for a in bundle["pose"]] if bundle["pose"] else None
+    return result
+
+
+def run(args):
+    root = Path(args.output)
+    root.mkdir()
+    report = {
+        "created_utc": datetime.now(UTC).isoformat(),
+        "started_monotonic_ns": time.monotonic_ns(),
+        "scope": "CPU-only bundle parity",
+        "synthetic_decoder_timeout": args.fault_decoder,
+    }
+    source_report = {}
+    identity = StreamStart(
+        run_id=root.name,
+        source_id="RAVNOVES00-prefix",
+        worker_id="worker-006",
+        epoch_id="epoch-1",
+        lease_generation=1,
+        profile_sha256=hashlib.sha256(Path("/out/candidate-profile.json").read_bytes()).hexdigest(),
+        image_sha256="664824aa25de1db178f177d67a81b01541a938b479812b9383ddbf03f6b59dbe",
+        effective_config_sha256=hashlib.sha256(b"cpu-bundle-parity-v1").hexdigest(),
+        calibration_sha256=hashlib.sha256(
+            b"no-calibration-or-model-scene-in-this-probe"
+        ).hexdigest(),
+        clock_domain_id="original-host-arrival-clock",
+        input_mode="recorded-source-paced",
+    )
+    runtime = StreamingLifecycle(
+        identity, Path("/tmp/cpu-bridge-lease"), StreamMailbox(), threading.Event()
+    )
+    heartbeat_stop = threading.Event()
+
+    def heartbeat():
+        while not heartbeat_stop.wait(0.25):
+            try:
+                runtime.renew(identity)
+            except WorkerLeaseError:
+                # Stop/lease failure is already terminal in the common runtime.
+                # Do not turn an expected asynchronous shutdown into thread noise.
+                return
+
+    renewer = threading.Thread(target=heartbeat, daemon=True)
+    renewer.start()
+    client = bridge = None
+    decoded = []
+    with (root / "decoder.log").open("wb") as log:
+        try:
+            command = ["python3", "-B", "/probe/pilot_fragment_decoder.py"]
+            if args.fault_decoder:
+                command = [
+                    "python3",
+                    "-B",
+                    "-c",
+                    "import sys,time; from pilot_ipc import receive,send; "
+                    "send(sys.stdout.buffer,{'ready':'fragment-h264','pyav':'18.0.0',"
+                    "'as_limit_mib':1024}); "
+                    "receive(sys.stdin.buffer); send(sys.stdout.buffer,{'initialized':True}); "
+                    "receive(sys.stdin.buffer); time.sleep(10)",
+                ]
+            child = runtime.spawn(
+                lambda: subprocess.Popen(
+                    command,
+                    stdin=subprocess.PIPE,
+                    stdout=subprocess.PIPE,
+                    stderr=log,
+                    start_new_session=True,
+                    env={**os.environ, "CUDA_VISIBLE_DEVICES": ""},
+                )
+            )
+            client = StreamingDecoderClient(runtime, child.stdout.fileno(), child.stdin.fileno())
+            runtime.ready()
+            source = RecordingSource(args, runtime, source_report)
+            bridge = BinaryGraphBridge(runtime, client, source, source_report)
+            bridge.start()
+            while (bundle := runtime.mailbox.take()) is not None:
+                decoded.append(digest(bundle))
+                runtime.mailbox.release(bundle)
+            if runtime.mailbox.error and not args.fault_decoder:
+                raise ValueError(runtime.mailbox.error)
+        finally:
+            heartbeat_stop.set()
+            renewer.join(timeout=1)
+            runtime.request_stop("completed" if len(decoded) == args.frames else "failed")
+            runtime.stop_children()
+            report["bridge_stopped"] = bridge.close() if bridge else True
+            if client:
+                client.close()
+            report["released"] = runtime.close()
+            report["runtime"] = runtime.snapshot()
+            report["source"] = source_report
+            report["peak_input_bytes"] = runtime.mailbox.peak_bytes
+            report["residual_input_bytes"] = runtime.mailbox.bytes
+            (root / "report.json").write_text(json.dumps(report, indent=2) + "\n")
+    if args.fault_decoder:
+        assert not decoded and report["released"] and not runtime.mailbox.bytes
+        assert "decoder RPC deadline" in source_report["binary_ingress"]["error"]
+        assert source_report["failed_camera_sequences"]
+        print(
+            json.dumps(
+                {
+                    "synthetic_decoder_timeout": True,
+                    "released": report["released"],
+                    "source_failed": source_report["failed_camera_sequences"],
+                    "input_bytes": runtime.mailbox.bytes,
+                }
+            )
+        )
+        return
+    # Independent oracle uses the old producer's exact assembly code AFTER the
+    # binary run. No prepared arrays feed the new decoder or graph. Zero camera
+    # bytes are used only here: image parity has its separate actual decoder test.
+    import run_joint_pilot as legacy
+
+    reference = []
+
+    class Sink:
+        error = None
+
+        def put(self, bundle):
+            reference.append(digest(bundle))
+            return True
+
+        def finish(self, error=None):
+            self.error = error
+
+    sink = Sink()
+    old_send, old_receive = legacy.send, legacy.receive
+    image = bytes(1440000)
+    legacy.send = lambda *args: None
+    legacy.receive = lambda *args: ({"decode_ms": 0}, image)
+    try:
+        legacy.produce(
+            args,
+            SimpleNamespace(stdin=None, stdout=None),
+            sink,
+            SimpleNamespace(is_set=lambda: False, wait=lambda _: False),
+            {},
+        )
+    finally:
+        legacy.send, legacy.receive = old_send, old_receive
+    assert sink.error is None, sink.error
+    report.update(
+        bundle_count=len(decoded),
+        oracle_count=len(reference),
+        exact_sensor_bundles=decoded == reference,
+        model_runs=0,
+        network_qualified=False,
+    )
+    (root / "decoded-bundles.json").write_text(json.dumps(decoded, indent=2) + "\n")
+    (root / "reference-bundles.json").write_text(json.dumps(reference, indent=2) + "\n")
+    (root / "report.json").write_text(json.dumps(report, indent=2) + "\n")
+    assert decoded == reference and len(decoded) == args.frames and report["released"]
+    print(
+        json.dumps(
+            {
+                k: report[k]
+                for k in (
+                    "bundle_count",
+                    "exact_sensor_bundles",
+                    "released",
+                    "peak_input_bytes",
+                    "residual_input_bytes",
+                )
+            }
+        )
+    )
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--output", required=True)
+    parser.add_argument("--frames", type=int, default=32)
+    parser.add_argument("--fault-decoder", action="store_true")
+    parser.add_argument("--camera-root", default="/camera")
+    parser.add_argument("--camera-index", default="/camera/index.jsonl")
+    parser.add_argument("--sensor-archive", default="/sensor-source.npz")
+    args = parser.parse_args()
+    if not 1 <= args.frames <= 128:
+        parser.error("CPU oracle is bounded to 128 cameras")
+    run(args)

experiments/perception/worker/streaming_profile_stage1/pilot_binary_source.py

@@ -0,0 +1,145 @@
+"""Recording adapter only: emits bounded camera/sensors at original 1x times.
+
+No model or decoder is called, and processing speed never sets source pacing.
+Source inventory/count belong here, never to the binary consumer or full graph.
+"""
+
+import hashlib
+import struct
+import time
+import traceback
+from collections import Counter
+from pathlib import Path
+
+from pilot_source import SensorArchive, camera_events, merged_events
+
+from k1link.compute.live_perception import LiveIngressEvent
+from k1link.perception.streaming_sender import StreamingSender
+
+
+def read_member(path, root, maximum):
+    resolved = path.resolve(strict=True)
+    if not resolved.is_relative_to(root.resolve(strict=True)):
+        raise ValueError("source member escaped camera root")
+    with resolved.open("rb") as stream:
+        raw = stream.read(maximum + 1)
+    if not 0 < len(raw) <= maximum:
+        raise ValueError("source member exceeds bound")
+    return raw
+
+
+class RecordingSource:
+    def __init__(self, args, runtime, report):
+        self.args, self.runtime, self.report = args, runtime, report
+        first = next(camera_events(args.camera_index, 1)).time_ns
+        self.source_zero = first - 500_000_000
+        self.wall_zero = time.monotonic_ns() + 50_000_000
+        self.released = []  # Diagnostic bounded prefix only, never runtime admission state.
+        report.update(
+            source_zero_ns=self.source_zero,
+            wall_zero_ns=self.wall_zero,
+            first_camera_source_ns=first,
+            source_clock_speed=1.0,
+            source_eof_required=False,
+            full_source_prepass=False,
+        )
+
+    def run(self, connection):
+        archive = None
+        sender = None
+        arrivals, skipped = Counter(), Counter()
+        lags = []
+        clock = self.runtime.stop_event
+        try:
+            archive = SensorArchive(Path(self.args.sensor_archive))
+            root = Path(self.args.camera_root)
+            sender = StreamingSender(
+                connection,
+                self.runtime.start,
+                "recorded-acquisition",
+                1,
+                lambda: self.runtime.check_current(self.runtime.start),
+            )
+            seq = 1
+            sender.send(
+                LiveIngressEvent(
+                    seq,
+                    "recorded-acquisition",
+                    1,
+                    "camera-init",
+                    "sensor.camera.right",
+                    0,
+                    0,
+                    self.source_zero,
+                    read_member(root / "init.mp4", root, 65536),
+                )
+            )
+            for event in merged_events(archive, self.args.camera_index, self.args.frames):
+                if event.time_ns < self.source_zero:
+                    skipped[event.channel] += 1
+                    continue
+                due = self.wall_zero + event.time_ns - self.source_zero
+                if clock.wait(max(0, (due - time.monotonic_ns()) / 1e9)):
+                    break
+                arrivals[event.channel] += 1
+                lags.append(max(0, time.monotonic_ns() - due) / 1e6)
+                utc = 0  # These normalized sensor rows have no UTC evidence.
+                if event.channel == "camera":
+                    self.released.append(event.sequence)
+                    row = event.value
+                    raw = read_member(root / row["path"], root, 1024 * 1024)
+                    if (
+                        len(raw) != row["length"]
+                        or hashlib.sha256(raw).hexdigest() != row["sha256"]
+                    ):
+                        raise ValueError("camera fragment integrity changed")
+                    modality, source_id, utc = (
+                        "camera-frame",
+                        "sensor.camera.right",
+                        row["host_epoch_ns"],
+                    )
+                elif event.channel == "points":
+                    xyz, intensity = event.value
+                    raw = struct.pack("<I", len(xyz)) + xyz.tobytes() + intensity.tobytes()
+                    modality, source_id = "lidar", "normalized-map-point-increments"
+                else:
+                    position, quaternion = event.value
+                    raw = position.tobytes() + quaternion.tobytes()
+                    modality, source_id = "pose", "normalized-map-from-lidar-pose"
+                seq += 1
+                sender.send(
+                    LiveIngressEvent(
+                        seq,
+                        "recorded-acquisition",
+                        1,
+                        modality,
+                        source_id,
+                        event.sequence,
+                        utc,
+                        event.time_ns,
+                        raw,
+                    )
+                )
+                del raw
+                if event.channel == "camera" and event.sequence == self.args.frames - 1:
+                    break  # Sensor streams may continue beyond this bounded camera probe.
+            if not clock.is_set():
+                self.report["end_sent_monotonic_ns"] = time.monotonic_ns()
+                sender.end()
+        except Exception:
+            self.report["source_error"] = traceback.format_exc()
+            self.runtime.mailbox.finish(self.report["source_error"])
+            self.runtime.request_stop("failed")
+        finally:
+            from run_joint_pilot import distribution
+
+            self.report.update(
+                arrivals=dict(arrivals),
+                skipped_prefix=dict(skipped),
+                release_lag_ms=distribution(lags),
+                incremental_reads=archive.counters() if archive else {},
+                window_end_monotonic_ns=time.monotonic_ns(),
+            )
+            if archive:
+                archive.close()
+            connection.close()

experiments/perception/worker/streaming_profile_stage1/pilot_fragment_decoder.py

@@ -0,0 +1,42 @@
+"""Supervised CPU decoder child. No source files, camera count, models or network."""
+
+import resource
+import sys
+import time
+
+from pilot_ipc import receive, send
+
+from k1link.perception.streaming_decoder import FragmentDecoder
+
+
+def main():
+    resource.setrlimit(resource.RLIMIT_AS, (1024**3, 1024**3))
+    decoder = FragmentDecoder()
+    send(sys.stdout.buffer, {"ready": "fragment-h264", "pyav": "18.0.0", "as_limit_mib": 1024})
+    try:
+        while True:
+            header, raw = receive(sys.stdin.buffer, max_payload=1024 * 1024)
+            begin = time.monotonic_ns()
+            if header == {"op": "init"}:
+                decoder.configure(raw)
+                send(sys.stdout.buffer, {"initialized": True})
+            elif header == {"op": "decode"}:
+                image = decoder.decode(raw)
+                send(
+                    sys.stdout.buffer,
+                    {
+                        "decode_ms": (time.monotonic_ns() - begin) / 1e6,
+                        "frame_index": decoder.frames - 1,
+                    },
+                    image.tobytes(),
+                )
+                del image
+            else:
+                raise ValueError("unsupported decoder operation")
+            del raw
+    finally:
+        decoder.close()
+
+
+if __name__ == "__main__":
+    main()

experiments/perception/worker/streaming_profile_stage1/pilot_ipc.py

@@ -20,13 +20,13 @@ def exact(stream, size):
     return bytes(chunks)
 
 
-def receive(stream):
+def receive(stream, max_payload=MAX_PAYLOAD):
     length = struct.unpack("<I", exact(stream, 4))[0]
     if not 1 <= length <= 65536:
         raise ValueError("IPC header exceeds budget")
     header = json.loads(exact(stream, length))
     size = header.pop("payload_bytes")
-    if type(size) is not int:
+    if type(size) is not int or not 0 <= size <= max_payload:
         raise ValueError("invalid IPC byte count")
     return header, exact(stream, size)
 

experiments/perception/worker/streaming_profile_stage1/pilot_sensor_binding.py

@@ -1,129 +1,23 @@
-"""Causal input binding with explicit preroll history and per-modality age.
+"""Compatibility imports for the common causal binding; no alternate rules."""
 
-No new inference or interpolation. The caller retains all preroll points in its
+from k1link.perception.streaming_sensors import (
-bounded rolling window; only the initial current-pair selection is narrowed.
+    NEWEST_POINT_AGE_NS,
-After the first camera, the original increment admission rules are unchanged.
+    OLDEST_POINT_AGE_NS,
-"""
+    POINT_POSE_SKEW_NS,
+    POSE_AGE_NS,
+    SensorBinding,
+    bind_sensors,
+    increment_identity,
+    milliseconds,
+)
 
-from dataclasses import dataclass
+__all__ = [
-
+    "NEWEST_POINT_AGE_NS",
-POSE_AGE_NS = 100_000_000
+    "OLDEST_POINT_AGE_NS",
-NEWEST_POINT_AGE_NS = 100_000_000
+    "POINT_POSE_SKEW_NS",
-OLDEST_POINT_AGE_NS = 250_000_000
+    "POSE_AGE_NS",
-POINT_POSE_SKEW_NS = 100_000_000
+    "SensorBinding",
-
+    "bind_sensors",
-
+    "increment_identity",
-def increment_identity(event):
+    "milliseconds",
-    return {
+]
-        "sequence": event.sequence,
-        "host_monotonic_ns": event.time_ns,
-        "points": len(event.value[0]),
-    }
-
-
-def milliseconds(value):
-    return None if value is None else value / 1e6
-
-
-@dataclass(frozen=True)
-class SensorBinding:
-    increments: tuple
-    history_only: tuple
-    pose_age_ns: int | None
-    newest_point_age_ns: int | None
-    oldest_point_age_ns: int | None
-    binding_age_ns: int | None
-    pose_state: str
-    points_state: str
-    reasons: tuple[str, ...]
-
-    @property
-    def available(self):
-        return not self.reasons
-
-    def document(self):
-        return {
-            "schema_version": "missioncore.pilot-sensor-binding/v1",
-            "pose_state": self.pose_state,
-            "points_state": self.points_state,
-            "current_pair_available": self.available,
-            "reason_codes": list(self.reasons),
-            "pose_age_ms": milliseconds(self.pose_age_ns),
-            "newest_point_age_ms": milliseconds(self.newest_point_age_ns),
-            "oldest_point_age_ms": milliseconds(self.oldest_point_age_ns),
-            "point_pose_skew_ms": milliseconds(self.binding_age_ns),
-            "preroll_history_only": [increment_identity(e) for e in self.history_only],
-            "preroll_history_disposition": "retained-in-bounded-rolling-window",
-        }
-
-
-def bind_sensors(camera_time_ns, pose, increments, *, previous_camera_time_ns=None):
-    increments = tuple(increments)
-    if previous_camera_time_ns is not None and previous_camera_time_ns >= camera_time_ns:
-        raise ValueError("camera binding clock must increase")
-    if pose is not None and pose.time_ns > camera_time_ns:
-        raise ValueError("future pose cannot bind to camera")
-    if any(e.time_ns > camera_time_ns for e in increments):
-        raise ValueError("future points cannot bind to camera")
-    if any(a.time_ns > b.time_ns for a, b in zip(increments, increments[1:], strict=False)):
-        raise ValueError("point binding clock moved backwards")
-
-    history_only = ()
-    if previous_camera_time_ns is None:
-        # History used to warm rolling geometry is not one current increment.
-        # Preserve its identity separately; do not retimestamp or silently drop it.
-        selected, history = [], []
-        for event in increments:
-            is_current = (
-                pose is not None
-                and camera_time_ns - event.time_ns <= OLDEST_POINT_AGE_NS
-                and abs(event.time_ns - pose.time_ns) <= POINT_POSE_SKEW_NS
-            )
-            (selected if is_current else history).append(event)
-        increments, history_only = tuple(selected), tuple(history)
-
-    pose_age = None if pose is None else camera_time_ns - pose.time_ns
-    newest_age = None if not increments else camera_time_ns - increments[-1].time_ns
-    oldest_age = None if not increments else camera_time_ns - increments[0].time_ns
-    skew = (
-        max(abs(e.time_ns - pose.time_ns) for e in increments)
-        if increments and pose is not None
-        else None
-    )
-    reasons = []
-    if pose is None:
-        pose_state = "unavailable"
-        reasons.append("pose-unavailable")
-    elif pose_age > POSE_AGE_NS:
-        pose_state = "stale"
-        reasons.append("pose-too-old")
-    else:
-        pose_state = (
-            "held"
-            if previous_camera_time_ns is not None and pose.time_ns <= previous_camera_time_ns
-            else "current"
-        )
-    if not increments or not any(len(e.value[0]) for e in increments):
-        points_state = "unavailable"
-        reasons.append("point-increment-unavailable")
-    else:
-        points_state = "current"
-        if newest_age > NEWEST_POINT_AGE_NS:
-            points_state = "stale"
-            reasons.append("newest-points-too-old")
-        if oldest_age > OLDEST_POINT_AGE_NS:
-            points_state = "stale"
-            reasons.append("oldest-points-too-old")
-    if skew is not None and skew > POINT_POSE_SKEW_NS:
-        reasons.append("point-pose-skew")
-    return SensorBinding(
-        increments,
-        history_only,
-        pose_age,
-        newest_age,
-        oldest_age,
-        skew,
-        pose_state,
-        points_state,
-        tuple(reasons),
-    )

experiments/perception/worker/streaming_profile_stage1/run_joint_pilot.py

@@ -35,6 +35,8 @@ from pilot_telemetry import NvmlSampler
 
 
 def distribution(values):
+    # Missing diagnostics are not zero; binary source lag is measured separately.
+    values = [value for value in values if value is not None]
     if not values:
         return None
     values = sorted(values)
@@ -271,6 +273,7 @@ def run(args):
         "triton_verbose": args.triton_verbose,
         "costmap_freshness_mode": args.costmap_freshness,
         "ddrnet_min_source_interval_ms": args.ddrnet_min_source_interval_ms,
+        "input_transport": args.input_transport,
     }
     children, logs, results, samples = [], [], [], []
     stop = threading.Event()
@@ -279,6 +282,7 @@ def run(args):
     producer = monitor = graph = gpu_stage = ddr_backend = None
     cpu_bundle = None
     controller = None
+    binary_bridge = decoder_client = None
     epoch_id = args.run_id
 
     def child(name, command, env=None):
@@ -385,10 +389,28 @@ def run(args):
                 ddr_backend.infer(tensor)
         decoder = child(
             "decoder",
-            [python, "-B", "/probe/pilot_model.py", "camera"],
+            (
-            {**os.environ, "PYTHONPATH": "/probe", "CUDA_VISIBLE_DEVICES": ""},
+                ["python3", "-B", "/probe/pilot_fragment_decoder.py"]
+                if args.input_transport == "binary-ipc"
+                else [python, "-B", "/probe/pilot_model.py", "camera"]
+            ),
+            {
+                **os.environ,
+                "PYTHONPATH": os.environ["PYTHONPATH"]
+                if args.input_transport == "binary-ipc"
+                else "/probe",
+                "CUDA_VISIBLE_DEVICES": "",
+            },
         )
-        report["decoder_ready"], _ = receive(decoder.stdout)
+        if args.input_transport == "binary-ipc":
+            from k1link.perception.streaming_decoder_client import StreamingDecoderClient
+
+            decoder_client = StreamingDecoderClient(
+                controller.runtime, decoder.stdout.fileno(), decoder.stdin.fileno()
+            )
+            report["decoder_ready"] = decoder_client.ready
+        else:
+            report["decoder_ready"], _ = receive(decoder.stdout)
         tgs = child("tgs", ["/usr/local/bin/pilot-tgs"])
         from pilot_graph import JointGraph
 
@@ -489,12 +511,22 @@ def run(args):
             from pilot_lifecycle import FencedIngress
 
             ingress = FencedIngress(controller)
-        producer = threading.Thread(
+        if args.input_transport == "binary-ipc":
-            target=produce, args=(args, decoder, ingress, stop, source_report), daemon=True
+            from pilot_binary_bridge import BinaryGraphBridge
-        )
+            from pilot_binary_source import RecordingSource
-        if controller:
+
-            controller.runtime.track_thread(producer)
+            source = RecordingSource(args, controller.runtime, source_report)
-        producer.start()
+            binary_bridge = BinaryGraphBridge(
+                controller.runtime, decoder_client, source, source_report
+            )
+            binary_bridge.start()
+        else:
+            producer = threading.Thread(
+                target=produce, args=(args, decoder, ingress, stop, source_report), daemon=True
+            )
+            if controller:
+                controller.runtime.track_thread(producer)
+            producer.start()
         if args.schedule == "overlap-cpu":
             gpu_stage = GpuStage(mailbox, compute_gpu, stop)
             if controller:
@@ -666,6 +698,10 @@ def run(args):
                     process.wait(timeout=1)
         if producer:
             producer.join(timeout=2)
+        if binary_bridge:
+            report["binary_bridge_stopped"] = binary_bridge.close()
+        if decoder_client and (not binary_bridge or report["binary_bridge_stopped"]):
+            decoder_client.close()
         if monitor:
             monitor.join(timeout=2)
         report["gpu_stage_stopped"] = gpu_stage.close() if gpu_stage else True
@@ -813,6 +849,10 @@ if __name__ == "__main__":
     parser.add_argument("--output", required=True)
     parser.add_argument("--sensor-archive", default="/sensor-source.npz")
     parser.add_argument("--camera-index", default="/camera-index.jsonl")
+    parser.add_argument("--camera-root", default="/camera")
+    parser.add_argument(
+        "--input-transport", choices=("legacy-pilot", "binary-ipc"), default="legacy-pilot"
+    )
     parser.add_argument(
         "--ddrnet-layout", choices=("reference", "channels-last"), default="reference"
     )
@@ -844,4 +884,6 @@ if __name__ == "__main__":
         parser.error("worker lease and launcher-verified image identity are required together")
     if args.stop_renew_after_sequence != -1 and not args.worker_lease_root:
         parser.error("lease-expiry injection requires a controller")
+    if args.input_transport == "binary-ipc" and not args.worker_lease_root:
+        parser.error("binary ingress requires the common lifecycle")
     raise SystemExit(run(args))

src/k1link/perception/streaming_decoder_client.py

@@ -0,0 +1,64 @@
+"""Fenced decoder RPC; native execution lives in a separately supervised child."""
+
+from __future__ import annotations
+
+import math
+from typing import Any
+
+from .graph_contracts import GraphState
+from .streaming_lifecycle import StreamingLifecycle
+from .streaming_pipe_rpc import HEADER_SCRATCH, MAX_OUTPUT, BoundedPipeRpc, PipeRpcError
+
+
+class StreamingDecoderClient:
+    def __init__(self, runtime: StreamingLifecycle, read_fd: int, write_fd: int) -> None:
+        if runtime.state != GraphState.STARTING:
+            raise PipeRpcError("decoder must be created during warmup")
+        self.runtime = runtime
+        self.reservation = runtime.mailbox.reserve_ingress(HEADER_SCRATCH)
+        self.closed = False
+        self.frames = 0
+        try:
+            self.rpc = BoundedPipeRpc(
+                read_fd, write_fd, lambda: runtime.check_current(runtime.start, starting=True)
+            )
+            self.ready = self.rpc.exchange(None, b"", bytearray(), timeout=10)
+            if self.ready != {"ready": "fragment-h264", "pyav": "18.0.0", "as_limit_mib": 1024}:
+                raise PipeRpcError("decoder startup identity changed")
+            self.rpc.check = lambda: runtime.check_current(runtime.start)
+        except BaseException:
+            runtime.request_stop("failed")
+            self.close()
+            raise
+
+    def configure(self, raw: bytes) -> None:
+        if self.closed or not 0 < len(raw) <= 65536:
+            raise PipeRpcError("decoder init outside bound")
+        if self.rpc.exchange({"op": "init"}, raw, bytearray()) != {"initialized": True}:
+            raise PipeRpcError("decoder init response changed")
+
+    def decode(self, raw: bytes, target: bytearray) -> dict[str, Any]:
+        # The caller has already reserved this exact BGR buffer in a decoded
+        # bundle ticket. RPC fills it directly, without an unaccounted copy.
+        if self.closed or len(target) != MAX_OUTPUT:
+            raise PipeRpcError("decoder output allocation changed")
+        result = self.rpc.exchange({"op": "decode"}, raw, target)
+        duration = result.get("decode_ms")
+        if (
+            set(result) != {"decode_ms", "frame_index"}
+            or type(result["frame_index"]) is not int
+            or result["frame_index"] != self.frames
+            or not isinstance(duration, (int, float))
+            or isinstance(duration, bool)
+            or not math.isfinite(duration)
+            or duration < 0
+        ):
+            raise PipeRpcError("decoder frame identity or measurement changed")
+        self.frames += 1
+        return result
+
+    def close(self) -> None:
+        # Caller joins its callback and stops the owned child before closing.
+        if not self.closed:
+            self.closed = True
+            self.reservation.release()

src/k1link/perception/streaming_lifecycle.py

@@ -21,7 +21,7 @@ from typing import Any, Literal
 from .graph_contracts import GraphState
 from .realtime_contract import StreamStart
 from .realtime_scene import _wire_integer
-from .streaming_queue import StreamBundle, StreamMailbox
+from .streaming_queue import IngressReservation, StreamBundle, StreamMailbox
 from .worker_lease import WorkerLease, WorkerLeaseError
 
 
@@ -116,6 +116,13 @@ class StreamingLifecycle:
             self._check(start)
             return self.mailbox.put(bundle)
 
+    def admit_reserved(
+        self, start: StreamStart, bundle: StreamBundle, reservation: IngressReservation
+    ) -> bool:
+        with self._lock:
+            self._check(start)
+            return self.mailbox.put_reserved(bundle, reservation)
+
     @contextmanager
     def work(self, start: StreamStart, lane: Literal["gpu", "cpu"]) -> Iterator[None]:
         with self._lock:
@@ -133,9 +140,10 @@ class StreamingLifecycle:
             with self._lock:
                 self._active[lane] -= 1
 
-    def check_current(self, start: StreamStart) -> None:
+    def check_current(self, start: StreamStart, *, starting: bool = False) -> None:
+        """Only child startup handshakes may opt into the warmup state."""
         with self._lock:
-            self._check(start)
+            self._check(start, starting=starting)
 
     def validate_result_binding(self, value: object) -> None:
         """Recheck at receipt/use; accepting a hash alone cannot renew authority."""

src/k1link/perception/streaming_pipe_rpc.py

@@ -0,0 +1,130 @@
+"""Bounded supervised-child RPC with caller-owned output buffers, POSIX only.
+
+Same small header/binary framing as the existing pilot RPC. No process creation,
+paths, models, queues or implicit allocation of image payloads. Caller reserves
+the target buffer and HEADER_SCRATCH before using this transport. Native child
+memory is separately constrained; timeout poisons this connection, never retries.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import select
+import struct
+import threading
+import time
+from collections.abc import Callable, Mapping
+from typing import Any
+
+HEADER_SCRATCH = 65536
+MAX_HEADER = 4096
+MAX_INPUT = 1024 * 1024
+MAX_OUTPUT = 1_440_000
+
+
+class PipeRpcError(ValueError):
+    pass
+
+
+def _unique(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
+    result: dict[str, Any] = {}
+    for key, value in pairs:
+        if key in result:
+            raise PipeRpcError("duplicate RPC field")
+        result[key] = value
+    return result
+
+
+class BoundedPipeRpc:
+    def __init__(self, read_fd: int, write_fd: int, check: Callable[[], None]) -> None:
+        self.read_fd, self.write_fd, self.check = read_fd, write_fd, check
+        os.set_blocking(read_fd, False)
+        os.set_blocking(write_fd, False)
+        self._lock = threading.Lock()
+        self.failed = False
+
+    def _check(self, deadline: float) -> None:
+        self.check()
+        if time.monotonic() >= deadline:
+            raise PipeRpcError("decoder RPC deadline exceeded")
+
+    def _read(self, target: memoryview, deadline: float) -> None:
+        offset = 0
+        while offset < len(target):
+            self._check(deadline)
+            if not select.select([self.read_fd], [], [], 0.01)[0]:
+                continue
+            try:
+                count = os.readv(self.read_fd, [target[offset:]])
+            except BlockingIOError:
+                continue
+            if not count:
+                raise PipeRpcError("truncated decoder RPC")
+            offset += count
+        self._check(deadline)
+
+    def _write(self, data: bytes | memoryview, deadline: float) -> None:
+        view = memoryview(data)
+        while view:
+            self._check(deadline)
+            if not select.select([], [self.write_fd], [], 0.01)[1]:
+                continue
+            try:
+                count = os.write(self.write_fd, view)
+            except BlockingIOError:
+                continue
+            if not count:
+                raise PipeRpcError("decoder RPC write made no progress")
+            view = view[count:]
+        self._check(deadline)
+
+    def exchange(
+        self,
+        header: Mapping[str, Any] | None,
+        payload: bytes,
+        target: bytearray,
+        *,
+        timeout: float = 0.25,
+    ) -> dict[str, Any]:
+        """None header reads a one-time startup message; no work is submitted."""
+        if not 0 < timeout <= (10 if header is None else 0.25):
+            raise PipeRpcError("RPC timeout outside bound")
+        if len(payload) > MAX_INPUT or len(target) > MAX_OUTPUT:
+            raise PipeRpcError("RPC payload outside bound")
+        if not self._lock.acquire(blocking=False):
+            raise PipeRpcError("concurrent decoder RPC is forbidden")
+        try:
+            if self.failed:
+                raise PipeRpcError("decoder RPC is poisoned")
+            deadline = time.monotonic() + timeout
+            self._check(deadline)
+            if header is not None:
+                raw = json.dumps(
+                    {**header, "payload_bytes": len(payload)}, allow_nan=False
+                ).encode()
+                if not 0 < len(raw) <= MAX_HEADER:
+                    raise PipeRpcError("RPC header outside bound")
+                self._write(struct.pack("<I", len(raw)), deadline)
+                self._write(raw, deadline)
+                self._write(payload, deadline)
+            prefix = bytearray(4)
+            self._read(memoryview(prefix), deadline)
+            size = struct.unpack("<I", prefix)[0]
+            if not 0 < size <= MAX_HEADER:
+                raise PipeRpcError("RPC header outside bound")
+            response = bytearray(size)
+            self._read(memoryview(response), deadline)
+            value = json.loads(response, object_pairs_hook=_unique)
+            if not isinstance(value, dict):
+                raise PipeRpcError("RPC header must be an object")
+            length = value.pop("payload_bytes", None)
+            if type(length) is not int or length != len(target):
+                raise PipeRpcError("decoder response byte count changed")
+            self._read(memoryview(target), deadline)
+            return value
+        except BaseException:
+            self.failed = True
+            raise
+        finally:
+            self._lock.release()

src/k1link/perception/streaming_queue.py

@@ -96,6 +96,30 @@ class StreamMailbox:
         self.drop_counts[reason] += 1
         self._recent_drops.append({"sequence": bundle["sequence"], "reason": reason})
 
+    def put_reserved(self, bundle: StreamBundle, reservation: IngressReservation) -> bool:
+        """Atomically transfer preallocated decoded bytes into queue ownership.
+
+        Success consumes the reservation. On rejection/error the caller still
+        owns it and must drop its buffers BEFORE releasing it. No unaccounted
+        interval or transient double charge; active inputs cannot be evicted.
+        """
+        with self.condition:
+            size = self._ingress.get(reservation)
+            if size is None or type(bundle["payload_bytes"]) is not int:
+                raise ValueError("decoded handoff ownership mismatch")
+            if bundle["payload_bytes"] != size:
+                raise ValueError("decoded handoff size mismatch")
+            del self._ingress[reservation]
+            self.bytes -= size
+            accepted = False
+            try:
+                accepted = self.put(bundle)
+                return accepted
+            finally:
+                if not accepted:
+                    self._ingress[reservation] = size
+                    self.bytes += size
+
     def _discard_pending(self, reason: str) -> None:
         old = self.pending.popleft()
         _, size = self._owned.pop(id(old))

src/k1link/perception/streaming_sensors.py

@@ -0,0 +1,239 @@
+"""Causal input binding with explicit preroll history and per-modality age.
+
+No new inference or interpolation. The caller retains all preroll points in its
+bounded rolling window; only the initial current-pair selection is narrowed.
+After the first camera, the original increment admission rules are unchanged.
+"""
+
+from collections import deque
+from collections.abc import Sequence
+from dataclasses import dataclass
+from typing import Any
+
+import numpy as np
+
+from .streaming_queue import StreamMailbox
+
+
+@dataclass(frozen=True)
+class SensorEvent:
+    time_ns: int
+    channel: str
+    sequence: int
+    value: Any
+
+
+POSE_AGE_NS = 100_000_000
+NEWEST_POINT_AGE_NS = 100_000_000
+OLDEST_POINT_AGE_NS = 250_000_000
+POINT_POSE_SKEW_NS = 100_000_000
+
+
+def increment_identity(event: SensorEvent) -> dict[str, int]:
+    return {
+        "sequence": event.sequence,
+        "host_monotonic_ns": event.time_ns,
+        "points": len(event.value[0]),
+    }
+
+
+def milliseconds(value: int | None) -> float | None:
+    return None if value is None else value / 1e6
+
+
+@dataclass(frozen=True)
+class SensorBinding:
+    increments: tuple[SensorEvent, ...]
+    history_only: tuple[SensorEvent, ...]
+    pose_age_ns: int | None
+    newest_point_age_ns: int | None
+    oldest_point_age_ns: int | None
+    binding_age_ns: int | None
+    pose_state: str
+    points_state: str
+    reasons: tuple[str, ...]
+
+    @property
+    def available(self) -> bool:
+        return not self.reasons
+
+    def document(self) -> dict[str, Any]:
+        return {
+            "schema_version": "missioncore.pilot-sensor-binding/v1",
+            "pose_state": self.pose_state,
+            "points_state": self.points_state,
+            "current_pair_available": self.available,
+            "reason_codes": list(self.reasons),
+            "pose_age_ms": milliseconds(self.pose_age_ns),
+            "newest_point_age_ms": milliseconds(self.newest_point_age_ns),
+            "oldest_point_age_ms": milliseconds(self.oldest_point_age_ns),
+            "point_pose_skew_ms": milliseconds(self.binding_age_ns),
+            "preroll_history_only": [increment_identity(e) for e in self.history_only],
+            "preroll_history_disposition": "retained-in-bounded-rolling-window",
+        }
+
+
+def bind_sensors(
+    camera_time_ns: int,
+    pose: SensorEvent | None,
+    increments: Sequence[SensorEvent],
+    *,
+    previous_camera_time_ns: int | None = None,
+) -> SensorBinding:
+    increments = tuple(increments)
+    if previous_camera_time_ns is not None and previous_camera_time_ns >= camera_time_ns:
+        raise ValueError("camera binding clock must increase")
+    if pose is not None and pose.time_ns > camera_time_ns:
+        raise ValueError("future pose cannot bind to camera")
+    if any(e.time_ns > camera_time_ns for e in increments):
+        raise ValueError("future points cannot bind to camera")
+    if any(a.time_ns > b.time_ns for a, b in zip(increments, increments[1:], strict=False)):
+        raise ValueError("point binding clock moved backwards")
+
+    history_only: tuple[SensorEvent, ...] = ()
+    if previous_camera_time_ns is None:
+        # History used to warm rolling geometry is not one current increment.
+        # Preserve its identity separately; do not retimestamp or silently drop it.
+        selected: list[SensorEvent] = []
+        history: list[SensorEvent] = []
+        for event in increments:
+            is_current = (
+                pose is not None
+                and camera_time_ns - event.time_ns <= OLDEST_POINT_AGE_NS
+                and abs(event.time_ns - pose.time_ns) <= POINT_POSE_SKEW_NS
+            )
+            (selected if is_current else history).append(event)
+        increments, history_only = tuple(selected), tuple(history)
+
+    pose_age = None if pose is None else camera_time_ns - pose.time_ns
+    newest_age = None if not increments else camera_time_ns - increments[-1].time_ns
+    oldest_age = None if not increments else camera_time_ns - increments[0].time_ns
+    skew = (
+        max(abs(e.time_ns - pose.time_ns) for e in increments)
+        if increments and pose is not None
+        else None
+    )
+    reasons = []
+    if pose is None:
+        pose_state = "unavailable"
+        reasons.append("pose-unavailable")
+    elif pose_age is not None and pose_age > POSE_AGE_NS:
+        pose_state = "stale"
+        reasons.append("pose-too-old")
+    else:
+        pose_state = (
+            "held"
+            if previous_camera_time_ns is not None and pose.time_ns <= previous_camera_time_ns
+            else "current"
+        )
+    if not increments or not any(len(e.value[0]) for e in increments):
+        points_state = "unavailable"
+        reasons.append("point-increment-unavailable")
+    else:
+        points_state = "current"
+        if newest_age is not None and newest_age > NEWEST_POINT_AGE_NS:
+            points_state = "stale"
+            reasons.append("newest-points-too-old")
+        if oldest_age is not None and oldest_age > OLDEST_POINT_AGE_NS:
+            points_state = "stale"
+            reasons.append("oldest-points-too-old")
+    if skew is not None and skew > POINT_POSE_SKEW_NS:
+        reasons.append("point-pose-skew")
+    return SensorBinding(
+        increments,
+        history_only,
+        pose_age,
+        newest_age,
+        oldest_age,
+        skew,
+        pose_state,
+        points_state,
+        tuple(reasons),
+    )
+
+
+def normalized_sensor(modality: str, sequence: int, time_ns: int, raw: bytes) -> SensorEvent:
+    """Explicit existing normalized-map wire layout, not a vendor packet parser."""
+    if modality == "lidar":
+        count = int.from_bytes(raw[:4], "little")
+        if len(raw) < 4 or not 0 <= count <= 50000 or len(raw) != 4 + count * 25:
+            raise ValueError("normalized point layout outside bound")
+        xyz = np.frombuffer(raw, "<f8", count=count * 3, offset=4).reshape(count, 3)
+        intensity = np.frombuffer(raw, "u1", count=count, offset=4 + count * 24)
+        # Bound validation temporaries inside the window's metadata/scratch
+        # allowance, even for a maximum-size incoming point increment.
+        flat = xyz.reshape(-1)
+        for offset in range(0, flat.size, 4096):
+            if not np.isfinite(flat[offset : offset + 4096]).all():
+                raise ValueError("nonfinite source cloud")
+        return SensorEvent(time_ns, "points", sequence, (xyz, intensity))
+    if modality != "pose" or len(raw) != 56:
+        raise ValueError("normalized pose layout outside bound")
+    values = np.frombuffer(raw, "<f8")
+    if not np.isfinite(values).all() or abs(float(np.linalg.norm(values[3:])) - 1) > 0.01:
+        raise ValueError("invalid source pose")
+    return SensorEvent(time_ns, "pose", sequence, (values[:3], values[3:]))
+
+
+class CausalSensorWindow:
+    """Single-consumer rolling/fresh caches; no async nearest/future lookup.
+
+    Reserve the maximum union of two 64000-point sets before retaining raw
+    views. Metadata and pose allowance is separate; Python objects/native RSS
+    still need the enclosing process limit. Sensor selection rules above are
+    exactly the existing pilot rules, including the initial history-only cut.
+    """
+
+    CACHE_BYTES = 2 * 64000 * 25 + 16384
+
+    def __init__(self, mailbox: StreamMailbox) -> None:
+        self._reservation = mailbox.reserve_ingress(self.CACHE_BYTES)
+        self.rolling: deque[SensorEvent] = deque()
+        self.fresh: list[SensorEvent] = []
+        self.pose: SensorEvent | None = None
+        self.previous_camera_time: int | None = None
+        self._last_time = -1
+        self.closed = False
+
+    def advance(self, time_ns: int) -> None:
+        if self.closed or time_ns < self._last_time:
+            raise ValueError("sensor stream is closed or moved backwards")
+        self._last_time = time_ns
+        while self.rolling and time_ns - self.rolling[0].time_ns > 1_000_000_000:
+            self.rolling.popleft()
+
+    def append(self, event: SensorEvent) -> None:
+        self.advance(event.time_ns)
+        if event.channel == "pose":
+            self.pose = event
+            return
+        if event.channel != "points":
+            raise ValueError("unsupported normalized sensor channel")
+        for items in (self.rolling, self.fresh):
+            if (
+                len(items) >= 64
+                or sum(len(e.value[0]) for e in items) + len(event.value[0]) > 64000
+            ):
+                raise ValueError("causal sensor cache exceeds bound")
+        self.rolling.append(event)
+        self.fresh.append(event)
+
+    def bind(self, time_ns: int) -> SensorBinding:
+        self.advance(time_ns)
+        return bind_sensors(
+            time_ns, self.pose, self.fresh, previous_camera_time_ns=self.previous_camera_time
+        )
+
+    def finish_camera(self, time_ns: int) -> None:
+        if self.closed or time_ns != self._last_time:
+            raise ValueError("sensor window changed during camera binding")
+        self.previous_camera_time = time_ns
+        self.fresh.clear()
+
+    def close(self) -> None:
+        if not self.closed:
+            self.rolling.clear()
+            self.fresh.clear()
+            self.pose = None
+            self.closed = True
+            self._reservation.release()

tests/test_perception_binary_bridge.py

@@ -0,0 +1,229 @@
+"""Small synthetic memory/clock/pipe checks; no real decode or models on the Mac."""
+
+import json
+import socket
+import struct
+from contextlib import contextmanager
+from pathlib import Path
+from types import SimpleNamespace
+
+import numpy as np
+import pytest
+
+from k1link.perception.streaming_pipe_rpc import BoundedPipeRpc, PipeRpcError
+from k1link.perception.streaming_queue import StreamMailbox
+from k1link.perception.streaming_sensors import CausalSensorWindow, normalized_sensor
+
+
+def bundle(seq=0, size=10):
+    return {"sequence": seq, "payload_bytes": size}
+
+
+def test_reservation_handoff_is_atomic_and_does_not_double_charge():
+    mailbox = StreamMailbox(byte_limit=10)
+    ticket = mailbox.reserve_ingress(10)
+    value = bundle()
+    assert mailbox.put_reserved(value, ticket)
+    assert mailbox.bytes == mailbox.peak_bytes == 10
+    with pytest.raises(ValueError, match="ownership"):
+        ticket.release()
+    assert mailbox.take() is value
+    mailbox.cancel()
+    assert mailbox.bytes == 10 and not mailbox.quiescent
+    mailbox.release(value)
+    assert mailbox.quiescent
+
+
+@pytest.mark.parametrize("case", ["closed", "size", "foreign", "sequence"])
+def test_rejected_handoff_keeps_caller_ownership(case):
+    mailbox = StreamMailbox(byte_limit=40)
+    ticket = mailbox.reserve_ingress(10)
+    value = bundle()
+    if case == "closed":
+        mailbox.cancel()
+        assert not mailbox.put_reserved(value, ticket)
+    else:
+        if case == "size":
+            value["payload_bytes"] = 11
+        if case == "sequence":
+            mailbox.put(bundle(1))
+        with pytest.raises(ValueError):
+            (StreamMailbox() if case == "foreign" else mailbox).put_reserved(value, ticket)
+    assert mailbox.bytes >= 10
+    ticket.size = 99999  # Only the stored reservation size is authoritative.
+    ticket.release()
+    mailbox.cancel()
+    assert mailbox.bytes == 0 and mailbox.quiescent
+
+
+def test_handoff_overflow_drops_pending_only():
+    mailbox = StreamMailbox(byte_limit=40)
+    mailbox.put(bundle(0))
+    active = mailbox.take()
+    mailbox.put(bundle(1, 5))
+    mailbox.put(bundle(2, 5))
+    ticket = mailbox.reserve_ingress(20)
+    assert mailbox.put_reserved(bundle(3, 20), ticket)
+    assert mailbox.dropped == [{"sequence": 1, "reason": "pending-overflow"}]
+    assert mailbox.bytes == 35
+    mailbox.cancel()
+    assert mailbox.bytes == 10
+    mailbox.release(active)
+    assert mailbox.quiescent
+
+
+@contextmanager
+def pipe(response=None, check=lambda: None):
+    client, server = socket.socketpair()
+    try:
+        if response:
+            server.sendall(response)
+        yield BoundedPipeRpc(client.fileno(), client.fileno(), check), server
+    finally:
+        client.close()
+        server.close()
+
+
+def message(header, payload=b""):
+    raw = json.dumps({**header, "payload_bytes": len(payload)}).encode()
+    return struct.pack("<I", len(raw)) + raw + payload
+
+
+def test_rpc_fills_only_the_callers_preallocated_buffer():
+    with pipe(message({"frame_index": 0}, b"1234")) as (rpc, server):
+        target = bytearray(4)
+        assert rpc.exchange({"op": "decode"}, b"fragment", target) == {"frame_index": 0}
+        assert target == b"1234"
+        raw = server.recv(1024)
+        size = struct.unpack("<I", raw[:4])[0]
+        assert json.loads(raw[4 : 4 + size]) == {"op": "decode", "payload_bytes": 8}
+        assert raw[4 + size :] == b"fragment"
+
+
+@pytest.mark.parametrize("case", ["timeout", "size", "header", "duplicate", "truncated", "guard"])
+def test_rpc_failures_are_bounded_and_cannot_be_retried(case):
+    raw = message({}, b"x") if case == "size" else None
+    if case == "header":
+        raw = struct.pack("<I", 4097)
+    if case == "duplicate":
+        body = b'{"payload_bytes":0,"payload_bytes":0}'
+        raw = struct.pack("<I", len(body)) + body
+
+    def guard():
+        if case == "guard":
+            raise ValueError("lost epoch")
+
+    with pipe(raw, guard) as (rpc, server):
+        if case == "truncated":
+            server.shutdown(socket.SHUT_WR)
+        with pytest.raises(ValueError):
+            rpc.exchange(None, b"", bytearray(), timeout=0.02)
+        assert rpc.failed
+        with pytest.raises(PipeRpcError, match="poisoned"):
+            rpc.exchange(None, b"", bytearray(), timeout=0.02)
+
+
+def point(stamp, seq=1):
+    raw = struct.pack("<I", 1) + np.array([[1, 2, 3]], "<f8").tobytes() + b"\xff"
+    return normalized_sensor("lidar", seq, stamp, raw)
+
+
+def pose(stamp):
+    return normalized_sensor("pose", 1, stamp, np.array([0, 0, 0, 0, 0, 0, 1], "<f8").tobytes())
+
+
+def test_causal_preroll_is_preserved_without_future_sensor_substitution():
+    mailbox = StreamMailbox()
+    window = CausalSensorWindow(mailbox)
+    window.append(point(500_000_000))
+    window.append(pose(990_000_000))
+    window.append(point(999_000_000, 2))
+    cut = window.bind(1_000_000_000)
+    assert cut.available and len(cut.increments) == len(cut.history_only) == 1
+    assert len(window.rolling) == 2 and cut.increments[0].sequence == 2
+    window.finish_camera(1_000_000_000)
+    window.append(pose(1_001_000_000))
+    with pytest.raises(ValueError, match="backwards"):
+        window.bind(1_000_000_000)
+    assert cut.pose_age_ns == 10_000_000  # New pose cannot mutate the existing cut.
+    mailbox.cancel()
+    assert mailbox.bytes == window.CACHE_BYTES and not mailbox.quiescent
+    window.close()
+    assert mailbox.quiescent
+
+
+def test_sensor_cache_cannot_grow_when_camera_stalls():
+    mailbox = StreamMailbox()
+    window = CausalSensorWindow(mailbox)
+    for sequence in range(64):
+        window.append(point(sequence, sequence))
+    with pytest.raises(ValueError, match="cache exceeds"):
+        window.append(point(65, 65))
+    window.close()
+    mailbox.finish()
+    assert mailbox.quiescent
+
+
+@pytest.mark.parametrize(
+    "modality,raw",
+    [
+        ("lidar", b""),
+        ("lidar", struct.pack("<I", 50001)),
+        ("pose", bytes(55)),
+        ("pose", bytes(56)),
+        ("unknown", bytes(56)),
+        ("lidar", struct.pack("<I", 1) + np.array([np.nan, 0, 0], "<f8").tobytes() + b"x"),
+    ],
+)
+def test_normalized_sensor_layout_is_explicit_and_validated(modality, raw):
+    with pytest.raises(ValueError):
+        normalized_sensor(modality, 1, 100, raw)
+
+
+def test_cloud_validation_checks_later_bounded_chunks():
+    xyz = np.ones((2000, 3), "<f8")
+    xyz[-1, -1] = np.inf
+    with pytest.raises(ValueError, match="nonfinite"):
+        normalized_sensor("lidar", 1, 100, struct.pack("<I", 2000) + xyz.tobytes() + bytes(2000))
+
+
+@pytest.mark.parametrize("failure", ["socket", "receiver-registration", "source-registration"])
+def test_bridge_initialization_failure_releases_unstarted_resources(monkeypatch, failure):
+    monkeypatch.syspath_prepend(
+        str(
+            Path(__file__).resolve().parents[1]
+            / "experiments/perception/worker/streaming_profile_stage1"
+        )
+    )
+    import pilot_binary_bridge as bridge
+
+    mailbox = StreamMailbox()
+    pair = socket.socketpair()
+    calls = []
+
+    def sockets():
+        if failure == "socket":
+            raise ValueError("injected socket failure")
+        return pair
+
+    def register(thread):
+        calls.append(thread)
+        if failure == "receiver-registration" or len(calls) == 2:
+            raise ValueError("injected registration failure")
+
+    monkeypatch.setattr(bridge.socket, "socketpair", sockets)
+    # Only constructor ownership is under test, without a live lease/thread.
+    monkeypatch.setattr("k1link.perception.streaming_ingress.wire.binding", lambda start: {})
+    runtime = SimpleNamespace(mailbox=mailbox, start=None, track_thread=register)
+    source = SimpleNamespace(source_zero=0, wall_zero=0, run=lambda sock: None)
+    try:
+        with pytest.raises(ValueError, match="injected"):
+            bridge.BinaryGraphBridge(runtime, None, source, {})
+        mailbox.finish()
+        assert mailbox.bytes == 0 and mailbox.quiescent
+        if failure != "socket":
+            assert all(s.fileno() == -1 for s in pair)
+        assert all(t.ident is None for t in calls)
+    finally:
+        for connection in pair:
+            connection.close()