Compare commits

...
4 Commits
Author SHA1 Message Date
showtan001 9076f8ddb0 feat: discover and add connected iOS devices
Tests / Test apps.device-host-agent.tests.test_mcp_token.test_load_or_create_concurrent_calls_do_not_corrupt failed
2026-09-07 18:33:49 +08:00
showtan001 8c99dc015a feat: add on-demand device screenshots
Tests / Test apps.device-host-agent.tests.test_mcp_token.test_load_or_create_concurrent_calls_do_not_corrupt failed
2026-08-31 10:43:37 +08:00
showtan001 60ee157e97 feat: preserve planner context across task steps
Tests / Test apps.device-host-agent.tests.test_mcp_token.test_load_or_create_concurrent_calls_do_not_corrupt failed
2026-08-30 22:59:19 +08:00
showtan001 dd8df33910 Log task planner conversations locally
Tests / Test apps.device-host-agent.tests.test_mcp_token.test_load_or_create_concurrent_calls_do_not_corrupt failed
2026-08-30 22:38:28 +08:00
24 changed files with 893 additions and 102 deletions
+5
View File
@@ -125,6 +125,11 @@ and final replies in a local SQLite database. View them at
`http://127.0.0.1:8765/conversations`; image bytes are excluded. Set `http://127.0.0.1:8765/conversations`; image bytes are excluded. Set
`HOST_AGENT_CONVERSATION_LOG_PATH` to change the database path. `HOST_AGENT_CONVERSATION_LOG_PATH` to change the database path.
The authenticated `Devices` page has an on-demand `Get screenshot` button for
each connected device. Screenshots are captured only after the operator clicks
the button; the page does not auto-refresh or capture screenshots as part of
heartbeat synchronization.
In local mode, Appium supervision is enabled by default. Host Agent probes In local mode, Appium supervision is enabled by default. Host Agent probes
`/status`, adopts a healthy existing Appium instance, starts Appium when no `/status`, adopts a healthy existing Appium instance, starts Appium when no
listener exists, restarts only processes it started if they crash, and stops listener exists, restarts only processes it started if they crash, and stops
+2 -1
View File
@@ -231,12 +231,14 @@ def create_application(
"MCP token generated at %s", mcp_token_path "MCP token generated at %s", mcp_token_path
) )
mcp_busy_tracker = McpBusyTracker(ttl_seconds=20.0) mcp_busy_tracker = McpBusyTracker(ttl_seconds=20.0)
conversation_log = ConversationLogStore(resolved_config.conversation_log_path) if resolved_config.mode == "local" else None
executor = AssignmentExecutor( executor = AssignmentExecutor(
create_execution_factories( create_execution_factories(
resolved_manager, resolved_manager,
metadata_store=metadata_store, metadata_store=metadata_store,
timeline=timeline, timeline=timeline,
host_agent_config=resolved_config, host_agent_config=resolved_config,
conversation_log=conversation_log,
), ),
mcp_busy_tracker=mcp_busy_tracker, mcp_busy_tracker=mcp_busy_tracker,
) )
@@ -246,7 +248,6 @@ def create_application(
status_tracker=status_tracker, status_tracker=status_tracker,
) )
planner_config = load_planner_config() planner_config = load_planner_config()
conversation_log = ConversationLogStore(resolved_config.conversation_log_path) if resolved_config.mode == "local" else None
conversation_agent = ( conversation_agent = (
ConversationAgent( ConversationAgent(
config=planner_config, config=planner_config,
@@ -62,11 +62,13 @@ class CloudProxyToolCallingClient:
screenshot: bytes | None, screenshot: bytes | None,
tools: list[ToolSpec], tools: list[ToolSpec],
timeout: float, timeout: float,
history: list[dict[str, Any]] | None = None,
) -> ToolCallDecision: ) -> ToolCallDecision:
payload: dict[str, Any] = { payload: dict[str, Any] = {
"host_id": self.config.host_id, "host_id": self.config.host_id,
"system_prompt": system_prompt, "system_prompt": system_prompt,
"user_prompt": user_prompt, "user_prompt": user_prompt,
"history": history or [],
"screenshot_base64": ( "screenshot_base64": (
base64.b64encode(screenshot).decode("ascii") base64.b64encode(screenshot).decode("ascii")
if screenshot is not None if screenshot is not None
+10 -2
View File
@@ -3,6 +3,7 @@ from __future__ import annotations
import os import os
from collections.abc import Callable from collections.abc import Callable
from dataclasses import dataclass, replace from dataclasses import dataclass, replace
from typing import Any
from device.manager import DeviceManager from device.manager import DeviceManager
from host_agent.cloud_planner_client import CloudProxyToolCallingClient from host_agent.cloud_planner_client import CloudProxyToolCallingClient
@@ -35,6 +36,7 @@ def create_execution_factories(
metadata_store: TaskMetadataStore | None = None, metadata_store: TaskMetadataStore | None = None,
timeline: Timeline | None = None, timeline: Timeline | None = None,
host_agent_config: HostAgentConfig | None = None, host_agent_config: HostAgentConfig | None = None,
conversation_log: Any | None = None,
) -> ExecutionFactories: ) -> ExecutionFactories:
shared_workflow_store = workflow_store or WorkflowStore() shared_workflow_store = workflow_store or WorkflowStore()
resolved_host_agent_config = host_agent_config resolved_host_agent_config = host_agent_config
@@ -48,7 +50,10 @@ def create_execution_factories(
), ),
metadata_store=metadata_store, metadata_store=metadata_store,
timeline=timeline, timeline=timeline,
planner=_host_agent_planner(resolved_host_agent_config), planner=_host_agent_planner(
resolved_host_agent_config,
event_logger=conversation_log.append if conversation_log is not None else None,
),
planner_config=_host_agent_planner_config(), planner_config=_host_agent_planner_config(),
device_platform_provider=lambda device_id: _device_platform( device_platform_provider=lambda device_id: _device_platform(
manager, device_id manager, device_id
@@ -86,6 +91,8 @@ def _host_agent_planner_config() -> PlannerConfig:
def _host_agent_planner( def _host_agent_planner(
host_agent_config: HostAgentConfig | None, host_agent_config: HostAgentConfig | None,
*,
event_logger: Callable[[dict[str, Any]], None] | None = None,
) -> Planner | None: ) -> Planner | None:
"""Build the `AIPlanner` explicitly when the cloud-proxy transport is """Build the `AIPlanner` explicitly when the cloud-proxy transport is
selected, so its `ToolCallingClient` is a `CloudProxyToolCallingClient` selected, so its `ToolCallingClient` is a `CloudProxyToolCallingClient`
@@ -100,11 +107,12 @@ def _host_agent_planner(
resolved_config = host_agent_config or load_host_agent_config() resolved_config = host_agent_config or load_host_agent_config()
if resolved_config.ai_planner_transport != "cloud": if resolved_config.ai_planner_transport != "cloud":
return None return AIPlanner(config=planner_config, event_logger=event_logger)
return AIPlanner( return AIPlanner(
client=CloudProxyToolCallingClient(resolved_config), client=CloudProxyToolCallingClient(resolved_config),
config=planner_config, config=planner_config,
event_logger=event_logger,
) )
@@ -0,0 +1,79 @@
from __future__ import annotations
import json
import subprocess
import tempfile
from pathlib import Path
from typing import Any
class IOSDiscoveryError(RuntimeError):
pass
def discover_connected_ios_devices(*, timeout_seconds: int = 10) -> list[dict[str, str]]:
"""Return paired, currently connected physical iOS devices from CoreDevice."""
with tempfile.TemporaryDirectory(prefix="ios-device-discovery-") as temp_dir:
output_path = Path(temp_dir) / "devices.json"
try:
completed = subprocess.run(
[
"xcrun",
"devicectl",
"list",
"devices",
"--json-output",
str(output_path),
"--timeout",
str(timeout_seconds),
"--quiet",
],
capture_output=True,
text=True,
timeout=timeout_seconds + 2,
check=False,
)
except FileNotFoundError as exc:
raise IOSDiscoveryError("xcrun is unavailable; install Xcode command line tools") from exc
except subprocess.TimeoutExpired as exc:
raise IOSDiscoveryError("iOS device discovery timed out") from exc
if completed.returncode != 0:
detail = completed.stderr.strip() or completed.stdout.strip()
raise IOSDiscoveryError(detail or "devicectl failed to discover iOS devices")
try:
payload = json.loads(output_path.read_text(encoding="utf-8"))
except (OSError, ValueError) as exc:
raise IOSDiscoveryError("devicectl returned invalid device data") from exc
raw_devices = payload.get("result", {}).get("devices", [])
devices: list[dict[str, str]] = []
for item in raw_devices if isinstance(raw_devices, list) else []:
if not isinstance(item, dict):
continue
hardware = item.get("hardwareProperties", {})
properties = item.get("deviceProperties", {})
connection = item.get("connectionProperties", {})
if not all(isinstance(value, dict) for value in (hardware, properties, connection)):
continue
udid = hardware.get("udid")
if (
hardware.get("platform") != "iOS"
or hardware.get("reality") != "physical"
or connection.get("pairingState") != "paired"
or connection.get("tunnelState") != "connected"
or not isinstance(udid, str)
or not udid
):
continue
devices.append(
{
"udid": udid,
"name": str(properties.get("name") or hardware.get("marketingName") or "iPhone"),
"model": str(hardware.get("marketingName") or hardware.get("productType") or "iPhone"),
"os_version": str(properties.get("osVersionNumber") or ""),
"transport": str(connection.get("transportType") or "unknown"),
}
)
return sorted(devices, key=lambda device: (device["name"], device["udid"]))
@@ -11,7 +11,9 @@ import jinja2
from fastapi import Depends, FastAPI, HTTPException, Request from fastapi import Depends, FastAPI, HTTPException, Request
from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse, Response from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse, Response
from core.errors import DeviceNotFoundError, DeviceOfflineError, DeviceRuntimeError
from device.manager import DeviceManager from device.manager import DeviceManager
from driver.registry import build_driver_factory
from host_agent.assignment import AssignmentExecutor from host_agent.assignment import AssignmentExecutor
from host_agent.client import ( from host_agent.client import (
HostAgentClient, HostAgentClient,
@@ -25,6 +27,7 @@ from host_agent.conversation_log import ConversationLogStore
from host_agent.devices import register_local_device, unregister_local_device from host_agent.devices import register_local_device, unregister_local_device
from host_agent.history import ConsoleHistoryStore from host_agent.history import ConsoleHistoryStore
from host_agent.identity import HostIdentityStore from host_agent.identity import HostIdentityStore
from host_agent.ios_discovery import IOSDiscoveryError, discover_connected_ios_devices
from host_agent.local_account import LocalAccountStore from host_agent.local_account import LocalAccountStore
from host_agent.mcp_lock import McpBusyTracker from host_agent.mcp_lock import McpBusyTracker
from host_agent.mcp_token import McpTokenStore from host_agent.mcp_token import McpTokenStore
@@ -488,6 +491,124 @@ def create_console_app(
error=None, error=None,
) )
@app.post("/api/devices/{device_id}/screenshot")
async def api_device_screenshot(
device_id: str,
session: SessionState = Depends(require_csrf),
) -> Response:
"""Capture one on-demand screenshot for a connected local device."""
try:
screenshot = await asyncio.to_thread(
lambda: manager.active_driver(device_id).screenshot()
)
except DeviceNotFoundError as exc:
raise HTTPException(status_code=404, detail=str(exc)) from exc
except DeviceOfflineError as exc:
raise HTTPException(status_code=503, detail=str(exc)) from exc
except DeviceRuntimeError as exc:
raise HTTPException(status_code=502, detail=str(exc)) from exc
except Exception as exc:
raise HTTPException(
status_code=502,
detail=str(exc) or "failed to capture device screenshot",
) from exc
if not isinstance(screenshot, bytes) or not screenshot:
raise HTTPException(
status_code=502,
detail="device returned an empty screenshot",
)
return Response(
content=screenshot,
media_type="image/png",
headers={
"Cache-Control": "no-store",
"Pragma": "no-cache",
"X-Content-Type-Options": "nosniff",
},
)
@app.get("/api/devices/discover-ios")
async def api_discover_ios_devices(
session: SessionState = Depends(require_session),
) -> JSONResponse:
try:
discovered = await asyncio.to_thread(discover_connected_ios_devices)
except IOSDiscoveryError as exc:
raise HTTPException(status_code=502, detail=str(exc)) from exc
configured = await asyncio.to_thread(config_store.list)
configured_udids = {
str(record["connection_info"].get("udid"))
for record in configured
if record["driver_type"] == "wda"
}
used_wda_ports = {
record["connection_info"].get("wda_local_port") for record in configured
}
used_mjpeg_ports = {
record["connection_info"].get("mjpegServerPort") for record in configured
}
next_wda_port = 8100
next_mjpeg_port = 9100
result = []
for device in discovered:
while next_wda_port in used_wda_ports:
next_wda_port += 1
while next_mjpeg_port in used_mjpeg_ports:
next_mjpeg_port += 1
result.append(
{
**device,
"configured": device["udid"] in configured_udids,
"suggested_wda_port": next_wda_port,
"suggested_mjpeg_port": next_mjpeg_port,
}
)
used_wda_ports.add(next_wda_port)
used_mjpeg_ports.add(next_mjpeg_port)
next_wda_port += 1
next_mjpeg_port += 1
return JSONResponse({"devices": result})
@app.post("/api/devices/test-connection")
async def api_device_test_connection(
request: Request,
session: SessionState = Depends(require_csrf),
) -> JSONResponse:
payload = await request.json()
if not isinstance(payload, dict):
raise HTTPException(status_code=400, detail="Request must be an object.")
driver_type = str(payload.get("driver_type", "")).strip()
connection_info = payload.get("connection_info")
if not driver_type or not isinstance(connection_info, dict):
raise HTTPException(
status_code=400,
detail="Driver type and connection info are required.",
)
driver = None
connected = False
try:
driver = build_driver_factory(driver_type, connection_info)()
await asyncio.to_thread(driver.connect)
connected = True
await asyncio.to_thread(driver.health_check)
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
except Exception as exc:
raise HTTPException(
status_code=502,
detail=str(exc) or "device connection test failed",
) from exc
finally:
if connected and driver is not None:
try:
await asyncio.to_thread(driver.disconnect)
except Exception:
pass
return JSONResponse({"ok": True, "driver_type": driver_type})
@app.post("/devices/save") @app.post("/devices/save")
async def devices_save( async def devices_save(
request: Request, request: Request,
@@ -514,6 +635,26 @@ def create_console_app(
else: else:
connection_info = parsed connection_info = parsed
if error is None:
# The console exposes the common Appium settings as regular form
# fields. Advanced JSON remains available for uncommon capabilities.
field_map = {
"server_url": "server_url",
"udid": "udid",
"device_name": "device_name",
}
for form_key, config_key in field_map.items():
value = str(form.get(form_key, "")).strip()
if value:
connection_info[config_key] = value
port_key = "wda_local_port" if driver_type == "wda" else "system_port"
port_value = str(form.get(port_key, "")).strip()
if port_value:
try:
connection_info[port_key] = int(port_value)
except ValueError:
error = f"{port_key} must be an integer."
if error is None: if error is None:
try: try:
await asyncio.to_thread( await asyncio.to_thread(
@@ -5,13 +5,20 @@
<p class="error">{{ error }}</p> <p class="error">{{ error }}</p>
{% endif %} {% endif %}
<table> <table>
<thead><tr><th>ID</th><th>Name</th><th>Driver</th><th>Cloud ID</th><th></th></tr></thead> <thead><tr><th>ID</th><th>Name</th><th>Driver</th><th>Cloud ID</th><th>Screenshot</th><th></th></tr></thead>
<tbody>{% for device in devices %} <tbody>{% for device in devices %}
<tr> <tr>
<td>{{ device["device_id"] }}</td> <td>{{ device["device_id"] }}</td>
<td>{{ device["name"] or "" }}</td> <td>{{ device["name"] or "" }}</td>
<td>{{ device["driver_type"] }}</td> <td>{{ device["driver_type"] }}</td>
<td>{{ device["cloud_device_id"] or "" }}</td> <td>{{ device["cloud_device_id"] or "" }}</td>
<td>
<button type="button" class="screenshot-button" data-device-id="{{ device["device_id"] }}">Get screenshot</button>
<div class="screenshot-preview" data-screenshot-preview hidden>
<p class="screenshot-status" data-screenshot-status></p>
<img alt="Current screen for {{ device["device_id"] }}" data-screenshot-image hidden>
</div>
</td>
<td> <td>
<a href="/devices?edit={{ device["device_id"] }}">Edit</a> <a href="/devices?edit={{ device["device_id"] }}">Edit</a>
<form class="inline" method="post" action="/devices/remove"> <form class="inline" method="post" action="/devices/remove">
@@ -24,14 +31,173 @@
{% endfor %}</tbody> {% endfor %}</tbody>
</table> </table>
<h2>{{ "Edit device" if edit_record else "Add device" }}</h2> <h2>{{ "Edit device" if edit_record else "Add device" }}</h2>
<button type="button" id="discover-ios">Scan connected iPhones</button>
<p id="discovery-status" class="screenshot-status"></p>
<div id="discovered-devices"></div>
<form method="post" action="/devices/save"> <form method="post" action="/devices/save">
<input type="hidden" name="csrf_token" value="{{ csrf_token }}"> <input type="hidden" name="csrf_token" value="{{ csrf_token }}">
<label>Device ID <input type="text" name="device_id" value="{{ edit_record["device_id"] if edit_record else "" }}" required></label><br> <label>Device ID <input type="text" name="device_id" value="{{ edit_record["device_id"] if edit_record else "" }}" required></label><br>
<label>Name <input type="text" name="name" value="{{ edit_record["name"] if edit_record else "" }}"></label><br> <label>Name <input type="text" name="name" value="{{ edit_record["name"] if edit_record else "" }}"></label><br>
<label>Driver type <input type="text" name="driver_type" value="{{ edit_record["driver_type"] if edit_record else "wda" }}" required></label><br> <label>Platform and protocol
<label>Connection info (JSON)<br> <select name="driver_type" id="driver-type" required>
<textarea name="connection_info" rows="3" cols="50">{{ connection_info_json }}</textarea> <option value="wda"{% if not edit_record or edit_record["driver_type"] == "wda" %} selected{% endif %}>iOS - Appium / XCUITest (WDA)</option>
<option value="uiautomator2"{% if edit_record and edit_record["driver_type"] == "uiautomator2" %} selected{% endif %}>Android - Appium / UiAutomator2</option>
</select>
</label><br> </label><br>
<label>Appium server URL <input type="url" name="server_url" value="{{ edit_record["connection_info"].get("server_url", "http://127.0.0.1:4723") if edit_record else "http://127.0.0.1:4723" }}" required></label><br>
<label>Device UDID <input type="text" name="udid" value="{{ edit_record["connection_info"].get("udid", "") if edit_record else "" }}" required></label><br>
<label>Device name <input type="text" name="device_name" value="{{ edit_record["connection_info"].get("device_name", "") if edit_record else "" }}" placeholder="iPhone"></label><br>
<label class="ios-setting">WDA local port <input type="number" min="1" max="65535" name="wda_local_port" value="{{ edit_record["connection_info"].get("wda_local_port", "") if edit_record else "" }}" placeholder="8100"></label>
<label class="android-setting">UiAutomator2 system port <input type="number" min="1" max="65535" name="system_port" value="{{ edit_record["connection_info"].get("system_port", "") if edit_record else "" }}" placeholder="8200"></label><br>
<details>
<summary>Advanced connection capabilities (JSON)</summary>
<textarea name="connection_info" rows="4" cols="60">{{ connection_info_json }}</textarea>
</details>
<p id="connection-test-status" class="screenshot-status"></p>
<button type="button" id="test-connection">Test connection</button>
<button type="submit">Save</button> <button type="submit">Save</button>
</form> </form>
<style>
.screenshot-preview { margin-top: 0.5rem; max-width: 260px; }
.screenshot-preview img { display: block; width: 100%; height: auto; border: 1px solid #c8d0d6; }
.screenshot-status { margin: 0 0 0.35rem; color: #5e6b73; }
.screenshot-status.error { color: #b00020; }
</style>
<script>
(() => {
const driverType = document.getElementById("driver-type");
const syncPlatformFields = () => {
const ios = driverType.value === "wda";
document.querySelectorAll(".ios-setting").forEach((el) => { el.hidden = !ios; });
document.querySelectorAll(".android-setting").forEach((el) => { el.hidden = ios; });
};
driverType.addEventListener("change", syncPlatformFields);
syncPlatformFields();
const csrfInput = document.querySelector('input[name="csrf_token"]');
const csrfToken = csrfInput ? csrfInput.value : "";
const form = document.querySelector('form[action="/devices/save"]');
const discoverButton = document.getElementById("discover-ios");
const discoveryStatus = document.getElementById("discovery-status");
const discoveredDevices = document.getElementById("discovered-devices");
discoverButton.addEventListener("click", async () => {
discoverButton.disabled = true;
discoveryStatus.classList.remove("error");
discoveryStatus.textContent = "Scanning...";
discoveredDevices.replaceChildren();
try {
const response = await fetch("/api/devices/discover-ios");
const payload = await response.json();
if (!response.ok) throw new Error(payload.detail || "Discovery failed.");
discoveryStatus.textContent = payload.devices.length
? `Found ${payload.devices.length} connected iPhone(s).`
: "No connected, paired iPhones found.";
payload.devices.forEach((device, index) => {
const row = document.createElement("p");
const description = document.createElement("span");
description.textContent = `${device.name} - ${device.model} - iOS ${device.os_version} (${device.transport}) `;
const select = document.createElement("button");
select.type = "button";
select.textContent = device.configured ? "Already added" : "Use this iPhone";
select.disabled = device.configured;
select.addEventListener("click", () => {
form.elements.driver_type.value = "wda";
form.elements.udid.value = device.udid;
form.elements.device_name.value = device.name;
form.elements.wda_local_port.value = device.suggested_wda_port;
form.elements.device_id.value ||= `ios-phone-${index + 1}`;
form.elements.name.value ||= device.name;
const advanced = JSON.parse(form.elements.connection_info.value || "{}");
advanced.mjpegServerPort = device.suggested_mjpeg_port;
advanced.derivedDataPath = `/tmp/wda-${device.udid}`;
form.elements.connection_info.value = JSON.stringify(advanced, null, 2);
syncPlatformFields();
form.scrollIntoView({ behavior: "smooth", block: "start" });
});
row.append(description, select);
discoveredDevices.append(row);
});
} catch (error) {
discoveryStatus.classList.add("error");
discoveryStatus.textContent = error.message || "Discovery failed.";
} finally {
discoverButton.disabled = false;
}
});
const testButton = document.getElementById("test-connection");
const testStatus = document.getElementById("connection-test-status");
const connectionInfo = () => {
const data = new FormData(form);
let info = {};
const advanced = String(data.get("connection_info") || "{}");
info = JSON.parse(advanced);
["server_url", "udid", "device_name"].forEach((key) => {
const value = String(data.get(key) || "").trim();
if (value) info[key] = value;
});
const portKey = data.get("driver_type") === "wda" ? "wda_local_port" : "system_port";
const port = String(data.get(portKey) || "").trim();
if (port) info[portKey] = Number(port);
return { driver_type: data.get("driver_type"), connection_info: info };
};
testButton.addEventListener("click", async () => {
testButton.disabled = true;
testStatus.classList.remove("error");
testStatus.textContent = "Testing connection...";
try {
const response = await fetch("/api/devices/test-connection", {
method: "POST",
headers: { "Content-Type": "application/json", "X-CSRF-Token": csrfToken },
body: JSON.stringify(connectionInfo()),
});
const payload = await response.json();
if (!response.ok) throw new Error(payload.detail || "Connection test failed.");
testStatus.textContent = "Connection successful.";
} catch (error) {
testStatus.classList.add("error");
testStatus.textContent = error.message || "Connection test failed.";
} finally {
testButton.disabled = false;
}
});
document.querySelectorAll(".screenshot-button").forEach((button) => {
button.addEventListener("click", async () => {
const deviceId = button.dataset.deviceId;
const preview = button.parentElement.querySelector("[data-screenshot-preview]");
const status = preview.querySelector("[data-screenshot-status]");
const image = preview.querySelector("[data-screenshot-image]");
const previousUrl = image.dataset.objectUrl;
if (previousUrl) URL.revokeObjectURL(previousUrl);
button.disabled = true;
preview.hidden = false;
image.hidden = true;
status.classList.remove("error");
status.textContent = "Capturing...";
try {
const response = await fetch(
"/api/devices/" + encodeURIComponent(deviceId) + "/screenshot",
{ method: "POST", headers: { "X-CSRF-Token": csrfToken } }
);
if (!response.ok) {
let detail = "Screenshot failed.";
try {
const payload = await response.json();
if (payload.detail) detail = payload.detail;
} catch (_) {}
throw new Error(detail);
}
const objectUrl = URL.createObjectURL(await response.blob());
image.src = objectUrl;
image.dataset.objectUrl = objectUrl;
image.hidden = false;
status.textContent = "Captured.";
} catch (error) {
status.classList.add("error");
status.textContent = error.message || "Screenshot failed.";
} finally {
button.disabled = false;
}
});
});
})();
</script>
{% endblock %} {% endblock %}
@@ -47,6 +47,7 @@ def test_devices_renders(env, sample_session) -> None:
**make_devices_context(sample_session) **make_devices_context(sample_session)
) )
assert '<form method="post" action="/devices/save">' in html assert '<form method="post" action="/devices/save">' in html
assert 'class="screenshot-button"' in html
def test_account_renders(env, sample_session) -> None: def test_account_renders(env, sample_session) -> None:
@@ -96,6 +96,36 @@ def test_decide_base64_encodes_screenshot() -> None:
assert body["screenshot_base64"] == "aGVsbG8=" assert body["screenshot_base64"] == "aGVsbG8="
def test_decide_forwards_planner_history() -> 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)
history = [
{
"user_prompt": "first screen",
"tool_name": "tap",
"arguments": {"x": 1, "y": 2},
"rationale": "Open it.",
"tool_result": {"success": True},
}
]
client.decide(
system_prompt="sp",
user_prompt="next screen",
screenshot=None,
tools=_TOOLS,
timeout=10.0,
history=history,
)
assert json.loads(seen_requests[0].content)["history"] == history
def test_decide_clamps_legacy_timeout_and_waits_for_cloud_profile_timeout() -> None: def test_decide_clamps_legacy_timeout_and_waits_for_cloud_profile_timeout() -> None:
seen_requests: list[httpx.Request] = [] seen_requests: list[httpx.Request] = []
@@ -242,6 +242,147 @@ def test_add_device_appears_in_devices_page_and_manager(tmp_path) -> None:
assert [device.id for device in context["manager"].list_devices()] == ["device-a"] assert [device.id for device in context["manager"].list_devices()] == ["device-a"]
def test_devices_page_captures_screenshot_only_when_button_endpoint_is_called(
tmp_path,
) -> None:
class ScreenshotDriver:
def __init__(self) -> None:
self.capture_count = 0
def connect(self) -> None:
pass
def disconnect(self) -> None:
pass
def screenshot(self) -> bytes:
self.capture_count += 1
return b"fake-png"
driver = ScreenshotDriver()
client, context = _build_client(tmp_path)
context["config_store"].add(
device_id="device-a",
name="Lab iPhone",
driver_type="wda",
connection_info={},
)
context["manager"].register_device("device-a", lambda: driver)
context["manager"].connect("device-a")
csrf_token = _login(client)
page = client.get("/devices")
assert page.status_code == 200
assert 'class="screenshot-button"' in page.text
assert 'data-device-id="device-a"' in page.text
assert driver.capture_count == 0
response = client.post(
"/api/devices/device-a/screenshot",
headers={"X-CSRF-Token": csrf_token},
)
assert response.status_code == 200
assert response.content == b"fake-png"
assert response.headers["content-type"] == "image/png"
assert response.headers["cache-control"] == "no-store"
assert driver.capture_count == 1
def test_device_screenshot_requires_csrf_and_connected_device(tmp_path) -> None:
class ScreenshotDriver:
def connect(self) -> None:
pass
def disconnect(self) -> None:
pass
def screenshot(self) -> bytes:
return b"fake-png"
client, context = _build_client(tmp_path)
context["manager"].register_device("device-a", ScreenshotDriver)
csrf_token = _login(client)
missing_csrf = client.post("/api/devices/device-a/screenshot")
assert missing_csrf.status_code == 403
offline = client.post(
"/api/devices/device-a/screenshot",
headers={"X-CSRF-Token": csrf_token},
)
assert offline.status_code == 503
def test_device_connection_test_connects_checks_health_and_disconnects(
tmp_path, monkeypatch
) -> None:
events: list[str] = []
class ProbeDriver:
def connect(self) -> None:
events.append("connect")
def health_check(self) -> None:
events.append("health")
def disconnect(self) -> None:
events.append("disconnect")
def factory(driver_type, connection_info):
assert driver_type == "wda"
assert connection_info["udid"] == "ios-udid"
return ProbeDriver
monkeypatch.setattr("host_agent.web.app.build_driver_factory", factory)
client, context = _build_client(tmp_path)
csrf_token = _login(client)
response = client.post(
"/api/devices/test-connection",
json={"driver_type": "wda", "connection_info": {"udid": "ios-udid"}},
headers={"X-CSRF-Token": csrf_token},
)
assert response.status_code == 200
assert response.json() == {"ok": True, "driver_type": "wda"}
assert events == ["connect", "health", "disconnect"]
assert context["config_store"].list() == []
def test_ios_discovery_returns_connected_devices_and_unique_ports(
tmp_path, monkeypatch
) -> None:
monkeypatch.setattr(
"host_agent.web.app.discover_connected_ios_devices",
lambda: [
{
"udid": "ios-new",
"name": "New iPhone",
"model": "iPhone 15",
"os_version": "18.0",
"transport": "wired",
}
],
)
client, context = _build_client(tmp_path)
context["config_store"].add(
device_id="existing",
driver_type="wda",
connection_info={"udid": "ios-old", "wda_local_port": 8100, "mjpegServerPort": 9100},
)
_login(client)
response = client.get("/api/devices/discover-ios")
assert response.status_code == 200
device = response.json()["devices"][0]
assert device["udid"] == "ios-new"
assert device["configured"] is False
assert device["suggested_wda_port"] == 8101
assert device["suggested_mjpeg_port"] == 9101
def test_remove_device_unregisters_from_manager(tmp_path) -> None: def test_remove_device_unregisters_from_manager(tmp_path) -> None:
client, context = _build_client(tmp_path) client, context = _build_client(tmp_path)
csrf_token = _login(client) csrf_token = _login(client)
+7 -1
View File
@@ -130,7 +130,13 @@ class DeviceManager:
if driver is None: if driver is None:
return False return False
try: try:
driver.screenshot() health_check = getattr(driver, "health_check", None)
if callable(health_check):
health_check()
else:
# Compatibility for drivers implemented before health_check
# existed. Built-in drivers use the non-screen health check.
driver.screenshot()
except Exception: except Exception:
self.mark_error(device_id, offline=True) self.mark_error(device_id, offline=True)
return False return False
+7
View File
@@ -81,6 +81,13 @@ class AndroidDriver(Driver):
except Exception as exc: except Exception as exc:
raise DriverError("screenshot failed") from exc raise DriverError("screenshot failed") from exc
def health_check(self) -> None:
client = self._require_client()
try:
client.get_status()
except Exception as exc:
raise DriverError("health check failed") from exc
def tap(self, x: float, y: float) -> None: def tap(self, x: float, y: float) -> None:
client = self._require_client() client = self._require_client()
try: try:
+9
View File
@@ -25,6 +25,15 @@ class Driver(ABC):
def screenshot(self) -> bytes: def screenshot(self) -> bytes:
"""Return the current screen as image bytes.""" """Return the current screen as image bytes."""
def health_check(self) -> None:
"""Verify the live session without reading the device screen.
Drivers with a transport-level status endpoint should override this
method. The default is a no-op for legacy drivers that do not expose
a separate health check.
"""
return None
@abstractmethod @abstractmethod
def tap(self, x: float, y: float) -> None: def tap(self, x: float, y: float) -> None:
"""Tap the screen at the given coordinates.""" """Tap the screen at the given coordinates."""
+7
View File
@@ -71,6 +71,13 @@ class WDADriver(Driver):
except Exception as exc: except Exception as exc:
raise DriverError("screenshot failed") from exc raise DriverError("screenshot failed") from exc
def health_check(self) -> None:
client = self._require_client()
try:
client.get_status()
except Exception as exc:
raise DriverError("health check failed") from exc
def tap(self, x: float, y: float) -> None: def tap(self, x: float, y: float) -> None:
client = self._require_client() client = self._require_client()
try: try:
@@ -6,6 +6,7 @@ import json
import logging import logging
from collections.abc import Awaitable, Callable from collections.abc import Awaitable, Callable
from datetime import timedelta from datetime import timedelta
from inspect import Parameter, signature
from time import monotonic from time import monotonic
from typing import TYPE_CHECKING from typing import TYPE_CHECKING
from uuid import uuid4 from uuid import uuid4
@@ -505,13 +506,21 @@ def create_internal_router(
started_at = monotonic() started_at = monotonic()
try: try:
decision = client.decide( decision_kwargs = {
system_prompt=payload.system_prompt, "system_prompt": payload.system_prompt,
user_prompt=payload.user_prompt, "user_prompt": payload.user_prompt,
screenshot=screenshot, "screenshot": screenshot,
tools=tools, "tools": tools,
timeout=planner_timeout, "timeout": planner_timeout,
) }
parameters = signature(client.decide).parameters.values()
if any(
parameter.name == "history"
or parameter.kind == Parameter.VAR_KEYWORD
for parameter in parameters
):
decision_kwargs["history"] = payload.history
decision = client.decide(**decision_kwargs)
except ToolCallUnavailable as exc: except ToolCallUnavailable as exc:
logger.info( logger.info(
"planner-decision request failed", "planner-decision request failed",
@@ -143,6 +143,7 @@ class PlannerDecisionRequest(BaseModel):
host_id: str = Field(min_length=1) host_id: str = Field(min_length=1)
system_prompt: str system_prompt: str
user_prompt: str user_prompt: str
history: list[dict[str, Any]] = Field(default_factory=list)
screenshot_base64: str | None = None screenshot_base64: str | None = None
tools: list[PlannerToolSpecModel] = Field(default_factory=list) tools: list[PlannerToolSpecModel] = Field(default_factory=list)
timeout_seconds: float = Field(default=30.0, gt=0, le=120) timeout_seconds: float = Field(default=30.0, gt=0, le=120)
+67 -24
View File
@@ -1,5 +1,7 @@
from __future__ import annotations from __future__ import annotations
from collections.abc import Callable
from inspect import Parameter, signature
from typing import TYPE_CHECKING, Any from typing import TYPE_CHECKING, Any
from core.errors import TaskFailedError from core.errors import TaskFailedError
@@ -23,9 +25,11 @@ class AIPlanner(Planner):
*, *,
client: ToolCallingClient | None = None, client: ToolCallingClient | None = None,
config: PlannerConfig | None = None, config: PlannerConfig | None = None,
event_logger: Callable[[dict[str, Any]], None] | None = None,
) -> None: ) -> None:
self.config = config or load_config() self.config = config or load_config()
self.client = client or build_client(self.config) self.client = client or build_client(self.config)
self.event_logger = event_logger
def plan( def plan(
self, self,
@@ -36,11 +40,11 @@ class AIPlanner(Planner):
world: "WorldState | None" = None, world: "WorldState | None" = None,
screenshot: bytes | None = None, screenshot: bytes | None = None,
) -> list[PlannedStep]: ) -> list[PlannedStep]:
_sync_tool_results(context)
scene_json = _without_ocr(scene.to_dict()) if self.config.multimodal else scene.to_dict() scene_json = _without_ocr(scene.to_dict()) if self.config.multimodal else scene.to_dict()
user_prompt = planner_user_prompt( user_prompt = planner_user_prompt(
goal=goal, goal=goal,
scene_json=scene_json, scene_json=scene_json,
history_summary=_history_summary(world),
device_platform=context.device_platform, device_platform=context.device_platform,
) )
if context.step_results and not context.step_results[-1].success: if context.step_results and not context.step_results[-1].success:
@@ -49,19 +53,45 @@ class AIPlanner(Planner):
"corrected action or finish the task if it cannot proceed.\n" "corrected action or finish the task if it cannot proceed.\n"
f"Previous failure: {context.step_results[-1].error or 'unknown error'}" f"Previous failure: {context.step_results[-1].error or 'unknown error'}"
) )
decision = self.client.decide( self._log({"type": "llm_request", "task_id": context.task_id, "goal": goal,
system_prompt=PLANNER_SYSTEM_PROMPT, "system_prompt": PLANNER_SYSTEM_PROMPT, "user_prompt": user_prompt,
user_prompt=user_prompt, "has_screenshot": screenshot is not None})
screenshot=screenshot, try:
tools=ALL_TOOL_SPECS, kwargs = {
timeout=self.config.timeout, "system_prompt": PLANNER_SYSTEM_PROMPT,
) "user_prompt": user_prompt,
"screenshot": screenshot,
"tools": ALL_TOOL_SPECS,
"timeout": self.config.timeout,
}
if _accepts_history(self.client.decide):
kwargs["history"] = context.planner_history[
-self.config.history_max_turns :
]
decision = self.client.decide(**kwargs)
except Exception as exc:
self._log({"type": "agent_error", "task_id": context.task_id, "error": str(exc)})
raise
self._log({"type": "llm_response", "task_id": context.task_id,
"content": decision.text_output, "thinking": decision.thinking,
"tool_name": decision.tool_name, "arguments": decision.arguments,
"purpose": decision.purpose, "expected_outcome": decision.expected_outcome})
if decision.tool_name == FINISH_TASK_TOOL: if decision.tool_name == FINISH_TASK_TOOL:
if decision.arguments.get("success"): if decision.arguments.get("success"):
return [] return []
raise TaskFailedError(decision.arguments.get("reason") or "task failed") raise TaskFailedError(decision.arguments.get("reason") or "task failed")
context.planner_history.append(
{
"user_prompt": user_prompt,
"tool_name": decision.tool_name,
"arguments": _conversation_arguments(decision),
"rationale": decision.text_output,
"tool_result": None,
}
)
return [ return [
PlannedStep( PlannedStep(
action=decision.tool_name, action=decision.tool_name,
@@ -85,22 +115,12 @@ class AIPlanner(Planner):
# (mapped to an empty plan above), never via this hook. # (mapped to an empty plan above), never via this hook.
return False return False
def _log(self, event: dict[str, Any]) -> None:
def _history_summary(world: "WorldState | None") -> list[dict[str, Any]]: if self.event_logger is not None:
if world is None: try:
return [] self.event_logger(event)
return [ except Exception:
{ pass
"page": event.page,
"action": event.action,
"arguments": dict(event.arguments),
"rationale": event.rationale,
"purpose": event.purpose,
"expected_outcome": event.expected_outcome,
"success": event.success,
}
for event in world.history
]
def _without_ocr(scene_json: dict[str, Any]) -> dict[str, Any]: def _without_ocr(scene_json: dict[str, Any]) -> dict[str, Any]:
@@ -114,3 +134,26 @@ def _without_ocr(scene_json: dict[str, Any]) -> dict[str, Any]:
] ]
cleaned.pop("ocr_elements", None) cleaned.pop("ocr_elements", None)
return cleaned return cleaned
def _sync_tool_results(context: TaskContext) -> None:
for turn, result in zip(context.planner_history, context.step_results, strict=False):
if turn.get("tool_result") is None:
turn["tool_result"] = result.to_dict()
def _conversation_arguments(decision: Any) -> dict[str, Any]:
arguments = dict(decision.arguments)
if decision.purpose is not None:
arguments["purpose"] = decision.purpose
if decision.expected_outcome is not None:
arguments["expected_outcome"] = decision.expected_outcome
return arguments
def _accepts_history(method: Any) -> bool:
parameters = signature(method).parameters.values()
return any(
parameter.name == "history" or parameter.kind == Parameter.VAR_KEYWORD
for parameter in parameters
)
+2 -1
View File
@@ -1,7 +1,7 @@
from __future__ import annotations from __future__ import annotations
from dataclasses import dataclass, field from dataclasses import dataclass, field
from typing import TYPE_CHECKING from typing import TYPE_CHECKING, Any
from core.models import Scene from core.models import Scene
@@ -18,6 +18,7 @@ class TaskContext:
scenes: list[Scene] = field(default_factory=list) scenes: list[Scene] = field(default_factory=list)
step_results: list["StepResult"] = field(default_factory=list) step_results: list["StepResult"] = field(default_factory=list)
world: "WorldState | None" = None world: "WorldState | None" = None
planner_history: list[dict[str, Any]] = field(default_factory=list)
def add_scene(self, scene: Scene) -> None: def add_scene(self, scene: Scene) -> None:
self.scenes.append(scene) self.scenes.append(scene)
+17
View File
@@ -11,6 +11,7 @@ DEFAULT_MODEL_BY_PROVIDER = {
"openai_compatible": "local-model", "openai_compatible": "local-model",
} }
DEFAULT_TIMEOUT_SECONDS = 30.0 DEFAULT_TIMEOUT_SECONDS = 30.0
DEFAULT_HISTORY_MAX_TURNS = 20
ENABLED_ENV = "AI_PLANNER_ENABLED" ENABLED_ENV = "AI_PLANNER_ENABLED"
PROVIDER_ENV = "AI_PLANNER_PROVIDER" PROVIDER_ENV = "AI_PLANNER_PROVIDER"
@@ -20,6 +21,7 @@ THINKING_BUDGET_ENV = "AI_PLANNER_THINKING_BUDGET_TOKENS"
API_KEY_ENV = "AI_PLANNER_API_KEY" API_KEY_ENV = "AI_PLANNER_API_KEY"
BASE_URL_ENV = "AI_PLANNER_BASE_URL" BASE_URL_ENV = "AI_PLANNER_BASE_URL"
MULTIMODAL_ENV = "AI_PLANNER_MULTIMODAL" MULTIMODAL_ENV = "AI_PLANNER_MULTIMODAL"
HISTORY_MAX_TURNS_ENV = "AI_PLANNER_HISTORY_MAX_TURNS"
SUPPORTED_PROVIDERS = frozenset(DEFAULT_MODEL_BY_PROVIDER) SUPPORTED_PROVIDERS = frozenset(DEFAULT_MODEL_BY_PROVIDER)
@@ -34,6 +36,7 @@ class PlannerConfig:
api_key: str | None = None api_key: str | None = None
base_url: str | None = None base_url: str | None = None
multimodal: bool = False multimodal: bool = False
history_max_turns: int = DEFAULT_HISTORY_MAX_TURNS
def resolved_model(self) -> str: def resolved_model(self) -> str:
return self.model or DEFAULT_MODEL_BY_PROVIDER[self.provider] return self.model or DEFAULT_MODEL_BY_PROVIDER[self.provider]
@@ -50,6 +53,10 @@ def load_config(env: Mapping[str, str] | None = None) -> PlannerConfig:
api_key=values.get(API_KEY_ENV) or _provider_key(values), api_key=values.get(API_KEY_ENV) or _provider_key(values),
base_url=values.get(BASE_URL_ENV) or None, base_url=values.get(BASE_URL_ENV) or None,
multimodal=_parse_bool(values.get(MULTIMODAL_ENV), default=False), multimodal=_parse_bool(values.get(MULTIMODAL_ENV), default=False),
history_max_turns=_parse_positive_int(
values.get(HISTORY_MAX_TURNS_ENV),
default=DEFAULT_HISTORY_MAX_TURNS,
),
) )
@@ -88,6 +95,16 @@ def _parse_thinking_budget(value: str | None) -> int | None:
return budget if budget > 0 else None return budget if budget > 0 else None
def _parse_positive_int(value: str | None, *, default: int) -> int:
if value is None:
return default
try:
parsed = int(value)
except ValueError:
return default
return parsed if parsed > 0 else default
def _provider_key(values: Mapping[str, str]) -> str | None: def _provider_key(values: Mapping[str, str]) -> str | None:
provider = (values.get(PROVIDER_ENV) or DEFAULT_PROVIDER).strip().lower() provider = (values.get(PROVIDER_ENV) or DEFAULT_PROVIDER).strip().lower()
if provider == "openai": if provider == "openai":
-3
View File
@@ -66,7 +66,6 @@ def planner_user_prompt(
*, *,
goal: str, goal: str,
scene_json: dict[str, Any], scene_json: dict[str, Any],
history_summary: list[dict[str, Any]],
device_platform: str | None = None, device_platform: str | None = None,
now: datetime | None = None, now: datetime | None = None,
) -> str: ) -> str:
@@ -83,8 +82,6 @@ def planner_user_prompt(
f"{goal}\n\n" f"{goal}\n\n"
"Current Scene (JSON):\n" "Current Scene (JSON):\n"
f"{json.dumps(scene_json, ensure_ascii=False, sort_keys=True)}\n\n" f"{json.dumps(scene_json, ensure_ascii=False, sort_keys=True)}\n\n"
"Recent history, oldest first (JSON):\n"
f"{json.dumps(history_summary, ensure_ascii=False, sort_keys=True)}\n\n"
"Call exactly one tool for this turn." "Call exactly one tool for this turn."
) )
+87 -4
View File
@@ -51,6 +51,7 @@ class ToolCallingClient(Protocol):
screenshot: bytes | None, screenshot: bytes | None,
tools: list[ToolSpec], tools: list[ToolSpec],
timeout: float, timeout: float,
history: list[dict[str, Any]] | None = None,
) -> ToolCallDecision: ... ) -> ToolCallDecision: ...
@@ -80,6 +81,7 @@ class AnthropicToolCallingClient:
screenshot: bytes | None, screenshot: bytes | None,
tools: list[ToolSpec], tools: list[ToolSpec],
timeout: float, timeout: float,
history: list[dict[str, Any]] | None = None,
) -> ToolCallDecision: ) -> ToolCallDecision:
try: try:
response = self._create_message( response = self._create_message(
@@ -87,6 +89,7 @@ class AnthropicToolCallingClient:
user_prompt, user_prompt,
screenshot, screenshot,
tools, tools,
history=history,
timeout=timeout, timeout=timeout,
forced=False, forced=False,
) )
@@ -103,6 +106,7 @@ class AnthropicToolCallingClient:
user_prompt, user_prompt,
screenshot, screenshot,
tools, tools,
history=history,
timeout=timeout, timeout=timeout,
forced=True, forced=True,
) )
@@ -123,6 +127,7 @@ class AnthropicToolCallingClient:
screenshot: bytes | None, screenshot: bytes | None,
tools: list[ToolSpec], tools: list[ToolSpec],
*, *,
history: list[dict[str, Any]] | None,
timeout: float, timeout: float,
forced: bool, forced: bool,
) -> Any: ) -> Any:
@@ -146,10 +151,8 @@ class AnthropicToolCallingClient:
} }
], ],
"messages": [ "messages": [
{ *_anthropic_history(history or []),
"role": "user", {"role": "user", "content": _anthropic_content(user_prompt, screenshot)},
"content": _anthropic_content(user_prompt, screenshot),
}
], ],
"tools": [_anthropic_tool(spec) for spec in tools], "tools": [_anthropic_tool(spec) for spec in tools],
"tool_choice": { "tool_choice": {
@@ -207,6 +210,7 @@ class OpenAIToolCallingClient:
screenshot: bytes | None, screenshot: bytes | None,
tools: list[ToolSpec], tools: list[ToolSpec],
timeout: float, timeout: float,
history: list[dict[str, Any]] | None = None,
) -> ToolCallDecision: ) -> ToolCallDecision:
try: try:
response = self._create_completion( response = self._create_completion(
@@ -214,6 +218,7 @@ class OpenAIToolCallingClient:
user_prompt, user_prompt,
screenshot, screenshot,
tools, tools,
history=history,
timeout=timeout, timeout=timeout,
forced=False, forced=False,
) )
@@ -227,6 +232,7 @@ class OpenAIToolCallingClient:
user_prompt, user_prompt,
screenshot, screenshot,
tools, tools,
history=history,
timeout=timeout, timeout=timeout,
forced=True, forced=True,
) )
@@ -247,6 +253,7 @@ class OpenAIToolCallingClient:
screenshot: bytes | None, screenshot: bytes | None,
tools: list[ToolSpec], tools: list[ToolSpec],
*, *,
history: list[dict[str, Any]] | None,
timeout: float, timeout: float,
forced: bool, forced: bool,
) -> Any: ) -> Any:
@@ -257,6 +264,7 @@ class OpenAIToolCallingClient:
"timeout": timeout, "timeout": timeout,
"messages": [ "messages": [
{"role": "system", "content": system_prompt}, {"role": "system", "content": system_prompt},
*_openai_history(history or []),
{"role": "user", "content": _openai_content(user_prompt, screenshot)}, {"role": "user", "content": _openai_content(user_prompt, screenshot)},
], ],
"tools": [_openai_tool(spec) for spec in tools], "tools": [_openai_tool(spec) for spec in tools],
@@ -320,6 +328,44 @@ def _anthropic_content(
return content return content
def _anthropic_history(history: list[dict[str, Any]]) -> list[dict[str, Any]]:
messages: list[dict[str, Any]] = []
for index, turn in enumerate(history):
result = turn.get("tool_result")
if result is None:
continue
tool_use_id = f"planner-turn-{index}"
assistant_content: list[dict[str, Any]] = []
rationale = turn.get("rationale")
if isinstance(rationale, str) and rationale:
assistant_content.append({"type": "text", "text": rationale})
assistant_content.append(
{
"type": "tool_use",
"id": tool_use_id,
"name": turn["tool_name"],
"input": dict(turn.get("arguments") or {}),
}
)
messages.extend(
[
{"role": "user", "content": turn["user_prompt"]},
{"role": "assistant", "content": assistant_content},
{
"role": "user",
"content": [
{
"type": "tool_result",
"tool_use_id": tool_use_id,
"content": json.dumps(result, ensure_ascii=False, default=str),
}
],
},
]
)
return messages
def _anthropic_tool(spec: ToolSpec) -> dict[str, Any]: def _anthropic_tool(spec: ToolSpec) -> dict[str, Any]:
return { return {
"name": spec.name, "name": spec.name,
@@ -393,6 +439,43 @@ def _openai_content(
] ]
def _openai_history(history: list[dict[str, Any]]) -> list[dict[str, Any]]:
messages: list[dict[str, Any]] = []
for index, turn in enumerate(history):
result = turn.get("tool_result")
if result is None:
continue
tool_call_id = f"planner-turn-{index}"
messages.extend(
[
{"role": "user", "content": turn["user_prompt"]},
{
"role": "assistant",
"content": turn.get("rationale"),
"tool_calls": [
{
"id": tool_call_id,
"type": "function",
"function": {
"name": turn["tool_name"],
"arguments": json.dumps(
turn.get("arguments") or {},
ensure_ascii=False,
),
},
}
],
},
{
"role": "tool",
"tool_call_id": tool_call_id,
"content": json.dumps(result, ensure_ascii=False, default=str),
},
]
)
return messages
def _openai_tool(spec: ToolSpec) -> dict[str, Any]: def _openai_tool(spec: ToolSpec) -> dict[str, Any]:
return { return {
"type": "function", "type": "function",
+57 -53
View File
@@ -8,6 +8,7 @@ from core.errors import TaskFailedError
from core.models import Bounds, Scene, SceneElement from core.models import Bounds, Scene, SceneElement
from runtime.ai_planner import AIPlanner from runtime.ai_planner import AIPlanner
from runtime.context import TaskContext from runtime.context import TaskContext
from runtime.executor import StepResult
from runtime.planner_config import PlannerConfig from runtime.planner_config import PlannerConfig
from runtime.tool_calling_client import ToolCallDecision from runtime.tool_calling_client import ToolCallDecision
from runtime.tool_specs import ALL_TOOL_SPECS from runtime.tool_specs import ALL_TOOL_SPECS
@@ -26,6 +27,7 @@ class FakeToolCallingClient:
screenshot: bytes | None, screenshot: bytes | None,
tools: list[Any], tools: list[Any],
timeout: float, timeout: float,
history: list[dict[str, Any]] | None = None,
) -> ToolCallDecision: ) -> ToolCallDecision:
self.calls.append( self.calls.append(
{ {
@@ -34,6 +36,7 @@ class FakeToolCallingClient:
"screenshot": screenshot, "screenshot": screenshot,
"tools": tools, "tools": tools,
"timeout": timeout, "timeout": timeout,
"history": list(history) if history is not None else None,
} }
) )
return self.decision return self.decision
@@ -242,63 +245,64 @@ def test_ai_planner_propagates_rationale_and_thinking_to_planned_step() -> None:
assert steps[0].expected_outcome == "The account settings page is visible." assert steps[0].expected_outcome == "The account settings page is visible."
def test_history_summary_returns_compact_format() -> None: def test_ai_planner_carries_completed_turn_into_the_next_llm_call() -> None:
from collections import deque client = FakeToolCallingClient(
from runtime.ai_planner import _history_summary ToolCallDecision(
from world.models import WorldEvent, WorldState tool_name="tap",
arguments={"x": 1, "y": 2},
state = WorldState( text_output="Opening the send control.",
history=deque( purpose="Open the send control.",
[ expected_outcome="The composer is focused.",
WorldEvent(
action="tap",
success=True,
rationale="Opened settings.",
arguments={"x": 1, "y": 2},
purpose="Open settings.",
expected_outcome="Settings is visible.",
page="Home",
),
WorldEvent(
action="swipe",
success=False,
rationale=None,
arguments={"start_y": 700, "end_y": 200},
page="Settings",
),
]
) )
) )
planner = AIPlanner(client=client)
context = _context()
summary = _history_summary(state) first_step = planner.plan(goal=context.goal, scene=_scene(), context=context)[0]
context.add_step_result(
StepResult(
step=first_step,
success=True,
attempts=1,
result={"ok": True},
)
)
planner.plan(goal=context.goal, scene=_scene(), context=context)
assert summary == [ assert client.calls[0]["history"] == []
history = client.calls[1]["history"]
assert history is not None
assert history[0]["tool_name"] == "tap"
assert history[0]["arguments"] == {
"x": 1,
"y": 2,
"purpose": "Open the send control.",
"expected_outcome": "The composer is focused.",
}
assert history[0]["tool_result"]["success"] is True
assert history[0]["tool_result"]["result"] == {"ok": True}
def test_ai_planner_limits_history_sent_to_the_llm() -> None:
client = FakeToolCallingClient(
ToolCallDecision(tool_name="tap", arguments={"x": 1, "y": 2})
)
planner = AIPlanner(client=client, config=PlannerConfig(history_max_turns=2))
context = _context()
context.planner_history.extend(
{ {
"page": "Home", "user_prompt": f"turn-{index}",
"action": "tap", "tool_name": "tap",
"arguments": {"x": 1, "y": 2}, "arguments": {},
"rationale": "Opened settings.",
"purpose": "Open settings.",
"expected_outcome": "Settings is visible.",
"success": True,
},
{
"page": "Settings",
"action": "swipe",
"arguments": {"start_y": 700, "end_y": 200},
"rationale": None, "rationale": None,
"purpose": None, "tool_result": {"success": True},
"expected_outcome": None, }
"success": False, for index in range(3)
}, )
planner.plan(goal=context.goal, scene=_scene(), context=context)
assert [turn["user_prompt"] for turn in client.calls[0]["history"]] == [
"turn-1",
"turn-2",
] ]
# Must not contain scene element data
for entry in summary:
assert "scene_summary" not in entry
assert "elements" not in entry
def test_history_summary_returns_empty_for_none_world() -> None:
from runtime.ai_planner import _history_summary
assert _history_summary(None) == []
+26
View File
@@ -46,3 +46,29 @@ def test_probe_marks_connected_device_offline_when_driver_is_unreachable() -> No
assert manager.probe("iphone-1") is False assert manager.probe("iphone-1") is False
assert manager.status("iphone-1") == "offline" assert manager.status("iphone-1") == "offline"
def test_probe_uses_health_check_without_capturing_screen() -> None:
class HealthCheckedDriver:
def __init__(self) -> None:
self.health_checks = 0
self.screenshots = 0
def connect(self) -> None:
return None
def screenshot(self) -> bytes:
self.screenshots += 1
return b"screen"
def health_check(self) -> None:
self.health_checks += 1
driver = HealthCheckedDriver()
manager = DeviceManager()
manager.register_device("iphone-1", lambda: driver) # type: ignore[arg-type]
manager.connect("iphone-1")
assert manager.probe("iphone-1") is True
assert driver.health_checks == 1
assert driver.screenshots == 0
+9 -2
View File
@@ -9,7 +9,6 @@ def test_planner_user_prompt_includes_time_zone_and_configured_device_type() ->
prompt = planner_user_prompt( prompt = planner_user_prompt(
goal="open settings", goal="open settings",
scene_json={"screen": {"width": 1, "height": 1}, "elements": []}, scene_json={"screen": {"width": 1, "height": 1}, "elements": []},
history_summary=[],
device_platform="ios", device_platform="ios",
now=datetime( now=datetime(
2026, 2026,
@@ -35,8 +34,16 @@ def test_planner_user_prompt_uses_scene_platform_when_context_is_unavailable() -
"elements": [], "elements": [],
"app": {"platform": "android"}, "app": {"platform": "android"},
}, },
history_summary=[],
now=datetime(2026, 7, 16, tzinfo=timezone.utc), now=datetime(2026, 7, 16, tzinfo=timezone.utc),
) )
assert "Device type: android" in prompt assert "Device type: android" in prompt
def test_planner_user_prompt_does_not_duplicate_conversation_history() -> None:
prompt = planner_user_prompt(
goal="open settings",
scene_json={"screen": {"width": 1, "height": 1}, "elements": []},
)
assert "Recent history" not in prompt