test(perception): exercise lifecycle and lease loss in full profile

experiments/perception/worker/streaming_profile_stage1/pilot_lease_probe.py

@@ -0,0 +1,69 @@
+"""CPU-only cross-container lease proof; never imports or starts a GPU model."""
+
+import argparse
+import json
+import os
+import threading
+import time
+from pathlib import Path
+
+from k1link.perception.realtime_contract import StreamStart
+from k1link.perception.streaming_lifecycle import StreamingLifecycle
+from k1link.perception.streaming_queue import StreamMailbox
+from k1link.perception.worker_lease import WorkerLease, WorkerLeaseError
+
+
+def binding(generation):
+    return StreamStart(
+        run_id=f"lease-probe-{generation}",
+        source_id="synthetic",
+        worker_id="worker-006",
+        epoch_id=f"lease-probe-epoch-{generation}",
+        lease_generation=generation,
+        profile_sha256="a" * 64,
+        image_sha256="b" * 64,
+        effective_config_sha256="c" * 64,
+        calibration_sha256="d" * 64,
+        clock_domain_id="synthetic-source-clock",
+        input_mode="live",
+    )
+
+
+def run(mode):
+    root = Path("/worker-control") / (
+        "crash-case" if mode in ("crash", "quarantine") else "cross-container"
+    )
+    if mode == "crash":
+        lease = WorkerLease(root, binding(1))
+        assert not lease.released
+        print(json.dumps({"active_record_written": True, "intentional_exit": 17}), flush=True)
+        os._exit(17)
+    if mode in ("contend", "quarantine"):
+        try:
+            WorkerLease(root, binding(2))
+        except WorkerLeaseError as exc:
+            expected = "already owned" if mode == "contend" else "unretired"
+            assert expected in str(exc), str(exc)
+            print(json.dumps({"mode": mode, "denied": True, "reason": str(exc)}), flush=True)
+            return
+        raise AssertionError("a second or unverified owner was admitted")
+    lifecycle = StreamingLifecycle(
+        binding(1 if mode == "hold" else 2), root, StreamMailbox(), threading.Event()
+    )
+    try:
+        lifecycle.ready()
+        print(json.dumps({"mode": mode, "ready": True}), flush=True)
+        if mode == "hold":
+            end = time.monotonic() + 8
+            while time.monotonic() < end:
+                lifecycle.renew(lifecycle.start)
+                threading.Event().wait(0.1)
+    finally:
+        assert lifecycle.close()
+    print(json.dumps(lifecycle.snapshot()), flush=True)
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser()
+    parser.add_argument("mode", choices=("hold", "contend", "reopen", "crash", "quarantine"))
+    run(parser.parse_args().mode)

experiments/perception/worker/streaming_profile_stage1/pilot_lifecycle.py

@@ -0,0 +1,125 @@
+"""Local controller adapter for the common lifecycle, not backend integration.
+
+The launcher verifies the diagnostic image and quiesces legacy GPU clients.
+This adapter binds its bounded static metadata and heartbeat to the common
+supervisor. It does not authenticate a remote client or claim network proof.
+"""
+
+import hashlib
+import json
+import os
+import threading
+import zipfile
+from pathlib import Path
+
+from pilot_freshness import CLOCK_DOMAIN
+
+from k1link.perception.realtime_contract import StreamStart
+from k1link.perception.streaming_lifecycle import StreamingLifecycle
+from k1link.perception.worker_lease import WorkerLeaseError
+
+
+def bounded_digest(path, limit=65536):
+    with Path(path).open("rb") as stream:
+        raw = stream.read(limit + 1)
+    if len(raw) > limit:
+        raise ValueError("controller metadata exceeds bound")
+    return hashlib.sha256(raw).hexdigest()
+
+
+def calibration_digest(path):
+    digest = hashlib.sha256()
+    with zipfile.ZipFile(path) as archive:
+        for name in ("intrinsic_fx_fy_cx_cy.npy", "distortion_kb4.npy", "t_camera_from_lidar.npy"):
+            if archive.getinfo(name).file_size > 4096:
+                raise ValueError("calibration metadata exceeds bound")
+            with archive.open(name) as stream:
+                raw = stream.read(4097)
+            if len(raw) > 4096:
+                raise ValueError("calibration metadata exceeds bound")
+            digest.update(name.encode() + b"\x00" + len(raw).to_bytes(4, "little") + raw)
+    return digest.hexdigest()
+
+
+class FencedIngress:
+    def __init__(self, controller):
+        self.controller = controller
+
+    def put(self, bundle):
+        return self.controller.runtime.admit(self.controller.start, bundle)
+
+    def finish(self, error=None):
+        self.controller.runtime.mailbox.finish(error)
+
+    @property
+    def error(self):
+        return self.controller.runtime.mailbox.error
+
+
+class PilotController:
+    def __init__(self, args, report, mailbox, stop):
+        config = {
+            "pilot_options": {
+                key: value
+                for key, value in report.items()
+                if key not in ("started_utc", "started_monotonic_ns", "run_id", "requested_frames")
+            },
+            "static_config_sha256": {
+                name: bounded_digest(Path("/code/config/perception") / name)
+                for name in (
+                    "m4-geometry-association-v1.json",
+                    "m4-temporal-motion-v1.json",
+                    "m4-rolling-local-map-v1.json",
+                    "m4-replay-threat-v3.json",
+                )
+            },
+            "numeric_threads": {
+                key: os.environ.get(key)
+                for key in ("OPENBLAS_NUM_THREADS", "OMP_NUM_THREADS", "MKL_NUM_THREADS")
+            },
+        }
+        self.start = StreamStart(
+            run_id=args.run_id,
+            source_id="RAVNOVES00-20260720T065719Z-viewer-live",
+            worker_id="worker-006",
+            epoch_id=args.run_id + "-epoch-1",
+            lease_generation=args.lease_generation,
+            profile_sha256=bounded_digest("/out/candidate-profile.json"),
+            image_sha256=args.controller_image_sha256,
+            effective_config_sha256=hashlib.sha256(
+                json.dumps(config, sort_keys=True, separators=(",", ":")).encode()
+            ).hexdigest(),
+            calibration_sha256=calibration_digest("/calibration.npz"),
+            clock_domain_id=CLOCK_DOMAIN,
+            input_mode="recorded-source-paced",
+        )
+        self.runtime = StreamingLifecycle(
+            self.start, Path(args.worker_lease_root), mailbox, stop, ttl_seconds=2.0
+        )
+        self.heartbeat_stop = threading.Event()
+        self.heartbeat_error = None
+        self.thread = threading.Thread(
+            target=self._renew, name="pilot-controller-heartbeat", daemon=True
+        )
+        self.thread.start()
+        report["runtime_binding"] = self.start.to_dict()
+        report["runtime_effective_config"] = config
+        report["lease_ttl_ms"] = 2000
+        report["lease_authority_scope"] = "cooperating-runtimes-sharing-fixed-worker-volume"
+        report["heartbeat_scope"] = "same-process-pilot-controller-not-network-client"
+
+    def _renew(self):
+        while not self.heartbeat_stop.wait(0.25):
+            try:
+                self.runtime.renew(self.start)
+            except WorkerLeaseError as exc:
+                self.heartbeat_error = str(exc)
+                return
+
+    def stop_renewals(self):
+        self.heartbeat_stop.set()
+        self.thread.join(timeout=1)
+
+    def close(self, reason):
+        self.stop_renewals()
+        return self.runtime.close(reason)

experiments/perception/worker/streaming_profile_stage1/run_joint_pilot.py

@@ -19,6 +19,7 @@ import threading
 import time
 import traceback
 from collections import Counter, deque
+from contextlib import nullcontext
 from dataclasses import asdict
 from datetime import UTC, datetime
 from pathlib import Path
@@ -85,6 +86,7 @@ def produce(args, decoder, mailbox, stop, report):
     skipped_prefix = Counter()
     binding_reasons = Counter()
     preroll_history_points = 0
+    pending_camera_sequence = None
     try:
         for event in merged_events(archive, args.camera_index, args.frames):
             if stop.is_set():
@@ -110,6 +112,7 @@ def produce(args, decoder, mailbox, stop, report):
                     raise ValueError("raw rolling cloud exceeds bounded 64000-point window")
                 continue
             # Decoder reads ONE frame after its original due time, never full-decodes.
+            pending_camera_sequence = event.sequence
             decode_start = time.monotonic_ns()
             send(decoder.stdin, {"op": "next"})
             decoded, raw = receive(decoder.stdout)
@@ -160,12 +163,15 @@ def produce(args, decoder, mailbox, stop, report):
             }
             bundle["payload_bytes"] = payload_size(bundle)
             mailbox.put(bundle)
+            pending_camera_sequence = None
             fresh = []
             previous_camera_time = event.time_ns
             report["last_camera_due_ns"] = due
             if arrivals["camera"] >= args.frames:
                 break
     except Exception:
+        if pending_camera_sequence is not None:
+            report["failed_camera_sequences"] = [pending_camera_sequence]
         mailbox.finish(traceback.format_exc())
     finally:
         report.update(
@@ -272,18 +278,24 @@ def run(args):
     source_report = {}
     producer = monitor = graph = gpu_stage = ddr_backend = None
     cpu_bundle = None
+    controller = None
+    epoch_id = args.run_id
 
     def child(name, command, env=None):
         log = (output / (name + ".log")).open("wb")
         logs.append(log)
-        process = subprocess.Popen(
+
-            command,
+        def spawn():
-            stdin=subprocess.PIPE,
+            return subprocess.Popen(
-            stdout=subprocess.PIPE,
+                command,
-            stderr=log,
+                stdin=subprocess.PIPE,
-            env=env,
+                stdout=subprocess.PIPE,
-            start_new_session=True,
+                stderr=log,
-        )
+                env=env,
+                start_new_session=True,
+            )
+
+        process = controller.runtime.spawn(spawn) if controller else spawn()
         children.append(process)
         return process
 
@@ -293,6 +305,11 @@ def run(args):
     previous = signal.signal(signal.SIGALRM, timeout_handler)
     signal.alarm(240)
     try:
+        if args.worker_lease_root:
+            from pilot_lifecycle import PilotController
+
+            controller = PilotController(args, report, mailbox, stop)
+            epoch_id = controller.start.epoch_id
         triton_models = ["rf_detr_large_native_kb4"]
         if args.ddrnet_runtime == "triton":
             triton_models.append("ddrnet_goose_fp32_mask")
@@ -392,7 +409,7 @@ def run(args):
 
         ddr_cache = None
 
-        def compute_gpu(bundle):
+        def compute_gpu_impl(bundle):
             nonlocal ddr_cache
             begin = time.monotonic_ns()
             execute_ddrnet = layer_refresh_due(
@@ -440,6 +457,8 @@ def run(args):
                 }
                 mask = ddr_cache["mask"]
             ddr_done = time.monotonic_ns()
+            if controller:
+                controller.runtime.check_current(controller.start)
             proposals = graph.detector.detect(graph.packet(bundle))
             gpu_done = time.monotonic_ns()
             ddr_layer = {
@@ -450,17 +469,36 @@ def run(args):
             }
             return begin, ddr_done, gpu_done, ddr_result, mask, proposals, ddr_layer
 
+        def compute_gpu(bundle):
+            context = (
+                controller.runtime.work(controller.start, "gpu") if controller else nullcontext()
+            )
+            with context:
+                return compute_gpu_impl(bundle)
+
+        if controller:
+            controller.runtime.ready()
+
         if args.telemetry_mode == "nvml":
             monitor = threading.Thread(
                 target=telemetry, args=(stop, samples, args.telemetry_device_state), daemon=True
             )
             monitor.start()
+        ingress = mailbox
+        if controller:
+            from pilot_lifecycle import FencedIngress
+
+            ingress = FencedIngress(controller)
         producer = threading.Thread(
-            target=produce, args=(args, decoder, mailbox, stop, source_report), daemon=True
+            target=produce, args=(args, decoder, ingress, stop, source_report), daemon=True
         )
+        if controller:
+            controller.runtime.track_thread(producer)
         producer.start()
         if args.schedule == "overlap-cpu":
             gpu_stage = GpuStage(mailbox, compute_gpu, stop)
+            if controller:
+                controller.runtime.track_thread(gpu_stage.thread)
         with (output / "scenes.jsonl").open("wb") as sink:
             while True:
                 if gpu_stage:
@@ -477,12 +515,18 @@ def run(args):
                     computed = compute_gpu(bundle)
                 begin, ddr_done, gpu_done, ddr_result, mask, proposals, ddr_layer = computed
                 cpu_started = time.monotonic_ns()
-                scene = graph.process(
+                context = (
-                    bundle,
+                    controller.runtime.work(controller.start, "cpu")
-                    mask,
+                    if controller
-                    proposals=proposals,
+                    else nullcontext()
-                    detector_ms=(gpu_done - ddr_done) / 1e6,
                 )
+                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()
@@ -491,7 +535,7 @@ def run(args):
                     scene,
                     bundle,
                     ddr_layer,
-                    epoch_id=args.run_id,
+                    epoch_id=epoch_id,
                     now_ns=freshness_started,
                     mode=args.costmap_freshness,
                 )
@@ -502,6 +546,9 @@ def run(args):
                     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")
@@ -509,11 +556,15 @@ def run(args):
                 received = json.loads(encoded)
                 if received["sequence"] != bundle["sequence"]:
                     raise ValueError("collector identity mismatch")
-                freshness = validate_receipt(received, bundle, epoch_id=args.run_id)
+                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"],
@@ -567,6 +618,8 @@ def run(args):
                         "source_eof_reached": False,
                         "source_end_window_reached": mailbox.done,
                     }
+                if controller and bundle["sequence"] == args.stop_renew_after_sequence:
+                    controller.stop_renewals()
                 if len(results) % 16 == 0:
                     print(
                         json.dumps(
@@ -581,6 +634,8 @@ def run(args):
                     )
         if mailbox.error:
             raise RuntimeError(mailbox.error)
+        if controller:
+            controller.runtime.check_current(controller.start)
         report["triton_statistics_delta"] = stats_delta(stats_before, read_stats(triton_models))
         report["execution_complete"] = True
     except Exception:
@@ -590,18 +645,25 @@ def run(args):
     finally:
         signal.alarm(0)
         signal.signal(signal.SIGALRM, previous)
+        if controller:
+            controller.runtime.request_stop(
+                "completed" if report["execution_complete"] else "failed"
+            )
         stop.set()
         shutdown_start = time.monotonic_ns()
-        for process in reversed(children):
+        if controller:
-            if process.poll() is None:
+            controller.runtime.stop_children()
-                os.killpg(process.pid, signal.SIGTERM)
+        else:
-        deadline = time.monotonic() + 4
+            for process in reversed(children):
-        for process in reversed(children):
+                if process.poll() is None:
-            try:
+                    os.killpg(process.pid, signal.SIGTERM)
-                process.wait(timeout=max(0.01, deadline - time.monotonic()))
+            deadline = time.monotonic() + 4
-            except subprocess.TimeoutExpired:
+            for process in reversed(children):
-                os.killpg(process.pid, signal.SIGKILL)
+                try:
-                process.wait(timeout=1)
+                    process.wait(timeout=max(0.01, deadline - time.monotonic()))
+                except subprocess.TimeoutExpired:
+                    os.killpg(process.pid, signal.SIGKILL)
+                    process.wait(timeout=1)
         if producer:
             producer.join(timeout=2)
         if monitor:
@@ -615,6 +677,11 @@ def run(args):
             graph.backend.close()
         if ddr_backend:
             ddr_backend.close()
+        if controller:
+            report["runtime_resources_released"] = controller.close(
+                "completed" if report["execution_complete"] else "failed"
+            )
+            report["runtime_final"] = controller.runtime.snapshot()
         for log in logs:
             log.close()
         report["stop_ms"] = (time.monotonic_ns() - shutdown_start) / 1e6
@@ -650,7 +717,11 @@ def run(args):
         "released": camera_count,
         "completed": len(results),
         "dropped": mailbox.dropped_count,
-        "unaccounted": camera_count - len(results) - mailbox.dropped_count,
+        "source_failed": len(source_report.get("failed_camera_sequences", [])),
+        "unaccounted": camera_count
+        - len(results)
+        - mailbox.dropped_count
+        - len(source_report.get("failed_camera_sequences", [])),
     }
     ddrnet_executed = sum(r.get("ddrnet_state") == "current" for r in results)
     report["model_cadence"] = {
@@ -694,7 +765,8 @@ def run(args):
         "stop_5s": report["stop_ms"] <= 5000
         and report["children_stopped"]
         and report["gpu_stage_stopped"]
-        and report["input_payloads_released"],
+        and report["input_payloads_released"]
+        and report.get("runtime_resources_released", True),
         "vram_22000MiB": bool(samples)
         and max(s.get("gpu_used_mib", 99999) for s in samples) <= 22000,
         "container_memory_8192MiB": bool(samples)
@@ -757,6 +829,10 @@ if __name__ == "__main__":
     parser.add_argument("--telemetry-device-state", action="store_true")
     parser.add_argument("--triton-verbose", action="store_true")
     parser.add_argument("--ddrnet-min-source-interval-ms", type=float, default=0.0)
+    parser.add_argument("--worker-lease-root")
+    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)
     args = parser.parse_args()
     if not 1 <= args.frames <= 256:
         parser.error("pilot window must be 1..256 frames")
@@ -764,4 +840,8 @@ if __name__ == "__main__":
         parser.error("DDRNet source interval must be 0..250 ms")
     if args.telemetry_device_state and args.telemetry_mode != "nvml":
         parser.error("device-state telemetry requires NVML")
+    if bool(args.worker_lease_root) != bool(args.controller_image_sha256):
+        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")
     raise SystemExit(run(args))

tests/test_perception_joint_pilot.py

@@ -51,6 +51,29 @@ def test_active_bytes_count_towards_memory_limit(pilot):
     assert queue.take() is None
 
 
+def test_interrupted_decode_accounts_for_released_but_not_admitted_camera(pilot, monkeypatch):
+    module = pilot("run_joint_pilot")
+    event = SimpleNamespace(time_ns=1_000_000_000, channel="camera", sequence=0, value={})
+    archive = SimpleNamespace(counters=lambda: {}, close=lambda: None)
+    monkeypatch.setattr(module, "SensorArchive", lambda _path: archive)
+    monkeypatch.setattr(module, "camera_events", lambda *args: iter([event]))
+    monkeypatch.setattr(module, "merged_events", lambda *args: iter([event]))
+
+    def interrupted(*args):
+        raise EOFError("decoder stopped by supervisor")
+
+    monkeypatch.setattr(module, "send", interrupted)
+    queue = pilot("pilot_queue").Mailbox()
+    report = {}
+    stop = SimpleNamespace(is_set=lambda: False, wait=lambda _seconds: False)
+    args = SimpleNamespace(sensor_archive="unused", camera_index="unused", frames=1)
+    module.produce(args, SimpleNamespace(stdin=None), queue, stop, report)
+    assert report["arrivals"]["camera"] == 1
+    assert report["failed_camera_sequences"] == [0]
+    assert "decoder stopped" in queue.error and queue.dropped_count == 0
+    assert queue.quiescent
+
+
 def test_numpy_member_is_read_incrementally_and_bounded(pilot):
     data = io.BytesIO()
     np.savez_compressed(data, points=np.arange(300, dtype="<f8").reshape(100, 3))