fix(k1): restore low-latency live recovery path

tests/test_viewer_diagnostics_api.py

@@ -87,6 +87,12 @@ def test_private_scanner_diagnostics_are_durable_structured_and_bounded(
             "camera_queue_bytes": 12_000_000,
             "camera_queue_segments": 96,
             "camera_retry_count": 3,
+            "camera_pcl_admission_ms": 184,
+            "camera_authority_wait_ms": 91,
+            "camera_ffmpeg_prepare_ms": 42,
+            "camera_post_spawn_commit_ms": 7,
+            "camera_activation_total_ms": 140,
+            "device_command_sent": False,
             "websocket_close_code": 4_008,
             "transport_epoch": 7,
             "unapproved_secret_field": "must-not-be-written",
@@ -108,6 +114,12 @@ def test_private_scanner_diagnostics_are_durable_structured_and_bounded(
     assert document["mqtt_loop_result_code"] == 7
     assert document["mqtt_loop_phase"] == "post-publish-drain"
     assert document["automatic_retry"] is False
+    assert document["camera_pcl_admission_ms"] == 184
+    assert document["camera_authority_wait_ms"] == 91
+    assert document["camera_ffmpeg_prepare_ms"] == 42
+    assert document["camera_post_spawn_commit_ms"] == 7
+    assert document["camera_activation_total_ms"] == 140
+    assert document["device_command_sent"] is False
     assert document["side_effect_status"] == "unknown"
     assert document["network_change_attempted"] is True
     assert document["device_write_attempted"] is True

tests/test_xgrids_acquisition_lifecycle.py

@@ -1047,6 +1047,33 @@ def service_with_fake_runtime(
     return service, runtime
 
 
+def test_passive_state_reuses_one_final_physical_snapshot(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    service, _runtime = service_with_fake_runtime(tmp_path)
+    coordinator = service._physical_command_coordinator  # noqa: SLF001
+    original_snapshot = coordinator.snapshot
+    snapshot_calls = 0
+
+    def counted_snapshot() -> dict[str, object]:
+        nonlocal snapshot_calls
+        snapshot_calls += 1
+        return original_snapshot()
+
+    monkeypatch.setattr(coordinator, "snapshot", counted_snapshot)
+
+    state = service.state()
+
+    # Two earlier reads are deliberate pre/post settlement proofs. Presentation
+    # policy and the nested control-session projection must share the one final
+    # verified durable snapshot instead of recursively reloading the ledger.
+    assert snapshot_calls == 3
+    assert state["application_control_session"]["physical_command"] == state[
+        "physical_command"
+    ]
+
+
 def test_service_owns_one_wifi_association_probe_across_monitor_recreation(
     monkeypatch: pytest.MonkeyPatch,
     tmp_path: Path,
@@ -10283,9 +10310,11 @@ def test_classified_scanning_cancels_stop_eof_and_preserves_or_rearms_camera_epo
         target: str,
         session_dir: Path,
         *,
+        pre_prepare_fence: Callable[[Callable[[], bool]], bool],
         commit_fence: Callable[[Callable[[], bool]], bool],
         committed_before_start: Callable[[dict[str, object]], None] | None = None,
     ) -> dict[str, object]:
+        assert pre_prepare_fence(lambda: True) is True
         select_camera(source_id, target)
         start_recording(session_dir)
         assert committed_before_start is not None
@@ -13763,9 +13792,11 @@ def test_confirmed_scanning_waits_for_first_authoritative_pcl_before_camera(
         target: str,
         session_dir: Path,
         *,
+        pre_prepare_fence: Callable[[Callable[[], bool]], bool],
         commit_fence: Callable[[Callable[[], bool]], bool],
         committed_before_start: Callable[[dict[str, object]], None] | None = None,
     ) -> dict[str, object]:
+        assert pre_prepare_fence(lambda: True) is True
         events.append(("select", (source_id, target)))
         camera_state["active_source_id"] = source_id
         events.append(("record", session_dir))
@@ -13849,6 +13880,27 @@ def test_confirmed_scanning_waits_for_first_authoritative_pcl_before_camera(
         "snapshot",
         snapshot_physical,
     )
+    full_control_snapshot_calls = 0
+    control_only_snapshot_calls = 0
+    real_control_snapshot = control.snapshot
+
+    def snapshot_control() -> dict[str, object]:
+        nonlocal full_control_snapshot_calls
+        full_control_snapshot_calls += 1
+        return real_control_snapshot()
+
+    def snapshot_control_only() -> dict[str, object]:
+        nonlocal control_only_snapshot_calls
+        control_only_snapshot_calls += 1
+        return real_control_snapshot()
+
+    monkeypatch.setattr(control, "snapshot", snapshot_control)
+    monkeypatch.setattr(
+        control,
+        "snapshot_control_only",
+        snapshot_control_only,
+        raising=False,
+    )
     runtime.pcl_frames = 1
     frame = DecodedPointCloudView(
         context=ConsumerFrameContext(
@@ -13863,6 +13915,8 @@ def test_confirmed_scanning_waits_for_first_authoritative_pcl_before_camera(
         positions_xyz=((0.0, 0.0, 0.0),),
     )
 
+    full_snapshots_before_pcl = full_control_snapshot_calls
+    control_only_snapshots_before_pcl = control_only_snapshot_calls
     service._observe_published_runtime_envelope(  # noqa: SLF001
         frame,
         runtime.producer_generation,
@@ -13871,6 +13925,8 @@ def test_confirmed_scanning_waits_for_first_authoritative_pcl_before_camera(
     # Public manual controls stay closed until that worker proves an exact
     # recording epoch, preventing a left/right selection race.
     assert activation_entered.wait(timeout=2.0)
+    assert full_control_snapshot_calls == full_snapshots_before_pcl
+    assert control_only_snapshot_calls == control_only_snapshots_before_pcl + 1
     activating_state = service.state()
     activating_camera_streams = [
         stream
@@ -13890,6 +13946,8 @@ def test_confirmed_scanning_waits_for_first_authoritative_pcl_before_camera(
             )
         )
     assert events == []
+    full_snapshots_before_camera_worker = full_control_snapshot_calls
+    control_only_snapshots_before_camera_worker = control_only_snapshot_calls
     release_activation.set()
     deadline = time.monotonic() + 2.0
     while len(events) < 2 and time.monotonic() < deadline:
@@ -13899,6 +13957,8 @@ def test_confirmed_scanning_waits_for_first_authoritative_pcl_before_camera(
         ("select", ("sensor.camera.right", "192.168.1.20")),
         ("record", out_dir),
     ]
+    assert full_control_snapshot_calls == full_snapshots_before_camera_worker
+    assert control_only_snapshot_calls == control_only_snapshots_before_camera_worker + 2
     assert camera_state["active_source_id"] == "sensor.camera.right"
     assert camera_state["recording"] == {
         "active": True,
@@ -14068,9 +14128,11 @@ def test_failed_post_pcl_camera_activation_retries_on_later_authoritative_frame(
         target: str,
         session_dir: Path,
         *,
+        pre_prepare_fence: Callable[[Callable[[], bool]], bool],
         commit_fence: Callable[[Callable[[], bool]], bool],
         committed_before_start: Callable[[dict[str, object]], None] | None = None,
     ) -> dict[str, object]:
+        assert pre_prepare_fence(lambda: True) is True
         del commit_fence, committed_before_start
         select_calls.append((source_id, target))
         recording_calls.append(session_dir)
@@ -14827,10 +14889,12 @@ def test_stop_priority_overtakes_blocked_initial_camera_popen(
         _target: str,
         _session_dir: Path,
         *,
+        pre_prepare_fence: Callable[[Callable[[], bool]], bool],
         commit_fence: Callable[[Callable[[], bool]], bool],
         committed_before_start: Callable[[dict[str, object]], None] | None = None,
     ) -> dict[str, object]:
         del committed_before_start
+        assert pre_prepare_fence(lambda: True) is True
         candidate_blocked.set()
         assert release_candidate.wait(3.0)
         if not commit_fence(lambda: candidate_committed.append(True) or True):

tests/test_xgrids_application_session.py

@@ -193,6 +193,7 @@ class FakePhysicalCommandCoordinator:
         self.resolutions: list[tuple[str, str | None]] = []
         self.phase_probe: Callable[[], str] | None = None
         self.snapshot_override: dict[str, object] | None = None
+        self.snapshot_calls = 0
 
     def bind_control_session(self, binding: PhysicalCommandRuntimeBinding) -> None:
         self.bindings.append(binding)
@@ -217,6 +218,7 @@ class FakePhysicalCommandCoordinator:
         self.resolutions.append((f"{action}-not-dispatched", phase))
 
     def snapshot(self) -> dict[str, object]:
+        self.snapshot_calls += 1
         if self.snapshot_override is not None:
             return self.snapshot_override
         return {
@@ -433,6 +435,36 @@ def _successful_start_checkpoint_observer(
     return None
 
 
+def test_control_only_snapshot_reuses_preverified_physical_command() -> None:
+    coordinator = FakePhysicalCommandCoordinator()
+    session = InteractiveApplicationControlSession(
+        FakeAuthorityLoader(),
+        transport_factory=FakeTransport,  # type: ignore[arg-type]
+        physical_command_coordinator=coordinator,  # type: ignore[arg-type]
+    )
+
+    control_only = session.snapshot_control_only()
+    assert control_only["state"] == "idle"
+    assert control_only["physical_command"] is None
+    assert coordinator.snapshot_calls == 0
+
+    verified_physical = {
+        "status": "resolved",
+        "requires_reconciliation": False,
+    }
+    joined = session.snapshot_with_physical_command(verified_physical)
+    assert joined["physical_command"] == verified_physical
+    assert coordinator.snapshot_calls == 0
+
+    regular = session.snapshot()
+    assert regular["physical_command"] == {
+        "status": "test",
+        "requires_reconciliation": False,
+        "automatic_replay_allowed": False,
+    }
+    assert coordinator.snapshot_calls == 1
+
+
 def test_canonical_stages_require_operator_events_but_device_standby_does_not(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:

tests/test_xgrids_camera_gateway.py

@@ -6,6 +6,7 @@ import json
 import sys
 import threading
 import time
+from collections.abc import Callable
 from io import BytesIO
 from pathlib import Path
 
@@ -1416,16 +1417,60 @@ def test_source_end_none_epoch_cas_waits_for_canonical_archive_seal(
         gateway.close()
 
 
+def test_acquisition_candidate_never_spawns_before_authority_reservation(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setenv("MISSIONCORE_FFMPEG_BINARY", str(_silent_ffmpeg(tmp_path)))
+    session = tmp_path / "sessions" / "camera-pre-popen-denied"
+    session.mkdir(parents=True)
+    gateway = XgridsK1CameraGateway(tmp_path, XGRIDS_K1_PLUGIN_ID)
+    popen_called = False
+
+    def forbidden_popen(*_: object, **__: object) -> object:
+        nonlocal popen_called
+        popen_called = True
+        raise AssertionError("FFmpeg must not start before authority reservation")
+
+    monkeypatch.setattr(camera_module.subprocess, "Popen", forbidden_popen)
+    try:
+        with pytest.raises(ValueError, match="authority"):
+            gateway.activate_recording_producer(
+                "sensor.camera.right",
+                "192.168.1.20",
+                session,
+                pre_prepare_fence=lambda _reserve: False,
+                commit_fence=lambda commit: commit(),
+            )
+
+        snapshot = gateway.snapshot()
+        assert popen_called is False
+        assert snapshot["active_source_id"] is None
+        assert snapshot["recording"]["active"] is False
+    finally:
+        gateway.close()
+
+
 def test_acquisition_candidate_binds_epoch_before_any_reader_thread_starts(
     tmp_path: Path,
     monkeypatch: pytest.MonkeyPatch,
+    caplog: pytest.LogCaptureFixture,
 ) -> None:
     monkeypatch.setenv("MISSIONCORE_FFMPEG_BINARY", str(_silent_ffmpeg(tmp_path)))
     session = tmp_path / "sessions" / "camera-bind-before-reader"
     session.mkdir(parents=True)
     gateway = XgridsK1CameraGateway(tmp_path, XGRIDS_K1_PLUGIN_ID)
+    caplog.set_level("INFO", logger="k1link.device_plugins.xgrids_k1.camera")
     bound_generation: list[int] = []
     reader_start_generation: list[int] = []
+    authority_reserved = False
+
+    def reserve_authority(reserve: Callable[[], bool]) -> bool:
+        nonlocal authority_reserved
+        assert authority_reserved is False
+        assert reserve() is True
+        authority_reserved = True
+        return True
 
     def bind(committed: dict[str, object]) -> None:
         recording = committed["recording"]
@@ -1436,6 +1481,7 @@ def test_acquisition_candidate_binds_epoch_before_any_reader_thread_starts(
 
     def start_threads(producer: object) -> None:
         generation = producer.generation  # type: ignore[attr-defined]
+        assert authority_reserved is True
         assert bound_generation == [generation]
         reader_start_generation.append(generation)
 
@@ -1445,6 +1491,7 @@ def test_acquisition_candidate_binds_epoch_before_any_reader_thread_starts(
             "sensor.camera.right",
             "192.168.1.20",
             session,
+            pre_prepare_fence=reserve_authority,
             commit_fence=lambda commit: commit(),
             committed_before_start=bind,
         )
@@ -1452,6 +1499,16 @@ def test_acquisition_candidate_binds_epoch_before_any_reader_thread_starts(
         assert bound_generation == [activated["recording"]["active_epoch"]]
         assert reader_start_generation == bound_generation
         assert activated["recording"]["media_ready"] is False
+        timing = [
+            record
+            for record in caplog.records
+            if getattr(record, "event_code", None) == "k1_camera_activation_timing"
+        ]
+        assert len(timing) == 1
+        assert timing[0].camera_generation == bound_generation[0]
+        assert timing[0].camera_authority_wait_ms >= 0
+        assert timing[0].camera_ffmpeg_prepare_ms >= 0
+        assert timing[0].camera_post_spawn_commit_ms >= 0
     finally:
         gateway.close()
 

tests/test_xgrids_physical_command_ledger.py

@@ -771,6 +771,71 @@ def test_immediate_archived_reconciliation_successor_fails_closed_on_archive_fau
         )
 
 
+def test_passive_snapshot_reuses_only_unchanged_verified_filesystem_generation(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    ledger = _ledger(tmp_path, monkeypatch)
+    _prepare_start(ledger)
+    real_reload = ledger._reload_locked
+    reload_count = 0
+
+    def counted_reload() -> None:
+        nonlocal reload_count
+        reload_count += 1
+        real_reload()
+
+    monkeypatch.setattr(ledger, "_reload_locked", counted_reload)
+
+    first = ledger.snapshot()
+    second = ledger.snapshot()
+    third = ledger.snapshot()
+
+    assert first == second == third
+    assert reload_count == 1
+
+
+def test_passive_snapshot_observes_atomic_write_from_second_ledger_instance(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    first = _ledger(tmp_path, monkeypatch)
+    _prepare_start(first)
+    assert first.snapshot().record is not None
+    second = PhysicalCommandLedger(tmp_path / "repository", clock=_clock)
+
+    second.mark_dispatching(START_OPERATION)
+
+    observed = first.snapshot()
+    assert observed.record is not None
+    assert observed.record.stage == "dispatching"
+    assert observed.record.revision == 2
+
+
+def test_passive_snapshot_ctime_fence_detects_same_size_archive_tamper(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    ledger = _ledger(tmp_path, monkeypatch)
+    _predecessor, _current, archive_path = _install_resolved_start_rebind_rollover(
+        ledger,
+        monkeypatch,
+    )
+    assert ledger.snapshot().status == "resolved"
+    original_metadata = archive_path.stat()
+    tampered = bytearray(archive_path.read_bytes())
+    tampered[-1] = ord("{") if tampered[-1] != ord("{") else ord("}")
+    archive_path.write_bytes(tampered)
+    os.utime(
+        archive_path,
+        ns=(original_metadata.st_atime_ns, original_metadata.st_mtime_ns),
+    )
+
+    assert archive_path.stat().st_size == original_metadata.st_size
+    assert archive_path.stat().st_mtime_ns == original_metadata.st_mtime_ns
+    assert ledger.snapshot().status == "corrupt"
+
+
 def test_classified_stop_ancestry_rejects_missing_confirmation_cycle_and_conflict(
     tmp_path: Path,
     monkeypatch: pytest.MonkeyPatch,