feat(perception): connect full graph to external streaming source and scene receiver

experiments/perception/worker/streaming_profile_stage1/pilot_binary_bridge.py

@@ -19,7 +19,7 @@ from k1link.perception.streaming_sensors import (
 )
 
 
-class BinaryGraphBridge:
+class BinaryGraphInput:
     def __init__(self, runtime, decoder, source, report, *, reset_temporal=None):
         self.runtime, self.decoder, self.source, self.report = runtime, decoder, source, report
         self.source_zero, self.wall_zero = source.source_zero, source.wall_zero
@@ -33,71 +33,14 @@ class BinaryGraphBridge:
         self.decode_sequence_start = 0
         self.init_timing = None
         self.epoch_reports, self.resumes, self.sync_skipped = [], [], []
-        self.left = right = None
-        try:
-            self.left, right = socket.socketpair()
-            self.receiver = StreamingIngress(
-                right, runtime, "recorded-acquisition", 1, self.consume, self.notice
-            )
-            self.producer = threading.Thread(
-                target=source.run,
-                args=(self.left,),
-                kwargs={"bridge": self} if reset_temporal is not None else {},
-                daemon=True,
-                name="binary-recorded-source",
-            )
-            runtime.track_thread(self.producer)
-        except BaseException:
-            # No threads have started. A failed registration/lease must not
-            # strand the preallocated cache or either end of the socket.
-            for connection in (self.left, right):
-                if connection is not None:
-                    connection.close()
-            self.window.close()
-            raise
-
-    def start(self):
-        self.receiver.start()
-        self.producer.start()
 
     def notice(self, value):
-        # A gap is not a new codec epoch. This profile requires a new run/lease
-        # and fresh decoder state; never continue predictive decoding across it.
-        raise ValueError("binary full profile requires restart after an explicit source gap")
+        raise ValueError("binary full profile requires controller-approved resynchronization")
 
-    def disconnect(self):
-        self.left.close()  # Real socket EOF; independent resident lease is kept.
-        self.runtime.pause_input(self.epoch, "input-disconnected")
-
-    def reconnect(self):
-        """Nonblocking source-side attempt; source clock keeps moving on failure."""
-        try:
-            epoch = self.runtime.begin_input(self.runtime.start)
-        except StreamSuspended:
-            return None
-        self.epoch_reports.append(self.receiver.snapshot())
-        if len(self.epoch_reports) > 256:
-            raise ValueError("bounded pilot reconnect count exceeded")
+    def begin_epoch(self, epoch):
         self.epoch = epoch
         self.decode_sequence_start = None
         self.window.reset()
-        self.left, right = socket.socketpair()
-        try:
-            self.receiver = StreamingIngress(
-                right,
-                self.runtime,
-                "recorded-acquisition",
-                1,
-                self.consume,
-                self.notice,
-                input_epoch=epoch,
-            )
-            self.receiver.start()
-        except BaseException:
-            self.left.close()
-            right.close()
-            raise
-        return self.left
 
     def consume(self, event):
         self.runtime.check_input(self.epoch, synchronizing=True)
@@ -228,6 +171,78 @@ class BinaryGraphBridge:
             if not transferred:
                 reservation.release()
 
+    def record_report(self):
+        self.report.update(
+            resumed_inputs=self.resumes,
+            synchronization_skipped_camera_sequences=self.sync_skipped,
+            decoder_bgr_sha256=self.bgr_hashes,
+            sensor_binding_reasons=dict(self.binding_reasons),
+            preroll_history_only_points=self.preroll,
+        )
+
+
+class BinaryGraphBridge(BinaryGraphInput):
+    def __init__(self, runtime, decoder, source, report, *, reset_temporal=None):
+        super().__init__(runtime, decoder, source, report, reset_temporal=reset_temporal)
+        self.left = right = None
+        try:
+            self.left, right = socket.socketpair()
+            self.receiver = StreamingIngress(
+                right, runtime, "recorded-acquisition", 1, self.consume, self.notice
+            )
+            self.producer = threading.Thread(
+                target=source.run,
+                args=(self.left,),
+                kwargs={"bridge": self} if reset_temporal is not None else {},
+                daemon=True,
+                name="binary-recorded-source",
+            )
+            runtime.track_thread(self.producer)
+        except BaseException:
+            # No threads have started. A failed registration/lease must not
+            # strand the preallocated cache or either end of the socket.
+            for connection in (self.left, right):
+                if connection is not None:
+                    connection.close()
+            self.window.close()
+            raise
+
+    def start(self):
+        self.receiver.start()
+        self.producer.start()
+
+    def disconnect(self):
+        self.left.close()  # Real socket EOF; independent resident lease is kept.
+        self.runtime.pause_input(self.epoch, "input-disconnected")
+
+    def reconnect(self):
+        """Nonblocking source-side attempt; source clock keeps moving on failure."""
+        try:
+            epoch = self.runtime.begin_input(self.runtime.start)
+        except StreamSuspended:
+            return None
+        self.epoch_reports.append(self.receiver.snapshot())
+        if len(self.epoch_reports) > 256:
+            raise ValueError("bounded pilot reconnect count exceeded")
+        self.begin_epoch(epoch)
+        self.left, right = socket.socketpair()
+        try:
+            self.receiver = StreamingIngress(
+                right,
+                self.runtime,
+                "recorded-acquisition",
+                1,
+                self.consume,
+                self.notice,
+                input_epoch=epoch,
+            )
+            self.receiver.start()
+        except BaseException:
+            self.left.close()
+            right.close()
+            raise
+        return self.left
+
     def close(self):
         if self.producer.ident is not None:
             self.producer.join(timeout=2)

experiments/perception/worker/streaming_profile_stage1/pilot_binary_source.py

@@ -45,6 +45,22 @@ def read_member(path, root, maximum):
     return raw
 
 
+def event_bytes(event, root):
+    """The same raw source representation for local IPC and network replay."""
+    if event.channel == "camera":
+        row = event.value
+        raw = read_member(root / row["path"], root, 1024 * 1024)
+        if len(raw) != row["length"] or hashlib.sha256(raw).hexdigest() != row["sha256"]:
+            raise ValueError("camera fragment integrity changed")
+        return "camera-frame", "sensor.camera.right", row["host_epoch_ns"], raw
+    if event.channel == "points":
+        xyz, intensity = event.value
+        raw = struct.pack("<I", len(xyz)) + xyz.tobytes() + intensity.tobytes()
+        return "lidar", "normalized-map-point-increments", 0, raw
+    position, quaternion = event.value
+    return "pose", "normalized-map-from-lidar-pose", 0, position.tobytes() + quaternion.tobytes()
+
+
 class RecordingSource:
     def __init__(self, args, runtime, report):
         self.args, self.runtime, self.report = args, runtime, report
@@ -150,27 +166,7 @@ class RecordingSource:
                             if event.sequence == self.args.frames - 1:
                                 break
                         continue
-                if event.channel == "camera":
-                    row = event.value
-                    raw = read_member(root / row["path"], root, 1024 * 1024)
-                    if (
-                        len(raw) != row["length"]
-                        or hashlib.sha256(raw).hexdigest() != row["sha256"]
-                    ):
-                        raise ValueError("camera fragment integrity changed")
-                    modality, source_id, utc = (
-                        "camera-frame",
-                        "sensor.camera.right",
-                        row["host_epoch_ns"],
-                    )
-                elif event.channel == "points":
-                    xyz, intensity = event.value
-                    raw = struct.pack("<I", len(xyz)) + xyz.tobytes() + intensity.tobytes()
-                    modality, source_id = "lidar", "normalized-map-point-increments"
-                else:
-                    position, quaternion = event.value
-                    raw = position.tobytes() + quaternion.tobytes()
-                    modality, source_id = "pose", "normalized-map-from-lidar-pose"
+                modality, source_id, utc, raw = event_bytes(event, root)
                 seq += 1
                 try:
                     sender.send(

experiments/perception/worker/streaming_profile_stage1/pilot_freshness.py

@@ -33,6 +33,8 @@ def payload_digest(scene, layer):
         "policy": ("policy_actions", "policy_counts"),
     }[layer]
     value = {key: scene[key] for key in fields}
+    if layer == "costmap" and "costmap_grid" in scene:
+        value["costmap_grid"] = scene["costmap_grid"]
     if layer == "costmap" and "costmap_cell_evidence" in scene:
         value["costmap_cell_evidence"] = scene["costmap_cell_evidence"]
         value["costmap_freshness_mode"] = scene["costmap_freshness_mode"]

experiments/perception/worker/streaming_profile_stage1/pilot_grpc_graph.py

@@ -0,0 +1,165 @@
+"""Network input/result adapter around the SAME decoder and causal graph input.
+
+No recording path, model constructor, lease renewal or controller authority is
+accepted from the network. Grant rotation stays in this trusted local adapter.
+"""
+
+import asyncio
+import threading
+import time
+import traceback
+from pathlib import Path
+from types import SimpleNamespace
+
+from pilot_binary_bridge import BinaryGraphInput
+from pilot_network_control import kernel_clock, read_control, write_control
+
+from k1link.perception.streaming_continuity import StreamSuspended
+from k1link.perception.streaming_grpc import GrpcStreamEndpoint
+
+
+class NetworkGraphBridge(BinaryGraphInput):
+    def __init__(
+        self,
+        runtime,
+        decoder,
+        report,
+        *,
+        control,
+        source_status,
+        source_zero,
+        certificate,
+        private_key,
+        address,
+        reset_temporal,
+        start_delay=2,
+    ):
+        self.clock = kernel_clock()
+        source = SimpleNamespace(
+            source_zero=source_zero, wall_zero=time.monotonic_ns() + int(start_delay * 1e9)
+        )
+        super().__init__(runtime, decoder, source, report, reset_temporal=reset_temporal)
+        self.control = Path(control)
+        self.source_status = Path(source_status)
+        self.certificate, self.private_key, self.address = certificate, private_key, address
+        self.endpoint = GrpcStreamEndpoint(runtime, self.consume, self.notice)
+        self.stopping, self.ready = threading.Event(), threading.Event()
+        self.failure = None
+        self.published = []
+        self.thread = threading.Thread(target=self._run, daemon=True, name="full-graph-grpc")
+        try:
+            runtime.track_thread(self.thread)
+        except BaseException:
+            self.window.close()
+            raise
+        report.update(
+            source_zero_ns=source.source_zero,
+            wall_zero_ns=source.wall_zero,
+            source_clock_speed=1.0,
+            source_eof_required=False,
+            full_source_prepass=False,
+            clock_proof=self.clock,
+            accounting_owner="external-source-receiver",
+        )
+
+    def _grant(self):
+        access = self.endpoint.issue(self.epoch, "recorded-acquisition", 1)
+        write_control(
+            self.control / "grant.json",
+            {
+                "epoch": access.epoch.to_dict(),
+                "token": access.token,
+                "source_zero_ns": self.source_zero,
+                "wall_zero_ns": self.wall_zero,
+                "cutoff_ns": self.runtime.continuity.cutoff_ns,
+                "clock": self.clock,
+            },
+        )
+
+    def _run(self):
+        try:
+            asyncio.run(self._serve())
+        except BaseException:
+            self.failure = traceback.format_exc()
+            self.runtime.request_stop("failed")
+        finally:
+            self.ready.set()
+
+    async def _serve(self):
+        server, _ = await self.endpoint.serve(
+            self.address,
+            certificate=Path(self.certificate).read_bytes(),
+            private_key=Path(self.private_key).read_bytes(),
+        )
+        try:
+            self._grant()
+            self.ready.set()
+            while not self.stopping.is_set() and not self.runtime.stop_event.is_set():
+                if self.endpoint.active is None:
+                    ended = self.source_status / "source-end.json"
+                    if ended.exists():
+                        terminal = read_control(ended)
+                        if terminal.get("run_id") != self.runtime.start.run_id:
+                            raise ValueError("external source completion binding mismatch")
+                        if terminal.get("error"):
+                            raise ValueError("external source failed; inspect its bounded report")
+                        self.runtime.mailbox.finish()
+                    elif self.runtime.continuity.phase == "waiting":
+                        try:
+                            epoch = self.runtime.begin_input(self.runtime.start)
+                        except StreamSuspended:
+                            pass
+                        else:
+                            self.epoch_reports.append(self.endpoint.last)
+                            if len(self.epoch_reports) > 16:
+                                raise ValueError("bounded network pilot epoch budget exceeded")
+                            self.begin_epoch(epoch)
+                            self._grant()
+                await asyncio.sleep(0.01)
+        finally:
+            await server.stop(0)
+            deadline = time.monotonic() + 2
+            while self.endpoint.active is not None and time.monotonic() < deadline:
+                await asyncio.sleep(0.01)
+
+    def start(self):
+        self.thread.start()
+        if not self.ready.wait(2) or self.failure:
+            raise RuntimeError("network graph listener failed to become ready")
+
+    def publish(self, epoch, sequence, payload):
+        accepted = self.endpoint.publish(epoch, sequence, payload)
+        self.published.append(
+            {
+                "sequence": sequence,
+                "epoch_id": epoch.epoch_id,
+                "accepted": accepted,
+                "bytes": len(payload),
+            }
+        )
+
+    def finish_results(self):
+        self.endpoint.finish_results(self.epoch)
+        deadline = time.monotonic() + 2
+        while self.endpoint.active is not None and time.monotonic() < deadline:
+            time.sleep(0.01)
+        if self.endpoint.active is not None:
+            raise TimeoutError("bounded result drain did not complete")
+
+    def close(self):
+        self.stopping.set()
+        if self.thread.ident is not None:
+            self.thread.join(timeout=3)
+        self.record_report()
+        self.report.update(
+            network_epochs=self.epoch_reports,
+            last_network_epoch=self.endpoint.last,
+            accepted_camera_sequences=self.accepted,
+            published=self.published,
+            adapter_failure=self.failure,
+        )
+        if not self.thread.is_alive() and self.endpoint.active is None:
+            self.window.close()
+            self.decoder.close()
+            return True
+        return False

experiments/perception/worker/streaming_profile_stage1/pilot_grpc_source.py

@@ -0,0 +1,308 @@
+"""External CPU-only source + real scene receiver for a bounded network pilot.
+
+Reads the existing recording incrementally at original 1x time. No decoder,
+models, GPU lease or whole-source upload. A source gap discards observations;
+reconnect consumes only a new locally-issued grant and future observations.
+"""
+
+import argparse
+import asyncio
+import hashlib
+import json
+import time
+import traceback
+from collections import Counter
+from datetime import UTC, datetime
+from pathlib import Path
+
+import grpc
+from pilot_binary_source import event_bytes, input_gaps, read_member
+from pilot_freshness import assess_receipt, validate_receipt
+from pilot_network_control import read_grant, write_control
+from pilot_source import SensorArchive, camera_events, merged_events
+
+from k1link.compute.live_perception import LiveIngressEvent
+from k1link.perception.streaming_grpc import GrpcStreamClient
+from k1link.perception.streaming_scene_payload import decode_scene_payload
+
+
+def connection_delay(wall_zero_ns, now_ns):
+    # Do not send init while the delayed source start is still farther away
+    # than the stream idle timeout. Connect just before the first release.
+    return max(0, (wall_zero_ns - now_ns - 100_000_000) / 1e9)
+
+
+async def run(args):
+    report = {
+        "started_utc": datetime.now(UTC).isoformat(),
+        "started_monotonic_ns": time.monotonic_ns(),
+        "frames": [],
+        "faults": [],
+        "transport_errors": [],
+        "input_transport": "grpc-tls-binary/v1",
+        "models": 0,
+        "source_clock_speed": 1.0,
+        "full_source_prepass": False,
+        "clock_scope": "two-containers-same-unshifted-Linux-kernel",
+        "error": None,
+        "actuation_allowed": False,
+        "commands_enabled": False,
+    }
+    output = Path(args.output)
+    output.mkdir(parents=True, exist_ok=False)
+    control, root = Path(args.control), Path(args.camera_root)
+    status = Path(args.source_status)
+    grant_path = control / "grant.json"
+    archive = client = reader = access = None
+    source_end = False
+    arrivals, skipped, lags, released = Counter(), Counter(), [], []
+    gaps, next_gap, outage_until = input_gaps(args.input_gap), 0, 0
+    previous_epoch = None
+    clock = None
+
+    async def disconnect():
+        nonlocal client, reader
+        if client is not None:
+            await client.close()
+        if reader is not None:
+            reader.cancel()
+            await asyncio.gather(reader, return_exceptions=True)
+        client = reader = None
+
+    try:
+        deadline = time.monotonic() + 120
+        while not grant_path.exists():
+            if time.monotonic() > deadline:
+                raise TimeoutError("trusted graph grant was not issued after warmup")
+            await asyncio.sleep(0.02)
+        clock, _ = read_grant(grant_path)
+        first = next(camera_events(args.camera_index, 1)).time_ns
+        if clock["source_zero_ns"] != first - 500_000_000:
+            raise ValueError("recording prefix does not match admitted source clock")
+        report.update(
+            source_zero_ns=clock["source_zero_ns"],
+            wall_zero_ns=clock["wall_zero_ns"],
+            clock_proof=clock["clock"],
+        )
+        archive = SensorArchive(Path(args.sensor_archive))
+        with (output / "scenes.jsonl").open("wb") as sink:
+
+            async def receive(stream, epoch):
+                while True:
+                    item = await stream.receive(timeout=5)
+                    if item is None:
+                        return
+                    sequence, payload = item
+                    arrived = time.monotonic_ns()
+                    scene, mask, shape = decode_scene_payload(payload, epoch, sequence)
+                    source_stamp = scene["original_source_ns"]
+                    if type(source_stamp) is not int:
+                        raise ValueError("invalid original source time")
+                    bundle = {
+                        "sequence": sequence,
+                        "time_ns": source_stamp,
+                        "due_ns": clock["wall_zero_ns"] + source_stamp - clock["source_zero_ns"],
+                    }
+                    freshness = validate_receipt(scene, bundle, epoch_id=epoch.epoch_id)
+                    checked_at = time.monotonic_ns()
+                    view, checked = assess_receipt(
+                        scene, freshness, bundle=bundle, now_ns=checked_at
+                    )
+                    consumer_ready = time.monotonic_ns()
+                    if len(report["frames"]) >= args.frames:
+                        raise ValueError("network result diagnostic bound exceeded")
+                    report["frames"].append(
+                        {
+                            "sequence": sequence,
+                            "epoch_id": epoch.epoch_id,
+                            "payload_sha256": hashlib.sha256(payload).hexdigest(),
+                            "payload_bytes": len(payload),
+                            "mask_shape": shape,
+                            "mask_sha256": hashlib.sha256(mask).hexdigest(),
+                            "arrived_monotonic_ns": arrived,
+                            "checked_monotonic_ns": checked_at,
+                            "consumer_ready_monotonic_ns": consumer_ready,
+                            "source_due_to_network_arrival_ms": (arrived - bundle["due_ns"]) / 1e6,
+                            "source_due_to_validated_receipt_ms": (checked_at - bundle["due_ns"])
+                            / 1e6,
+                            "source_due_to_consumer_ready_ms": (consumer_ready - bundle["due_ns"])
+                            / 1e6,
+                            "freshness_at_receipt": checked.to_dict(),
+                            "received_before_source_end": not source_end,
+                            "policy_counts": view["policy_counts"],
+                        }
+                    )
+                    # Evidence write AFTER receipt; source pacing/lag includes its real cost.
+                    sink.write(
+                        json.dumps(scene, allow_nan=False, separators=(",", ":")).encode() + b"\n"
+                    )
+                    del scene, view, mask, payload, item
+
+            async def connect():
+                nonlocal client, reader, access, previous_epoch
+                value, candidate = read_grant(grant_path)
+                if candidate.epoch.epoch_id == previous_epoch:
+                    return False
+                if any(
+                    value[key] != clock[key] for key in ("source_zero_ns", "wall_zero_ns", "clock")
+                ):
+                    raise ValueError("source clock changed between input epochs")
+                access, previous_epoch = candidate, candidate.epoch.epoch_id
+                client = GrpcStreamClient(args.target, Path(args.certificate).read_bytes(), access)
+                await client.open()
+                await client.send(
+                    LiveIngressEvent(
+                        1,
+                        "recorded-acquisition",
+                        1,
+                        "camera-init",
+                        "sensor.camera.right",
+                        0,
+                        0,
+                        max(value["cutoff_ns"], value["source_zero_ns"]),
+                        read_member(root / "init.mp4", root, 65536),
+                    )
+                )
+                reader = asyncio.create_task(receive(client, access.epoch))
+                report.setdefault("input_epochs", []).append(access.epoch.to_dict())
+                if len(report["input_epochs"]) > 16:
+                    raise ValueError("bounded pilot epoch limit")
+                return True
+
+            await asyncio.sleep(connection_delay(clock["wall_zero_ns"], time.monotonic_ns()))
+            report["initial_connect_monotonic_ns"] = time.monotonic_ns()
+            await connect()
+            seq = 1
+            for event in merged_events(archive, args.camera_index, args.frames):
+                if event.time_ns < clock["source_zero_ns"]:
+                    skipped[event.channel + "-prefix"] += 1
+                    continue
+                due = clock["wall_zero_ns"] + event.time_ns - clock["source_zero_ns"]
+                await asyncio.sleep(max(0, (due - time.monotonic_ns()) / 1e9))
+                lag = max(0, time.monotonic_ns() - due) / 1e6
+                arrivals[event.channel] += 1
+                lags.append(lag)
+                if event.channel == "camera":
+                    released.append(event.sequence)
+                try:
+                    if reader is not None and reader.done():
+                        reader.result()  # Invalid payload is fatal; transport interruption is not.
+                        raise ConnectionError("result stream ended before source EOF")
+                    if (
+                        event.channel == "camera"
+                        and next_gap < len(gaps)
+                        and event.sequence >= gaps[next_gap][0]
+                    ):
+                        duration = gaps[next_gap][1]
+                        await disconnect()
+                        outage_until = time.monotonic_ns() + duration * 1_000_000
+                        report["faults"].append(
+                            {
+                                "sequence": event.sequence,
+                                "duration_ms": duration,
+                                "injected_monotonic_ns": time.monotonic_ns(),
+                            }
+                        )
+                        next_gap += 1
+                    if lag > 250:
+                        await disconnect()  # Never catch up a stale predictive camera backlog.
+                    if client is None:
+                        if time.monotonic_ns() >= outage_until and await connect():
+                            seq = 1
+                        skipped[event.channel + "-wait"] += 1
+                        # Current observation predates the new cutoff; no retimestamp/replay.
+                    else:
+                        modality, source_id, utc, raw = event_bytes(event, root)
+                        seq += 1
+                        await client.send(
+                            LiveIngressEvent(
+                                seq,
+                                "recorded-acquisition",
+                                1,
+                                modality,
+                                source_id,
+                                event.sequence,
+                                utc,
+                                event.time_ns,
+                                raw,
+                            )
+                        )
+                        del raw
+                except (grpc.RpcError, TimeoutError, OSError) as exc:
+                    report["transport_errors"].append(
+                        {"sequence": event.sequence, "type": type(exc).__name__}
+                    )
+                    if len(report["transport_errors"]) > 16:
+                        raise ValueError("bounded transport fault budget exceeded") from exc
+                    await disconnect()
+                    skipped[event.channel + "-interrupted"] += 1
+                if event.channel == "camera" and event.sequence == args.frames - 1:
+                    break
+            source_end = True
+            report["end_sent_monotonic_ns"] = time.monotonic_ns()
+            if client is not None:
+                await client.end()
+                await asyncio.wait_for(reader, timeout=5)
+            write_control(
+                status / "source-end.json", {"run_id": access.epoch.run_id, "error": False}
+            )
+    except Exception:
+        report["error"] = traceback.format_exc()
+        if access is not None:
+            write_control(
+                status / "source-end.json", {"run_id": access.epoch.run_id, "error": True}
+            )
+    finally:
+        await disconnect()
+        from run_joint_pilot import distribution
+
+        report.update(
+            arrivals=dict(arrivals),
+            skipped=dict(skipped),
+            released_camera_sequences=released,
+            release_lag_ms=distribution(lags),
+            incremental_reads=archive.counters() if archive else {},
+            ended_monotonic_ns=time.monotonic_ns(),
+        )
+        if archive:
+            archive.close()
+        report["distributions_ms"] = {
+            key: distribution([row[key] for row in report["frames"]])
+            for key in (
+                "source_due_to_network_arrival_ms",
+                "source_due_to_validated_receipt_ms",
+                "source_due_to_consumer_ready_ms",
+            )
+        }
+        (output / "report.json").write_text(json.dumps(report, allow_nan=False, indent=2) + "\n")
+    print(
+        json.dumps(
+            {
+                "frames": len(report["frames"]),
+                "error": report["error"],
+                "distributions_ms": report["distributions_ms"],
+            }
+        ),
+        flush=True,
+    )
+    return 1 if report["error"] else 0
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--target", required=True)
+    parser.add_argument("--control", required=True)
+    parser.add_argument("--source-status", required=True)
+    parser.add_argument("--certificate", required=True)
+    parser.add_argument("--output", required=True)
+    parser.add_argument("--camera-index", default="/camera/index.jsonl")
+    parser.add_argument("--camera-root", default="/camera")
+    parser.add_argument("--sensor-archive", default="/sensor-source.npz")
+    parser.add_argument("--frames", type=int, default=128)
+    parser.add_argument("--input-gap", action="append", default=[])
+    args = parser.parse_args()
+    if not 1 <= args.frames <= 256 or any(
+        seq >= args.frames for seq, _ in input_gaps(args.input_gap)
+    ):
+        parser.error("bounded source/fault range must be 1..256")
+    raise SystemExit(asyncio.run(run(args)))

experiments/perception/worker/streaming_profile_stage1/pilot_network_control.py

@@ -0,0 +1,57 @@
+"""Temporary trusted pilot control files, NOT a product grant/clock service.
+
+Both containers must share a Linux boot and unshifted monotonic clock. Reject
+foreign clocks; a future Mac/live adapter needs an actual clock mapping proof.
+The controller owns the grant file; source never owns or renews the GPU lease.
+"""
+
+import json
+import os
+from pathlib import Path
+
+from k1link.perception.realtime_contract import StreamStart
+from k1link.perception.streaming_grpc import StreamAccess
+from k1link.perception.streaming_wire import parse_header
+
+
+def kernel_clock():
+    boot = Path("/proc/sys/kernel/random/boot_id").read_text().strip()
+    offsets = Path("/proc/self/timens_offsets").read_text().splitlines()
+    if len(boot) != 36 or not offsets or any(row.split()[1:] != ["0", "0"] for row in offsets):
+        raise ValueError("pilot requires a proven unshifted Linux monotonic clock")
+    return {"boot_id": boot, "time_namespace_offsets": offsets}
+
+
+def read_control(path):
+    with Path(path).open("rb") as stream:
+        raw = stream.read(65537)
+    if not 0 < len(raw) <= 65536:
+        raise ValueError("control document exceeds pilot bound")
+    return parse_header(bytearray(raw))
+
+
+def write_control(path, value):
+    path = Path(path)
+    raw = json.dumps(value, allow_nan=False, separators=(",", ":")).encode()
+    if len(raw) > 65536:
+        raise ValueError("control document exceeds pilot bound")
+    temporary = path.with_suffix(".pending")
+    fd = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
+    with os.fdopen(fd, "wb") as stream:
+        stream.write(raw)
+    os.replace(temporary, path)
+
+
+def read_grant(path):
+    value = read_control(path)
+    if value["clock"] != kernel_clock():
+        raise ValueError("foreign source/worker clocks require measured mapping")
+    for key in ("source_zero_ns", "wall_zero_ns", "cutoff_ns"):
+        if type(value[key]) is not int or value[key] < 0:
+            raise ValueError("invalid pilot source clock anchor")
+    access = StreamAccess(
+        StreamStart.from_dict(value["epoch"]), "recorded-acquisition", 1, value["token"]
+    )
+    if not isinstance(access.token, str) or len(access.token) != 64:
+        raise ValueError("invalid pilot grant")
+    return value, access

experiments/perception/worker/streaming_profile_stage1/run_joint_pilot.py

@@ -395,18 +395,18 @@ def run(args):
             "decoder",
             (
                 ["python3", "-B", "/probe/pilot_fragment_decoder.py"]
-                if args.input_transport == "binary-ipc"
+                if args.input_transport in ("binary-ipc", "grpc")
                 else [python, "-B", "/probe/pilot_model.py", "camera"]
             ),
             {
                 **os.environ,
                 "PYTHONPATH": os.environ["PYTHONPATH"]
-                if args.input_transport == "binary-ipc"
+                if args.input_transport in ("binary-ipc", "grpc")
                 else "/probe",
                 "CUDA_VISIBLE_DEVICES": "",
             },
         )
-        if args.input_transport == "binary-ipc":
+        if args.input_transport in ("binary-ipc", "grpc"):
             from k1link.perception.streaming_decoder_client import StreamingDecoderClient
 
             decoder_client = StreamingDecoderClient(
@@ -535,7 +535,25 @@ def run(args):
             from pilot_lifecycle import FencedIngress
 
             ingress = FencedIngress(controller)
-        if args.input_transport == "binary-ipc":
+        if args.input_transport == "grpc":
+            from pilot_grpc_graph import NetworkGraphBridge
+
+            binary_bridge = NetworkGraphBridge(
+                controller.runtime,
+                decoder_client,
+                source_report,
+                control=args.network_control,
+                source_status=args.network_source_status,
+                source_zero=args.network_source_zero_ns,
+                certificate=args.network_certificate,
+                private_key=args.network_private_key,
+                address=args.network_address,
+                reset_temporal=reset_temporal,
+            )
+            controller.source_clock = binary_bridge.source
+            report["scope"] = "full-graph-network-candidate; receiver evidence is separate"
+            binary_bridge.start()
+        elif args.input_transport == "binary-ipc":
             from pilot_binary_bridge import BinaryGraphBridge
             from pilot_binary_source import RecordingSource
 
@@ -596,6 +614,9 @@ def run(args):
                         json.dumps(scene["costmap_states"], separators=(",", ":")).encode()
                     ).hexdigest()
                     freshness_started = time.monotonic_ns()
+                    if args.input_transport == "grpc":
+                        # Return the actual bounded grid, not a hash or a path to Worker files.
+                        scene["costmap_grid"] = report["costmap_grid"]
                     prepare_publication(
                         scene,
                         bundle,
@@ -617,6 +638,14 @@ def run(args):
                     encoded = json.dumps(scene, allow_nan=False, separators=(",", ":")).encode()
                     if len(encoded) > 1024 * 1024:
                         raise ValueError("scene exceeds bounded collector message budget")
+                    if args.input_transport == "grpc":
+                        from k1link.perception.streaming_scene_payload import encode_scene_payload
+
+                        binary_bridge.publish(
+                            input_start(bundle),
+                            bundle["sequence"],
+                            encode_scene_payload(encoded, mask.tobytes(), mask.shape),
+                        )
                     # Actual bounded local receiver parse. Durable export excluded below.
                     received = json.loads(encoded)
                     if received["sequence"] != bundle["sequence"]:
@@ -715,6 +744,8 @@ def run(args):
             raise RuntimeError(mailbox.error)
         if controller:
             controller.runtime.check_current(controller.start)
+        if args.input_transport == "grpc":
+            binary_bridge.finish_results()
         report["triton_statistics_delta"] = stats_delta(stats_before, read_stats(triton_models))
         report["execution_complete"] = True
     except Exception:
@@ -806,6 +837,16 @@ def run(args):
         - mailbox.dropped_count
         - len(source_report.get("failed_camera_sequences", [])),
     }
+    if args.input_transport == "grpc":
+        report["accounting"] = {
+            "scope": "worker-accepted-input-only; reconcile with external source report",
+            "accepted": len(source_report.get("accepted_camera_sequences", [])),
+            "completed": len(results),
+            "dropped": mailbox.dropped_count,
+            "released": None,
+            "source_failed": None,
+            "unaccounted": None,
+        }
     ddrnet_executed = sum(r.get("ddrnet_state") == "current" for r in results)
     report["model_cadence"] = {
         "ddrnet_executed": ddrnet_executed,
@@ -898,8 +939,14 @@ if __name__ == "__main__":
     parser.add_argument("--camera-index", default="/camera-index.jsonl")
     parser.add_argument("--camera-root", default="/camera")
     parser.add_argument(
-        "--input-transport", choices=("legacy-pilot", "binary-ipc"), default="legacy-pilot"
+        "--input-transport", choices=("legacy-pilot", "binary-ipc", "grpc"), default="legacy-pilot"
     )
+    parser.add_argument("--network-control")
+    parser.add_argument("--network-source-status")
+    parser.add_argument("--network-source-zero-ns", type=int)
+    parser.add_argument("--network-certificate")
+    parser.add_argument("--network-private-key")
+    parser.add_argument("--network-address", default="[::]:50061")
     parser.add_argument(
         "--ddrnet-layout", choices=("reference", "channels-last"), default="reference"
     )
@@ -921,7 +968,9 @@ if __name__ == "__main__":
     parser.add_argument("--controller-image-sha256")
     parser.add_argument("--worker-control-requests")
     parser.add_argument("--worker-control-responses")
-    parser.add_argument("--telemetry-snapshot", help="Optional host-agent current snapshot, never readiness")
+    parser.add_argument(
+        "--telemetry-snapshot", help="Optional host-agent current snapshot, never readiness"
+    )
     parser.add_argument(
         "--worker-readiness-mode",
         choices=("strict-envelope", "labelled-experiment"),
@@ -941,15 +990,26 @@ if __name__ == "__main__":
         parser.error("worker lease and launcher-verified image identity are required together")
     if args.stop_renew_after_sequence != -1 and not args.worker_lease_root:
         parser.error("lease-expiry injection requires a controller")
-    if args.input_transport == "binary-ipc" and not args.worker_lease_root:
+    if args.input_transport in ("binary-ipc", "grpc") and not args.worker_lease_root:
         parser.error("binary ingress requires the common lifecycle")
+    if args.input_transport == "grpc" and (
+        not args.recover_input
+        or not args.network_control
+        or not args.network_source_status
+        or not args.network_certificate
+        or not args.network_private_key
+        or not args.network_source_zero_ns
+    ):
+        parser.error(
+            "network graph requires recovery, trusted grants, TLS and a source clock anchor"
+        )
     from pilot_binary_source import input_gaps
 
     try:
         gaps = input_gaps(args.input_gap)
     except (ValueError, TypeError):
         parser.error("invalid bounded input gap plan")
-    if args.recover_input and args.input_transport != "binary-ipc":
+    if args.recover_input and args.input_transport not in ("binary-ipc", "grpc"):
         parser.error("resumable full profile requires binary input")
     if bool(args.worker_control_requests) != bool(args.worker_control_responses):
         parser.error("host control needs separate request/response mounts")

src/k1link/perception/streaming_scene_payload.py

@@ -0,0 +1,72 @@
+"""One bounded scene document plus its actual uint8 segmentation plane.
+
+No image re-encoding, base64, hidden array cache or new domain ontology. This
+codec only frames the existing scene and plane; domain layer digests/freshness
+must still be checked by the receiving profile adapter before presentation.
+"""
+
+from __future__ import annotations
+
+import struct
+from hashlib import sha256
+from typing import Any
+
+from . import streaming_wire as wire
+from .realtime_contract import StreamStart
+from .realtime_scene import SceneFreshness
+from .streaming_network import MAX_RESULT
+
+PREFIX = struct.Struct("!4sIHH")
+MAX_SCENE_JSON = 512 * 1024
+MAX_MASK_BYTES = 512 * 1024
+
+
+def encode_scene_payload(scene: bytes, mask: bytes, shape: tuple[int, int]) -> bytes:
+    height, width = shape
+    if (
+        type(scene) is not bytes
+        or type(mask) is not bytes
+        or type(height) is not int
+        or type(width) is not int
+        or not 0 < height <= 2048
+        or not 0 < width <= 2048
+        or not 0 < len(scene) <= MAX_SCENE_JSON
+        or not 0 < height * width == len(mask) <= MAX_MASK_BYTES
+        or PREFIX.size + len(scene) + len(mask) > MAX_RESULT
+    ):
+        raise wire.StreamWireError("scene/segmentation exceeds bounded output contract")
+    return PREFIX.pack(b"MCS1", len(scene), height, width) + scene + mask
+
+
+def decode_scene_payload(
+    raw: bytes, epoch: StreamStart, sequence: int
+) -> tuple[dict[str, Any], memoryview, tuple[int, int]]:
+    if not PREFIX.size < len(raw) <= MAX_RESULT:
+        raise wire.StreamWireError("invalid scene payload length")
+    magic, length, height, width = PREFIX.unpack_from(raw)
+    if (
+        magic != b"MCS1"
+        or not 0 < length <= MAX_SCENE_JSON
+        or not 0 < height <= 2048
+        or not 0 < width <= 2048
+        or not 0 < height * width <= MAX_MASK_BYTES
+        or PREFIX.size + length + height * width != len(raw)
+    ):
+        raise wire.StreamWireError("invalid scene/segmentation framing")
+    scene = wire.parse_header(bytearray(raw[PREFIX.size : PREFIX.size + length]))
+    mask = memoryview(raw)[PREFIX.size + length :]
+    freshness = SceneFreshness.from_dict(scene.get("freshness"))
+    if (
+        StreamStart.from_dict(scene.get("runtime_binding")) != epoch
+        or type(scene.get("sequence")) is not int
+        or scene["sequence"] != sequence
+        or freshness.source_sequence != sequence
+        or freshness.epoch_id != epoch.epoch_id
+        or freshness.clock_domain_id != epoch.clock_domain_id
+        or scene.get("segmentation_sha256") != sha256(mask).hexdigest()
+        or freshness.layers[0].payload_sha256 != scene["segmentation_sha256"]
+        or scene.get("commands_enabled") is not False
+        or scene.get("actuation_allowed") is not False
+    ):
+        raise wire.StreamWireError("scene identity, mask digest or authority mismatch")
+    return scene, mask, (height, width)

src/k1link/perception/streaming_wire.py

@@ -105,8 +105,11 @@ def _unique_pairs(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
 
 
 def parse_header(raw: bytearray) -> dict[str, Any]:
+    def reject_constant(value: str) -> Any:
+        raise StreamWireError("non-finite JSON metadata")
+
     try:
-        value = json.loads(raw, object_pairs_hook=_unique_pairs)
+        value = json.loads(raw, object_pairs_hook=_unique_pairs, parse_constant=reject_constant)
     except (ValueError, UnicodeError, RecursionError) as exc:
         raise StreamWireError("invalid bounded metadata") from exc
     if not isinstance(value, dict):

tests/test_perception_network_graph_adapter.py

@@ -0,0 +1,116 @@
+"""Small network/controller seam test, not decoder or GPU qualification."""
+
+import asyncio
+import importlib
+import threading
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+
+pytest.importorskip("grpc")
+
+from test_perception_streaming_grpc import eventually, identity, tls  # noqa: F401,E402
+
+from k1link.perception.streaming_grpc import GrpcStreamClient  # noqa: E402
+from k1link.perception.streaming_lifecycle import StreamingLifecycle  # noqa: E402
+from k1link.perception.streaming_queue import StreamMailbox  # noqa: E402
+
+
+@pytest.fixture
+def adapter(monkeypatch):
+    monkeypatch.syspath_prepend(
+        str(
+            Path(__file__).resolve().parents[1]
+            / "experiments/perception/worker/streaming_profile_stage1"
+        )
+    )
+    module = importlib.import_module("pilot_grpc_graph")
+    control = importlib.import_module("pilot_network_control")
+    proof = {"boot_id": "synthetic-clock-proof", "time_namespace_offsets": ["monotonic 0 0"]}
+    monkeypatch.setattr(module, "kernel_clock", lambda: proof)
+    monkeypatch.setattr(control, "kernel_clock", lambda: proof)
+    return module, control
+
+
+def test_adapter_rotates_grant_only_after_old_stream_drains(tmp_path, tls, adapter):  # noqa: F811
+    module, control = adapter
+    cert, key = tmp_path / "cert", tmp_path / "key"
+    cert.write_bytes(tls[0])
+    key.write_bytes(tls[1])
+    status = tmp_path / "status"
+    status.mkdir()
+    runtime = StreamingLifecycle(
+        identity(),
+        tmp_path / "lease",
+        StreamMailbox(),
+        threading.Event(),
+        clock_ns=lambda: 1_000_000_000,
+        recover_input=True,
+        source_clock_ns=lambda: 1_000_000_000,
+    )
+    runtime.ready()
+    closed = []
+    bridge = module.NetworkGraphBridge(
+        runtime,
+        SimpleNamespace(close=lambda: closed.append(True)),
+        {},
+        control=tmp_path,
+        source_status=status,
+        source_zero=1,
+        certificate=cert,
+        private_key=key,
+        address="localhost:0",
+        reset_temporal=lambda: None,
+    )
+    # Capture dynamically bound port without opening another listener.
+    real_serve = bridge.endpoint.serve
+
+    async def serve(*args, **kwargs):
+        server, port = await real_serve(*args, **kwargs)
+        bridge.port = port
+        return server, port
+
+    bridge.endpoint.serve = serve
+
+    async def check():
+        bridge.start()
+        _, first = control.read_grant(tmp_path / "grant.json")
+        client = GrpcStreamClient(f"localhost:{bridge.port}", tls[0], first)
+        try:
+            await client.open()
+            await client.close()  # True network disconnect, not a synthetic runtime pause.
+            await eventually(
+                lambda: control.read_grant(tmp_path / "grant.json")[1].epoch != first.epoch
+            )
+            _, second = control.read_grant(tmp_path / "grant.json")
+            assert runtime.continuity.phase == "synchronizing"
+            assert second.epoch.lease_generation == first.epoch.lease_generation
+            assert runtime.start == first.epoch and not runtime.stop_event.is_set()
+            assert bridge.endpoint.active is None
+            assert len(bridge.epoch_reports) == 1
+        finally:
+            await client.close()
+
+    try:
+        asyncio.run(check())
+    finally:
+        runtime.request_stop("completed")
+        assert bridge.close()
+        assert runtime.close()
+        assert runtime.mailbox.bytes == 0 and closed
+
+
+def test_foreign_clock_cannot_be_silently_subtracted(tmp_path, adapter):
+    _, control = adapter
+    path = tmp_path / "grant.json"
+    control.write_control(path, {"clock": {"boot_id": "other"}})
+    with pytest.raises(ValueError, match="foreign"):
+        control.read_grant(path)
+
+
+def test_delayed_source_start_does_not_open_an_idle_connection(adapter):
+    source = importlib.import_module("pilot_grpc_source")
+    assert source.connection_delay(3_000_000_000, 1_000_000_000) == 1.9
+    assert source.connection_delay(3_000_000_000, 2_950_000_000) == 0
+    assert source.connection_delay(3_000_000_000, 3_010_000_000) == 0

tests/test_perception_scene_payload.py

@@ -0,0 +1,148 @@
+"""Bounded actual scene/mask codec; no models or real recordings."""
+
+import importlib
+import json
+from hashlib import sha256
+from pathlib import Path
+
+import pytest
+
+from k1link.perception.realtime_contract import StreamStart
+from k1link.perception.streaming_scene_payload import (
+    MAX_SCENE_JSON,
+    decode_scene_payload,
+    encode_scene_payload,
+)
+
+
+@pytest.fixture
+def sample(monkeypatch):
+    monkeypatch.syspath_prepend(
+        str(
+            Path(__file__).resolve().parents[1]
+            / "experiments/perception/worker/streaming_profile_stage1"
+        )
+    )
+    pilot = importlib.import_module("pilot_freshness")
+    epoch = StreamStart(
+        "run", "source", "worker", "epoch", 1, *(["a" * 64] * 4), pilot.CLOCK_DOMAIN, "live"
+    )
+    mask = b"\x01\x02\x03\x04"
+    stamp = 9_007_199_254_740_993
+    scene = dict(
+        segmentation_sha256=sha256(mask).hexdigest(),
+        proposals=[],
+        observations=[],
+        tracks=[],
+        threats=[],
+        surface_state="valid",
+        range_estimator={},
+        costmap_states=[1],
+        costmap_material=[1],
+        policy_actions=[0],
+        costmap_grid=[[0.0, 0.0, 0.45, 0.45]],
+        policy_counts={"ALLOW_candidate": 1, "HIGH_COST": 0, "NO_GO": 0},
+        tgs_counts={"oldest_permissive_cell_source_ns": stamp},
+        commands_enabled=False,
+        actuation_allowed=False,
+        sequence=2,
+        original_source_ns=stamp,
+        runtime_binding=epoch.to_dict(),
+    )
+    bundle = dict(
+        sequence=2,
+        time_ns=stamp,
+        due_ns=1_000_000_000,
+        available=True,
+        lineage={
+            "pose_host_monotonic_ns": stamp,
+            "point_increments": [{"host_monotonic_ns": stamp}],
+        },
+    )
+    ddr = dict(state="current", source_sequence=2, source_host_monotonic_ns=stamp)
+    pilot.prepare_publication(scene, bundle, ddr, epoch_id="epoch", now_ns=bundle["due_ns"])
+    return scene, mask, epoch, bundle, pilot
+
+
+def test_actual_mask_grid_and_int64_source_identity_roundtrip(sample):
+    scene, mask, epoch, bundle, pilot = sample
+    raw = encode_scene_payload(json.dumps(scene).encode(), mask, (2, 2))
+    received, plane, shape = decode_scene_payload(raw, epoch, 2)
+    assert received == scene and bytes(plane) == mask and shape == (2, 2)
+    assert plane.obj is raw and plane.readonly  # no second retained mask allocation
+    freshness = pilot.validate_receipt(received, bundle, epoch_id="epoch")
+    view, assessed = pilot.assess_receipt(
+        received, freshness, bundle=bundle, now_ns=bundle["due_ns"] + 251_000_000
+    )
+    assert not assessed.fresh_complete and view["policy_actions"] == [2]
+    assert received["policy_actions"] == [0]  # transit age never rewrites published evidence
+
+
+@pytest.mark.parametrize(
+    "mutation",
+    [
+        "mask",
+        "shape",
+        "truncated",
+        "extra",
+        "magic",
+        "sequence",
+        "epoch",
+        "authority",
+        "digest",
+        "nan",
+        "duplicate",
+    ],
+)
+def test_payload_rejects_bad_framing_binding_and_integrity(sample, mutation):
+    scene, mask, epoch, _, _ = sample
+    if mutation == "sequence":
+        scene["sequence"] = 3
+    if mutation == "epoch":
+        scene["runtime_binding"]["epoch_id"] = "old"
+    if mutation == "authority":
+        scene["actuation_allowed"] = True
+    if mutation == "digest":
+        scene["segmentation_sha256"] = "0" * 64
+    if mutation == "nan":
+        scene["range_estimator"] = {"range": float("nan")}
+    encoded = json.dumps(scene).encode()
+    if mutation == "duplicate":
+        encoded = b'{"sequence":2,' + encoded[1:]
+    raw = encode_scene_payload(encoded, mask, (2, 2))
+    if mutation == "mask":
+        raw = raw[:-1] + b"z"
+    if mutation == "shape":
+        raw = raw[:10] + b"\x00\x03" + raw[12:]
+    if mutation == "truncated":
+        raw = raw[:-1]
+    if mutation == "extra":
+        raw += b"z"
+    if mutation == "magic":
+        raw = b"NOPE" + raw[4:]
+    with pytest.raises(ValueError):
+        decode_scene_payload(raw, epoch, 2)
+
+
+def test_grid_is_covered_by_the_costmap_layer_digest(sample):
+    scene, mask, epoch, bundle, pilot = sample
+    scene["costmap_grid"][0][0] = 500
+    raw = encode_scene_payload(json.dumps(scene).encode(), mask, (2, 2))
+    received, _, _ = decode_scene_payload(raw, epoch, 2)
+    with pytest.raises(ValueError, match="digest"):
+        pilot.validate_receipt(received, bundle, epoch_id="epoch")
+
+
+@pytest.mark.parametrize(
+    "shape,mask,scene",
+    [
+        ((0, 2), b"", b"{}"),
+        ((True, 1), b"x", b"{}"),
+        ((2, 2), b"abc", b"{}"),
+        ((1, 1), b"x", b" " * (MAX_SCENE_JSON + 1)),
+        ((1, 1), b"x", b""),
+    ],
+)
+def test_encoder_bounds(shape, mask, scene):
+    with pytest.raises(ValueError):
+        encode_scene_payload(scene, mask, shape)