feat(k1): stabilize LAB bridge and isolate onboard device integration

tests/test_xgrids_camera_gateway.py

@@ -299,6 +299,99 @@ def test_camera_selection_is_exclusive_and_hides_device_transport(
         gateway.close()
 
 
+@pytest.mark.parametrize("entrypoint", ["acquisition", "selected-preview"])
+def test_camera_records_in_configured_evidence_outside_source_checkout(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, entrypoint: str,
+) -> None:
+    checkout = tmp_path / "checkout"
+    checkout.mkdir()
+    evidence = tmp_path / "durable" / "sessions"
+    session = evidence / "synthetic-camera-session"
+    session.mkdir(parents=True)
+    monkeypatch.setenv("MISSIONCORE_FFMPEG_BINARY", str(_fake_ffmpeg(tmp_path)))
+    gateway = XgridsK1CameraGateway(checkout, XGRIDS_K1_PLUGIN_ID, evidence_root=evidence)
+    try:
+        if entrypoint == "acquisition":
+            gateway.activate_recording_producer(
+                "sensor.camera.right", "192.168.1.20", session,
+                pre_prepare_fence=lambda reserve: reserve(),
+                commit_fence=lambda commit: commit(),
+            )
+        else:
+            gateway.select("sensor.camera.right", "192.168.1.20")
+            gateway.start_recording(session)
+        _wait_until(lambda: gateway.snapshot()["recording"]["media_ready"] is True)
+        assert gateway.snapshot()["recording"]["session"] == session.name
+        assert not list(checkout.rglob("*.m4s"))
+    finally:
+        gateway.close()
+    assert any(session.rglob("*.m4s"))
+
+
+@pytest.mark.parametrize("entrypoint", ["acquisition", "selected-preview"])
+@pytest.mark.parametrize("escape", ["sibling", "symlink"])
+def test_camera_rejects_paths_outside_configured_evidence_before_admission(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, entrypoint: str, escape: str,
+) -> None:
+    checkout = tmp_path / "checkout"
+    checkout.mkdir()
+    evidence = tmp_path / "evidence"
+    evidence.mkdir()
+    outside = checkout / "not-evidence"
+    outside.mkdir()
+    session = outside
+    if escape == "symlink":
+        session = evidence / "escaped-session"
+        session.symlink_to(outside, target_is_directory=True)
+    gateway = XgridsK1CameraGateway(checkout, XGRIDS_K1_PLUGIN_ID, evidence_root=evidence)
+
+    def forbidden(*_args: object, **_kwargs: object) -> bool:
+        raise AssertionError("An escaped path must not reach authority or FFmpeg")
+
+    monkeypatch.setattr(camera_module.subprocess, "Popen", forbidden)
+    try:
+        with pytest.raises(ValueError, match="configured evidence root"):
+            if entrypoint == "acquisition":
+                gateway.activate_recording_producer(
+                    "sensor.camera.right", "192.168.1.20", session,
+                    pre_prepare_fence=forbidden, commit_fence=forbidden,
+                )
+            else:
+                gateway.start_recording(session)
+        assert gateway.snapshot()["recording"]["active"] is False
+    finally:
+        gateway.close()
+
+
+def test_service_camera_uses_the_same_external_evidence_root_as_acquisition(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    checkout = tmp_path / "checkout"
+    checkout.mkdir()
+    evidence = tmp_path / "configured-evidence"
+    session = evidence / "synthetic-session"
+    session.mkdir(parents=True)
+    monkeypatch.setenv("MISSIONCORE_EVIDENCE_DIR", str(evidence))
+    service = XgridsK1CompatibilityService(checkout)
+    authority_entered = False
+
+    def deny_authority(_reserve: Callable[[], bool]) -> bool:
+        nonlocal authority_entered
+        authority_entered = True
+        return False
+
+    try:
+        with pytest.raises(ValueError, match="authority"):
+            service.camera_preview.activate_recording_producer(
+                "sensor.camera.right", "192.168.1.20", session,
+                pre_prepare_fence=deny_authority, commit_fence=lambda _commit: False,
+            )
+        assert authority_entered is True
+        assert service.camera_preview.snapshot()["active_source_id"] is None
+    finally:
+        service.close()
+
+
 def test_camera_derived_observer_runs_only_after_durable_archive_commit(
     tmp_path: Path,
     monkeypatch: pytest.MonkeyPatch,