feat(k1): wire dormant control lease

tests/test_xgrids_acquisition_lifecycle.py

@@ -17,10 +17,14 @@ from k1link.device_plugins.xgrids_k1.facade import (
     CompatibilityAttestationRequest,
     ConnectRequest,
     PrepareAcquisitionRequest,
+    ShadowApplicationControlArmRequest,
     StartAcquisitionRequest,
     StopAcquisitionRequest,
     XgridsK1CompatibilityService,
 )
+from k1link.device_plugins.xgrids_k1.protocol.application_bootstrap import (
+    ApplicationControlAuthority,
+)
 
 ATTESTATION = CompatibilityAttestationRequest(
     firmware_version="3.0.2",
@@ -30,6 +34,16 @@ ATTESTATION = CompatibilityAttestationRequest(
 PRIMARY_TEST_CREDENTIAL = "x" * 24
 SECONDARY_TEST_CREDENTIAL = "y" * 24
 PROJECT_NAME = "K1 lifecycle test"
+PRIVATE_APPLICATION_AUTHORITY = "11111111-2222-3333-4444-555555555555"
+
+
+class FakeApplicationAuthorityLoader:
+    def __init__(self) -> None:
+        self.calls = 0
+
+    def load(self) -> ApplicationControlAuthority:
+        self.calls += 1
+        return ApplicationControlAuthority(openapi_key=PRIVATE_APPLICATION_AUTHORITY)
 
 
 class FakeVisualizationRuntime:
@@ -148,6 +162,104 @@ def test_project_name_is_normalized_and_control_characters_are_rejected() -> Non
             )
 
 
+def test_facade_arms_bounded_shadow_lease_without_installing_publish_transport(
+    tmp_path: Path,
+) -> None:
+    loader = FakeApplicationAuthorityLoader()
+    service = XgridsK1CompatibilityService(
+        tmp_path,
+        application_authority_loader=loader,
+    )
+    runtime = FakeVisualizationRuntime()
+    service.runtime = runtime  # type: ignore[assignment]
+    service._selected_device_id = "test-ble-transport"  # noqa: SLF001
+    service._k1_ip = "192.168.1.20"  # noqa: SLF001
+    service._compatibility_attestation = {  # noqa: SLF001
+        "firmware_version": "3.0.2",
+        "topology": "direct-lan",
+        "basis": "operator-attested",
+        "observed_at": "2026-07-18T00:00:00Z",
+    }
+
+    state = service.arm_application_control_shadow(
+        ShadowApplicationControlArmRequest(
+            operator_confirmed=True,
+            lease_seconds=60.0,
+            timezone_name="Europe/Moscow",
+        )
+    )
+
+    execution = state["application_control_execution"]
+    assert loader.calls == 1
+    assert execution["state"] == "armed-shadow-only"
+    assert execution["lease"]["authority_cached"] is True
+    assert execution["can_emit_requests"] is False
+    assert execution["live_transport_installed"] is False
+    assert execution["publisher"]["vendor_writes_enabled"] is False
+    assert execution["publisher"]["transport_calls"] == 0
+    assert PRIVATE_APPLICATION_AUTHORITY not in str(state)
+
+    disarmed = service.disarm_application_control_shadow()
+    assert disarmed["application_control_execution"]["state"] == "disarmed"
+    assert disarmed["application_control_execution"]["lease"] is None
+
+
+def test_shadow_arm_requires_idle_connected_attested_device_before_keychain_read(
+    tmp_path: Path,
+) -> None:
+    loader = FakeApplicationAuthorityLoader()
+    service = XgridsK1CompatibilityService(
+        tmp_path,
+        application_authority_loader=loader,
+    )
+    service.runtime = FakeVisualizationRuntime()  # type: ignore[assignment]
+
+    with pytest.raises(RuntimeError, match="подключите K1"):
+        service.arm_application_control_shadow(
+            ShadowApplicationControlArmRequest(
+                operator_confirmed=True,
+                timezone_name="Europe/Moscow",
+            )
+        )
+
+    assert loader.calls == 0
+
+
+def test_shadow_arm_contract_requires_explicit_operator_confirmation() -> None:
+    with pytest.raises(ValidationError):
+        ShadowApplicationControlArmRequest.model_validate({"timezone_name": "Europe/Moscow"})
+
+
+def test_prepare_acquisition_revokes_existing_shadow_authority_lease(tmp_path: Path) -> None:
+    loader = FakeApplicationAuthorityLoader()
+    service = XgridsK1CompatibilityService(
+        tmp_path,
+        application_authority_loader=loader,
+    )
+    service.runtime = FakeVisualizationRuntime()  # type: ignore[assignment]
+    service._selected_device_id = "test-ble-transport"  # noqa: SLF001
+    service._k1_ip = "192.168.1.20"  # noqa: SLF001
+    service._compatibility_attestation = {"profile": "exact"}  # noqa: SLF001
+    service.arm_application_control_shadow(
+        ShadowApplicationControlArmRequest(
+            operator_confirmed=True,
+            timezone_name="Europe/Moscow",
+        )
+    )
+
+    prepared = service.prepare_acquisition(
+        PrepareAcquisitionRequest(
+            project_name=PROJECT_NAME,
+            host="192.168.1.20",
+            duration_seconds=60,
+            compatibility_attestation=ATTESTATION,
+        )
+    )
+
+    assert prepared["application_control_execution"]["state"] == "disarmed"
+    assert prepared["application_control_execution"]["lease"] is None
+
+
 def test_operator_manual_start_is_confirmed_only_by_real_point_data(tmp_path: Path) -> None:
     service, runtime = service_with_fake_runtime(tmp_path)
     prepared = service.prepare_acquisition(