fix(viewer): gate recorded perception readiness

tests/test_e10_integrated_perception.py

@@ -3,10 +3,16 @@ from __future__ import annotations
 import importlib.util
 import sys
 from pathlib import Path
+from types import SimpleNamespace
 
 import numpy as np
+import pytest
 
-from k1link.compute.integrated_perception import _CuboidPresentationState
+import k1link.compute.integrated_perception as integrated_module
+from k1link.compute.integrated_perception import (
+    IntegratedPerceptionOverlayStore,
+    _CuboidPresentationState,
+)
 
 
 def _worker_modules() -> tuple[object, object]:
@@ -35,6 +41,60 @@ def _worker_modules() -> tuple[object, object]:
         sys.path.pop(0)
 
 
+def test_integrated_overlay_catalog_validation_is_memoized_until_publication(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    jobs_root = tmp_path / "jobs"
+    results_root = tmp_path / "results"
+    lidar_packs_root = tmp_path / "lidar-packs"
+    cache_root = tmp_path / "cache"
+    for root in (jobs_root, results_root, lidar_packs_root):
+        root.mkdir()
+    job_root = jobs_root / "job-1"
+    job_root.mkdir()
+    first_result = results_root / f"e10-integrated-perception-{'a' * 64}"
+    first_result.mkdir()
+
+    validation_calls = {"job": 0, "result": 0}
+    job = SimpleNamespace(session_id="session-1")
+
+    def validate_job(_root: Path) -> object:
+        validation_calls["job"] += 1
+        return job
+
+    def validate_result(_job_root: Path, candidate: Path, _packs: Path) -> object:
+        validation_calls["result"] += 1
+        return SimpleNamespace(
+            accepted=True,
+            publication_scope="recorded-integrated-realtime-qualification-only",
+            created_at_utc=candidate.name,
+            result_id=candidate.name,
+        )
+
+    monkeypatch.setattr(integrated_module, "validate_camera_compute_job", validate_job)
+    monkeypatch.setattr(
+        integrated_module,
+        "validate_integrated_perception_result",
+        validate_result,
+    )
+    store = IntegratedPerceptionOverlayStore(
+        jobs_root=jobs_root,
+        results_root=results_root,
+        lidar_packs_root=lidar_packs_root,
+        cache_root=cache_root,
+        ffmpeg_path=tmp_path / "ffmpeg",
+    )
+
+    assert store._latest("session-1") is not None
+    assert store._latest("session-1") is not None
+    assert validation_calls == {"job": 1, "result": 1}
+
+    (results_root / f"e10-integrated-perception-{'b' * 64}").mkdir()
+    assert store._latest("session-1") is not None
+    assert validation_calls == {"job": 2, "result": 3}
+
+
 def test_e10_profile_pins_integrated_realtime_budget() -> None:
     _fusion, runner = _worker_modules()
     profile_path = (