Support Anthropic conversation logging and retries

README.md

@@ -72,6 +72,7 @@ export AI_PLANNER_PROVIDER=openai-compatible
 export AI_PLANNER_MODEL=qwen2.5
 export AI_PLANNER_API_KEY=local-key
 export AI_PLANNER_BASE_URL=http://127.0.0.1:11434/v1
+export AI_PLANNER_MULTIMODAL=true
 uv run --package device-host-agent device-host-agent setup
 uv run --package device-host-agent device-host-agent
 ```
@@ -86,12 +87,16 @@ The local Host console also exposes an authenticated conversational Agent API:
 POST http://127.0.0.1:8765/api/chat
 ```
 
-Send a JSON body containing `messages` (`user`/`assistant` roles). The Agent can
+Send a JSON body containing `device_id` and `messages` (`user`/`assistant` roles). The Agent can
 return ordinary assistant text or call the same device-operation tool contracts
 used by the Runtime; each tool result is fed back to the model before the final
 reply is returned. The endpoint uses the Host console session cookie, so it is
 not an unauthenticated device-control endpoint.
 
+Every chat request is bound to exactly one registered `device_id`. Keep a
+separate message history and client session for each phone; the Agent injects
+the bound device into device tools and rejects cross-device tool arguments.
+
 For vision-capable OpenAI models, a user message may contain standard OpenAI
 multimodal blocks:
 
@@ -111,9 +116,14 @@ The compact form `{ "role": "user", "text": "...", "image_base64": "..." }`
 is also accepted. The model can inspect the supplied image and then call a
 phone tool such as `tap` in the same conversation.
 
+Set `AI_PLANNER_MULTIMODAL=true` for a vision model. The Planner sends the
+screenshot and omits OCR-only elements and OCR metadata from the structured
+scene payload, avoiding duplicate OCR text.
+
 Local mode records LLM responses, reasoning fields, tool calls, tool results,
-and final replies as JSONL in `host_agent_data/conversations.jsonl`. Image
-bytes are excluded; set `HOST_AGENT_CONVERSATION_LOG_PATH` to change the path.
+and final replies in a local SQLite database. View them at
+`http://127.0.0.1:8765/conversations`; image bytes are excluded. Set
+`HOST_AGENT_CONVERSATION_LOG_PATH` to change the database path.
 
 In local mode, Appium supervision is enabled by default. Host Agent probes
 `/status`, adopts a healthy existing Appium instance, starts Appium when no

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

@@ -276,6 +276,7 @@ def create_application(
             mcp_token_store=mcp_token_store,
             mcp_busy_tracker=mcp_busy_tracker,
             conversation_agent=conversation_agent,
+            conversation_log=conversation_log,
         )
         console_server = _EmbeddedConsoleServer(
             uvicorn.Config(

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

@@ -49,7 +49,7 @@ class HostAgentConfig:
     dependency_restart_max_attempts: int = 5
     task_progress_db_path: Path = Path("host_agent_data/task_progress.sqlite3")
     task_artifact_dir: Path = Path("host_agent_data/history")
-    conversation_log_path: Path = Path("host_agent_data/conversations.jsonl")
+    conversation_log_path: Path = Path("host_agent_data/conversations.sqlite3")
     task_retention_max_count: int = 50
     task_retention_max_age_days: int = 7
     skill_sync_interval_seconds: float = 300.0
@@ -160,7 +160,7 @@ def load_host_agent_config(
                 "HOST_AGENT_TASK_ARTIFACT_DIR", "host_agent_data/history"
             ).strip()
         ),
-        conversation_log_path=Path(values.get("HOST_AGENT_CONVERSATION_LOG_PATH", "host_agent_data/conversations.jsonl").strip()),
+        conversation_log_path=Path(values.get("HOST_AGENT_CONVERSATION_LOG_PATH", "host_agent_data/conversations.sqlite3").strip()),
         task_retention_max_count=_positive_int(
             values, "HOST_AGENT_TASK_RETENTION_MAX_COUNT", 50
         ),

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

@@ -40,12 +40,13 @@ class ConversationAgent:
         self.event_logger = event_logger
         self._client: Any | None = None
 
-    def chat(self, messages: list[dict[str, Any]]) -> ChatResult:
+    def chat(self, messages: list[dict[str, Any]], *, tools: dict[str, Callable[..., Any]] | None = None) -> ChatResult:
+        if self.config.provider == "anthropic":
+            return self._chat_anthropic(messages, tools=tools)
         if self.config.provider not in {"openai", "openai_compatible"}:
-            raise RuntimeError(
-                "conversation chat currently requires an OpenAI-compatible provider"
-            )
+            raise RuntimeError("unsupported conversation provider")
         client = self._get_client()
+        active_tools = tools or self.tools
         history: list[dict[str, Any]] = [
             {
                 "role": "system",
@@ -88,7 +89,7 @@ class ConversationAgent:
                     arguments = decoded_arguments
                     arguments.pop("purpose", None)
                     arguments.pop("expected_outcome", None)
-                    result = self.tools[name](**arguments)
+                    result = active_tools[name](**arguments)
                 except Exception as exc:
                     result = {"ok": False, "error": str(exc)}
                 calls += 1
@@ -103,6 +104,58 @@ class ConversationAgent:
                 )
         raise RuntimeError("conversation exceeded the maximum tool-call rounds")
 
+    def _chat_anthropic(self, messages: list[dict[str, Any]], *, tools: dict[str, Callable[..., Any]] | None = None) -> ChatResult:
+        import anthropic
+
+        kwargs: dict[str, Any] = {}
+        if self.config.api_key:
+            kwargs["api_key"] = self.config.api_key
+        if self.config.base_url:
+            kwargs["base_url"] = self.config.base_url
+        client = anthropic.Anthropic(**kwargs)
+        history = [_normalize_anthropic_message(message) for message in messages]
+        active_tools = tools or self.tools
+        calls = 0
+        for _ in range(self.max_rounds):
+            response = client.messages.create(
+                model=self.config.resolved_model(),
+                max_tokens=2048,
+                system="You are a mobile device assistant. Reply naturally, or call one device tool when needed. Never claim an action succeeded unless its tool result confirms it.",
+                messages=history,
+                tools=[_anthropic_tool(spec) for spec in [*ACTION_TOOL_SPECS, *READ_TOOL_SPECS]],
+                timeout=self.config.timeout,
+            )
+            blocks = getattr(response, "content", []) or []
+            text_parts = [getattr(block, "text", "") for block in blocks if getattr(block, "type", None) == "text"]
+            thinking_parts = [getattr(block, "thinking", "") for block in blocks if getattr(block, "type", None) == "thinking"]
+            uses = [block for block in blocks if getattr(block, "type", None) == "tool_use"]
+            self._log({"type": "llm_response", "content": "\n".join(x for x in text_parts if x), "thinking": "\n".join(x for x in thinking_parts if x), "tool_calls": [{"id": u.id, "name": u.name, "arguments": u.input} for u in uses]})
+            if not uses:
+                content = "\n".join(x for x in text_parts if x)
+                self._log({"type": "final_reply", "content": content})
+                return ChatResult(content=content, tool_calls=calls)
+            history.append({"role": "assistant", "content": [_anthropic_block_dict(block) for block in blocks if getattr(block, "type", None) != "thinking"]})
+            results = []
+            for use in uses:
+                arguments = dict(use.input) if isinstance(use.input, dict) else {}
+                try:
+                    result = active_tools[use.name](**arguments)
+                except Exception as exc:
+                    result = {"ok": False, "error": str(exc)}
+                calls += 1
+                self._log({"type": "tool_result", "tool_call_id": use.id, "tool_name": use.name, "arguments": arguments, "result": result})
+                results.append({"type": "tool_result", "tool_use_id": use.id, "content": json.dumps(result, ensure_ascii=True, default=str)})
+            history.append({"role": "user", "content": results})
+        raise RuntimeError("conversation exceeded the maximum tool-call rounds")
+
+    def chat_for_device(self, device_id: str, messages: list[dict[str, Any]]) -> ChatResult:
+        """Run a conversation bound to exactly one device."""
+        device_tools = {
+            name: _bind_device(tool, device_id, name=name)
+            for name, tool in self.tools.items()
+        }
+        return self.chat(messages, tools=device_tools)
+
     def _log(self, event: dict[str, Any]) -> None:
         if self.event_logger is not None:
             try:
@@ -141,6 +194,37 @@ def _openai_tool(spec: ToolSpec) -> dict[str, Any]:
     }
 
 
+def _anthropic_tool(spec: ToolSpec) -> dict[str, Any]:
+    return {"name": spec.name, "description": spec.description, "input_schema": spec.parameters}
+
+
+def _anthropic_block_dict(block: Any) -> dict[str, Any]:
+    block_type = getattr(block, "type", "")
+    if block_type == "text":
+        return {"type": "text", "text": getattr(block, "text", "")}
+    if block_type == "thinking":
+        return {"type": "thinking", "thinking": getattr(block, "thinking", "")}
+    return {"type": "tool_use", "id": block.id, "name": block.name, "input": block.input}
+
+
+def _normalize_anthropic_message(message: dict[str, Any]) -> dict[str, Any]:
+    normalized = _normalize_message(message)
+    content = normalized["content"]
+    if isinstance(content, str):
+        return normalized
+    blocks = []
+    for block in content:
+        if block.get("type") == "text":
+            blocks.append(block)
+        elif block.get("type") == "image_url":
+            url = block.get("image_url", {}).get("url", "")
+            if isinstance(url, str) and url.startswith("data:"):
+                header, data = url.split(",", 1)
+                media_type = header[5:].split(";", 1)[0]
+                blocks.append({"type": "image", "source": {"type": "base64", "media_type": media_type, "data": data}})
+    return {"role": normalized["role"], "content": blocks}
+
+
 def _message_dict(message: Any) -> dict[str, Any]:
     result: dict[str, Any] = {"role": "assistant"}
     content = getattr(message, "content", None)
@@ -192,3 +276,16 @@ def _normalize_message(message: dict[str, Any]) -> dict[str, Any]:
         blocks.append({"type": "image_url", "image_url": {"url": f"data:{mime};base64,{image}"}})
         return {"role": role, "content": blocks}
     raise ValueError("message content must be text, image blocks, or image_base64")
+
+
+def _bind_device(tool: Callable[..., Any], device_id: str, *, name: str) -> Callable[..., Any]:
+    def bound(**arguments: Any) -> Any:
+        if name == "list_devices":
+            return tool(**arguments)
+        requested = arguments.get("device_id")
+        if requested is not None and requested != device_id:
+            raise ValueError(f"conversation is bound to device {device_id}")
+        arguments["device_id"] = device_id
+        return tool(**arguments)
+
+    return bound

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

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import json
+import sqlite3
 from datetime import UTC, datetime
 from pathlib import Path
 from threading import Lock
@@ -8,17 +9,35 @@ from typing import Any
 
 
 class ConversationLogStore:
-    """Append-only local audit log for chat and tool activity."""
+    """SQLite-backed local audit log for chat and tool activity."""
 
     def __init__(self, path: str | Path) -> None:
         self.path = Path(path)
         self._lock = Lock()
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+        with self._connect() as connection:
+            connection.execute("CREATE TABLE IF NOT EXISTS conversation_events (id INTEGER PRIMARY KEY AUTOINCREMENT, occurred_at TEXT NOT NULL, event_type TEXT NOT NULL, payload_json TEXT NOT NULL)")
 
     def append(self, event: dict[str, Any]) -> None:
-        record = {"timestamp": datetime.now(UTC).isoformat(), **_safe(event)}
-        self.path.parent.mkdir(parents=True, exist_ok=True)
-        with self._lock, self.path.open("a", encoding="utf-8") as stream:
-            stream.write(json.dumps(record, ensure_ascii=True, default=str) + "\n")
+        with self._lock, self._connect() as connection:
+            connection.execute("INSERT INTO conversation_events (occurred_at, event_type, payload_json) VALUES (?, ?, ?)", (datetime.now(UTC).isoformat(), str(event.get("type") or "event"), json.dumps(_safe(event), ensure_ascii=True, default=str)))
+
+    def list_recent(self, *, limit: int = 200) -> list[dict[str, Any]]:
+        with self._connect() as connection:
+            rows = connection.execute("SELECT id, occurred_at, event_type, payload_json FROM conversation_events ORDER BY id DESC LIMIT ?", (max(1, min(limit, 1000)),)).fetchall()
+        result = []
+        for row in rows:
+            try:
+                payload = json.loads(row["payload_json"])
+            except (TypeError, ValueError):
+                payload = {}
+            result.append({"id": row["id"], "occurred_at": row["occurred_at"], "event_type": row["event_type"], **payload})
+        return result
+
+    def _connect(self) -> sqlite3.Connection:
+        connection = sqlite3.connect(self.path)
+        connection.row_factory = sqlite3.Row
+        return connection
 
 
 def _safe(value: Any) -> Any:

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

@@ -21,6 +21,7 @@ from host_agent.client import (
 )
 from host_agent.config import HostAgentConfig
 from host_agent.conversation import ConversationAgent
+from host_agent.conversation_log import ConversationLogStore
 from host_agent.devices import register_local_device, unregister_local_device
 from host_agent.history import ConsoleHistoryStore
 from host_agent.identity import HostIdentityStore
@@ -234,6 +235,7 @@ def create_console_app(
     mcp_token_store: McpTokenStore | None = None,
     mcp_busy_tracker: McpBusyTracker | None = None,
     conversation_agent: ConversationAgent | None = None,
+    conversation_log: ConversationLogStore | None = None,
 ) -> FastAPI:
     app = FastAPI(title="Host Agent Console")
     cookie_secure = config.console_bind_host not in _LOOPBACK_BIND_HOSTS
@@ -425,6 +427,11 @@ def create_console_app(
         if conversation_agent is None:
             raise HTTPException(status_code=503, detail="chat agent is not configured")
         payload = await request.json()
+        device_id = payload.get("device_id") if isinstance(payload, dict) else None
+        if not isinstance(device_id, str) or not device_id.strip():
+            raise HTTPException(status_code=400, detail="device_id is required")
+        if device_id not in {device.id for device in manager.list_devices()}:
+            raise HTTPException(status_code=404, detail="unknown device")
         raw_messages = payload.get("messages") if isinstance(payload, dict) else None
         if not isinstance(raw_messages, list) or not raw_messages:
             raise HTTPException(status_code=400, detail="messages must be a non-empty list")
@@ -437,7 +444,22 @@ def create_console_app(
         ]
         if not messages:
             raise HTTPException(status_code=400, detail="messages are invalid")
-        result = await asyncio.to_thread(conversation_agent.chat, messages)
+        if conversation_log is not None:
+            await asyncio.to_thread(
+                conversation_log.append,
+                {"type": "user_request", "device_id": device_id.strip(), "messages": messages},
+            )
+        try:
+            result = await asyncio.to_thread(
+                conversation_agent.chat_for_device, device_id.strip(), messages
+            )
+        except Exception as exc:
+            if conversation_log is not None:
+                await asyncio.to_thread(
+                    conversation_log.append,
+                    {"type": "agent_error", "device_id": device_id.strip(), "error": str(exc)},
+                )
+            raise HTTPException(status_code=502, detail=str(exc)) from exc
         return JSONResponse(
             {"content": result.content, "tool_calls": result.tool_calls}
         )
@@ -605,6 +627,15 @@ def create_console_app(
             entries=entries,
         )
 
+    @app.get("/conversations", response_class=HTMLResponse)
+    async def conversations_page(
+        session: SessionState = Depends(require_session),
+    ) -> HTMLResponse:
+        events = await asyncio.to_thread(
+            conversation_log.list_recent if conversation_log is not None else (lambda: [])
+        )
+        return _render("conversations.html", title="Conversations", session=session, events=events)
+
     def _tasks_list_context(
         session: SessionState,
         *,

apps/device-host-agent/host_agent/web/templates/base.html

@@ -26,6 +26,7 @@ form.inline { display: inline; margin: 0; }
       <a href="/tasks">Tasks</a>
       <a href="/account">Account</a>
       <a href="/history">History</a>
+      <a href="/conversations">Conversations</a>
       <form class="inline" method="post" action="/logout">
         <input type="hidden" name="csrf_token" value="{{ session.csrf_token }}">
         <button type="submit">Logout</button>

apps/device-host-agent/host_agent/web/templates/conversations.html

@@ -0,0 +1,14 @@
+{% extends "base.html" %}
+{% block body %}
+<h1>Conversations</h1>
+{% if not events %}<p>No conversation activity recorded yet.</p>{% endif %}
+{% for event in events %}
+<article>
+  <h2>{{ event["event_type"] }} <small>{{ event["occurred_at"] }}</small></h2>
+  {% if event.get("content") %}<pre>{{ event["content"] }}</pre>{% endif %}
+  {% if event.get("thinking") %}<details><summary>LLM reasoning</summary><pre>{{ event["thinking"] }}</pre></details>{% endif %}
+  {% if event.get("tool_calls") %}<details open><summary>Tool calls</summary><pre>{{ event["tool_calls"] | tojson(indent=2) }}</pre></details>{% endif %}
+  {% if event.get("tool_name") %}<p><strong>{{ event["tool_name"] }}</strong></p><pre>{{ event.get("arguments") | tojson(indent=2) }}</pre><pre>{{ event.get("result") | tojson(indent=2) }}</pre>{% endif %}
+</article>
+{% endfor %}
+{% endblock %}

runtime/ai_planner.py

@@ -36,12 +36,19 @@ class AIPlanner(Planner):
         world: "WorldState | None" = None,
         screenshot: bytes | None = None,
     ) -> list[PlannedStep]:
+        scene_json = _without_ocr(scene.to_dict()) if self.config.multimodal else scene.to_dict()
         user_prompt = planner_user_prompt(
             goal=goal,
-            scene_json=scene.to_dict(),
+            scene_json=scene_json,
             history_summary=_history_summary(world),
             device_platform=context.device_platform,
         )
+        if context.step_results and not context.step_results[-1].success:
+            user_prompt += (
+                "\n\nThe previous tool call failed. Diagnose the failure and choose a "
+                "corrected action or finish the task if it cannot proceed.\n"
+                f"Previous failure: {context.step_results[-1].error or 'unknown error'}"
+            )
         decision = self.client.decide(
             system_prompt=PLANNER_SYSTEM_PROMPT,
             user_prompt=user_prompt,
@@ -94,3 +101,16 @@ def _history_summary(world: "WorldState | None") -> list[dict[str, Any]]:
         }
         for event in world.history
     ]
+
+
+def _without_ocr(scene_json: dict[str, Any]) -> dict[str, Any]:
+    cleaned = dict(scene_json)
+    elements = cleaned.get("elements")
+    if isinstance(elements, list):
+        cleaned["elements"] = [
+            {key: value for key, value in element.items() if key not in {"source", "confidence", "foreground_color", "background_color"}}
+            for element in elements
+            if isinstance(element, dict) and element.get("source") != "ocr"
+        ]
+    cleaned.pop("ocr_elements", None)
+    return cleaned

runtime/planner_config.py

@@ -19,6 +19,7 @@ TIMEOUT_ENV = "AI_PLANNER_TIMEOUT_SECONDS"
 THINKING_BUDGET_ENV = "AI_PLANNER_THINKING_BUDGET_TOKENS"
 API_KEY_ENV = "AI_PLANNER_API_KEY"
 BASE_URL_ENV = "AI_PLANNER_BASE_URL"
+MULTIMODAL_ENV = "AI_PLANNER_MULTIMODAL"
 
 SUPPORTED_PROVIDERS = frozenset(DEFAULT_MODEL_BY_PROVIDER)
 
@@ -32,6 +33,7 @@ class PlannerConfig:
     thinking_budget_tokens: int | None = None
     api_key: str | None = None
     base_url: str | None = None
+    multimodal: bool = False
 
     def resolved_model(self) -> str:
         return self.model or DEFAULT_MODEL_BY_PROVIDER[self.provider]
@@ -47,6 +49,7 @@ def load_config(env: Mapping[str, str] | None = None) -> PlannerConfig:
         thinking_budget_tokens=_parse_thinking_budget(values.get(THINKING_BUDGET_ENV)),
         api_key=values.get(API_KEY_ENV) or _provider_key(values),
         base_url=values.get(BASE_URL_ENV) or None,
+        multimodal=_parse_bool(values.get(MULTIMODAL_ENV), default=False),
     )
 
 

runtime/task.py

@@ -187,13 +187,18 @@ class TaskRunner:
                         "failed",
                         result.error or "step failed",
                     )
-                    self._update_task(
-                        task,
-                        status="failed",
-                        completed=True,
-                        failure_reason=result.error or "step failed",
-                    )
-                    return task
+                    # AI planners can use the structured failure feedback to
+                    # correct malformed arguments or choose another action.
+                    # Deterministic planners retain their fail-fast behavior.
+                    if self.planner.__class__.__name__ != "AIPlanner":
+                        self._update_task(
+                            task,
+                            status="failed",
+                            completed=True,
+                            failure_reason=result.error or "step failed",
+                        )
+                        return task
+                    break
 
         self._emit_step_progress(
             len(context.step_results),

storage/task_metadata.py

@@ -133,4 +133,24 @@ class TaskMetadataStore:
     def _connect(self) -> sqlite3.Connection:
         connection = sqlite3.connect(self.db_path)
         connection.row_factory = sqlite3.Row
+        # A previous interrupted startup can leave a zero-byte SQLite file
+        # behind before ``__init__`` reaches ``_ensure_schema``. Ensure the
+        # table exists on every connection so the console can recover without
+        # manual deletion or database repair.
+        connection.execute(
+            """
+            create table if not exists tasks (
+                id text primary key,
+                goal text not null,
+                device_id text not null,
+                status text not null,
+                created_at text not null,
+                updated_at text not null,
+                completed_at text,
+                failure_reason text,
+                source_task_id text,
+                source_attempt integer
+            )
+            """
+        )
         return connection

tests/test_task_metadata.py

@@ -0,0 +1,10 @@
+from storage.task_metadata import TaskMetadataStore
+
+
+def test_store_recovers_from_preexisting_empty_database(tmp_path) -> None:
+    path = tmp_path / "task_progress.sqlite3"
+    path.touch()
+
+    store = TaskMetadataStore(path)
+
+    assert store.list_tasks() == []