feat(perception): evaluate fixed-class detector candidates

tests/test_fixed_class_detector_tournament.py

@@ -0,0 +1,80 @@
+from __future__ import annotations
+
+from copy import deepcopy
+
+import pytest
+
+from k1link.perception.fixed_class_detector_tournament import (
+    EXACT_FRAME_NAMES,
+    WORKER_RUN_SCHEMA,
+    CandidateWorkerRun,
+    FixedClassTournamentError,
+    false_authority,
+)
+
+
+def _worker_document() -> dict[str, object]:
+    frames = []
+    for name in EXACT_FRAME_NAMES:
+        detections = []
+        if name == "frame-000253.png":
+            detections = [
+                {
+                    "class_id": 16,
+                    "label": "dog",
+                    "score": 0.72,
+                    "bbox_xyxy": [100.0, 200.0, 160.0, 280.0],
+                    "valid_fov_fraction": 1.0,
+                }
+            ]
+        frames.append(
+            {
+                "frame_name": name,
+                "source_sha256": "a" * 64,
+                "detections": detections,
+                "timing_ms": {"end_to_end": 12.0},
+            }
+        )
+    return {
+        "schema_version": WORKER_RUN_SCHEMA,
+        "profile_id": "candidate/v0",
+        "provider_id": "shadow-candidate/v0",
+        "upstream_revision": "revision",
+        "checkpoint_sha256": "b" * 64,
+        "completed": True,
+        "execution": {"inference_passes_per_evidence_frame": 1},
+        "frames": frames,
+        "metrics": {"capacity_fps": 80.0},
+        "authority": false_authority(),
+    }
+
+
+def test_candidate_worker_run_reports_risk_only_quality() -> None:
+    result = CandidateWorkerRun.from_document(_worker_document())
+
+    summary = result.quality_summary(threshold=0.5)
+
+    assert summary["detection_count"] == 1
+    assert summary["risk_group_counts"] == {"animal": 1}
+    assert summary["frame_000253_dog_detected"] is True
+    assert summary["frame_000253_dog_max_score"] == 0.72
+
+
+def test_candidate_worker_run_rejects_missing_frame() -> None:
+    document = _worker_document()
+    frames = document["frames"]
+    assert isinstance(frames, list)
+    frames.pop()
+
+    with pytest.raises(FixedClassTournamentError, match="exact M48S slice"):
+        CandidateWorkerRun.from_document(document)
+
+
+def test_candidate_worker_run_rejects_authority() -> None:
+    document = deepcopy(_worker_document())
+    authority = document["authority"]
+    assert isinstance(authority, dict)
+    authority["candidate_accepted"] = True
+
+    with pytest.raises(FixedClassTournamentError, match="false authority"):
+        CandidateWorkerRun.from_document(document)

tests/test_m48s_rf_detr_deployment_result.py

@@ -0,0 +1,60 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+REPOSITORY_ROOT = Path(__file__).resolve().parents[1]
+RESULT_ID = (
+    "m48s-rf-detr-deployment-gate-"
+    "2feb9e1b12a5588951ad35d63bf23cf6bdd579d54b5329d46d7696f88c444547"
+)
+RESULT_ROOT = (
+    REPOSITORY_ROOT
+    / ".runtime/compute-experiments/m48s-semantic-shadow/rf-detr-deployment-results"
+    / RESULT_ID
+)
+
+
+def test_rf_detr_tensorrt_detector_is_ready_only_for_reference_graph_shadow() -> None:
+    manifest = json.loads((RESULT_ROOT / "manifest.json").read_text("utf-8"))
+    decision = manifest["decision"]
+    evidence = manifest["evidence"]
+    load = evidence["source_paced_load"]
+
+    assert manifest["result_id"] == RESULT_ID
+    assert manifest["completed"] is True
+    assert manifest["accepted"] is False
+    assert decision == {
+        "detector_source_paced_load_gate_passed": True,
+        "integrated_world_state_gate_evaluated": False,
+        "next_gate": (
+            "run the RF-DETR shadow provider inside the complete reference graph and "
+            "require world-state p95 <= 175 ms without changing false authority"
+        ),
+        "production_accepted": False,
+        "ready_for_reference_graph_shadow": True,
+        "tensorrt_numeric_parity_passed": True,
+        "tournament_finalist": True,
+    }
+    assert evidence["engine_sha256"] == (
+        "986399ce706b7380472cf5e473232249fed6e628971d8007f6609e83128d46b8"
+    )
+    assert evidence["tensorrt_parity"]["passed"] is True
+    assert evidence["tensorrt_parity"]["frame_000253_dog_present_in_tensorrt"] is True
+    assert evidence["pytorch_quality_at_0_5"]["class_counts"] == evidence[
+        "triton_quality_at_0_5"
+    ]["class_counts"]
+    assert load["execution"]["source_frames_consumed"] == 18_008
+    assert load["execution"]["source_frame_replacements"] == 0
+    assert load["execution"]["effective_consumed_fps"] >= 9.5
+    assert load["detector_completion_age_ms"]["p95"] <= 175.0
+    assert load["gpu"]["gpu_memory_used_mib"]["maximum"] <= 20 * 1024
+    assert load["gpu"]["longest_100_percent_gpu_sample_run"] == 0
+    assert all(load["checks"].values())
+    assert manifest["authority"] == {
+        "actuation_allowed": False,
+        "candidate_accepted": False,
+        "commands_enabled": False,
+        "ground_truth": False,
+        "navigation_or_safety_accepted": False,
+    }

tests/test_m48s_yolox_all_coco_result.py

@@ -0,0 +1,43 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+REPOSITORY_ROOT = Path(__file__).resolve().parents[1]
+RESULT_ID = (
+    "m48s-yolox-all-coco-shadow-"
+    "7dbe6043b3fc12c7ddb162f609f883d86b34a4f2dd3785a632795f257e192d06"
+)
+RESULT_ROOT = (
+    REPOSITORY_ROOT
+    / ".runtime/compute-experiments/m48s-semantic-shadow/yolox-all-coco-results"
+    / RESULT_ID
+)
+
+
+def test_all_coco_yolox_uses_one_inference_pass_with_bounded_postprocess_cost() -> None:
+    manifest = json.loads((RESULT_ROOT / "manifest.json").read_text("utf-8"))
+    metrics = manifest["metrics"]
+
+    assert manifest["result_id"] == RESULT_ID
+    assert manifest["completed"] is True
+    assert manifest["accepted"] is False
+    assert metrics["frames"] == {"completed": 11, "requested": 11}
+    assert metrics["inference_passes_per_frame"] == 1
+    assert metrics["frozen_detection_count"] == 44
+    assert metrics["all_coco_detection_count"] == 45
+    assert metrics["added_detection_count"] == 1
+    assert metrics["all_coco_class_counts"] == {
+        "car": 36,
+        "handbag": 1,
+        "person": 3,
+        "truck": 5,
+    }
+    benchmark = metrics["postprocess_benchmark"]
+    assert benchmark["iterations_per_profile_per_frame"] == 20
+    frozen_mean = benchmark["timing_ms"]["frozen_ms"]["mean"]
+    all_coco_mean = benchmark["timing_ms"]["all_coco_ms"]["mean"]
+    assert all_coco_mean - frozen_mean < 1.0
+    assert metrics["all_coco_core_capacity_fps"] > 30.0
+    assert manifest["authority"]["commands_enabled"] is False
+    assert manifest["authority"]["navigation_or_safety_accepted"] is False

tests/test_rf_detr_detector_provider.py

@@ -0,0 +1,202 @@
+from __future__ import annotations
+
+import json
+import math
+from pathlib import Path
+from typing import cast
+
+import numpy as np
+import pytest
+from numpy.typing import NDArray
+
+from k1link.perception.contracts import (
+    ClockBasis,
+    ModalityOutcome,
+    ModalityStatus,
+    SourceEnvelope,
+    TimestampBundle,
+)
+from k1link.perception.detector import (
+    RF_DETR_SHADOW_MODEL_ID,
+    RF_DETR_SHADOW_PREPROCESS_ID,
+    RF_DETR_SHADOW_PROVIDER_ID,
+    RfDetrShadowDetectorProvider,
+)
+from k1link.perception.providers import SourcePacket
+from k1link.perception.rf_detr_object_detector import (
+    RF_DETR_CONFIG,
+    RF_DETR_ENGINE_SHA256,
+    RF_DETR_FP16_ONNX_SHA256,
+    RfDetrConfig,
+    RfDetrDetectorError,
+    RfDetrRawOutput,
+    TritonRfDetrHttpInferenceBackend,
+    postprocess_rf_detr,
+    preprocess_raw_kb4_rf_detr,
+)
+
+REPOSITORY_ROOT = Path(__file__).resolve().parents[1]
+
+
+def _status() -> ModalityStatus:
+    return ModalityStatus(True, ModalityOutcome.AVAILABLE, "test-available")
+
+
+def _packet(sequence: int, image: object) -> SourcePacket:
+    return SourcePacket(
+        envelope=SourceEnvelope(
+            source_id="RAVNOVES00",
+            session_id="20260720T065719Z_viewer_live",
+            frame_id=f"frame-{sequence:06d}",
+            sequence=sequence,
+            timestamps=TimestampBundle(
+                utc_ns=1_000 + sequence,
+                monotonic_ns=2_000 + sequence,
+                source_ns=3_000 + sequence,
+                clock_basis=ClockBasis.RECORDED_HOST,
+            ),
+            source_age_ns=0,
+            binding_reason="test-recorded-source",
+            calibration_id="camera-1-kb4-test",
+            representation_id="registered-map-increment-v1",
+            image=_status(),
+            registered_point_increment=_status(),
+            pose=_status(),
+        ),
+        image_payload=image,
+        registered_point_increment_payload=("points", sequence),
+        pose_payload=("pose", sequence),
+    )
+
+
+class _Resizer:
+    def __init__(self) -> None:
+        self.source: NDArray[np.uint8] | None = None
+
+    def resize(
+        self, image: NDArray[np.uint8], width: int, height: int
+    ) -> NDArray[np.uint8]:
+        self.source = image.copy()
+        output = np.empty((height, width, 3), dtype=np.uint8)
+        output[:, :, 0] = 255
+        output[:, :, 1] = 0
+        output[:, :, 2] = 127
+        return output
+
+
+class _Backend:
+    def __init__(self, output: RfDetrRawOutput) -> None:
+        self.output = output
+        self.calls = 0
+
+    def infer(self, tensor: NDArray[np.float32]) -> RfDetrRawOutput:
+        assert tensor.shape == (1, 3, 704, 704)
+        assert tensor.dtype == np.float32
+        self.calls += 1
+        return self.output
+
+
+def _output() -> RfDetrRawOutput:
+    boxes = np.zeros((1, 300, 4), dtype=np.float16)
+    logits = np.full((1, 300, 91), -20.0, dtype=np.float16)
+    boxes[0, 0] = (0.5, 0.5, 0.25, np.float16(1 / 3))
+    logits[0, 0, 18] = np.float16(math.log(3.0))  # dog, score 0.75
+    boxes[0, 1] = (0.25, 0.25, 0.1, 0.2)
+    logits[0, 1, 1] = np.float16(math.log(4.0))  # person, score 0.80
+    boxes[0, 2] = (0.75, 0.25, 0.1, 0.2)
+    logits[0, 2, 62] = np.float16(math.log(9.0))  # chair, non-risk
+    logits[0, 3, 12] = np.float16(math.log(9.0))  # unused COCO slot
+    return RfDetrRawOutput(boxes=boxes, logits=logits)
+
+
+def test_preprocess_masks_bgr_converts_rgb_stretches_and_normalizes() -> None:
+    image = np.zeros((600, 800, 3), dtype=np.uint8)
+    image[:, :] = (10, 20, 30)
+    mask = np.ones((600, 800), dtype=np.bool_)
+    mask[0, 0] = False
+    resizer = _Resizer()
+
+    tensor = preprocess_raw_kb4_rf_detr(image, mask, resizer=resizer)
+
+    assert resizer.source is not None
+    assert tuple(resizer.source[1, 1]) == (30, 20, 10)
+    assert tuple(resizer.source[0, 0]) == (114, 114, 114)
+    assert tensor.shape == (1, 3, 704, 704)
+    assert tensor.dtype == np.float32
+    assert tensor[0, 0, 0, 0] == pytest.approx((1.0 - 0.485) / 0.229)
+    assert tensor[0, 1, 0, 0] == pytest.approx((0.0 - 0.456) / 0.224)
+    assert tensor[0, 2, 0, 0] == pytest.approx((127 / 255.0 - 0.406) / 0.225)
+
+
+def test_postprocess_maps_sparse_coco_slots_and_emits_only_risk_classes() -> None:
+    result = postprocess_rf_detr(_output(), np.ones((600, 800), dtype=np.bool_))
+
+    assert tuple(item.label for item in result.detections) == ("person", "dog")
+    assert result.detections[0].score == pytest.approx(0.8, abs=0.001)
+    assert result.detections[1].score == pytest.approx(0.75, abs=0.001)
+    assert result.detections[1].bbox_xyxy == pytest.approx(
+        (300.0, 200.0, 500.0, 400.0), abs=0.03
+    )
+    assert dict(result.rejected) == {"non-risk-class": 1, "unmapped-class-slot": 1}
+
+    with pytest.raises(RfDetrDetectorError, match="tensor types"):
+        postprocess_rf_detr(
+            RfDetrRawOutput(
+                boxes=cast(NDArray[np.float16], _output().boxes.astype(np.float32)),
+                logits=_output().logits,
+            ),
+            np.ones((600, 800), dtype=np.bool_),
+        )
+
+
+def test_shadow_provider_uses_one_pass_and_preserves_semantic_hints() -> None:
+    backend = _Backend(_output())
+    provider = RfDetrShadowDetectorProvider(
+        mask=np.ones((600, 800), dtype=np.bool_),
+        backend=backend,
+        resizer=_Resizer(),
+        clock_ns=iter((10, 30)).__next__,
+    )
+
+    proposals = provider.detect(_packet(7, np.zeros((600, 800, 3), dtype=np.uint8)))
+
+    assert backend.calls == 1
+    assert tuple(item.semantic_hint for item in proposals) == ("person", "dog")
+    assert all(item.provider_id == RF_DETR_SHADOW_PROVIDER_ID for item in proposals)
+    assert all(item.model_id == RF_DETR_SHADOW_MODEL_ID for item in proposals)
+    assert all(item.preprocess_id == RF_DETR_SHADOW_PREPROCESS_ID for item in proposals)
+    assert provider.snapshot().completed_frames == 1
+    assert provider.snapshot().proposal_count == 2
+    assert provider.snapshot().core_duration_ns == 20
+
+
+def test_shadow_profile_is_fixed_and_transport_pins_model_version() -> None:
+    assert RF_DETR_CONFIG.minimum_score == 0.25
+    with pytest.raises(RfDetrDetectorError, match="cannot be tuned"):
+        RfDetrConfig(minimum_score=0.5)
+
+    backend = TritonRfDetrHttpInferenceBackend("http://127.0.0.1:8100")
+    try:
+        assert backend.path == "/v2/models/rf_detr_large/versions/1/infer"
+    finally:
+        backend.close()
+    with pytest.raises(RfDetrDetectorError, match="explicit HTTP origin"):
+        TritonRfDetrHttpInferenceBackend("http://user:secret@127.0.0.1:8100")
+
+
+def test_shadow_profile_pins_worker_engine_and_retains_false_authority() -> None:
+    profile = json.loads(
+        (REPOSITORY_ROOT / "config/perception/rf-detr-large-risk-shadow-v0.json").read_text(
+            "utf-8"
+        )
+    )
+
+    assert profile["model"]["strongly_typed_fp16_onnx_sha256"] == RF_DETR_FP16_ONNX_SHA256
+    assert (
+        profile["model"]["worker_006_rtx4090_tensorrt_11_engine_sha256"]
+        == RF_DETR_ENGINE_SHA256
+    )
+    assert profile["emission"]["single_inference_per_source_frame"] is True
+    assert profile["emission"]["geometry_owns_static_occupancy"] is True
+    assert profile["emission"]["unlisted_semantic_classes_emitted"] is False
+    assert not any(profile["authority"].values())

tests/test_yolox_all_coco_profile.py

@@ -0,0 +1,29 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+from k1link.perception.detector import ALL_COCO_YOLOX_PROVIDER_ID
+from k1link.perception.yolox_object_detector import (
+    ALL_COCO_CLASS_IDS,
+    ALL_COCO_YOLOX_CONFIG,
+    COCO_CLASSES,
+    YOLOX_MODEL_SHA256,
+)
+
+REPOSITORY_ROOT = Path(__file__).resolve().parents[1]
+PROFILE_PATH = REPOSITORY_ROOT / "config/perception/yolox-s-all-coco-shadow-v2.json"
+
+
+def test_all_coco_profile_matches_executable_provider_contract() -> None:
+    profile = json.loads(PROFILE_PATH.read_text("utf-8"))
+
+    assert profile["schema_version"] == "missioncore.yolox-detector-profile/v2"
+    assert profile["provider_id"] == ALL_COCO_YOLOX_PROVIDER_ID
+    assert profile["model"]["model_sha256"] == YOLOX_MODEL_SHA256
+    assert profile["model"]["additional_inference_passes"] == 0
+    assert tuple(profile["postprocess"]["target_class_ids"]) == ALL_COCO_CLASS_IDS
+    assert ALL_COCO_YOLOX_CONFIG.target_class_ids == tuple(range(len(COCO_CLASSES)))
+    assert "dog" in profile["class_policy"]["risk_groups"]["animal"]
+    assert profile["authority"]["commands_enabled"] is False
+    assert profile["authority"]["navigation_or_safety_accepted"] is False

tests/test_yolox_detector_provider.py

@@ -17,13 +17,19 @@ from k1link.perception.contracts import (
     TimestampBundle,
 )
 from k1link.perception.detector import (
+    ALL_COCO_YOLOX_PROVIDER_ID,
     FROZEN_YOLOX_PROVIDER_ID,
+    AllCocoYoloxDetectorProvider,
     DetectorProviderError,
     FrozenYoloxDetectorProvider,
     proposals_from_detections,
 )
 from k1link.perception.providers import SourcePacket
 from k1link.perception.yolox_object_detector import (
+    ALL_COCO_CLASS_IDS,
+    ALL_COCO_YOLOX_CONFIG,
+    COCO_CLASSES,
+    AllCocoYoloxConfig,
     FrozenYoloxConfig,
     TritonHttpInferenceBackend,
     YoloxDetection,
@@ -102,6 +108,14 @@ def _one_person_output() -> NDArray[np.float32]:
     return output
 
 
+def _one_dog_output() -> NDArray[np.float32]:
+    output = np.zeros((1, 8400, 85), dtype=np.float32)
+    output[0, 0, :4] = [40.0, 30.0, math.log(10.0), math.log(10.0)]
+    output[0, 0, 4] = 0.9
+    output[0, 0, 5 + 16] = 0.9
+    return output
+
+
 def test_frozen_preprocess_and_postprocess_match_the_e46j_contract() -> None:
     image = np.full((600, 800, 3), 7, dtype=np.uint8)
     mask = np.ones((600, 800), dtype=np.bool_)
@@ -174,6 +188,39 @@ def test_frozen_profile_rejects_in_place_threshold_tuning() -> None:
         FrozenYoloxConfig(minimum_score=0.51)
 
 
+def test_all_coco_profile_emits_dog_without_another_inference_pass() -> None:
+    mask = np.ones((600, 800), dtype=np.bool_)
+
+    assert postprocess_yolox(_one_dog_output(), mask).detections == ()
+    all_coco = postprocess_yolox(
+        _one_dog_output(),
+        mask,
+        config=ALL_COCO_YOLOX_CONFIG,
+    )
+    backend = _Backend(_one_dog_output())
+    provider = AllCocoYoloxDetectorProvider(
+        mask=mask,
+        backend=backend,
+        resizer=_Resizer(),
+    )
+    proposals = provider.detect(
+        _packet(16, np.zeros((600, 800, 3), dtype=np.uint8))
+    )
+
+    assert tuple(range(80)) == ALL_COCO_CLASS_IDS
+    assert len(COCO_CLASSES) == 80
+    assert tuple(item.label for item in all_coco.detections) == ("dog",)
+    assert backend.calls == 1
+    assert len(proposals) == 1
+    assert proposals[0].provider_id == ALL_COCO_YOLOX_PROVIDER_ID
+    assert proposals[0].semantic_hint == "dog"
+
+
+def test_all_coco_profile_is_versioned_and_cannot_be_tuned_in_place() -> None:
+    with pytest.raises(YoloxDetectorError, match="all-COCO.*cannot be tuned"):
+        AllCocoYoloxConfig(target_class_ids=(0, 16))
+
+
 def test_triton_transport_pins_the_frozen_model_version() -> None:
     backend = TritonHttpInferenceBackend("http://127.0.0.1:8000")
     try: