feat(perception): add bounded full-graph streaming and latency pilots

experiments/perception/worker/streaming_profile_stage1/Dockerfile.dependency-probe

@@ -0,0 +1,17 @@
+# Stage-1 disposable ABI/process-isolation probe, NOT a released full profile.
+# Before building, verify both local base tags against the IDs in the report.
+FROM ndc/mission-core-installed-lab-v1-ddrnet-step:439127908dba AS goose
+FROM ndc/mission-core-m49-t3-travel:20260826 AS travel
+COPY baseline_tgs.cpp /tmp/baseline_tgs.cpp
+RUN g++ -std=c++17 -O3 -DNDEBUG -pthread \
+    -I/opt/travel/src/TRAVEL/cpp/travel/core -I/usr/include/eigen3 \
+    /tmp/baseline_tgs.cpp -o /tmp/stage1-tgs
+
+FROM nvcr.io/nvidia/tritonserver:26.06-py3
+COPY --from=goose /opt/conda /opt/conda
+COPY --from=travel /tmp/stage1-tgs /usr/local/bin/stage1-tgs
+LABEL com.nodedc.product="mission-core" \
+      com.nodedc.stack="observatory" \
+      com.nodedc.role="stage1-dependency-probe" \
+      com.nodedc.managed-by="codex-local"
+ENTRYPOINT ["/bin/bash"]

experiments/perception/worker/streaming_profile_stage1/Dockerfile.joint-pilot

@@ -0,0 +1,40 @@
+# Bounded stage-1 common-graph pilot, NOT the stage-2 standalone release.
+# Verify local base tags against the recorded image IDs before build.
+FROM ndc/mission-core-installed-lab-v1-ddrnet-step:439127908dba AS ddr
+FROM ndc/mission-core-m49-t3-travel:20260826 AS tgs
+COPY pilot_tgs.cpp /tmp/pilot_tgs.cpp
+RUN g++ -std=c++17 -O3 -DNDEBUG -pthread \
+    -I/opt/travel/src/TRAVEL/cpp/travel/core -I/usr/include/eigen3 \
+    /tmp/pilot_tgs.cpp -o /tmp/pilot-tgs
+
+FROM nvcr.io/nvidia/tritonserver:26.06-py3 AS pilot
+COPY --from=ddr /opt/conda /opt/conda
+# Exact version from repository uv.lock; only the CPU supervisor needs this.
+RUN python3 -m pip install --no-cache-dir --only-binary=:all: pillow==12.3.0
+COPY --from=tgs /tmp/pilot-tgs /usr/local/bin/pilot-tgs
+COPY source-code.tar.gz /tmp/source-code.tar.gz
+RUN mkdir /code && tar --no-same-owner -xzf /tmp/source-code.tar.gz -C /code \
+    && mkdir -p /models/rf_detr_large_native_kb4/1 /root/sg_logs
+COPY *.py /probe/
+COPY run_goose_vegetation_benchmark.py /probe/run_goose_vegetation_benchmark.py
+COPY rfdetr-config.pbtxt /models/rf_detr_large_native_kb4/config.pbtxt
+ENV PYTHONPATH=/code/src \
+    OPENBLAS_NUM_THREADS=1 OMP_NUM_THREADS=1 MKL_NUM_THREADS=1 \
+    MPLCONFIGDIR=/tmp/matplotlib
+LABEL com.nodedc.product="mission-core" com.nodedc.stack="observatory" \
+      com.nodedc.role="stage1-joint-pilot" com.nodedc.managed-by="codex-local"
+ENTRYPOINT ["python3", "-B", "/probe/run_joint_pilot.py"]
+
+# Bounded local fallback for a daemon with missing compressed registry blobs.
+# Uses only already-built files; no new registry access or credential change.
+# This overlay is a diagnostic pilot, NEVER a standalone image release.
+FROM scratch AS pilot-assets
+COPY --from=pilot /probe /probe
+COPY --from=pilot /code /code
+COPY --from=pilot /models /models
+COPY --from=pilot /usr/local/bin/pilot-tgs /bin/pilot-tgs
+COPY --from=pilot /usr/local/lib/python3.12/dist-packages/PIL /python/PIL
+COPY --from=pilot /usr/local/lib/python3.12/dist-packages/pillow.libs /python/pillow.libs
+
+# Default docker build remains a runnable image, not the diagnostic overlay.
+FROM pilot AS joint-pilot

experiments/perception/worker/streaming_profile_stage1/baseline_ddrnet.py

@@ -0,0 +1,127 @@
+#!/usr/bin/env python3
+"""Bounded component baseline, NOT full-profile or transport qualification.
+
+Run only in the pinned installed DDRNet image on an exclusive Worker. The
+existing video is decoded incrementally; it is not copied, hashed end-to-end,
+cached as PNGs or fully decoded before the first measured frame.
+"""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import importlib.machinery
+import importlib.util
+import json
+import signal
+import statistics
+import time
+from datetime import datetime, timezone
+from pathlib import Path
+
+
+def digest(path: Path) -> str:
+    result = hashlib.sha256()
+    with path.open("rb") as stream:
+        for chunk in iter(lambda: stream.read(1024 * 1024), b""):
+            result.update(chunk)
+    return result.hexdigest()
+
+
+def distribution(values: list[float]) -> dict[str, float]:
+    ordered = sorted(values)
+    return {
+        "mean": statistics.fmean(ordered),
+        "p50": ordered[int((len(ordered) - 1) * 0.50)],
+        "p95": ordered[int((len(ordered) - 1) * 0.95)],
+        "p99": ordered[int((len(ordered) - 1) * 0.99)],
+        "maximum": max(ordered),
+    }
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--video", type=Path, required=True)
+    parser.add_argument("--runner", type=Path, required=True)
+    parser.add_argument("--checkpoint", type=Path, required=True)
+    parser.add_argument("--samples", type=int, default=64, choices=range(8, 129))
+    parser.add_argument("--warmup", type=int, default=8, choices=range(1, 17))
+    args = parser.parse_args()
+    signal.alarm(180)
+    started = time.monotonic_ns()
+    utc = datetime.now(timezone.utc).isoformat()
+    runner = args.runner
+    checkpoint = args.checkpoint
+    expected_checkpoint = "b99c2838051bcd7b092fd3970aa62a77d5c0bbb809c9b9afb2ff4b0ebdaa4ee6"
+    if digest(checkpoint) != expected_checkpoint:
+        raise RuntimeError("checkpoint identity changed")
+    if digest(runner) != "b18ad60f277eea69a240a28f290611b94627fb9707faf1bb3e6e22102dad67c1":
+        raise RuntimeError("reused runner identity changed")
+    loader = importlib.machinery.SourceFileLoader("goose_baseline_runner", str(runner))
+    spec = importlib.util.spec_from_loader(loader.name, loader)
+    if spec is None:
+        raise RuntimeError("runner is unavailable")
+    module = importlib.util.module_from_spec(spec)
+    loader.exec_module(module)
+    import cv2
+    import torch
+    from PIL import Image
+
+    capture = cv2.VideoCapture(str(args.video))
+    try:
+        ok, bgr = capture.read()
+        if not ok or bgr is None or bgr.shape != (600, 800, 3):
+            raise RuntimeError("first native KB4 frame unavailable")
+        tensor, _ = module.preprocess(Image.fromarray(cv2.cvtColor(bgr, cv2.COLOR_BGR2RGB)))
+        model, name, _ = module.load_model("ddrnet", checkpoint)
+        for _ in range(args.warmup):
+            module.infer(model, tensor)
+        warm_ns = time.monotonic_ns()
+        torch.cuda.reset_peak_memory_stats()
+        rows = []
+        mask_digest = hashlib.sha256()
+        for sequence in range(args.samples):
+            begin = time.monotonic_ns()
+            ok, bgr = capture.read()
+            decoded = time.monotonic_ns()
+            if not ok or bgr is None or bgr.shape != (600, 800, 3):
+                raise RuntimeError("bounded source ended or changed shape")
+            tensor, _ = module.preprocess(Image.fromarray(cv2.cvtColor(bgr, cv2.COLOR_BGR2RGB)))
+            preprocessed = time.monotonic_ns()
+            mask, forward_ms = module.infer(model, tensor)
+            completed = time.monotonic_ns()
+            mask_digest.update(mask.tobytes())
+            rows.append({
+                "sequence": sequence + 1,
+                "decode_ms": (decoded - begin) / 1e6,
+                "preprocess_ms": (preprocessed - decoded) / 1e6,
+                "h2d_forward_post_d2h_ms": (completed - preprocessed) / 1e6,
+                "forward_post_ms": forward_ms,
+                "component_total_ms": (completed - begin) / 1e6,
+            })
+        result = {
+            "schema_version": "missioncore.perception-component-baseline/v1",
+            "component": "ddrnet", "started_utc": utc,
+            "started_monotonic_ns": started, "finished_monotonic_ns": time.monotonic_ns(),
+            "checkpoint_sha256": expected_checkpoint, "runner_sha256": digest(runner),
+            "model": name, "torch": torch.__version__, "cuda": torch.version.cuda,
+            "gpu": torch.cuda.get_device_name(0), "precision": "existing-fp32",
+            "samples": args.samples, "warmup_iterations": args.warmup,
+            "load_first_decode_warmup_seconds": (warm_ns - started) / 1e9,
+            "source_whole_file_copied_or_hashed": False,
+            "source_frames_decoded": args.samples + 1,
+            "source_paced": False, "full_profile_qualified": False,
+            "mask_sequence_sha256": mask_digest.hexdigest(),
+            "peak_allocated_vram_mib": torch.cuda.max_memory_allocated() / 1048576,
+            "peak_reserved_vram_mib": torch.cuda.max_memory_reserved() / 1048576,
+            "timing_ms": {key: distribution([row[key] for row in rows]) for key in rows[0]
+                          if key != "sequence"},
+            "frames": rows,
+        }
+        print("STAGE1_RESULT=" + json.dumps(result, sort_keys=True), flush=True)
+    finally:
+        capture.release()
+
+
+if __name__ == "__main__":
+    main()

experiments/perception/worker/streaming_profile_stage1/baseline_surface.py

@@ -0,0 +1,32 @@
+"""CPU component probe using synthetic, explicitly stationary geometry only."""
+
+import json
+import time
+
+import numpy as np
+
+from k1link.compute.lidar_local_surface_shadow import (
+    K1LocalSurfaceShadowEstimator,
+    K1LocalSurfaceShadowInput,
+)
+
+random = np.random.default_rng(20260901)
+estimator = K1LocalSurfaceShadowEstimator()
+rows = []
+for index in range(64):
+    xy = random.uniform(-8, 8, (4000, 2))
+    z = random.normal(0, 0.015, (4000, 1))
+    z[3800:] += 1.0
+    value = K1LocalSurfaceShadowInput(
+        frame_index=index, source_frame_index=index, session_seconds=index / 10,
+        pose_binding_age_ms=0, points_map=np.column_stack((xy, z)),
+        position_map=np.zeros(3), published_monotonic_ns=time.monotonic_ns(),
+    )
+    started = time.monotonic_ns()
+    result = estimator.process(value)
+    rows.append({"sequence": index, "wall_ms": (time.monotonic_ns() - started) / 1e6,
+                 "state": result.state})
+print(json.dumps({"schema_version": "missioncore.perception-component-baseline/v1",
+                  "component": "online-local-surface", "synthetic": True,
+                  "points_per_frame": 4000, "frames": rows,
+                  "full_profile_qualified": False}, sort_keys=True))

experiments/perception/worker/streaming_profile_stage1/baseline_tgs.cpp

@@ -0,0 +1,61 @@
+// Component-only CPU measurement on 64 existing prepared rolling clouds.
+// Does NOT measure source decoding, rolling input preparation, costmap or transport.
+#include <chrono>
+#include <cmath>
+#include <fstream>
+#include <iomanip>
+#include <iostream>
+#include <sstream>
+#include <stdexcept>
+#include <string>
+#include "travel/point_types.hpp"
+#include "travel/tgs.hpp"
+
+struct QuietLibraryOutput {
+    std::ostringstream buffer;
+    std::streambuf* previous;
+    QuietLibraryOutput() : previous(std::cout.rdbuf(buffer.rdbuf())) {}
+    ~QuietLibraryOutput() { std::cout.rdbuf(previous); }
+};
+
+int main(int argc, char** argv) {
+    if (argc != 2) return 2;
+    using Clock = std::chrono::steady_clock;
+    try {
+        std::cout << "sequence,points,ground,nonground,wall_ms,algorithm_ms\n";
+        for (int index = 0; index < 64; ++index) {
+            std::ostringstream path;
+            path << argv[1] << '/' << std::setw(6) << std::setfill('0') << index << ".bin";
+            std::ifstream file(path.str(), std::ios::binary);
+            if (!file) throw std::runtime_error("bounded cloud unavailable");
+            travel::PointCloud<PointXYZILID> input;
+            float row[4];
+            while (file.read(reinterpret_cast<char*>(row), sizeof(row))) {
+                if (input.size() >= 500000) throw std::runtime_error("cloud exceeds point budget");
+                PointXYZILID point{};
+                point.x = row[0]; point.y = row[1]; point.z = row[2]; point.intensity = row[3];
+                if (!std::isfinite(point.x) || !std::isfinite(point.y) || !std::isfinite(point.z))
+                    throw std::runtime_error("nonfinite cloud");
+                input.emplace_back(point);
+            }
+            if (file.gcount() || input.empty()) throw std::runtime_error("invalid cloud bytes");
+            const auto start = Clock::now();
+            travel::PointCloud<PointXYZILID> ground, nonground;
+            double algorithm_seconds = 0;
+            {
+                QuietLibraryOutput quiet;
+                travel::TravelGroundSeg<PointXYZILID> tgs;
+                tgs.setParams(80.0, 1.0, 8.0, 3, 5, 10, 0.5, 0.125, 0.3, 0.940,
+                              200.0, 0.03, 0.1, 1.0, true, false);
+                tgs.estimateGround(input, ground, nonground, algorithm_seconds);
+            }
+            const auto end = Clock::now();
+            const auto wall = std::chrono::duration<double, std::milli>(end - start).count();
+            std::cout << index << ',' << input.size() << ',' << ground.size() << ','
+                      << nonground.size() << ',' << wall << ',' << algorithm_seconds * 1000 << '\n';
+        }
+    } catch (const std::exception& error) {
+        std::cerr << error.what() << '\n';
+        return 1;
+    }
+}

experiments/perception/worker/streaming_profile_stage1/pilot_ddrnet_runtime.py

@@ -0,0 +1,138 @@
+"""Measured FP32 DDRNet execution; preserves sigmoid and first-index ties.
+
+An experimental memory-layout variant, not a new weight/precision/label model.
+Python 3.9 compatible. GPU calls remain serialized by the owning supervisor.
+"""
+
+import time
+
+
+def layout_scores(scores, layout):
+    if layout == "reference":
+        return scores, 1
+    if layout == "channels-last":
+        return scores.permute(0, 2, 3, 1).contiguous(), 3
+    raise ValueError("unknown DDRNet postprocess layout")
+
+
+class DdrnetRuntime:
+    def __init__(self, model, logits_from_output, layout, execution="eager"):
+        import torch
+
+        if layout not in ("reference", "channels-last"):
+            raise ValueError("unknown DDRNet postprocess layout")
+        if execution not in ("eager", "cuda-graph"):
+            raise ValueError("unknown DDRNet execution mode")
+        self.torch = torch
+        self.model = model
+        self.logits_from_output = logits_from_output
+        self.layout = layout
+        self.execution = execution
+        self.graph = None
+        self.events = [torch.cuda.Event(enable_timing=True) for _ in range(5)]
+
+    def validate_ties(self):
+        torch = self.torch
+        # Sigmoid saturates 100 and 101 to the same score. Argmax(logits)
+        # would change the original label; the layout optimization must not.
+        logits = torch.tensor(
+            [
+                [100.0, 0.0, -100.0, 4.0],
+                [101.0, 0.0, -99.0, 4.0],
+                [-100.0, 0.0, 3.0, 0.0],
+                [-9.0, 0.0, -9.0, 0.0],
+            ],
+            dtype=torch.float32,
+            device="cuda",
+        ).reshape(1, 4, 2, 2)
+        scores = torch.sigmoid(logits)
+        reference = torch.argmax(scores, dim=1)
+        changed, dimension = layout_scores(scores, "channels-last")
+        candidate = torch.argmax(changed, dim=dimension)
+        valid = bool(torch.equal(reference, candidate) and reference[0, 0, 0].item() == 0)
+        if not valid:
+            raise RuntimeError("DDRNet layout changed saturated/tied-score labels")
+        return valid
+
+    def infer(self, tensor):
+        if self.execution == "cuda-graph":
+            return self.infer_graph(tensor)
+        torch = self.torch
+        h2d_started = time.perf_counter_ns()
+        tensor = tensor.cuda(non_blocking=True)
+        torch.cuda.synchronize()
+        started = time.perf_counter_ns()
+        with torch.inference_mode():
+            self.events[0].record()
+            logits = self.logits_from_output(self.model(tensor))
+            self.events[1].record()
+            scores = torch.sigmoid(logits)
+            self.events[2].record()
+            scores, dimension = layout_scores(scores, self.layout)
+            self.events[3].record()
+            prediction = torch.argmax(scores, dim=dimension)
+            self.events[4].record()
+        torch.cuda.synchronize()
+        predicted = time.perf_counter_ns()
+        mask = prediction[0].to(device="cpu", dtype=torch.uint8).numpy()
+        copied = time.perf_counter_ns()
+        names = ("model_gpu_ms", "sigmoid_gpu_ms", "layout_gpu_ms", "argmax_gpu_ms")
+        stages = {
+            name: self.events[i].elapsed_time(self.events[i + 1]) for i, name in enumerate(names)
+        }
+        stages.update(
+            h2d_sync_wall_ms=(started - h2d_started) / 1e6,
+            d2h_wall_ms=(copied - predicted) / 1e6,
+            forward_post_wall_ms=(predicted - started) / 1e6,
+        )
+        return mask, (predicted - started) / 1e6, stages
+
+    def _compute(self, tensor):
+        logits = self.logits_from_output(self.model(tensor))
+        scores, dimension = layout_scores(self.torch.sigmoid(logits), self.layout)
+        return self.torch.argmax(scores, dim=dimension)
+
+    def _capture(self, tensor):
+        torch = self.torch
+        # Capture only fixed-shape DDRNet work, never source reads, RF-DETR,
+        # host decisions, or another profile. Capture is local to this GPU/run.
+        self.static_input = tensor.cuda().clone()
+        stream = torch.cuda.Stream()
+        stream.wait_stream(torch.cuda.current_stream())
+        with torch.cuda.stream(stream), torch.inference_mode():
+            for _ in range(3):
+                self._compute(self.static_input)
+        torch.cuda.current_stream().wait_stream(stream)
+        torch.cuda.synchronize()
+        self.graph = torch.cuda.CUDAGraph()
+        with torch.cuda.graph(self.graph), torch.inference_mode():
+            self.static_prediction = self._compute(self.static_input)
+        torch.cuda.synchronize()
+
+    def infer_graph(self, tensor):
+        torch = self.torch
+        if self.graph is None:
+            self._capture(tensor)
+        if tensor.shape != self.static_input.shape or tensor.dtype != self.static_input.dtype:
+            raise ValueError("DDRNet CUDA graph input shape/dtype changed")
+        h2d_started = time.perf_counter_ns()
+        self.static_input.copy_(tensor, non_blocking=True)
+        torch.cuda.synchronize()
+        started = time.perf_counter_ns()
+        self.events[0].record()
+        self.graph.replay()
+        self.events[1].record()
+        torch.cuda.synchronize()
+        predicted = time.perf_counter_ns()
+        mask = self.static_prediction[0].to(device="cpu", dtype=torch.uint8).numpy()
+        copied = time.perf_counter_ns()
+        return (
+            mask,
+            (predicted - started) / 1e6,
+            {
+                "cuda_graph_gpu_ms": self.events[0].elapsed_time(self.events[1]),
+                "h2d_sync_wall_ms": (started - h2d_started) / 1e6,
+                "d2h_wall_ms": (copied - predicted) / 1e6,
+                "forward_post_wall_ms": (predicted - started) / 1e6,
+            },
+        )

experiments/perception/worker/streaming_profile_stage1/pilot_graph.py

@@ -0,0 +1,451 @@
+"""Joint-pilot wiring of existing algorithms to current, causal source data.
+
+Experiment adapter only: no installed LAB or product registry mutation.
+Historical profile files supply algorithm parameters, never derived evidence.
+"""
+
+from __future__ import annotations
+
+import csv
+import hashlib
+import importlib.util
+import struct
+import sys
+import time
+import zipfile
+from collections import Counter, OrderedDict
+from dataclasses import asdict
+from pathlib import Path
+
+import numpy as np
+from pilot_ipc import exact
+
+from k1link.compute.lidar_local_surface_shadow import (
+    K1LocalSurfaceShadowEstimator,
+    K1LocalSurfaceShadowInput,
+)
+from k1link.perception.contracts import (
+    ClockBasis,
+    LocalObstacleMap,
+    ModalityOutcome,
+    ModalityStatus,
+    SourceAccounting,
+    SourceEnvelope,
+    TemporalState,
+    TimestampBundle,
+)
+from k1link.perception.detector import NativeRfDetrShadowDetectorProvider
+from k1link.perception.geometry import (
+    GeometryFrame,
+    Ravnoves00GeometryAssociationProvider,
+    load_geometry_profile,
+)
+from k1link.perception.geometry_math import (
+    Kb4ProjectionProfile,
+    project_map_points_kb4,
+    quaternion_xyzw_to_rotation_matrix,
+)
+from k1link.perception.motion import ClassIndependentMotionEstimator
+from k1link.perception.providers import SourcePacket
+from k1link.perception.rf_detr_native_object_detector import TritonNativeRfDetrHttpInferenceBackend
+from k1link.perception.rolling_map import RollingLocalObstacleMapProvider, load_rolling_map_profile
+from k1link.perception.temporal import BoundedSpatialTemporalProvider, load_temporal_motion_profile
+from k1link.perception.threat import (
+    DualEvidenceReplayThreatProvider,
+    ReplayBodyFrame,
+    load_replay_threat_profile,
+)
+from k1link.perception.yolox_object_detector import load_valid_fov_mask
+
+
+def module_from_file(name, path):
+    spec = importlib.util.spec_from_file_location(name, path)
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[name] = module
+    spec.loader.exec_module(module)
+    return module
+
+
+def calibration(path):
+    # Read only the three small static arrays, never the archive's point data.
+    with zipfile.ZipFile(path) as archive:
+        arrays = {}
+        for name in ("intrinsic_fx_fy_cx_cy", "distortion_kb4", "t_camera_from_lidar"):
+            if archive.getinfo(name + ".npy").file_size > 4096:
+                raise ValueError("calibration exceeds metadata budget")
+            with archive.open(name + ".npy") as stream:
+                arrays[name] = np.lib.format.read_array(stream, allow_pickle=False)
+    return Kb4ProjectionProfile(
+        800,
+        600,
+        tuple(arrays["intrinsic_fx_fy_cx_cy"]),
+        tuple(arrays["distortion_kb4"]),
+        arrays["t_camera_from_lidar"],
+    )
+
+
+def grid_indices(point_cells, grid):
+    """Exact equivalent of per-point tuple/dict lookup, including holes."""
+    grid_xy = grid[:, :2].astype(np.int64)
+    lower = grid_xy.min(axis=0)
+    shape = grid_xy.max(axis=0) - lower + 1
+    lookup = np.full(tuple(shape), -1, np.int64)
+    offset = grid_xy - lower
+    lookup[offset[:, 0], offset[:, 1]] = np.arange(len(grid))
+    relative = point_cells - lower
+    inside = np.all((relative >= 0) & (relative < shape), axis=1)
+    result = np.full(len(point_cells), -1, np.int64)
+    result[inside] = lookup[relative[inside, 0], relative[inside, 1]]
+    return result
+
+
+class CurrentStore:
+    """One current frame, no NPZ store, historical lookup or future trajectory."""
+
+    def __init__(self, profile):
+        self.profile = profile
+        self.current = None
+        self.pose = None
+        self.body = None
+        self.frame_id = None
+        self.body_history = OrderedDict()
+
+    def frame(self, packet):
+        if packet.envelope.frame_id != self.frame_id:
+            raise ValueError("online geometry frame identity mismatch")
+        return self.current
+
+    def current_points(self, packet):
+        frame = self.frame(packet)
+        return frame.points_map if frame is not None and frame.surface_valid else None
+
+    def pose_values_for_frame(self, frame_id):
+        if frame_id != self.frame_id:
+            raise ValueError("past/future pose lookup forbidden")
+        return self.pose
+
+    def body_frame_for_frame(self, frame_id):
+        if frame_id > self.frame_id:
+            raise ValueError("future body lookup forbidden")
+        return self.body_history.get(frame_id)
+
+    def remember_body(self):
+        self.body_history[self.frame_id] = self.body
+        while len(self.body_history) > 64:
+            self.body_history.popitem(last=False)
+
+
+def current_body(frame, surface):
+    if (
+        not surface.valid
+        or surface.sensor_height_m is None
+        or abs(surface.sensor_height_m - 1.25) > 0.45
+        or surface.slope_deg is None
+        or surface.slope_deg > 10.0
+    ):
+        return None
+    rotation = quaternion_xyzw_to_rotation_matrix(frame.sensor_orientation_xyzw)
+    forward = rotation @ frame.projection.t_camera_from_lidar[2, :3]
+    forward[2] = 0
+    norm = np.linalg.norm(forward)
+    normal = np.asarray(surface.plane_coefficients_map[:3])
+    if norm < 1e-9 or abs(normal[2]) < 1e-6:
+        return None
+    forward /= norm
+    up = np.array([0.0, 0.0, 1.0])
+    left = np.cross(up, forward)
+    origin = frame.sensor_position_map - up * (surface.sensor_height_m / abs(normal[2]))
+    basis = np.column_stack((forward, left, up))
+    return ReplayBodyFrame(
+        f"frame-{frame.frame_index:06d}",
+        tuple(origin),
+        tuple(tuple(row) for row in basis),
+        surface.sensor_height_m,
+        surface.slope_deg,
+        "causal-camera-forward-simulation",
+        0.0,
+    )
+
+
+class JointGraph:
+    def __init__(self, code_root, tgs, *, mask_path, mapping_path, calibration_pack):
+        root = Path(code_root)
+        config = root / "config/perception"
+        self.projection = calibration(calibration_pack)
+        self.surface = K1LocalSurfaceShadowEstimator()
+        self.store = CurrentStore(load_geometry_profile(config / "m4-geometry-association-v1.json"))
+        self.geometry = Ravnoves00GeometryAssociationProvider(store=self.store)
+        temporal_profile = load_temporal_motion_profile(config / "m4-temporal-motion-v1.json")
+        self.temporal = BoundedSpatialTemporalProvider(
+            point_resolver=self.store, profile=temporal_profile
+        )
+        self.motion = ClassIndependentMotionEstimator(profile=temporal_profile)
+        self.rolling = RollingLocalObstacleMapProvider(
+            pose_resolver=self.store,
+            profile=load_rolling_map_profile(config / "m4-rolling-local-map-v1.json"),
+        )
+        self.threat = DualEvidenceReplayThreatProvider(
+            body_frame_resolver=self.store,
+            profile=load_replay_threat_profile(config / "m4-replay-threat-v3.json"),
+        )
+        self.backend = TritonNativeRfDetrHttpInferenceBackend(
+            "http://127.0.0.1:8000", timeout_seconds=5
+        )
+        self.detector = NativeRfDetrShadowDetectorProvider(
+            mask=load_valid_fov_mask(Path(mask_path)), backend=self.backend
+        )
+        self.tgs = tgs
+        scripts = root / "experiments/perception/worker/m49_t3_travel"
+        sys.path.insert(0, str(scripts))
+        from build_tgs_fail_closed_evidence import costmap_grid
+        from build_tgs_full_shadow_evidence import rasterize
+
+        self.rasterize = rasterize
+        self.grid = costmap_grid(12.0, 0.45)
+        self.cell_lookup = {(int(row[0]), int(row[1])): i for i, row in enumerate(self.grid)}
+        policy_module = module_from_file(
+            "pilot_vegetation_policy", root / "src/k1link/laboratory/vegetation_mission_policy.py"
+        )
+        self.policy = policy_module.load_vegetation_mission_policy(
+            config / "lab-v1-vegetation-mission-policy-v1.json", repository_root=root
+        )
+        mapping = policy_module.load_vegetation_provider_label_map(
+            config / "lab-v1-vegetation-provider-label-map-v1.json", policy=self.policy
+        )
+        if (
+            hashlib.sha256(Path(mapping_path).read_bytes()).hexdigest()
+            != "88ae319ba5a3877dd3ae0773f693a6a5fdc283934140de9dfaff029108aefd7f"
+        ):
+            raise ValueError("GOOSE class mapping changed")
+        with Path(mapping_path).open(encoding="utf-8-sig", newline="") as stream:
+            labels = {int(row["label_key"]): row["class_name"] for row in csv.DictReader(stream)}
+        if set(labels) != set(range(64)):
+            raise ValueError("GOOSE mapping incomplete")
+        material_names = ["unknown", *self.policy["material_classes"]]
+        provider_map = mapping["providers"]["goose-fine-64"]["labels"]
+        self.material_names = material_names
+        self.material_lut = np.asarray(
+            [material_names.index(provider_map.get(labels[i], "unknown")) for i in range(64)]
+        )
+        self.action_lut = np.full(
+            (len(material_names), 4), 2, np.uint8
+        )  # 0=ALLOW,1=HIGH_COST,2=NO_GO
+        for i, material in enumerate(material_names):
+            decision = policy_module.resolve_terrain_policy(
+                self.policy,
+                preset_id="rural",
+                material_class=None if material == "unknown" else material,
+                evidence_state="SUPPORTED_GROUND",
+            )
+            self.action_lut[i, 1] = {"ALLOW": 0, "HIGH_COST": 1, "NO_GO": 2}[
+                decision.effective_action
+            ]
+            # Pilot intent: ONLY coarse hard_surface can be an ALLOW candidate.
+            # This additional conservative interlock cannot weaken legacy rules.
+            if material != "hard_surface":
+                self.action_lut[i, :] = 2
+
+    def packet(self, bundle):
+        available = bundle["available"]
+        status = ModalityStatus(
+            available,
+            ModalityOutcome.AVAILABLE if available else ModalityOutcome.UNAVAILABLE,
+            "causal-source-bound" if available else "no-fresh-past-cloud-or-pose",
+        )
+        env = SourceEnvelope(
+            source_id="RAVNOVES00",
+            session_id="20260720T065719Z_viewer_live",
+            frame_id=f"frame-{bundle['sequence']:06d}",
+            sequence=bundle["sequence"],
+            timestamps=TimestampBundle(
+                utc_ns=bundle["utc_ns"],
+                monotonic_ns=bundle["time_ns"],
+                source_ns=bundle["source_ns"],
+                clock_basis=ClockBasis.RECORDED_HOST,
+            ),
+            source_age_ns=0,
+            binding_reason="past-only-source-arrival",
+            calibration_id="camera-1-kb4-05f3ad9b",
+            representation_id=self.rolling.profile.representation_id,
+            image=ModalityStatus(True, ModalityOutcome.AVAILABLE, "current-camera"),
+            registered_point_increment=status,
+            pose=status,
+        )
+        return SourcePacket(
+            env,
+            bundle["image"],
+            bundle["points"] if available else None,
+            bundle["pose"] if available else None,
+        )
+
+    def process(self, bundle, segmentation, *, proposals=None, detector_ms=None):
+        packet = self.packet(bundle)
+        timing = {}
+        if proposals is None:
+            begin = time.monotonic_ns()
+            proposals = self.detector.detect(packet)
+            detector_ms = (time.monotonic_ns() - begin) / 1e6
+        timing["detector_ms"] = detector_ms
+        self.store.frame_id = packet.envelope.frame_id
+        self.store.current = self.store.pose = self.store.body = None
+        surface = None
+        begin = time.monotonic_ns()
+        if bundle["available"]:
+            position, quaternion = bundle["pose"]
+            surface = self.surface.process(
+                K1LocalSurfaceShadowInput(
+                    frame_index=bundle["sequence"],
+                    source_frame_index=bundle["sequence"],
+                    session_seconds=bundle["source_ns"] / 1e9,
+                    pose_binding_age_ms=bundle["binding_age_ms"],
+                    points_map=bundle["points"],
+                    position_map=position,
+                    published_monotonic_ns=bundle["due_ns"],
+                )
+            )
+            self.store.current = GeometryFrame(
+                bundle["sequence"],
+                bundle["points"],
+                surface.point_class,
+                position,
+                quaternion,
+                self.projection,
+                surface.valid,
+            )
+            self.store.pose = (tuple(position), tuple(quaternion))
+            self.store.body = current_body(self.store.current, surface)
+        self.store.remember_body()
+        timing["online_surface_ms"] = (time.monotonic_ns() - begin) / 1e6
+        begin = time.monotonic_ns()
+        observations = self.geometry.associate(packet, proposals)
+        timing["association_ms"] = (time.monotonic_ns() - begin) / 1e6
+        begin = time.monotonic_ns()
+        temporal = self.temporal.update(packet, observations)
+        moving = self.motion.estimate(packet, temporal)
+        retained = self.rolling.update(packet, moving)
+        current = tuple(item for item in moving if item.state is TemporalState.CURRENT)
+        unknown = tuple(item for item in moving if item.state is not TemporalState.CURRENT)
+        associated = {
+            pid
+            for item in observations
+            if item.metric_geometry is not None
+            for pid in item.proposal_ids
+        }
+        scene_map = LocalObstacleMap(
+            "RAVNOVES00",
+            "20260720T065719Z_viewer_live",
+            packet.envelope.frame_id,
+            "stage1-joint-online-pilot/v1",
+            time.monotonic_ns(),
+            max(0, time.monotonic_ns() - bundle["due_ns"]),
+            (*current, *retained),
+            unknown,
+            tuple(p for p in proposals if p.proposal_id not in associated),
+            SourceAccounting(1, 1, 0, 0),
+        )
+        threats = self.threat.assess(scene_map)
+        timing["motion_rolling_threat_ms"] = (time.monotonic_ns() - begin) / 1e6
+        begin = time.monotonic_ns()
+        costmap, actions, material, tgs_counts = self.costmap(bundle, segmentation, scene_map)
+        timing["tgs_costmap_policy_ms"] = (time.monotonic_ns() - begin) / 1e6
+        return {
+            "timing_ms": timing,
+            "surface_state": surface.state if surface else "unavailable",
+            "segmentation_sha256": hashlib.sha256(segmentation.tobytes()).hexdigest(),
+            "proposals": [asdict(x) for x in proposals],
+            "observations": [asdict(x) for x in observations],
+            "tracks": [asdict(x) for x in moving],
+            "threats": [asdict(x) for x in threats],
+            "tgs_counts": tgs_counts,
+            "costmap_states": costmap.tolist(),
+            "costmap_material": material.tolist(),
+            "policy_actions": actions.tolist(),
+            "policy_counts": {
+                key: int(np.count_nonzero(actions == value))
+                for key, value in (("ALLOW_candidate", 0), ("HIGH_COST", 1), ("NO_GO", 2))
+            },
+            "range_estimator": {
+                "detected": "median-camera-z",
+                "geometry_only": "nearest-euclidean-sensor",
+            },
+            "commands_enabled": False,
+            "actuation_allowed": False,
+        }
+
+    def costmap(self, bundle, mask, obstacle_map):
+        count = len(self.grid)
+        material = np.zeros(count, np.int32)
+        if not bundle["available"]:
+            return (
+                np.zeros(count, np.uint8),
+                np.full(count, 2, np.uint8),
+                material,
+                {"unavailable": True},
+            )
+        position, quaternion = bundle["pose"]
+        points = bundle["rolling_points"]
+        stamps = bundle["rolling_times"]
+        local = points - position
+        keep = np.sum(local[:, :2] ** 2, axis=1) <= 144.0
+        local, points, stamps = local[keep], points[keep], stamps[keep]
+        if not 0 < len(local) <= 64000:
+            raise ValueError("TGS rolling input outside pilot bound")
+        native = np.column_stack((local, np.zeros(len(local)))).astype("<f4")
+        self.tgs.stdin.write(struct.pack("<I", len(local)) + native.tobytes())
+        self.tgs.stdin.flush()
+        n, algorithm_ms = struct.unpack("<Id", exact(self.tgs.stdout, 12))
+        if n != len(local):
+            raise ValueError("TGS point accounting changed")
+        states = np.frombuffer(exact(self.tgs.stdout, n), np.uint8)
+        if np.any((states < 1) | (states > 3)):
+            raise ValueError("TGS returned invalid state")
+        cells, _ = self.rasterize(native[:, :3], states, self.grid, 0.45)
+        point_cells = np.floor(local[:, :2] / 0.45).astype(np.int32)
+        ids = grid_indices(point_cells, self.grid)
+        valid = ids >= 0
+        last_seen = np.full(count, -1, np.int64)
+        np.maximum.at(last_seen, ids[valid], stamps[valid])
+        stale = (last_seen >= 0) & (bundle["time_ns"] - last_seen > 250_000_000)
+        cells[stale & (cells != 2)] = 3
+        projected = project_map_points_kb4(
+            points,
+            position_map_xyz=position,
+            orientation_map_from_lidar_xyzw=quaternion,
+            profile=self.projection,
+        )
+        uv = projected.pixels_xy
+        crop = (uv[:, 0] >= 100) & (uv[:, 0] < 700) & (uv[:, 1] >= 0) & (uv[:, 1] < 600)
+        point_ids = projected.source_indices[crop]
+        uv = uv[crop]
+        labels = mask[
+            (uv[:, 1] * 512 / 600).astype(int), ((uv[:, 0] - 100) * 512 / 600).astype(int)
+        ]
+        votes = np.zeros((count, len(self.material_names)), np.int32)
+        cell_ids = ids[point_ids]
+        good = (cell_ids >= 0) & (states[point_ids] == 1)
+        np.add.at(votes, (cell_ids[good], self.material_lut[labels[good]]), 1)
+        material = votes.argmax(axis=1).astype(np.int32)
+        # Current and retained geometry can only add a prohibition, never clear TGS.
+        for obstacle in (*obstacle_map.occupied, *obstacle_map.unknown):
+            for cell in obstacle.cells:
+                xy = (
+                    np.asarray((cell.x, cell.y)) + 0.5
+                ) * self.temporal.config.voxel_size_m - position[:2]
+                index = self.cell_lookup.get(tuple(np.floor(xy / 0.45).astype(int)))
+                if index is not None:
+                    cells[index] = 2
+        actions = self.action_lut[material, cells]
+        return (
+            cells,
+            actions,
+            material,
+            {
+                "points": n,
+                "algorithm_ms": algorithm_ms,
+                "ground": int(np.count_nonzero(states == 1)),
+                "occupied": int(np.count_nonzero(states == 2)),
+                "rejected": int(np.count_nonzero(states == 3)),
+                "stale_cells": int(np.count_nonzero(stale)),
+                "state_counts": dict(Counter(int(x) for x in cells)),
+            },
+        )

experiments/perception/worker/streaming_profile_stage1/pilot_ipc.py

@@ -0,0 +1,41 @@
+"""Bounded local probe RPC, compatible with Python 3.9 and 3.12."""
+
+from __future__ import annotations
+
+import json
+import struct
+
+MAX_PAYLOAD = 16 * 1024 * 1024
+
+
+def exact(stream, size):
+    if not 0 <= size <= MAX_PAYLOAD:
+        raise ValueError("IPC read exceeds budget")
+    chunks = bytearray()
+    while len(chunks) < size:
+        part = stream.read(size - len(chunks))
+        if not part:
+            raise EOFError("truncated IPC message")
+        chunks.extend(part)
+    return bytes(chunks)
+
+
+def receive(stream):
+    length = struct.unpack("<I", exact(stream, 4))[0]
+    if not 1 <= length <= 65536:
+        raise ValueError("IPC header exceeds budget")
+    header = json.loads(exact(stream, length))
+    size = header.pop("payload_bytes")
+    if type(size) is not int:
+        raise ValueError("invalid IPC byte count")
+    return header, exact(stream, size)
+
+
+def send(stream, header, payload=b""):
+    if len(payload) > MAX_PAYLOAD:
+        raise ValueError("IPC payload exceeds budget")
+    raw = json.dumps({**header, "payload_bytes": len(payload)}, allow_nan=False).encode()
+    if len(raw) > 65536:
+        raise ValueError("IPC header exceeds budget")
+    stream.write(struct.pack("<I", len(raw)) + raw + payload)
+    stream.flush()

experiments/perception/worker/streaming_profile_stage1/pilot_model.py

@@ -0,0 +1,110 @@
+"""Persistent decode OR DDRNet child; never loads an alternative model."""
+
+from __future__ import annotations
+
+import argparse
+import importlib.util
+import os
+import sys
+import time
+from pathlib import Path
+
+from baseline_ddrnet import digest
+from pilot_ddrnet_runtime import DdrnetRuntime
+from pilot_ipc import receive, send
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("mode", choices=("camera", "ddrnet"))
+    parser.add_argument("--video", default="/source.mp4")
+    parser.add_argument("--runner", default="/probe/run_goose_vegetation_benchmark.py")
+    parser.add_argument("--checkpoint", default="/checkpoint.pth")
+    parser.add_argument(
+        "--ddrnet-layout", choices=("reference", "channels-last"), default="reference"
+    )
+    parser.add_argument("--ddrnet-execution", choices=("eager", "cuda-graph"), default="eager")
+    args = parser.parse_args()
+    output = os.fdopen(os.dup(sys.stdout.fileno()), "wb", buffering=0)
+    os.dup2(sys.stderr.fileno(), sys.stdout.fileno())
+    import cv2
+    import numpy as np
+
+    started = time.monotonic_ns()
+    capture = None
+    if args.mode == "camera":
+        capture = cv2.VideoCapture(args.video)
+        if not capture.isOpened():
+            raise RuntimeError("camera decoder unavailable")
+    else:
+        if digest(Path(args.checkpoint)) != (
+            "b99c2838051bcd7b092fd3970aa62a77d5c0bbb809c9b9afb2ff4b0ebdaa4ee6"
+        ):
+            raise RuntimeError("DDRNet checkpoint changed")
+        if digest(Path(args.runner)) != (
+            "b18ad60f277eea69a240a28f290611b94627fb9707faf1bb3e6e22102dad67c1"
+        ):
+            raise RuntimeError("DDRNet preprocessing runner changed")
+        spec = importlib.util.spec_from_file_location("pinned_goose_runner", args.runner)
+        module = importlib.util.module_from_spec(spec)
+        spec.loader.exec_module(module)
+        import torch
+        from PIL import Image
+
+        model, _, _ = module.load_model("ddrnet", Path(args.checkpoint))
+        runtime = DdrnetRuntime(
+            model, module.logits_from_output, args.ddrnet_layout, args.ddrnet_execution
+        )
+        runtime.validate_ties()
+        warm, _ = module.preprocess(Image.fromarray(np.zeros((600, 800, 3), np.uint8)))
+        for _ in range(8):
+            runtime.infer(warm)
+        torch.cuda.reset_peak_memory_stats()
+    send(
+        output,
+        {
+            "ready": args.mode,
+            "warmup_ms": (time.monotonic_ns() - started) / 1e6,
+            "postprocess_layout": args.ddrnet_layout if args.mode == "ddrnet" else None,
+            "postprocess_tie_check": args.mode == "ddrnet",
+            "execution_mode": args.ddrnet_execution if args.mode == "ddrnet" else None,
+        },
+    )
+    try:
+        while True:
+            header, payload = receive(sys.stdin.buffer)
+            if header.get("op") == "stop":
+                return
+            begin = time.monotonic_ns()
+            if args.mode == "camera":
+                if header != {"op": "next"} or payload:
+                    raise ValueError("invalid decoder operation")
+                ok, bgr = capture.read()
+                if not ok or bgr is None or bgr.shape != (600, 800, 3):
+                    raise RuntimeError("camera ended or changed shape")
+                send(output, {"decode_ms": (time.monotonic_ns() - begin) / 1e6}, bgr.tobytes())
+            else:
+                if header != {"op": "infer"} or len(payload) != 600 * 800 * 3:
+                    raise ValueError("invalid DDRNet observation")
+                bgr = np.frombuffer(payload, np.uint8).reshape(600, 800, 3)
+                tensor, _ = module.preprocess(Image.fromarray(cv2.cvtColor(bgr, cv2.COLOR_BGR2RGB)))
+                preprocess_ms = (time.monotonic_ns() - begin) / 1e6
+                mask, forward_ms, stages = runtime.infer(tensor)
+                stages["preprocess_wall_ms"] = preprocess_ms
+                send(
+                    output,
+                    {
+                        "component_ms": (time.monotonic_ns() - begin) / 1e6,
+                        "forward_ms": forward_ms,
+                        "stages_ms": stages,
+                        "postprocess_layout": args.ddrnet_layout,
+                    },
+                    mask.astype(np.uint8).tobytes(),
+                )
+    finally:
+        if capture is not None:
+            capture.release()
+
+
+if __name__ == "__main__":
+    main()

experiments/perception/worker/streaming_profile_stage1/pilot_preflight.py

@@ -0,0 +1,69 @@
+"""CPU-only preflight: metadata, one increment, TGS identity and graph wiring."""
+
+import json
+import struct
+import subprocess
+from pathlib import Path
+
+import numpy as np
+from pilot_graph import JointGraph
+from pilot_ipc import exact
+from pilot_source import SensorArchive, camera_events
+
+archive = SensorArchive(Path("/sensor-source.npz"))
+try:
+    point = next(archive.points())
+    pose = next(archive.poses())
+    camera = next(camera_events("/camera-index.jsonl", 1))
+    print(
+        json.dumps(
+            {
+                "first_point": point.time_ns,
+                "first_pose": pose.time_ns,
+                "first_camera": camera.time_ns,
+                "point_shape": point.value[0].shape,
+                "reads": archive.counters(),
+            }
+        ),
+        flush=True,
+    )
+finally:
+    archive.close()
+process = subprocess.Popen(
+    ["/usr/local/bin/pilot-tgs"], stdin=subprocess.PIPE, stdout=subprocess.PIPE
+)
+try:
+    grid = np.array(
+        [(x, y, -1.25, 0) for x in np.linspace(-8, 8, 24) for y in np.linspace(-8, 8, 24)],
+        dtype="<f4",
+    )
+    process.stdin.write(struct.pack("<I", len(grid)) + grid.tobytes())
+    process.stdin.flush()
+    n, ms = struct.unpack("<Id", exact(process.stdout, 12))
+    states = np.frombuffer(exact(process.stdout, n), np.uint8)
+    assert n == len(grid) and np.all((states >= 1) & (states <= 3))
+    graph = JointGraph(
+        "/code",
+        process,
+        mask_path="/valid-fov.png",
+        mapping_path="/goose.csv",
+        calibration_pack="/calibration.npz",
+    )
+    assert np.count_nonzero(graph.action_lut == 0) == 1
+    assert graph.action_lut[graph.material_names.index("hard_surface"), 1] == 0
+    graph.backend.close()
+    print(
+        json.dumps(
+            {
+                "graph_initialized": True,
+                "tgs_identity_accounted": n,
+                "tgs_ms": ms,
+                "cells": len(graph.grid),
+            }
+        ),
+        flush=True,
+    )
+finally:
+    process.stdin.write(struct.pack("<I", 0))
+    process.stdin.flush()
+    process.wait(timeout=5)

experiments/perception/worker/streaming_profile_stage1/pilot_queue.py

@@ -0,0 +1,74 @@
+"""One producer, one graph, bounded pending queue and active payload accounting."""
+
+from collections import deque
+from threading import Condition
+
+
+class Mailbox:
+    def __init__(self, capacity=2, byte_limit=16 * 1024 * 1024):
+        self.capacity = capacity
+        self.byte_limit = byte_limit
+        self.condition = Condition()
+        self.pending = deque()
+        self.bytes = 0
+        self.peak_bytes = 0
+        self.peak_pending = 0
+        self.external_pending = 0
+        self.dropped = []
+        self.done = False
+        self.error = None
+
+    def put(self, bundle):
+        size = bundle["payload_bytes"]
+        with self.condition:
+            while self.pending and (
+                len(self.pending) + self.external_pending >= self.capacity
+                or self.bytes + size > self.byte_limit
+            ):
+                old = self.pending.popleft()
+                self.bytes -= old["payload_bytes"]
+                self.dropped.append({"sequence": old["sequence"], "reason": "pending-overflow"})
+            if len(self.pending) + self.external_pending >= self.capacity:
+                self.dropped.append({"sequence": bundle["sequence"], "reason": "pending-overflow"})
+                return
+            if self.bytes + size > self.byte_limit:
+                self.dropped.append({"sequence": bundle["sequence"], "reason": "byte-budget"})
+                return
+            self.pending.append(bundle)
+            self.bytes += size
+            self.peak_bytes = max(self.peak_bytes, self.bytes)
+            self.peak_pending = max(self.peak_pending, len(self.pending) + self.external_pending)
+            self.condition.notify_all()
+
+    def reserve_completed(self):
+        """A finished GPU result shares the SAME pending budget as ingress."""
+        with self.condition:
+            if self.external_pending:
+                raise ValueError("only one completed GPU slot is permitted")
+            while len(self.pending) + 1 > self.capacity:
+                old = self.pending.popleft()
+                self.bytes -= old["payload_bytes"]
+                self.dropped.append({"sequence": old["sequence"], "reason": "handoff-overflow"})
+            self.external_pending = 1
+            self.peak_pending = max(self.peak_pending, len(self.pending) + 1)
+
+    def take_completed(self):
+        with self.condition:
+            if self.external_pending != 1:
+                raise ValueError("completed GPU slot accounting mismatch")
+            self.external_pending = 0
+
+    def take(self):
+        with self.condition:
+            self.condition.wait_for(lambda: self.pending or self.done)
+            return self.pending.popleft() if self.pending else None
+
+    def release(self, bundle):
+        with self.condition:
+            self.bytes -= bundle["payload_bytes"]
+
+    def finish(self, error=None):
+        with self.condition:
+            self.done = True
+            self.error = error
+            self.condition.notify_all()

experiments/perception/worker/streaming_profile_stage1/pilot_scheduler.py

@@ -0,0 +1,61 @@
+"""One serial GPU stage can overlap one chronological CPU fusion stage.
+
+No GPU model concurrency. Ingress and completed-GPU results share two pending
+slots. Active payloads remain in the shared mailbox byte budget.
+"""
+
+import queue
+import threading
+import traceback
+
+
+class GpuStage:
+    def __init__(self, mailbox, compute, stop):
+        self.mailbox = mailbox
+        self.compute = compute
+        self.stop = stop
+        self.output = queue.Queue(maxsize=1)
+        # Reserve the output slot BEFORE starting another GPU call. Otherwise a
+        # blocked put would hide a third pending frame outside the two queues.
+        self.output_slot = threading.Semaphore(1)
+        self.finished = threading.Event()
+        self.error = None
+        self.peak_pending = 0
+        self.thread = threading.Thread(target=self._run, daemon=True)
+        self.thread.start()
+
+    def _run(self):
+        try:
+            while not self.stop.is_set():
+                if not self.output_slot.acquire(timeout=0.05):
+                    continue
+                bundle = self.mailbox.take()
+                if bundle is None:
+                    break
+                result = self.compute(bundle)
+                self.mailbox.reserve_completed()
+                self.output.put_nowait((bundle, result))
+                self.peak_pending = max(self.peak_pending, self.output.qsize())
+        except Exception:
+            self.error = traceback.format_exc()
+        finally:
+            self.finished.set()
+
+    def take(self):
+        while True:
+            try:
+                result = self.output.get(timeout=0.05)
+                self.mailbox.take_completed()
+                self.output_slot.release()
+                return result
+            except queue.Empty:
+                if self.finished.is_set():
+                    if self.error:
+                        raise RuntimeError(self.error) from None
+                    return None
+
+    def close(self):
+        self.stop.set()
+        self.mailbox.finish(self.mailbox.error)
+        self.thread.join(timeout=2)
+        return not self.thread.is_alive()

experiments/perception/worker/streaming_profile_stage1/pilot_source.py

@@ -0,0 +1,154 @@
+"""Incremental existing normalized sensor archive; no NPZ materialization.
+
+Only original capture clocks/points/pose are read. No camera-indexed E10
+nearest-neighbor association, surface NPZ, masks, TGS outputs or threat ledger.
+One-row lookahead belongs to the source reader, never to perception providers.
+"""
+
+from __future__ import annotations
+
+import heapq
+import json
+import math
+import zipfile
+from dataclasses import dataclass
+from pathlib import Path
+
+import numpy as np
+
+
+class NpyRows:
+    def __init__(self, archive, name, *, columns=(), dtype=None):
+        self.stream = archive.open(name + ".npy")
+        version = np.lib.format.read_magic(self.stream)
+        if version == (1, 0):
+            shape, fortran, actual = np.lib.format.read_array_header_1_0(self.stream)
+        elif version == (2, 0):
+            shape, fortran, actual = np.lib.format.read_array_header_2_0(self.stream)
+        else:
+            raise ValueError("unsupported bounded NPY header")
+        if fortran or actual.hasobject or tuple(shape[1:]) != columns:
+            raise ValueError("sensor archive layout changed")
+        if dtype is not None and actual != np.dtype(dtype):
+            raise ValueError("sensor archive dtype changed")
+        self.count = shape[0]
+        self.columns = columns
+        self.dtype = actual
+        self.row_bytes = actual.itemsize * math.prod(columns)
+        self.position = 0
+        self.bytes_read = 0
+
+    def take(self, count=1):
+        size = count * self.row_bytes
+        if count < 0 or self.position + count > self.count or size > 8 * 1024 * 1024:
+            raise ValueError("sensor row exceeds source budget")
+        raw = self.stream.read(size)
+        if len(raw) != size:
+            raise EOFError("truncated sensor array")
+        self.position += count
+        self.bytes_read += size
+        return np.frombuffer(raw, self.dtype).reshape((count, *self.columns))
+
+
+@dataclass(frozen=True)
+class SensorEvent:
+    time_ns: int
+    channel: str
+    sequence: int
+    value: object
+
+
+class SensorArchive:
+    def __init__(self, path: Path):
+        self.archive = zipfile.ZipFile(path)
+        self.readers = {}
+
+    def rows(self, name, **kwargs):
+        reader = NpyRows(self.archive, name, **kwargs)
+        self.readers[name] = reader
+        return reader
+
+    def points(self):
+        times = self.rows("point_received_monotonic_ns", dtype="<i8")
+        seqs = self.rows("point_capture_sequence", dtype="<i8")
+        offsets = self.rows("point_offsets", dtype="<i8")
+        xyz = self.rows("point_xyz_map", columns=(3,), dtype="<f8")
+        intensity = self.rows("point_intensity", dtype="u1")
+        begin = int(offsets.take()[0])
+        if begin != 0:
+            raise ValueError("point offsets do not start at zero")
+        previous = -1
+        for _ in range(times.count):
+            stamp = int(times.take()[0])
+            sequence = int(seqs.take()[0])
+            end = int(offsets.take()[0])
+            if stamp < previous or not 0 <= end - begin <= 50000:
+                raise ValueError("point clock/size outside source bounds")
+            points = xyz.take(end - begin)
+            intensities = intensity.take(end - begin)
+            if not np.isfinite(points).all():
+                raise ValueError("nonfinite source cloud")
+            yield SensorEvent(stamp, "points", sequence, (points, intensities))
+            begin, previous = end, stamp
+
+    def poses(self):
+        times = self.rows("pose_received_monotonic_ns", dtype="<i8")
+        seqs = self.rows("pose_capture_sequence", dtype="<i8")
+        positions = self.rows("pose_positions_map", columns=(3,), dtype="<f8")
+        quats = self.rows("pose_quaternions_map_from_lidar", columns=(4,), dtype="<f8")
+        previous = -1
+        for _ in range(times.count):
+            stamp = int(times.take()[0])
+            if stamp < previous:
+                raise ValueError("pose clock moved backwards")
+            position, quaternion = positions.take()[0], quats.take()[0]
+            if (
+                not np.isfinite(position).all()
+                or not np.isfinite(quaternion).all()
+                or abs(float(np.linalg.norm(quaternion)) - 1.0) > 0.01
+            ):
+                raise ValueError("invalid source pose")
+            yield SensorEvent(stamp, "pose", int(seqs.take()[0]), (position, quaternion))
+            previous = stamp
+
+    def counters(self):
+        return {
+            name: {"rows_read": r.position, "bytes_read": r.bytes_read, "source_rows": r.count}
+            for name, r in self.readers.items()
+        }
+
+    def close(self):
+        for reader in self.readers.values():
+            reader.stream.close()
+        self.archive.close()
+
+
+def camera_events(path, limit):
+    previous = -1
+    with Path(path).open("rb") as stream:
+        for index in range(limit):
+            raw = stream.readline(65537)
+            if not raw or len(raw) > 65536:
+                raise ValueError("camera index is truncated or unbounded")
+            row = json.loads(raw)
+            if (
+                row.get("schema_version") != "missioncore.camera-recording-index/v1"
+                or row.get("kind") != "media"
+                or row.get("sequence") != index + 1
+            ):
+                raise ValueError("camera source identity/order changed")
+            stamp = row["host_monotonic_ns"]
+            if type(stamp) is not int or stamp <= previous:
+                raise ValueError("camera source clock is not increasing")
+            yield SensorEvent(stamp, "camera", index, row)
+            previous = stamp
+
+
+def merged_events(archive, camera_index, limit):
+    # Stable per-channel order; a tie admits sensor data before the camera.
+    return heapq.merge(
+        archive.poses(),
+        archive.points(),
+        camera_events(camera_index, limit),
+        key=lambda event: event.time_ns,
+    )

experiments/perception/worker/streaming_profile_stage1/pilot_tgs.cpp

@@ -0,0 +1,62 @@
+// Persistent bounded TGS adapter. stdin: uint32 count + XYZ-intensity float32.
+// Original point identity survives segmentation in PointXYZILID::id (<=64000).
+// stdout: uint32 count + double algorithm_ms + uint8 state per input point.
+#include <cmath>
+#include <cstdint>
+#include <cstdio>
+#include <stdexcept>
+#include <unistd.h>
+#include <vector>
+#include "travel/point_types.hpp"
+#include "travel/tgs.hpp"
+
+int main() {
+    FILE* protocol = fdopen(dup(STDOUT_FILENO), "wb");
+    dup2(STDERR_FILENO, STDOUT_FILENO);  // TGS diagnostics cannot corrupt RPC.
+    try {
+        std::uint32_t count;
+        while (std::fread(&count, sizeof(count), 1, stdin) == 1) {
+            if (count == 0) break;
+            if (count > 64000) throw std::runtime_error("TGS point budget exceeded");
+            std::vector<float> data(static_cast<std::size_t>(count) * 4);
+            if (std::fread(data.data(), sizeof(float), data.size(), stdin) != data.size())
+                throw std::runtime_error("truncated TGS observation");
+            travel::PointCloud<PointXYZILID> input, ground, nonground;
+            input.reserve(count);
+            for (std::uint32_t index = 0; index < count; ++index) {
+                PointXYZILID p{};
+                p.x = data[index*4]; p.y = data[index*4+1]; p.z = data[index*4+2];
+                p.intensity = data[index*4+3]; p.id = static_cast<std::uint16_t>(index);
+                if (!std::isfinite(p.x) || !std::isfinite(p.y) || !std::isfinite(p.z))
+                    throw std::runtime_error("nonfinite TGS input");
+                input.push_back(p);
+            }
+            double seconds = 0;
+            travel::TravelGroundSeg<PointXYZILID> tgs;
+            tgs.setParams(80.0, 1.0, 8.0, 3, 5, 10, 0.5, 0.125, 0.3, 0.940,
+                          200.0, 0.03, 0.1, 1.0, true, false);
+            tgs.estimateGround(input, ground, nonground, seconds);
+            std::vector<std::uint8_t> states(count, 3);  // Rejected != unobserved/free.
+            for (const auto& p : ground.points) {
+                if (p.id >= count || states[p.id] != 3)
+                    throw std::runtime_error("TGS ground identity collision");
+                states[p.id] = 1;
+            }
+            for (const auto& p : nonground.points) {
+                if (p.id >= count || states[p.id] != 3)
+                    throw std::runtime_error("TGS nonground identity collision");
+                states[p.id] = 2;
+            }
+            const double ms = seconds * 1000.0;
+            if (std::fwrite(&count, sizeof(count), 1, protocol) != 1 ||
+                std::fwrite(&ms, sizeof(ms), 1, protocol) != 1 ||
+                std::fwrite(states.data(), 1, count, protocol) != count)
+                throw std::runtime_error("TGS result transport failed");
+            std::fflush(protocol);
+        }
+        std::fclose(protocol);
+    } catch (const std::exception& e) {
+        std::fprintf(stderr, "TGS pilot: %s\n", e.what());
+        return 1;
+    }
+}

experiments/perception/worker/streaming_profile_stage1/rfdetr-config.pbtxt

@@ -0,0 +1,9 @@
+name: "rf_detr_large_native_kb4"
+platform: "tensorrt_plan"
+max_batch_size: 0
+input [{ name: "raw_kb4_bgr" data_type: TYPE_UINT8 dims: [1,600,800,3] }]
+output [
+  { name: "dets" data_type: TYPE_FP16 dims: [1,300,4] },
+  { name: "labels" data_type: TYPE_FP16 dims: [1,300,91] }
+]
+instance_group [{ count: 1 kind: KIND_GPU gpus: [0] }]

experiments/perception/worker/streaming_profile_stage1/run_joint_pilot.py

@@ -0,0 +1,605 @@
+"""Bounded source-paced Worker pilot, not an installed or qualified profile.
+
+Only persistent DDRNet and RF-DETR share the GPU; inference is serialized.
+Source is paced independently of the graph, using original host arrival times.
+The in-container collector is measured; network to Mission Core is NOT measured.
+"""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import http.client
+import json
+import math
+import os
+import signal
+import subprocess
+import threading
+import time
+import traceback
+from collections import Counter, deque
+from dataclasses import asdict
+from datetime import UTC, datetime
+from pathlib import Path
+
+import numpy as np
+from pilot_ipc import receive, send
+from pilot_queue import Mailbox
+from pilot_scheduler import GpuStage
+from pilot_source import SensorArchive, camera_events, merged_events
+
+
+def distribution(values):
+    if not values:
+        return None
+    values = sorted(values)
+    return {
+        "count": len(values),
+        "mean": sum(values) / len(values),
+        "min": values[0],
+        "p50": values[math.ceil(len(values) * 0.5) - 1],
+        "p95": values[math.ceil(len(values) * 0.95) - 1],
+        "p99": values[math.ceil(len(values) * 0.99) - 1],
+        "max": values[-1],
+    }
+
+
+def payload_size(bundle):
+    return (
+        sum(bundle[key].nbytes for key in ("image", "points", "rolling_points", "rolling_times"))
+        + 4096
+    )
+
+
+def produce(args, decoder, mailbox, stop, report):
+    archive = SensorArchive(Path(args.sensor_archive))
+    first_camera = next(camera_events(args.camera_index, args.frames)).time_ns
+    source_zero = first_camera - 500_000_000
+    wall_zero = time.monotonic_ns() + 50_000_000
+    report.update(
+        source_zero_ns=source_zero,
+        wall_zero_ns=wall_zero,
+        first_camera_source_ns=first_camera,
+        source_clock_speed=1.0,
+        source_eof_required=False,
+        full_source_prepass=False,
+    )
+    pose = None
+    rolling = deque()
+    fresh = []
+    arrivals = Counter()
+    release_lags = []
+    skipped_prefix = Counter()
+    try:
+        for event in merged_events(archive, args.camera_index, args.frames):
+            if stop.is_set():
+                break
+            if event.time_ns < source_zero:
+                skipped_prefix[event.channel] += 1
+                continue
+            due = wall_zero + event.time_ns - source_zero
+            if stop.wait(max(0.0, (due - time.monotonic_ns()) / 1e9)):
+                break
+            arrived = time.monotonic_ns()
+            release_lags.append(max(0, arrived - due) / 1e6)
+            arrivals[event.channel] += 1
+            while rolling and event.time_ns - rolling[0].time_ns > 1_000_000_000:
+                rolling.popleft()
+            if event.channel == "pose":
+                pose = event
+                continue
+            if event.channel == "points":
+                rolling.append(event)
+                fresh.append(event)
+                if sum(len(e.value[0]) for e in rolling) > 64000:
+                    raise ValueError("raw rolling cloud exceeds bounded 64000-point window")
+                continue
+            # Decoder reads ONE frame after its original due time, never full-decodes.
+            decode_start = time.monotonic_ns()
+            send(decoder.stdin, {"op": "next"})
+            decoded, raw = receive(decoder.stdout)
+            image = np.frombuffer(raw, np.uint8).reshape(600, 800, 3)
+            points = np.concatenate([e.value[0] for e in fresh]) if fresh else np.empty((0, 3))
+            rolling_points = (
+                np.concatenate([e.value[0] for e in rolling]) if rolling else np.empty((0, 3))
+            )
+            rolling_times = (
+                np.concatenate([np.full(len(e.value[0]), e.time_ns, np.int64) for e in rolling])
+                if rolling
+                else np.empty(0, np.int64)
+            )
+            pose_age = event.time_ns - pose.time_ns if pose else None
+            point_age = event.time_ns - fresh[-1].time_ns if fresh else None
+            oldest_age = event.time_ns - fresh[0].time_ns if fresh else None
+            binding_age = (
+                max(abs(e.time_ns - pose.time_ns) for e in fresh) if fresh and pose else None
+            )
+            available = bool(
+                len(points)
+                and pose
+                and 0 <= pose_age <= 100_000_000
+                and 0 <= point_age <= 100_000_000
+                and oldest_age <= 250_000_000
+                and binding_age <= 100_000_000
+            )
+            bundle = {
+                "sequence": event.sequence,
+                "time_ns": event.time_ns,
+                "source_ns": event.time_ns - source_zero,
+                "due_ns": due,
+                "utc_ns": event.value["host_epoch_ns"],
+                "image": image,
+                "points": points,
+                "rolling_points": rolling_points,
+                "rolling_times": rolling_times,
+                "pose": pose.value if pose else None,
+                "available": available,
+                "binding_age_ms": binding_age / 1e6 if binding_age is not None else None,
+                "lineage": {
+                    "camera_index_sequence": event.value["sequence"],
+                    "camera_host_monotonic_ns": event.time_ns,
+                    "pose_sequence": pose.sequence if pose else None,
+                    "pose_host_monotonic_ns": pose.time_ns if pose else None,
+                    "point_increments": [
+                        {
+                            "sequence": e.sequence,
+                            "host_monotonic_ns": e.time_ns,
+                            "points": len(e.value[0]),
+                        }
+                        for e in fresh
+                    ],
+                    "pose_age_ms": pose_age / 1e6 if pose_age is not None else None,
+                    "newest_point_age_ms": point_age / 1e6 if point_age is not None else None,
+                    "oldest_point_age_ms": oldest_age / 1e6 if oldest_age is not None else None,
+                },
+                "source_release_lag_ms": max(0, arrived - due) / 1e6,
+                "decode_ms": decoded["decode_ms"],
+                "decode_rpc_ms": (time.monotonic_ns() - decode_start) / 1e6,
+                "enqueued_ns": time.monotonic_ns(),
+            }
+            bundle["payload_bytes"] = payload_size(bundle)
+            mailbox.put(bundle)
+            fresh = []
+            report["last_camera_due_ns"] = due
+            if arrivals["camera"] >= args.frames:
+                break
+    except Exception:
+        mailbox.finish(traceback.format_exc())
+    finally:
+        report.update(
+            arrivals=dict(arrivals),
+            skipped_prefix=dict(skipped_prefix),
+            release_lag_ms=distribution(release_lags),
+            incremental_reads=archive.counters(),
+            window_end_monotonic_ns=time.monotonic_ns(),
+        )
+        archive.close()
+        mailbox.finish(mailbox.error)
+
+
+def telemetry(stop, samples):
+    while not stop.is_set():
+        sample = {"monotonic_ns": time.monotonic_ns()}
+        try:
+            result = subprocess.run(
+                [
+                    "nvidia-smi",
+                    "--query-gpu=memory.used,utilization.gpu",
+                    "--format=csv,noheader,nounits",
+                ],
+                capture_output=True,
+                text=True,
+                timeout=2,
+                check=True,
+            )
+            mem, util = result.stdout.strip().split(",")
+            sample.update(gpu_used_mib=int(mem), gpu_utilization=int(util))
+            sample["cgroup_memory_mib"] = (
+                int(Path("/sys/fs/cgroup/memory.current").read_text()) / 1048576
+            )
+            sample["cpu_stat"] = {
+                key: int(value)
+                for key, value in (
+                    line.split()
+                    for line in Path("/sys/fs/cgroup/cpu.stat").read_text().splitlines()
+                )
+            }
+        except Exception as error:
+            sample["error"] = str(error)
+        samples.append(sample)
+        stop.wait(0.5)
+
+
+def wait_triton(process, seconds=90):
+    deadline = time.monotonic() + seconds
+    while time.monotonic() < deadline:
+        if process.poll() is not None:
+            raise RuntimeError("Triton exited during initialization")
+        connection = http.client.HTTPConnection("127.0.0.1", 8000, timeout=1)
+        try:
+            connection.request("GET", "/v2/models/rf_detr_large_native_kb4/ready")
+            if connection.getresponse().status == 200:
+                return
+        except OSError:
+            pass
+        finally:
+            connection.close()
+        time.sleep(0.1)
+    raise TimeoutError("Triton initialization exceeded budget")
+
+
+def run(args):
+    output = Path(args.output)
+    output.mkdir(parents=True, exist_ok=False)
+    started = time.monotonic_ns()
+    report = {
+        "schema_version": "missioncore.stage1-joint-pilot/v1",
+        "run_id": args.run_id,
+        "started_utc": datetime.now(UTC).isoformat(),
+        "started_monotonic_ns": started,
+        "requested_frames": args.frames,
+        "profile": "K1 Perception — DDRNet-39 + RF-DETR + TGS",
+        "scope": "one-container-worker-graph-to-local-collector",
+        "external_transport_measured": False,
+        "standalone_release": False,
+        "authority": {"commands_enabled": False, "actuation_allowed": False},
+        "effective_policy": "hard-surface-only-plus-geometry-TGS-stale-no-go",
+        "body_frame": "causal-camera-forward-simulation; no future trajectory",
+        "clock_quality": "original host arrival best effort; not hardware synchronization",
+        "ddrnet_postprocess_layout": args.ddrnet_layout,
+        "ddrnet_execution": args.ddrnet_execution,
+        "schedule": args.schedule,
+    }
+    children, logs, results, samples = [], [], [], []
+    stop = threading.Event()
+    mailbox = Mailbox(capacity=2)
+    source_report = {}
+    producer = monitor = graph = gpu_stage = None
+
+    def child(name, command, env=None):
+        log = (output / (name + ".log")).open("wb")
+        logs.append(log)
+        process = subprocess.Popen(
+            command,
+            stdin=subprocess.PIPE,
+            stdout=subprocess.PIPE,
+            stderr=log,
+            env=env,
+            start_new_session=True,
+        )
+        children.append(process)
+        return process
+
+    def timeout_handler(signum, frame):
+        raise TimeoutError("joint pilot wall-clock watchdog fired")
+
+    previous = signal.signal(signal.SIGALRM, timeout_handler)
+    signal.alarm(240)
+    try:
+        triton = child(
+            "triton",
+            [
+                "tritonserver",
+                "--model-repository=/models",
+                "--model-control-mode=explicit",
+                "--load-model=rf_detr_large_native_kb4",
+                "--allow-grpc=false",
+                "--allow-metrics=false",
+                "--http-address=127.0.0.1",
+                "--pinned-memory-pool-byte-size=16777216",
+                "--cuda-memory-pool-byte-size=0:16777216",
+            ],
+        )
+        # Triton logs stdout too: drain directly to the same bounded-run log.
+        threading.Thread(target=lambda: drain(triton.stdout, logs[0]), daemon=True).start()
+        wait_triton(triton)
+        python = "/opt/conda/envs/goose/bin/python"
+        ddr = child(
+            "ddrnet",
+            [
+                python,
+                "-B",
+                "/probe/pilot_model.py",
+                "ddrnet",
+                "--ddrnet-layout",
+                args.ddrnet_layout,
+                "--ddrnet-execution",
+                args.ddrnet_execution,
+            ],
+            {**os.environ, "PYTHONPATH": "/probe"},
+        )
+        report["ddrnet_ready"], _ = receive(ddr.stdout)
+        decoder = child(
+            "decoder",
+            [python, "-B", "/probe/pilot_model.py", "camera"],
+            {**os.environ, "PYTHONPATH": "/probe", "CUDA_VISIBLE_DEVICES": ""},
+        )
+        report["decoder_ready"], _ = receive(decoder.stdout)
+        tgs = child("tgs", ["/usr/local/bin/pilot-tgs"])
+        from pilot_graph import JointGraph
+
+        graph = JointGraph(
+            "/code",
+            tgs,
+            mask_path="/valid-fov.png",
+            mapping_path="/goose.csv",
+            calibration_pack="/calibration.npz",
+        )
+        report["rfdetr_warmup"] = asdict(graph.detector.warm_up())
+        report["warmup_ms"] = (time.monotonic_ns() - started) / 1e6
+        report["costmap_grid"] = graph.grid.tolist()
+        report["material_names"] = graph.material_names
+
+        def compute_gpu(bundle):
+            begin = time.monotonic_ns()
+            send(ddr.stdin, {"op": "infer"}, bundle["image"].tobytes())
+            ddr_result, raw = receive(ddr.stdout)
+            mask = np.frombuffer(raw, np.uint8).reshape(512, 512)
+            ddr_done = time.monotonic_ns()
+            proposals = graph.detector.detect(graph.packet(bundle))
+            gpu_done = time.monotonic_ns()
+            return begin, ddr_done, gpu_done, ddr_result, mask, proposals
+
+        monitor = threading.Thread(target=telemetry, args=(stop, samples), daemon=True)
+        monitor.start()
+        producer = threading.Thread(
+            target=produce, args=(args, decoder, mailbox, stop, source_report), daemon=True
+        )
+        producer.start()
+        if args.schedule == "overlap-cpu":
+            gpu_stage = GpuStage(mailbox, compute_gpu, stop)
+        with (output / "scenes.jsonl").open("wb") as sink:
+            while True:
+                if gpu_stage:
+                    item = gpu_stage.take()
+                    if item is None:
+                        break
+                    bundle, computed = item
+                else:
+                    bundle = mailbox.take()
+                    if bundle is None:
+                        break
+                    computed = compute_gpu(bundle)
+                begin, ddr_done, gpu_done, ddr_result, mask, proposals = computed
+                cpu_started = time.monotonic_ns()
+                scene = graph.process(
+                    bundle,
+                    mask,
+                    proposals=proposals,
+                    detector_ms=(gpu_done - ddr_done) / 1e6,
+                )
+                # Late evidence remains inspectable but cannot authorize terrain.
+                scene["oldest_required_input_age_ms"] = (
+                    time.monotonic_ns() - bundle["due_ns"]
+                ) / 1e6 + max(
+                    bundle["lineage"]["pose_age_ms"] or 0,
+                    bundle["lineage"]["oldest_point_age_ms"] or 0,
+                )
+                scene["stale_at_publication"] = scene["oldest_required_input_age_ms"] > 250
+                if scene["stale_at_publication"]:
+                    scene["policy_actions"] = [2] * len(scene["policy_actions"])
+                    scene["policy_counts"] = {
+                        "ALLOW_candidate": 0,
+                        "HIGH_COST": 0,
+                        "NO_GO": len(scene["policy_actions"]),
+                    }
+                scene.update(
+                    sequence=bundle["sequence"],
+                    lineage=bundle["lineage"],
+                    available=bundle["available"],
+                    original_source_ns=bundle["time_ns"],
+                )
+                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")
+                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,
+                }
+                result = {
+                    "sequence": bundle["sequence"],
+                    "finished_monotonic_ns": finished,
+                    "timing_ms": timing,
+                    "available": bundle["available"],
+                    "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": scene["policy_counts"],
+                    "scene_bytes": len(encoded),
+                    "scene_sha256": hashlib.sha256(encoded).hexdigest(),
+                }
+                results.append(result)
+                mailbox.release(bundle)
+                sink.write(encoded + b"\n")
+                if len(results) == 1:
+                    report["reads_at_first_result"] = {
+                        "results": 1,
+                        "source_eof_reached": False,
+                        "source_end_window_reached": mailbox.done,
+                    }
+                if len(results) % 16 == 0:
+                    print(
+                        json.dumps(
+                            {
+                                "completed": len(results),
+                                "last_sequence": bundle["sequence"],
+                                "last_age_ms": timing["source_due_to_receiver_ms"],
+                                "drops": len(mailbox.dropped),
+                            }
+                        ),
+                        flush=True,
+                    )
+        if mailbox.error:
+            raise RuntimeError(mailbox.error)
+        report["execution_complete"] = True
+    except Exception:
+        report["execution_complete"] = False
+        report["error"] = traceback.format_exc()
+        print(report["error"], flush=True)
+    finally:
+        signal.alarm(0)
+        signal.signal(signal.SIGALRM, previous)
+        stop.set()
+        shutdown_start = time.monotonic_ns()
+        for process in reversed(children):
+            if process.poll() is None:
+                os.killpg(process.pid, signal.SIGTERM)
+        deadline = time.monotonic() + 4
+        for process in reversed(children):
+            try:
+                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:
+            monitor.join(timeout=2)
+        report["gpu_stage_stopped"] = gpu_stage.close() if gpu_stage else True
+        if graph:
+            graph.backend.close()
+        for log in logs:
+            log.close()
+        report["stop_ms"] = (time.monotonic_ns() - shutdown_start) / 1e6
+        report["children_stopped"] = all(p.poll() is not None for p in children)
+    report["source"] = source_report
+    report["frames"] = results
+    report["telemetry"] = samples
+    report["queue"] = {
+        "capacity": mailbox.capacity,
+        "gpu_completed_capacity": 1 if gpu_stage else 0,
+        "completed_shares_pending_limit": bool(gpu_stage),
+        "global_pending_limit": 2,
+        "gpu_peak_pending": gpu_stage.peak_pending if gpu_stage else 0,
+        "byte_limit": mailbox.byte_limit,
+        "peak_pending": mailbox.peak_pending,
+        "peak_bytes": mailbox.peak_bytes,
+        "drops": mailbox.dropped,
+        "residual_bytes": mailbox.bytes,
+        "residual_completed": mailbox.external_pending,
+    }
+    report["distributions_ms"] = (
+        {
+            key: distribution([r["timing_ms"][key] for r in results])
+            for key in results[0]["timing_ms"]
+        }
+        if results
+        else {}
+    )
+    camera_count = source_report.get("arrivals", {}).get("camera", 0)
+    report["accounting"] = {
+        "released": camera_count,
+        "completed": len(results),
+        "dropped": len(mailbox.dropped),
+        "unaccounted": camera_count - len(results) - len(mailbox.dropped),
+    }
+    latency = report["distributions_ms"].get("source_due_to_receiver_ms")
+    queue_lags = [
+        r["timing_ms"]["queue_wait_ms"] + r["timing_ms"]["gpu_to_cpu_queue_wait_ms"]
+        for r in results
+    ]
+    growth = (
+        max(
+            0.0,
+            sum(queue_lags[-8:]) / len(queue_lags[-8:]) - sum(queue_lags[:8]) / len(queue_lags[:8]),
+        )
+        if queue_lags
+        else None
+    )
+    report["backlog_growth_ms"] = growth
+    report["backlog_growth_basis"] = "ingress-plus-completed-GPU-wait-last8-minus-first8"
+    first_ms = (
+        ((results[0]["finished_monotonic_ns"] - source_report["wall_zero_ns"]) / 1e6)
+        if results
+        else None
+    )
+    report["first_result_from_stream_start_ms"] = first_ms
+    report["gates"] = {
+        "execution": report["execution_complete"],
+        "zero_drops_complete_accounting": camera_count == args.frames == len(results)
+        and not mailbox.dropped,
+        "worker_p95_p99_125ms": bool(latency and latency["p95"] <= 125 and latency["p99"] <= 125),
+        "source_release_lag_25ms": bool(
+            source_report.get("release_lag_ms") and source_report["release_lag_ms"]["max"] <= 25
+        ),
+        "backlog_growth_25ms": growth is not None and growth <= 25,
+        "first_result_1s": first_ms is not None and first_ms <= 1000,
+        "warmup_120s": report.get("warmup_ms", math.inf) <= 120000,
+        "stop_5s": report["stop_ms"] <= 5000
+        and report["children_stopped"]
+        and report["gpu_stage_stopped"],
+        "vram_22000MiB": bool(samples)
+        and max(s.get("gpu_used_mib", 99999) for s in samples) <= 22000,
+        "container_memory_8192MiB": bool(samples)
+        and max(s.get("cgroup_memory_mib", 99999) for s in samples) <= 8192,
+        "all_modalities_fresh": len(results) == args.frames
+        and all(r["available"] for r in results),
+        "network_end_to_end_qualified": False,
+    }
+    report["profile_realtime_qualified"] = False
+    (output / "report.json").write_text(json.dumps(report, indent=2, allow_nan=False) + "\n")
+    print(
+        json.dumps(
+            {
+                key: report[key]
+                for key in (
+                    "run_id",
+                    "execution_complete",
+                    "accounting",
+                    "gates",
+                    "distributions_ms",
+                )
+            }
+        ),
+        flush=True,
+    )
+    return 0 if report["execution_complete"] else 1
+
+
+def drain(stream, sink):
+    while True:
+        raw = stream.read(4096)
+        if not raw:
+            break
+        sink.write(raw)
+        sink.flush()
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--frames", type=int, default=128)
+    parser.add_argument("--run-id", required=True)
+    parser.add_argument("--output", required=True)
+    parser.add_argument("--sensor-archive", default="/sensor-source.npz")
+    parser.add_argument("--camera-index", default="/camera-index.jsonl")
+    parser.add_argument(
+        "--ddrnet-layout", choices=("reference", "channels-last"), default="reference"
+    )
+    parser.add_argument("--ddrnet-execution", choices=("eager", "cuda-graph"), default="eager")
+    parser.add_argument("--schedule", choices=("serial", "overlap-cpu"), default="serial")
+    args = parser.parse_args()
+    if not 1 <= args.frames <= 256:
+        parser.error("pilot window must be 1..256 frames")
+    raise SystemExit(run(args))

tests/test_perception_joint_pilot.py

@@ -0,0 +1,216 @@
+"""Small synthetic checks only; model/real-source runs belong on Worker 006."""
+
+import importlib
+import io
+import json
+import threading
+import zipfile
+from pathlib import Path
+
+import numpy as np
+import pytest
+
+
+@pytest.fixture
+def pilot(monkeypatch):
+    root = Path(__file__).resolve().parents[1]
+    monkeypatch.syspath_prepend(
+        str(root / "experiments/perception/worker/streaming_profile_stage1")
+    )
+    return lambda name: importlib.import_module(name)
+
+
+def test_pending_overflow_is_explicit_and_does_not_evict_active(pilot):
+    queue = pilot("pilot_queue").Mailbox(capacity=2, byte_limit=100)
+
+    def make(sequence):
+        return {"sequence": sequence, "payload_bytes": 10}
+
+    queue.put(make(0))
+    active = queue.take()
+    for sequence in (1, 2, 3):
+        queue.put(make(sequence))
+    assert queue.dropped == [{"sequence": 1, "reason": "pending-overflow"}]
+    assert queue.bytes == 30
+    assert queue.peak_pending == 2
+    queue.release(active)
+    assert queue.take()["sequence"] == 2
+    assert queue.take()["sequence"] == 3
+
+
+def test_active_bytes_count_towards_memory_limit(pilot):
+    queue = pilot("pilot_queue").Mailbox(byte_limit=20)
+    queue.put({"sequence": 0, "payload_bytes": 20})
+    active = queue.take()
+    queue.put({"sequence": 1, "payload_bytes": 1})
+    assert queue.dropped == [{"sequence": 1, "reason": "byte-budget"}]
+    assert queue.peak_bytes == 20
+    queue.release(active)
+    queue.finish()
+    assert queue.take() is None
+
+
+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))
+    data.seek(0)
+    with zipfile.ZipFile(data) as archive:
+        rows = pilot("pilot_source").NpyRows(archive, "points", columns=(3,), dtype="<f8")
+        np.testing.assert_array_equal(rows.take(2), np.arange(6).reshape(2, 3))
+        assert rows.count == 100 and rows.position == 2 and rows.bytes_read == 48
+        with pytest.raises(ValueError, match="budget"):
+            rows.take(99)
+
+
+def test_original_camera_clock_not_fixed_frame_rate(pilot, tmp_path):
+    path = tmp_path / "camera.jsonl"
+    stamps = [123000, 127777, 987654]
+    rows = [
+        {
+            "schema_version": "missioncore.camera-recording-index/v1",
+            "kind": "media",
+            "sequence": i + 1,
+            "host_monotonic_ns": stamp,
+        }
+        for i, stamp in enumerate(stamps)
+    ]
+    path.write_text("\n".join(json.dumps(row) for row in rows))
+    events = list(pilot("pilot_source").camera_events(path, 2))
+    assert [e.time_ns for e in events] == stamps[:2]
+    assert [e.sequence for e in events] == [0, 1]
+
+
+def test_ipc_preserves_binary_and_rejects_oversized_header(pilot):
+    ipc = pilot("pilot_ipc")
+    stream = io.BytesIO()
+    ipc.send(stream, {"op": "infer"}, b"\x00\xff")
+    stream.seek(0)
+    assert ipc.receive(stream) == ({"op": "infer"}, b"\x00\xff")
+    with pytest.raises(ValueError, match="header"):
+        ipc.receive(io.BytesIO((65537).to_bytes(4, "little")))
+
+
+def test_nearest_rank_tail_metrics_keep_outlier(pilot):
+    metrics = pilot("run_joint_pilot").distribution([1] * 99 + [200])
+    assert metrics["p99"] == 1
+    assert metrics["max"] == 200
+    assert pilot("run_joint_pilot").distribution([1, 200])["p99"] == 200
+
+
+def test_body_history_is_bounded_and_never_reads_future(pilot):
+    store = pilot("pilot_graph").CurrentStore(None)
+    for index in range(100):
+        store.frame_id = f"frame-{index:06d}"
+        store.body = index
+        store.remember_body()
+    assert len(store.body_history) == 64
+    assert store.body_frame_for_frame("frame-000098") == 98
+    assert store.body_frame_for_frame("frame-000000") is None
+    with pytest.raises(ValueError, match="future"):
+        store.body_frame_for_frame("frame-000100")
+
+
+def test_vectorized_grid_lookup_preserves_exact_cells_and_holes(pilot):
+    grid = np.asarray([[-2, -1], [0, 0], [2, 1]])
+    points = np.asarray([[x, y] for x in range(-5, 6) for y in range(-5, 6)])
+    lookup = {tuple(row): i for i, row in enumerate(grid)}
+    expected = np.asarray([lookup.get(tuple(row), -1) for row in points])
+    np.testing.assert_array_equal(pilot("pilot_graph").grid_indices(points, grid), expected)
+
+
+def test_ddrnet_layout_preserves_class_order_and_ties(pilot):
+    class Scores:
+        def __init__(self, array):
+            self.array = array
+
+        def permute(self, *axes):
+            return Scores(self.array.transpose(axes))
+
+        def contiguous(self):
+            return Scores(np.ascontiguousarray(self.array))
+
+    # Include saturated/equal probabilities: first class must win every tie.
+    scores = Scores(np.array([[[[1.0, 0.5]], [[1.0, 0.5]], [[0.0, 0.9]], [[1.0, 0.9]]]]))
+    function = pilot("pilot_ddrnet_runtime").layout_scores
+    reference, reference_dim = function(scores, "reference")
+    candidate, candidate_dim = function(scores, "channels-last")
+    np.testing.assert_array_equal(
+        reference.array.argmax(axis=reference_dim), candidate.array.argmax(axis=candidate_dim)
+    )
+    np.testing.assert_array_equal(candidate.array.argmax(axis=candidate_dim), [[[0, 2]]])
+    assert candidate.array.flags.c_contiguous
+    with pytest.raises(ValueError, match="unknown"):
+        function(scores, "silent-change")
+
+
+def test_gpu_stage_preserves_order_and_reserves_bounded_output_before_compute(pilot):
+    mailbox = pilot("pilot_queue").Mailbox(capacity=2, byte_limit=30)
+    started = [threading.Event(), threading.Event()]
+    stop = threading.Event()
+
+    def compute(bundle):
+        started[bundle["sequence"]].set()
+        return bundle["sequence"] * 10
+
+    stage = pilot("pilot_scheduler").GpuStage(mailbox, compute, stop)
+    try:
+        mailbox.put({"sequence": 0, "payload_bytes": 10})
+        assert started[0].wait(1)
+        mailbox.put({"sequence": 1, "payload_bytes": 10})
+        mailbox.finish()
+        assert not started[1].wait(0.05)  # No hidden completed third slot.
+        assert mailbox.bytes == 20  # GPU completion cannot release CPU-owned input.
+        first, result = stage.take()
+        assert (first["sequence"], result) == (0, 0)
+        assert started[1].wait(1)
+        second, result = stage.take()
+        assert (second["sequence"], result) == (1, 10)
+        assert mailbox.bytes == 20
+        mailbox.release(first)
+        mailbox.release(second)
+        assert stage.take() is None
+        assert mailbox.bytes == 0 and not mailbox.dropped
+        assert stage.peak_pending <= 1
+    finally:
+        assert stage.close()
+
+
+def test_gpu_stage_propagates_failure_and_stops_waiting_for_input(pilot):
+    mailbox = pilot("pilot_queue").Mailbox(capacity=1)
+
+    def compute(bundle):
+        raise ValueError("synthetic GPU failure")
+
+    stage = pilot("pilot_scheduler").GpuStage(mailbox, compute, threading.Event())
+    mailbox.put({"sequence": 0, "payload_bytes": 10})
+    try:
+        with pytest.raises(RuntimeError, match="synthetic GPU failure"):
+            stage.take()
+    finally:
+        assert stage.close()
+    waiting = pilot("pilot_scheduler").GpuStage(
+        pilot("pilot_queue").Mailbox(capacity=1), compute, threading.Event()
+    )
+    assert waiting.close()
+
+
+def test_completed_gpu_and_ingress_share_two_pending_slots(pilot):
+    mailbox = pilot("pilot_queue").Mailbox(capacity=2, byte_limit=100)
+    mailbox.put({"sequence": 0, "payload_bytes": 10})
+    active = mailbox.take()
+    for sequence in (1, 2):
+        mailbox.put({"sequence": sequence, "payload_bytes": 10})
+    mailbox.reserve_completed()
+    assert mailbox.dropped == [{"sequence": 1, "reason": "handoff-overflow"}]
+    assert mailbox.external_pending + len(mailbox.pending) == 2
+    mailbox.put({"sequence": 3, "payload_bytes": 10})
+    assert mailbox.dropped[-1] == {"sequence": 2, "reason": "pending-overflow"}
+    assert mailbox.peak_pending == 2 and mailbox.bytes == 20
+    mailbox.take_completed()
+    mailbox.release(active)
+    last = mailbox.take()
+    assert last["sequence"] == 3
+    mailbox.release(last)
+    assert mailbox.bytes == 0
+    with pytest.raises(ValueError, match="accounting"):
+        mailbox.take_completed()