From 7b717a05a947765f16c9d255e6825f789e2aecce Mon Sep 17 00:00:00 2001 From: Jerry Yan <792602257@qq.com> Date: Mon, 13 Jul 2026 08:21:44 +0800 Subject: [PATCH] test(openspec): map cloud integration scenarios --- .../cloud-control-plane-integration/tasks.md | 2 +- .../verification.md | 71 +++++++++++++++++++ .../cloud-platform/cloud/internal_api/api.py | 19 ++++- packages/cloud-platform/cloud/pool.py | 14 +++- packages/cloud-platform/cloud/repository.py | 2 + .../cloud-platform/cloud/sql_repository.py | 11 +++ tests/test_cloud_repository_contract.py | 31 ++++++++ tests/test_host_agent_internal_api.py | 56 +++++++++++++++ 8 files changed, 200 insertions(+), 6 deletions(-) create mode 100644 openspec/changes/cloud-control-plane-integration/verification.md diff --git a/openspec/changes/cloud-control-plane-integration/tasks.md b/openspec/changes/cloud-control-plane-integration/tasks.md index 0f19d04..aa2412b 100644 --- a/openspec/changes/cloud-control-plane-integration/tasks.md +++ b/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. diff --git a/openspec/changes/cloud-control-plane-integration/verification.md b/openspec/changes/cloud-control-plane-integration/verification.md new file mode 100644 index 0000000..a8b6c36 --- /dev/null +++ b/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`. diff --git a/packages/cloud-platform/cloud/internal_api/api.py b/packages/cloud-platform/cloud/internal_api/api.py index 5b70d91..e2d99c8 100644 --- a/packages/cloud-platform/cloud/internal_api/api.py +++ b/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 diff --git a/packages/cloud-platform/cloud/pool.py b/packages/cloud-platform/cloud/pool.py index c1d0b76..54aaabd 100644 --- a/packages/cloud-platform/cloud/pool.py +++ b/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, diff --git a/packages/cloud-platform/cloud/repository.py b/packages/cloud-platform/cloud/repository.py index ac08aa6..8f538b0 100644 --- a/packages/cloud-platform/cloud/repository.py +++ b/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]: ... diff --git a/packages/cloud-platform/cloud/sql_repository.py b/packages/cloud-platform/cloud/sql_repository.py index 1a0820a..8959724 100644 --- a/packages/cloud-platform/cloud/sql_repository.py +++ b/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( diff --git a/tests/test_cloud_repository_contract.py b/tests/test_cloud_repository_contract.py index abda2cd..3ff10c4 100644 --- a/tests/test_cloud_repository_contract.py +++ b/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 diff --git a/tests/test_host_agent_internal_api.py b/tests/test_host_agent_internal_api.py index 85ce2ec..ebbc090 100644 --- a/tests/test_host_agent_internal_api.py +++ b/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)