feat(perception): add bounded binary stream ingress

experiments/perception/worker/streaming_profile_stage1/pilot_stream_ingress_probe.py

@@ -0,0 +1,312 @@
+"""CPU-only, bounded IPC evidence on Worker; NOT a model/network qualification.
+
+Recorded mode transfers original fMP4 init/media plus existing normalized
+point/pose increments at original 1x release times. Receiver is never told the
+source length. The consumer only hashes raw input, it does not decode/infer.
+Truncated mode is explicitly synthetic and exercises multi-fragment failure.
+"""
+
+import argparse
+import hashlib
+import json
+import resource
+import socket
+import struct
+import subprocess
+import sys
+import threading
+import time
+from datetime import UTC, datetime
+from pathlib import Path
+
+from pilot_source import SensorArchive, camera_events, merged_events
+
+from k1link.compute.live_perception import LiveIngressEvent
+from k1link.perception import streaming_wire as wire
+from k1link.perception.realtime_contract import StreamStart
+from k1link.perception.streaming_ingress import StreamingIngress
+from k1link.perception.streaming_lifecycle import StreamingLifecycle
+from k1link.perception.streaming_queue import StreamMailbox
+from k1link.perception.streaming_sender import StreamingSender
+
+
+def record(event):
+    return {
+        **wire.event_metadata(event),
+        "payload_bytes": len(event.payload),
+        "payload_sha256": hashlib.sha256(event.payload).hexdigest(),
+    }
+
+
+def emit_json(stream, value):
+    stream.write(json.dumps(value, sort_keys=True) + "\n")
+    stream.flush()
+
+
+def bounded_file(path, maximum):
+    with path.open("rb") as stream:
+        raw = stream.read(maximum + 1)
+    if not 0 < len(raw) <= maximum:
+        raise ValueError("source member exceeds bound")
+    return raw
+
+
+def recorded_events(root):
+    camera_root = Path("/camera")
+    camera_index = camera_root / "index.jsonl"
+    first = next(camera_events(camera_index, 1)).time_ns
+    zero = first - 500_000_000
+    wall = time.monotonic_ns() + 50_000_000
+    seq = 1
+    yield (
+        LiveIngressEvent(
+            seq,
+            "recorded-acquisition",
+            1,
+            "camera-init",
+            "sensor.camera.right",
+            0,
+            0,
+            zero,
+            bounded_file(camera_root / "init.mp4", wire.MAX_FRAGMENT),
+        ),
+        wall,
+    )
+    archive = SensorArchive(Path("/sensor-source.npz"))
+    try:
+        for observation in merged_events(archive, camera_index, 32):
+            if observation.time_ns < zero:
+                continue
+            due = wall + observation.time_ns - zero
+            threading.Event().wait(max(0, (due - time.monotonic_ns()) / 1e9))
+            seq += 1
+            utc = 0  # The normalized sensor archive has no UTC evidence; do not invent it.
+            if observation.channel == "camera":
+                row = observation.value
+                path = (camera_root / row["path"]).resolve()
+                if not path.is_relative_to(camera_root):
+                    raise ValueError("camera member escaped source root")
+                raw = bounded_file(path, wire.MAX_FRAGMENT)
+                if len(raw) != row["length"] or hashlib.sha256(raw).hexdigest() != row["sha256"]:
+                    raise ValueError("camera chunk integrity failed")
+                modality, source_id, utc = (
+                    "camera-frame",
+                    "sensor.camera.right",
+                    row["host_epoch_ns"],
+                )
+            elif observation.channel == "points":
+                xyz, intensity = observation.value
+                raw = struct.pack("<I", len(xyz)) + xyz.tobytes() + intensity.tobytes()
+                modality, source_id = "lidar", "normalized-map-point-increments"
+            else:
+                position, quaternion = observation.value
+                raw = position.tobytes() + quaternion.tobytes()
+                modality, source_id = "pose", "normalized-map-from-lidar-pose"
+            yield (
+                LiveIngressEvent(
+                    seq,
+                    "recorded-acquisition",
+                    1,
+                    modality,
+                    source_id,
+                    observation.sequence,
+                    utc,
+                    observation.time_ns,
+                    raw,
+                ),
+                due,
+            )
+            if observation.channel == "camera" and observation.sequence == 31:
+                break
+    finally:
+        (root / "source-reads.json").write_text(json.dumps(archive.counters(), indent=2))
+        archive.close()
+
+
+def producer(args):
+    root = Path(args.output)
+    identity = StreamStart.from_dict(json.loads((root / "start.json").read_text()))
+    connection = socket.socket(fileno=args.fd)
+    if connection.recv(1) != b"R":
+        raise ValueError("runtime did not become ready")
+    with (root / "sent.jsonl").open("w") as journal:
+        if args.case == "recorded":
+            sender = StreamingSender(connection, identity, "recorded-acquisition", 1, lambda: None)
+            for event, due in recorded_events(root):
+                sent = time.monotonic_ns()
+                sender.send(event)
+                emit_json(
+                    journal,
+                    {
+                        **record(event),
+                        "due_monotonic_ns": str(due),
+                        "send_start_monotonic_ns": str(sent),
+                        "send_end_monotonic_ns": str(time.monotonic_ns()),
+                    },
+                )
+            end_ns = time.monotonic_ns()
+            sender.end()
+        else:
+            connection.sendall(wire.open_packet(identity, "recorded-acquisition", 1))
+            event = LiveIngressEvent(
+                1,
+                "recorded-acquisition",
+                1,
+                "lidar",
+                "synthetic",
+                1,
+                0,
+                1,
+                b"x" * (wire.MAX_FRAGMENT + 17),
+            )
+            header, payload = next(wire.event_packets(identity, event))
+            connection.sendall(header)
+            connection.sendall(payload)
+            emit_json(journal, record(event))
+            end_ns = time.monotonic_ns()
+            connection.close()  # Deliberate EOF between fragments; no End.
+    (root / "source-end.json").write_text(json.dumps({"end_monotonic_ns": str(end_ns)}))
+    # The supervised source child remains alive until owned-group cleanup.
+    # Its normal completion is not mistaken for an unexpected model death.
+    threading.Event().wait(30)
+
+
+def run_case(base, case, generation):
+    root = base / case
+    root.mkdir()
+    specification = {
+        "case": case,
+        "transport": wire.SCHEMA,
+        "consumer": "raw-digest-only",
+        "camera": "original-fmp4-init-and-media",
+        "points": "u32le-count+xyz-f64le+intensity-u8",
+        "pose": "position-f64le3+map-from-lidar-quaternion-f64le4",
+        "fragment_timeout_ms": 250,
+        "idle_timeout_ms": 2000,
+    }
+    identity = StreamStart(
+        run_id=f"binary-ingress-{case}",
+        source_id="RAVNOVES00-prefix" if case == "recorded" else "synthetic",
+        worker_id="worker-006",
+        epoch_id=f"ipc-{generation}",
+        lease_generation=generation,
+        profile_sha256=hashlib.sha256(
+            bounded_file(base / "candidate-profile.json", 65536)
+        ).hexdigest(),
+        image_sha256="664824aa25de1db178f177d67a81b01541a938b479812b9383ddbf03f6b59dbe",
+        effective_config_sha256=hashlib.sha256(wire.canonical(specification)).hexdigest(),
+        # Opaque passthrough doesn't resolve or use calibration. Explicit synthetic
+        # identity is NOT eligible for a real perception scene.
+        calibration_sha256=hashlib.sha256(b"opaque-transport-no-calibration-use").hexdigest(),
+        clock_domain_id="original-host-arrival-clock",
+        input_mode="recorded-source-paced",
+    )
+    (root / "start.json").write_text(json.dumps(identity.to_dict(), indent=2))
+    run = StreamingLifecycle(
+        identity, Path("/tmp/worker-lease"), StreamMailbox(), threading.Event()
+    )
+    heartbeat_stop = threading.Event()
+
+    def renew():
+        while not heartbeat_stop.wait(0.25):
+            run.renew(identity)
+
+    heartbeat = threading.Thread(target=renew, daemon=True)
+    heartbeat.start()
+    left, right = socket.socketpair()
+    receiver = None
+    started = time.monotonic_ns()
+    with (root / "producer.log").open("wb") as log, (root / "received.jsonl").open("w") as journal:
+        try:
+            run.spawn(
+                lambda: subprocess.Popen(
+                    [
+                        sys.executable,
+                        "-B",
+                        __file__,
+                        "--producer",
+                        "--case",
+                        case,
+                        "--fd",
+                        str(left.fileno()),
+                        "--output",
+                        str(root),
+                    ],
+                    pass_fds=(left.fileno(),),
+                    start_new_session=True,
+                    stdin=subprocess.DEVNULL,
+                    stdout=log,
+                    stderr=log,
+                )
+            )
+            left.close()
+
+            def consume(event):
+                emit_json(
+                    journal,
+                    {
+                        **record(event),
+                        "received_monotonic_ns": str(time.monotonic_ns()),
+                        "original_received_monotonic_ns": str(event.received_monotonic_ns),
+                    },
+                )
+
+            receiver = StreamingIngress(
+                right,
+                run,
+                "recorded-acquisition",
+                1,
+                consume,
+                lambda notice: emit_json(journal, {"notice": notice}),
+            )
+            run.ready()
+            right.sendall(b"R")
+            receiver.start()
+            if not receiver.join(timeout=15):
+                raise TimeoutError("bounded stream probe did not finish")
+        finally:
+            run.request_stop("completed" if receiver and receiver.terminal == "end" else "failed")
+            run.stop_children()
+            left.close()
+            right.close()
+            if receiver:
+                receiver.join()
+            heartbeat_stop.set()
+            heartbeat.join(timeout=1)
+            released = run.close()
+    result = {
+        "created_utc": datetime.now(UTC).isoformat(),
+        "started_monotonic_ns": str(started),
+        "finished_monotonic_ns": str(time.monotonic_ns()),
+        "specification": specification,
+        "receiver": receiver.snapshot(),
+        "lifecycle": run.snapshot(),
+        "released": released,
+        "peak_input_bytes": run.mailbox.peak_bytes,
+        "residual_input_bytes": run.mailbox.bytes,
+        "maximum_rss_kib": resource.getrusage(resource.RUSAGE_SELF).ru_maxrss,
+        "child_maximum_rss_kib": resource.getrusage(resource.RUSAGE_CHILDREN).ru_maxrss,
+        "cgroup_memory_peak": Path("/sys/fs/cgroup/memory.peak").read_text().strip(),
+        "model_runs": 0,
+        "network_qualified": False,
+        "actuation_allowed": False,
+    }
+    (root / "report.json").write_text(json.dumps(result, indent=2))
+    assert released and run.mailbox.bytes == 0
+    assert receiver.terminal == ("end" if case == "recorded" else "failed")
+    assert receiver.snapshot()["incomplete_observations"] == (0 if case == "recorded" else 1)
+    print(json.dumps(result), flush=True)
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--producer", action="store_true")
+    parser.add_argument("--fd", type=int)
+    parser.add_argument("--case", choices=("recorded", "truncated"))
+    parser.add_argument("--output", default="/out")
+    args = parser.parse_args()
+    if args.producer:
+        producer(args)
+    else:
+        for generation, case in enumerate(("recorded", "truncated"), 1):
+            run_case(Path(args.output), case, generation)

src/k1link/perception/streaming_ingress.py

@@ -0,0 +1,293 @@
+"""One controller-supplied IPC socket feeding the common streaming lifecycle.
+
+No listener, acquisition, archive reader, model RPC, network authentication or
+automatic reconnect. Reconnect needs a new StreamStart/lease and decoder state.
+The synchronous consumer is a trusted adapter: raw events are borrowed for that
+call only. It must reserve decoder scratch before allocation and retain decoded
+work only through the common mailbox. Slow/failed consumers fail this stream;
+they do not create another buffering queue.
+"""
+
+from __future__ import annotations
+
+import select
+import socket
+import threading
+import time
+from collections import Counter
+from collections.abc import Callable
+from hashlib import sha256
+from typing import Any
+
+from k1link.compute.live_perception import LiveIngressEvent
+
+from . import streaming_wire as wire
+from .streaming_lifecycle import StreamingLifecycle
+
+
+class StreamingIngress:
+    def __init__(
+        self,
+        connection: socket.socket,
+        runtime: StreamingLifecycle,
+        session_id: str,
+        session_generation: int,
+        consume: Callable[[LiveIngressEvent], None],
+        notice: Callable[[dict[str, Any]], None],
+        *,
+        fragment_timeout: float = 0.25,
+        idle_timeout: float = 2.0,
+    ) -> None:
+        if not 0 < fragment_timeout <= 2 or not 0 < idle_timeout <= 30:
+            raise ValueError("invalid bounded ingress timeouts")
+        wire.identifier(session_id)
+        wire.decimal(session_generation)
+        if session_generation < 1:
+            raise ValueError("invalid acquisition generation")
+        self.connection, self.runtime = connection, runtime
+        connection.setblocking(False)
+        self.session_id, self.session_generation = session_id, session_generation
+        self.consume, self.notice = consume, notice
+        self.fragment_timeout, self.idle_timeout = fragment_timeout, idle_timeout
+        self.binding = wire.binding(runtime.start)
+        self.opened = False
+        self.error: str | None = None
+        self.terminal: str | None = None
+        self.counts: Counter[str] = Counter()
+        self.last_ingress_sequence = 0
+        self.channel_sequences: dict[str, tuple[str, int, int]] = {}
+        self.camera_initialized = False
+        self.thread = threading.Thread(
+            target=self._serve, name="perception-binary-ingress", daemon=True
+        )
+        runtime.track_thread(self.thread)
+
+    def start(self) -> None:
+        self.runtime.check_current(self.runtime.start)
+        self.thread.start()
+
+    def join(self, timeout: float = 1.0) -> bool:
+        self.thread.join(timeout=timeout)
+        return not self.thread.is_alive()
+
+    def _check(self, deadline: float) -> None:
+        self.runtime.check_current(self.runtime.start)
+        if time.monotonic() >= deadline:
+            raise wire.StreamWireError("ingress deadline exceeded")
+
+    def _read_into(self, target: memoryview, deadline: float) -> None:
+        offset = 0
+        while offset < len(target):
+            self._check(deadline)
+            if not select.select(
+                [self.connection], [], [], min(0.05, max(0, deadline - time.monotonic()))
+            )[0]:
+                continue
+            try:
+                read = self.connection.recv_into(target[offset:])
+            except BlockingIOError:
+                continue
+            if read == 0:
+                raise wire.StreamWireError("unexpected EOF without explicit End")
+            self.counts["wire_bytes"] += read
+            offset += read
+        self._check(deadline)
+
+    def _header(self, deadline: float) -> tuple[int, dict[str, Any]]:
+        prefix = bytearray(wire.PREFIX.size)
+        self._read_into(memoryview(prefix), deadline)
+        magic, kind, size = wire.PREFIX.unpack(prefix)
+        if (
+            magic != wire.MAGIC
+            or kind not in (wire.OPEN, wire.FRAGMENT, wire.END, wire.CANCEL, wire.GAP)
+            or not 0 < size <= wire.MAX_HEADER
+        ):
+            raise wire.StreamWireError("invalid frame prefix or header size")
+        raw = bytearray(size)
+        self._read_into(memoryview(raw), deadline)
+        return kind, wire.parse_header(raw)
+
+    def _bound(self, header: dict[str, Any]) -> None:
+        if header.get("binding") != self.binding:
+            raise wire.StreamWireError("frame belongs to another StreamStart")
+        self.runtime.check_current(self.runtime.start)
+
+    def _fragment(self, header: dict[str, Any]) -> tuple[dict[str, Any], int, int, int]:
+        if set(header) != {
+            "binding",
+            "event",
+            "total_bytes",
+            "offset",
+            "fragment_bytes",
+            "payload_sha256",
+            "fragment_sha256",
+        }:
+            raise wire.StreamWireError("fragment fields changed")
+        self._bound(header)
+        event = wire.validate_event_metadata(header["event"])
+        if (
+            event["session_id"] != self.session_id
+            or event["session_generation"] != self.session_generation
+        ):
+            raise wire.StreamWireError("event acquisition binding changed")
+        total, offset, size = (header[x] for x in ("total_bytes", "offset", "fragment_bytes"))
+        if any(type(x) is not int for x in (total, offset, size)) or not (
+            0 < total <= wire.PAYLOAD_LIMITS[event["modality"]]
+            and 0 <= offset < total
+            and 0 < size <= wire.MAX_FRAGMENT
+            and size == min(wire.MAX_FRAGMENT, total - offset)
+        ):
+            raise wire.StreamWireError("invalid fragment length or offset")
+        wire.digest(header["payload_sha256"])
+        wire.digest(header["fragment_sha256"])
+        return event, total, offset, size
+
+    def _observation(self, first: dict[str, Any], deadline: float) -> None:
+        event, total, offset, size = self._fragment(first)
+        modality = event["modality"]
+        if offset != 0 or event["ingress_sequence"] <= self.last_ingress_sequence:
+            raise wire.StreamWireError("duplicate, out-of-order or incomplete observation")
+        previous = self.channel_sequences.get(modality)
+        if previous and (
+            event["source_id"] != previous[0]
+            or event["source_sequence"] <= previous[1]
+            or event["received_monotonic_ns"] < previous[2]
+        ):
+            raise wire.StreamWireError("channel identity, sequence or source clock regressed")
+        if modality == "camera-frame" and not self.camera_initialized:
+            raise wire.StreamWireError("camera frame requires new codec initialization")
+        if modality == "camera-frame" and previous and event["source_sequence"] != previous[1] + 1:
+            raise wire.StreamWireError("camera sequence gap requires new codec initialization")
+        self.counts["observations_started"] += 1
+        reservation = self.runtime.mailbox.reserve_ingress(2 * total)
+        raw = None
+        admitted = None
+        try:
+            raw = bytearray(total)
+            header = first
+            while True:
+                chunk = memoryview(raw)[offset : offset + size]
+                try:
+                    self._read_into(chunk, deadline)
+                    if sha256(chunk).hexdigest() != header["fragment_sha256"]:
+                        raise wire.StreamWireError("fragment integrity mismatch")
+                finally:
+                    chunk.release()
+                self.counts["fragments"] += 1
+                offset += size
+                if offset == total:
+                    break
+                kind, header = self._header(deadline)
+                if kind != wire.FRAGMENT:
+                    raise wire.StreamWireError("incomplete observation before control event")
+                next_event, next_total, next_offset, size = self._fragment(header)
+                if (
+                    next_event != event
+                    or next_total != total
+                    or next_offset != offset
+                    or header["payload_sha256"] != first["payload_sha256"]
+                ):
+                    raise wire.StreamWireError("interleaved or discontinuous fragments")
+            if sha256(raw).hexdigest() != first["payload_sha256"]:
+                raise wire.StreamWireError("observation integrity mismatch")
+            self._check(deadline)
+            admitted = LiveIngressEvent(**event, payload=bytes(raw))
+            raw = None  # Immutable callback bytes remain fully reserved.
+            self.consume(admitted)
+            self._check(deadline)
+            self.counts["ingress_sequence_gaps"] += (
+                event["ingress_sequence"] - self.last_ingress_sequence - 1
+            )
+            self.last_ingress_sequence = event["ingress_sequence"]
+            self.channel_sequences[modality] = (
+                event["source_id"],
+                event["source_sequence"],
+                event["received_monotonic_ns"],
+            )
+            if modality == "camera-init":
+                self.camera_initialized = True
+                self.channel_sequences.pop("camera-frame", None)
+            self.counts["observations_completed"] += 1
+        finally:
+            admitted = raw = None
+            reservation.release()
+
+    def _gap(self, header: dict[str, Any]) -> None:
+        if set(header) != {"binding", "modality", "reason", "count"}:
+            raise wire.StreamWireError("gap fields changed")
+        if header["modality"] not in wire.PAYLOAD_LIMITS or header["reason"] not in (
+            "unavailable",
+            "source-gap",
+            "overload",
+        ):
+            raise wire.StreamWireError("invalid gap notice")
+        self.counts["declared_gap_observations"] += wire.uint64(header["count"])
+        if header["modality"].startswith("camera"):
+            self.camera_initialized = False
+        self.notice(dict(header))
+        self.runtime.check_current(self.runtime.start)
+        self.counts["gap_notices"] += 1
+
+    def _serve(self) -> None:
+        reservation = None
+        try:
+            # Covers raw header + JSON decoding copies; Python object overhead
+            # belongs to the separately measured RSS envelope, not payload bytes.
+            reservation = self.runtime.mailbox.reserve_ingress(
+                3 * wire.MAX_HEADER + wire.PREFIX.size
+            )
+            kind, header = self._header(time.monotonic() + self.idle_timeout)
+            expected = wire.parse_header(
+                bytearray(
+                    wire.open_packet(
+                        self.runtime.start,
+                        self.session_id,
+                        self.session_generation,
+                    )[wire.PREFIX.size :]
+                )
+            )
+            if kind != wire.OPEN or header != expected:
+                raise wire.StreamWireError("StreamStart or acquisition handshake mismatch")
+            self.opened = True
+            while True:
+                kind, header = self._header(time.monotonic() + self.idle_timeout)
+                self._bound(header)
+                if kind == wire.FRAGMENT:
+                    self._observation(header, time.monotonic() + self.fragment_timeout)
+                elif kind == wire.GAP:
+                    self._gap(header)
+                elif kind in (wire.END, wire.CANCEL) and set(header) == {"binding"}:
+                    self.terminal = "cancelled" if kind == wire.CANCEL else "end"
+                    if kind == wire.CANCEL:
+                        self.runtime.request_stop("cancelled")
+                    else:
+                        self.runtime.mailbox.finish()
+                    return
+                else:
+                    raise wire.StreamWireError("unexpected stream control event")
+        except Exception as exc:
+            self.error = str(exc)
+            self.terminal = "failed"
+            if self.opened:
+                self.runtime.mailbox.finish(self.error)
+                self.runtime.request_stop("failed")
+        finally:
+            self.connection.close()
+            if reservation is not None:
+                reservation.release()
+
+    def snapshot(self) -> dict[str, Any]:
+        return {
+            "schema_version": wire.SCHEMA,
+            "opened": self.opened,
+            "terminal": self.terminal,
+            "error": self.error,
+            "counts": dict(self.counts),
+            "incomplete_observations": self.counts["observations_started"]
+            - self.counts["observations_completed"],
+            "source_duration_known": False,
+            "requires_eof_before_delivery": False,
+            "transport": "controller-supplied-ipc-socket",
+            "network_qualified": False,
+            "actuation_allowed": False,
+        }

src/k1link/perception/streaming_queue.py

@@ -17,6 +17,21 @@ StreamBundle = Mapping[str, Any]
 DISCARD_REASONS = frozenset(("cancelled", "gpu-failed", "processing-failed"))
 
 
+class IngressReservation:
+    """Raw/reassembly bytes retained outside the decoded work queue.
+
+    Trusted adapters reserve BEFORE allocation and release AFTER borrowed raw
+    buffers/callbacks are gone. Closing ingress never frees a live reservation.
+    This is byte ownership, not another pending camera slot.
+    """
+
+    def __init__(self, mailbox: StreamMailbox, size: int) -> None:
+        self._mailbox, self.size = mailbox, size
+
+    def release(self) -> None:
+        self._mailbox._release_ingress(self)
+
+
 class StreamMailbox:
     def __init__(self, capacity: int = 2, byte_limit: int = 16 * 1024 * 1024) -> None:
         if type(capacity) is not int or not 1 <= capacity <= 2:
@@ -31,6 +46,7 @@ class StreamMailbox:
         self._recent_drops: deque[dict[str, Any]] = deque(maxlen=256)
         self._owned: dict[int, tuple[StreamBundle, int]] = {}
         self._active: set[int] = set()
+        self._ingress: dict[IngressReservation, int] = {}
         self._last_sequence = -1
         self.done = False
         self.error: str | None = None
@@ -49,7 +65,32 @@ class StreamMailbox:
     @property
     def quiescent(self) -> bool:
         with self.condition:
-            return self.done and not self._owned and not self.external_pending
+            return self.done and not self._owned and not self.external_pending and not self._ingress
+
+    def reserve_ingress(self, size: int) -> IngressReservation:
+        """Charge raw buffers to the SAME budget as queued and active inputs.
+
+        Fail closed on exhaustion; do not wait and silently stretch source time.
+        Decoder scratch must also be reserved before allocation; retained decoded
+        work uses the existing bundle ownership. No adapter may hide a buffer.
+        """
+        if type(size) is not int or size <= 0:
+            raise ValueError("invalid ingress reservation")
+        with self.condition:
+            if self.done or self.bytes + size > self.byte_limit or len(self._ingress) >= 8:
+                raise ValueError("ingress byte budget exhausted or closed")
+            reservation = IngressReservation(self, size)
+            self._ingress[reservation] = size
+            self.bytes += size
+            self.peak_bytes = max(self.peak_bytes, self.bytes)
+            return reservation
+
+    def _release_ingress(self, reservation: IngressReservation) -> None:
+        with self.condition:
+            if reservation not in self._ingress:
+                raise ValueError("ingress release ownership mismatch")
+            self.bytes -= self._ingress.pop(reservation)
+            self.condition.notify_all()
 
     def _drop(self, bundle: StreamBundle, reason: str) -> None:
         self.drop_counts[reason] += 1

src/k1link/perception/streaming_sender.py

@@ -0,0 +1,91 @@
+"""Synchronous, deadline-bounded IPC writer for existing raw ingress events.
+
+The caller owns pacing and already-committed source bytes. No background queue,
+recording inventory or retry is created; timeout is an explicit failed source
+outcome. A new connection requires a new controller-approved StreamStart.
+"""
+
+from __future__ import annotations
+
+import select
+import socket
+import time
+from collections.abc import Callable, Iterable
+
+from k1link.compute.live_perception import LiveIngressEvent
+
+from . import streaming_wire as wire
+from .realtime_contract import StreamStart
+
+
+class StreamingSender:
+    def __init__(
+        self,
+        connection: socket.socket,
+        start: StreamStart,
+        session_id: str,
+        session_generation: int,
+        check_current: Callable[[], None],
+        *,
+        timeout: float = 0.25,
+    ) -> None:
+        if not 0 < timeout <= 2:
+            raise ValueError("invalid bounded send timeout")
+        self.connection, self.identity, self.check_current = connection, start, check_current
+        self.session_id, self.session_generation = session_id, session_generation
+        self.timeout, self.closed = timeout, False
+        connection.setblocking(False)
+        try:
+            self._send((wire.open_packet(start, session_id, session_generation),))
+        except Exception:
+            self.close()
+            raise
+
+    def _send(self, pieces: Iterable[bytes | memoryview]) -> None:
+        if self.closed:
+            raise wire.StreamWireError("sender is closed")
+        deadline = time.monotonic() + self.timeout
+        try:
+            for piece in pieces:
+                view = memoryview(piece)
+                offset = 0
+                while offset < len(view):
+                    self.check_current()
+                    remaining = deadline - time.monotonic()
+                    if remaining <= 0:
+                        raise wire.StreamWireError("send deadline exceeded")
+                    if not select.select([], [self.connection], [], min(0.05, remaining))[1]:
+                        continue
+                    try:
+                        count = self.connection.send(view[offset:])
+                    except BlockingIOError:
+                        continue
+                    if count == 0:
+                        raise wire.StreamWireError("source connection closed")
+                    offset += count
+            self.check_current()
+            if time.monotonic() >= deadline:
+                raise wire.StreamWireError("send deadline exceeded")
+        except Exception:
+            self.close()
+            raise
+
+    def send(self, event: LiveIngressEvent) -> None:
+        if (
+            event.session_id != self.session_id
+            or event.session_generation != self.session_generation
+        ):
+            self.close()
+            raise wire.StreamWireError("source acquisition binding changed")
+        self._send(piece for pair in wire.event_packets(self.identity, event) for piece in pair)
+
+    def gap(self, *, modality: str, reason: str, count: int) -> None:
+        self._send((wire.gap_packet(self.identity, modality=modality, reason=reason, count=count),))
+
+    def end(self, *, cancel: bool = False) -> None:
+        self._send((wire.terminal_packet(self.identity, cancel=cancel),))
+        self.close()
+
+    def close(self) -> None:
+        self.closed = True
+        self.connection.close()

src/k1link/perception/streaming_wire.py

@@ -0,0 +1,211 @@
+"""Bounded v2 mapping of existing raw-first LiveIngressEvent semantics.
+
+This IPC candidate carries binary payloads, not base64/PNG/archive uploads.
+It is not the selected/authenticated network transport. A trusted controller
+binds one acquisition session to StreamStart; hashes provide integrity, NOT
+authentication. No peer-supplied paths, commands, models or calibration loading.
+"""
+
+from __future__ import annotations
+
+import json
+import re
+import struct
+from collections.abc import Iterator
+from hashlib import sha256
+from typing import Any, Final
+
+from k1link.compute.live_perception import LiveIngressEvent
+
+from .realtime_contract import StreamStart
+
+SCHEMA: Final = "missioncore.live-perception-wire/v2"
+MAGIC: Final = b"MCI2"
+PREFIX: Final = struct.Struct("!4sBI")
+OPEN, FRAGMENT, END, CANCEL, GAP = range(1, 6)
+MAX_HEADER: Final = 64 * 1024
+MAX_FRAGMENT: Final = 1024 * 1024
+PAYLOAD_LIMITS: Final = {
+    "camera-init": MAX_FRAGMENT,
+    "camera-frame": MAX_FRAGMENT,
+    "lidar": 2 * MAX_FRAGMENT,
+    "pose": 2 * MAX_FRAGMENT,
+}
+EVENT_FIELDS: Final = frozenset(
+    (
+        "ingress_sequence",
+        "session_id",
+        "session_generation",
+        "modality",
+        "source_id",
+        "source_sequence",
+        "captured_at_epoch_ns",
+        "received_monotonic_ns",
+    )
+)
+INTEGER_FIELDS: Final = EVENT_FIELDS - {"session_id", "modality", "source_id"}
+_DECIMAL: Final = re.compile(r"^(0|[1-9][0-9]{0,19})$")
+_DIGEST: Final = re.compile(r"^[a-f0-9]{64}$")
+
+
+class StreamWireError(ValueError):
+    """Malformed, stale, oversized or incomplete transport input."""
+
+
+def uint64(value: object) -> int:
+    if not isinstance(value, str) or not _DECIMAL.fullmatch(value):
+        raise StreamWireError("wire uint64 must be a canonical decimal string")
+    result = int(value)
+    if result > (1 << 64) - 1:
+        raise StreamWireError("wire uint64 overflow")
+    return result
+
+
+def decimal(value: object) -> str:
+    if type(value) is not int or not 0 <= value < 1 << 64:
+        raise StreamWireError("invalid uint64")
+    return str(value)
+
+
+def identifier(value: object) -> str:
+    if not isinstance(value, str) or not 1 <= len(value) <= 160:
+        raise StreamWireError("invalid bounded source identity")
+    if not value.isascii() or any(ord(c) < 33 or ord(c) > 126 for c in value):
+        raise StreamWireError("invalid bounded source identity")
+    return value
+
+
+def digest(value: object) -> str:
+    if not isinstance(value, str) or not _DIGEST.fullmatch(value):
+        raise StreamWireError("invalid payload digest")
+    return value
+
+
+def canonical(value: dict[str, Any]) -> bytes:
+    return json.dumps(value, sort_keys=True, separators=(",", ":"), allow_nan=False).encode()
+
+
+def start_document(start: StreamStart) -> dict[str, Any]:
+    result = start.to_dict()
+    result["lease_generation"] = decimal(start.lease_generation)
+    return result
+
+
+def binding(start: StreamStart) -> str:
+    return sha256(canonical(start_document(start))).hexdigest()
+
+
+def _unique_pairs(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
+    result: dict[str, Any] = {}
+    for key, value in pairs:
+        if key in result:
+            raise StreamWireError("duplicate metadata field")
+        result[key] = value
+    return result
+
+
+def parse_header(raw: bytearray) -> dict[str, Any]:
+    try:
+        value = json.loads(raw, object_pairs_hook=_unique_pairs)
+    except (ValueError, UnicodeError, RecursionError) as exc:
+        raise StreamWireError("invalid bounded metadata") from exc
+    if not isinstance(value, dict):
+        raise StreamWireError("metadata must be an object")
+    return value
+
+
+def packet(kind: int, header: dict[str, Any]) -> bytes:
+    raw = canonical(header)
+    if kind not in (OPEN, FRAGMENT, END, CANCEL, GAP) or not 0 < len(raw) <= MAX_HEADER:
+        raise StreamWireError("invalid frame header")
+    return PREFIX.pack(MAGIC, kind, len(raw)) + raw
+
+
+def open_packet(start: StreamStart, session_id: str, session_generation: int) -> bytes:
+    return packet(
+        OPEN,
+        {
+            "schema_version": SCHEMA,
+            "start": start_document(start),
+            "session_id": identifier(session_id),
+            "session_generation": decimal(session_generation),
+        },
+    )
+
+
+def event_metadata(event: LiveIngressEvent) -> dict[str, Any]:
+    result = {name: getattr(event, name) for name in EVENT_FIELDS}
+    for name in INTEGER_FIELDS:
+        result[name] = decimal(result[name])
+    validate_event_metadata(result)
+    return result
+
+
+def validate_event_metadata(value: object) -> dict[str, Any]:
+    if not isinstance(value, dict) or set(value) != EVENT_FIELDS:
+        raise StreamWireError("event fields changed")
+    result = dict(value)
+    for name in INTEGER_FIELDS:
+        result[name] = uint64(value[name])
+    for name in ("session_id", "source_id"):
+        identifier(value[name])
+    if (
+        not isinstance(value["modality"], str)
+        or value["modality"] not in PAYLOAD_LIMITS
+        or result["ingress_sequence"] < 1
+        or result["session_generation"] < 1
+    ):
+        raise StreamWireError("invalid raw modality or generation")
+    return result
+
+
+def event_packets(
+    start: StreamStart,
+    event: LiveIngressEvent,
+) -> Iterator[tuple[bytes, memoryview]]:
+    """Iterate one already committed raw event; never assemble a route/archive.
+
+    Source adapter owns the original event bytes. Returned views borrow them;
+    sender must finish each fragment before advancing this iterator.
+    """
+    metadata = event_metadata(event)
+    total = len(event.payload)
+    if not 0 < total <= PAYLOAD_LIMITS[event.modality]:
+        raise StreamWireError("raw event exceeds modality bound")
+    whole_hash = sha256(event.payload).hexdigest()
+    view = memoryview(event.payload)
+    for offset in range(0, total, MAX_FRAGMENT):
+        chunk = view[offset : offset + MAX_FRAGMENT]
+        yield (
+            packet(
+                FRAGMENT,
+                {
+                    "binding": binding(start),
+                    "event": metadata,
+                    "total_bytes": total,
+                    "offset": offset,
+                    "fragment_bytes": len(chunk),
+                    "payload_sha256": whole_hash,
+                    "fragment_sha256": sha256(chunk).hexdigest(),
+                },
+            ),
+            chunk,
+        )
+
+
+def terminal_packet(start: StreamStart, *, cancel: bool = False) -> bytes:
+    return packet(CANCEL if cancel else END, {"binding": binding(start)})
+
+
+def gap_packet(start: StreamStart, *, modality: str, reason: str, count: int) -> bytes:
+    if modality not in PAYLOAD_LIMITS or reason not in ("unavailable", "source-gap", "overload"):
+        raise StreamWireError("invalid gap notice")
+    return packet(
+        GAP,
+        {
+            "binding": binding(start),
+            "modality": modality,
+            "reason": reason,
+            "count": decimal(count),
+        },
+    )

tests/test_perception_streaming_ingress.py

@@ -0,0 +1,313 @@
+"""Bounded synthetic IPC fixtures, no model or local load tests."""
+
+import json
+import socket
+import threading
+from contextlib import contextmanager
+from dataclasses import replace
+
+import pytest
+
+from k1link.compute.live_perception import LiveIngressEvent
+from k1link.perception import streaming_wire as wire
+from k1link.perception.graph_contracts import GraphState
+from k1link.perception.realtime_contract import StreamStart
+from k1link.perception.streaming_ingress import StreamingIngress
+from k1link.perception.streaming_lifecycle import StreamingLifecycle
+from k1link.perception.streaming_queue import StreamMailbox
+from k1link.perception.streaming_sender import StreamingSender
+
+
+def start():
+    return StreamStart(
+        run_id="test-run",
+        source_id="recorded-or-live",
+        worker_id="worker-006",
+        epoch_id="epoch-1",
+        lease_generation=1,
+        profile_sha256="a" * 64,
+        image_sha256="b" * 64,
+        effective_config_sha256="c" * 64,
+        calibration_sha256="d" * 64,
+        clock_domain_id="original-host-clock",
+        input_mode="live",
+    )
+
+
+def event(seq=1, *, modality="lidar", payload=b"points", source_sequence=None):
+    return LiveIngressEvent(
+        ingress_sequence=seq,
+        session_id="capture-1",
+        session_generation=1,
+        modality=modality,
+        source_id=f"raw/{modality}",
+        source_sequence=seq if source_sequence is None else source_sequence,
+        captured_at_epoch_ns=1_799_999_999_123_456_789,
+        received_monotonic_ns=9_007_199_254_740_993 + seq,
+        payload=payload,
+    )
+
+
+def send_event(sock, observation, *, identity=None):
+    for header, payload in wire.event_packets(identity or start(), observation):
+        sock.sendall(header)
+        sock.sendall(payload)
+
+
+@contextmanager
+def harness(tmp_path, *, consume=None, byte_limit=16 * 1024 * 1024, clock=None, **kwargs):
+    mailbox = StreamMailbox(byte_limit=byte_limit)
+    run = StreamingLifecycle(
+        start(),
+        tmp_path,
+        mailbox,
+        threading.Event(),
+        **({"clock_ns": clock} if clock else {}),
+    )
+    left, right = socket.socketpair()
+    left.settimeout(1)
+    received, notices = [], []
+    run.ready()
+    receiver = StreamingIngress(
+        right, run, "capture-1", 1, consume or received.append, notices.append, **kwargs
+    )
+    receiver.start()
+    try:
+        yield left, receiver, run, received, notices
+    finally:
+        left.close()
+        run.request_stop("cancelled")
+        assert receiver.join()
+        assert run.close()
+
+
+def hello(sock):
+    sock.sendall(wire.open_packet(start(), "capture-1", 1))
+
+
+def test_unknown_duration_delivers_before_end_and_preserves_every_raw_byte(tmp_path):
+    delivered = threading.Event()
+    seen = []
+
+    def consume(value):
+        seen.append(value)
+        delivered.set()
+
+    with harness(tmp_path, consume=consume) as (sock, receiver, run, _, _):
+        hello(sock)
+        original = event(payload=b"x" * (wire.MAX_FRAGMENT + 19))
+        send_event(sock, original)
+        assert delivered.wait(1)
+        assert receiver.thread.is_alive() and not run.mailbox.done
+        assert seen == [original]  # Includes int64 values above JS exact-number range.
+        sock.sendall(wire.terminal_packet(start()))
+        assert receiver.join()
+        assert receiver.terminal == "end" and receiver.error is None
+        assert receiver.counts["fragments"] == 2
+        assert receiver.snapshot()["incomplete_observations"] == 0
+        assert run.mailbox.bytes == 0
+        assert (
+            run.mailbox.peak_bytes
+            <= 2 * len(original.payload) + 3 * wire.MAX_HEADER + wire.PREFIX.size
+        )
+
+
+@pytest.mark.parametrize("value", [0, 1, (1 << 53) + 1, (1 << 64) - 1])
+def test_uint64_roundtrip(value):
+    assert wire.uint64(wire.decimal(value)) == value
+
+
+@pytest.mark.parametrize("value", [1, True, "-1", "01", "1.0", str(1 << 64), "1e6"])
+def test_uint64_rejects_lossy_or_ambiguous_projection(value):
+    with pytest.raises(wire.StreamWireError):
+        wire.uint64(value)
+
+
+def test_duplicate_json_fields_and_unbounded_headers_fail_before_body(tmp_path):
+    with pytest.raises(wire.StreamWireError):
+        wire.parse_header(bytearray(b'{"binding":"a","binding":"b"}'))
+    with harness(tmp_path) as (sock, receiver, run, _, _):
+        sock.sendall(wire.PREFIX.pack(wire.MAGIC, wire.OPEN, wire.MAX_HEADER + 1))
+        assert receiver.join()
+        assert "header size" in receiver.error
+        assert run.state == GraphState.RUNNING  # Unbound peer cannot stop owner.
+        assert run.mailbox.bytes == 0
+
+
+@pytest.mark.parametrize(
+    "field,value", [("epoch_id", "old"), ("lease_generation", 2), ("image_sha256", "f" * 64)]
+)
+def test_wrong_handshake_does_not_cancel_current_owner(tmp_path, field, value):
+    with harness(tmp_path) as (sock, receiver, run, _, _):
+        sock.sendall(wire.open_packet(replace(start(), **{field: value}), "capture-1", 1))
+        assert receiver.join()
+        assert not receiver.opened and run.state == GraphState.RUNNING
+        run.check_current(start())
+
+
+@pytest.mark.parametrize(
+    "mutation,expected",
+    [
+        (lambda h: h.update(binding="e" * 64), "another StreamStart"),
+        (lambda h: h.update(total_bytes=2 * wire.MAX_FRAGMENT + 1), "length"),
+        (lambda h: h.update(fragment_bytes=True), "length"),
+        (lambda h: h.update(offset=1), "length"),
+        (lambda h: h.update(payload_sha256="e" * 64), "observation integrity"),
+        (lambda h: h.update(fragment_sha256="e" * 64), "fragment integrity"),
+        (lambda h: h["event"].update(session_generation="2"), "acquisition"),
+        (lambda h: h["event"].update(received_monotonic_ns=9007199254740993), "decimal string"),
+    ],
+)
+def test_invalid_fragments_fail_closed_without_delivery(tmp_path, mutation, expected):
+    with harness(tmp_path) as (sock, receiver, run, received, _):
+        hello(sock)
+        header, payload = next(wire.event_packets(start(), event()))
+        value = json.loads(header[wire.PREFIX.size :])
+        mutation(value)
+        sock.sendall(wire.packet(wire.FRAGMENT, value) + payload)
+        assert receiver.join()
+        assert expected in receiver.error and not received
+        assert run.state == GraphState.STOPPING and run.mailbox.bytes == 0
+
+
+@pytest.mark.parametrize("case", ["truncated", "interleaved", "end", "timeout"])
+def test_incomplete_reassembly_has_terminal_accounting_and_releases_bytes(tmp_path, case):
+    with harness(tmp_path, fragment_timeout=0.05) as (sock, receiver, run, received, _):
+        hello(sock)
+        pieces = list(wire.event_packets(start(), event(payload=b"x" * (wire.MAX_FRAGMENT + 1))))
+        sock.sendall(pieces[0][0])
+        sock.sendall(pieces[0][1])
+        if case == "truncated":
+            sock.shutdown(socket.SHUT_WR)
+        elif case == "interleaved":
+            send_event(sock, event(seq=2))
+        elif case == "end":
+            sock.sendall(wire.terminal_packet(start()))
+        assert receiver.join()
+        assert receiver.terminal == "failed" and not received
+        assert receiver.snapshot()["incomplete_observations"] == 1
+        assert run.mailbox.bytes == 0 and run.mailbox.quiescent
+
+
+def test_raw_and_decoded_work_share_one_budget(tmp_path):
+    # Small bound proves rejection without creating large local pressure.
+    with harness(tmp_path, byte_limit=3 * wire.MAX_HEADER + 100) as (
+        sock,
+        receiver,
+        run,
+        received,
+        _,
+    ):
+        active = {"sequence": 1, "payload_bytes": 85}
+        assert run.admit(start(), active)
+        assert run.mailbox.take() is active
+        hello(sock)
+        send_event(sock, event())  # Needs 12 extra bytes, only 6 remain.
+        assert receiver.join()
+        assert "byte budget" in receiver.error and not received
+        assert run.mailbox.bytes == 85 and not run.mailbox.quiescent
+        run.mailbox.release(active)
+
+
+def test_stop_cannot_release_borrowed_payload_until_callback_exits(tmp_path):
+    entered, leave = threading.Event(), threading.Event()
+
+    def consume(_value):
+        entered.set()
+        assert leave.wait(2)
+
+    with harness(tmp_path, consume=consume) as (sock, receiver, run, _, _):
+        hello(sock)
+        send_event(sock, event())
+        assert entered.wait(1)
+        try:
+            assert not run.close("cancelled")
+            assert run.mailbox.bytes > 0 and not run.lease.released
+        finally:
+            leave.set()
+        assert receiver.join() and run.mailbox.bytes == 0
+        assert run.close() and run.lease.released
+
+
+def test_lease_expiry_wakes_idle_receiver_without_new_bytes(tmp_path):
+    now = [100]
+    with harness(tmp_path, clock=lambda: now[0]) as (sock, receiver, run, _, _):
+        hello(sock)
+        now[0] += 3_000_000_000
+        assert receiver.join()
+        assert run.stop_event.is_set() and run.reason == "lease-lost"
+        assert run.mailbox.bytes == 0
+
+
+def test_gaps_reset_camera_codec_and_cancel_is_not_successful_end(tmp_path):
+    with harness(tmp_path) as (sock, receiver, run, received, notices):
+        hello(sock)
+        send_event(sock, event(1, modality="camera-init"))
+        send_event(sock, event(2, modality="camera-frame"))
+        sock.sendall(
+            wire.gap_packet(start(), modality="camera-frame", reason="source-gap", count=1)
+        )
+        send_event(sock, event(4, modality="camera-init"))
+        send_event(sock, event(5, modality="camera-frame"))
+        sock.sendall(wire.terminal_packet(start(), cancel=True))
+        assert receiver.join()
+        assert receiver.terminal == "cancelled" and len(received) == 4
+        assert len(notices) == 1 and receiver.counts["declared_gap_observations"] == 1
+        assert receiver.counts["ingress_sequence_gaps"] == 1
+        assert run.reason == "cancelled"
+
+
+@pytest.mark.parametrize("case", ["no-init", "duplicate", "camera-gap", "clock-regression"])
+def test_source_order_and_codec_prerequisites_are_enforced(tmp_path, case):
+    with harness(tmp_path) as (sock, receiver, _, received, _):
+        hello(sock)
+        if case != "no-init":
+            send_event(sock, event(1, modality="camera-init"))
+            send_event(sock, event(2, modality="camera-frame"))
+        bad = event(
+            2 if case == "duplicate" else 3,
+            modality="camera-frame",
+            source_sequence=4 if case == "camera-gap" else None,
+        )
+        if case == "clock-regression":
+            bad = replace(bad, received_monotonic_ns=1)
+        send_event(sock, bad)
+        assert receiver.join() and receiver.terminal == "failed"
+        assert len(received) == (0 if case == "no-init" else 2)
+
+
+def test_external_reservation_is_not_freed_by_cancel_and_cannot_double_release():
+    queue = StreamMailbox(byte_limit=100)
+    raw = queue.reserve_ingress(20)
+    queue.cancel()
+    assert not queue.quiescent and queue.bytes == 20
+    raw.release()
+    assert queue.quiescent and queue.bytes == 0
+    with pytest.raises(ValueError, match="ownership"):
+        raw.release()
+
+
+def test_sender_streams_existing_events_and_explicit_gap_then_end(tmp_path):
+    with harness(tmp_path) as (sock, receiver, run, received, notices):
+        sender = StreamingSender(sock, start(), "capture-1", 1, lambda: run.check_current(start()))
+        sender.send(event())
+        sender.gap(modality="pose", reason="unavailable", count=0)
+        sender.send(event(2, modality="pose"))
+        sender.end()
+        assert receiver.join() and receiver.terminal == "end"
+        assert len(received) == 2 and notices[0]["reason"] == "unavailable"
+        with pytest.raises(wire.StreamWireError, match="closed"):
+            sender.send(event(3))
+
+
+def test_sender_timeout_closes_transport_instead_of_building_backlog():
+    left, right = socket.socketpair()
+    left.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 4096)
+    try:
+        sender = StreamingSender(left, start(), "capture-1", 1, lambda: None, timeout=0.01)
+        with pytest.raises(wire.StreamWireError, match="deadline"):
+            sender.send(event(payload=b"x" * 65536))
+        assert sender.closed and left.fileno() == -1
+    finally:
+        left.close()
+        right.close()