feat(k1): wire canonical control session to UI

tests/test_xgrids_acquisition_lifecycle.py

@@ -16,6 +16,7 @@ from k1link.device_plugins.xgrids_k1.facade import (
     CameraPreviewSelectRequest,
     CompatibilityAttestationRequest,
     ConnectRequest,
+    OperatorPresenceRequest,
     PrepareAcquisitionRequest,
     ShadowApplicationControlArmRequest,
     StartAcquisitionRequest,
@@ -113,6 +114,61 @@ class FakeVisualizationRuntime:
         self.stop()
 
 
+class FakeInteractiveControlSession:
+    def __init__(self) -> None:
+        self.state = "workspace-ready"
+        self.start_projects: list[str] = []
+        self.stop_calls = 0
+        self.confirm_calls = 0
+
+    def snapshot(self) -> dict[str, object]:
+        return {
+            "state": self.state,
+            "can_confirm_standby": self.state == "awaiting-standby-confirmation",
+        }
+
+    def open_project_prompt(self) -> dict[str, object]:
+        assert self.state == "workspace-ready"
+        self.state = "project-ready"
+        return self.snapshot()
+
+    def request_start(self, *, project_name: str, confirmation: object) -> dict[str, object]:
+        assert confirmation is not None
+        assert self.state == "project-ready"
+        self.start_projects.append(project_name)
+        self.state = "scanning"
+        return self.snapshot()
+
+    def request_stop(self, *, confirmation: object) -> dict[str, object]:
+        assert confirmation is not None
+        assert self.state == "scanning"
+        self.stop_calls += 1
+        self.state = "awaiting-standby-confirmation"
+        return self.snapshot()
+
+    def confirm_standby(self) -> dict[str, object]:
+        assert self.state == "awaiting-standby-confirmation"
+        self.confirm_calls += 1
+        self.state = "completed"
+        return self.snapshot()
+
+    def close_prestart(self) -> dict[str, object]:
+        self.state = "closed"
+        return self.snapshot()
+
+    def close(self) -> None:
+        self.state = "closed"
+
+
+PHYSICAL_ACCEPTANCE = OperatorPresenceRequest(
+    operator_present=True,
+    owner_controlled_device=True,
+    lixelgo_closed=True,
+    battery_storage_confirmed=True,
+    expected_physical_state_confirmed=True,
+)
+
+
 def service_with_fake_runtime(
     tmp_path: Path,
 ) -> tuple[XgridsK1CompatibilityService, FakeVisualizationRuntime]:
@@ -291,6 +347,68 @@ def test_operator_manual_start_is_confirmed_only_by_real_point_data(tmp_path: Pa
     assert acquiring["last_operation"]["result"]["confirmation"] == "point-frame"
 
 
+def test_plugin_commanded_acquisition_keeps_start_and_stop_as_explicit_actions(
+    tmp_path: Path,
+) -> None:
+    service, runtime = service_with_fake_runtime(tmp_path)
+    control = FakeInteractiveControlSession()
+    service._application_control_session = control  # type: ignore[assignment]  # noqa: SLF001
+    service._k1_ip = "192.168.1.20"  # noqa: SLF001
+    service._compatibility_attestation = {"profile": "exact"}  # noqa: SLF001
+
+    prepared = service.prepare_acquisition(
+        PrepareAcquisitionRequest(
+            project_name="TEST001",
+            host="192.168.1.20",
+            compatibility_attestation=ATTESTATION,
+        )
+    )
+    acquisition_id = prepared["acquisition"]["acquisition_id"]
+    assert prepared["acquisition"]["control_mode"] == "plugin-commanded"
+    assert control.state == "project-ready"
+    assert runtime.start_calls == []
+
+    with pytest.raises(ValueError, match="подтверждения присутствия"):
+        service.start_acquisition(StartAcquisitionRequest(acquisition_id=acquisition_id))
+    assert control.start_projects == []
+
+    service.start_acquisition(
+        StartAcquisitionRequest(
+            acquisition_id=acquisition_id,
+            physical_acceptance=PHYSICAL_ACCEPTANCE,
+        )
+    )
+    assert control.start_projects == ["TEST001"]
+    runtime.mark_ready()
+    runtime.pcl_frames = 1
+    acquiring = service.state()
+    assert acquiring["acquisition"]["state"] == "acquiring"
+
+    stopping = service.stop_acquisition(
+        StopAcquisitionRequest(
+            acquisition_id=acquisition_id,
+            mode="graceful",
+            physical_acceptance=PHYSICAL_ACCEPTANCE,
+        )
+    )
+    stop_operation = stopping["last_operation"]
+    assert control.stop_calls == 1
+    assert stopping["acquisition"]["state"] == "awaiting_external_stop"
+
+    completed = service.stop_acquisition(
+        StopAcquisitionRequest(
+            acquisition_id=acquisition_id,
+            mode="graceful",
+            operator_confirmed=True,
+            operation_id=stop_operation["operation_id"],
+        )
+    )
+    assert control.stop_calls == 1
+    assert control.confirm_calls == 1
+    assert completed["acquisition"]["state"] == "completed"
+    assert runtime.stop_calls == 1
+
+
 def test_second_start_conflict_does_not_tear_down_start_owner(tmp_path: Path) -> None:
     service, runtime = service_with_fake_runtime(tmp_path)
     prepared = service.prepare_acquisition(