fix(perception): reject material ties and trace gpu pacing

2026-09-02 09:34:20 +03:00
parent ce09d30c1a
commit c6c42c29e8
8 changed files with 714 additions and 8 deletions
@@ -0,0 +1,7 @@
+# Diagnostic only: never use 64 MiB logits as the streaming profile output.
+name: "ddrnet_goose_fp32_logits"
+platform: "tensorrt_plan"
+max_batch_size: 0
+input [{ name: "input" data_type: TYPE_FP32 dims: [1,3,512,512] }]
+output [{ name: "logits" data_type: TYPE_FP32 dims: [1,64,512,512] }]
+instance_group [{ count: 1 kind: KIND_GPU gpus: [0] }]
@@ -0,0 +1,128 @@
+"""Offline single-cell attribution using the unchanged production-pilot voting.
+
+No detector, model GPU calls or obstacle-history simulation. Replays only causal
+raw point/pose increments to attribute material voting, not to qualify policy.
+"""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import signal
+import subprocess
+import time
+from collections import deque
+from datetime import datetime, timezone
+from pathlib import Path
+from types import SimpleNamespace
+
+import numpy as np
+from pilot_graph import JointGraph
+from pilot_sensor_binding import bind_sensors
+from pilot_source import SensorArchive, camera_events, merged_events
+
+
+def source_bundle(sequence):
+    archive = SensorArchive(Path("/sensor-source.npz"))
+    zero = next(camera_events("/camera-index.jsonl", 1)).time_ns - 500_000_000
+    rolling, fresh, pose, previous = deque(), [], None, None
+    try:
+        for event in merged_events(archive, "/camera-index.jsonl", sequence + 1):
+            if event.time_ns < zero:
+                continue
+            while rolling and event.time_ns - rolling[0].time_ns > 1_000_000_000:
+                rolling.popleft()
+            if event.channel == "pose":
+                pose = event
+            elif event.channel == "points":
+                rolling.append(event)
+                fresh.append(event)
+                if sum(len(e.value[0]) for e in rolling) > 64000:
+                    raise ValueError("diagnostic rolling history exceeds pilot bound")
+            else:
+                binding = bind_sensors(event.time_ns, pose, fresh, previous_camera_time_ns=previous)
+                if event.sequence == sequence:
+                    return {
+                        "available": binding.available,
+                        "time_ns": event.time_ns,
+                        "sensor_binding": binding.document(),
+                        "pose": pose.value if pose else None,
+                        "rolling_points": np.concatenate([e.value[0] for e in rolling]),
+                        "rolling_times": np.concatenate(
+                            [np.full(len(e.value[0]), e.time_ns, np.int64) for e in rolling]
+                        ),
+                    }
+                previous, fresh = event.time_ns, []
+        raise RuntimeError("diagnostic source frame unavailable")
+    finally:
+        archive.close()
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--output", type=Path, required=True)
+    parser.add_argument("--masks", type=Path, required=True)
+    parser.add_argument("--sequence", type=int, choices=range(128), default=115)
+    parser.add_argument("--cell", type=int, default=1900)
+    args = parser.parse_args()
+    signal.alarm(45)
+    args.output.mkdir(parents=True, exist_ok=False)
+    started = time.monotonic_ns()
+    started_utc = datetime.now(timezone.utc).isoformat()
+    bundle = source_bundle(args.sequence)
+    with (args.output / "tgs.log").open("wb") as log:
+        tgs = subprocess.Popen(
+            ["/usr/local/bin/pilot-tgs"], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=log
+        )
+        try:
+            graph = JointGraph(
+                "/code",
+                tgs,
+                mask_path="/valid-fov.png",
+                mapping_path="/goose.csv",
+                calibration_pack="/calibration.npz",
+            )
+            results = {}
+            with np.load(args.masks, allow_pickle=False) as masks:
+                for name in masks.files:
+                    mask = masks[name]
+                    if mask.shape != (512, 512) or mask.dtype != np.uint8:
+                        raise ValueError("diagnostic mask contract changed")
+                    cells, actions, material, evidence = graph.costmap(
+                        bundle,
+                        mask,
+                        SimpleNamespace(occupied=(), unknown=()),
+                        diagnostic_cells=(args.cell,),
+                    )
+                    results[name] = {
+                        "mask_sha256": hashlib.sha256(mask.tobytes()).hexdigest(),
+                        "cell_state": int(cells[args.cell]),
+                        "material": graph.material_names[material[args.cell]],
+                        "action_before_obstacle_history": int(actions[args.cell]),
+                        **evidence["diagnostic_cells"][str(args.cell)],
+                    }
+            report = {
+                "schema_version": "missioncore.ddrnet-costmap-attribution/v1",
+                "source_sequence": args.sequence,
+                "cell_index": args.cell,
+                "grid_cell": graph.grid[args.cell].tolist(),
+                "started_utc": started_utc,
+                "started_monotonic_ns": started,
+                "elapsed_ms": (time.monotonic_ns() - started) / 1e6,
+                "source_paced": False,
+                "qualified": False,
+                "actuation_enabled": False,
+                "obstacle_history_replayed": False,
+                "sensor_binding": bundle["sensor_binding"],
+                "results": results,
+            }
+            (args.output / "report.json").write_text(json.dumps(report, indent=2) + "\n")
+            print(json.dumps(report), flush=True)
+        finally:
+            tgs.stdin.close()
+            tgs.wait(timeout=3)
+
+
+if __name__ == "__main__":
+    main()
@@ -0,0 +1,256 @@
+"""Single-frame numerical diagnosis, never a production inference path.
+
+The 64 MiB logits response is confined to this bounded offline diagnostic.
+Reference, alternative arithmetic and TensorRT calls execute sequentially.
+"""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import http.client
+import importlib.util
+import json
+import signal
+import subprocess
+import time
+from datetime import datetime, timezone
+from pathlib import Path
+
+import numpy as np
+
+LOGITS_SHAPE = (1, 64, 512, 512)
+LOGITS_BYTES = int(np.prod(LOGITS_SHAPE)) * 4
+
+
+def read_logits(tensor):
+    binary = tensor.tobytes()
+    header = json.dumps(
+        {
+            "inputs": [
+                {
+                    "name": "input",
+                    "shape": [1, 3, 512, 512],
+                    "datatype": "FP32",
+                    "parameters": {"binary_data_size": len(binary)},
+                }
+            ],
+            "outputs": [{"name": "logits", "parameters": {"binary_data": True}}],
+        }
+    ).encode()
+    connection = http.client.HTTPConnection("127.0.0.1", 8000, timeout=30)
+    try:
+        connection.request(
+            "POST",
+            "/v2/models/ddrnet_goose_fp32_logits/infer",
+            body=header + binary,
+            headers={
+                "Inference-Header-Content-Length": str(len(header)),
+                "Content-Type": "application/octet-stream",
+            },
+        )
+        response = connection.getresponse()
+        payload = response.read(LOGITS_BYTES + 65537)
+        length = int(response.getheader("Inference-Header-Content-Length") or "0")
+        if response.status != 200 or not 0 < length <= 65536:
+            raise RuntimeError("invalid diagnostic logits response")
+        output = json.loads(payload[:length])["outputs"]
+        if (
+            len(payload) != length + LOGITS_BYTES
+            or len(output) != 1
+            or not (
+                output[0]["name"] == "logits"
+                and output[0]["datatype"] == "FP32"
+                and output[0]["shape"] == list(LOGITS_SHAPE)
+                and output[0]["parameters"]["binary_data_size"] == LOGITS_BYTES
+            )
+        ):
+            raise RuntimeError("diagnostic logits contract changed")
+        return np.frombuffer(payload, dtype="<f4", offset=length).reshape(LOGITS_SHAPE).copy()
+    finally:
+        connection.close()
+
+
+def compare_masks(reference, candidate):
+    if reference.shape != candidate.shape or reference.ndim != 2:
+        raise ValueError("mask shape mismatch")
+    changed = reference != candidate
+    pairs, counts = np.unique(
+        np.stack((reference[changed], candidate[changed]), axis=1), axis=0, return_counts=True
+    )
+    return {
+        "different_pixels": int(changed.sum()),
+        "transitions": [
+            {"reference": int(pair[0]), "candidate": int(pair[1]), "pixels": int(count)}
+            for pair, count in zip(pairs, counts)
+        ],
+    }
+
+
+def score_details(logits, scores, mask, positions):
+    rows = []
+    for y, x in positions:
+        values, probabilities = logits[0, :, y, x], scores[0, :, y, x]
+        top = np.argsort(-values, kind="stable")[:2]
+        rows.append(
+            {
+                "y": int(y),
+                "x": int(x),
+                "selected": int(mask[y, x]),
+                "top_logits": [
+                    {"class": int(k), "logit": float(values[k]), "score": float(probabilities[k])}
+                    for k in top
+                ],
+                "logit_margin": float(values[top[0]] - values[top[1]]),
+                "max_score_ties": np.flatnonzero(probabilities == probabilities.max()).tolist(),
+            }
+        )
+    return rows
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--sequence", type=int, choices=range(128), default=115)
+    parser.add_argument("--output", type=Path, required=True)
+    args = parser.parse_args()
+    signal.alarm(180)
+    args.output.mkdir(parents=True, exist_ok=False)
+    started = time.monotonic_ns()
+    started_utc = datetime.now(timezone.utc).isoformat()
+    import cv2
+    import torch
+    from compare_ddrnet_triton import wait_ready
+    from export_ddrnet_onnx import CHECKPOINT_SHA256, RUNNER_SHA256, digest
+    from pilot_ddrnet_numpy import preprocess_bgr
+    from pilot_ddrnet_preprocess import preprocess_bgr as pinned_preprocess
+    from pilot_ddrnet_triton import TritonDdrnetMaskHttpInferenceBackend
+
+    runner = Path("/probe/run_goose_vegetation_benchmark.py")
+    if digest(runner) != RUNNER_SHA256 or digest(Path("/checkpoint.pth")) != CHECKPOINT_SHA256:
+        raise RuntimeError("diagnostic model identity changed")
+    spec = importlib.util.spec_from_file_location("pinned_numeric_runner", runner)
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    capture = cv2.VideoCapture("/source.mp4")
+    try:
+        for _ in range(args.sequence + 1):
+            ok, bgr = capture.read()
+            if not ok or bgr is None:
+                raise RuntimeError("diagnostic frame unavailable")
+    finally:
+        capture.release()
+    tensor = preprocess_bgr(bgr)
+    if not np.array_equal(tensor, pinned_preprocess(module.preprocess, bgr)):
+        raise RuntimeError("diagnostic preprocess parity failed")
+    defaults = {
+        "cudnn_allow_tf32": torch.backends.cudnn.allow_tf32,
+        "matmul_allow_tf32": torch.backends.cuda.matmul.allow_tf32,
+        "cudnn_benchmark": torch.backends.cudnn.benchmark,
+        "cudnn_deterministic": torch.backends.cudnn.deterministic,
+    }
+    model, model_name, _ = module.load_model("ddrnet", Path("/checkpoint.pth"))
+
+    def postprocess(logits):
+        scores = torch.sigmoid(logits)
+        mask = scores.argmax(dim=1)[0].to(dtype=torch.uint8)
+        torch.cuda.synchronize()
+        return logits.cpu().numpy(), scores.cpu().numpy(), mask.cpu().numpy()
+
+    def infer(value):
+        with torch.inference_mode():
+            return postprocess(module.logits_from_output(model(torch.from_numpy(value).cuda())))
+
+    for _ in range(8):
+        infer(np.zeros_like(tensor))
+    reference = infer(tensor)
+    repeat = infer(tensor)
+    torch.backends.cudnn.allow_tf32 = False
+    torch.backends.cuda.matmul.allow_tf32 = False
+    no_tf32 = infer(tensor)
+    torch.backends.cudnn.allow_tf32 = defaults["cudnn_allow_tf32"]
+    torch.backends.cuda.matmul.allow_tf32 = defaults["matmul_allow_tf32"]
+    log = (args.output / "triton.log").open("wb")
+    triton = subprocess.Popen(
+        [
+            "tritonserver",
+            "--model-repository=/models",
+            "--model-control-mode=explicit",
+            "--load-model=ddrnet_goose_fp32_mask",
+            "--load-model=ddrnet_goose_fp32_logits",
+            "--allow-grpc=false",
+            "--allow-metrics=false",
+            "--http-address=127.0.0.1",
+        ],
+        stdout=log,
+        stderr=subprocess.STDOUT,
+    )
+    backend = None
+    try:
+        wait_ready(triton)
+        backend = TritonDdrnetMaskHttpInferenceBackend()
+        for _ in range(8):
+            backend.infer(np.zeros_like(tensor))
+        candidate_mask = backend.infer(tensor)
+        candidate_repeat = backend.infer(tensor)
+        candidate_logits = read_logits(tensor)
+        with torch.inference_mode():
+            candidate = postprocess(torch.from_numpy(candidate_logits).cuda())
+        positions = np.argwhere(reference[2] != candidate_mask)
+        selected = positions[:512]
+        report = {
+            "schema_version": "missioncore.ddrnet-numeric-diagnostic/v1",
+            "source_sequence": args.sequence,
+            "source_paced": False,
+            "qualified": False,
+            "actuation_enabled": False,
+            "started_utc": started_utc,
+            "started_monotonic_ns": started,
+            "torch_version": torch.__version__,
+            "cuda_version": torch.version.cuda,
+            "cudnn_version": torch.backends.cudnn.version(),
+            "defaults": defaults,
+            "model": model_name,
+            "input_sha256": hashlib.sha256(tensor.tobytes()).hexdigest(),
+            "reference_repeat": compare_masks(reference[2], repeat[2]),
+            "reference_no_tf32": compare_masks(reference[2], no_tf32[2]),
+            "reference_trt_mask": compare_masks(reference[2], candidate_mask),
+            "no_tf32_trt_mask": compare_masks(no_tf32[2], candidate_mask),
+            "trt_repeat": compare_masks(candidate_mask, candidate_repeat),
+            "trt_logits_torch_postprocess_vs_trt_mask": compare_masks(candidate_mask, candidate[2]),
+            "reference_raw_argmax_vs_sigmoid": compare_masks(
+                reference[2], reference[0].argmax(1)[0]
+            ),
+            "trt_raw_argmax_vs_sigmoid": compare_masks(
+                candidate_mask, candidate_logits.argmax(1)[0]
+            ),
+            "reference_logit_repeat_max_abs": float(np.abs(reference[0] - repeat[0]).max()),
+            "reference_trt_logit_max_abs": float(np.abs(reference[0] - candidate_logits).max()),
+            "details_truncated": len(positions) > len(selected),
+            "reference_details": score_details(*reference, selected),
+            "candidate_logit_engine_details": score_details(*candidate, selected),
+            "elapsed_ms": (time.monotonic_ns() - started) / 1e6,
+        }
+        np.savez_compressed(
+            args.output / "masks.npz",
+            reference=reference[2],
+            no_tf32=no_tf32[2],
+            trt_mask=candidate_mask,
+            trt_logits_torch_postprocess=candidate[2],
+        )
+        (args.output / "report.json").write_text(
+            json.dumps(report, indent=2, allow_nan=False) + "\n"
+        )
+        print(
+            json.dumps({k: v for k, v in report.items() if not k.endswith("details")}), flush=True
+        )
+    finally:
+        if backend is not None:
+            backend.close()
+        if triton.poll() is None:
+            triton.terminate()
+            triton.wait(timeout=5)
+        log.close()
+
+
+if __name__ == "__main__":
+    main()
@@ -0,0 +1,177 @@
+"""Bounded, sequential trtexec pause/DMA diagnostic; not a source-paced gate."""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import http.client
+import json
+import signal
+import subprocess
+import threading
+import time
+from datetime import datetime, timezone
+from pathlib import Path
+
+from pilot_telemetry import NvmlSampler
+
+
+def triton_run(args, payload):
+    """Same fixed tensor via loopback HTTP; only one model/call in flight."""
+    import numpy as np
+    from pilot_ddrnet_triton import MODEL_ID, TritonDdrnetMaskHttpInferenceBackend
+    from pilot_triton_stats import read_stats, stats_delta
+
+    command = [
+        "tritonserver",
+        "--model-repository=/models",
+        "--model-control-mode=explicit",
+        f"--load-model={MODEL_ID}",
+        "--allow-grpc=false",
+        "--allow-metrics=false",
+        "--http-address=127.0.0.1",
+        "--log-verbose=1",
+    ]
+    tensor = np.frombuffer(payload, dtype="<f4").reshape(1, 3, 512, 512)
+    backend = TritonDdrnetMaskHttpInferenceBackend(timeout=5)
+    rows = []
+    with (args.output / "triton.log").open("wb") as log:
+        process = subprocess.Popen(command, stdout=log, stderr=subprocess.STDOUT)
+        try:
+            deadline = time.monotonic() + 20
+            while True:
+                if process.poll() is not None or time.monotonic() >= deadline:
+                    raise RuntimeError("diagnostic Triton not ready within bound")
+                connection = http.client.HTTPConnection("127.0.0.1", 8000, timeout=1)
+                try:
+                    connection.request("GET", f"/v2/models/{MODEL_ID}/ready")
+                    if connection.getresponse().status == 200:
+                        break
+                except OSError:
+                    pass
+                finally:
+                    connection.close()
+                time.sleep(0.05)
+            for _ in range(8):
+                backend.infer(np.zeros_like(tensor))
+            before = read_stats((MODEL_ID,))
+            for index in range(32):
+                if index:
+                    time.sleep(args.idle_ms / 1000)
+                started = time.monotonic_ns()
+                mask = backend.infer(tensor)
+                finished = time.monotonic_ns()
+                rows.append(
+                    {
+                        "index": index,
+                        "started_monotonic_ns": started,
+                        "finished_monotonic_ns": finished,
+                        "wall_ms": (finished - started) / 1e6,
+                        "mask_sha256": hashlib.sha256(mask.tobytes()).hexdigest(),
+                        **backend.last_timing_ms,
+                    }
+                )
+            stats = stats_delta(before, read_stats((MODEL_ID,)))
+            return {"returncode": 0, "command": command, "rows": rows, "triton_stats": stats}
+        finally:
+            backend.close()
+            if process.poll() is None:
+                process.terminate()
+                process.wait(timeout=5)
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--output", type=Path, required=True)
+    parser.add_argument("--idle-ms", type=int, choices=(0, 50, 100), default=0)
+    parser.add_argument("--include-transfers", action="store_true")
+    parser.add_argument("--blocking-sync", action="store_true")
+    parser.add_argument("--runtime", choices=("trtexec", "triton"), default="trtexec")
+    args = parser.parse_args()
+    if args.runtime == "triton" and (args.blocking_sync or args.include_transfers):
+        parser.error("Triton transfer/sync settings come from its model config")
+    signal.alarm(55)
+    args.output.mkdir(parents=True, exist_ok=False)
+    payload = Path("/input.f32").read_bytes()
+    if len(payload) != 1 * 3 * 512 * 512 * 4:
+        raise ValueError("diagnostic input contract changed")
+    command = [
+        "/usr/bin/trtexec",
+        "--loadEngine=/model.plan",
+        "--loadInputs=input:/input.f32",
+        "--warmUp=500",
+        "--duration=0",
+        "--iterations=32",
+        "--avgRuns=32",
+        "--infStreams=1",
+        "--exposeDMA",
+        f"--idleTime={args.idle_ms}",
+        f"--exportTimes={args.output / 'times.json'}",
+    ]
+    if args.include_transfers:
+        command.append("--includeDataTransfers")
+    if args.blocking_sync:
+        command.append("--noSpinWait")
+    started = time.monotonic_ns()
+    started_utc = datetime.now(timezone.utc).isoformat()
+    sampler, stopped = NvmlSampler(), threading.Event()
+    samples, errors = [], []
+
+    def sample():
+        try:
+            for _ in range(1000):
+                samples.append(
+                    {
+                        "monotonic_ns": time.monotonic_ns(),
+                        **sampler.sample(),
+                        **sampler.sample_device_state(),
+                    }
+                )
+                if stopped.wait(0.05):
+                    return
+            errors.append("diagnostic telemetry sample bound reached")
+        except Exception as error:
+            errors.append(str(error))
+
+    thread = threading.Thread(target=sample, daemon=True)
+    thread.start()
+    try:
+        if args.runtime == "triton":
+            result = triton_run(args, payload)
+        else:
+            with (args.output / "trtexec.log").open("wb") as log:
+                completed = subprocess.run(
+                    command, stdout=log, stderr=subprocess.STDOUT, timeout=45
+                )
+            result = {"command": command, "returncode": completed.returncode}
+    finally:
+        stopped.set()
+        thread.join(timeout=2)
+        if thread.is_alive():
+            raise RuntimeError("diagnostic telemetry did not stop")
+        sampler.close()
+    report = {
+        "schema_version": "missioncore.ddrnet-pacing-diagnostic/v1",
+        "started_utc": started_utc,
+        "started_monotonic_ns": started,
+        "elapsed_ms": (time.monotonic_ns() - started) / 1e6,
+        "repeated_same_frame": True,
+        "source_paced": False,
+        "qualified": False,
+        "actuation_enabled": False,
+        "input_sha256": hashlib.sha256(payload).hexdigest(),
+        "idle_ms": args.idle_ms,
+        "transfers_included": True if args.runtime == "triton" else args.include_transfers,
+        "blocking_sync": None if args.runtime == "triton" else args.blocking_sync,
+        "runtime": args.runtime,
+        **result,
+        "telemetry_errors": errors,
+        "samples": samples,
+    }
+    (args.output / "report.json").write_text(json.dumps(report, indent=2) + "\n")
+    print(json.dumps({k: v for k, v in report.items() if k not in ("samples", "rows")}), flush=True)
+    return result["returncode"]
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
@@ -167,6 +167,17 @@ def current_body(frame, surface):
    )


+def select_material(votes):
+    """A tied plurality is unknown, never an allowance by class ordering."""
+    if votes.ndim != 2 or votes.shape[1] < 1 or np.any(votes < 0):
+        raise ValueError("invalid material votes")
+    maximum = votes.max(axis=1)
+    unique = (votes == maximum[:, None]).sum(axis=1) == 1
+    winner = votes.argmax(axis=1).astype(np.int32)
+    winner[(maximum == 0) | ~unique] = 0  # material index 0 is always unknown
+    return winner
+
+
 class JointGraph:
    def __init__(self, code_root, tgs, *, mask_path, mapping_path, calibration_pack):
        root = Path(code_root)
@@ -372,7 +383,11 @@ class JointGraph:
            "actuation_allowed": False,
        }

-    def costmap(self, bundle, mask, obstacle_map):
+    def costmap(self, bundle, mask, obstacle_map, *, diagnostic_cells=()):
+        if len(diagnostic_cells) > 8 or any(
+            type(i) is not int or not 0 <= i < len(self.grid) for i in diagnostic_cells
+        ):
+            raise ValueError("costmap diagnostic cell bound exceeded")
        count = len(self.grid)
        material = np.zeros(count, np.int32)
        if not bundle["available"]:
@@ -424,7 +439,24 @@ class JointGraph:
        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)
+        material = select_material(votes)
+        diagnostics = {}
+        for index in diagnostic_cells:
+            matching = np.flatnonzero(good & (cell_ids == index))
+            diagnostics[str(index)] = {
+                "votes": dict(zip(self.material_names, votes[index].tolist())),
+                "projected_ground_points": len(matching),
+                "points_truncated": len(matching) > 128,
+                "points": [
+                    {
+                        "mask_y": int(uv[j, 1] * 512 / 600),
+                        "mask_x": int((uv[j, 0] - 100) * 512 / 600),
+                        "label": int(labels[j]),
+                        "material": self.material_names[self.material_lut[labels[j]]],
+                    }
+                    for j in matching[:128]
+                ],
+            }
        # 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:
@@ -447,5 +479,6 @@ class JointGraph:
                "rejected": int(np.count_nonzero(states == 3)),
                "stale_cells": int(np.count_nonzero(stale)),
                "state_counts": dict(Counter(int(x) for x in cells)),
+                **({"diagnostic_cells": diagnostics} if diagnostics else {}),
            },
        )
@@ -81,6 +81,38 @@ class NvmlSampler:
            "gpu_utilization": int(utilization.gpu),
        }

+    def sample_device_state(self):
+        """Optional read-only diagnostic; unsupported fields are not zeroes.
+
+        Kept out of the default streaming sampler to avoid changing its cost.
+        NVML clock enums: SM=1, memory=2. Power usage is in milliwatts.
+        """
+        if self.closed:
+            raise RuntimeError("NVML session is closed")
+        result, errors = {}, {}
+        for name, symbol, extra in (
+            ("sm_clock_mhz", "nvmlDeviceGetClockInfo", (1,)),
+            ("memory_clock_mhz", "nvmlDeviceGetClockInfo", (2,)),
+            ("pstate", "nvmlDeviceGetPerformanceState", ()),
+            ("power_mw", "nvmlDeviceGetPowerUsage", ()),
+        ):
+            function = getattr(self.library, symbol, None)
+            if function is None:
+                result[name], errors[name] = None, "symbol-unavailable"
+                continue
+            function.argtypes = [
+                ctypes.c_void_p,
+                *([ctypes.c_uint] * len(extra)),
+                ctypes.POINTER(ctypes.c_uint),
+            ]
+            function.restype = ctypes.c_int
+            value = ctypes.c_uint()
+            code = function(self.handle, *extra, ctypes.byref(value))
+            result[name] = int(value.value) if code == 0 else None
+            if code:
+                errors[name] = int(code)
+        return {**result, "device_state_errors": errors}
+
    def close(self):
        if self.closed:
            return
@@ -180,7 +180,7 @@ def produce(args, decoder, mailbox, stop, report):
        mailbox.finish(mailbox.error)


-def telemetry(stop, samples):
+def telemetry(stop, samples, device_state=False):
    sampler = None
    try:
        sampler = NvmlSampler()
@@ -188,6 +188,8 @@ def telemetry(stop, samples):
            sample = {"monotonic_ns": time.monotonic_ns()}
            try:
                sample.update(sampler.sample())
+                if device_state:
+                    sample.update(sampler.sample_device_state())
                sample["cgroup_memory_mib"] = (
                    int(Path("/sys/fs/cgroup/memory.current").read_text()) / 1048576
                )
@@ -235,6 +237,7 @@ def run(args):
    output.mkdir(parents=True, exist_ok=False)
    started = time.monotonic_ns()
    report = {
+        "material_vote_rule": "unique-plurality-or-unknown/v1",
        "schema_version": "missioncore.stage1-joint-pilot/v1",
        "run_id": args.run_id,
        "started_utc": datetime.now(UTC).isoformat(),
@@ -257,6 +260,8 @@ def run(args):
        "ddrnet_preprocess": args.ddrnet_preprocess,
        "schedule": args.schedule,
        "telemetry_mode": args.telemetry_mode,
+        "telemetry_device_state": args.telemetry_device_state,
+        "triton_verbose": args.triton_verbose,
        "ddrnet_min_source_interval_ms": args.ddrnet_min_source_interval_ms,
    }
    children, logs, results, samples = [], [], [], []
@@ -298,6 +303,7 @@ def run(args):
            "--http-address=127.0.0.1",
            "--pinned-memory-pool-byte-size=16777216",
            "--cuda-memory-pool-byte-size=0:16777216",
+            *(["--log-verbose=1"] if args.triton_verbose else []),
        ]
        triton = child(
            "triton",
@@ -446,7 +452,9 @@ def run(args):
            return begin, ddr_done, gpu_done, ddr_result, mask, proposals, ddr_layer

        if args.telemetry_mode == "nvml":
-            monitor = threading.Thread(target=telemetry, args=(stop, samples), daemon=True)
+            monitor = threading.Thread(
+                target=telemetry, args=(stop, samples, args.telemetry_device_state), daemon=True
+            )
            monitor.start()
        producer = threading.Thread(
            target=produce, args=(args, decoder, mailbox, stop, source_report), daemon=True
@@ -753,10 +761,14 @@ if __name__ == "__main__":
    )
    parser.add_argument("--schedule", choices=("serial", "overlap-cpu"), default="serial")
    parser.add_argument("--telemetry-mode", choices=("nvml", "none"), default="nvml")
+    parser.add_argument("--telemetry-device-state", action="store_true")
+    parser.add_argument("--triton-verbose", action="store_true")
    parser.add_argument("--ddrnet-min-source-interval-ms", type=float, default=0.0)
    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")
+    if args.telemetry_device_state and args.telemetry_mode != "nvml":
+        parser.error("device-state telemetry requires NVML")
    raise SystemExit(run(args))
@@ -135,10 +135,16 @@ def test_triton_statistics_exclude_warmup_and_reject_counter_reset(pilot):
 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},
-    }]}
+    descriptor = {
+        "outputs": [
+            {
+                "name": "mask",
+                "datatype": "UINT8",
+                "shape": [1, 512, 512],
+                "parameters": {"binary_data_size": mask.nbytes},
+            }
+        ]
+    }
    header = json.dumps(descriptor).encode()

    class Response:
@@ -214,6 +220,61 @@ def test_nvml_sampler_reuses_one_session_without_spawning_processes(pilot):
    assert library.nvmlDeviceGetUtilizationRates.calls == 2


+def test_material_vote_tie_never_allows_by_class_order(pilot):
+    choose = pilot("pilot_graph").select_material
+    # unknown=0, hard_surface=1, bare_soil=2, grass=3
+    votes = np.array(
+        [[0, 0, 0, 2], [0, 1, 0, 1], [0, 2, 0, 1], [0, 0, 0, 0], [1, 1, 0, 0], [0, 0, 2, 2]]
+    )
+    np.testing.assert_array_equal(choose(votes), [3, 0, 1, 0, 0, 0])
+    with pytest.raises(ValueError, match="invalid material"):
+        choose(np.array([[0, -1, 0]]))
+
+
+def test_numeric_diagnostic_counts_transitions_and_score_ties(pilot):
+    module = pilot("diagnose_ddrnet_numeric")
+    reference, candidate = np.array([[3, 3], [1, 0]]), np.array([[1, 3], [0, 0]])
+    assert module.compare_masks(reference, reference)["different_pixels"] == 0
+    assert module.compare_masks(reference, candidate) == {
+        "different_pixels": 2,
+        "transitions": [
+            {"reference": 1, "candidate": 0, "pixels": 1},
+            {"reference": 3, "candidate": 1, "pixels": 1},
+        ],
+    }
+    logits = np.array([[[[100.0]], [[101.0]]]], dtype=np.float32)
+    details = module.score_details(logits, np.ones_like(logits), np.array([[0]]), [(0, 0)])
+    assert details[0]["selected"] == 0 and details[0]["logit_margin"] == 1
+    assert details[0]["max_score_ties"] == [0, 1]
+
+
+def test_optional_nvml_device_state_distinguishes_unavailable_from_zero(pilot):
+    class Function:
+        def __init__(self, code, value):
+            self.code, self.value = code, value
+
+        def __call__(self, *args):
+            args[-1]._obj.value = self.value
+            return self.code
+
+    module = pilot("pilot_telemetry")
+    sampler = object.__new__(module.NvmlSampler)
+    sampler.closed, sampler.handle = False, None
+    sampler.library = SimpleNamespace(
+        nvmlDeviceGetClockInfo=Function(0, 2685),
+        nvmlDeviceGetPerformanceState=Function(0, 0),
+        nvmlDeviceGetPowerUsage=Function(3, 0),
+    )
+    state = sampler.sample_device_state()
+    assert state["sm_clock_mhz"] == 2685 and state["pstate"] == 0
+    assert state["power_mw"] is None and state["device_state_errors"] == {"power_mw": 3}
+    del sampler.library.nvmlDeviceGetClockInfo
+    assert sampler.sample_device_state()["sm_clock_mhz"] is None
+    sampler.closed = True
+    with pytest.raises(RuntimeError, match="closed"):
+        sampler.sample_device_state()
+
+
 def test_body_history_is_bounded_and_never_reads_future(pilot):
    store = pilot("pilot_graph").CurrentStore(None)
    for index in range(100):