feat(perception): add bounded authenticated gRPC stream transport

2026-09-02 18:11:42 +03:00
parent df0158e1ba
commit 0310592ee3
5 changed files with 996 additions and 0 deletions
@@ -29,6 +29,9 @@ dependencies = [
 [tool.uv.sources]
 missioncore-plugin-sdk = { path = "packages/plugin-sdk", editable = true }

+[project.optional-dependencies]
+perception-stream = ["grpcio>=1.76,<2"]
+
 [project.scripts]
 k1link = "k1link.device_plugins.xgrids_k1.cli:app"
 missioncore-datasets = "k1link.datasets.cli:app"
@@ -0,0 +1,409 @@
+"""TLS gRPC candidate carrying the existing raw-first binary wire unchanged.
+
+One stream per resident profile. The trusted local controller issues one-shot,
+short-lived capabilities AFTER choosing the input epoch. Network code cannot
+activate models, renew leases, reset temporal stores or authorize a reconnect.
+No insecure listener/channel, route upload, background source queue or retry.
+
+The gRPC method uses identity byte serialization (not generated protobufs):
+each request is <=64 KiB of the existing MCI2 byte stream, each response one
+bounded MCR3 envelope. Source/result semantics remain in the existing adapters.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import hmac
+import secrets
+import socket
+import threading
+import time
+from collections.abc import Callable, Iterable
+from contextlib import suppress
+from dataclasses import dataclass, field
+from hashlib import sha256
+from typing import Any
+
+import grpc  # type: ignore[import-untyped]  # Upstream wheel does not ship type stubs.
+
+from k1link.compute.live_perception import LiveIngressEvent
+
+from . import streaming_wire as wire
+from .realtime_contract import StreamStart
+from .streaming_ingress import StreamingIngress
+from .streaming_lifecycle import StreamingLifecycle
+from .streaming_network import (
+    BRIDGE_RESERVATION,
+    INPUT_CHUNK,
+    MAX_REPLY,
+    LatestReplies,
+    decode_reply,
+    encode_reply,
+)
+
+SERVICE = "missioncore.perception.v1.BinaryStream"
+METHOD = f"/{SERVICE}/Exchange"
+OPTIONS = (
+    ("grpc.enable_retries", 0),
+    ("grpc.http2.bdp_probe", 0),
+    # A fixed 1 MiB HTTP/2 receive window, not the default 64 KiB which
+    # serializes large observations into many WAN/SSH round trips. Keep BDP
+    # auto-growth disabled; native buffering remains part of the RSS gate.
+    ("grpc.http2.lookahead_bytes", wire.MAX_FRAGMENT),
+    ("grpc.http2.write_buffer_size", INPUT_CHUNK),
+    ("grpc.max_metadata_size", 4096),
+    ("grpc.max_concurrent_streams", 2),
+)
+
+
+@dataclass(frozen=True)
+class StreamAccess:
+    epoch: StreamStart
+    session_id: str
+    session_generation: int
+    token: str = field(repr=False)
+
+    def metadata(self) -> tuple[tuple[str, str], ...]:
+        return (
+            ("authorization", "Bearer " + self.token),
+            ("x-stream-binding", wire.binding(self.epoch)),
+        )
+
+
+class GrpcStreamEndpoint:
+    def __init__(
+        self,
+        runtime: StreamingLifecycle,
+        consume: Callable[[LiveIngressEvent], None],
+        notice: Callable[[dict[str, Any]], None],
+        *,
+        idle_timeout: float = 2.0,
+        io_timeout: float = 0.25,
+    ) -> None:
+        if runtime.continuity is None:
+            raise ValueError("external stream requires a resumable resident runtime")
+        if not 0 < idle_timeout <= 30 or not 0 < io_timeout <= 2:
+            raise ValueError("invalid bounded network timeouts")
+        self.runtime, self.consume, self.notice = runtime, consume, notice
+        self.idle_timeout, self.io_timeout = idle_timeout, io_timeout
+        self.lock = threading.Lock()
+        self.grant: tuple[StreamAccess, bytes, float] | None = None
+        self.active: tuple[StreamStart, LatestReplies] | None = None
+        self.last: dict[str, Any] | None = None
+        self._quarantine_thread: threading.Thread | None = None
+
+    def issue(self, epoch: StreamStart, session_id: str, session_generation: int) -> StreamAccess:
+        """Local control plane only. Delivery of this secret is outside data-plane."""
+        wire.identifier(session_id)
+        wire.decimal(session_generation)
+        if session_generation < 1:
+            raise ValueError("invalid acquisition generation")
+        self.runtime.check_input(epoch, synchronizing=True)
+        with self.lock:
+            if self.active is not None:
+                raise wire.StreamWireError("previous network stream has not drained")
+            access = StreamAccess(epoch, session_id, session_generation, secrets.token_hex(32))
+            # Do not retain the plaintext secret on the endpoint.
+            self.grant = (
+                StreamAccess(epoch, session_id, session_generation, ""),
+                sha256(access.token.encode()).digest(),
+                time.monotonic() + 30,
+            )
+            return access
+
+    def _claim(self, metadata: Iterable[tuple[str, str | bytes]]) -> StreamAccess:
+        values: dict[str, str | bytes] = {}
+        for key, value in metadata:
+            if key in ("authorization", "x-stream-binding"):
+                if key in values:
+                    raise wire.StreamWireError("stream authorization rejected")
+                values[key] = value
+        token = values.get("authorization")
+        if not isinstance(token, str) or len(token) != 71 or not token.startswith("Bearer "):
+            raise wire.StreamWireError("stream authorization rejected")
+        with self.lock:
+            grant = self.grant
+            if (
+                grant is None
+                or self.active is not None
+                or time.monotonic() >= grant[2]
+                or values.get("x-stream-binding") != wire.binding(grant[0].epoch)
+                or not hmac.compare_digest(sha256(token[7:].encode()).digest(), grant[1])
+            ):
+                raise wire.StreamWireError("stream authorization rejected")
+            self.runtime.check_input(grant[0].epoch, synchronizing=True)
+            self.grant = None
+            self.active = (grant[0].epoch, LatestReplies())
+            return grant[0]
+
+    def publish(self, epoch: StreamStart, sequence: int, payload: bytes) -> bool:
+        self.runtime.check_input(epoch)
+        with self.lock:
+            if self.active is None or self.active[0] != epoch:
+                return False
+            return self.active[1].put(sequence, payload)
+
+    def finish_results(self, epoch: StreamStart) -> None:
+        """Graph controller signals output EOF AFTER all admitted work is accounted."""
+        with self.lock:
+            if self.active is not None and self.active[0] == epoch:
+                self.active[1].finish()
+
+    async def serve(self, address: str, *, certificate: bytes, private_key: bytes) -> Any:
+        if not certificate or not private_key:
+            raise ValueError("TLS identity is required")
+        server = grpc.aio.server(
+            options=OPTIONS
+            + (
+                ("grpc.max_receive_message_length", INPUT_CHUNK),
+                ("grpc.max_send_message_length", MAX_REPLY),
+            ),
+            maximum_concurrent_rpcs=2,
+            compression=grpc.Compression.NoCompression,
+        )
+        server.add_generic_rpc_handlers(
+            (
+                grpc.method_handlers_generic_handler(
+                    SERVICE, {"Exchange": grpc.stream_stream_rpc_method_handler(self.exchange)}
+                ),
+            )
+        )
+        port = server.add_secure_port(
+            address, grpc.ssl_server_credentials(((private_key, certificate),))
+        )
+        if port == 0:
+            raise RuntimeError("TLS stream listener failed to bind")
+        await server.start()
+        return server, port
+
+    async def exchange(self, unused_iterator: Any, context: Any) -> None:
+        if b"ssl" not in context.auth_context().get("transport_security_type", ()):
+            await context.abort(grpc.StatusCode.UNAUTHENTICATED, "TLS required")
+        try:
+            access = self._claim(context.invocation_metadata())
+        except (ValueError, RuntimeError):
+            await context.abort(grpc.StatusCode.UNAUTHENTICATED, "stream authorization rejected")
+            return
+        assert self.active is not None
+        outbox = self.active[1]
+        left = right = receiver = reservation = None
+        tasks: list[asyncio.Task[None]] = []
+        try:
+            reservation = self.runtime.mailbox.reserve_ingress(BRIDGE_RESERVATION)
+            left, right = socket.socketpair()
+            for sock in (left, right):
+                sock.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 32 * 1024)
+                sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 32 * 1024)
+                sock.setblocking(False)
+            receiver = StreamingIngress(
+                right,
+                self.runtime,
+                access.session_id,
+                access.session_generation,
+                self.consume,
+                self.notice,
+                input_epoch=access.epoch,
+                fragment_timeout=self.io_timeout,
+                idle_timeout=self.idle_timeout,
+            )
+            receiver.start()
+            await context.send_initial_metadata((("x-stream-binding", wire.binding(access.epoch)),))
+
+            async def read_input() -> None:
+                while True:
+                    async with asyncio.timeout(self.idle_timeout):
+                        raw = await context.read()
+                    if raw is grpc.aio.EOF:
+                        # Explicit wire End can already have closed the peer.
+                        with suppress(OSError):
+                            left.shutdown(socket.SHUT_WR)
+                        return
+                    if not isinstance(raw, bytes) or not 0 < len(raw) <= INPUT_CHUNK:
+                        raise wire.StreamWireError("invalid network chunk")
+                    async with asyncio.timeout(self.io_timeout):
+                        await asyncio.get_running_loop().sock_sendall(left, raw)
+
+            async def write_output() -> None:
+                while True:
+                    if not receiver.thread.is_alive() and receiver.terminal != "end":
+                        raise ConnectionError("input interrupted or rejected")
+                    self.runtime.check_input(access.epoch, synchronizing=True)
+                    item = outbox.take()
+                    if item is not None:
+                        self.runtime.check_input(access.epoch)
+                        raw = encode_reply(access.epoch, *item)
+                        async with asyncio.timeout(self.io_timeout):
+                            await context.write(raw)
+                        del raw
+                        item = None
+                    elif receiver.terminal == "end" and outbox.drained():
+                        return
+                    else:
+                        await asyncio.sleep(0.01)
+
+            tasks = [asyncio.create_task(read_input()), asyncio.create_task(write_output())]
+            await asyncio.gather(*tasks)
+        except (TimeoutError, OSError, ValueError, RuntimeError):
+            try:
+                # Even error-status delivery can block behind an unread result.
+                # It must never delay closing ingress/releasing local buffers.
+                async with asyncio.timeout(self.io_timeout):
+                    await context.abort(
+                        grpc.StatusCode.UNAVAILABLE,
+                        "stream interrupted; fresh controller grant required",
+                    )
+            except TimeoutError:
+                context.set_code(grpc.StatusCode.UNAVAILABLE)
+            except Exception as exc:
+                # A blocked peer can cancel the HTTP/2 call while the server is
+                # sending its final status. Cleanup still owns the real outcome.
+                if not isinstance(exc, grpc.RpcError):
+                    raise
+        finally:
+            for task in tasks:
+                task.cancel()
+            await asyncio.gather(*tasks, return_exceptions=True)
+            if left is not None:
+                left.close()
+            if receiver is not None and receiver.thread.ident is not None:
+                deadline = time.monotonic() + 1
+                while receiver.thread.is_alive() and time.monotonic() < deadline:
+                    await asyncio.sleep(0.01)
+            elif right is not None:
+                right.close()
+            outbox.close()
+            drained = receiver is None or not receiver.thread.is_alive()
+            if drained:
+                self._complete(receiver, reservation, outbox)
+            else:
+                self._quarantine_thread = threading.Thread(
+                    target=self._finish_quarantine,
+                    args=(receiver, reservation, outbox),
+                    daemon=True,
+                    name="perception-network-quarantine",
+                )
+                self._quarantine_thread.start()
+            # A hung trusted consumer is quarantined: no replacement grant or
+            # premature reservation release while it still owns raw bytes.
+
+    def _finish_quarantine(
+        self, receiver: StreamingIngress, reservation: Any, outbox: LatestReplies
+    ) -> None:
+        receiver.thread.join()  # Bounded to this one already-quarantined stream.
+        self._complete(receiver, reservation, outbox)
+
+    def _complete(
+        self, receiver: StreamingIngress | None, reservation: Any, outbox: LatestReplies
+    ) -> None:
+        if reservation is not None:
+            reservation.release()
+        with self.lock:
+            self.last = {
+                "ingress": receiver.snapshot() if receiver else None,
+                "reply_drops": outbox.dropped,
+                "network_threads_drained": True,
+                "transport": "grpc-tls-binary/v1",
+                "actuation_allowed": False,
+            }
+            self.active = None
+
+
+class GrpcStreamClient:
+    """Explicit async source/result API. Exactly one writer and one reader."""
+
+    def __init__(
+        self, target: str, roots: bytes, access: StreamAccess, *, timeout: float = 0.25
+    ) -> None:
+        if not roots or not 0 < timeout <= 2:
+            raise ValueError("trusted TLS roots and bounded timeout required")
+        self.access, self.timeout = access, timeout
+        self.channel = grpc.aio.secure_channel(
+            target,
+            grpc.ssl_channel_credentials(root_certificates=roots),
+            options=OPTIONS
+            + (
+                ("grpc.max_send_message_length", INPUT_CHUNK),
+                ("grpc.max_receive_message_length", MAX_REPLY),
+            ),
+            compression=grpc.Compression.NoCompression,
+        )
+        self.call = self.channel.stream_stream(METHOD)(
+            metadata=access.metadata(), wait_for_ready=False
+        )
+        self.last_reply = -1
+        self._sending = self._reading = False
+
+    async def open(self) -> None:
+        try:
+            async with asyncio.timeout(2):
+                metadata = await self.call.initial_metadata()
+                if dict(metadata).get("x-stream-binding") != wire.binding(self.access.epoch):
+                    raise wire.StreamWireError("server did not admit current input")
+            await self._send(
+                (
+                    wire.open_packet(
+                        self.access.epoch, self.access.session_id, self.access.session_generation
+                    ),
+                )
+            )
+        except BaseException:
+            self.call.cancel()
+            raise
+
+    async def _send(self, pieces: Iterable[bytes | memoryview]) -> None:
+        if self._sending:
+            self.call.cancel()
+            raise wire.StreamWireError("one source writer per stream")
+        self._sending = True
+        try:
+            async with asyncio.timeout(self.timeout):
+                for piece in pieces:
+                    for offset in range(0, len(piece), INPUT_CHUNK):
+                        await self.call.write(bytes(piece[offset : offset + INPUT_CHUNK]))
+        except BaseException:
+            self.call.cancel()
+            raise
+        finally:
+            self._sending = False
+
+    async def send(self, event: LiveIngressEvent) -> None:
+        if (event.session_id, event.session_generation) != (
+            self.access.session_id,
+            self.access.session_generation,
+        ):
+            self.call.cancel()
+            raise wire.StreamWireError("source acquisition binding changed")
+        await self._send(
+            piece for pair in wire.event_packets(self.access.epoch, event) for piece in pair
+        )
+
+    async def end(self) -> None:
+        await self._send((wire.terminal_packet(self.access.epoch),))
+        await self.call.done_writing()
+
+    async def receive(self, *, timeout: float = 2) -> tuple[int, bytes] | None:
+        if not 0 < timeout <= 30:
+            raise ValueError("invalid bounded result deadline")
+        if self._reading:
+            self.call.cancel()
+            raise wire.StreamWireError("one result reader per stream")
+        self._reading = True
+        try:
+            async with asyncio.timeout(timeout):
+                raw = await self.call.read()
+            if raw is grpc.aio.EOF:
+                return None
+            sequence, payload = decode_reply(self.access.epoch, raw)
+            if sequence <= self.last_reply:
+                raise wire.StreamWireError("network result sequence regressed")
+            self.last_reply = sequence
+            return sequence, payload
+        except BaseException:
+            self.call.cancel()
+            raise
+        finally:
+            self._reading = False
+
+    async def close(self) -> None:
+        self.call.cancel()
+        await self.channel.close()
@@ -0,0 +1,93 @@
+"""Bounded network envelopes; no model, acquisition or ownership authority.
+
+Replies wrap existing domain bytes, not a second scene ontology. A complete
+reply is at most 1 MiB in this candidate; larger outputs need a separately
+reviewed fragmentation adapter, never an automatic enlargement of the queue.
+"""
+
+from __future__ import annotations
+
+import struct
+import threading
+from collections import deque
+from hashlib import sha256
+
+from . import streaming_wire as wire
+from .realtime_contract import StreamStart
+
+INPUT_CHUNK = 64 * 1024
+MAX_RESULT = 1024 * 1024
+REPLY = struct.Struct("!4sQ32s32s")
+MAX_REPLY = MAX_RESULT + REPLY.size
+# Application-owned bytes, including a reply being serialized/written and two
+# pending replies. Native gRPC/HTTP2/socket overhead also needs measured RSS.
+BRIDGE_RESERVATION = 4 * MAX_REPLY + 4 * INPUT_CHUNK
+
+
+def encode_reply(epoch: StreamStart, sequence: int, payload: bytes) -> bytes:
+    wire.decimal(sequence)
+    if type(payload) is not bytes or not 0 < len(payload) <= MAX_RESULT:
+        raise wire.StreamWireError("result exceeds bounded binary envelope")
+    return (
+        REPLY.pack(b"MCR3", sequence, bytes.fromhex(wire.binding(epoch)), sha256(payload).digest())
+        + payload
+    )
+
+
+def decode_reply(epoch: StreamStart, raw: bytes) -> tuple[int, bytes]:
+    if not REPLY.size < len(raw) <= MAX_REPLY:
+        raise wire.StreamWireError("invalid reply length")
+    magic, sequence, binding, digest = REPLY.unpack_from(raw)
+    payload = raw[REPLY.size :]
+    if (
+        magic != b"MCR3"
+        or binding.hex() != wire.binding(epoch)
+        or sha256(payload).digest() != digest
+    ):
+        raise wire.StreamWireError("reply identity or integrity mismatch")
+    return sequence, payload
+
+
+class LatestReplies:
+    """Two pending immutable results, one consumer, no event-loop callback queue."""
+
+    def __init__(self) -> None:
+        self.lock = threading.Lock()
+        self.pending: deque[tuple[int, bytes]] = deque()
+        self.last_sequence = -1
+        self.dropped = 0
+        self.finished = self.closed = False
+
+    def put(self, sequence: int, payload: bytes) -> bool:
+        wire.decimal(sequence)
+        if type(payload) is not bytes or not 0 < len(payload) <= MAX_RESULT:
+            raise wire.StreamWireError("result exceeds bounded binary envelope")
+        with self.lock:
+            if self.closed or self.finished:
+                return False
+            if sequence <= self.last_sequence:
+                raise wire.StreamWireError("result sequence regressed")
+            self.last_sequence = sequence
+            if len(self.pending) == 2:
+                self.pending.popleft()
+                self.dropped += 1
+            self.pending.append((sequence, payload))
+            return True
+
+    def take(self) -> tuple[int, bytes] | None:
+        with self.lock:
+            return self.pending.popleft() if self.pending else None
+
+    def finish(self) -> None:
+        with self.lock:
+            self.finished = True
+
+    def drained(self) -> bool:
+        with self.lock:
+            return self.finished and not self.pending
+
+    def close(self) -> None:
+        with self.lock:
+            self.closed = True
+            self.dropped += len(self.pending)
+            self.pending.clear()
@@ -0,0 +1,463 @@
+"""Small TLS/socket integration checks, not a model benchmark or load test."""
+
+# ruff: noqa: E402 -- optional transport extra must be checked before its import.
+
+import asyncio
+import os
+import subprocess
+import threading
+from contextlib import asynccontextmanager
+from dataclasses import replace
+from hashlib import sha256
+
+import pytest
+
+grpc = pytest.importorskip("grpc", reason="install the perception-stream extra")
+
+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_continuity import ResumeEvidence, StreamSuspended
+from k1link.perception.streaming_grpc import GrpcStreamClient, GrpcStreamEndpoint
+from k1link.perception.streaming_lifecycle import StreamingLifecycle
+from k1link.perception.streaming_network import (
+    MAX_RESULT,
+    LatestReplies,
+    decode_reply,
+    encode_reply,
+)
+from k1link.perception.streaming_queue import StreamMailbox
+
+
+def identity():
+    return StreamStart("run", "source", "worker", "epoch", 1, *(["a" * 64] * 4), "clock", "live")
+
+
+def event(sequence=1, payload=b"synthetic-raw-points", stamp=1_000_000_000):
+    return LiveIngressEvent(
+        sequence,
+        "capture",
+        1,
+        "lidar",
+        "lidar",
+        sequence,
+        1_799_999_999_123_456_789,
+        stamp,
+        payload,
+    )
+
+
+@pytest.fixture(scope="module")
+def tls(tmp_path_factory):
+    root = tmp_path_factory.mktemp("stream-tls")
+    key, cert = root / "key.pem", root / "cert.pem"
+    subprocess.run(
+        [
+            "openssl",
+            "req",
+            "-x509",
+            "-newkey",
+            "rsa:2048",
+            "-nodes",
+            "-days",
+            "1",
+            "-subj",
+            "/CN=localhost",
+            "-keyout",
+            str(key),
+            "-out",
+            str(cert),
+        ],
+        check=True,
+        capture_output=True,
+        timeout=10,
+    )
+    return cert.read_bytes(), key.read_bytes()
+
+
+async def eventually(predicate):
+    async with asyncio.timeout(2):
+        while not predicate():
+            await asyncio.sleep(0.01)
+
+
+@asynccontextmanager
+async def harness(tmp_path, tls, *, consume=None, **options):
+    clock = [1_000_000_000]
+    run = StreamingLifecycle(
+        identity(),
+        tmp_path,
+        StreamMailbox(),
+        threading.Event(),
+        clock_ns=lambda: clock[0],
+        recover_input=True,
+        source_clock_ns=lambda: clock[0],
+    )
+    run.ready()
+    seen = []
+
+    def accept(value):
+        seen.append(value)
+        if consume:
+            consume(value)
+        else:
+            endpoint.publish(
+                run.continuity.epoch, value.ingress_sequence, sha256(value.payload).digest()
+            )
+
+    endpoint = GrpcStreamEndpoint(run, accept, lambda _: None, **options)
+    server, port = await endpoint.serve("localhost:0", certificate=tls[0], private_key=tls[1])
+    clients = []
+
+    def client(access=None, **kwargs):
+        access = access or endpoint.issue(run.continuity.epoch, "capture", 1)
+        result = GrpcStreamClient(f"localhost:{port}", tls[0], access, **kwargs)
+        clients.append(result)
+        return result
+
+    try:
+        yield run, endpoint, client, seen, clock
+    finally:
+        for item in clients:
+            await item.close()
+        await server.stop(0)
+        await eventually(lambda: endpoint.active is None)
+        assert run.close()
+        assert run.mailbox.bytes == 0
+
+
+def test_delivers_before_eof_both_directions_and_explicit_result_drain(tmp_path, tls):
+    async def check():
+        async with harness(tmp_path, tls) as (run, endpoint, client, seen, _):
+            stream = client()
+            await stream.open()
+            original = event()
+            await stream.send(original)
+            assert await stream.receive() == (1, sha256(original.payload).digest())
+            assert seen == [original]
+            assert not run.mailbox.done
+            await stream.end()
+            # Input EOF alone must not discard delayed graph results.
+            await eventually(lambda: run.mailbox.done)
+            assert endpoint.active is not None
+            endpoint.finish_results(identity())
+            assert await stream.receive() is None
+            await eventually(lambda: endpoint.active is None)
+            assert endpoint.last["ingress"]["terminal"] == "end"
+
+    asyncio.run(check())
+
+
+@pytest.mark.parametrize("fault", ["token", "binding", "expired", "duplicate"])
+def test_untrusted_grant_cannot_reserve_memory_or_stop_owner(tmp_path, tls, fault):
+    async def check():
+        async with harness(tmp_path, tls) as (run, endpoint, client, _, _):
+            access = endpoint.issue(identity(), "capture", 1)
+            if fault == "token":
+                bad = replace(access, token="0" * 64)
+            elif fault == "binding":
+                bad = replace(access, epoch=replace(identity(), calibration_sha256="f" * 64))
+            elif fault == "expired":
+                a, b, _ = endpoint.grant
+                endpoint.grant = (a, b, 0)
+                bad = access
+            else:
+                with pytest.raises(ValueError):
+                    endpoint._claim((*access.metadata(), *access.metadata()))
+                bad = replace(access, token="0" * 64)
+            stream = client(bad)
+            with pytest.raises((grpc.aio.AioRpcError, wire.StreamWireError)):
+                await stream.open()
+            assert endpoint.active is None and run.mailbox.bytes == 0
+            assert run.state == GraphState.RUNNING and endpoint.grant is not None
+
+    asyncio.run(check())
+
+
+def test_disconnect_new_controller_epoch_stale_secret_and_local_lease(tmp_path, tls):
+    async def check():
+        async with harness(tmp_path, tls) as (run, endpoint, client, _, clock):
+            stream = client()
+            original_access = stream.access
+            await stream.open()
+            await stream.send(event())
+            await stream.receive()
+            with pytest.raises(ValueError):
+                endpoint.issue(identity(), "capture", 1)
+            await stream.close()
+            await eventually(lambda: endpoint.active is None)
+            assert run.state == GraphState.RUNNING
+            assert run.continuity.phase == "waiting" and run.lease.start == identity()
+            clock[0] += 100_000_000
+            run.renew(identity())  # Local controller, unrelated to network traffic.
+            epoch = run.begin_input(identity())
+            access = endpoint.issue(epoch, "capture", 1)
+            stale = client(original_access)
+            with pytest.raises((grpc.aio.AioRpcError, wire.StreamWireError)):
+                await stale.open()
+            assert run.continuity.epoch == epoch and run.state == GraphState.RUNNING
+            resumed = client(access)
+            await resumed.open()
+            # Transport does NOT invent a decoded keyframe/causal sensor proof.
+            assert run.continuity.phase == "synchronizing"
+            with pytest.raises(StreamSuspended):
+                endpoint.publish(epoch, 1, b"premature")
+            run.resume_input(epoch, ResumeEvidence(*([clock[0]] * 4), True), lambda: None)
+            await resumed.send(event(2, stamp=clock[0]))
+            assert (await resumed.receive())[0] == 2
+            with pytest.raises(StreamSuspended):
+                endpoint.publish(identity(), 3, b"obsolete")
+            await resumed.end()
+            endpoint.finish_results(epoch)
+            assert await resumed.receive() is None
+
+    asyncio.run(check())
+
+
+def test_input_idle_deadline_pauses_without_source_eof(tmp_path, tls):
+    async def check():
+        async with harness(tmp_path, tls, idle_timeout=0.15) as (run, endpoint, client, _, _):
+            stream = client()
+            await stream.open()
+            await stream.send(event())
+            await stream.receive()
+            await eventually(lambda: endpoint.active is None)
+            assert run.state == GraphState.RUNNING
+            assert run.continuity.phase == "waiting" and not run.mailbox.done
+
+    asyncio.run(check())
+
+
+def test_rpc_half_close_without_wire_end_is_not_success(tmp_path, tls):
+    async def check():
+        async with harness(tmp_path, tls) as (run, endpoint, client, _, _):
+            stream = client()
+            await stream.open()
+            await stream.send(event())
+            await stream.receive()
+            await stream.call.done_writing()
+            with pytest.raises(grpc.aio.AioRpcError):
+                await stream.receive()
+            await eventually(lambda: endpoint.active is None)
+            assert run.continuity.phase == "waiting" and not run.mailbox.done
+
+    asyncio.run(check())
+
+
+def test_authorized_malformed_wire_remains_fatal(tmp_path, tls):
+    async def check():
+        async with harness(tmp_path, tls) as (run, endpoint, client, _, _):
+            stream = client()
+            await stream.open()
+            await stream.send(event())
+            await stream.receive()
+            await stream.call.write(wire.PREFIX.pack(b"BAD!", wire.OPEN, 1) + b"x")
+            with pytest.raises(grpc.aio.AioRpcError):
+                await stream.receive()
+            await eventually(lambda: endpoint.active is None)
+            assert run.state == GraphState.STOPPING
+
+    asyncio.run(check())
+
+
+def test_oversized_rpc_rejected_before_domain_consumer(tmp_path, tls):
+    async def check():
+        async with harness(tmp_path, tls) as (run, endpoint, client, seen, _):
+            stream = client()
+            await stream.open()
+            with pytest.raises(grpc.aio.AioRpcError):
+                await stream.call.write(b"x" * (64 * 1024 + 1))
+                await stream.receive()
+            assert seen == []
+
+    asyncio.run(check())
+
+
+def test_latest_reply_bounds_no_hidden_queue():
+    queue = LatestReplies()
+    for i in range(8):
+        assert queue.put(i, b"result")
+    assert len(queue.pending) == 2 and queue.dropped == 6
+    assert queue.take() == (6, b"result")
+    assert queue.take() == (7, b"result")
+    with pytest.raises(ValueError):
+        queue.put(7, b"regressed")
+    queue.finish()
+    assert not queue.put(8, b"late") and queue.drained()
+
+
+@pytest.mark.parametrize("payload", [b"", b"x" * (MAX_RESULT + 1), bytearray(b"x")])
+def test_result_size_type_rejected_before_queue(payload):
+    with pytest.raises(ValueError):
+        LatestReplies().put(0, payload)
+
+
+def test_reply_digest_epoch_and_exact_uint64():
+    raw = encode_reply(identity(), (1 << 53) + 1, b"existing-domain-wire")
+    assert decode_reply(identity(), raw) == ((1 << 53) + 1, b"existing-domain-wire")
+    for altered in (raw[:-1] + b"?", raw[:20]):
+        with pytest.raises(ValueError):
+            decode_reply(identity(), altered)
+    with pytest.raises(ValueError):
+        decode_reply(replace(identity(), epoch_id="another"), raw)
+
+
+def test_tls_name_verification_is_not_disabled(tmp_path, tls):
+    async def check():
+        async with harness(tmp_path, tls) as (run, endpoint, _, seen, _):
+            access = endpoint.issue(identity(), "capture", 1)
+            server, port = await endpoint.serve(
+                "127.0.0.1:0", certificate=tls[0], private_key=tls[1]
+            )
+            stream = GrpcStreamClient(f"127.0.0.1:{port}", tls[0], access)
+            try:
+                # The certificate is for localhost, not this numeric address.
+                with pytest.raises((grpc.aio.AioRpcError, wire.StreamWireError)):
+                    await stream.open()
+                assert endpoint.active is None and not seen and run.mailbox.bytes == 0
+            finally:
+                await stream.close()
+                await server.stop(0)
+
+    asyncio.run(check())
+
+
+def test_slow_result_sink_times_out_without_killing_resident_runtime(tmp_path):
+    class Aborted(Exception):
+        pass
+
+    async def check():
+        run = StreamingLifecycle(
+            identity(),
+            tmp_path,
+            StreamMailbox(),
+            threading.Event(),
+            recover_input=True,
+            source_clock_ns=lambda: 1_000_000_000,
+        )
+        run.ready()
+        endpoint = GrpcStreamEndpoint(
+            run,
+            lambda e: endpoint.publish(identity(), e.ingress_sequence, b"result"),
+            lambda _: None,
+            io_timeout=0.05,
+        )
+        access = endpoint.issue(identity(), "capture", 1)
+        packets = [wire.open_packet(identity(), "capture", 1)]
+        packets += [
+            bytes(piece) for pair in wire.event_packets(identity(), event()) for piece in pair
+        ]
+
+        class Context:
+            def auth_context(self):
+                return {"transport_security_type": (b"ssl",)}
+
+            def invocation_metadata(self):
+                return access.metadata()
+
+            async def send_initial_metadata(self, _):
+                pass
+
+            async def read(self):
+                if packets:
+                    return packets.pop(0)
+                await asyncio.Event().wait()
+
+            async def write(self, _):
+                await asyncio.Event().wait()
+
+            async def abort(self, *_):
+                raise Aborted()
+
+        try:
+            with pytest.raises(Aborted):
+                async with asyncio.timeout(1):
+                    await endpoint.exchange(None, Context())
+            assert run.state == GraphState.RUNNING and run.continuity.phase == "waiting"
+            assert endpoint.active is None and run.mailbox.bytes == 0
+        finally:
+            assert run.close()
+
+    asyncio.run(check())
+
+
+def test_slow_trusted_callback_quarantines_then_releases_without_replacing_owner(tmp_path, tls):
+    entered, leave = threading.Event(), threading.Event()
+
+    def consume(_):
+        entered.set()
+        leave.wait(3)
+
+    async def check():
+        async with harness(tmp_path, tls, consume=consume) as (run, endpoint, client, _, _):
+            stream = client()
+            try:
+                await stream.open()
+                await stream.send(event())
+                await eventually(entered.is_set)
+                await stream.close()
+                await eventually(lambda: endpoint._quarantine_thread is not None)
+                assert endpoint.active is not None and run.mailbox.bytes > 0
+                with pytest.raises((ValueError, StreamSuspended)):
+                    endpoint.issue(identity(), "capture", 1)
+            finally:
+                leave.set()
+            await eventually(lambda: endpoint.active is None)
+            assert run.state == GraphState.RUNNING and run.mailbox.bytes == 0
+            await eventually(lambda: not endpoint._quarantine_thread.is_alive())
+
+    asyncio.run(check())
+
+
+@pytest.mark.skipif(
+    os.environ.get("NDC_STREAM_WORKER_PROBE") != "1", reason="Worker-only bounded probe"
+)
+def test_worker_maximum_result_crosses_tls_intact(tmp_path, tls):
+    payload = b"r" * MAX_RESULT
+
+    async def check():
+        async with harness(
+            tmp_path,
+            tls,
+            consume=lambda e: endpoint.publish(identity(), e.ingress_sequence, payload),
+        ) as (_, endpoint, client, _, _):
+            stream = client()
+            await stream.open()
+            await stream.send(event())
+            assert await stream.receive() == (1, payload)
+            await stream.end()
+            endpoint.finish_results(identity())
+            assert await stream.receive() is None
+
+    asyncio.run(check())
+
+
+@pytest.mark.skipif(
+    os.environ.get("NDC_STREAM_WORKER_PROBE") != "1", reason="Worker-only bounded probe"
+)
+def test_worker_actual_grpc_slow_reader_releases_stream_not_runtime(tmp_path, tls):
+    payload = b"r" * MAX_RESULT
+
+    async def check():
+        async with harness(
+            tmp_path,
+            tls,
+            consume=lambda e: endpoint.publish(identity(), e.ingress_sequence, payload),
+            io_timeout=0.1,
+        ) as (run, endpoint, client, _, _):
+            stream = client()
+            await stream.open()
+            # At most 24 tiny inputs; server outputs at most 24 x 1 MiB on Worker.
+            # Never read results: real gRPC/TLS flow control must reach its bound.
+            for sequence in range(1, 25):
+                try:
+                    await stream.send(event(sequence))
+                except grpc.aio.AioRpcError:
+                    break
+                await asyncio.sleep(0.02)
+            await eventually(lambda: endpoint.active is None)
+            assert run.state == GraphState.RUNNING and run.continuity.phase == "waiting"
+            assert run.mailbox.bytes == 0 and endpoint.last["reply_drops"] > 0
+
+    asyncio.run(check())
@@ -146,6 +146,27 @@ wheels = [
    { url = "https://files.pythonhosted.org/packages/c7/e1/8ccb4a985c5baf82947e48cff18483c2125ea11e55e8a28950730ab6c065/foxglove_sdk-0.25.3-cp310-abi3-win_amd64.whl", hash = "sha256:ac881ae307ba432766e6141d9098ced2059838226d225fbeb20de1c57d782a9f", size = 16496466, upload-time = "2026-06-25T00:24:24.906Z" },
 ]

+[[package]]
+name = "grpcio"
+version = "1.83.1"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "typing-extensions" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/e2/b1/46539f5050d7c316a13396d185451f95084a74ddc68b12d818595bef0377/grpcio-1.83.1.tar.gz", hash = "sha256:9cee6fcbf2eb57c4b49451787bfa87be8efc1ca02a0b327dd4b54d44502e362b", size = 13445033, upload-time = "2026-08-28T07:09:11.464Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/85/9e/a3ba13e08bbee5bf6e57597dfe4823961fd7e94c0b8afe3a4bb7dca639f3/grpcio-1.83.1-cp312-cp312-linux_armv7l.whl", hash = "sha256:5acd14c6ddf047de62cbf8745b11103ea91abbf57d1b8edd5395ccd9fcd13abb", size = 6303170, upload-time = "2026-08-28T07:08:08.188Z" },
+    { url = "https://files.pythonhosted.org/packages/1c/ae/65ce56a2527faa17d02cba4c2231c74047ad898be339486ba87f093bfb66/grpcio-1.83.1-cp312-cp312-macosx_11_0_universal2.whl", hash = "sha256:16138031a47b771860a16a975b53087f4fd5bbdbb2c03a188c5d90ad65d2bdae", size = 12165806, upload-time = "2026-08-28T07:08:10.309Z" },
+    { url = "https://files.pythonhosted.org/packages/4e/91/40432480088a2243d360864de072ed5b78c4ebbaabd29c28918f1e1b1454/grpcio-1.83.1-cp312-cp312-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:5ccc26715fd4defca5e129e280dd883b1737b65045ec50ffe22ce42104089519", size = 6872490, upload-time = "2026-08-28T07:08:12.355Z" },
+    { url = "https://files.pythonhosted.org/packages/c8/62/3da2300c8c79fd20a78a8a4bb6251e5068d9af33bc8fd389b98fec35e8a3/grpcio-1.83.1-cp312-cp312-manylinux2014_i686.manylinux_2_17_i686.whl", hash = "sha256:b74f2a1d9ab1dfa3e263ef33d581613679b78d0884babf11671af26e45570ead", size = 7618367, upload-time = "2026-08-28T07:08:14.025Z" },
+    { url = "https://files.pythonhosted.org/packages/bc/19/9fc702e31a631262d7a752fa699f6022821e707fefc8bff49b1550a57729/grpcio-1.83.1-cp312-cp312-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:72578aa07a4008f17521ef52debcc3acfd1e2c5426243bc3ffb56a38bfe610b7", size = 7040936, upload-time = "2026-08-28T07:08:15.963Z" },
+    { url = "https://files.pythonhosted.org/packages/ec/56/95933cc44cba2429765fa065c951dd529e5771b119d9d2481b4646f1d6a5/grpcio-1.83.1-cp312-cp312-musllinux_1_2_aarch64.whl", hash = "sha256:c12e1fc59c6dc26d10d9144453ddc6cbfe4cd4c31e874ed2d0132f88e685eb8b", size = 7573096, upload-time = "2026-08-28T07:08:17.729Z" },
+    { url = "https://files.pythonhosted.org/packages/ac/80/af63359da06b016de48cb111f144703a10043850dafa43ae0a038907b9e8/grpcio-1.83.1-cp312-cp312-musllinux_1_2_i686.whl", hash = "sha256:4910b62f7d12197160bfb7de06d876d64dd12d43483e8292f98f49ca09b628d9", size = 8609442, upload-time = "2026-08-28T07:08:19.777Z" },
+    { url = "https://files.pythonhosted.org/packages/3d/fa/f0586c56bdfb8a7a2adda01e0ac2413447cde3141ab09411a5d5afdcffd3/grpcio-1.83.1-cp312-cp312-musllinux_1_2_x86_64.whl", hash = "sha256:9e703effe3ae779925c82ac24fdb82cf4105e1096810151ed9501c5f34546b9c", size = 7984321, upload-time = "2026-08-28T07:08:22.114Z" },
+    { url = "https://files.pythonhosted.org/packages/25/8a/14ec05669f9eb295801e26c2ea8c561a1b786b0e3557c2c22131165ab010/grpcio-1.83.1-cp312-cp312-win32.whl", hash = "sha256:a2aea8bd6e0a34f12cbaddb7bb70bec836818789fa5c7ab7572c6b745396a2d4", size = 4395604, upload-time = "2026-08-28T07:08:24.08Z" },
+    { url = "https://files.pythonhosted.org/packages/e9/37/8c2f7cc16089e36a3fbacaacc7a3d043912aa0d2dfae5556f6450414ea6e/grpcio-1.83.1-cp312-cp312-win_amd64.whl", hash = "sha256:583bf2e8255040a4a312f9572dfe62a05271437b149550e1a536d5c47d2d1e8a", size = 5161512, upload-time = "2026-08-28T07:08:25.81Z" },
+]
+
 [[package]]
 name = "h11"
 version = "0.16.0"
@@ -336,6 +357,11 @@ dependencies = [
    { name = "uvicorn", extra = ["standard"] },
 ]

+[package.optional-dependencies]
+perception-stream = [
+    { name = "grpcio" },
+]
+
 [package.dev-dependencies]
 dev = [
    { name = "mypy" },
@@ -349,6 +375,7 @@ requires-dist = [
    { name = "bleak", specifier = "==3.0.2" },
    { name = "fastapi", specifier = ">=0.116,<1" },
    { name = "foxglove-sdk", specifier = "==0.25.3" },
+    { name = "grpcio", marker = "extra == 'perception-stream'", specifier = ">=1.76,<2" },
    { name = "httpx", specifier = ">=0.28,<1" },
    { name = "lz4", specifier = ">=4.4,<5" },
    { name = "missioncore-plugin-sdk", editable = "packages/plugin-sdk" },
@@ -360,6 +387,7 @@ requires-dist = [
    { name = "typer", specifier = ">=0.15,<1" },
    { name = "uvicorn", extras = ["standard"], specifier = ">=0.35,<1" },
 ]
+provides-extras = ["perception-stream"]

 [package.metadata.requires-dev]
 dev = [