Стабильное подключение в идеальных условиях K1

tests/test_xgrids_application_session.py

@@ -500,6 +500,182 @@ def test_canonical_stages_require_operator_events_but_device_standby_does_not(
     ]
 
 
+def test_session_checkpoints_do_not_repeat_native_dispatch_proof(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Slow host-path proof belongs to the exact publish lease, not UI checkpoints."""
+
+    class DispatchingExecutor(FakeExecutor):
+        def _dispatch_one_packet(self) -> None:
+            guard = self.transport.dispatch_guard
+            assert guard is not None
+            release = guard()
+            try:
+                self.transport.publish_attempts += 1
+            finally:
+                if callable(release):
+                    release()
+
+        def run_workspace_entry_stage(
+            self,
+            orchestrator: object,
+            checkpoint: object,
+            *,
+            dispatch_guard: Callable[[], None] | None = None,
+        ) -> LiveDeviceControlBinding:
+            if dispatch_guard is not None:
+                dispatch_guard()
+            self._dispatch_one_packet()
+            return super().run_workspace_entry_stage(
+                orchestrator,
+                checkpoint,
+                dispatch_guard=None,
+            )
+
+        def run_project_prompt_stage(
+            self,
+            orchestrator: object,
+            checkpoint: object,
+            *,
+            dispatch_guard: Callable[[], None] | None = None,
+        ) -> LiveDeviceControlBinding:
+            if dispatch_guard is not None:
+                dispatch_guard()
+            self._dispatch_one_packet()
+            return super().run_project_prompt_stage(
+                orchestrator,
+                checkpoint,
+                dispatch_guard=None,
+            )
+
+        def execute_canonical_start(self, *_args: object, **kwargs: object) -> object:
+            dispatch_guard = kwargs.get("dispatch_guard")
+            if callable(dispatch_guard):
+                dispatch_guard()
+            self._dispatch_one_packet()
+            return super().execute_canonical_start(*_args, **kwargs)
+
+        def execute_canonical_stop(self, *_args: object, **kwargs: object) -> object:
+            dispatch_guard = kwargs.get("dispatch_guard")
+            if callable(dispatch_guard):
+                dispatch_guard()
+            self._dispatch_one_packet()
+            return super().execute_canonical_stop(*_args, **kwargs)
+
+    FakeExecutor.records = []
+    monkeypatch.setattr(
+        session_module,
+        "PhysicalAcceptanceDialogueExecutor",
+        DispatchingExecutor,
+    )
+    initial_path_proofs: list[ApplicationConnectionBinding] = []
+    full_binding_proofs: list[ApplicationConnectionBinding] = []
+    snapshot_proofs: list[ApplicationConnectionBinding] = []
+    dispatch_proofs: list[ApplicationConnectionBinding] = []
+    dispatch_releases: list[ApplicationConnectionBinding] = []
+
+    def acquire_dispatch_proof(
+        binding: ApplicationConnectionBinding,
+        _deadline_reached: Callable[[], bool] | None,
+    ) -> Callable[[], None]:
+        dispatch_proofs.append(binding)
+        return lambda: dispatch_releases.append(binding)
+
+    transport = FakeTransport("192.168.1.20")
+    session = InteractiveApplicationControlSession(
+        FakeAuthorityLoader(),
+        transport_factory=lambda _host: transport,  # type: ignore[arg-type]
+        connection_path_validator=lambda binding: initial_path_proofs.append(binding),
+        connection_binding_validator=lambda binding: full_binding_proofs.append(binding),
+        connection_binding_snapshot_validator=lambda binding: snapshot_proofs.append(binding),
+        connection_dispatch_lease=acquire_dispatch_proof,
+    )
+    session.open(
+        host="192.168.1.20",
+        timezone_name="Europe/Moscow",
+        confirmation=_confirmation(),
+        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="TEST001", confirmation=_confirmation())
+    _wait_phase(session, "scanning")
+    session.request_stop(confirmation=_confirmation())
+    _wait_phase(session, "completed")
+
+    assert initial_path_proofs == [_connection_binding()]
+    assert full_binding_proofs == []
+    assert len(snapshot_proofs) >= 12
+    assert dispatch_proofs == [_connection_binding()] * 4
+    assert dispatch_releases == dispatch_proofs
+
+
+def test_exact_publish_lease_still_rejects_epoch_drift_after_snapshot_check(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A cheap checkpoint pass can never bypass the final physical route fence."""
+
+    class DispatchRejectedExecutor(FakeExecutor):
+        def run_workspace_entry_stage(
+            self,
+            orchestrator: object,
+            checkpoint: object,
+            *,
+            dispatch_guard: Callable[[], None] | None = None,
+        ) -> LiveDeviceControlBinding:
+            if dispatch_guard is not None:
+                dispatch_guard()
+            guard = self.transport.dispatch_guard
+            assert guard is not None
+            guard()
+            return super().run_workspace_entry_stage(
+                orchestrator,
+                checkpoint,
+                dispatch_guard=None,
+            )
+
+    FakeExecutor.records = []
+    monkeypatch.setattr(
+        session_module,
+        "PhysicalAcceptanceDialogueExecutor",
+        DispatchRejectedExecutor,
+    )
+    snapshot_proofs: list[ApplicationConnectionBinding] = []
+
+    def reject_dispatch(
+        _binding: ApplicationConnectionBinding,
+        _deadline_reached: Callable[[], bool] | None,
+    ) -> Callable[[], None]:
+        raise ApplicationConnectionBindingLost("test host-path epoch changed")
+
+    session = InteractiveApplicationControlSession(
+        FakeAuthorityLoader(),
+        transport_factory=lambda host: FakeTransport(host),  # type: ignore[arg-type]
+        connection_path_validator=lambda _binding: True,
+        connection_binding_validator=lambda _binding: True,
+        connection_binding_snapshot_validator=lambda binding: snapshot_proofs.append(binding),
+        connection_dispatch_lease=reject_dispatch,
+    )
+    session.open(
+        host="192.168.1.20",
+        timezone_name="Europe/Moscow",
+        confirmation=_confirmation(),
+        connection_binding=_connection_binding(),
+    )
+    _wait_phase(session, "connection-ready")
+    session.enter_workspace()
+    failed = _wait_phase(session, "failed")
+
+    assert snapshot_proofs
+    assert "workspace:7" not in FakeExecutor.records
+    assert failed["failure"]["reason_code"] == (  # type: ignore[index]
+        ApplicationConnectionBindingLost.reason_code
+    )
+
+
 def test_read_only_device_info_open_does_not_require_physical_acceptance(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None: