test(openspec): map cloud integration scenarios

openspec/changes/cloud-control-plane-integration/tasks.md

@@ -74,5 +74,5 @@
 - [x] 9.1 Run formatting, static checks, all non-integration tests, and targeted PostgreSQL integration/concurrency tests.
 - [x] 9.2 Run an end-to-end cloud submission through a Host Agent and fake device until the public SDK reports done and a failure case until it reports failed.
 - [x] 9.3 Verify existing local REST/MCP/console behavior and dependency-boundary tests remain unchanged.
-- [ ] 9.4 Run OpenSpec validation for `cloud-control-plane-integration` and map automated tests to every new or modified scenario.
+- [x] 9.4 Run OpenSpec validation for `cloud-control-plane-integration` and map automated tests to every new or modified scenario.
 - [ ] 9.5 Update the project index, architecture/deployment documentation, and runtime maturity memory after implementation verification.

openspec/changes/cloud-control-plane-integration/verification.md

@@ -0,0 +1,71 @@
+# Verification Mapping
+
+Strict validation:
+
+```text
+openspec validate cloud-control-plane-integration --strict
+Change 'cloud-control-plane-integration' is valid
+```
+
+## cloud-control-plane
+
+- Start the cloud application: `apps/cloud-api/tests/test_app.py::test_create_app_returns_independent_cloud_application` and `test_cloud_application_owns_database_lifecycle`.
+- Local Runtime remains separately runnable: `tests/test_cloud_composition_safety.py::test_existing_modules_do_not_import_cloud` and `tests/test_rest_api.py::test_rest_start_task_and_poll_until_complete`.
+- Queued work becomes assigned: `apps/cloud-api/tests/test_app.py::test_lifespan_runs_scheduler_and_reaper_until_shutdown`.
+- Lifecycle iteration fails transiently: `apps/cloud-api/tests/test_app.py::test_lifecycle_workers_log_failures_and_continue`.
+- Application shuts down: `apps/cloud-api/tests/test_app.py::test_lifespan_runs_scheduler_and_reaper_until_shutdown`.
+- Start with PostgreSQL: PostgreSQL parameter of `tests/test_cloud_repository_contract.py::test_repository_crud_contract`.
+- Start in local SQLite mode: `tests/test_cloud_database.py::test_sqlite_database_owns_working_repository_lifecycle`.
+- Upgrade an existing cloud database: `tests/test_cloud_migrations.py::test_legacy_data_survives_upgrade_and_downgrade`.
+- Schema is behind at startup: `tests/test_cloud_migrations.py::test_schema_readiness_requires_head_revision`.
+- Restart with queued tasks: `apps/cloud-api/tests/test_app.py::test_persisted_queue_is_recovered_after_control_plane_restart`.
+- Restart after a lease expires: `apps/cloud-api/tests/test_app.py::test_expired_lease_is_recovered_after_control_plane_restart`.
+- Database is unavailable: `apps/cloud-api/tests/test_app.py::test_liveness_stays_up_when_database_readiness_fails`.
+- Task lifecycle is logged: `tests/test_cloud_repository_contract.py::test_task_lifecycle_logs_structured_identifiers`, `tests/test_cloud_observability.py::test_sensitive_payload_fields_are_redacted_recursively`, and `apps/cloud-api/tests/test_app.py::test_request_correlation_id_is_propagated_without_sensitive_headers`.
+
+## device-pool
+
+- Valid remote snapshot: `tests/test_host_agent_internal_api.py::test_authenticated_heartbeat_replaces_complete_snapshot`.
+- Invalid snapshot is rejected atomically: `tests/test_host_agent_internal_api.py::test_invalid_duplicate_snapshot_preserves_previous_devices` and `tests/test_host_agent_protocol_models.py::test_device_snapshot_rejects_invalid_host_status`.
+- Two live hosts report the same device id: `tests/test_host_agent_internal_api.py::test_live_device_owner_conflict_is_rejected_without_partial_sync`.
+- Previous owner is stale: `tests/test_host_agent_internal_api.py::test_stale_device_owner_can_be_replaced_by_live_host` and the SQLite/PostgreSQL parameters of `tests/test_cloud_repository_contract.py::test_repository_can_atomically_transfer_explicit_device_takeover`.
+
+## host-agent-protocol
+
+- Host authenticates as itself: `apps/device-host-agent/tests/test_client.py::test_client_sends_authenticated_heartbeat_and_claim` and `tests/test_cloud_auth.py::test_host_credential_authenticates_as_one_bound_host`.
+- Host attempts to impersonate another host: `tests/test_host_agent_internal_api.py::test_host_token_cannot_submit_heartbeat_for_another_host`.
+- Host reports devices: `apps/device-host-agent/tests/test_heartbeat.py::test_build_device_snapshot_copies_complete_non_secret_state` and `tests/test_host_agent_internal_api.py::test_authenticated_heartbeat_replaces_complete_snapshot`.
+- Host reports no devices: `tests/test_host_agent_internal_api.py::test_empty_heartbeat_removes_only_reporting_hosts_devices`.
+- Assigned work is available: `tests/test_host_agent_internal_api.py::test_long_poll_claim_returns_at_most_one_owned_assignment`.
+- No work becomes available: `tests/test_host_agent_internal_api.py::test_empty_long_poll_timeout_is_normal_response`.
+- Lease renewal succeeds: `tests/test_host_agent_internal_api.py::test_lease_renewal_extends_active_assignment` and `apps/device-host-agent/tests/test_lease.py::test_lease_renews_while_execution_is_active`.
+- Lease is stale or foreign: `tests/test_host_agent_internal_api.py::test_stale_renewal_returns_typed_conflict` and `apps/device-host-agent/tests/test_lease.py::test_stale_lease_stops_later_interruptible_actions`.
+- Execute a goal assignment: `apps/device-host-agent/tests/test_assignment.py::test_goal_assignment_executes_through_task_runner` and `apps/device-host-agent/tests/test_e2e.py::test_public_sdk_reports_fake_device_success_and_runtime_failure`.
+- Execute a workflow assignment: `apps/device-host-agent/tests/test_assignment.py::test_workflow_assignment_loads_and_executes_definition`.
+- Report a successful result: `tests/test_host_agent_internal_api.py::test_terminal_result_is_idempotent_through_internal_api` and `apps/device-host-agent/tests/test_e2e.py::test_public_sdk_reports_fake_device_success_and_runtime_failure`.
+- Retry a result after response loss: `apps/device-host-agent/tests/test_client.py::test_result_report_retries_identical_payload_after_response_loss`.
+- Stale attempt reports after requeue: `tests/test_host_agent_internal_api.py::test_superseded_attempt_result_cannot_overwrite_current_lease`.
+- Host is behind NAT: `apps/device-host-agent/tests/test_e2e.py::test_nat_style_host_requires_only_outbound_requests`.
+
+## platform-sdk
+
+- Production starts without configured credentials: `apps/cloud-api/tests/test_app.py::test_production_app_rejects_missing_credentials`.
+- Explicit local anonymous override: `tests/test_cloud_control_config.py::test_auth_provider_requires_explicit_anonymous_override`.
+- Custom AuthProvider is honored: `tests/test_cloud_sdk_api.py::test_rejecting_auth_provider_blocks_every_route`.
+- Submit token has task scope: `tests/test_cloud_sdk_api.py::test_every_public_route_enforces_its_scope`.
+- Non-admin token attempts plugin registration: `tests/test_cloud_sdk_api.py::test_plugin_admin_scope_is_checked_before_registration`.
+- Query an active remote task: `tests/test_cloud_sdk_api.py::test_task_status_exposes_distributed_metadata_without_lease_secret`.
+- Query a failed remote task: `tests/test_cloud_sdk_api.py::test_task_status_exposes_distributed_metadata_without_lease_secret` and `apps/device-host-agent/tests/test_e2e.py::test_public_sdk_reports_fake_device_success_and_runtime_failure`.
+- Client configured with bearer token: `tests/test_cloud_client.py::test_client_applies_bearer_token_to_every_public_method`.
+- Client receives authorization failure: `tests/test_cloud_client.py::test_client_raises_typed_authorization_error_without_exposing_token`.
+
+## task-scheduler
+
+- Scheduler assigns an idle device: `tests/test_cloud_repository_contract.py::test_atomic_assignment_creates_lease_attempt_and_reservation`.
+- Later scheduler iteration sees stale idle snapshot: `tests/test_cloud_repository_contract.py::test_active_assignment_blocks_reuse_of_stale_idle_snapshot`.
+- Owning host claims once: `tests/test_cloud_repository_contract.py::test_owning_host_claims_one_active_assignment`.
+- Concurrent claims race: `tests/test_cloud_repository_concurrency.py::test_postgresql_concurrent_claim_has_one_winner`.
+- Lease expires with attempts remaining: `tests/test_cloud_repository_contract.py::test_expired_lease_requeues_with_auditable_history`.
+- Lease expires at attempt limit: `tests/test_cloud_repository_contract.py::test_expired_lease_fails_at_attempt_limit`.
+- Active lease reports completion: `tests/test_cloud_repository_contract.py::test_terminal_result_is_recorded_idempotently_and_releases_reservation`.
+- Superseded lease reports completion: `tests/test_cloud_repository_contract.py::test_stale_foreign_or_expired_result_is_rejected`.

packages/cloud-platform/cloud/internal_api/api.py

@@ -72,7 +72,11 @@ def create_internal_router(
         request: Request,
     ) -> HeartbeatResponse:
         authorize_host(request, host_id)
-        _validate_snapshot(pool, host_id=host_id, payload=payload)
+        allow_device_takeover = _validate_snapshot(
+            pool,
+            host_id=host_id,
+            payload=payload,
+        )
         devices = [
             Device(
                 id=device.device_id,
@@ -82,7 +86,12 @@ def create_internal_router(
             )
             for device in payload.devices
         ]
-        pool.sync_host_devices(host_id, devices, address=payload.address)
+        pool.sync_host_devices(
+            host_id,
+            devices,
+            address=payload.address,
+            allow_device_takeover=allow_device_takeover,
+        )
         return HeartbeatResponse(
             host_id=host_id,
             accepted_devices=len(devices),
@@ -228,7 +237,7 @@ def _validate_snapshot(
     *,
     host_id: str,
     payload: HeartbeatRequest,
-) -> None:
+) -> bool:
     if payload.host_id != host_id:
         raise HTTPException(
             status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
@@ -244,6 +253,7 @@ def _validate_snapshot(
     now = utc_now()
     hosts = {host.host_id: host for host in pool.store.list_hosts()}
     conflicts: list[str] = []
+    stale_owner_found = False
     requested_ids = set(device_ids)
     for device in pool.store.list_devices():
         if device.device_id not in requested_ids or device.host_id == host_id:
@@ -254,8 +264,11 @@ def _validate_snapshot(
         age_seconds = (now - owner.last_seen_at).total_seconds()
         if age_seconds <= pool.config.stale_after_seconds:
             conflicts.append(device.device_id)
+        else:
+            stale_owner_found = True
     if conflicts:
         raise HTTPException(
             status_code=status.HTTP_409_CONFLICT,
             detail=f"device ownership conflict: {sorted(set(conflicts))}",
         )
+    return stale_owner_found

packages/cloud-platform/cloud/pool.py

@@ -61,6 +61,7 @@ class DevicePool:
         snapshot: list[Device],
         *,
         address: str | None = None,
+        allow_device_takeover: bool = False,
     ) -> None:
         """Push a host's current device snapshot into the pool.
 
@@ -71,7 +72,14 @@ class DevicePool:
         now = utc_now()
         self.store.upsert_host(host_id, address=address, last_seen_at=now)
         devices = [self._to_pooled(device, host_id, now) for device in snapshot]
-        self.store.replace_host_devices(host_id, devices)
+        if allow_device_takeover:
+            self.store.replace_host_devices(
+                host_id,
+                devices,
+                allow_device_takeover=True,
+            )
+        else:
+            self.store.replace_host_devices(host_id, devices)
 
     def list_devices(self) -> list[PooledDevice]:
         devices = self.store.list_devices()
@@ -105,7 +113,9 @@ class DevicePool:
         host_id: str,
         synced_at: datetime,
     ) -> PooledDevice:
-        raw_status = device.status if device.status in _HOST_REPORTED_STATUSES else "idle"
+        raw_status = (
+            device.status if device.status in _HOST_REPORTED_STATUSES else "idle"
+        )
         tags = list(device.capability_tags or [])
         return PooledDevice(
             device_id=device.id,

packages/cloud-platform/cloud/repository.py

@@ -58,6 +58,8 @@ class CloudRepository(Protocol):
         self,
         host_id: str,
         devices: list[PooledDevice],
+        *,
+        allow_device_takeover: bool = False,
     ) -> None: ...
 
     def list_hosts(self) -> list[HostRegistration]: ...

packages/cloud-platform/cloud/sql_repository.py

@@ -59,11 +59,22 @@ class SQLAlchemyCloudRepository:
         self,
         host_id: str,
         devices: list[Any],
+        *,
+        allow_device_takeover: bool = False,
     ) -> None:
         with self._sessions.begin() as session:
             session.execute(
                 delete(PooledDeviceRow).where(PooledDeviceRow.host_id == host_id)
             )
+            if allow_device_takeover:
+                device_ids = [device.device_id for device in devices]
+                if device_ids:
+                    session.execute(
+                        delete(PooledDeviceRow).where(
+                            PooledDeviceRow.host_id != host_id,
+                            PooledDeviceRow.device_id.in_(device_ids),
+                        )
+                    )
             session.add_all(
                 [
                     PooledDeviceRow(

tests/test_cloud_repository_contract.py

@@ -123,6 +123,37 @@ def test_repository_crud_contract(database_url: str) -> None:
         database.close()
 
 
+def test_repository_can_atomically_transfer_explicit_device_takeover(
+    database_url: str,
+) -> None:
+    database = CloudDatabase(database_url)
+    repository = database.repository
+    old_host_id = _unique_id("old-host")
+    new_host_id = _unique_id("new-host")
+    device_id = _unique_id("shared-device")
+
+    try:
+        repository.replace_host_devices(
+            old_host_id,
+            [_device(device_id, old_host_id)],
+        )
+        repository.replace_host_devices(
+            new_host_id,
+            [_device(device_id, new_host_id)],
+            allow_device_takeover=True,
+        )
+
+        matching = [
+            device
+            for device in repository.list_devices()
+            if device.device_id == device_id
+        ]
+        assert len(matching) == 1
+        assert matching[0].host_id == new_host_id
+    finally:
+        database.close()
+
+
 def test_failed_snapshot_transaction_rolls_back(database_url: str) -> None:
     database = CloudDatabase(database_url)
     repository = database.repository

tests/test_host_agent_internal_api.py

@@ -77,6 +77,33 @@ def test_authenticated_heartbeat_replaces_complete_snapshot(tmp_path) -> None:
     }
 
 
+def test_empty_heartbeat_removes_only_reporting_hosts_devices(tmp_path) -> None:
+    client, pool = _build_client(tmp_path)
+    client.put(
+        "/internal/v1/hosts/host-a/heartbeat",
+        headers={"Authorization": "Bearer token-a"},
+        json=_heartbeat_payload("host-a", "device-a"),
+    )
+    client.put(
+        "/internal/v1/hosts/host-b/heartbeat",
+        headers={"Authorization": "Bearer token-b"},
+        json=_heartbeat_payload("host-b", "device-b"),
+    )
+
+    response = client.put(
+        "/internal/v1/hosts/host-a/heartbeat",
+        headers={"Authorization": "Bearer token-a"},
+        json=_heartbeat_payload("host-a"),
+    )
+
+    assert response.status_code == 200
+    assert response.json()["accepted_devices"] == 0
+    devices = pool.store.list_devices()
+    assert [(device.device_id, device.host_id) for device in devices] == [
+        ("device-b", "host-b")
+    ]
+
+
 def test_invalid_duplicate_snapshot_preserves_previous_devices(tmp_path) -> None:
     client, pool = _build_client(tmp_path)
     client.put(
@@ -127,6 +154,35 @@ def test_live_device_owner_conflict_is_rejected_without_partial_sync(tmp_path) -
     assert devices[0].host_id == "host-a"
 
 
+def test_stale_device_owner_can_be_replaced_by_live_host(tmp_path) -> None:
+    client, pool = _build_client(tmp_path)
+    stale_at = datetime.now(UTC) - timedelta(seconds=61)
+    pool.store.upsert_host("host-a", address=None, last_seen_at=stale_at)
+    pool.store.replace_host_devices(
+        "host-a",
+        [
+            PooledDevice(
+                device_id="shared-device",
+                host_id="host-a",
+                driver_type="wda",
+                status="idle",
+                synced_at=stale_at,
+            )
+        ],
+    )
+
+    response = client.put(
+        "/internal/v1/hosts/host-b/heartbeat",
+        headers={"Authorization": "Bearer token-b"},
+        json=_heartbeat_payload("host-b", "shared-device"),
+    )
+
+    assert response.status_code == 200
+    device = pool.store.list_devices()[0]
+    assert device.device_id == "shared-device"
+    assert device.host_id == "host-b"
+
+
 def test_host_token_cannot_submit_heartbeat_for_another_host(tmp_path) -> None:
     client, pool = _build_client(tmp_path)