diff --git a/apps/device-host-agent/tests/test_e2e.py b/apps/device-host-agent/tests/test_e2e.py index ed64a28..ce92bdc 100644 --- a/apps/device-host-agent/tests/test_e2e.py +++ b/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()) diff --git a/openspec/changes/cloud-control-plane-integration/tasks.md b/openspec/changes/cloud-control-plane-integration/tasks.md index 539f135..72ba577 100644 --- a/openspec/changes/cloud-control-plane-integration/tasks.md +++ b/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.