test(cloud): verify public execution outcomes

apps/device-host-agent/tests/test_e2e.py

@@ -8,23 +8,35 @@ from pathlib import Path
 
 import httpx
 import pytest
+from fastapi.testclient import TestClient
 
 from cloud.auth import BearerCredential
 from cloud.control_config import CloudControlConfig
+from cloud.sdk.client import CloudClient
 from cloud.scheduler import TaskConstraints
 from cloud_api.app import create_app
+from core.models import Scene
 from device.manager import DeviceManager
 from driver.base import Driver
-from host_agent.assignment import AssignmentExecutionResult
+from host_agent.assignment import AssignmentExecutionResult, AssignmentExecutor
 from host_agent.client import HostAgentClient, StaleLeaseError
 from host_agent.config import HostAgentConfig
+from host_agent.execution import ExecutionFactories
 from host_agent.heartbeat import HeartbeatSynchronizer
+from host_agent.lease import ActiveAssignmentRunner
 from host_agent.processor import AssignmentProcessor
+from runtime.executor import Executor, default_tool_registry
+from runtime.planner import PlannedStep, Planner
+from runtime.task import TaskRunner, TaskRunnerConfig
+from workflow.store import WorkflowStore
 
 
 class FakeDriver(Driver):
+    def __init__(self) -> None:
+        self.calls: list[tuple[str, tuple[object, ...]]] = []
+
     def connect(self) -> None:
-        return None
+        self.calls.append(("connect", ()))
 
     def disconnect(self) -> None:
         return None
@@ -33,7 +45,7 @@ class FakeDriver(Driver):
         return b""
 
     def tap(self, x: float, y: float) -> None:
-        return None
+        self.calls.append(("tap", (x, y)))
 
     def swipe(
         self,
@@ -76,6 +88,14 @@ def _credential(host_id: str) -> BearerCredential:
     )
 
 
+def _public_credential() -> BearerCredential:
+    return BearerCredential(
+        principal_id="sdk",
+        token="public-token",
+        scopes=frozenset({"tasks:submit", "tasks:read"}),
+    )
+
+
 def _config(host_id: str) -> HostAgentConfig:
     return HostAgentConfig(
         control_plane_url="http://control.test",
@@ -171,6 +191,58 @@ class _SuccessfulExecution:
         return None
 
 
+class _AsyncTestClient:
+    def __init__(self, client: TestClient) -> None:
+        self.client = client
+
+    async def request(self, method: str, path: str, **kwargs) -> httpx.Response:
+        kwargs.pop("timeout", None)
+        return await asyncio.to_thread(self.client.request, method, path, **kwargs)
+
+
+class _TapPlanner(Planner):
+    def plan(self, *, goal, scene, context):
+        if context.step_results:
+            return []
+        return [PlannedStep(action="tap", description="tap", args={"x": 2, "y": 3})]
+
+    def goal_reached(self, *, goal, scene, context):
+        return bool(context.step_results)
+
+
+class _FailingPlanner(Planner):
+    def plan(self, *, goal, scene, context):
+        raise RuntimeError("planner unavailable")
+
+
+def _assignment_processor(
+    tmp_path: Path,
+    client: HostAgentClient,
+    manager: DeviceManager,
+    *,
+    planner: Planner,
+) -> AssignmentProcessor:
+    scene = Scene(width=10, height=20, elements=[])
+
+    def task_runner_factory() -> TaskRunner:
+        return TaskRunner(
+            planner=planner,
+            executor=Executor(tools=default_tool_registry(manager=manager)),
+            observer=lambda device_id: scene,
+            screenshot_provider=lambda device_id: b"",
+            config=TaskRunnerConfig(max_steps=3),
+        )
+
+    workflow_store = WorkflowStore(tmp_path / "workflows.sqlite3")
+    factories = ExecutionFactories(
+        task_runner_factory=task_runner_factory,
+        workflow_runner_factory=lambda: object(),  # type: ignore[arg-type,return-value]
+        workflow_store=workflow_store,
+    )
+    executor = AssignmentExecutor(factories)
+    return AssignmentProcessor(client, ActiveAssignmentRunner(client, executor))
+
+
 def test_one_host_executes_assignment_through_outbound_protocol(tmp_path) -> None:
     async def scenario() -> None:
         paths: list[str] = []
@@ -323,3 +395,71 @@ def test_lease_loss_rejects_stale_host_result(tmp_path) -> None:
                 assert task.terminal_result is None
 
     asyncio.run(scenario())
+
+
+def test_public_sdk_reports_fake_device_success_and_runtime_failure(tmp_path) -> None:
+    app = create_app(
+        config=CloudControlConfig(
+            environment="test",
+            database_url=f"sqlite:///{(tmp_path / 'sdk-e2e.sqlite3').as_posix()}",
+            scheduler_interval_seconds=60,
+            lease_reaper_interval_seconds=60,
+            lease_duration_seconds=30,
+            credentials=(_public_credential(), _credential("host-a")),
+        )
+    )
+    driver = FakeDriver()
+    manager = DeviceManager()
+    manager.register_device("device-a", lambda: driver, status="idle")
+
+    with TestClient(app) as http_client:
+        cloud_client = CloudClient(
+            "http://testserver",
+            http_client=http_client,
+            token="public-token",
+        )
+
+        async def scenario() -> None:
+            host_client = HostAgentClient(
+                _config("host-a"),
+                http_client=_AsyncTestClient(http_client),  # type: ignore[arg-type]
+            )
+            heartbeat = HeartbeatSynchronizer(manager, host_client, _config("host-a"))
+            heartbeat.connect_devices()
+            await heartbeat.sync_once()
+
+            successful_task_id = cloud_client.submit_task(goal="tap screen")[
+                "task_id"
+            ]
+            app.state.cloud_services.scheduler.assign()
+            successful_assignment = await host_client.claim()
+            assert successful_assignment is not None
+            await _assignment_processor(
+                tmp_path / "success",
+                host_client,
+                manager,
+                planner=_TapPlanner(),
+            ).process(successful_assignment)
+
+            successful_status = cloud_client.get_task_status(successful_task_id)
+            assert successful_status["status"] == "done"
+            assert ("tap", (2, 3)) in driver.calls
+
+            failed_task_id = cloud_client.submit_task(goal="fail planning")["task_id"]
+            app.state.cloud_services.scheduler.assign()
+            failed_assignment = await host_client.claim()
+            assert failed_assignment is not None
+            await _assignment_processor(
+                tmp_path / "failure",
+                host_client,
+                manager,
+                planner=_FailingPlanner(),
+            ).process(failed_assignment)
+
+            failed_status = cloud_client.get_task_status(failed_task_id)
+            assert failed_status["status"] == "failed"
+            assert failed_status["failure_reason"] == (
+                "RuntimeError: planner unavailable"
+            )
+
+        asyncio.run(scenario())

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

@@ -72,7 +72,7 @@
 ## 9. Verification And Project Records
 
 - [x] 9.1 Run formatting, static checks, all non-integration tests, and targeted PostgreSQL integration/concurrency tests.
-- [ ] 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.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.
 - [ ] 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.
 - [ ] 9.5 Update the project index, architecture/deployment documentation, and runtime maturity memory after implementation verification.