feat(perception): add bounded triton runtime diagnostics

experiments/perception/worker/streaming_profile_stage1/compare_ddrnet_triton.py

@@ -0,0 +1,142 @@
+#!/usr/bin/env python3
+"""Compare the pinned PyTorch DDRNet mask with its FP32 TensorRT export."""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import http.client
+import json
+import os
+import signal
+import subprocess
+import time
+from pathlib import Path
+
+import cv2
+import numpy as np
+from pilot_ddrnet_preprocess import load_preprocess, preprocess_bgr
+from pilot_ddrnet_triton import TritonDdrnetMaskHttpInferenceBackend
+from pilot_ipc import receive, send
+
+
+def wait_ready(process):
+    deadline = time.monotonic() + 90
+    while time.monotonic() < deadline:
+        if process.poll() is not None:
+            raise RuntimeError("Triton exited before DDRNet readiness")
+        connection = http.client.HTTPConnection("127.0.0.1", 8000, timeout=1)
+        try:
+            connection.request("GET", "/v2/models/ddrnet_goose_fp32_mask/ready")
+            if connection.getresponse().status == 200:
+                return
+        except OSError:
+            pass
+        finally:
+            connection.close()
+        time.sleep(0.1)
+    raise TimeoutError("DDRNet Triton readiness exceeded budget")
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--video", default="/source.mp4")
+    parser.add_argument("--frames", type=int, default=128)
+    parser.add_argument("--output", type=Path, required=True)
+    args = parser.parse_args()
+    signal.alarm(240)
+    started = time.monotonic_ns()
+    args.output.parent.mkdir(parents=True, exist_ok=True)
+    log = args.output.with_suffix(".triton.log").open("wb")
+    triton = subprocess.Popen(
+        [
+            "tritonserver",
+            "--model-repository=/models",
+            "--model-control-mode=explicit",
+            "--load-model=ddrnet_goose_fp32_mask",
+            "--allow-grpc=false",
+            "--allow-metrics=false",
+            "--http-address=127.0.0.1",
+        ],
+        stdout=log,
+        stderr=subprocess.STDOUT,
+        start_new_session=True,
+    )
+    reference_log = args.output.with_suffix(".reference.log").open("wb")
+    reference = candidate = capture = None
+    try:
+        wait_ready(triton)
+        reference = subprocess.Popen(
+            [
+                "/opt/conda/envs/goose/bin/python",
+                "-B",
+                "/probe/pilot_model.py",
+                "ddrnet",
+            ],
+            stdin=subprocess.PIPE,
+            stdout=subprocess.PIPE,
+            stderr=reference_log,
+            env={**os.environ, "PYTHONPATH": "/probe"},
+        )
+        ready, _ = receive(reference.stdout)
+        if ready.get("ready") != "ddrnet":
+            raise RuntimeError("PyTorch DDRNet reference did not become ready")
+        candidate = TritonDdrnetMaskHttpInferenceBackend()
+        preprocess = load_preprocess()
+        zero = np.zeros((600, 800, 3), np.uint8)
+        for _ in range(8):
+            candidate.infer(preprocess_bgr(preprocess, zero))
+        capture = cv2.VideoCapture(args.video)
+        reference_hash = hashlib.sha256()
+        candidate_hash = hashlib.sha256()
+        rows = []
+        for sequence in range(args.frames):
+            ok, bgr = capture.read()
+            if not ok or bgr is None or bgr.shape != (600, 800, 3):
+                raise RuntimeError("DDRNet parity source ended or changed shape")
+            send(reference.stdin, {"op": "infer"}, bgr.tobytes())
+            reference_result, raw = receive(reference.stdout)
+            reference_mask = np.frombuffer(raw, np.uint8).reshape(512, 512)
+            candidate_started = time.monotonic_ns()
+            candidate_mask = candidate.infer(preprocess_bgr(preprocess, bgr))
+            candidate_ms = (time.monotonic_ns() - candidate_started) / 1e6
+            different = int(np.count_nonzero(reference_mask != candidate_mask))
+            reference_hash.update(reference_mask.tobytes())
+            candidate_hash.update(candidate_mask.tobytes())
+            rows.append(
+                {
+                    "sequence": sequence,
+                    "different_pixels": different,
+                    "reference_component_ms": reference_result["component_ms"],
+                    "candidate_preprocess_infer_ms": candidate_ms,
+                }
+            )
+        report = {
+            "schema_version": "missioncore.ddrnet-tensorrt-parity/v1",
+            "frames": args.frames,
+            "exact_frames": sum(row["different_pixels"] == 0 for row in rows),
+            "different_pixels": sum(row["different_pixels"] for row in rows),
+            "reference_mask_sequence_sha256": reference_hash.hexdigest(),
+            "candidate_mask_sequence_sha256": candidate_hash.hexdigest(),
+            "elapsed_ms": (time.monotonic_ns() - started) / 1e6,
+            "rows": rows,
+        }
+        args.output.write_text(json.dumps(report, indent=2, allow_nan=False) + "\n")
+        print(json.dumps({key: report[key] for key in report if key != "rows"}), flush=True)
+    finally:
+        if capture is not None:
+            capture.release()
+        if candidate is not None:
+            candidate.close()
+        if reference is not None and reference.poll() is None:
+            send(reference.stdin, {"op": "stop"})
+            reference.wait(timeout=5)
+        if triton.poll() is None:
+            triton.terminate()
+            triton.wait(timeout=5)
+        log.close()
+        reference_log.close()
+
+
+if __name__ == "__main__":
+    main()

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

@@ -0,0 +1,6 @@
+name: "ddrnet_goose_fp32_mask"
+platform: "tensorrt_plan"
+max_batch_size: 0
+input [{ name: "input" data_type: TYPE_FP32 dims: [1,3,512,512] }]
+output [{ name: "mask" data_type: TYPE_UINT8 dims: [1,512,512] }]
+instance_group [{ count: 1 kind: KIND_GPU gpus: [0] }]

experiments/perception/worker/streaming_profile_stage1/ddrnet-cuda-graphs-config.pbtxt

@@ -0,0 +1,7 @@
+name: "ddrnet_goose_fp32_mask"
+platform: "tensorrt_plan"
+max_batch_size: 0
+input [{ name: "input" data_type: TYPE_FP32 dims: [1,3,512,512] }]
+output [{ name: "mask" data_type: TYPE_UINT8 dims: [1,512,512] }]
+instance_group [{ count: 1 kind: KIND_GPU gpus: [0] }]
+optimization { cuda { graphs: true busy_wait_events: false } }

experiments/perception/worker/streaming_profile_stage1/ddrnet-cuda-graphs-spin-config.pbtxt

@@ -0,0 +1,7 @@
+name: "ddrnet_goose_fp32_mask"
+platform: "tensorrt_plan"
+max_batch_size: 0
+input [{ name: "input" data_type: TYPE_FP32 dims: [1,3,512,512] }]
+output [{ name: "mask" data_type: TYPE_UINT8 dims: [1,512,512] }]
+instance_group [{ count: 1 kind: KIND_GPU gpus: [0] }]
+optimization { cuda { graphs: true busy_wait_events: true } }

experiments/perception/worker/streaming_profile_stage1/export_ddrnet_onnx.py

@@ -0,0 +1,99 @@
+#!/usr/bin/env python3
+"""Export the pinned FP32 DDRNet logits graph for a bounded TensorRT probe."""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import importlib.util
+import json
+import signal
+import time
+from pathlib import Path
+
+CHECKPOINT_SHA256 = "b99c2838051bcd7b092fd3970aa62a77d5c0bbb809c9b9afb2ff4b0ebdaa4ee6"
+RUNNER_SHA256 = "b18ad60f277eea69a240a28f290611b94627fb9707faf1bb3e6e22102dad67c1"
+
+
+def digest(path):
+    value = hashlib.sha256()
+    with path.open("rb") as stream:
+        for chunk in iter(lambda: stream.read(1024 * 1024), b""):
+            value.update(chunk)
+    return value.hexdigest()
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        "--runner",
+        type=Path,
+        default=Path("/probe/run_goose_vegetation_benchmark.py"),
+    )
+    parser.add_argument("--checkpoint", type=Path, default=Path("/checkpoint.pth"))
+    parser.add_argument("--output", type=Path, required=True)
+    parser.add_argument("--output-kind", choices=("logits", "mask"), default="logits")
+    args = parser.parse_args()
+    signal.alarm(180)
+    if digest(args.runner) != RUNNER_SHA256 or digest(args.checkpoint) != CHECKPOINT_SHA256:
+        raise RuntimeError("DDRNet export input identity changed")
+    spec = importlib.util.spec_from_file_location("pinned_goose_export_runner", args.runner)
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    import onnx
+    import torch
+
+    class LogitsOnly(torch.nn.Module):
+        def __init__(self, model):
+            super().__init__()
+            self.model = model
+
+        def forward(self, value):
+            logits = module.logits_from_output(self.model(value))
+            if args.output_kind == "logits":
+                return logits
+            return torch.argmax(torch.sigmoid(logits), dim=1).to(torch.uint8)
+
+    started = time.monotonic_ns()
+    model, model_name, failures = module.load_model("ddrnet", args.checkpoint)
+    wrapper = LogitsOnly(model).eval()
+    example = torch.zeros((1, 3, 512, 512), dtype=torch.float32, device="cuda")
+    with torch.inference_mode():
+        output = wrapper(example)
+    expected = (
+        ((1, 64, 512, 512), torch.float32, "logits")
+        if args.output_kind == "logits"
+        else ((1, 512, 512), torch.uint8, "mask")
+    )
+    if tuple(output.shape) != expected[0] or output.dtype != expected[1]:
+        raise RuntimeError("DDRNet export output contract changed")
+    args.output.parent.mkdir(parents=True, exist_ok=True)
+    torch.onnx.export(
+        wrapper,
+        example,
+        str(args.output),
+        input_names=["input"],
+        output_names=[expected[2]],
+        opset_version=17,
+        do_constant_folding=True,
+    )
+    graph = onnx.load(str(args.output))
+    onnx.checker.check_model(graph)
+    result = {
+        "schema_version": "missioncore.ddrnet-onnx-export/v1",
+        "model": model_name,
+        "precision": "fp32",
+        "input": {"name": "input", "shape": [1, 3, 512, 512]},
+        "output": {"name": expected[2], "shape": list(expected[0])},
+        "checkpoint_sha256": CHECKPOINT_SHA256,
+        "runner_sha256": RUNNER_SHA256,
+        "onnx_sha256": digest(args.output),
+        "onnx_bytes": args.output.stat().st_size,
+        "elapsed_ms": (time.monotonic_ns() - started) / 1e6,
+        "load_failures": failures,
+    }
+    print("STAGE1_RESULT=" + json.dumps(result, sort_keys=True), flush=True)
+
+
+if __name__ == "__main__":
+    main()

experiments/perception/worker/streaming_profile_stage1/pilot_ddrnet_numpy.py

@@ -0,0 +1,68 @@
+"""Fixed KB4 DDRNet preprocessing without a second Python runtime or IPC."""
+
+from __future__ import annotations
+
+import numpy as np
+
+# Pinned runner: center crop 800x600 -> 600x600, Pillow NEAREST -> 512x512.
+# Integer pixel-center coordinates avoid a float rounding convention change.
+_NEAREST = ((2 * np.arange(512) + 1) * 600) // (2 * 512)
+
+
+def preprocess_bgr(bgr):
+    if bgr.shape != (600, 800, 3) or bgr.dtype != np.uint8:
+        raise ValueError("DDRNet native image contract changed")
+    rgb = bgr[_NEAREST[:, None], (100 + _NEAREST)[None, :], ::-1]
+    normalized = np.asarray(rgb, dtype=np.float32) / 255.0
+    return np.ascontiguousarray(normalized.transpose(2, 0, 1)[None], dtype="<f4")
+
+
+def main():
+    import argparse
+    import hashlib
+    import json
+    import os
+    import time
+    from pathlib import Path
+
+    import cv2
+    from pilot_ddrnet_preprocess import load_preprocess
+    from pilot_ddrnet_preprocess import preprocess_bgr as pinned_preprocess
+
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--video", default="/source.mp4")
+    parser.add_argument("--frames", type=int, default=128)
+    parser.add_argument("--output", type=Path, required=True)
+    args = parser.parse_args()
+    if os.environ.get("CUDA_VISIBLE_DEVICES") != "" or not 1 <= args.frames <= 256:
+        raise ValueError("preprocess parity is a bounded CPU-only check")
+    started = time.monotonic_ns()
+    pinned = load_preprocess()
+    reference_hash, candidate_hash = hashlib.sha256(), hashlib.sha256()
+    capture = cv2.VideoCapture(args.video)
+    try:
+        for _ in range(args.frames):
+            ok, bgr = capture.read()
+            if not ok:
+                raise ValueError("preprocess parity source ended")
+            reference, candidate = pinned_preprocess(pinned, bgr), preprocess_bgr(bgr)
+            if not np.array_equal(reference, candidate):
+                raise ValueError("native DDRNet preprocessing differs from pinned runner")
+            reference_hash.update(reference.tobytes())
+            candidate_hash.update(candidate.tobytes())
+    finally:
+        capture.release()
+    report = {
+        "schema_version": "missioncore.ddrnet-preprocess-parity/v1",
+        "frames": args.frames,
+        "exact_frames": args.frames,
+        "reference_tensor_sequence_sha256": reference_hash.hexdigest(),
+        "candidate_tensor_sequence_sha256": candidate_hash.hexdigest(),
+        "elapsed_ms": (time.monotonic_ns() - started) / 1e6,
+    }
+    args.output.write_text(json.dumps(report, indent=2) + "\n")
+    print(json.dumps(report), flush=True)
+
+
+if __name__ == "__main__":
+    main()

experiments/perception/worker/streaming_profile_stage1/pilot_ddrnet_preprocess.py

@@ -0,0 +1,64 @@
+#!/usr/bin/env python3
+"""Persistent exact DDRNet preprocessor with CUDA explicitly hidden."""
+
+from __future__ import annotations
+
+import hashlib
+import importlib.util
+import os
+import sys
+import time
+from pathlib import Path
+
+import cv2
+import numpy as np
+from PIL import Image
+from pilot_ipc import receive, send
+
+RUNNER_SHA256 = "b18ad60f277eea69a240a28f290611b94627fb9707faf1bb3e6e22102dad67c1"
+
+
+def digest(path):
+    return hashlib.sha256(path.read_bytes()).hexdigest()
+
+
+def load_preprocess(path=Path("/probe/run_goose_vegetation_benchmark.py")):
+    if digest(path) != RUNNER_SHA256:
+        raise RuntimeError("DDRNet preprocessing runner changed")
+    spec = importlib.util.spec_from_file_location("pinned_goose_preprocess_runner", path)
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module.preprocess
+
+
+def preprocess_bgr(preprocess, bgr):
+    if bgr.shape != (600, 800, 3) or bgr.dtype != np.uint8:
+        raise ValueError("DDRNet native image contract changed")
+    tensor, _ = preprocess(Image.fromarray(cv2.cvtColor(bgr, cv2.COLOR_BGR2RGB)))
+    value = np.ascontiguousarray(tensor.numpy(), dtype="<f4")
+    if value.shape != (1, 3, 512, 512):
+        raise ValueError("DDRNet preprocessing output changed")
+    return value
+
+
+def main():
+    if os.environ.get("CUDA_VISIBLE_DEVICES") != "":
+        raise RuntimeError("DDRNet preprocessing child must not see CUDA")
+    output = os.fdopen(os.dup(sys.stdout.fileno()), "wb", buffering=0)
+    os.dup2(sys.stderr.fileno(), sys.stdout.fileno())
+    preprocess = load_preprocess()
+    send(output, {"ready": "ddrnet-preprocess", "cuda_visible_devices": ""})
+    while True:
+        header, payload = receive(sys.stdin.buffer)
+        if header.get("op") == "stop":
+            return
+        if header != {"op": "preprocess"} or len(payload) != 600 * 800 * 3:
+            raise ValueError("invalid DDRNet preprocessing observation")
+        begin = time.monotonic_ns()
+        bgr = np.frombuffer(payload, np.uint8).reshape(600, 800, 3)
+        tensor = preprocess_bgr(preprocess, bgr)
+        send(output, {"preprocess_ms": (time.monotonic_ns() - begin) / 1e6}, tensor.tobytes())
+
+
+if __name__ == "__main__":
+    main()

experiments/perception/worker/streaming_profile_stage1/pilot_ddrnet_triton.py

@@ -0,0 +1,86 @@
+"""Bounded fixed-shape transport for the unqualified DDRNet TensorRT candidate."""
+
+from __future__ import annotations
+
+import http.client
+import json
+import time
+
+import numpy as np
+
+MODEL_ID = "ddrnet_goose_fp32_mask"
+MAX_RESPONSE_BYTES = 65536 + 512 * 512
+
+
+class TritonDdrnetMaskHttpInferenceBackend:
+    def __init__(self, host="127.0.0.1", port=8000, timeout=60):
+        self.connection = http.client.HTTPConnection(host, port, timeout=timeout)
+        self.path = f"/v2/models/{MODEL_ID}/versions/1/infer"
+        self.last_timing_ms = {}
+
+    def close(self):
+        self.connection.close()
+
+    def infer(self, tensor):
+        started = time.monotonic_ns()
+        value = np.ascontiguousarray(tensor, dtype="<f4")
+        if value.shape != (1, 3, 512, 512):
+            raise ValueError("DDRNet Triton input contract changed")
+        binary = value.tobytes()
+        descriptor = {
+            "inputs": [
+                {
+                    "name": "input",
+                    "shape": [1, 3, 512, 512],
+                    "datatype": "FP32",
+                    "parameters": {"binary_data_size": len(binary)},
+                }
+            ],
+            "outputs": [{"name": "mask", "parameters": {"binary_data": True}}],
+        }
+        header = json.dumps(descriptor, sort_keys=True, separators=(",", ":")).encode()
+        encoded = time.monotonic_ns()
+        self.connection.request(
+            "POST",
+            self.path,
+            body=header + binary,
+            headers={
+                "Content-Type": "application/octet-stream",
+                "Inference-Header-Content-Length": str(len(header)),
+            },
+        )
+        sent = time.monotonic_ns()
+        response = self.connection.getresponse()
+        responded = time.monotonic_ns()
+        payload = response.read(MAX_RESPONSE_BYTES + 1)
+        if len(payload) > MAX_RESPONSE_BYTES:
+            self.close()
+            raise RuntimeError("DDRNet Triton response exceeds byte budget")
+        received = time.monotonic_ns()
+        self.last_timing_ms = {
+            "http_encode_ms": (encoded - started) / 1e6,
+            "http_send_ms": (sent - encoded) / 1e6,
+            "http_wait_headers_ms": (responded - sent) / 1e6,
+            "http_read_body_ms": (received - responded) / 1e6,
+        }
+        if response.status != 200:
+            raise RuntimeError(f"DDRNet Triton inference failed: HTTP {response.status}")
+        try:
+            header_length = int(response.getheader("Inference-Header-Content-Length") or "")
+            if not 0 < header_length <= 65536:
+                raise ValueError("invalid header length")
+            outputs = json.loads(payload[:header_length])["outputs"]
+            if not isinstance(outputs, list) or len(outputs) != 1:
+                raise ValueError("invalid output count")
+            output = outputs[0]
+            valid = (
+                output["name"] == "mask"
+                and output["datatype"] == "UINT8"
+                and output["shape"] == [1, 512, 512]
+                and output["parameters"]["binary_data_size"] == 512 * 512
+            )
+        except (KeyError, TypeError, ValueError, json.JSONDecodeError) as error:
+            raise RuntimeError("DDRNet Triton response contract is invalid") from error
+        if not valid or len(payload) != header_length + 512 * 512:
+            raise RuntimeError("DDRNet Triton mask contract changed")
+        return np.frombuffer(payload, np.uint8, offset=header_length).reshape(512, 512).copy()

experiments/perception/worker/streaming_profile_stage1/pilot_telemetry.py

@@ -0,0 +1,88 @@
+"""Low-overhead Worker GPU telemetry through the already loaded NVML library."""
+
+from __future__ import annotations
+
+import ctypes
+
+
+class _MemoryInfo(ctypes.Structure):
+    _fields_ = [
+        ("total", ctypes.c_ulonglong),
+        ("free", ctypes.c_ulonglong),
+        ("used", ctypes.c_ulonglong),
+    ]
+
+
+class _Utilization(ctypes.Structure):
+    _fields_ = [("gpu", ctypes.c_uint), ("memory", ctypes.c_uint)]
+
+
+class NvmlSampler:
+    """Keep one NVML session instead of spawning ``nvidia-smi`` per sample."""
+
+    def __init__(self, library=None):
+        self.library = library or ctypes.CDLL("libnvidia-ml.so.1")
+        self.handle = ctypes.c_void_p()
+        self.closed = False
+        self._configure()
+        self._check(self.library.nvmlInit_v2(), "initialize")
+        try:
+            self._check(
+                self.library.nvmlDeviceGetHandleByIndex_v2(0, ctypes.byref(self.handle)),
+                "select device",
+            )
+        except Exception:
+            self.library.nvmlShutdown()
+            self.closed = True
+            raise
+
+    def _configure(self):
+        self.library.nvmlInit_v2.argtypes = []
+        self.library.nvmlInit_v2.restype = ctypes.c_int
+        self.library.nvmlShutdown.argtypes = []
+        self.library.nvmlShutdown.restype = ctypes.c_int
+        self.library.nvmlDeviceGetHandleByIndex_v2.argtypes = [
+            ctypes.c_uint,
+            ctypes.POINTER(ctypes.c_void_p),
+        ]
+        self.library.nvmlDeviceGetHandleByIndex_v2.restype = ctypes.c_int
+        self.library.nvmlDeviceGetMemoryInfo.argtypes = [
+            ctypes.c_void_p,
+            ctypes.POINTER(_MemoryInfo),
+        ]
+        self.library.nvmlDeviceGetMemoryInfo.restype = ctypes.c_int
+        self.library.nvmlDeviceGetUtilizationRates.argtypes = [
+            ctypes.c_void_p,
+            ctypes.POINTER(_Utilization),
+        ]
+        self.library.nvmlDeviceGetUtilizationRates.restype = ctypes.c_int
+
+    @staticmethod
+    def _check(code, operation):
+        if code != 0:
+            raise RuntimeError(f"NVML failed to {operation}: code {code}")
+
+    def sample(self):
+        memory = _MemoryInfo()
+        utilization = _Utilization()
+        self._check(
+            self.library.nvmlDeviceGetMemoryInfo(self.handle, ctypes.byref(memory)),
+            "read memory",
+        )
+        self._check(
+            self.library.nvmlDeviceGetUtilizationRates(
+                self.handle,
+                ctypes.byref(utilization),
+            ),
+            "read utilization",
+        )
+        return {
+            "gpu_used_mib": int((memory.used + 1048575) // 1048576),
+            "gpu_utilization": int(utilization.gpu),
+        }
+
+    def close(self):
+        if self.closed:
+            return
+        self.closed = True
+        self._check(self.library.nvmlShutdown(), "shutdown")

experiments/perception/worker/streaming_profile_stage1/pilot_triton_stats.py

@@ -0,0 +1,45 @@
+"""Warmup-excluded server statistics; queried outside the measured stream."""
+
+from __future__ import annotations
+
+import http.client
+import json
+
+
+def read_stats(model_ids):
+    connection = http.client.HTTPConnection("127.0.0.1", 8000, timeout=5)
+    result = {}
+    try:
+        for model_id in model_ids:
+            connection.request("GET", f"/v2/models/{model_id}/versions/1/stats")
+            response = connection.getresponse()
+            payload = response.read(1024 * 1024 + 1)
+            if response.status != 200 or len(payload) > 1024 * 1024:
+                raise RuntimeError("Triton statistics unavailable or unbounded")
+            rows = json.loads(payload)["model_stats"]
+            if len(rows) != 1 or rows[0]["name"] != model_id or rows[0]["version"] != "1":
+                raise RuntimeError("Triton statistics model identity changed")
+            result[model_id] = rows[0]["inference_stats"]
+    finally:
+        connection.close()
+    return result
+
+
+def stats_delta(before, after):
+    result = {}
+    for model_id, stages in before.items():
+        result[model_id] = {}
+        for name in (
+            "success", "fail", "queue", "compute_input", "compute_infer", "compute_output"
+        ):
+            first, last = stages[name], after[model_id][name]
+            count = int(last["count"]) - int(first["count"])
+            duration_ns = int(last["ns"]) - int(first["ns"])
+            if count < 0 or duration_ns < 0:
+                raise RuntimeError("Triton statistics counters went backwards")
+            result[model_id][name] = {
+                "count": count,
+                "total_ms": duration_ns / 1e6,
+                "mean_ms": duration_ns / count / 1e6 if count else None,
+            }
+    return result

experiments/perception/worker/streaming_profile_stage1/profile_ddrnet_tensor.py

@@ -0,0 +1,58 @@
+"""Bounded component diagnostic on one real frame; not a realtime proof."""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import subprocess
+import time
+from pathlib import Path
+
+import cv2
+from pilot_ddrnet_numpy import preprocess_bgr
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--output", type=Path, required=True)
+    args = parser.parse_args()
+    args.output.mkdir(parents=True, exist_ok=False)
+    started = time.monotonic_ns()
+    capture = cv2.VideoCapture("/source.mp4")
+    try:
+        ok, image = capture.read()
+        if not ok:
+            raise ValueError("real tensor source unavailable")
+        payload = preprocess_bgr(image).tobytes()
+    finally:
+        capture.release()
+    tensor_path = args.output / "frame-000000.f32"
+    tensor_path.write_bytes(payload)
+    command = [
+        "/usr/bin/trtexec", "--loadEngine=/model.plan",
+        f"--loadInputs=input:{tensor_path}", "--warmUp=500", "--duration=0",
+        "--iterations=64", "--avgRuns=64", "--dumpProfile",
+        f"--exportProfile={args.output / 'layers.json'}",
+    ]
+    with (args.output / "trtexec.log").open("wb") as log:
+        result = subprocess.run(command, stdout=log, stderr=subprocess.STDOUT, timeout=30)
+    report = {
+        "schema_version": "missioncore.ddrnet-real-tensor-component-profile/v1",
+        "source_sequence": 0,
+        "repeated_same_frame": True,
+        "source_paced": False,
+        "transport_included": False,
+        "input_sha256": hashlib.sha256(payload).hexdigest(),
+        "input_bytes": len(payload),
+        "command": command,
+        "returncode": result.returncode,
+        "elapsed_ms": (time.monotonic_ns() - started) / 1e6,
+    }
+    (args.output / "report.json").write_text(json.dumps(report, indent=2) + "\n")
+    print(json.dumps(report), flush=True)
+    return result.returncode
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

experiments/perception/worker/streaming_profile_stage1/run_joint_pilot.py

@@ -29,6 +29,7 @@ from pilot_queue import Mailbox
 from pilot_scheduler import GpuStage
 from pilot_sensor_binding import bind_sensors, increment_identity, milliseconds
 from pilot_source import SensorArchive, camera_events, merged_events
+from pilot_telemetry import NvmlSampler
 
 
 def distribution(values):
@@ -53,6 +54,14 @@ def payload_size(bundle):
     )
 
 
+def layer_refresh_due(last_source_ns, current_source_ns, minimum_interval_ms):
+    if last_source_ns is None:
+        return True
+    if current_source_ns < last_source_ns:
+        raise ValueError("layer source clock moved backwards")
+    return current_source_ns - last_source_ns >= minimum_interval_ms * 1_000_000
+
+
 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
@@ -172,52 +181,51 @@ def produce(args, decoder, mailbox, stop, report):
 
 
 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()
+    sampler = None
+    try:
+        sampler = NvmlSampler()
+        while not stop.is_set():
+            sample = {"monotonic_ns": time.monotonic_ns()}
+            try:
+                sample.update(sampler.sample())
+                sample["cgroup_memory_mib"] = (
+                    int(Path("/sys/fs/cgroup/memory.current").read_text()) / 1048576
                 )
-            }
-        except Exception as error:
-            sample["error"] = str(error)
-        samples.append(sample)
-        stop.wait(0.5)
+                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)
+    except Exception as error:
+        samples.append({"monotonic_ns": time.monotonic_ns(), "error": str(error)})
+    finally:
+        if sampler is not None:
+            sampler.close()
 
 
-def wait_triton(process, seconds=90):
+def wait_triton(process, model_ids, 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()
+        ready = True
+        for model_id in model_ids:
+            connection = http.client.HTTPConnection("127.0.0.1", 8000, timeout=1)
+            try:
+                connection.request("GET", f"/v2/models/{model_id}/ready")
+                ready = ready and connection.getresponse().status == 200
+            except OSError:
+                ready = False
+            finally:
+                connection.close()
+        if ready:
+            return
         time.sleep(0.1)
     raise TimeoutError("Triton initialization exceeded budget")
 
@@ -242,13 +250,20 @@ def run(args):
         "clock_quality": "original host arrival best effort; not hardware synchronization",
         "ddrnet_postprocess_layout": args.ddrnet_layout,
         "ddrnet_execution": args.ddrnet_execution,
+        "ddrnet_runtime": args.ddrnet_runtime,
+        "ddrnet_backend_status": (
+            "reference" if args.ddrnet_runtime == "pytorch" else "experimental-unqualified"
+        ),
+        "ddrnet_preprocess": args.ddrnet_preprocess,
         "schedule": args.schedule,
+        "telemetry_mode": args.telemetry_mode,
+        "ddrnet_min_source_interval_ms": args.ddrnet_min_source_interval_ms,
     }
     children, logs, results, samples = [], [], [], []
     stop = threading.Event()
     mailbox = Mailbox(capacity=2)
     source_report = {}
-    producer = monitor = graph = gpu_stage = None
+    producer = monitor = graph = gpu_stage = ddr_backend = None
 
     def child(name, command, env=None):
         log = (output / (name + ".log")).open("wb")
@@ -270,39 +285,78 @@ def run(args):
     previous = signal.signal(signal.SIGALRM, timeout_handler)
     signal.alarm(240)
     try:
+        triton_models = ["rf_detr_large_native_kb4"]
+        if args.ddrnet_runtime == "triton":
+            triton_models.append("ddrnet_goose_fp32_mask")
+        triton_command = [
+            "tritonserver",
+            "--model-repository=/models",
+            "--model-control-mode=explicit",
+            *[f"--load-model={model_id}" for model_id in triton_models],
+            "--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 = 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_command,
         )
         # 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)
+        wait_triton(triton, triton_models)
+        report["triton_config_sha256"] = {
+            model_id: hashlib.sha256(
+                Path(f"/models/{model_id}/config.pbtxt").read_bytes()
+            ).hexdigest()
+            for model_id in triton_models
+        }
         python = "/opt/conda/envs/goose/bin/python"
-        ddr = child(
-            "ddrnet",
-            [
-                python,
-                "-B",
-                "/probe/pilot_model.py",
+        if args.ddrnet_runtime == "pytorch":
+            ddr = child(
                 "ddrnet",
-                "--ddrnet-layout",
-                args.ddrnet_layout,
-                "--ddrnet-execution",
-                args.ddrnet_execution,
-            ],
-            {**os.environ, "PYTHONPATH": "/probe"},
-        )
-        report["ddrnet_ready"], _ = receive(ddr.stdout)
+                [
+                    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)
+        else:
+            from pilot_ddrnet_numpy import preprocess_bgr as native_preprocess
+            from pilot_ddrnet_triton import TritonDdrnetMaskHttpInferenceBackend
+
+            ddr = None
+            if args.ddrnet_preprocess == "pinned-child":
+                ddr = child(
+                    "ddrnet-preprocess",
+                    [python, "-B", "/probe/pilot_ddrnet_preprocess.py"],
+                    {
+                        **os.environ,
+                        "PYTHONPATH": "/probe",
+                        "CUDA_VISIBLE_DEVICES": "",
+                    },
+                )
+                report["ddrnet_ready"], _ = receive(ddr.stdout)
+            else:
+                report["ddrnet_ready"] = {"ready": "native-numpy-preprocess"}
+            ddr_backend = TritonDdrnetMaskHttpInferenceBackend()
+            zero = np.zeros((600, 800, 3), np.uint8)
+            for _ in range(8):
+                if ddr:
+                    send(ddr.stdin, {"op": "preprocess"}, zero.tobytes())
+                    _, raw = receive(ddr.stdout)
+                    tensor = np.frombuffer(raw, "<f4").reshape(1, 3, 512, 512)
+                else:
+                    tensor = native_preprocess(zero)
+                ddr_backend.infer(tensor)
         decoder = child(
             "decoder",
             [python, "-B", "/probe/pilot_model.py", "camera"],
@@ -323,19 +377,77 @@ def run(args):
         report["warmup_ms"] = (time.monotonic_ns() - started) / 1e6
         report["costmap_grid"] = graph.grid.tolist()
         report["material_names"] = graph.material_names
+        from pilot_triton_stats import read_stats, stats_delta
+
+        stats_before = read_stats(triton_models)
+
+        ddr_cache = None
 
         def compute_gpu(bundle):
+            nonlocal ddr_cache
             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)
+            execute_ddrnet = layer_refresh_due(
+                ddr_cache["source_ns"] if ddr_cache is not None else None,
+                bundle["time_ns"],
+                args.ddrnet_min_source_interval_ms,
+            )
+            if execute_ddrnet:
+                if args.ddrnet_runtime == "pytorch":
+                    send(ddr.stdin, {"op": "infer"}, bundle["image"].tobytes())
+                    ddr_result, raw = receive(ddr.stdout)
+                    mask = np.frombuffer(raw, np.uint8).reshape(512, 512)
+                else:
+                    if ddr:
+                        send(ddr.stdin, {"op": "preprocess"}, bundle["image"].tobytes())
+                        preprocess_result, raw = receive(ddr.stdout)
+                        tensor = np.frombuffer(raw, "<f4").reshape(1, 3, 512, 512)
+                    else:
+                        tensor = native_preprocess(bundle["image"])
+                        preprocess_result = {
+                            "preprocess_ms": (time.monotonic_ns() - begin) / 1e6
+                        }
+                    preprocessed = time.monotonic_ns()
+                    mask = ddr_backend.infer(tensor)
+                    inferred = time.monotonic_ns()
+                    ddr_result = {
+                        "component_ms": (inferred - begin) / 1e6,
+                        "forward_ms": (inferred - preprocessed) / 1e6,
+                        "stages_ms": {
+                            "preprocess_wall_ms": preprocess_result["preprocess_ms"],
+                            "preprocess_rpc_ms": (preprocessed - begin) / 1e6,
+                            "triton_infer_wall_ms": (inferred - preprocessed) / 1e6,
+                            **ddr_backend.last_timing_ms,
+                        },
+                    }
+                ddr_cache = {
+                    "source_ns": bundle["time_ns"],
+                    "source_sequence": bundle["sequence"],
+                    "result": ddr_result,
+                    "mask": mask,
+                }
+            else:
+                ddr_result = {
+                    "component_ms": 0.0,
+                    "forward_ms": 0.0,
+                    "stages_ms": {
+                        key: 0.0 for key in ddr_cache["result"]["stages_ms"]
+                    },
+                }
+                mask = ddr_cache["mask"]
             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
+            ddr_layer = {
+                "state": "current" if execute_ddrnet else "reused",
+                "source_sequence": ddr_cache["source_sequence"],
+                "source_host_monotonic_ns": ddr_cache["source_ns"],
+                "source_age_ms": (bundle["time_ns"] - ddr_cache["source_ns"]) / 1e6,
+            }
+            return begin, ddr_done, gpu_done, ddr_result, mask, proposals, ddr_layer
 
-        monitor = threading.Thread(target=telemetry, args=(stop, samples), daemon=True)
-        monitor.start()
+        if args.telemetry_mode == "nvml":
+            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
         )
@@ -354,7 +466,7 @@ def run(args):
                     if bundle is None:
                         break
                     computed = compute_gpu(bundle)
-                begin, ddr_done, gpu_done, ddr_result, mask, proposals = computed
+                begin, ddr_done, gpu_done, ddr_result, mask, proposals, ddr_layer = computed
                 cpu_started = time.monotonic_ns()
                 scene = graph.process(
                     bundle,
@@ -363,12 +475,35 @@ def run(args):
                     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(
+                policy_age_ms = (time.monotonic_ns() - bundle["due_ns"]) / 1e6
+                sensor_source_age_ms = max(
                     bundle["lineage"]["pose_age_ms"] or 0,
                     bundle["lineage"]["oldest_point_age_ms"] or 0,
                 )
+                scene["layer_freshness"] = {
+                    "segmentation": {
+                        **ddr_layer,
+                        "age_at_policy_ms": policy_age_ms + ddr_layer["source_age_ms"],
+                    },
+                    "detection": {
+                        "state": "current",
+                        "source_sequence": bundle["sequence"],
+                        "source_host_monotonic_ns": bundle["time_ns"],
+                        "source_age_ms": 0.0,
+                        "age_at_policy_ms": policy_age_ms,
+                    },
+                    "geometry": {
+                        "state": "current" if bundle["available"] else "unavailable",
+                        "source_sequence": bundle["sequence"],
+                        "source_host_monotonic_ns": bundle["time_ns"],
+                        "source_age_ms": sensor_source_age_ms,
+                        "age_at_policy_ms": policy_age_ms + sensor_source_age_ms,
+                    },
+                }
+                scene["oldest_required_input_age_ms"] = policy_age_ms + max(
+                    sensor_source_age_ms,
+                    ddr_layer["source_age_ms"],
+                )
                 scene["stale_at_publication"] = scene["oldest_required_input_age_ms"] > 250
                 if scene["stale_at_publication"]:
                     scene["policy_actions"] = [2] * len(scene["policy_actions"])
@@ -412,6 +547,8 @@ def run(args):
                     "finished_monotonic_ns": finished,
                     "timing_ms": timing,
                     "available": bundle["available"],
+                    "ddrnet_state": ddr_layer["state"],
+                    "ddrnet_source_age_ms": ddr_layer["source_age_ms"],
                     "surface_state": scene["surface_state"],
                     "proposal_count": len(scene["proposals"]),
                     "observation_count": len(scene["observations"]),
@@ -447,6 +584,9 @@ def run(args):
                     )
         if mailbox.error:
             raise RuntimeError(mailbox.error)
+        report["triton_statistics_delta"] = stats_delta(
+            stats_before, read_stats(triton_models)
+        )
         report["execution_complete"] = True
     except Exception:
         report["execution_complete"] = False
@@ -474,6 +614,8 @@ def run(args):
         report["gpu_stage_stopped"] = gpu_stage.close() if gpu_stage else True
         if graph:
             graph.backend.close()
+        if ddr_backend:
+            ddr_backend.close()
         for log in logs:
             log.close()
         report["stop_ms"] = (time.monotonic_ns() - shutdown_start) / 1e6
@@ -511,6 +653,15 @@ def run(args):
         "dropped": len(mailbox.dropped),
         "unaccounted": camera_count - len(results) - len(mailbox.dropped),
     }
+    ddrnet_executed = sum(r.get("ddrnet_state") == "current" for r in results)
+    report["model_cadence"] = {
+        "ddrnet_executed": ddrnet_executed,
+        "ddrnet_reused": len(results) - ddrnet_executed,
+        "ddrnet_source_age_ms": distribution(
+            [r["ddrnet_source_age_ms"] for r in results]
+        ),
+        "detector_executed": len(results),
+    }
     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"]
@@ -552,6 +703,8 @@ def run(args):
         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),
+        "layer_source_age_250ms": bool(results)
+        and all(r["ddrnet_source_age_ms"] <= 250 for r in results),
         "network_end_to_end_qualified": False,
     }
     report["profile_realtime_qualified"] = False
@@ -594,8 +747,16 @@ if __name__ == "__main__":
         "--ddrnet-layout", choices=("reference", "channels-last"), default="reference"
     )
     parser.add_argument("--ddrnet-execution", choices=("eager", "cuda-graph"), default="eager")
+    parser.add_argument("--ddrnet-runtime", choices=("pytorch", "triton"), default="pytorch")
+    parser.add_argument(
+        "--ddrnet-preprocess", choices=("pinned-child", "native-numpy"), default="pinned-child"
+    )
     parser.add_argument("--schedule", choices=("serial", "overlap-cpu"), default="serial")
+    parser.add_argument("--telemetry-mode", choices=("nvml", "none"), default="nvml")
+    parser.add_argument("--ddrnet-min-source-interval-ms", type=float, default=0.0)
     args = parser.parse_args()
     if not 1 <= args.frames <= 256:
         parser.error("pilot window must be 1..256 frames")
+    if not 0 <= args.ddrnet_min_source_interval_ms <= 250:
+        parser.error("DDRNet source interval must be 0..250 ms")
     raise SystemExit(run(args))

tests/test_perception_joint_pilot.py

@@ -98,6 +98,122 @@ def test_nearest_rank_tail_metrics_keep_outlier(pilot):
     assert pilot("run_joint_pilot").distribution([1, 200])["p99"] == 200
 
 
+def test_layer_refresh_cadence_is_source_clocked_and_fail_closed(pilot):
+    refresh = pilot("run_joint_pilot").layer_refresh_due
+
+    assert refresh(None, 1_000_000_000, 50)
+    assert not refresh(1_000_000_000, 1_049_999_999, 50)
+    assert refresh(1_000_000_000, 1_050_000_000, 50)
+    with pytest.raises(ValueError, match="backwards"):
+        refresh(1_000_000_000, 999_999_999, 50)
+
+
+def test_native_ddrnet_preprocess_matches_pillow_nearest_exactly(pilot):
+    from PIL import Image
+
+    bgr = np.random.default_rng(41).integers(0, 256, (600, 800, 3), dtype=np.uint8)
+    rgb = Image.fromarray(bgr[:, :, ::-1]).crop((100, 0, 700, 600))
+    resized = rgb.resize((512, 512), Image.Resampling.NEAREST)
+    expected = (np.asarray(resized, dtype=np.float32) / 255.0).transpose(2, 0, 1)[None]
+    actual = pilot("pilot_ddrnet_numpy").preprocess_bgr(bgr)
+    np.testing.assert_array_equal(actual, expected)
+    assert actual.flags.c_contiguous and actual.dtype == np.dtype("<f4")
+    with pytest.raises(ValueError, match="contract"):
+        pilot("pilot_ddrnet_numpy").preprocess_bgr(bgr[:512])
+
+
+def test_triton_statistics_exclude_warmup_and_reject_counter_reset(pilot):
+    names = ("success", "fail", "queue", "compute_input", "compute_infer", "compute_output")
+    before = {"model": {name: {"count": "8", "ns": "1000000"} for name in names}}
+    after = {"model": {name: {"count": "10", "ns": "5000000"} for name in names}}
+    delta = pilot("pilot_triton_stats").stats_delta(before, after)
+    assert delta["model"]["compute_infer"] == {"count": 2, "total_ms": 4, "mean_ms": 2}
+    with pytest.raises(RuntimeError, match="backwards"):
+        pilot("pilot_triton_stats").stats_delta(after, before)
+
+
+def test_ddrnet_triton_transport_is_bounded_and_validates_output_identity(pilot):
+    module = pilot("pilot_ddrnet_triton")
+    mask = np.zeros((512, 512), np.uint8)
+    descriptor = {"outputs": [{
+        "name": "mask", "datatype": "UINT8", "shape": [1, 512, 512],
+        "parameters": {"binary_data_size": mask.nbytes},
+    }]}
+    header = json.dumps(descriptor).encode()
+
+    class Response:
+        status = 200
+        payload = header + mask.tobytes()
+
+        def read(self, limit):
+            assert limit == module.MAX_RESPONSE_BYTES + 1
+            return self.payload[:limit]
+
+        def getheader(self, _name):
+            return str(len(header))
+
+    response = Response()
+    backend = module.TritonDdrnetMaskHttpInferenceBackend()
+    backend.connection = SimpleNamespace(
+        request=lambda *args, **kwargs: None,
+        getresponse=lambda: response,
+        close=lambda: None,
+    )
+    tensor = np.zeros((1, 3, 512, 512), np.float32)
+    np.testing.assert_array_equal(backend.infer(tensor), mask)
+    response.payload = b"x" * (module.MAX_RESPONSE_BYTES + 1)
+    with pytest.raises(RuntimeError, match="budget"):
+        backend.infer(tensor)
+    response.payload = header.replace(b'"UINT8"', b'"FP32" ') + mask.tobytes()
+    with pytest.raises(RuntimeError, match="contract"):
+        backend.infer(tensor)
+
+
+def test_nvml_sampler_reuses_one_session_without_spawning_processes(pilot):
+    class Function:
+        def __init__(self, callback):
+            self.callback = callback
+            self.calls = 0
+
+        def __call__(self, *args):
+            self.calls += 1
+            return self.callback(*args)
+
+    def set_handle(_index, pointer):
+        pointer._obj.value = 7
+        return 0
+
+    def set_memory(_handle, pointer):
+        pointer._obj.total = 24 * 1024 * 1024
+        pointer._obj.free = 20 * 1024 * 1024
+        pointer._obj.used = 3 * 1024 * 1024 + 1
+        return 0
+
+    def set_utilization(_handle, pointer):
+        pointer._obj.gpu = 42
+        pointer._obj.memory = 3
+        return 0
+
+    library = SimpleNamespace(
+        nvmlInit_v2=Function(lambda: 0),
+        nvmlShutdown=Function(lambda: 0),
+        nvmlDeviceGetHandleByIndex_v2=Function(set_handle),
+        nvmlDeviceGetMemoryInfo=Function(set_memory),
+        nvmlDeviceGetUtilizationRates=Function(set_utilization),
+    )
+    sampler = pilot("pilot_telemetry").NvmlSampler(library)
+
+    assert sampler.sample() == {"gpu_used_mib": 4, "gpu_utilization": 42}
+    assert sampler.sample() == {"gpu_used_mib": 4, "gpu_utilization": 42}
+    sampler.close()
+    sampler.close()
+
+    assert library.nvmlInit_v2.calls == 1
+    assert library.nvmlShutdown.calls == 1
+    assert library.nvmlDeviceGetMemoryInfo.calls == 2
+    assert library.nvmlDeviceGetUtilizationRates.calls == 2
+
+
 def test_body_history_is_bounded_and_never_reads_future(pilot):
     store = pilot("pilot_graph").CurrentStore(None)
     for index in range(100):