From 6469505c41722b7754fecc765e42458d72f4e7bc Mon Sep 17 00:00:00 2001 From: DCCONSTRUCTIONS Date: Wed, 2 Sep 2026 19:04:41 +0300 Subject: [PATCH] feat(perception): connect full graph to external streaming source and scene receiver --- .../pilot_binary_bridge.py | 135 ++++---- .../pilot_binary_source.py | 38 +-- .../pilot_freshness.py | 2 + .../pilot_grpc_graph.py | 165 ++++++++++ .../pilot_grpc_source.py | 308 ++++++++++++++++++ .../pilot_network_control.py | 57 ++++ .../run_joint_pilot.py | 76 ++++- .../perception/streaming_scene_payload.py | 72 ++++ src/k1link/perception/streaming_wire.py | 5 +- .../test_perception_network_graph_adapter.py | 116 +++++++ tests/test_perception_scene_payload.py | 148 +++++++++ 11 files changed, 1032 insertions(+), 90 deletions(-) create mode 100644 experiments/perception/worker/streaming_profile_stage1/pilot_grpc_graph.py create mode 100644 experiments/perception/worker/streaming_profile_stage1/pilot_grpc_source.py create mode 100644 experiments/perception/worker/streaming_profile_stage1/pilot_network_control.py create mode 100644 src/k1link/perception/streaming_scene_payload.py create mode 100644 tests/test_perception_network_graph_adapter.py create mode 100644 tests/test_perception_scene_payload.py diff --git a/experiments/perception/worker/streaming_profile_stage1/pilot_binary_bridge.py b/experiments/perception/worker/streaming_profile_stage1/pilot_binary_bridge.py index 139aab4..8e6ec86 100644 --- a/experiments/perception/worker/streaming_profile_stage1/pilot_binary_bridge.py +++ b/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) diff --git a/experiments/perception/worker/streaming_profile_stage1/pilot_binary_source.py b/experiments/perception/worker/streaming_profile_stage1/pilot_binary_source.py index 02b2585..cf0a20f 100644 --- a/experiments/perception/worker/streaming_profile_stage1/pilot_binary_source.py +++ b/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(" 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 diff --git a/experiments/perception/worker/streaming_profile_stage1/pilot_grpc_source.py b/experiments/perception/worker/streaming_profile_stage1/pilot_grpc_source.py new file mode 100644 index 0000000..80af724 --- /dev/null +++ b/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))) diff --git a/experiments/perception/worker/streaming_profile_stage1/pilot_network_control.py b/experiments/perception/worker/streaming_profile_stage1/pilot_network_control.py new file mode 100644 index 0000000..1d079b4 --- /dev/null +++ b/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 diff --git a/experiments/perception/worker/streaming_profile_stage1/run_joint_pilot.py b/experiments/perception/worker/streaming_profile_stage1/run_joint_pilot.py index ef50201..1c4f562 100644 --- a/experiments/perception/worker/streaming_profile_stage1/run_joint_pilot.py +++ b/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") diff --git a/src/k1link/perception/streaming_scene_payload.py b/src/k1link/perception/streaming_scene_payload.py new file mode 100644 index 0000000..600a5ad --- /dev/null +++ b/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) diff --git a/src/k1link/perception/streaming_wire.py b/src/k1link/perception/streaming_wire.py index d246a1b..80d1839 100644 --- a/src/k1link/perception/streaming_wire.py +++ b/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): diff --git a/tests/test_perception_network_graph_adapter.py b/tests/test_perception_network_graph_adapter.py new file mode 100644 index 0000000..7e2b7af --- /dev/null +++ b/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 diff --git a/tests/test_perception_scene_payload.py b/tests/test_perception_scene_payload.py new file mode 100644 index 0000000..8cc1091 --- /dev/null +++ b/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)