Implement cloud-planner-proxy: AI planner routes through Cloud API

Implements all 19 tasks of the cloud-planner-proxy OpenSpec change:

- Cloud API: cloud.planner_config (CloudPlannerConfig, load/build helpers)
  reusing runtime.tool_calling_client provider clients (no new dependency
  needed -- device-cloud-platform already depends on device-agent-runtime).
- Cloud API: new host-scoped POST /internal/v1/hosts/{host_id}/planner/decide
  internal endpoint, reusing existing bearer auth; logs only metadata
  (host id, tool name, latency, error class), never prompt/screenshot
  content.
- Host Agent: new AI_PLANNER_TRANSPORT config (direct default | cloud) and
  host_agent/cloud_planner_client.py::CloudProxyToolCallingClient, a
  synchronous ToolCallingClient implementation (structural, not importing
  runtime) that calls the new endpoint via its own httpx.Client -- avoids
  bridging the async HostAgentClient across the worker-thread boundary
  that AIPlanner.plan() runs in (asyncio.to_thread in lease.py).
- Host Agent wiring: create_execution_factories()/_host_agent_planner()
  select the cloud-proxy client only when AI_PLANNER_TRANSPORT=cloud;
  direct/unset transport is unchanged (still the default).
- Tests: 22 new tests across Cloud API config, the new endpoint, the new
  client, and transport-selection wiring; full non-integration suite
  (492 tests) passes with no regressions.
- Docs: docs/CLOUD_DEPLOYMENT.md documents the cloud transport, its
  trade-offs, and the credential split between Host Agent and Cloud API.

proposal.md/design.md were corrected during implementation to reflect two
findings: no new anthropic/openai dependency is actually needed, and
CloudProxyToolCallingClient uses its own sync httpx.Client rather than a
new HostAgentClient method, per the thread-boundary reasoning above.

apps/device-host-agent/host_agent/app.py

@@ -196,7 +196,12 @@ def create_application(
             else None
         ),
     )
-    executor = AssignmentExecutor(create_execution_factories(resolved_manager))
+    executor = AssignmentExecutor(
+        create_execution_factories(
+            resolved_manager,
+            host_agent_config=resolved_config,
+        )
+    )
     active_runner = ActiveAssignmentRunner(client, executor)
     processor = AssignmentProcessor(
         client,

apps/device-host-agent/host_agent/cloud_planner_client.py

@@ -0,0 +1,109 @@
+"""``ToolCallingClient`` implementation that proxies AI Planner decisions
+through the Cloud Control Plane instead of calling an LLM provider directly.
+
+Lives in ``host_agent``, not ``runtime`` (design decision D3): the shared
+Runtime package's boundary test
+(``apps/device-host-agent/tests/test_execution.py::test_runtime_owned_packages_do_not_import_host_or_cloud_concerns``)
+forbids ``runtime`` from importing ``cloud`` or ``host_agent``, so this class
+satisfies ``runtime.tool_calling_client.ToolCallingClient`` structurally
+from outside that package instead of living inside it.
+
+Uses its own synchronous ``httpx.Client`` (mirroring
+``HostAgentEnrollmentClient``'s pattern) rather than wrapping the async
+``HostAgentClient``: ``ToolCallingClient.decide()`` is a synchronous Protocol
+method invoked from a worker thread via ``asyncio.to_thread`` (see
+``host_agent/lease.py``), off the main event loop, so reusing an
+``httpx.AsyncClient`` bound to that loop would require event-loop bridging
+for no real benefit over a plain synchronous client with the same
+host-scoped bearer auth.
+"""
+
+from __future__ import annotations
+
+import base64
+from typing import Any
+
+import httpx
+
+from cloud.internal_api.models import PlannerDecisionError, PlannerDecisionResponse
+from host_agent.config import HostAgentConfig
+from runtime.tool_calling_client import ToolCallDecision, ToolCallUnavailable
+from runtime.tool_specs import ToolSpec
+
+
+class CloudProxyToolCallingClient:
+    def __init__(
+        self,
+        config: HostAgentConfig,
+        *,
+        http_client: httpx.Client | None = None,
+    ) -> None:
+        self.config = config
+        self._owns_client = http_client is None
+        self._client = http_client or httpx.Client(base_url=config.control_plane_url)
+
+    def decide(
+        self,
+        *,
+        system_prompt: str,
+        user_prompt: str,
+        screenshot: bytes | None,
+        tools: list[ToolSpec],
+        timeout: float,
+    ) -> ToolCallDecision:
+        payload: dict[str, Any] = {
+            "host_id": self.config.host_id,
+            "system_prompt": system_prompt,
+            "user_prompt": user_prompt,
+            "screenshot_base64": (
+                base64.b64encode(screenshot).decode("ascii")
+                if screenshot is not None
+                else None
+            ),
+            "tools": [
+                {
+                    "name": spec.name,
+                    "description": spec.description,
+                    "parameters": spec.parameters,
+                }
+                for spec in tools
+            ],
+            "timeout_seconds": timeout,
+        }
+        try:
+            response = self._client.post(
+                f"/internal/v1/hosts/{self.config.host_id}/planner/decide",
+                json=payload,
+                headers={"Authorization": f"Bearer {self.config.token}"},
+                timeout=timeout + 5,
+            )
+        except httpx.HTTPError as exc:
+            raise ToolCallUnavailable(str(exc)) from exc
+
+        if response.is_success:
+            decoded = PlannerDecisionResponse.model_validate(response.json())
+            return ToolCallDecision(
+                tool_name=decoded.tool_name,
+                arguments=dict(decoded.arguments),
+            )
+        raise ToolCallUnavailable(_error_detail(response))
+
+    def close(self) -> None:
+        if self._owns_client:
+            self._client.close()
+
+
+def _error_detail(response: httpx.Response) -> str:
+    try:
+        payload = response.json()
+    except ValueError:
+        return f"planner-decision request failed with status {response.status_code}"
+    try:
+        error = PlannerDecisionError.model_validate(payload)
+    except Exception:
+        detail = payload.get("detail") if isinstance(payload, dict) else None
+        return (
+            detail
+            or f"planner-decision request failed with status {response.status_code}"
+        )
+    return error.detail

apps/device-host-agent/host_agent/config.py

@@ -12,6 +12,7 @@ class HostAgentConfigurationError(ValueError):
 
 
 _LOOPBACK_BIND_HOSTS = frozenset({"127.0.0.1", "localhost", "::1"})
+_AI_PLANNER_TRANSPORTS = frozenset({"direct", "cloud"})
 
 
 @dataclass(frozen=True)
@@ -34,6 +35,7 @@ class HostAgentConfig:
     console_allow_non_loopback: bool = False
     console_session_ttl_seconds: float = 43200.0
     console_history_limit: int = 200
+    ai_planner_transport: str = "direct"
 
 
 def load_host_agent_config(
@@ -112,6 +114,9 @@ def load_host_agent_config(
             "HOST_AGENT_CONSOLE_HISTORY_LIMIT",
             200,
         ),
+        ai_planner_transport=_parse_ai_planner_transport(
+            values.get("AI_PLANNER_TRANSPORT")
+        ),
     )
     if config.max_retry_backoff_seconds < config.retry_backoff_seconds:
         raise HostAgentConfigurationError(
@@ -129,6 +134,18 @@ def load_host_agent_config(
     return config
 
 
+def _parse_ai_planner_transport(value: str | None) -> str:
+    if value is None:
+        return "direct"
+    transport = value.strip().lower()
+    if transport not in _AI_PLANNER_TRANSPORTS:
+        raise HostAgentConfigurationError(
+            "AI_PLANNER_TRANSPORT must be one of "
+            f"{', '.join(sorted(_AI_PLANNER_TRANSPORTS))}"
+        )
+    return transport
+
+
 def _positive_float(
     values: Mapping[str, str],
     name: str,

apps/device-host-agent/host_agent/execution.py

@@ -5,7 +5,11 @@ from collections.abc import Callable
 from dataclasses import dataclass, replace
 
 from device.manager import DeviceManager
+from host_agent.cloud_planner_client import CloudProxyToolCallingClient
+from host_agent.config import HostAgentConfig, load_host_agent_config
+from runtime.ai_planner import AIPlanner
 from runtime.executor import Executor, default_tool_registry
+from runtime.planner import Planner
 from runtime.planner_config import PlannerConfig, load_config as load_planner_config
 from runtime.task import TaskRunner
 from storage.task_metadata import TaskMetadataStore
@@ -27,14 +31,17 @@ def create_execution_factories(
     workflow_store: WorkflowStore | None = None,
     metadata_store: TaskMetadataStore | None = None,
     timeline: Timeline | None = None,
+    host_agent_config: HostAgentConfig | None = None,
 ) -> ExecutionFactories:
     shared_workflow_store = workflow_store or WorkflowStore()
+    resolved_host_agent_config = host_agent_config
 
     def create_task_runner() -> TaskRunner:
         return TaskRunner(
             executor=Executor(tools=default_tool_registry(manager=manager)),
             metadata_store=metadata_store,
             timeline=timeline,
+            planner=_host_agent_planner(resolved_host_agent_config),
             planner_config=_host_agent_planner_config(),
         )
 
@@ -64,3 +71,28 @@ def _host_agent_planner_config() -> PlannerConfig:
     if os.environ.get("AI_PLANNER_ENABLED") is None:
         config = replace(config, enabled=True)
     return config
+
+
+def _host_agent_planner(
+    host_agent_config: HostAgentConfig | None,
+) -> Planner | None:
+    """Build the `AIPlanner` explicitly when the cloud-proxy transport is
+    selected, so its `ToolCallingClient` is a `CloudProxyToolCallingClient`
+    instead of a local Anthropic/OpenAI SDK client.
+
+    Returns `None` (letting `TaskRunner` fall back to its own
+    `_default_planner()`) for the `direct` transport, which preserves the
+    existing default-enabled/direct-to-provider behavior unchanged.
+    """
+    planner_config = _host_agent_planner_config()
+    if not planner_config.enabled:
+        return None
+
+    resolved_config = host_agent_config or load_host_agent_config()
+    if resolved_config.ai_planner_transport != "cloud":
+        return None
+
+    return AIPlanner(
+        client=CloudProxyToolCallingClient(resolved_config),
+        config=planner_config,
+    )

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

@@ -0,0 +1,152 @@
+from __future__ import annotations
+
+import json
+
+import httpx
+import pytest
+
+from host_agent.cloud_planner_client import CloudProxyToolCallingClient
+from host_agent.config import HostAgentConfig
+from runtime.tool_calling_client import ToolCallDecision, ToolCallUnavailable
+from runtime.tool_specs import ToolSpec
+
+_CONFIG = HostAgentConfig(
+    control_plane_url="https://control-plane.example",
+    host_id="host-a",
+    token="token-a",
+)
+
+_TOOLS = [ToolSpec(name="tap", description="tap an element", parameters={})]
+
+
+def _client(handler) -> CloudProxyToolCallingClient:
+    transport = httpx.MockTransport(handler)
+    http_client = httpx.Client(base_url=_CONFIG.control_plane_url, transport=transport)
+    return CloudProxyToolCallingClient(_CONFIG, http_client=http_client)
+
+
+def test_decide_returns_tool_call_decision_on_success() -> None:
+    seen_requests: list[httpx.Request] = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        seen_requests.append(request)
+        return httpx.Response(
+            200,
+            json={"tool_name": "tap", "arguments": {"x": 1, "y": 2}},
+        )
+
+    client = _client(handler)
+
+    decision = client.decide(
+        system_prompt="you are a planner",
+        user_prompt="tap login",
+        screenshot=None,
+        tools=_TOOLS,
+        timeout=30.0,
+    )
+
+    assert decision == ToolCallDecision(tool_name="tap", arguments={"x": 1, "y": 2})
+    assert len(seen_requests) == 1
+    request = seen_requests[0]
+    assert request.url.path == "/internal/v1/hosts/host-a/planner/decide"
+    assert request.headers["Authorization"] == "Bearer token-a"
+    body = json.loads(request.content)
+    assert body["host_id"] == "host-a"
+    assert body["system_prompt"] == "you are a planner"
+    assert body["screenshot_base64"] is None
+    assert body["timeout_seconds"] == 30.0
+
+
+def test_decide_base64_encodes_screenshot() -> None:
+    seen_requests: list[httpx.Request] = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        seen_requests.append(request)
+        return httpx.Response(200, json={"tool_name": "tap", "arguments": {}})
+
+    client = _client(handler)
+
+    client.decide(
+        system_prompt="sp",
+        user_prompt="up",
+        screenshot=b"hello",
+        tools=_TOOLS,
+        timeout=10.0,
+    )
+
+    body = json.loads(seen_requests[0].content)
+    assert body["screenshot_base64"] == "aGVsbG8="
+
+
+def test_decide_raises_tool_call_unavailable_on_network_error() -> None:
+    def handler(request: httpx.Request) -> httpx.Response:
+        raise httpx.ConnectError("connection refused", request=request)
+
+    client = _client(handler)
+
+    with pytest.raises(ToolCallUnavailable):
+        client.decide(
+            system_prompt="sp",
+            user_prompt="up",
+            screenshot=None,
+            tools=_TOOLS,
+            timeout=5.0,
+        )
+
+
+def test_decide_raises_tool_call_unavailable_on_structured_error_response() -> None:
+    def handler(request: httpx.Request) -> httpx.Response:
+        return httpx.Response(
+            502,
+            json={"code": "planner_unavailable", "detail": "provider timed out"},
+        )
+
+    client = _client(handler)
+
+    with pytest.raises(ToolCallUnavailable, match="provider timed out"):
+        client.decide(
+            system_prompt="sp",
+            user_prompt="up",
+            screenshot=None,
+            tools=_TOOLS,
+            timeout=5.0,
+        )
+
+
+def test_decide_raises_tool_call_unavailable_on_unstructured_error_response() -> None:
+    def handler(request: httpx.Request) -> httpx.Response:
+        return httpx.Response(500, text="internal server error")
+
+    client = _client(handler)
+
+    with pytest.raises(ToolCallUnavailable):
+        client.decide(
+            system_prompt="sp",
+            user_prompt="up",
+            screenshot=None,
+            tools=_TOOLS,
+            timeout=5.0,
+        )
+
+
+def test_close_closes_an_internally_constructed_http_client() -> None:
+    client = CloudProxyToolCallingClient(_CONFIG)
+    client.close()
+
+    assert client._client.is_closed
+
+
+def test_close_leaves_an_externally_supplied_http_client_open() -> None:
+    """Only a client built internally (no ``http_client`` override) is closed
+    by ``close()``; a caller-supplied ``http_client`` is left open for the
+    caller to manage."""
+    transport = httpx.MockTransport(lambda request: httpx.Response(200, json={}))
+    external_http_client = httpx.Client(
+        base_url=_CONFIG.control_plane_url, transport=transport
+    )
+
+    client = CloudProxyToolCallingClient(_CONFIG, http_client=external_http_client)
+    client.close()
+
+    assert not external_http_client.is_closed
+    external_http_client.close()

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

@@ -5,6 +5,8 @@ from pathlib import Path
 import pytest
 
 from device.manager import DeviceManager
+from host_agent.cloud_planner_client import CloudProxyToolCallingClient
+from host_agent.config import HostAgentConfig
 from host_agent.execution import create_execution_factories
 from runtime.ai_planner import AIPlanner
 from runtime.planner import Planner
@@ -58,6 +60,54 @@ def test_created_task_runner_honors_explicit_ai_planner_opt_out(
     assert type(task_runner.planner) is Planner
 
 
+def test_cloud_transport_builds_ai_planner_with_cloud_proxy_client(
+    tmp_path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.delenv("AI_PLANNER_ENABLED", raising=False)
+    manager = DeviceManager()
+    host_agent_config = HostAgentConfig(
+        control_plane_url="https://control-plane.example",
+        host_id="host-a",
+        token="token-a",
+        ai_planner_transport="cloud",
+    )
+    factories = create_execution_factories(
+        manager,
+        workflow_store=WorkflowStore(tmp_path / "workflows.sqlite3"),
+        host_agent_config=host_agent_config,
+    )
+
+    task_runner = factories.task_runner_factory()
+
+    assert isinstance(task_runner.planner, AIPlanner)
+    assert isinstance(task_runner.planner.client, CloudProxyToolCallingClient)
+    assert task_runner.planner.client.config is host_agent_config
+
+
+@pytest.mark.parametrize("transport", [None, "direct"])
+def test_direct_transport_preserves_existing_local_provider_construction(
+    tmp_path, monkeypatch: pytest.MonkeyPatch, transport: str | None
+) -> None:
+    monkeypatch.delenv("AI_PLANNER_ENABLED", raising=False)
+    manager = DeviceManager()
+    host_agent_config = HostAgentConfig(
+        control_plane_url="https://control-plane.example",
+        host_id="host-a",
+        token="token-a",
+        **({} if transport is None else {"ai_planner_transport": transport}),
+    )
+    factories = create_execution_factories(
+        manager,
+        workflow_store=WorkflowStore(tmp_path / "workflows.sqlite3"),
+        host_agent_config=host_agent_config,
+    )
+
+    task_runner = factories.task_runner_factory()
+
+    assert isinstance(task_runner.planner, AIPlanner)
+    assert not isinstance(task_runner.planner.client, CloudProxyToolCallingClient)
+
+
 def test_runtime_owned_packages_do_not_import_host_or_cloud_concerns() -> None:
     root = Path(__file__).resolve().parents[3]
     forbidden = ("import cloud", "from cloud", "import host_agent", "from host_agent")