Стабилизация подключения K1 после сна и ожидания Bluetooth

docs/04_K1_WIFI_PROVISIONING_PROFILE.md

@@ -105,6 +105,13 @@ dispatch, a timeout, or an unchanged read-only observation without that exact
 confirmed write remains ambiguous and cannot authorize host association or an
 automatic retry.
 
+The reviewed 15-second limit applies to the byte-51 AP-ready observation after
+the one AP-enable write. It is not the CoreBluetooth connection timeout. The
+pre-write connection is one separate pending CoreBluetooth request with a
+bounded 30-second K1-appearance window. It performs no fallback discovery or
+retry; a timeout invalidates that exact captured handle with zero device writes
+and requires another explicit operator scan.
+
 Static review of the original client also established a lifecycle requirement:
 LixelGO keeps the same BLE manager connected after AP-ready and invokes native
 Wi-Fi association from that live session. Mission Core now retains the same

src/k1link/device_plugins/xgrids_k1/ble/ap_activation.py

@@ -120,6 +120,7 @@ async def _device_ap_activation_session_impl(
     poll_interval_seconds: float = 0.5,
     write_mode: WriteMode = "auto",
     *,
+    connect_timeout_seconds: float,
     captured_device: CapturedDiscoveredDevice | None,
     recovery_device_session_id: str | None,
     on_write_dispatch: Callable[[WifiStatus, ResolvedWriteMode], None] | None,
@@ -137,6 +138,8 @@ async def _device_ap_activation_session_impl(
 
     if timeout_seconds <= 0:
         raise ValueError("timeout_seconds must be positive")
+    if connect_timeout_seconds <= 0:
+        raise ValueError("connect_timeout_seconds must be positive")
     if poll_interval_seconds <= 0:
         raise ValueError("poll_interval_seconds must be positive")
     if write_mode not in ("auto", "with_response", "without_response"):
@@ -229,7 +232,7 @@ async def _device_ap_activation_session_impl(
             progress.operation_stage = operation_stage
             try:
                 client = await client_stack.enter_async_context(
-                    BleakClient(device, timeout=timeout_seconds, pair=False)
+                    BleakClient(device, timeout=connect_timeout_seconds, pair=False)
                 )
             except Exception as exc:
                 if active_captured_device is not None:
@@ -447,6 +450,7 @@ async def device_ap_activation_session(
     poll_interval_seconds: float = 0.5,
     write_mode: WriteMode = "auto",
     *,
+    connect_timeout_seconds: float | None = None,
     captured_device: CapturedDiscoveredDevice | None = None,
     recovery_device_session_id: str | None = None,
     on_write_dispatch: Callable[[WifiStatus, ResolvedWriteMode], None] | None = None,
@@ -459,15 +463,26 @@ async def device_ap_activation_session(
         )
     if recovery_device_session_id == "":
         raise ValueError("recovery_device_session_id must not be empty")
+    resolved_connect_timeout_seconds = (
+        timeout_seconds
+        if connect_timeout_seconds is None
+        else connect_timeout_seconds
+    )
+    if resolved_connect_timeout_seconds <= 0:
+        raise ValueError("connect_timeout_seconds must be positive")
     progress = BleOperationProgress(operation_stage="resolution")
     async with run_ble_operation_session(
         "ap-enable",
-        hard_setup_timeout_seconds=(timeout_seconds + BLE_AP_ENABLE_HARD_TIMEOUT_GRACE_SECONDS),
+        hard_setup_timeout_seconds=(
+            resolved_connect_timeout_seconds
+            + BLE_AP_ENABLE_HARD_TIMEOUT_GRACE_SECONDS
+        ),
         operation=lambda operation_progress: _device_ap_activation_session_impl(
             device_macos_uuid,
             timeout_seconds=timeout_seconds,
             poll_interval_seconds=poll_interval_seconds,
             write_mode=write_mode,
+            connect_timeout_seconds=resolved_connect_timeout_seconds,
             captured_device=captured_device,
             recovery_device_session_id=recovery_device_session_id,
             on_write_dispatch=on_write_dispatch,
@@ -484,6 +499,7 @@ async def activate_device_ap_once(
     poll_interval_seconds: float = 0.5,
     write_mode: WriteMode = "auto",
     *,
+    connect_timeout_seconds: float | None = None,
     captured_device: CapturedDiscoveredDevice | None = None,
     recovery_device_session_id: str | None = None,
     on_write_dispatch: Callable[[WifiStatus, ResolvedWriteMode], None] | None = None,
@@ -499,6 +515,7 @@ async def activate_device_ap_once(
         timeout_seconds=timeout_seconds,
         poll_interval_seconds=poll_interval_seconds,
         write_mode=write_mode,
+        connect_timeout_seconds=connect_timeout_seconds,
         captured_device=captured_device,
         recovery_device_session_id=recovery_device_session_id,
         on_write_dispatch=on_write_dispatch,

src/k1link/device_plugins/xgrids_k1/facade.py

@@ -452,6 +452,12 @@ _LocalStopRetirementDisposition = Literal[
 # LixelGo-parity discovery attempt at the old forty-second boundary.
 CONNECTION_VERIFY_HARD_TIMEOUT_SECONDS = 125.0
 CONNECTION_VERIFY_EXACT_UUID_SCAN_TIMEOUT_SECONDS = 30.0
+# The reviewed 15-second Quick Connect timeout starts only after the AP-enable
+# write while Mission Core waits for byte 51. Opening the one CoreBluetooth
+# session is a separate pre-write boundary. Give that single pending connect
+# the same bounded K1-appearance window as exact UUID observation; never turn
+# this into a second connect, scan or AP-enable write.
+QUICK_CONNECT_BLE_CONNECT_TIMEOUT_SECONDS = 30.0
 # A saved Quick Connect verification changes only the controller's Wi-Fi
 # association.  The K1 AP/profile pair was already durably established by the
 # reviewed BLE activation flow, so this recovery path must neither rediscover
@@ -7326,6 +7332,11 @@ class XgridsK1CompatibilityService:
             stop_operation_id=prepared_stop_operation_id,
         )
         self._retire_ephemeral_connection_binding_on_proven_loss(application_control_session)
+        # Proven-loss reduction may atomically move the existing capture
+        # producer from LIVE to RECONNECTING.  Re-sample that local fact before
+        # acquisition reduction so a START-active sleep/wake recovery cannot
+        # be overtaken by the stale pre-wake LIVE snapshot captured above.
+        runtime = self.runtime.snapshot()
         # Proven-loss retirement may synchronously reset a terminal control
         # owner to idle. Continue the same atomic snapshot from that factual
         # state instead of returning the stale pre-retirement failure row.
@@ -11414,6 +11425,9 @@ class XgridsK1CompatibilityService:
                     async with device_ap_activation_session(
                         request.device_id,
                         timeout_seconds=15.0,
+                        connect_timeout_seconds=(
+                            QUICK_CONNECT_BLE_CONNECT_TIMEOUT_SECONDS
+                        ),
                         # The accepted macOS / K1 FW 3.0.2 transport is one
                         # write-with-response frame. Do not reinterpret the live
                         # characteristic metadata into another write mode here.
@@ -27774,6 +27788,42 @@ class XgridsK1CompatibilityService:
         provisioning remains forbidden until fresh READY evidence resolves it.
         """
 
+        # A suspend may let the data socket resume while the older control
+        # request loses only its trailing read-only response.  Admit the same
+        # reviewed recovery used for a dead MQTT loop, but only after the
+        # executor reached exact initialized SCANNING and before any STOP was
+        # attempted.  The recovery admission independently requires the
+        # resolved composite START and active durable checkpoint; otherwise it
+        # returns false and the existing fail-closed reduction remains intact.
+        post_start_failure = application_control_session.get("failure")
+        post_start_read_only_timeout = bool(
+            application_control_session.get("state") == "failed"
+            and isinstance(post_start_failure, Mapping)
+            and post_start_failure.get("reason_code") == "mqtt_response_timeout"
+            and post_start_failure.get("failed_phase") == "initializing"
+            and post_start_failure.get("dialogue_stage") == "start-active-observed"
+            and post_start_failure.get("modeling_command_attempted") is True
+            and post_start_failure.get("stop_command_attempted") is False
+        )
+        if post_start_read_only_timeout and self._request_active_stream_recovery_for_local_loss(
+            "mqtt_response_timeout"
+        ):
+            with self._lock:
+                self._connection_loss_signature = None
+                self._connection_loss_evidence_token = None
+                self._connection_loss_confirmation_count = 0
+            logger.info(
+                "K1 post-START read-only timeout entered active stream recovery",
+                extra={
+                    "event_code": "k1_post_start_read_only_timeout_recovery_wake",
+                    "reason_code": "mqtt_response_timeout",
+                    "automatic_command_retry": False,
+                    "device_command_performed": False,
+                    "network_mutation_performed": False,
+                },
+            )
+            return
+
         # A composite-confirmed active START owns the sole automatic exception:
         # preserve its exact topology while the data owner performs a bounded,
         # inspection-only rebind.  This path never authorizes a command retry.

src/k1link/device_plugins/xgrids_k1/protocol/application_acceptance.py

@@ -277,6 +277,7 @@ class PhysicalAcceptanceDialogueExecutor:
         self._bootstrap_complete = False
         self._command_complete = False
         self._dialogue_stage = "new"
+        self._start_active_confirmed = False
         self._start_complete = False
         self._active_session_adopted = False
         self._stop_attempted = False
@@ -493,6 +494,7 @@ class PhysicalAcceptanceDialogueExecutor:
         permit: PhysicalAcceptancePermit,
         checkpoint: OperatorDialogueCheckpoint,
         dispatch_guard: Callable[[], None] | None = None,
+        start_active_observer: Callable[[], None] | None = None,
     ) -> ModelingResponse:
         """Execute retained operations 11-14 on one continuously serviced socket."""
 
@@ -564,6 +566,17 @@ class PhysicalAcceptanceDialogueExecutor:
                 allowed_response_topics={MODELING_STATUS_RESPONSE_TOPIC},
             )
 
+        # The correlated START response plus fresh, bound, initialized
+        # SCANNING is the complete physical side-effect proof.  Persist that
+        # fact before the trailing read-only refresh so a host suspend between
+        # ordinals 12 and 13/14 cannot relabel an already-running K1 as an
+        # unknown START.  The observer performs storage only; it neither
+        # grants STOP authority nor changes the canonical transcript.
+        self._start_active_confirmed = True
+        self._dialogue_stage = "start-active-observed"
+        if start_active_observer is not None:
+            start_active_observer()
+
         refresh = post_start.post_initialization_refresh
         required_operations = {
             f"dialogue:{request.ordinal}:{request.message_type}" for request in refresh
@@ -797,6 +810,7 @@ class PhysicalAcceptanceDialogueExecutor:
             "bootstrap_complete": self._bootstrap_complete,
             "command_complete": self._command_complete,
             "start_attempted": self._command_complete,
+            "start_active_confirmed": self._start_active_confirmed,
             "start_complete": self._start_complete,
             "active_session_adopted": self._active_session_adopted,
             "stop_attempted": self._stop_attempted,

src/k1link/device_plugins/xgrids_k1/protocol/application_session.py

@@ -1013,28 +1013,56 @@ class InteractiveApplicationControlSession:
                 )
                 self._validate_connection_binding_snapshot("start-pre-dispatch")
                 self._set_phase("initializing")
-                executor.execute_canonical_start(
+                start_active_observed = False
-                    start_command,
+                start_transition_gate_acquired = False
-                    build_canonical_post_start_observation(authority, binding),
+
-                    authority=authority,
+                def observe_start_active() -> None:
-                    binding=binding,
+                    nonlocal start_active_observed, start_transition_gate_acquired
-                    permit=start_permit,
+                    if start_active_observed:
-                    checkpoint=start_checkpoint,
+                        return
-                    dispatch_guard=lambda: self._validate_connection_binding_snapshot(
+                    self._scanning_transition_gate.acquire()
-                        "start-dispatch"
+                    start_transition_gate_acquired = True
-                    ),
+                    try:
-                )
+                        if coordinator is not None:
-                self._validate_connection_binding_snapshot("start-post-response")
+                            coordinator.resolve("start")
-                if coordinator is not None:
+                        # Activate the durable recovery checkpoint at the same
-                    coordinator.resolve("start")
+                        # exact SCANNING proof as the physical ledger.  The
-                with self._scanning_transition_gate:
+                        # gate remains held while the read-only ordinals 13-14
-                    # Activate the exact durable checkpoint before publishing
+                        # finish, so STOP cannot overtake the later public
-                    # the STOP-admitting SCANNING phase.  The observer performs
+                        # ``scanning`` transition.
-                    # storage only; it never sends a device command.
+                        if not self._scanning_observer_confirmed:
-                    self._scanning_observer_confirmed = (
+                            self._scanning_observer_confirmed = (
-                        self._notify_scanning_observer()
+                                self._notify_scanning_observer()
+                            )
+                        start_active_observed = True
+                    except BaseException:
+                        self._scanning_transition_gate.release()
+                        start_transition_gate_acquired = False
+                        raise
+
+                try:
+                    executor.execute_canonical_start(
+                        start_command,
+                        build_canonical_post_start_observation(authority, binding),
+                        authority=authority,
+                        binding=binding,
+                        permit=start_permit,
+                        checkpoint=start_checkpoint,
+                        dispatch_guard=lambda: self._validate_connection_binding_snapshot(
+                            "start-dispatch"
+                        ),
+                        start_active_observer=observe_start_active,
                     )
+                    # Compatibility executors used by lower-level integrations may
+                    # return without invoking the new proof callback.  A normal
+                    # production executor invokes it before the read-only refresh;
+                    # this idempotent fallback remains strictly post-success.
+                    observe_start_active()
+                    self._validate_connection_binding_snapshot("start-post-response")
                     self._set_phase("scanning")
+                finally:
+                    if start_transition_gate_acquired:
+                        self._scanning_transition_gate.release()
 
             executor.maintain_active_until_stop_requested(self._stop_requested.is_set)
             (

tests/test_xgrids_acquisition_lifecycle.py

@@ -15417,6 +15417,56 @@ def test_control_first_loss_freezes_topology_before_ephemeral_retirement(
     assert service._physical_command_coordinator.snapshot() is physical  # noqa: SLF001
 
 
+def test_post_start_refresh_timeout_wakes_existing_read_only_recovery(
+    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",
+    )
+    with service._lock:  # noqa: SLF001
+        acquisition = service._acquisition  # noqa: SLF001
+        assert acquisition is not None
+        acquisition.state = "awaiting_external_start"
+        acquisition.message_code = "acquisition.start.device_initializing"
+        acquisition.state_revision += 1
+    control.state = "failed"
+    control.outcome_unknown = True
+    control.failure = {
+        "reason_code": "mqtt_response_timeout",
+        "failed_phase": "initializing",
+        "dialogue_stage": "start-active-observed",
+        "modeling_command_attempted": True,
+        "stop_command_attempted": False,
+        "safe_to_retry": False,
+    }
+    start_projects_before = list(control.start_projects)
+    stop_calls_before = control.stop_calls
+
+    state = service.state()
+
+    assert runtime.recovery_requests == [
+        ("mqtt_response_timeout", runtime.producer_generation),
+    ]
+    assert state["phase"] == "reconnecting"
+    assert state["connection_recovery"]["automatic_read_only_rebind"] is True
+    assert state["acquisition"]["state"] == "awaiting_external_start"
+    assert state["acquisition"]["cleanup_pending"] is False
+    assert state["acquisition"]["result"] is None
+    assert state["last_operation"]["status"] == "running"
+    assert state["selected_device_id"] == "test-ble-transport"
+    assert state["source_mode"] == "live"
+    assert runtime.stop_calls == 0
+    assert control.start_projects == start_projects_before
+    assert control.stop_calls == stop_calls_before
+    assert service._physical_command_coordinator.snapshot() is physical  # noqa: SLF001
+
+
 def test_incident_host_route_streak_wakes_recovery_before_binding_retirement(
     tmp_path: Path,
     monkeypatch: pytest.MonkeyPatch,
@@ -20311,7 +20361,7 @@ def test_quick_connect_activates_the_device_ap_then_associates_the_host(
         lambda _target: True,
     )
     _set_scanned_devices(service, [{"device_id": "k1-a", "name": "XGR-TEST-A"}])
-    activation_calls: list[tuple[str, str]] = []
+    activation_calls: list[tuple[str, str, float, float]] = []
     association_calls: list[tuple[Path, str, str]] = []
     ble_session_open = False
 
@@ -20331,12 +20381,21 @@ def test_quick_connect_activates_the_device_ap_then_associates_the_host(
     async def fake_activation_session(
         device_id: str,
         *,
+        timeout_seconds: float,
+        connect_timeout_seconds: float,
         write_mode: str,
         on_write_dispatch: Callable[[dict[str, Any], str], None] | None = None,
         **_: object,
     ) -> AsyncIterator[dict[str, Any]]:
         nonlocal ble_session_open
-        activation_calls.append((device_id, write_mode))
+        activation_calls.append(
+            (
+                device_id,
+                write_mode,
+                timeout_seconds,
+                connect_timeout_seconds,
+            )
+        )
         _dispatch_test_network_write(on_write_dispatch)
         ble_session_open = True
         try:
@@ -20396,7 +20455,14 @@ def test_quick_connect_activates_the_device_ap_then_associates_the_host(
         )
     )
 
-    assert activation_calls == [("k1-a", "with_response")]
+    assert activation_calls == [
+        (
+            "k1-a",
+            "with_response",
+            15.0,
+            facade_module.QUICK_CONNECT_BLE_CONNECT_TIMEOUT_SECONDS,
+        )
+    ]
     assert not ble_session_open
     assert len(association_calls) == 1
     assert association_calls[0][0].name == "associate_wifi.swift"

tests/test_xgrids_ap_activation.py

@@ -79,7 +79,7 @@ def test_ap_activation_uses_retained_handle_without_rediscovery(
     device_id = "synthetic-corebluetooth-uuid"
     retained_handle = BLEDevice(device_id, "XGR-K1", details=object())
     rediscovery_calls: list[tuple[str, float]] = []
-    client_calls: list[object] = []
+    client_calls: list[tuple[object, float, bool]] = []
 
     class SelectedHandleObserved(RuntimeError):
         pass
@@ -89,8 +89,14 @@ def test_ap_activation_uses_retained_handle_without_rediscovery(
         raise AssertionError("a fresh explicit scan handle must be used directly")
 
     class CapturingClient:
-        def __init__(self, device: object, **_kwargs: object) -> None:
+        def __init__(
-            client_calls.append(device)
+            self,
+            device: object,
+            *,
+            timeout: float,
+            pair: bool,
+        ) -> None:
+            client_calls.append((device, timeout, pair))
             raise SelectedHandleObserved
 
     monkeypatch.setattr(scanner_module, "monotonic", lambda: 100.0)
@@ -117,6 +123,7 @@ def test_ap_activation_uses_retained_handle_without_rediscovery(
             await ap_module.activate_device_ap_once(
                 device_id,
                 timeout_seconds=1.0,
+                connect_timeout_seconds=30.0,
                 captured_device=captured,
             )
         assert (
@@ -131,7 +138,7 @@ def test_ap_activation_uses_retained_handle_without_rediscovery(
     error = asyncio.run(scenario())
 
     assert rediscovery_calls == []
-    assert client_calls == [retained_handle]
+    assert client_calls == [(retained_handle, 30.0, False)]
     assert error.operation_stage == "connect"  # type: ignore[attr-defined]
     assert error.device_write_attempted is False  # type: ignore[attr-defined]
     assert error.device_write_confirmed is False  # type: ignore[attr-defined]

tests/test_xgrids_application_acceptance.py

@@ -545,6 +545,97 @@ def test_start_initialization_wait_has_fail_closed_watchdog() -> None:
     assert executor.snapshot()["start_complete"] is False
 
 
+def test_initialized_scanning_is_observed_before_post_start_refresh_timeout() -> None:
+    class PostStartRefreshTimeoutTransport(SyntheticAcceptanceTransport):
+        def exchange_batch_once(
+            self,
+            envelopes: Sequence[OneShotPublishEnvelope],
+            *,
+            required_response_operation_keys: Collection[str],
+            dispatch_admission_deadline_reached: Callable[[], bool] | None = None,
+            dispatch_admission_commit: Callable[[], None] | None = None,
+        ) -> dict[str, bytes]:
+            operation_keys = tuple(envelope.operation_key for envelope in envelopes)
+            if operation_keys == (
+                "dialogue:13:DeviceInfoRequest",
+                "dialogue:14:ModelingStatusRequest",
+            ):
+                self.batches.append(operation_keys)
+                raise ApplicationMqttTransportError(
+                    "post-START read-only response timed out after host wake",
+                    reason_code="mqtt_response_timeout",
+                )
+            return super().exchange_batch_once(
+                envelopes,
+                required_response_operation_keys=required_response_operation_keys,
+                dispatch_admission_deadline_reached=(
+                    dispatch_admission_deadline_reached
+                ),
+                dispatch_admission_commit=dispatch_admission_commit,
+            )
+
+    transport = PostStartRefreshTimeoutTransport()
+    executor = PhysicalAcceptanceDialogueExecutor(transport)
+    authority = ApplicationControlAuthority(openapi_key=APPLICATION_KEY)
+    orchestrator = ShadowApplicationBootstrapOrchestrator(
+        authority,
+        epoch_seconds=1_752_680_000,
+        timezone_name="Europe/Moscow",
+    )
+    binding = executor.run_connection_stage(orchestrator)
+    executor.run_workspace_entry_stage(
+        orchestrator,
+        executor.wait_for_operator_checkpoint("workspace-entered", lambda: True),
+    )
+    executor.run_project_prompt_stage(
+        orchestrator,
+        executor.wait_for_operator_checkpoint("project-prompt-opened", lambda: True),
+    )
+    command = ShadowModelingCommand.from_command(
+        encode_modeling_start(
+            CommandHeaderIdentity(
+                device_id=binding.vendor_device_id,
+                openapi_key=APPLICATION_KEY,
+            ),
+            project_name="SLEEP_WAKE",
+            record_mode=RecordMode.RECORD_AND_CALCULATE,
+            scan_mode=ScanMode.LCC,
+            mount_type=MountType.HANDHELD,
+        )
+    )
+    observed_stages: list[str] = []
+
+    with pytest.raises(ApplicationMqttTransportError) as raised:
+        executor.execute_canonical_start(
+            command,
+            build_canonical_post_start_observation(authority, binding),
+            authority=authority,
+            binding=binding,
+            permit=PhysicalAcceptancePermit(_checklist(ModelingAction.START)),
+            checkpoint=executor.wait_for_operator_checkpoint(
+                "start-confirmed",
+                lambda: True,
+            ),
+            start_active_observer=lambda: observed_stages.append(
+                str(executor.snapshot()["dialogue_stage"])
+            ),
+        )
+
+    assert raised.value.reason_code == "mqtt_response_timeout"
+    assert observed_stages == ["start-active-observed"]
+    assert transport.batches[-3:] == [
+        ("modeling:start",),
+        ("dialogue:12:ModelingStatusRequest",),
+        (
+            "dialogue:13:DeviceInfoRequest",
+            "dialogue:14:ModelingStatusRequest",
+        ),
+    ]
+    assert executor.snapshot()["start_active_confirmed"] is True
+    assert executor.snapshot()["start_complete"] is False
+    assert transport.stop_emitted is False
+
+
 @pytest.mark.parametrize(
     ("device_model", "device_type"),
     (

tests/test_xgrids_application_session.py

@@ -2222,6 +2222,90 @@ def test_start_outcome_unknown_blocks_reopen_even_when_transport_is_closed(
         session.retire_for_network_change()
 
 
+def test_post_start_read_only_timeout_preserves_durable_active_proof_without_stop_authority(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    class PostStartRefreshFailureExecutor(FakeExecutor):
+        start_attempted = False
+        start_active_confirmed = False
+
+        def execute_canonical_start(self, *_args: object, **kwargs: object) -> object:
+            type(self).start_attempted = True
+            self.records.append("start:11-12")
+            self.transport.ready = False
+            observer = kwargs.get("start_active_observer")
+            assert callable(observer)
+            observer()
+            type(self).start_active_confirmed = True
+            self.records.append("refresh:13-14-timeout")
+            raise ApplicationMqttTransportError(
+                "post-START read-only response timed out after host wake",
+                reason_code="mqtt_response_timeout",
+            )
+
+        def snapshot(self) -> dict[str, object]:
+            return {
+                "dialogue_stage": (
+                    "start-active-observed"
+                    if type(self).start_active_confirmed
+                    else "start-attempted"
+                ),
+                "start_attempted": type(self).start_attempted,
+                "start_active_confirmed": type(self).start_active_confirmed,
+                "start_complete": False,
+                "stop_attempted": False,
+                "response_evidence": [],
+                "automatic_retry": False,
+            }
+
+    FakeExecutor.records = []
+    PostStartRefreshFailureExecutor.start_attempted = False
+    PostStartRefreshFailureExecutor.start_active_confirmed = False
+    monkeypatch.setattr(
+        session_module,
+        "PhysicalAcceptanceDialogueExecutor",
+        PostStartRefreshFailureExecutor,
+    )
+    coordinator = FakePhysicalCommandCoordinator()
+    scanning_observed = threading.Event()
+    session = InteractiveApplicationControlSession(
+        FakeAuthorityLoader(),
+        transport_factory=lambda host: FakeTransport(host),  # type: ignore[arg-type]
+        physical_command_coordinator=coordinator,  # type: ignore[arg-type]
+        scanning_observer=scanning_observed.set,
+    )
+    coordinator.phase_probe = lambda: str(session.snapshot()["state"])
+
+    session.open(
+        host="192.168.1.20",
+        timezone_name="Europe/Moscow",
+        connection_binding=_connection_binding(),
+    )
+    _wait_phase(session, "connection-ready")
+    session.enter_workspace()
+    _wait_phase(session, "workspace-ready")
+    session.open_project_prompt()
+    _wait_phase(session, "project-ready")
+    session.request_start(
+        project_name="SLEEP_WAKE",
+        confirmation=_confirmation(),
+        command_context=_physical_context("start"),
+        preparation_checkpoint_observer=_successful_start_checkpoint_observer,  # type: ignore[arg-type]
+    )
+    failed = _wait_phase(session, "failed")
+
+    assert coordinator.resolutions == [("start", "initializing")]
+    assert scanning_observed.is_set()
+    assert failed["can_stop"] is False
+    assert failed["outcome_unknown"] is True
+    assert failed["failure"]["reason_code"] == "mqtt_response_timeout"  # type: ignore[index]
+    assert failed["dialogue"]["start_active_confirmed"] is True  # type: ignore[index]
+    assert FakeExecutor.records[-2:] == [
+        "start:11-12",
+        "refresh:13-14-timeout",
+    ]
+
+
 @pytest.mark.parametrize(
     ("disconnect_error", "expected_reason_code"),
     [