fix(k1): stabilize live recovery and media admission

tests/test_canonical_pipeline.py

@@ -326,6 +326,7 @@ def test_live_rerun_process_and_close_failure_preserve_capture_then_recover(
     allow_second_bridge = threading.Event()
     second_bridge_ready = threading.Event()
     recovery_confirmed = threading.Event()
+    raw_capture_completed = threading.Event()
     raw_sequences: list[int] = []
     should_stop_before_explicit_stop: list[bool] = []
     confirmed_attempts: list[int] = []
@@ -389,6 +390,7 @@ def test_live_rerun_process_and_close_failure_preserve_capture_then_recover(
         assert second_bridge_ready.wait(timeout=2.0)
         raw_sequences.append(10)
         enqueue(_captured_live_point_cloud(10))  # type: ignore[operator]
+        raw_capture_completed.set()
         assert recovery_confirmed.wait(timeout=2.0)
 
         while not should_stop():  # type: ignore[operator]
@@ -409,6 +411,7 @@ def test_live_rerun_process_and_close_failure_preserve_capture_then_recover(
     )
 
     assert recovery_confirmed.wait(timeout=3.0)
+    assert raw_capture_completed.wait(timeout=3.0)
     snapshot = runtime.snapshot()
     assert snapshot["phase"] == "live"
     assert snapshot["source_ready"] is True
@@ -510,6 +513,66 @@ def test_live_latest_point_cloud_survives_pose_and_perception_pressure(
     runtime.close()
 
 
+def test_live_pose_is_published_during_continuous_point_cloud_pressure(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    producer_started = threading.Event()
+    producer_finished = threading.Event()
+    pose_published_during_pressure = threading.Event()
+
+    class SlowPointCloudBridge(MetricRuntimeBridgeStub):
+        def process(self, envelope: object) -> None:
+            if isinstance(envelope, DecodedPointCloudView):
+                # Keep another PCL waiting in the latest-wins slot. The
+                # scheduler must still admit pose while pressure continues.
+                time.sleep(0.01)
+            elif (
+                isinstance(envelope, DecodedPoseView)
+                and not producer_finished.is_set()
+            ):
+                pose_published_during_pressure.set()
+            super().process(envelope)
+
+    def fake_capture_mqtt(
+        _host: str,
+        _out_dir: Path,
+        **callbacks: object,
+    ) -> dict[str, object]:
+        callbacks["on_clock_established"]()  # type: ignore[operator]
+        callbacks["on_ready"]()  # type: ignore[operator]
+        enqueue = callbacks["on_message_recorded"]
+        producer_started.set()
+        for index in range(300):
+            enqueue(_captured_live_point_cloud(index * 2 + 1))  # type: ignore[operator]
+            enqueue(_captured_live_pose(index * 2 + 2))  # type: ignore[operator]
+            time.sleep(0.001)
+        producer_finished.set()
+
+        should_stop = callbacks["should_stop"]
+        while not should_stop():  # type: ignore[operator]
+            time.sleep(0.005)
+        return {"message_count": 600}
+
+    monkeypatch.setattr(runtime_module, "capture_mqtt", fake_capture_mqtt)
+    runtime = VisualizationRuntime(
+        bridge_factory=lambda **kwargs: SlowPointCloudBridge(kwargs["metrics"]),  # type: ignore[arg-type]
+        normalizer=normalize_k1_message,
+    )
+    runtime.start_live(
+        "192.168.1.50",
+        tmp_path / "pose-fairness",
+        duration_seconds=None,
+        project_name="POSEFAIRNESS001",
+    )
+
+    assert producer_started.wait(timeout=2.0)
+    assert producer_finished.wait(timeout=2.0)
+    runtime.stop(wait_seconds=2.0)
+    runtime.close()
+    assert pose_published_during_pressure.is_set()
+
+
 @pytest.mark.parametrize("attempt", [5, 1025, 10**100])
 def test_live_rerun_recovery_backoff_saturates(attempt: int) -> None:
     assert runtime_module._rerun_recovery_backoff_seconds(attempt) == 5.0  # noqa: SLF001

tests/test_connection_supervisor.py

@@ -1817,7 +1817,7 @@ def test_same_path_positive_reducer_observation_resets_technical_failure_streak(
     asyncio.run(scenario())
 
 
-def test_unrelated_supervisor_revisions_do_not_reset_technical_failure_streak() -> None:
+def test_unrelated_supervisor_revisions_preserve_exact_configured_route() -> None:
     async def scenario() -> None:
         supervisor, epoch = _configured_unverified_supervisor()
         path = _available_path()
@@ -1836,8 +1836,12 @@ def test_unrelated_supervisor_revisions_do_not_reset_technical_failure_streak()
         async def host_probe(_target: EndpointTarget) -> HostPathProbeResult:
             return technical_timeout
 
+        tcp_calls = 0
+
         async def tcp_probe(_target: EndpointTarget) -> bool:
-            raise AssertionError("technical failure must skip TCP")
+            nonlocal tcp_calls
+            tcp_calls += 1
+            return True
 
         monitor = ReadOnlyConnectionMonitor(
             supervisor,
@@ -1847,8 +1851,10 @@ def test_unrelated_supervisor_revisions_do_not_reset_technical_failure_streak()
         )
 
         first = await monitor.poll_once()
-        assert first.revision == initial.revision
-        assert first.host_path is initial.host_path
+        assert first.host_path.available is True
+        assert first.host_path.epoch == initial.host_path.epoch
+        assert first.endpoint.tcp_state == "reachable"
+        assert first.authority.control_allowed is False
         assert supervisor.observe_endpoint(
             target=TARGET,
             intent_id="bridge-1",
@@ -1857,11 +1863,14 @@ def test_unrelated_supervisor_revisions_do_not_reset_technical_failure_streak()
         )
         endpoint_refresh = supervisor.snapshot()
         assert endpoint_refresh.revision > first.revision
-        assert endpoint_refresh.host_path is initial.host_path
+        assert endpoint_refresh.host_path.epoch == initial.host_path.epoch
+        assert endpoint_refresh.host_path.fingerprint == initial.host_path.fingerprint
 
         second = await monitor.poll_once()
-        assert second.revision == endpoint_refresh.revision
-        assert second.host_path is initial.host_path
+        assert second.host_path.available is True
+        assert second.host_path.epoch == initial.host_path.epoch
+        assert second.endpoint.tcp_state == "reachable"
+        assert second.authority.control_allowed is False
         assert supervisor.observe_endpoint(
             target=TARGET,
             intent_id="bridge-1",
@@ -1870,15 +1879,17 @@ def test_unrelated_supervisor_revisions_do_not_reset_technical_failure_streak()
         )
 
         confirmed = await monitor.poll_once()
-        assert confirmed.host_path.available is False
-        assert confirmed.host_path.reason_code == "host-wifi-operation-timeout"
+        assert confirmed.host_path.available is True
+        assert confirmed.host_path.epoch == initial.host_path.epoch
+        assert confirmed.endpoint.tcp_state == "reachable"
         assert confirmed.authority.control_allowed is False
+        assert tcp_calls == 3
         await monitor.close()
 
     asyncio.run(scenario())
 
 
-def test_inflight_third_timeout_cannot_overwrite_a_concurrent_positive_observation() -> None:
+def test_inflight_timeout_cannot_overwrite_a_concurrent_positive_observation() -> None:
     async def scenario() -> None:
         supervisor, initial_epoch = _configured_unverified_supervisor()
         path = _available_path()
@@ -1892,20 +1903,20 @@ def test_inflight_third_timeout_cannot_overwrite_a_concurrent_positive_observati
             observation_failure_class="association-observer",
             kernel_route_fingerprint=path.kernel_route_fingerprint,
         )
-        third_probe_entered = asyncio.Event()
-        release_third_probe = asyncio.Event()
+        timeout_probe_entered = asyncio.Event()
+        release_timeout_probe = asyncio.Event()
         host_calls = 0
 
         async def host_probe(_target: EndpointTarget) -> HostPathProbeResult:
             nonlocal host_calls
             host_calls += 1
-            if host_calls == 3:
-                third_probe_entered.set()
-                await release_third_probe.wait()
+            if host_calls == 1:
+                timeout_probe_entered.set()
+                await release_timeout_probe.wait()
             return technical_timeout
 
         async def tcp_probe(_target: EndpointTarget) -> bool:
-            raise AssertionError("technical failure must skip TCP")
+            return True
 
         monitor = ReadOnlyConnectionMonitor(
             supervisor,
@@ -1913,32 +1924,20 @@ def test_inflight_third_timeout_cannot_overwrite_a_concurrent_positive_observati
             tcp_probe=tcp_probe,
             target_provider=lambda: TARGET,
         )
-        first = await monitor.poll_once()
-        second = await monitor.poll_once()
-        assert first.authority.control_allowed is False
-        assert second.authority.control_allowed is False
-
-        inflight_third = asyncio.create_task(monitor.poll_once())
-        await third_probe_entered.wait()
+        inflight_timeout = asyncio.create_task(monitor.poll_once())
+        await timeout_probe_entered.wait()
         refreshed_epoch = supervisor.observe_host_path(path)
         assert refreshed_epoch == initial_epoch
         external_positive = supervisor.snapshot()
-        release_third_probe.set()
-        raced_timeout = await inflight_third
+        release_timeout_probe.set()
+        raced_timeout = await inflight_timeout
 
-        assert raced_timeout.revision == external_positive.revision
-        assert raced_timeout.host_path is external_positive.host_path
+        assert raced_timeout.revision >= external_positive.revision
+        assert raced_timeout.host_path.available is True
+        assert raced_timeout.host_path.epoch == external_positive.host_path.epoch
+        assert raced_timeout.host_path.fingerprint == external_positive.host_path.fingerprint
+        assert raced_timeout.endpoint.tcp_state == "reachable"
         assert raced_timeout.authority.control_allowed is False
-
-        second_after_positive = await monitor.poll_once()
-        assert second_after_positive.revision == external_positive.revision
-        assert second_after_positive.host_path is external_positive.host_path
-        assert second_after_positive.authority.control_allowed is False
-
-        confirmed = await monitor.poll_once()
-        assert confirmed.host_path.available is False
-        assert confirmed.host_path.reason_code == "host-wifi-operation-timeout"
-        assert confirmed.authority.control_allowed is False
         await monitor.close()
 
     asyncio.run(scenario())

tests/test_plugin_runtime.py

@@ -37,6 +37,7 @@ from k1link.device_plugins.xgrids_k1.facade import (
     ACTION_PHYSICAL_COMMAND_RECONCILE,
     ACTION_PHYSICAL_COMMAND_REOPEN_RETIRED_RECONCILIATION,
     ACTION_PHYSICAL_COMMAND_RETIRE_UNAVAILABLE,
+    ACTION_STATE_READ,
     ACTION_STREAM_START_LIVE,
     ACTION_STREAM_START_REPLAY,
     ACTION_STREAM_STOP,
@@ -1285,6 +1286,57 @@ def test_sync_runtime_actions_run_outside_the_api_event_loop() -> None:
     assert service.thread_id != event_loop_thread
 
 
+def test_state_reads_are_single_flight_and_survive_one_cancelled_waiter() -> None:
+    class BlockingStateService(FakeXgridsService):
+        def __init__(self) -> None:
+            super().__init__()
+            self.state_calls = 0
+            self.started = threading.Event()
+            self.release = threading.Event()
+
+        def state(self) -> dict[str, Any]:
+            self.state_calls += 1
+            self.started.set()
+            if not self.release.wait(timeout=2):
+                raise AssertionError("state read was not released by the test")
+            return {"phase": "idle", "revision": self.state_calls}
+
+    service = BlockingStateService()
+    adapter = XgridsK1PluginFacade(service)
+
+    def invocation(invocation_id: str) -> RuntimeActionInvocation:
+        return RuntimeActionInvocation(
+            invocation_id=invocation_id,
+            plugin_id=adapter.plugin_id,
+            action_id=ACTION_STATE_READ,
+            requested_at=datetime.now(UTC),
+            parameters={},
+        )
+
+    async def exercise() -> None:
+        first = asyncio.create_task(adapter.invoke(invocation("state-read-first")))
+        while not service.started.is_set():
+            await asyncio.sleep(0)
+        second = asyncio.create_task(adapter.invoke(invocation("state-read-second")))
+        await asyncio.sleep(0.02)
+        assert service.state_calls == 1
+
+        first.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await first
+        assert service.state_calls == 1
+
+        service.release.set()
+        assert await second == {"phase": "idle", "revision": 1}
+        assert await adapter.invoke(invocation("state-read-fresh")) == {
+            "phase": "idle",
+            "revision": 2,
+        }
+
+    asyncio.run(exercise())
+    assert service.state_calls == 2
+
+
 def test_runtime_requires_successful_handshake_before_dispatch() -> None:
     adapter = XgridsK1PluginFacade(FakeXgridsService())
     runtime = _in_process_runtime(adapter, activate=False)

tests/test_session_catalog_lifecycle.py

@@ -200,6 +200,33 @@ def test_startup_scan_baselines_historical_sessions_without_enqueuing(
         manager.close()
 
 
+def test_archive_revision_tracks_session_lifecycle_without_capture_churn(
+    tmp_path: Path,
+) -> None:
+    sessions = tmp_path / "sessions"
+    sessions.mkdir()
+    initial = app_module.observation_archive_revision((sessions,))
+
+    session = sessions / "20260716T205632Z_viewer_live"
+    session.mkdir()
+    started = app_module.observation_archive_revision((sessions,))
+    assert started != initial
+
+    capture = session / "captures" / "mqtt_live"
+    capture.mkdir(parents=True)
+    (capture / "mqtt.raw.k1mqtt").write_bytes(b"growing-capture")
+    assert app_module.observation_archive_revision((sessions,)) == started
+
+    marker = sessions / ".current_session"
+    marker.write_text(f"{session.name}\n", encoding="utf-8")
+    active = app_module.observation_archive_revision((sessions,))
+    assert active != started
+
+    marker.unlink()
+    finalized = app_module.observation_archive_revision((sessions,))
+    assert finalized != active
+
+
 def test_reconciliation_skips_one_stale_session_and_prepares_later_valid_session(
     tmp_path: Path,
     monkeypatch: pytest.MonkeyPatch,

tests/test_web_validation_security.py

@@ -40,6 +40,24 @@ class _FailingSendWebSocket:
         del code, reason
 
 
+class _DisconnectAfterSecondSendWebSocket:
+    def __init__(self) -> None:
+        self.accepted = False
+        self.send_count = 0
+
+    async def accept(self) -> None:
+        self.accepted = True
+
+    async def send_json(self, payload: dict[str, Any]) -> None:
+        del payload
+        self.send_count += 1
+        if self.send_count == 2:
+            raise WebSocketDisconnect(code=1001)
+
+    async def close(self, *, code: int, reason: str) -> None:
+        del code, reason
+
+
 async def _post_json(path: str, payload: dict[str, Any]) -> tuple[int, str]:
     body = json.dumps(payload).encode()
     request_sent = False
@@ -198,6 +216,25 @@ def test_device_plugin_events_treats_proven_transport_disconnect_as_completion(
     assert websocket.accepted is True
 
 
+def test_device_plugin_events_keeps_heavy_state_poll_off_live_media_cadence(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    websocket = _DisconnectAfterSecondSendWebSocket()
+    sleeps: list[float] = []
+
+    async def record_sleep(delay: float) -> None:
+        sleeps.append(delay)
+
+    monkeypatch.setattr(app_module, "plugin_dispatcher", _StateDispatcher())
+    monkeypatch.setattr(app_module.asyncio, "sleep", record_sleep)
+
+    asyncio.run(app_module.device_plugin_events(websocket, "test.plugin"))
+
+    assert websocket.accepted is True
+    assert sleeps == [app_module.DEVICE_PLUGIN_EVENT_POLL_INTERVAL_SECONDS]
+    assert sleeps[0] == 2.0
+
+
 def test_device_plugin_events_propagates_arbitrary_send_runtime_error(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:

tests/test_xgrids_acquisition_lifecycle.py

@@ -5662,7 +5662,7 @@ def test_read_only_verify_scan_is_rejected_after_atomic_fence_before_admission(
     original_retire = service._retire_application_control_for_network_change  # noqa: SLF001
     original_apply = service._apply_read_only_device_topology  # noqa: SLF001
 
-    def retire_with_competing_scan() -> None:
+    def retire_with_competing_scan(*, allow_terminal_failure: bool = False) -> None:
         assert service._provisioning_active is True  # noqa: SLF001
         scan_outcomes.append(
             _attempt_competing_scan_from_sync_boundary(
@@ -5670,7 +5670,7 @@ def test_read_only_verify_scan_is_rejected_after_atomic_fence_before_admission(
                 operation_id="op-00000000-0000-4000-8000-000000000102",
             )
         )
-        original_retire()
+        original_retire(allow_terminal_failure=allow_terminal_failure)
 
     def apply_with_fence_assertion(**kwargs: Any) -> str:
         nonlocal admission_calls
@@ -7692,6 +7692,58 @@ def test_terminal_prepared_start_without_publish_settles_and_stops_local_capture
     assert service._acquisition_session_lease is None  # noqa: SLF001
 
 
+def test_live_prepared_start_reduction_does_not_touch_publish_fence(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    service, _ = service_with_fake_runtime(tmp_path)
+    acquisition_id = "acq-live-prepared"
+    operation_id = "op-live-prepared"
+    monkeypatch.setattr(
+        service._physical_command_coordinator,  # noqa: SLF001
+        "snapshot",
+        lambda: {
+            "status": "unresolved",
+            "record": {
+                "operation_id": operation_id,
+                "acquisition_id": acquisition_id,
+                "action": "start",
+                "stage": "prepared",
+                "resolution": None,
+                "publish_call_returned": None,
+                "packet_id": None,
+                "qos2_completed": False,
+                "application_response": None,
+            },
+        },
+    )
+    monkeypatch.setattr(
+        service._application_control_session,  # noqa: SLF001
+        "snapshot",
+        lambda: {"state": "initializing", "failure": None},
+    )
+
+    class PublishFenceMustStayFree:
+        def acquire(self, *args: object, **kwargs: object) -> bool:
+            raise AssertionError("live START reduction touched the publish fence")
+
+        def release(self) -> None:
+            raise AssertionError("live START reduction released an unowned fence")
+
+    monkeypatch.setattr(
+        service,
+        "_k1_command_dispatch_gate",
+        PublishFenceMustStayFree(),
+    )
+
+    outcome = service._settle_prepared_start_worker_failure(  # noqa: SLF001
+        acquisition_id=acquisition_id,
+        start_operation_id=operation_id,
+    )
+
+    assert outcome == "not-applicable"
+
+
 def _activate_real_checkpoint_for_prepared_stop_fixture(
     service: XgridsK1CompatibilityService,
     *,
@@ -12273,6 +12325,7 @@ def _stop_response_without_terminal_status_fixture(
     tmp_path: Path,
     *,
     qos2_completed: bool = True,
+    application_response: bool = True,
 ) -> SimpleNamespace:
     clock_value = [datetime(2026, 8, 10, 8, 10, tzinfo=UTC)]
     service, runtime = service_with_fake_runtime(tmp_path)
@@ -12424,20 +12477,21 @@ def _stop_response_without_terminal_status_fixture(
             packet_id=82,
         )
         ledger.mark_qos2_completed(stop_operation_id, packet_id=82)
-    ledger.record_application_response(
-        stop_operation_id,
-        PhysicalCommandApplicationResponse(
-            operation_id=stop_operation_id,
-            action="stop",
-            control_session_id=physical_connection.control_session_id,
-            host_path_epoch=physical_connection.host_path_epoch,
-            producer_generation=physical_connection.producer_generation,
-            result_code=PHYSICAL_COMMAND_APPLICATION_SUCCESS_CODE,
-            success=True,
-            payload_sha256="a" * 64,
-            observed_at_utc="2026-08-10T08:10:02.000Z",
-        ),
-    )
+    if application_response:
+        ledger.record_application_response(
+            stop_operation_id,
+            PhysicalCommandApplicationResponse(
+                operation_id=stop_operation_id,
+                action="stop",
+                control_session_id=physical_connection.control_session_id,
+                host_path_epoch=physical_connection.host_path_epoch,
+                producer_generation=physical_connection.producer_generation,
+                result_code=PHYSICAL_COMMAND_APPLICATION_SUCCESS_CODE,
+                success=True,
+                payload_sha256="a" * 64,
+                observed_at_utc="2026-08-10T08:10:02.000Z",
+            ),
+        )
     return SimpleNamespace(
         service=service,
         runtime=runtime,
@@ -12450,6 +12504,60 @@ def _stop_response_without_terminal_status_fixture(
     )
 
 
+def test_unknown_stop_retires_only_terminal_control_and_keeps_live_capture(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    fixture = _stop_response_without_terminal_status_fixture(
+        tmp_path,
+        qos2_completed=False,
+        application_response=False,
+    )
+    service = fixture.service
+    control = fixture.control
+    close_calls = 0
+    worker_retired = False
+    original_snapshot = control.snapshot
+    original_close = control.close
+
+    def terminal_snapshot() -> dict[str, object]:
+        snapshot = original_snapshot()
+        if control.state == "failed" and not worker_retired:
+            snapshot["can_open"] = False
+        return snapshot
+
+    def retire_failed_worker() -> None:
+        nonlocal close_calls, worker_retired
+        close_calls += 1
+        worker_retired = True
+        original_close()
+
+    monkeypatch.setattr(control, "snapshot", terminal_snapshot)
+    monkeypatch.setattr(control, "close", retire_failed_worker)
+    control.state = "failed"
+    control.outcome_unknown = True
+    control.failure = {
+        "code": "ApplicationCommandOutcomeUnknown",
+        "reason_code": "mqtt_response_timeout",
+        "stop_command_attempted": True,
+        "stop_publish_attempts": 1,
+        "safe_to_retry": False,
+    }
+    service._acquire_application_control_process_lease()  # noqa: SLF001
+
+    recovered = service.state()
+
+    assert close_calls == 1
+    assert recovered["application_control_session"]["state"] == "idle"
+    assert recovered["acquisition"]["state"] == "awaiting_external_stop"
+    assert recovered["source_mode"] == "live"
+    assert recovered["physical_command"]["status"] == "unresolved"
+    assert recovered["physical_command"]["requires_reconciliation"] is True
+    assert fixture.runtime.stop_calls == 0
+    assert control.stop_calls == 1
+    assert service._application_control_process_lease_holders == set()  # noqa: SLF001
+
+
 @pytest.mark.parametrize("qos2_completed", [True, False])
 def test_stop_success_without_terminal_status_times_out_into_local_only_recovery(
     tmp_path: Path,
@@ -13729,7 +13837,18 @@ def test_confirmed_scanning_waits_for_first_authoritative_pcl_before_camera(
         binding=binding,
         resolved=True,
     )
-    monkeypatch.setattr(service._physical_command_coordinator, "snapshot", lambda: physical)  # noqa: SLF001
+    physical_snapshot_calls = 0
+
+    def snapshot_physical() -> dict[str, object]:
+        nonlocal physical_snapshot_calls
+        physical_snapshot_calls += 1
+        return physical
+
+    monkeypatch.setattr(
+        service._physical_command_coordinator,  # noqa: SLF001
+        "snapshot",
+        snapshot_physical,
+    )
     runtime.pcl_frames = 1
     frame = DecodedPointCloudView(
         context=ConsumerFrameContext(
@@ -13818,9 +13937,11 @@ def test_confirmed_scanning_waits_for_first_authoritative_pcl_before_camera(
     )
 
     # Every later authoritative PCL for the same lineage is idempotent.
+    physical_snapshot_calls_before = physical_snapshot_calls
     service._observe_published_runtime_envelope(frame, runtime.producer_generation)  # noqa: SLF001
     time.sleep(0.05)
     assert len(events) == 2
+    assert physical_snapshot_calls == physical_snapshot_calls_before
 
 
 def test_stale_post_publish_pcl_cannot_activate_camera(
@@ -15316,6 +15437,85 @@ def test_active_stream_recovery_scanning_resumes_same_physical_lineage_without_c
     )
 
 
+def test_active_stream_recovery_privacy_bridge_retains_exact_route_for_read_only_rebind(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    service, runtime = service_with_fake_runtime(tmp_path)
+    control, physical = _install_composite_active_recovery_fixture(
+        service,
+        runtime,
+        monkeypatch,
+    )
+    service.state()
+    lineage = service._active_stream_recovery_lineage  # noqa: SLF001
+    assert lineage is not None
+    supervisor = service._connection_supervisor  # noqa: SLF001
+    baseline = supervisor.snapshot()
+    candidate = replace(
+        baseline,
+        device_identity=replace(baseline.device_identity, state="unverified"),
+        control_plane=replace(
+            baseline.control_plane,
+            state="lost",
+            session_id=None,
+        ),
+        lease=replace(baseline.lease, state="configured-unverified"),
+        authority=replace(
+            baseline.authority,
+            control_allowed=False,
+            acquisition_start_allowed=False,
+        ),
+    )
+    start_projects_before = list(control.start_projects)
+    stop_calls_before = control.stop_calls
+    operation_journal_before = service._operations.snapshot()  # noqa: SLF001
+    physical_ledger_before = service._physical_command_ledger.snapshot()  # noqa: SLF001
+    network_ledger_before = service._network_mutation_ledger.snapshot()  # noqa: SLF001
+    monkeypatch.setattr(facade_module, "_inspect_host_path", _direct_host_path)
+    monkeypatch.setattr(
+        service._host_wifi_association_probe,  # noqa: SLF001
+        "observe",
+        lambda *_args, **_kwargs: {
+            "schema_version": 1,
+            "adapter": "CoreWLAN",
+            "wifi_interface": True,
+            "association_state": "unavailable",
+            "evidence_quality": "unavailable",
+            "continuity_proven": False,
+            "continuity_token": "d" * 64,
+            "reason_code": "association-identity-unavailable",
+        },
+    )
+    monkeypatch.setattr(
+        supervisor,
+        "association_timeout_retention_candidate",
+        lambda *, expected_target: (
+            candidate
+            if expected_target
+            == EndpointTarget(lineage.target_ipv4, lineage.target_port)
+            else None
+        ),
+    )
+
+    sampled = service._sample_host_path(lineage.target_ipv4)  # noqa: SLF001
+
+    assert sampled.available is True
+    assert sampled.reason_code is None
+    assert sampled.fingerprint == baseline.host_path.fingerprint
+    assert sampled.kernel_route_fingerprint == (
+        baseline.host_path.kernel_route_fingerprint
+    )
+    assert control.start_projects == start_projects_before
+    assert control.stop_calls == stop_calls_before
+    assert service._operations.snapshot() == operation_journal_before  # noqa: SLF001
+    assert (  # noqa: SLF001
+        service._physical_command_ledger.snapshot() == physical_ledger_before
+    )
+    assert service._network_mutation_ledger.snapshot() == network_ledger_before  # noqa: SLF001
+    assert service._physical_command_coordinator.snapshot() is physical  # noqa: SLF001
+
+
 def test_active_stream_control_adoption_gets_fresh_budget_after_slow_proof(
     tmp_path: Path,
     monkeypatch: pytest.MonkeyPatch,
@@ -15638,7 +15838,7 @@ def test_active_stream_recovery_wrong_identity_blocks_without_retry_or_camera_re
     monkeypatch.setattr(
         service,
         "_probe_control_endpoint",
-        lambda _target: facade_module._CorrelatedEndpointObservation(  # noqa: SLF001
+        lambda _target, **_kwargs: facade_module._CorrelatedEndpointObservation(  # noqa: SLF001
             path=_direct_host_path(lineage.target_ipv4),
             reachable=True,
             reason_code=None,
@@ -15699,8 +15899,14 @@ def test_active_stream_recovery_owned_path_is_read_only_and_resumes_exact_lineag
         events.append(("monitor", True))
         return True
 
-    def probe_exact_target(target_ipv4: str) -> facade_module._CorrelatedEndpointObservation:  # noqa: SLF001
-        events.append(("probe", target_ipv4))
+    def probe_exact_target(  # noqa: SLF001
+        target_ipv4: str,
+        *,
+        association_timeout_seconds: float,
+    ) -> facade_module._CorrelatedEndpointObservation:
+        events.append(
+            ("probe", (target_ipv4, association_timeout_seconds))
+        )
         return facade_module._CorrelatedEndpointObservation(  # noqa: SLF001
             path=_direct_host_path(target_ipv4),
             reachable=True,
@@ -15770,7 +15976,13 @@ def test_active_stream_recovery_owned_path_is_read_only_and_resumes_exact_lineag
     assert events == [
         ("monitor", True),
         ("lease-acquire", "network"),
-        ("probe", lineage.target_ipv4),
+        (
+            "probe",
+            (
+                lineage.target_ipv4,
+                facade_module.ACTIVE_STREAM_RECOVERY_ASSOCIATION_TIMEOUT_SECONDS,
+            ),
+        ),
         (
             "device-info-status",
             (
@@ -15910,6 +16122,7 @@ def test_incident_recovery_retires_epoch_one_control_and_retries_fresh_inspectio
 
     def probe_current_epoch(
         target_ipv4: str,
+        **_kwargs: object,
     ) -> facade_module._CorrelatedEndpointObservation:  # noqa: SLF001
         path = next(probe_paths)
         service._observe_connection_transport(  # noqa: SLF001
@@ -16065,7 +16278,7 @@ def test_active_stream_recovery_waits_for_terminal_control_worker_retirement(
     monkeypatch.setattr(
         service,
         "_probe_control_endpoint",
-        lambda _target: facade_module._CorrelatedEndpointObservation(  # noqa: SLF001
+        lambda _target, **_kwargs: facade_module._CorrelatedEndpointObservation(  # noqa: SLF001
             path=_direct_host_path(lineage.target_ipv4),
             reachable=True,
             reason_code=None,
@@ -16140,7 +16353,7 @@ def test_active_stream_recovery_exact_device_system_fault_is_terminal_without_co
     monkeypatch.setattr(
         service,
         "_probe_control_endpoint",
-        lambda _target: facade_module._CorrelatedEndpointObservation(  # noqa: SLF001
+        lambda _target, **_kwargs: facade_module._CorrelatedEndpointObservation(  # noqa: SLF001
             path=_direct_host_path(lineage.target_ipv4),
             reachable=True,
             reason_code=None,
@@ -16223,7 +16436,7 @@ def test_active_stream_recovery_bootstrap_system_error_is_normalized_terminal_fa
     monkeypatch.setattr(
         service,
         "_probe_control_endpoint",
-        lambda _target: facade_module._CorrelatedEndpointObservation(  # noqa: SLF001
+        lambda _target, **_kwargs: facade_module._CorrelatedEndpointObservation(  # noqa: SLF001
             path=_direct_host_path(lineage.target_ipv4),
             reachable=True,
             reason_code=None,
@@ -16476,6 +16689,111 @@ def test_active_stream_recovery_never_retries_invalid_fmp4_after_media_commit(
     assert control.stop_calls == 0
 
 
+def test_bound_scan_over_without_stop_dominates_late_empty_camera_epoch_failure(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    service, runtime = service_with_fake_runtime(tmp_path)
+    control, physical = _install_composite_active_recovery_fixture(
+        service,
+        runtime,
+        monkeypatch,
+        runtime_phase="live",
+        camera_phase="streaming",
+    )
+    camera = service.camera_preview.snapshot()
+    recording = camera["recording"]
+    assert isinstance(recording, dict)
+    recording.update(
+        {
+            "active": True,
+            "active_epoch": 2,
+            "committed_media_segment_count": 4_833,
+            "producer_alive": False,
+            "completed_epochs": 2,
+            "last_summary": {
+                "codec_epoch": 2,
+                "status": "failed",
+                "media_segment_count": 0,
+                "failure_code": "invalid-fmp4",
+            },
+        }
+    )
+    camera.update(
+        {
+            "phase": "error",
+            "error": {
+                "code": "invalid-fmp4",
+                "message": "synthetic empty post-power-loss epoch",
+            },
+        }
+    )
+    physical.update(
+        {
+            "runtime_bound": True,
+            "reconciliation_ready": True,
+            "observed_session_state": "scan_over",
+        }
+    )
+    control.state = "failed"
+    control.state_revision += 1
+    control.failure = {
+        "reason_code": "application_acceptance_failed",
+        "failed_phase": "scanning",
+        "modeling_command_attempted": True,
+        "stop_command_attempted": False,
+        "diagnostic_snapshot_unavailable": [],
+        "diagnostic_evidence_unavailable": [],
+        "safe_to_retry": False,
+    }
+
+    def scan_over_control_snapshot() -> dict[str, object]:
+        snapshot = FakeInteractiveControlSession.snapshot(control)
+        snapshot["transport"] = {
+            "state": "failed",
+            "publish_attempts": 7,
+            "device_status_reports": 9,
+            "latest_device_session_state": "scan_over",
+            "latest_device_project_bound": True,
+            "latest_device_init_ready": False,
+            "latest_system_error_code": None,
+            "automatic_retry": False,
+            "automatic_reconnect": False,
+        }
+        return snapshot
+
+    monkeypatch.setattr(control, "snapshot", scan_over_control_snapshot)
+    monkeypatch.setattr(service, "_seal_acquisition_capture_clock", lambda: None)
+    camera_stop_calls: list[str] = []
+    monkeypatch.setattr(
+        service.camera_preview,
+        "stop_current",
+        lambda: camera_stop_calls.append("stop") or {},
+    )
+    start_projects_before = list(control.start_projects)
+    stop_calls_before = control.stop_calls
+
+    terminal = service.state()
+
+    assert terminal["acquisition"]["state"] == "interrupted"
+    assert terminal["acquisition"]["message_code"] == (
+        "acquisition.recovery.device_standby_observed"
+    )
+    assert terminal["acquisition"]["result"] == {
+        "receiver_stopped": True,
+        "device_state": "scan_over",
+        "device_stop": "not-sent",
+        "automatic_command_retry": False,
+        "read_only_recovery": True,
+        "physical_reconciliation_required": True,
+    }
+    assert terminal["acquisition"]["cleanup_pending"] is False
+    assert camera_stop_calls == ["stop"]
+    assert runtime.stop_calls == 1
+    assert control.start_projects == start_projects_before
+    assert control.stop_calls == stop_calls_before == 0
+
+
 def test_active_stream_recovery_reopens_exact_camera_epoch_once_and_fences_stale_lineage(
     tmp_path: Path,
     monkeypatch: pytest.MonkeyPatch,
@@ -24955,6 +25273,59 @@ def test_configured_unverified_monitor_keeps_same_route_without_promoting_author
         assert "verify-control-device-info" in retained.allowed_actions
 
 
+def test_monitor_layer_retains_exact_configured_route_across_association_lock_timeout(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    service, _ = service_with_fake_runtime(tmp_path)
+    binding = _seed_supervised_connection(service, with_control=False)
+    baseline = service._connection_supervisor.snapshot()  # noqa: SLF001
+    assert baseline.host_path.kernel_route_fingerprint is not None
+    tcp_calls: list[str] = []
+
+    monkeypatch.setattr(
+        service,
+        "_sample_host_path",
+        lambda target, **_kwargs: HostPathProbeResult(
+            available=False,
+            fingerprint=None,
+            interface=baseline.host_path.interface,
+            source_ipv4=baseline.host_path.source_ipv4,
+            route_class="unavailable",
+            reason_code="host-wifi-operation-timeout",
+            observation_failure_class="association-observer",
+            kernel_route_fingerprint=baseline.host_path.kernel_route_fingerprint,
+        ),
+    )
+    monkeypatch.setattr(
+        facade_module,
+        "_probe_control_endpoint_socket",
+        lambda target: (
+            tcp_calls.append(target)
+            or facade_module.TcpReachabilityProbeResult(reachable=True)
+        ),
+    )
+
+    async def poll_three_times() -> list[facade_module.ConnectionSupervisorSnapshot]:
+        return [
+            await service._connection_monitor.poll_once()  # noqa: SLF001
+            for _ in range(3)
+        ]
+
+    snapshots = asyncio.run(poll_three_times())
+
+    assert tcp_calls == [binding.target_ipv4] * 3
+    for retained in snapshots:
+        assert retained.host_path.available is True
+        assert retained.host_path.epoch == baseline.host_path.epoch
+        assert retained.host_path.fingerprint == baseline.host_path.fingerprint
+        assert retained.endpoint.tcp_state == "reachable"
+        assert retained.device_identity.state == "unverified"
+        assert retained.authority.control_allowed is False
+        assert retained.authority.acquisition_start_allowed is False
+        assert "verify-control-device-info" in retained.allowed_actions
+
+
 def test_association_timeout_cannot_hide_a_real_kernel_route_change(
     monkeypatch: pytest.MonkeyPatch,
     tmp_path: Path,
@@ -29825,6 +30196,91 @@ def test_explicit_verify_reconciles_persisted_start_to_ready_without_device_io(
     assert service._application_control_process_lease_holders == {"control"}  # noqa: SLF001
 
 
+def test_long_first_verify_refreshes_aged_transport_before_exposing_prestart_control(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    """A long checkpoint reconciliation cannot orphan its new ready socket."""
+
+    service, _ = service_with_fake_runtime(tmp_path)
+    coordinator = _VerifyPhysicalRecoveryCoordinator(
+        observed_session_state="ready",
+        reconciliation_ready=True,
+    )
+    _install_synthetic_verify_recovery(service, coordinator)
+    supervisor = service._connection_supervisor  # noqa: SLF001
+    monotonic_now = [100.0]
+    suspend_aware_now = [1_000.0]
+    supervisor._monotonic_clock = lambda: monotonic_now[0]  # noqa: SLF001
+    supervisor._suspend_aware_clock = lambda: suspend_aware_now[0]  # noqa: SLF001
+    before = supervisor.snapshot()
+    assert before.intent is not None
+    assert before.device_network.target is not None
+    host_epoch = supervisor.observe_host_path(
+        _association_bound_direct_host_path(before.device_network.target.ipv4)
+    )
+    assert supervisor.observe_endpoint(
+        target=before.device_network.target,
+        intent_id=before.intent.intent_id,
+        host_path_epoch=host_epoch,
+        reachable=True,
+    )
+
+    monkeypatch.setattr(facade_module, "_inspect_host_path", _direct_host_path)
+    tcp_probes: list[str] = []
+
+    def reachable(target: str) -> bool:
+        tcp_probes.append(target)
+        return True
+
+    monkeypatch.setattr(facade_module, "_control_endpoint_reachable", reachable)
+    reconcile = service._reconcile_physical_command_after_verify_owned  # noqa: SLF001
+
+    async def reconcile_after_transport_ages(
+        bound_service: XgridsK1CompatibilityService,
+        *,
+        verify_operation_id: str,
+        allow_receiver_rehydrate: bool = True,
+    ) -> dict[str, Any]:
+        result = await reconcile(
+            verify_operation_id=verify_operation_id,
+            allow_receiver_rehydrate=allow_receiver_rehydrate,
+        )
+        monotonic_now[0] += 20.0
+        suspend_aware_now[0] += 20.0
+        return result
+
+    service._reconcile_physical_command_after_verify_owned = MethodType(  # type: ignore[method-assign]  # noqa: SLF001
+        reconcile_after_transport_ages,
+        service,
+    )
+    operation_id = "op-00000000-0000-4000-8000-000000001202"
+
+    verified = asyncio.run(
+        service.verify_connection(
+            _retained_physical_recovery_verify_request(operation_id=operation_id)
+        )
+    )
+
+    assert tcp_probes == ["192.168.1.20"]
+    assert verified["last_operation"]["status"] == "succeeded"
+    assert verified["active_connection_mode"] == "bridge"
+    assert verified["connection_policy"]["actions"]["start-acquisition"]["allowed"] is True
+    assert service._application_control_session.snapshot()["state"] == "connection-ready"  # noqa: SLF001
+    assert service._application_control_process_lease_holders == {"control"}  # noqa: SLF001
+    fresh = supervisor.snapshot()
+    assert fresh.intent is not None
+    assert fresh.device_network.target is not None
+    assert supervisor.endpoint_observation_has_remaining_lease(
+        target=fresh.device_network.target,
+        intent_id=fresh.intent.intent_id,
+        host_path_epoch=fresh.host_path.epoch,
+        minimum_remaining_seconds=(
+            facade_module.VERIFY_POST_TRANSITION_ENDPOINT_MIN_REMAINING_SECONDS
+        ),
+    )
+
+
 def test_explicit_verify_reconciliation_does_not_require_normal_command_authority(
     tmp_path: Path,
 ) -> None:

tests/test_xgrids_active_acquisition_checkpoint_integration.py

@@ -1843,6 +1843,25 @@ def test_chained_stop_facade_cessation_uses_real_ledger_ancestry(
         stale_reconciliation,
         final_record,
     )
+    # Reproduce the append-only field shape: an otherwise valid historical
+    # standby reconciliation is inherited by the newer ordinary STOP.  The
+    # checkpoint reducer must use the current STOP's own READY status, not the
+    # last historical reconciliation merely because it is terminal standby.
+    snapshot_with_stale_standby = replace(
+        ledger.snapshot(),
+        record=replace(
+            final_record,
+            reconciliations=(
+                *final_record.reconciliations,
+                stale_reconciliation,
+            ),
+        ),
+    )
+    monkeypatch.setattr(
+        ledger,
+        "snapshot",
+        lambda: snapshot_with_stale_standby,
+    )
 
     service._application_control_session.snapshot = lambda: {  # type: ignore[method-assign]
         "verified_control": _verified_control(first_rebind)
@@ -1862,6 +1881,88 @@ def test_chained_stop_facade_cessation_uses_real_ledger_ancestry(
     assert checkpoint.cessation_physical_proof.ancestor_chain == ancestry
 
 
+def test_restart_ceases_active_checkpoint_after_composite_ordinary_stop(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    service = _service(tmp_path, monkeypatch)
+    connection = _connection(
+        control_session_id="control-before-ordinary-stop-restart",
+        host_path_epoch=1,
+        producer_generation=1,
+    )
+    _prepare_and_activate_checkpoint(service, connection)
+    ledger = service._physical_command_ledger  # noqa: SLF001
+    start = ledger.snapshot().record
+    assert start is not None and start.operation_id == START_OPERATION_ID
+    identity = PhysicalCommandIdentity(
+        vendor_device_id_sha256=VENDOR_SHA256,
+        device_serial_sha256=SERIAL_SHA256,
+    )
+    stop_operation_id = "physical-stop-ordinary-restart"
+    ledger.prepare(
+        operation_id=stop_operation_id,
+        parent_operation_id=START_OPERATION_ID,
+        acquisition_id=ACQUISITION_ID,
+        action="stop",
+        identity=identity,
+        connection=connection,
+        compatibility_profile_id=XGRIDS_K1_COMPATIBILITY_PROFILE_ID,
+        payload_sha256="6" * 64,
+        baseline_status=_status(
+            connection,
+            "scanning",
+            observed_at_utc="2026-08-13T12:02:00.000Z",
+        ),
+    )
+    ledger.mark_dispatching(stop_operation_id)
+    ledger.mark_observing(
+        stop_operation_id,
+        publish_call_returned=True,
+        packet_id=43,
+    )
+    ledger.mark_qos2_completed(stop_operation_id, packet_id=43)
+    ledger.record_application_response(
+        stop_operation_id,
+        PhysicalCommandApplicationResponse(
+            operation_id=stop_operation_id,
+            action="stop",
+            control_session_id=connection.control_session_id,
+            host_path_epoch=connection.host_path_epoch,
+            producer_generation=connection.producer_generation,
+            result_code=PHYSICAL_COMMAND_APPLICATION_SUCCESS_CODE,
+            success=True,
+            payload_sha256="7" * 64,
+            observed_at_utc="2026-08-13T12:02:01.000Z",
+        ),
+    )
+    ledger.record_status_observation(
+        stop_operation_id,
+        _status(
+            connection,
+            "ready",
+            observed_at_utc="2026-08-13T12:02:02.000Z",
+        ),
+    )
+    stop = ledger.resolve(
+        stop_operation_id,
+        resolution="stop-standby-observed",
+    )
+    before_restart = ActiveAcquisitionRecoveryCheckpointStore(tmp_path).snapshot()
+    assert before_restart.status == "active"
+
+    restarted = XgridsK1CompatibilityService(tmp_path)
+
+    after_restart = ActiveAcquisitionRecoveryCheckpointStore(tmp_path).snapshot()
+    assert after_restart.status == "ceased"
+    assert after_restart.checkpoint is not None
+    assert after_restart.checkpoint.cessation_physical_proof is not None
+    assert after_restart.checkpoint.cessation_physical_proof.operation_id == (
+        stop.operation_id
+    )
+    assert restarted._active_acquisition_checkpoint_trust == "trusted"  # noqa: SLF001
+
+
 def test_restart_ready_classification_ceases_active_checkpoint_after_undispatched_stop(
     tmp_path: Path,
     monkeypatch: pytest.MonkeyPatch,

tests/test_xgrids_active_acquisition_restart_rehydration.py

@@ -887,7 +887,7 @@ def test_ambiguous_prepared_restart_ready_settles_after_dhcp_target_change(
         mqtt_retained=False,
         observed_at_utc="2026-08-13T12:20:01.000Z",
     )
-    record = restarted._physical_command_ledger.reconcile_ambiguous(  # noqa: SLF001
+    restarted._physical_command_ledger.reconcile_ambiguous(  # noqa: SLF001
         restart_support.START_OPERATION_ID,
         reconciliation_id="reconciliation-ready-new-dhcp-address",
         resolution="physical-standby-observed",
@@ -1057,7 +1057,7 @@ def test_restart_receiver_link_loss_before_first_pcl_rebinds_read_only_then_prom
     monkeypatch.setattr(
         restarted,
         "_probe_control_endpoint",
-        lambda target_ipv4: restart_support.facade_module._CorrelatedEndpointObservation(  # noqa: SLF001
+        lambda target_ipv4, **_kwargs: restart_support.facade_module._CorrelatedEndpointObservation(  # noqa: SLF001
             path=lifecycle_support._direct_host_path(target_ipv4),  # noqa: SLF001
             reachable=True,
             reason_code=None,
@@ -1237,7 +1237,7 @@ def test_ambiguous_prepared_restart_second_rebind_promotes_only_after_fresh_pcl(
     monkeypatch.setattr(
         restarted,
         "_probe_control_endpoint",
-        lambda target_ipv4: restart_support.facade_module._CorrelatedEndpointObservation(  # noqa: SLF001
+        lambda target_ipv4, **_kwargs: restart_support.facade_module._CorrelatedEndpointObservation(  # noqa: SLF001
             path=lifecycle_support._direct_host_path(target_ipv4),  # noqa: SLF001
             reachable=True,
             reason_code=None,

tests/test_xgrids_application_session.py

@@ -394,7 +394,10 @@ def _wait_phase(
     session: InteractiveApplicationControlSession,
     expected: str,
 ) -> dict[str, object]:
-    deadline = time.monotonic() + 2.0
+    # FakeExecutor may use its full two-second checkpoint timeout before the
+    # worker publishes the terminal state. Keep the observer deadline strictly
+    # larger so this helper does not race the transition it is asserting.
+    deadline = time.monotonic() + 3.0
     while time.monotonic() < deadline:
         snapshot = session.snapshot()
         if snapshot["state"] == expected: