agentic-mobile-control/apps/device-host-agent/host_agent/cloud_planner_client.py

"""``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 host_agent.planner_context import current_planner_execution_context
from runtime.tool_calling_client import (
    ToolCallDecision,
    ToolCallUnavailable,
    ToolCallUsage,
)
from runtime.tool_specs import ToolSpec


_CLOUD_PROVIDER_MAX_TIMEOUT_SECONDS = 120.0
_CLOUD_PROXY_TRANSPORT_GRACE_SECONDS = 5.0
_CLOUD_PROXY_HTTP_TIMEOUT_SECONDS = (
    _CLOUD_PROVIDER_MAX_TIMEOUT_SECONDS + _CLOUD_PROXY_TRANSPORT_GRACE_SECONDS
)


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,
        history: list[dict[str, Any]] | None = None,
    ) -> ToolCallDecision:
        payload: dict[str, Any] = {
            "host_id": self.config.host_id,
            "system_prompt": system_prompt,
            "user_prompt": user_prompt,
            "history": history or [],
            "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
            ],
            # Legacy Cloud API versions use this field. Current Cloud API versions
            # resolve the provider timeout from the active Provider profile.
            "timeout_seconds": min(timeout, _CLOUD_PROVIDER_MAX_TIMEOUT_SECONDS),
        }
        context = current_planner_execution_context()
        if context is not None:
            payload.update(
                {
                    "task_id": context.task_id,
                    "attempt": context.attempt,
                    "lease_id": context.lease_id,
                }
            )
        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=_CLOUD_PROXY_HTTP_TIMEOUT_SECONDS,
            )
        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),
                usage=(
                    ToolCallUsage(
                        input_tokens=decoded.input_tokens,
                        output_tokens=decoded.output_tokens,
                        total_tokens=decoded.total_tokens,
                    )
                    if any(
                        value is not None
                        for value in (
                            decoded.input_tokens,
                            decoded.output_tokens,
                            decoded.total_tokens,
                        )
                    )
                    else None
                ),
                text_output=decoded.rationale,
                thinking=decoded.thinking,
                purpose=decoded.purpose,
                expected_outcome=decoded.expected_outcome,
            )
        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