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 d449947..139aab4 100644 --- a/experiments/perception/worker/streaming_profile_stage1/pilot_binary_bridge.py +++ b/experiments/perception/worker/streaming_profile_stage1/pilot_binary_bridge.py @@ -8,6 +8,8 @@ from collections import Counter import numpy as np +from k1link.media_fragments import ParseBudget, video_fragment_timing, video_timing +from k1link.perception.streaming_continuity import ResumeEvidence, StreamSuspended from k1link.perception.streaming_ingress import StreamingIngress from k1link.perception.streaming_sensors import ( CausalSensorWindow, @@ -18,7 +20,7 @@ from k1link.perception.streaming_sensors import ( class BinaryGraphBridge: - def __init__(self, runtime, decoder, source, report): + 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 self.window = CausalSensorWindow(runtime.mailbox) @@ -26,6 +28,11 @@ class BinaryGraphBridge: self.binding_reasons = Counter() self.preroll = 0 self.bgr_hashes = [] + self.epoch = runtime.start + self.reset_temporal = reset_temporal + 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() @@ -33,7 +40,11 @@ class BinaryGraphBridge: right, runtime, "recorded-acquisition", 1, self.consume, self.notice ) self.producer = threading.Thread( - target=source.run, args=(self.left,), daemon=True, name="binary-recorded-source" + 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: @@ -54,14 +65,52 @@ class BinaryGraphBridge: # and fresh decoder state; never continue predictive decoding across it. raise ValueError("binary full profile requires restart after an explicit source gap") + 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.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_current(self.runtime.start) + self.runtime.check_input(self.epoch, synchronizing=True) stamp = event.received_monotonic_ns self.window.advance(stamp) if event.modality == "camera-init": if event.source_id != "sensor.camera.right": raise ValueError("camera source mapping changed") - self.decoder.configure(event.payload) + if self.decode_sequence_start is None: + self.decoder.reset(self.epoch, event.payload) + else: + self.decoder.configure(event.payload) + self.init_timing = video_timing(event.payload, ParseBudget(128, 1)) return if event.modality in ("lidar", "pose"): expected = { @@ -84,6 +133,16 @@ class BinaryGraphBridge: fresh = binding.increments count = sum(len(e.value[0]) for e in fresh) rolling_count = sum(len(e.value[0]) for e in rolling) + resuming = self.decode_sequence_start is None + if resuming and ( + not binding.available + or not video_fragment_timing( + event.payload, self.init_timing, ParseBudget(128, 1) + ).random_access + ): + self.sync_skipped.append(event.source_sequence) + self.window.finish_camera(stamp) + return # No decode of dependent pictures; wait for a current keyframe/pair. size = 1440000 + count * 24 + rolling_count * 32 + 4096 reservation = self.runtime.mailbox.reserve_ingress(size) raw = image = points = rolling_points = rolling_times = bundle = None @@ -94,7 +153,25 @@ class BinaryGraphBridge: decoded = self.decoder.decode(event.payload, raw) image = np.frombuffer(raw, "u1").reshape(600, 800, 3) image.setflags(write=False) - if decoded["frame_index"] != event.source_sequence: + if resuming: + evidence = ResumeEvidence( + stamp, + pose.time_ns, + fresh[0].time_ns, + fresh[-1].time_ns, + decoded["frame_index"] == 0, + ) + self.runtime.resume_input(self.epoch, evidence, self.reset_temporal) + self.decode_sequence_start = event.source_sequence + self.resumes.append( + { + "sequence": event.source_sequence, + "input_start": self.epoch.to_dict(), + "source_ns": stamp, + "completed_monotonic_ns": time.monotonic_ns(), + } + ) + if decoded["frame_index"] != event.source_sequence - self.decode_sequence_start: raise ValueError("camera and decoder frame identity diverged") points = np.concatenate([e.value[0] for e in fresh]) if fresh else np.empty((0, 3)) rolling_points = ( @@ -108,6 +185,7 @@ class BinaryGraphBridge: offset += length due = self.wall_zero + stamp - self.source_zero bundle = { + "input_start": self.epoch, "sequence": event.source_sequence, "time_ns": stamp, "source_ns": stamp - self.source_zero, @@ -139,7 +217,7 @@ class BinaryGraphBridge: } self.bgr_hashes.append(hashlib.sha256(image).hexdigest()) bundle["enqueued_ns"] = time.monotonic_ns() - transferred = self.runtime.admit_reserved(self.runtime.start, bundle, reservation) + transferred = self.runtime.admit_reserved(self.epoch, bundle, reservation) self.accepted.append(event.source_sequence) self.binding_reasons.update(binding.reasons) self.preroll += sum(len(e.value[0]) for e in binding.history_only) @@ -160,6 +238,9 @@ class BinaryGraphBridge: joined = self.receiver.join(timeout=2) if self.receiver.thread.ident is not None else True self.report.update( binary_ingress=self.receiver.snapshot(), + previous_input_epochs=self.epoch_reports, + resumed_inputs=self.resumes, + synchronization_skipped_camera_sequences=self.sync_skipped, decoder_bgr_sha256=self.bgr_hashes, failed_camera_sequences=sorted(set(self.source.released) - set(self.accepted)), sensor_binding_reasons=dict(self.binding_reasons), diff --git a/experiments/perception/worker/streaming_profile_stage1/pilot_binary_bridge_probe.py b/experiments/perception/worker/streaming_profile_stage1/pilot_binary_bridge_probe.py index 334f0e7..e0f0448 100644 --- a/experiments/perception/worker/streaming_profile_stage1/pilot_binary_bridge_probe.py +++ b/experiments/perception/worker/streaming_profile_stage1/pilot_binary_bridge_probe.py @@ -15,6 +15,7 @@ from pilot_binary_bridge import BinaryGraphBridge from pilot_binary_source import RecordingSource from k1link.perception.realtime_contract import StreamStart +from k1link.perception.streaming_continuity import StreamSuspended from k1link.perception.streaming_decoder_client import StreamingDecoderClient from k1link.perception.streaming_lifecycle import StreamingLifecycle from k1link.perception.streaming_queue import StreamMailbox @@ -60,7 +61,16 @@ def run(args): lease_generation=1, profile_sha256=hashlib.sha256(Path("/out/candidate-profile.json").read_bytes()).hexdigest(), image_sha256="664824aa25de1db178f177d67a81b01541a938b479812b9383ddbf03f6b59dbe", - effective_config_sha256=hashlib.sha256(b"cpu-bundle-parity-v1").hexdigest(), + effective_config_sha256=hashlib.sha256( + json.dumps( + { + "probe": "cpu-bundle-parity-v1", + "recover_input": args.recover_input, + "input_gap": args.input_gap, + }, + sort_keys=True, + ).encode() + ).hexdigest(), calibration_sha256=hashlib.sha256( b"no-calibration-or-model-scene-in-this-probe" ).hexdigest(), @@ -68,7 +78,12 @@ def run(args): input_mode="recorded-source-paced", ) runtime = StreamingLifecycle( - identity, Path("/tmp/cpu-bridge-lease"), StreamMailbox(), threading.Event() + identity, + Path("/tmp/cpu-bridge-lease"), + StreamMailbox(), + threading.Event(), + recover_input=args.recover_input, + source_clock_ns=lambda: source.source_zero + time.monotonic_ns() - source.wall_zero, ) heartbeat_stop = threading.Event() @@ -113,9 +128,20 @@ def run(args): client = StreamingDecoderClient(runtime, child.stdout.fileno(), child.stdin.fileno()) runtime.ready() source = RecordingSource(args, runtime, source_report) - bridge = BinaryGraphBridge(runtime, client, source, source_report) + bridge = BinaryGraphBridge( + runtime, + client, + source, + source_report, + reset_temporal=(lambda: None) if args.recover_input else None, + ) bridge.start() while (bundle := runtime.mailbox.take()) is not None: + try: + runtime.check_input(bundle["input_start"]) + except StreamSuspended: + runtime.mailbox.release(bundle, discard_reason="input-gap") + continue timings.append( { "sequence": bundle["sequence"], @@ -131,7 +157,7 @@ def run(args): finally: heartbeat_stop.set() renewer.join(timeout=1) - runtime.request_stop("completed" if len(decoded) == args.frames else "failed") + runtime.request_stop("completed" if not runtime.mailbox.error and decoded else "failed") runtime.stop_children() report["bridge_stopped"] = bridge.close() if bridge else True if client: @@ -142,7 +168,35 @@ def run(args): report["peak_input_bytes"] = runtime.mailbox.peak_bytes report["residual_input_bytes"] = runtime.mailbox.bytes report["timings"] = timings + report["model_runs"] = 0 + report["queue_dropped"] = runtime.mailbox.dropped + report["bundle_count"] = len(decoded) (root / "report.json").write_text(json.dumps(report, indent=2) + "\n") + if args.recover_input: + report["continuity_passed"] = ( + len(source_report["resumed_inputs"]) == len(args.input_gap) + and len(decoded) + + runtime.mailbox.dropped_count + + len(source_report["failed_camera_sequences"]) + == args.frames + and report["released"] + and not runtime.mailbox.bytes + and not runtime.mailbox.error + ) + (root / "decoded-bundles.json").write_text(json.dumps(decoded, indent=2) + "\n") + (root / "report.json").write_text(json.dumps(report, indent=2) + "\n") + assert report["continuity_passed"], report + print( + json.dumps( + { + "continuity_passed": True, + "resumes": len(source_report["resumed_inputs"]), + "decoded": len(decoded), + "model_runs": 0, + } + ) + ) + return if args.fault_decoder: assert not decoded and report["released"] and not runtime.mailbox.bytes assert "decoder RPC deadline" in source_report["binary_ingress"]["error"] @@ -223,6 +277,8 @@ if __name__ == "__main__": parser.add_argument("--output", required=True) parser.add_argument("--frames", type=int, default=32) parser.add_argument("--fault-decoder", action="store_true") + parser.add_argument("--recover-input", action="store_true") + parser.add_argument("--input-gap", action="append", default=[]) parser.add_argument("--camera-root", default="/camera") parser.add_argument("--camera-index", default="/camera/index.jsonl") parser.add_argument("--sensor-archive", default="/sensor-source.npz") 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 7e72ff1..02b2585 100644 --- a/experiments/perception/worker/streaming_profile_stage1/pilot_binary_source.py +++ b/experiments/perception/worker/streaming_profile_stage1/pilot_binary_source.py @@ -14,7 +14,24 @@ from pathlib import Path from pilot_source import SensorArchive, camera_events, merged_events from k1link.compute.live_perception import LiveIngressEvent +from k1link.perception.streaming_continuity import StreamSuspended from k1link.perception.streaming_sender import StreamingSender +from k1link.perception.streaming_wire import StreamWireError + + +def input_gaps(values): + """Bounded diagnostic fault plan, not a source or model configuration.""" + result = [] + for value in values: + sequence, milliseconds = (int(v) for v in value.split(":")) + if not 0 < sequence < 256 or not 1 <= milliseconds <= 5000: + raise ValueError("input gap outside pilot bounds") + if result and sequence <= result[-1][0]: + raise ValueError("input gaps must increase") + result.append((sequence, milliseconds)) + if len(result) > 4: + raise ValueError("at most four input gaps") + return result def read_member(path, root, maximum): @@ -44,36 +61,44 @@ class RecordingSource: full_source_prepass=False, ) - def run(self, connection): + def run(self, connection, *, bridge=None): archive = None sender = None arrivals, skipped = Counter(), Counter() lags = [] clock = self.runtime.stop_event + gaps = input_gaps(getattr(self.args, "input_gap", [])) + gap_index, outage_until = 0, 0 + faults, skipped_cameras = [], [] try: archive = SensorArchive(Path(self.args.sensor_archive)) root = Path(self.args.camera_root) - sender = StreamingSender( - connection, - self.runtime.start, - "recorded-acquisition", - 1, - lambda: self.runtime.check_current(self.runtime.start), - ) - seq = 1 - sender.send( - LiveIngressEvent( - seq, + + def connect(sock, epoch, stamp): + stream = StreamingSender( + sock, + epoch, "recorded-acquisition", 1, - "camera-init", - "sensor.camera.right", - 0, - 0, - self.source_zero, - read_member(root / "init.mp4", root, 65536), + lambda: self.runtime.check_current(self.runtime.start), ) - ) + stream.send( + LiveIngressEvent( + 1, + "recorded-acquisition", + 1, + "camera-init", + "sensor.camera.right", + 0, + 0, + stamp, + read_member(root / "init.mp4", root, 65536), + ) + ) + return stream + + sender = connect(connection, self.runtime.start, self.source_zero) + seq = 1 for event in merged_events(archive, self.args.camera_index, self.args.frames): if event.time_ns < self.source_zero: skipped[event.channel] += 1 @@ -86,6 +111,46 @@ class RecordingSource: utc = 0 # These normalized sensor rows have no UTC evidence. if event.channel == "camera": self.released.append(event.sequence) + if bridge is not None: + if ( + event.channel == "camera" + and gap_index < len(gaps) + and event.sequence >= gaps[gap_index][0] + ): + milliseconds = gaps[gap_index][1] + bridge.disconnect() + sender = None + outage_until = time.monotonic_ns() + milliseconds * 1_000_000 + faults.append( + { + "sequence": event.sequence, + "source_ns": event.time_ns, + "injected_monotonic_ns": time.monotonic_ns(), + "duration_ms": milliseconds, + "runtime": self.runtime.snapshot(), + } + ) + gap_index += 1 + if self.runtime.continuity.phase == "waiting": + if sender is not None: + sender.close() + sender = None + if time.monotonic_ns() >= outage_until: + connection = bridge.reconnect() + if connection is not None: + sender = connect( + connection, bridge.epoch, self.runtime.continuity.cutoff_ns + ) + seq = 1 + # The current event predates the reconnect cutoff. Do + # not retimestamp or replay it; wait for the next event. + skipped[event.channel + "-input-gap"] += 1 + if event.channel == "camera": + skipped_cameras.append(event.sequence) + 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 ( @@ -107,25 +172,40 @@ class RecordingSource: raw = position.tobytes() + quaternion.tobytes() modality, source_id = "pose", "normalized-map-from-lidar-pose" seq += 1 - sender.send( - LiveIngressEvent( - seq, - "recorded-acquisition", - 1, - modality, - source_id, - event.sequence, - utc, - event.time_ns, - raw, + try: + sender.send( + LiveIngressEvent( + seq, + "recorded-acquisition", + 1, + modality, + source_id, + event.sequence, + utc, + event.time_ns, + raw, + ) ) - ) + except (StreamWireError, OSError, StreamSuspended): + if bridge is None: + raise + # Bounded IPC write failure: pause, never retry old bytes. + bridge.disconnect() + sender = None + skipped[event.channel + "-send-interrupted"] += 1 + if event.channel == "camera": + skipped_cameras.append(event.sequence) del raw if event.channel == "camera" and event.sequence == self.args.frames - 1: break # Sensor streams may continue beyond this bounded camera probe. - if not clock.is_set(): + if not clock.is_set() and sender is not None: self.report["end_sent_monotonic_ns"] = time.monotonic_ns() sender.end() + elif not clock.is_set(): + # The bounded diagnostic source ended during an outage. There + # are no further observations to recover from; do not hang. + self.report["source_ended_while_waiting"] = True + self.runtime.mailbox.finish() except Exception: self.report["source_error"] = traceback.format_exc() self.runtime.mailbox.finish(self.report["source_error"]) @@ -139,7 +219,10 @@ class RecordingSource: release_lag_ms=distribution(lags), incremental_reads=archive.counters() if archive else {}, window_end_monotonic_ns=time.monotonic_ns(), + input_faults=faults, + input_gap_skipped_camera_sequences=skipped_cameras, ) if archive: archive.close() - connection.close() + if connection is not None: + connection.close() diff --git a/experiments/perception/worker/streaming_profile_stage1/pilot_graph.py b/experiments/perception/worker/streaming_profile_stage1/pilot_graph.py index bfacbb5..ff89c60 100644 --- a/experiments/perception/worker/streaming_profile_stage1/pilot_graph.py +++ b/experiments/perception/worker/streaming_profile_stage1/pilot_graph.py @@ -200,6 +200,7 @@ class JointGraph: body_frame_resolver=self.store, profile=load_replay_threat_profile(config / "m4-replay-threat-v3.json"), ) + self.temporal_resets = 0 self.backend = TritonNativeRfDetrHttpInferenceBackend( "http://127.0.0.1:8000", timeout_seconds=5 ) @@ -257,6 +258,39 @@ class JointGraph: if material != "hard_surface": self.action_lut[i, :] = 2 + def reset_temporal(self): + """Quiescent input epoch boundary; keep model backend and TGS process. + + TGS C++ creates a new estimator per request; costmap arrays are local. + Cross-frame ground, associations, motion and rolling evidence live here. + Configuration objects are retained; no model reload or host I/O. + """ + previous = self.temporal_state_counts() + surface = K1LocalSurfaceShadowEstimator(self.surface.profile) + store = CurrentStore(self.store.profile) + geometry = Ravnoves00GeometryAssociationProvider(store=store) + temporal = BoundedSpatialTemporalProvider( + point_resolver=store, profile=self.temporal.profile + ) + motion = ClassIndependentMotionEstimator(profile=self.motion.profile) + rolling = RollingLocalObstacleMapProvider(pose_resolver=store, profile=self.rolling.profile) + threat = DualEvidenceReplayThreatProvider( + body_frame_resolver=store, profile=self.threat.profile + ) + self.surface, self.store, self.geometry = surface, store, geometry + self.temporal, self.motion, self.rolling, self.threat = temporal, motion, rolling, threat + self.temporal_resets += 1 + return {"previous": previous, "current": self.temporal_state_counts()} + + def temporal_state_counts(self): + """Bounded diagnostics, never admission evidence from a remote client.""" + return { + "surface_cache": len(self.surface._cache), + "body_history": len(self.store.body_history), + "components": len(self.temporal._components), + "rolling_cells": len(self.rolling._cells), + } + def packet(self, bundle): available = bundle["available"] status = ModalityStatus( diff --git a/experiments/perception/worker/streaming_profile_stage1/pilot_lifecycle.py b/experiments/perception/worker/streaming_profile_stage1/pilot_lifecycle.py index 7b7eabc..a8280ee 100644 --- a/experiments/perception/worker/streaming_profile_stage1/pilot_lifecycle.py +++ b/experiments/perception/worker/streaming_profile_stage1/pilot_lifecycle.py @@ -9,6 +9,7 @@ import hashlib import json import os import threading +import time import zipfile from pathlib import Path @@ -94,8 +95,15 @@ class PilotController: input_mode="recorded-source-paced", ) self.runtime = StreamingLifecycle( - self.start, Path(args.worker_lease_root), mailbox, stop, ttl_seconds=2.0 + self.start, + Path(args.worker_lease_root), + mailbox, + stop, + ttl_seconds=2.0, + recover_input=getattr(args, "recover_input", False), + source_clock_ns=self.source_now, ) + self.source_clock = None self.heartbeat_stop = threading.Event() self.heartbeat_error = None self.thread = threading.Thread( @@ -108,6 +116,11 @@ class PilotController: report["lease_authority_scope"] = "cooperating-runtimes-sharing-fixed-worker-volume" report["heartbeat_scope"] = "same-process-pilot-controller-not-network-client" + def source_now(self): + if self.source_clock is None: + raise WorkerLeaseError("source clock mapping not configured") + return self.source_clock.source_zero + time.monotonic_ns() - self.source_clock.wall_zero + def _renew(self): while not self.heartbeat_stop.wait(0.25): try: 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 a494afd..2922b65 100644 --- a/experiments/perception/worker/streaming_profile_stage1/run_joint_pilot.py +++ b/experiments/perception/worker/streaming_profile_stage1/run_joint_pilot.py @@ -33,6 +33,8 @@ from pilot_sensor_binding import bind_sensors, increment_identity, milliseconds from pilot_source import SensorArchive, camera_events, merged_events from pilot_telemetry import NvmlSampler +from k1link.perception.streaming_continuity import StreamSuspended + def distribution(values): # Missing diagnostics are not zero; binary source lag is measured separately. @@ -274,6 +276,8 @@ def run(args): "costmap_freshness_mode": args.costmap_freshness, "ddrnet_min_source_interval_ms": args.ddrnet_min_source_interval_ms, "input_transport": args.input_transport, + "recover_input": getattr(args, "recover_input", False), + "input_gap_plan": getattr(args, "input_gap", []), } children, logs, results, samples = [], [], [], [] stop = threading.Event() @@ -430,6 +434,26 @@ def run(args): stats_before = read_stats(triton_models) ddr_cache = None + report["resident_pids"] = [p.pid for p in children] + report["temporal_reset_events"] = [] + + def reset_temporal(): + nonlocal ddr_cache + state = graph.reset_temporal() + ddr_cache = None + report["temporal_reset_events"].append( + { + "state": state, + "ddr_cache_cleared": True, + "input_epoch_id": controller.runtime.continuity.epoch.epoch_id, + "resident_pids": [p.pid for p in children], + "all_children_alive": all(p.poll() is None for p in children), + "monotonic_ns": time.monotonic_ns(), + } + ) + + def input_start(bundle): + return bundle.get("input_start", controller.start) if controller else None def compute_gpu_impl(bundle): nonlocal ddr_cache @@ -480,7 +504,7 @@ def run(args): mask = ddr_cache["mask"] ddr_done = time.monotonic_ns() if controller: - controller.runtime.check_current(controller.start) + controller.runtime.check_input(input_start(bundle)) proposals = graph.detector.detect(graph.packet(bundle)) gpu_done = time.monotonic_ns() ddr_layer = { @@ -493,7 +517,7 @@ def run(args): def compute_gpu(bundle): context = ( - controller.runtime.work(controller.start, "gpu") if controller else nullcontext() + controller.runtime.work(input_start(bundle), "gpu") if controller else nullcontext() ) with context: return compute_gpu_impl(bundle) @@ -516,8 +540,13 @@ def run(args): from pilot_binary_source import RecordingSource source = RecordingSource(args, controller.runtime, source_report) + controller.source_clock = source binary_bridge = BinaryGraphBridge( - controller.runtime, decoder_client, source, source_report + controller.runtime, + decoder_client, + source, + source_report, + reset_temporal=reset_temporal if report["recover_input"] else None, ) binary_bridge.start() else: @@ -538,109 +567,126 @@ def run(args): if item is None: break bundle, computed = item + item = None # Do not retain the old input while waiting for the next epoch. cpu_bundle = bundle else: bundle = mailbox.take() if bundle is None: break cpu_bundle = bundle - computed = compute_gpu(bundle) - begin, ddr_done, gpu_done, ddr_result, mask, proposals, ddr_layer = computed - cpu_started = time.monotonic_ns() - context = ( - controller.runtime.work(controller.start, "cpu") - if controller - else nullcontext() - ) - with context: - scene = graph.process( - bundle, - mask, - proposals=proposals, - detector_ms=(gpu_done - ddr_done) / 1e6, + computed = None + try: + if computed is None: + computed = compute_gpu(bundle) + begin, ddr_done, gpu_done, ddr_result, mask, proposals, ddr_layer = computed + cpu_started = time.monotonic_ns() + context = ( + controller.runtime.work(input_start(bundle), "cpu") + if controller + else nullcontext() ) - raw_costmap_sha256 = hashlib.sha256( - json.dumps(scene["costmap_states"], separators=(",", ":")).encode() - ).hexdigest() - freshness_started = time.monotonic_ns() - prepare_publication( - scene, - bundle, - ddr_layer, - epoch_id=epoch_id, - now_ns=freshness_started, - mode=args.costmap_freshness, - ) - scene.update( - sequence=bundle["sequence"], - lineage=bundle["lineage"], - sensor_binding=bundle["sensor_binding"], - available=bundle["available"], - original_source_ns=bundle["time_ns"], - ) - if controller: - controller.runtime.check_current(controller.start) - scene["runtime_binding"] = controller.start.to_dict() - encoded = json.dumps(scene, allow_nan=False, separators=(",", ":")).encode() - if len(encoded) > 1024 * 1024: - raise ValueError("scene exceeds bounded collector message budget") - # Actual bounded local receiver parse. Durable export excluded below. - received = json.loads(encoded) - if received["sequence"] != bundle["sequence"]: - raise ValueError("collector identity mismatch") - if controller: - controller.runtime.validate_result_binding(received["runtime_binding"]) - freshness = validate_receipt(received, bundle, epoch_id=epoch_id) - checked_at = time.monotonic_ns() - view, checked = assess_receipt( - received, freshness, bundle=bundle, now_ns=checked_at - ) - if controller: - controller.runtime.validate_result_binding(view["runtime_binding"]) - finished = time.monotonic_ns() - timing = { - **scene["timing_ms"], - **{"ddrnet_" + key: value for key, value in ddr_result["stages_ms"].items()}, - "ddrnet_rpc_ms": (ddr_done - begin) / 1e6, - "ddrnet_component_ms": ddr_result["component_ms"], - "ddrnet_forward_ms": ddr_result["forward_ms"], - "decode_ms": bundle["decode_ms"], - "decode_rpc_ms": bundle["decode_rpc_ms"], - "source_release_lag_ms": bundle["source_release_lag_ms"], - "queue_wait_ms": (begin - bundle["enqueued_ns"]) / 1e6, - "gpu_to_cpu_queue_wait_ms": (cpu_started - gpu_done) / 1e6, - "cpu_tail_to_receiver_ms": (finished - cpu_started) / 1e6, - "compute_to_receiver_ms": (finished - begin) / 1e6, - "source_due_to_receiver_ms": (finished - bundle["due_ns"]) / 1e6, - "freshness_encode_receive_ms": (finished - freshness_started) / 1e6, - } - result = { - "sequence": bundle["sequence"], - "finished_monotonic_ns": finished, - "timing_ms": timing, - "available": bundle["available"], - "ddrnet_state": ddr_layer["state"], - "ddrnet_source_age_ms": ddr_layer["source_age_ms"], - "surface_state": scene["surface_state"], - "proposal_count": len(scene["proposals"]), - "observation_count": len(scene["observations"]), - "track_count": len(scene["tracks"]), - "metric_count": sum( - x["metric_geometry"] is not None for x in scene["observations"] - ), - "tgs_counts": scene["tgs_counts"], - "policy_counts": view["policy_counts"], - "freshness_at_receipt": checked.to_dict(), - "policy_counts_at_publication": scene["policy_counts"], - "cell_assessment_at_receipt": view.get("cell_assessment"), - "raw_costmap_states_sha256": raw_costmap_sha256, - "effective_costmap_states_sha256": hashlib.sha256( - json.dumps(view["costmap_states"], separators=(",", ":")).encode() - ).hexdigest(), - "scene_bytes": len(encoded), - "scene_sha256": hashlib.sha256(encoded).hexdigest(), - } - results.append(result) + with context: + scene = graph.process( + bundle, + mask, + proposals=proposals, + detector_ms=(gpu_done - ddr_done) / 1e6, + ) + raw_costmap_sha256 = hashlib.sha256( + json.dumps(scene["costmap_states"], separators=(",", ":")).encode() + ).hexdigest() + freshness_started = time.monotonic_ns() + prepare_publication( + scene, + bundle, + ddr_layer, + epoch_id=input_start(bundle).epoch_id if controller else epoch_id, + now_ns=freshness_started, + mode=args.costmap_freshness, + ) + scene.update( + sequence=bundle["sequence"], + lineage=bundle["lineage"], + sensor_binding=bundle["sensor_binding"], + available=bundle["available"], + original_source_ns=bundle["time_ns"], + ) + if controller: + controller.runtime.check_input(input_start(bundle)) + scene["runtime_binding"] = input_start(bundle).to_dict() + encoded = json.dumps(scene, allow_nan=False, separators=(",", ":")).encode() + if len(encoded) > 1024 * 1024: + raise ValueError("scene exceeds bounded collector message budget") + # Actual bounded local receiver parse. Durable export excluded below. + received = json.loads(encoded) + if received["sequence"] != bundle["sequence"]: + raise ValueError("collector identity mismatch") + if controller: + controller.runtime.validate_result_binding(received["runtime_binding"]) + freshness = validate_receipt( + received, + bundle, + epoch_id=input_start(bundle).epoch_id if controller else epoch_id, + ) + checked_at = time.monotonic_ns() + view, checked = assess_receipt( + received, freshness, bundle=bundle, now_ns=checked_at + ) + if controller: + controller.runtime.validate_result_binding(view["runtime_binding"]) + finished = time.monotonic_ns() + timing = { + **scene["timing_ms"], + **{ + "ddrnet_" + key: value for key, value in ddr_result["stages_ms"].items() + }, + "ddrnet_rpc_ms": (ddr_done - begin) / 1e6, + "ddrnet_component_ms": ddr_result["component_ms"], + "ddrnet_forward_ms": ddr_result["forward_ms"], + "decode_ms": bundle["decode_ms"], + "decode_rpc_ms": bundle["decode_rpc_ms"], + "source_release_lag_ms": bundle["source_release_lag_ms"], + "queue_wait_ms": (begin - bundle["enqueued_ns"]) / 1e6, + "gpu_to_cpu_queue_wait_ms": (cpu_started - gpu_done) / 1e6, + "cpu_tail_to_receiver_ms": (finished - cpu_started) / 1e6, + "compute_to_receiver_ms": (finished - begin) / 1e6, + "source_due_to_receiver_ms": (finished - bundle["due_ns"]) / 1e6, + "freshness_encode_receive_ms": (finished - freshness_started) / 1e6, + } + result = { + "input_epoch_id": input_start(bundle).epoch_id if controller else epoch_id, + "sequence": bundle["sequence"], + "finished_monotonic_ns": finished, + "timing_ms": timing, + "available": bundle["available"], + "ddrnet_state": ddr_layer["state"], + "ddrnet_source_age_ms": ddr_layer["source_age_ms"], + "surface_state": scene["surface_state"], + "proposal_count": len(scene["proposals"]), + "observation_count": len(scene["observations"]), + "track_count": len(scene["tracks"]), + "metric_count": sum( + x["metric_geometry"] is not None for x in scene["observations"] + ), + "tgs_counts": scene["tgs_counts"], + "policy_counts": view["policy_counts"], + "freshness_at_receipt": checked.to_dict(), + "policy_counts_at_publication": scene["policy_counts"], + "cell_assessment_at_receipt": view.get("cell_assessment"), + "raw_costmap_states_sha256": raw_costmap_sha256, + "effective_costmap_states_sha256": hashlib.sha256( + json.dumps(view["costmap_states"], separators=(",", ":")).encode() + ).hexdigest(), + "scene_bytes": len(encoded), + "scene_sha256": hashlib.sha256(encoded).hexdigest(), + } + results.append(result) + except StreamSuspended: + mailbox.release(bundle, discard_reason="input-gap") + cpu_bundle = bundle = computed = mask = proposals = scene = received = view = ( + None + ) + continue mailbox.release(bundle) cpu_bundle = None sink.write(encoded + b"\n") @@ -664,6 +710,7 @@ def run(args): ), flush=True, ) + bundle = computed = mask = proposals = scene = received = view = None if mailbox.error: raise RuntimeError(mailbox.error) if controller: @@ -873,6 +920,8 @@ if __name__ == "__main__": parser.add_argument("--lease-generation", type=int, default=1) parser.add_argument("--controller-image-sha256") parser.add_argument("--stop-renew-after-sequence", type=int, default=-1) + parser.add_argument("--recover-input", action="store_true") + parser.add_argument("--input-gap", action="append", default=[], metavar="SEQUENCE:MILLISECONDS") args = parser.parse_args() if not 1 <= args.frames <= 256: parser.error("pilot window must be 1..256 frames") @@ -886,4 +935,14 @@ if __name__ == "__main__": parser.error("lease-expiry injection requires a controller") if args.input_transport == "binary-ipc" and not args.worker_lease_root: parser.error("binary ingress requires the common lifecycle") + 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": + parser.error("resumable full profile requires binary input") + if gaps and (not args.recover_input or any(seq >= args.frames for seq, _ in gaps)): + parser.error("input gaps require recovery and must be inside the bounded source") raise SystemExit(run(args)) diff --git a/src/k1link/perception/streaming_lifecycle.py b/src/k1link/perception/streaming_lifecycle.py index c073314..890e2c7 100644 --- a/src/k1link/perception/streaming_lifecycle.py +++ b/src/k1link/perception/streaming_lifecycle.py @@ -227,8 +227,10 @@ class StreamingLifecycle: self._check(start) if self.continuity is None or self._source_clock_ns is None: raise ValueError("resumable input was not enabled") - if any(self._active.values()) or ( - self._ingress_thread is not None and self._ingress_thread.is_alive() + if ( + not self.mailbox.epoch_drained + or any(self._active.values()) + or (self._ingress_thread is not None and self._ingress_thread.is_alive()) ): raise StreamSuspended("old epoch callbacks or connection still active") if self.continuity.phase != "waiting": diff --git a/src/k1link/perception/streaming_queue.py b/src/k1link/perception/streaming_queue.py index 0e2dd71..6b8d0a4 100644 --- a/src/k1link/perception/streaming_queue.py +++ b/src/k1link/perception/streaming_queue.py @@ -228,3 +228,9 @@ class StreamMailbox: if self.done or self._owned or self.external_pending: raise ValueError("old epoch still owns work or mailbox is closed") self._last_sequence = -1 + + @property + def epoch_drained(self) -> bool: + """Resident scratch may remain; no old input/result may still be owned.""" + with self.condition: + return not self.done and not self._owned and not self.external_pending diff --git a/tests/test_perception_joint_pilot.py b/tests/test_perception_joint_pilot.py index 7199e2a..e4f659a 100644 --- a/tests/test_perception_joint_pilot.py +++ b/tests/test_perception_joint_pilot.py @@ -21,6 +21,67 @@ def pilot(monkeypatch): return lambda name: importlib.import_module(name) +def test_full_graph_reset_replaces_temporal_state_not_models(pilot): + module = pilot("pilot_graph") + root = Path(__file__).resolve().parents[1] / "config/perception" + graph = module.JointGraph.__new__(module.JointGraph) + graph.surface = module.K1LocalSurfaceShadowEstimator() + graph.store = module.CurrentStore( + module.load_geometry_profile(root / "m4-geometry-association-v1.json") + ) + profile = module.load_temporal_motion_profile(root / "m4-temporal-motion-v1.json") + graph.temporal = module.BoundedSpatialTemporalProvider( + point_resolver=graph.store, profile=profile + ) + graph.motion = module.ClassIndependentMotionEstimator(profile=profile) + graph.rolling = module.RollingLocalObstacleMapProvider( + pose_resolver=graph.store, + profile=module.load_rolling_map_profile(root / "m4-rolling-local-map-v1.json"), + ) + graph.threat = module.DualEvidenceReplayThreatProvider( + body_frame_resolver=graph.store, + profile=module.load_replay_threat_profile(root / "m4-replay-threat-v3.json"), + ) + graph.temporal_resets = 0 + graph.backend = graph.detector = graph.tgs = model = object() + for cache in ( + graph.surface._cache, + graph.store.body_history, + graph.temporal._components, + graph.rolling._cells, + ): + cache["old"] = object() + graph.surface._previous_surface = (1, 2, 3, 4) + graph.temporal._previous_sequence = graph.rolling._previous_sequence = 32 + old_store = graph.store + state = graph.reset_temporal() + assert set(state["previous"].values()) == {1} + assert set(state["current"].values()) == {0} + assert graph.temporal._previous_sequence is graph.rolling._previous_sequence is None + assert graph.surface._previous_surface is None and graph.temporal_resets == 1 + assert graph.store is not old_store + assert ( + graph.temporal.point_resolver + is graph.rolling.pose_resolver + is graph.threat.body_frame_resolver + is graph.store + ) + assert graph.backend is graph.detector is graph.tgs is model + assert graph.temporal.profile is graph.motion.profile is profile + + +@pytest.mark.parametrize( + "values", [["0:1"], ["10:0"], ["10:5001"], ["a:1"], ["10:1", "9:1"], ["1:1"] * 5] +) +def test_source_gap_plan_is_bounded(pilot, values): + with pytest.raises(ValueError): + pilot("pilot_binary_source").input_gaps(values) + + +def test_source_gap_plan_preserves_explicit_sequence_and_duration(pilot): + assert pilot("pilot_binary_source").input_gaps(["16:150", "72:2200"]) == [(16, 150), (72, 2200)] + + def test_pending_overflow_is_explicit_and_does_not_evict_active(pilot): queue = pilot("pilot_queue").Mailbox(capacity=2, byte_limit=100) diff --git a/tests/test_perception_streaming_continuity.py b/tests/test_perception_streaming_continuity.py index 10a065e..ea8e5dd 100644 --- a/tests/test_perception_streaming_continuity.py +++ b/tests/test_perception_streaming_continuity.py @@ -137,7 +137,7 @@ def test_active_old_callback_cannot_publish_or_be_freed_early(harness): run.check_current(run.start) # In-flight IPC can drain without poisoning the child. with pytest.raises(StreamSuspended): run.begin_input(run.start) - with pytest.raises(ValueError, match="still owns"): + with pytest.raises(StreamSuspended, match="still active"): run.begin_input(run.start) run.mailbox.release(packet, discard_reason="input-gap") epoch = run.begin_input(run.start)