diff --git a/docs/CONSTITUTION.md b/docs/CONSTITUTION.md index 75626fd..4ef340a 100644 --- a/docs/CONSTITUTION.md +++ b/docs/CONSTITUTION.md @@ -21,7 +21,13 @@ driver such as `WDADriver`. ## Perception Boundary -`Scene` is the only perception artifact the LLM sees. It is produced through +`Scene` is the only perception artifact the LLM sees, with one narrow, +explicit exception: the runtime-layer AI `Planner` (and only that Planner) +may additionally receive the raw screenshot bytes for the current step, +alongside `Scene`, to support vision-grounded decision-making. No other +layer — `api`, `tools`, `perception`, `storage`, or any other LLM consumer — +may receive raw screenshot bytes; every other perception consumer still +receives `Scene` only. `Scene` itself is still produced exclusively through `PerceptionProvider`, not by direct calls to OCR, UI tree parsing, or `scene_builder` from runtime and API layers. diff --git a/openspec/changes/ai-planner-runtime/.openspec.yaml b/openspec/changes/ai-planner-runtime/.openspec.yaml new file mode 100644 index 0000000..8803b47 --- /dev/null +++ b/openspec/changes/ai-planner-runtime/.openspec.yaml @@ -0,0 +1,2 @@ +schema: spec-driven +created: 2026-07-12 diff --git a/openspec/changes/ai-planner-runtime/design.md b/openspec/changes/ai-planner-runtime/design.md new file mode 100644 index 0000000..fe79ce6 --- /dev/null +++ b/openspec/changes/ai-planner-runtime/design.md @@ -0,0 +1,289 @@ +## Context + +`docs/ROADMAP.md` places the Planner inside Milestone 3 (Agent Runtime): +"Planner/Executor structure, tool execution, retry behavior, and task +runner," already implemented by `apex-agent-mvp`. What `apex-agent-mvp` +actually shipped for the Planner half is `runtime/planner.py::Planner` — a +stub that returns one hardcoded `describe_screen` step, then `[]` forever. +Every later milestone built real capability around this stub (`Executor` +retry/backoff, `TaskContext` memory, `semantic-scene` enrichment, +`world-model` cross-step state) without ever giving the loop a real decision +-maker. This change closes that gap: an `AIPlanner` that uses native LLM +tool/function calling to choose one grounded action per turn, wired into +`TaskRunner` behind a default-off flag so applying this change does not +change any existing task's behavior, cost, or dependencies unless explicitly +enabled. + +Two decisions were fixed before design work started (not re-litigated here): +a **pluggable dual-provider** abstraction (Anthropic tool use and OpenAI +function calling, switchable via config) rather than a single hardcoded +provider, and **Scene + screenshot** (vision multimodal) as the Planner's +perception input rather than Scene-only — the latter requires the Constitution +amendment covered in Decision D5. + +## Goals / Non-Goals + +**Goals:** +- Replace the stub decision logic with a real LLM call that chooses exactly + one tool (action or `finish_task`) per `plan()` invocation, using the + current `Scene`, optionally the current screenshot, and recent + `WorldState` history as input. +- Support both Anthropic and OpenAI as interchangeable providers behind one + internal `ToolCallingClient` interface, selected by configuration alone. +- Keep the change strictly additive and default-disabled: with + `AI_PLANNER_ENABLED` unset (or false), `TaskRunner`'s constructed planner, + control flow, and dependencies are byte-for-byte the same as before this + change. +- Make task completion and task failure both explicit, model-driven signals + (`finish_task(success, reason)`) rather than inferring completion from the + absence of a tool call. +- Preserve every Constitution invariant except the one explicitly amended + (Perception Boundary), and amend that one as narrowly as the vision + requirement allows. + +**Non-Goals:** +- No multi-step lookahead planning — `AIPlanner.plan()` always returns 0 or 1 + step; `TaskRunner.run()` already re-observes and re-plans every loop + iteration, so pre-planning multiple steps ahead would only drift from the + actual screen state. +- No `wait`/no-op tool. Forcing "exactly one tool call per turn" means the + model cannot express "do nothing this turn" — a real but low-severity gap + (a spurious action self-corrects next turn once the loop re-observes). + Candidate for a v1.1 follow-up, not this change. +- No dynamic tool subset per task or app — the six-tool face + (`ACTION_TOOL_SPECS` + `finish_task`) is fixed. +- No `find_text_on_screen`/`find_icon_on_screen`/`describe_screen_semantic` + exposed as Planner tools — the per-turn `Scene` JSON already enumerates + every element's id/type/text/bounds, so a "search for X" tool call would + only add latency and hallucination surface without adding information the + model doesn't already have. +- No automatic retry of a failed LLM call inside `AIPlanner` or + `ToolCallingClient` — a transient failure surfaces as a failed task via the + `TaskRunner.run()` fix in D7, matching the project's existing convention + that retry is `Executor`'s concern for tool execution, not the Planner's + concern for its own decision calls. +- No provider/model selection exposed as an API request parameter — v1 is a + deployment-time/environment-variable choice only. +- No backfill of `openspec/specs/agent-runtime/`, which does not exist + because `apex-agent-mvp`'s `agent-runtime` capability was never archived + into `openspec/specs/`. This is a pre-existing gap unrelated to this + change; this change's `specs/agent-runtime/spec.md` uses `## ADDED + Requirements` against that unarchived baseline, the same way + `semantic-scene-runtime` and `world-model-runtime` each did. + +## Decisions + +### D1: Single-step, ReAct-style `plan()` — never multi-step +`AIPlanner.plan()` returns a list of 0 or 1 `PlannedStep`. `TaskRunner.run()` +already calls `observer()` then `_plan()` fresh on every loop iteration +(`runtime/task.py`), so a Planner that tried to hand back several steps at +once would either have those extra steps silently ignored by the loop's +current per-iteration contract, or require changing that contract to consume +a queue — both worse than just re-deciding every turn against the freshly +observed `Scene`. +**Alternative considered**: return a short queued plan (e.g. up to 3 steps) +and only re-plan when a step's expectation is violated. Rejected — UI state +can change after any single action (a dialog appears, a keyboard covers an +element), so a queued step is frequently stale by the time it would execute; +single-step keeps every action grounded in the turn's actual `Scene`. + +### D2: Explicit `finish_task(success, reason)` control tool, not an implicit "no call" signal +Task completion and task failure are both signaled by the model calling +`finish_task` — never by the model declining to call any tool (which native +tool-calling APIs do not reliably support as a distinguishable "done" state +across both providers) and never by a heuristic on the Planner's side. +`finish_task(success=True)` maps to `AIPlanner.plan()` returning `[]`, which +reuses `TaskRunner.run()`'s existing `if not steps or ...: return +self._complete_task(task)` short-circuit unchanged. `finish_task(success= +False, reason)` maps to `AIPlanner.plan()` raising `TaskFailedError(reason)` +— reusing `core/errors.py`'s `TaskFailedError`, defined since +`apex-agent-mvp` but never previously constructed anywhere in the codebase. +These two outcomes are deliberately routed through different mechanisms +(empty list vs. exception) rather than both returning `[]` with a status +flag, because `run()`'s short-circuit condition is `if not steps or +self.planner.goal_reached(...)` — a bare empty list on failure would be +silently read as success by that exact line. +**Alternative considered**: signal success via `goal_reached()` returning +`True`. Rejected — see D3. + +### D3: `AIPlanner.goal_reached()` always returns `False` +This is a required invariant, not a style choice. `run()` evaluates `if not +steps or self.planner.goal_reached(...)` — the `or` means `goal_reached()` is +still consulted even when `steps` is non-empty. If `goal_reached()` could +ever return `True` on a turn where `plan()` also returned a real action, that +action would be silently discarded and the task would be marked complete one +turn early. Routing every completion/failure signal exclusively through +`finish_task` (D2) removes any reason for `goal_reached()` to do anything; +it is a permanent no-op override, documented as such at the call site. + +### D4: Dual-provider abstraction via `Protocol`, forced single-tool-call on both +`runtime/tool_calling_client.py` defines `ToolCallingClient` as a +structural `Protocol` (matching the existing precedent of +`semantic/enricher.py::SemanticLLMClient` and +`skills_learning/embeddings.py::EmbeddingClient` — no ABC is used anywhere in +this codebase for a swappable single-method client), with concrete +`AnthropicToolCallingClient` and `OpenAIToolCallingClient` implementations +selected by `build_client(config)` based on `AI_PLANNER_PROVIDER`. Both +implementations force the API to return exactly one tool call per turn — +Anthropic via `tool_choice={"type": "any", "disable_parallel_tool_use": +True}`, OpenAI via `tool_choice="required"` plus the top-level +`parallel_tool_calls=False` — so a single `ToolCallDecision` can always be +parsed deterministically regardless of provider. Both implementations use +the lazy-import-plus-injectable-transport pattern already established by +`semantic/llm_client.py::AnthropicSemanticClient` (SDK imported only inside a +`_client()` method; constructor accepts an optional `transport` for tests), +and wrap every SDK/network/parse failure into one internal +`ToolCallUnavailable` exception, mirroring `EnrichmentUnavailable`. +**Alternative considered**: a single hardcoded Anthropic-only client (the +simpler default for a first LLM-driven Planner). Explicitly rejected by +product decision before this design was drafted — dual-provider +pluggability is a requirement, not a nice-to-have, for this change. + +### D5: Perception Boundary amendment — narrow, Planner-only screenshot exception +The Constitution's Perception Boundary previously stated `Scene` is the +*only* perception artifact the LLM sees. Vision-grounded action selection +(precise tap/swipe coordinates, disambiguating visually-similar elements +that OCR/UI-tree fusion can conflate) requires the Planner to also see the +raw screenshot for the current step — a deliberate product decision, not an +oversight. The amendment is written to be as narrow as that requirement +actually is: it names one consumer (`the runtime-layer AI Planner, and only +that Planner`) and explicitly reaffirms that every other layer/consumer +(`api`, `tools`, `perception`, `storage`, or any other LLM consumer) still +never receives raw screenshot bytes, and that `Scene` itself is still +produced exclusively through `PerceptionProvider`. See "Constitution +Compliance" below for the exact before/after text. +**Alternative considered**: keep `Scene` as the Planner's only input +(Scene-only, no vision). Explicitly rejected by product decision before this +design was drafted, for the reasons above. +**Alternative considered**: broaden the exception to "any runtime-layer LLM +consumer" instead of naming the Planner specifically, anticipating future +vision consumers. Rejected — Change Discipline requires each change to +justify its own scope; a hypothetical future consumer should justify its own +amendment when it exists, not inherit a pre-approved blank check today. + +### D6: Default-disabled via `AI_PLANNER_ENABLED`, mirroring existing LLM-feature flags +`PlannerConfig.enabled` defaults to `False`, and `load_config()` reads +`AI_PLANNER_ENABLED` the same way `semantic/config.py`, +`skills_learning/config.py`, and `agents/config.py` gate their own +LLM-backed behavior — off unless explicitly turned on. (`world/config.py`'s +`WorldConfig.enabled` defaults `True`, but that capability makes zero LLM +calls and is not a counterexample to this convention.) `TaskRunner`'s +`_default_planner()` follows the same "self-loaded config, enabled flag +picks the implementation" shape already used for `world_model` and +`on_task_succeeded` (`runtime/task.py`), so no change to `api/rest.py` or any +other caller of `TaskRunner()` is required to preserve current behavior. + +### D7: `TaskRunner.run()` gains a try/except around observe+plan +Before this change, `run()` had no error handling around +`self.observer(...)` or `self._plan(...)` — safe only because the stub +`Planner` never raised. `AIPlanner` can raise for real reasons (network +failure, malformed provider response, `TaskFailedError` from D2's +`finish_task(success=False)` path), so `run()` now catches any exception +from that per-iteration observe-then-plan sequence and marks the task +`status="failed"` with `failure_reason=f"{type(exc).__name__}: {exc}"`, +returning immediately. Two failure modes are avoided by doing this +explicitly rather than skipping it: an uncaught exception leaving a +`BackgroundTasks`-invoked task stuck at `status="running"` forever, and a +catch-and-return-`[]` degrade pattern (as used by +`semantic/enricher.py::enrich_scene()`) that would be misread as task +success by the same `if not steps or ...` short-circuit discussed in D2/D3. +**Alternative considered**: copy `enrich_scene()`'s catch-and-degrade-to-None +pattern. Rejected — that pattern is correct for an optional enrichment pass +where the caller has a defined fallback (use the raw `Scene`). The Planner +has no such fallback: if it cannot decide, the task cannot proceed, and +pretending otherwise (via an empty list) means silent, incorrect "success." + +### D8: `_planner_accepts_world()` generalized to `_planner_accepts(name)` +`TaskRunner` already used `inspect.signature` reflection to decide whether to +pass a `world=` kwarg to `Planner.plan()`, so that narrow-signature test +Planners (e.g. `NoWorldPlanner`) would not receive an unexpected keyword +argument. This change generalizes that one method to take the parameter name +as an argument and reuses it for both `"world"` and the new `"screenshot"` +kwarg, preserving backward compatibility with any existing `Planner` +subclass that does not declare a `screenshot` parameter (or `**kwargs`). + +## Constitution Compliance + +- **Device Boundary**: unaffected — this change adds no driver code and does + not touch `driver/`/`device/`. +- **Tool Boundary**: unaffected — `AIPlanner` calls tools exclusively through + `PlannedStep` → `Executor`, the same as the stub `Planner`; no new code + imports a concrete driver or bypasses `tools/`. +- **Perception Boundary**: amended, narrowly, per D5. The invariant that + `Scene` is the only perception artifact for every consumer *except* the + named Planner is preserved and made explicit in the same sentence as the + exception, rather than being loosened globally. +- **Runtime Boundary**: preserved and, in fact, fulfilled by this change — + the existing text already states "LLM dependencies enter at `runtime` + through Planner behavior," which is exactly what `AIPlanner` is. The + Executor remains the only component that calls tools and handles + tool-execution retries; `AIPlanner` does not call tools directly or + implement its own retry loop (D7 fails the task instead of retrying). +- **Change Discipline**: the dependency direction (`core -> driver/device -> + tools -> perception -> storage -> runtime -> api`) is preserved — all new + code lives in `runtime/`, imports flow downward only (`runtime/ai_planner.py` + imports from `runtime/`, `core/`; nothing in `core`, `driver`, `device`, or + `tools` imports anything LLM-related). The new external integration (the + OpenAI SDK, newly *used* though already declared as a dependency) is added + at the adapter layer that owns the concern (`runtime/tool_calling_client.py`), + not at the domain or device boundary. + +## Risks / Trade-offs + +- **[Risk] Per-step LLM call adds latency and cost to every planning turn** + → Mitigation: default-disabled (D6); `AI_PLANNER_TIMEOUT_SECONDS` bounds + worst-case latency; no in-`AIPlanner` retry (D7) means a slow/failing + provider fails fast instead of compounding delay. +- **[Risk] Hallucinated tool arguments (e.g. tap coordinates outside any + element's bounds)** → Mitigation: the system prompt instructs the model to + ground every coordinate in the current turn's `Scene` element bounds (and + screenshot, when present); a bad tap still degrades gracefully into an + ordinary failed/retried step through `Executor`'s existing mechanism, + unchanged by this design. +- **[Risk] Provider wire-format drift (Anthropic/OpenAI SDK or API changes) + silently breaks request construction or response parsing** → Mitigation: + all format-specific logic is isolated into small, independently unit- + tested functions (`_anthropic_tool`, `_openai_tool`, + `_decision_from_anthropic_response`, `_decision_from_openai_response`) + against fake transports, so a drift shows up as a specific, localized test + failure rather than a silent behavior change. +- **[Risk] Two providers could behave inconsistently** (e.g. one honors + `finish_task` semantics more reliably than the other) → Mitigation: both + are constrained by the identical `ALL_TOOL_SPECS` JSON Schema and the same + system prompt; provider-specific behavior differences are a model-quality + concern to observe via the optional real-integration test, not something + this design can fully eliminate structurally. +- **[Risk] Perception Boundary amendment could be read as a precedent for + loosening the invariant further** → Mitigation: the amendment text itself + names exactly one consumer and restates the "no one else" constraint in + the same breath (D5); any future consumer needs its own amendment. + +## Migration Plan + +Purely additive; no data migration: +1. Add the five new `runtime/*` files and the `runtime/planner.py`/ + `runtime/task.py` edits described in Impact. With `AI_PLANNER_ENABLED` + unset, `TaskRunner()` continues to construct the stub `Planner` exactly as + before. +2. Amend `docs/CONSTITUTION.md`'s Perception Boundary section (D5). +3. No changes to `storage/`'s timeline format, `api/rest.py`, or `api/mcp.py`. +4. Rollback is simply leaving `AI_PLANNER_ENABLED` unset/false, or reverting + the changed files; no other capability depends on this one. +5. Enabling in a real environment requires setting `AI_PLANNER_ENABLED=true`, + `AI_PLANNER_PROVIDER` (`anthropic` or `openai`), and the corresponding + provider's API key in the process environment (already-existing SDK + convention, not a new config surface this change introduces). + +## Open Questions + +- Whether a `wait`/no-op tool is actually needed in practice, or whether the + self-correcting-next-turn behavior of a spurious action is good enough + indefinitely — deferred to real usage observation, not decided here. +- Whether `AI_PLANNER_TIMEOUT_SECONDS`'s default (30s) is well-tuned against + real provider latency for image-bearing requests — needs tuning against + real usage data once this is enabled somewhere with real traffic; not + fixed by this design. +- Whether provider/model selection should eventually move from + environment-variable/deployment-time to a per-task or per-request choice — + leaning toward "not until a concrete need appears" (YAGNI), consistent + with `semantic-scene-runtime`'s D4 reasoning for its own model choice. diff --git a/openspec/changes/ai-planner-runtime/proposal.md b/openspec/changes/ai-planner-runtime/proposal.md new file mode 100644 index 0000000..672abe1 --- /dev/null +++ b/openspec/changes/ai-planner-runtime/proposal.md @@ -0,0 +1,106 @@ +## Why + +`runtime/planner.py::Planner` — the Planner half of Milestone 3 (Agent +Runtime)'s `agent-runtime` capability, implemented by `apex-agent-mvp` and +still the only Planner in the codebase — is a stub: it always returns exactly +one hardcoded `describe_screen` step on a task's first call, then `[]` on +every call after any step has executed, regardless of the goal or the current +`Scene`. `TaskRunner`'s Observe→Plan→Act→Observe loop, `Executor`'s +retry/backoff, `TaskContext`'s per-task memory, and the already-implemented +`semantic-scene` and `world-model` capabilities are all real and wired +end-to-end — every later milestone (Task Memory, Semantic Scene, World Model, +Skill Learning, Workflow Orchestration, Multi-Agent Runtime) has been built on +top of, or alongside, this stub without ever replacing it. Nothing in the +runtime actually decides what to do. This change gives the runtime its first +real decision-maker: an LLM-driven Planner that uses native tool/function +calling to choose exactly one grounded action per turn, completing what +Milestone 3 always intended the Planner role to be. + +## What Changes + +- Add `runtime/ai_planner.py::AIPlanner`, a `Planner` implementation that + calls an LLM with native tool/function calling once per `plan()` + invocation, translating the model's single chosen tool call into 0 or 1 + `PlannedStep`. Single-step, ReAct-style: `TaskRunner.run()` already + re-observes and re-plans every iteration, so `AIPlanner` never attempts + multi-step lookahead. +- Add a **pluggable dual-provider** LLM abstraction + (`runtime/tool_calling_client.py`): both Anthropic native tool use and + OpenAI function calling are supported, selected via configuration + (`AI_PLANNER_PROVIDER`), each forced to return exactly one tool call per + turn so a response always resolves to a single, unambiguous decision. +- Add a fixed, six-tool tool-face (`runtime/tool_specs.py`): `tap`, `swipe`, + `input_text`, `launch_app`, `terminate_app` for action, plus an explicit + `finish_task(success, reason)` control tool the model calls to end the task + (on success or failure) instead of relying on an ambiguous "no tool call" + signal. +- **Amend the Perception Boundary invariant** (`docs/CONSTITUTION.md`) with + one narrow, explicit exception: the AI `Planner` (only that Planner) may + additionally receive the current step's raw screenshot bytes alongside + `Scene`, to support vision-grounded action grounding (precise tap/swipe + coordinates). No other layer or LLM consumer gains access to raw screenshot + bytes; `Scene` remains the only perception artifact everywhere else. +- Add configuration (`runtime/planner_config.py`: `AI_PLANNER_ENABLED`, + `AI_PLANNER_PROVIDER`, `AI_PLANNER_MODEL`, `AI_PLANNER_TIMEOUT_SECONDS`) + defaulting `AI_PLANNER_ENABLED=False`, matching the existing enable/disable + convention used by every other LLM-backed capability (`semantic-scene`, + `skill-learning`). `runtime/task.py::TaskRunner` builds an `AIPlanner` by + default only when enabled; otherwise its existing stub `Planner` behavior + is unchanged. +- Fix a latent correctness gap in `TaskRunner.run()`'s loop, exposed by + giving the Planner a real chance to fail: observe/plan exceptions are now + caught per iteration and turned into a `status="failed"` task with a + `failure_reason`, instead of propagating uncaught (previously harmless only + because the stub Planner never raised). +- **BREAKING**: none. Default-disabled; when disabled, `TaskRunner`'s + constructed `Planner` and control flow are unchanged from before this + change. + +## Capabilities + +### New Capabilities + +(none — this change fulfills the Planner role already scoped by Milestone +3's `agent-runtime` capability; see Impact for why no new capability is +declared) + +### Modified Capabilities + +- `agent-runtime`: the Planner requirement ("given a goal and the current + Scene, produces steps") gains a real, LLM-backed implementation with + native tool calling, a defined single-action-per-turn contract, an + explicit task-completion/failure signal (`finish_task`), and — as a + Planner-only exception to the Perception Boundary — optional access to the + current step's screenshot. `agent-runtime`'s base spec was never archived + into `openspec/specs/` (a pre-existing gap from `apex-agent-mvp`, out of + scope for this change); this change's delta spec below uses `## ADDED + Requirements` against that as-yet-unarchived baseline, the same way + `semantic-scene-runtime` and `world-model-runtime` each layered their own + delta on top of it without attempting to backfill it (see `design.md`). + +## Impact + +- **New files**: `runtime/planner_config.py`, `runtime/tool_specs.py`, + `runtime/tool_calling_client.py`, `runtime/planner_prompts.py`, + `runtime/ai_planner.py`. +- **Modified**: `runtime/planner.py` (base `Planner.plan()` gains an optional + `screenshot` parameter, stub behavior unchanged), `runtime/task.py` + (`TaskRunner` builds an `AIPlanner` when enabled; generalized its existing + `_planner_accepts_world()` reflection helper to also conditionally inject + `screenshot`; wrapped the observe+plan step in try/except), + `docs/CONSTITUTION.md` (Perception Boundary amendment described above). +- **No change** to `api/rest.py` — `TaskRunner()`'s existing zero-argument + construction picks up the new behavior automatically once + `AI_PLANNER_ENABLED=true` is set in the environment; no new request + parameters, no MCP surface change (`api/mcp.py` is a separate, curated + tool-handler surface unaffected by this change). +- **No new dependencies**: `pyproject.toml` already declares both + `anthropic` and `openai` SDKs (the former added for + `semantic-scene-runtime`); this change is the first to actually construct + an OpenAI client. +- **Out of scope**: no `wait`/no-op tool (v1.1 candidate, see `design.md`); + no dynamic tool subset selection; no provider/model choice exposed as an + API request parameter; no automatic retry of failed LLM calls inside + `AIPlanner` (transient failures surface as a failed task; tool-execution + retry remains solely `Executor`'s concern, unchanged by this change); no + backfill of the missing `openspec/specs/agent-runtime/` base spec. diff --git a/openspec/changes/ai-planner-runtime/specs/agent-runtime/spec.md b/openspec/changes/ai-planner-runtime/specs/agent-runtime/spec.md new file mode 100644 index 0000000..9cfc686 --- /dev/null +++ b/openspec/changes/ai-planner-runtime/specs/agent-runtime/spec.md @@ -0,0 +1,90 @@ +## ADDED Requirements + +### Requirement: LLM-driven Planner selects exactly one grounded action per turn +The system SHALL provide a Planner implementation that, given a goal, the +current Scene, and recent task history, uses native LLM tool/function +calling to select exactly one action (or the completion signal defined +below) per `plan()` invocation, grounding any coordinates in the current +turn's Scene element bounds. + +#### Scenario: Planner selects a single action for the current turn +- **WHEN** the AI Planner is invoked with a goal and the current Scene +- **THEN** it returns at most one `PlannedStep`, whose action and arguments + come from exactly one tool call chosen by the underlying LLM for that turn + +#### Scenario: Planner re-decides every turn from the current Scene +- **WHEN** the AI Planner is invoked again after a prior step has executed +- **THEN** its decision is grounded in the newly observed Scene for that + turn, not in coordinates or assumptions carried over from a previous turn + +### Requirement: Explicit finish_task completion and failure signal +The system SHALL treat task completion and task failure as explicit, +model-driven signals via a dedicated `finish_task(success, reason)` tool, +rather than inferring either outcome from the model declining to call any +tool. + +#### Scenario: Model signals successful completion +- **WHEN** the model calls `finish_task` with `success=True` +- **THEN** the Planner returns an empty step list and the task is marked + completed + +#### Scenario: Model signals it cannot complete the goal +- **WHEN** the model calls `finish_task` with `success=False` and a `reason` +- **THEN** the task is marked failed with that reason, without attempting + any further planning steps + +### Requirement: Pluggable dual-provider tool-calling abstraction +The system SHALL support at least two interchangeable LLM providers +(Anthropic native tool use and OpenAI function calling) for the AI Planner's +decision calls, selectable via configuration, with both providers +constrained to return exactly one tool call per request. + +#### Scenario: Provider selected via configuration +- **WHEN** the AI Planner is configured with a given provider identifier +- **THEN** it constructs and uses the tool-calling client for that provider + without requiring any change to `AIPlanner`'s own decision logic + +#### Scenario: Provider response resolves to a single decision +- **WHEN** either supported provider returns a response to a tool-calling + request +- **THEN** the response is parsed into exactly one tool name and one + arguments object, regardless of which provider produced it + +### Requirement: AI Planner is disabled by default and additive to the existing Planner +The system SHALL default to the existing non-LLM Planner unless the AI +Planner is explicitly enabled via configuration, and SHALL NOT alter the +existing Planner's behavior, dependencies, or any caller's construction of +`TaskRunner` when left disabled. + +#### Scenario: AI Planner disabled (default) +- **WHEN** `TaskRunner` is constructed without an explicit `planner` and + without the AI Planner enabled in configuration +- **THEN** it uses the existing non-LLM Planner, unchanged from before this + capability existed + +#### Scenario: AI Planner enabled via configuration +- **WHEN** `TaskRunner` is constructed without an explicit `planner` and with + the AI Planner enabled in configuration +- **THEN** it uses the AI Planner, configured with the selected provider and + model + +### Requirement: Screenshot access is a Planner-only, narrow exception to the Perception Boundary +The system SHALL allow the AI Planner, and only the AI Planner, to receive +the current step's raw screenshot bytes alongside the Scene for +vision-grounded decision-making, while every other perception consumer +SHALL continue to receive only the Scene. + +#### Scenario: Planner receives both Scene and screenshot +- **WHEN** a screenshot for the current step is available +- **THEN** the AI Planner's decision call includes both the Scene JSON and + the raw screenshot bytes for that step + +#### Scenario: Screenshot unavailable does not block planning +- **WHEN** a screenshot for the current step cannot be obtained +- **THEN** the AI Planner still produces a decision using the Scene alone, + and this is not treated as a task failure + +#### Scenario: No other consumer receives raw screenshot bytes +- **WHEN** any component other than the AI Planner (for example, `api`, + `tools`, `perception`, or `storage`) consumes perception output +- **THEN** it receives only the Scene, never raw screenshot bytes diff --git a/openspec/changes/ai-planner-runtime/tasks.md b/openspec/changes/ai-planner-runtime/tasks.md new file mode 100644 index 0000000..c6ff25d --- /dev/null +++ b/openspec/changes/ai-planner-runtime/tasks.md @@ -0,0 +1,93 @@ +## 1. Configuration and tool face + +- [x] 1.1 Create `runtime/planner_config.py`: `PlannerConfig` (`enabled`, + `provider`, `model`, `timeout`), `load_config()` reading + `AI_PLANNER_ENABLED`/`AI_PLANNER_PROVIDER`/`AI_PLANNER_MODEL`/ + `AI_PLANNER_TIMEOUT_SECONDS`, `enabled` defaulting `False` +- [x] 1.2 Create `runtime/tool_specs.py`: vendor-neutral `ToolSpec` dataclass + plus `TAP_SPEC`/`SWIPE_SPEC`/`INPUT_TEXT_SPEC`/`LAUNCH_APP_SPEC`/ + `TERMINATE_APP_SPEC`/`FINISH_TASK_SPEC`, matching real tool signatures, + exported as `ACTION_TOOL_SPECS` (5) and `ALL_TOOL_SPECS` (6) +- [x] 1.3 Create `runtime/planner_prompts.py`: `PLANNER_SYSTEM_PROMPT` and + `planner_user_prompt(*, goal, scene_json, history_summary)` + +## 2. Dual-provider tool-calling client + +- [x] 2.1 Create `runtime/tool_calling_client.py`: `ToolCallDecision`, + `ToolCallUnavailable`, `ToolCallingClient` Protocol +- [x] 2.2 Implement `AnthropicToolCallingClient` (lazy SDK import, injectable + transport, forced single tool call via + `tool_choice={"type": "any", "disable_parallel_tool_use": True}`, + image content block when a screenshot is present) +- [x] 2.3 Implement `OpenAIToolCallingClient` (lazy SDK import, injectable + transport, `max_completion_tokens`, forced single tool call via + `tool_choice="required"` + `parallel_tool_calls=False`, `image_url` + data-URI block when a screenshot is present) +- [x] 2.4 Implement `build_client(config)` provider selection + +## 3. AI Planner and runtime wiring + +- [x] 3.1 Create `runtime/ai_planner.py::AIPlanner` (single-step decision, + `finish_task` → empty plan / `TaskFailedError`, `goal_reached()` + permanently `False`) +- [x] 3.2 Add `screenshot: bytes | None = None` to base + `runtime/planner.py::Planner.plan()` +- [x] 3.3 Wire `runtime/task.py::TaskRunner`: `planner_config` constructor + param, `_default_planner()` (AI Planner when enabled, stub otherwise), + generalize `_planner_accepts_world()` into `_planner_accepts(name)`, + add `_planning_screenshot()`, conditionally inject `screenshot=` in + `_plan()` +- [x] 3.4 Wrap `TaskRunner.run()`'s per-iteration observe+plan in try/except, + marking the task `status="failed"` with a `failure_reason` on any + exception instead of propagating it uncaught + +## 4. Constitution amendment + +- [x] 4.1 Amend `docs/CONSTITUTION.md`'s Perception Boundary section with the + narrow, Planner-only screenshot exception + +## 5. openspec change artifacts + +- [x] 5.1 Write `proposal.md`, `design.md` (with explicit Constitution + Compliance section), `tasks.md` +- [x] 5.2 Write `specs/agent-runtime/spec.md` (`## ADDED Requirements` only) + +## 6. Tests + +- [x] 6.1 `tests/test_planner_config.py`: env var parsing, defaults, + `enabled` defaults `False`, invalid/negative timeout falls back to + default +- [x] 6.2 `tests/test_tool_specs.py`: each `ToolSpec.parameters` schema + (required fields, `additionalProperties: False`), `ALL_TOOL_SPECS` has + exactly 6 entries +- [x] 6.3 `tests/test_tool_calling_client.py`: fake-transport tests per + provider — forced single-tool-call fields present, image block + present/absent based on screenshot, successful response parses to + `ToolCallDecision`, malformed/error response raises + `ToolCallUnavailable` +- [x] 6.4 `tests/test_ai_planner.py`: fake `ToolCallingClient` — action + decision → single `PlannedStep`; `finish_task(success=True)` → `[]`; + `finish_task(success=False, reason=...)` → raises `TaskFailedError`; + `goal_reached()` always `False` +- [x] 6.5 Added `tests/test_ai_planner_task_runner.py` (kept + `tests/test_task_loop.py` untouched rather than extending it): planner + exception mid-task → task ends `status="failed"` with a + `failure_reason` (not stuck `running`); observer exception is caught + the same way; a narrow-signature Planner does not receive an + unexpected `screenshot` kwarg while one that declares it does; + `PlannerConfig(enabled=False)` (default) still yields the stub + `Planner`; `PlannerConfig(enabled=True)` without an explicit + `planner=` yields an `AIPlanner` +- [x] 6.6 `tests/test_ai_planner_integration.py` (`@pytest.mark.integration`, + skipped without a real API key): one real call per provider + +## 7. Verification + +- [x] 7.1 Ran `pytest -m "not integration"`: 314 passed, 6 deselected (up + from the pre-existing 269-test baseline plus the 45 new tests added by + this change) +- [x] 7.2 Confirmed default-off behavior: no `AI_PLANNER_*` environment + variables set in the shell → `TaskRunner()` constructs + `planner_config=PlannerConfig(enabled=False, ...)` and + `type(runner.planner) is Planner`; `uvicorn api.rest:create_app + --factory` starts cleanly with no API key configured diff --git a/runtime/ai_planner.py b/runtime/ai_planner.py new file mode 100644 index 0000000..98ae069 --- /dev/null +++ b/runtime/ai_planner.py @@ -0,0 +1,73 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING, Any + +from core.errors import TaskFailedError +from core.models import Scene +from runtime.context import TaskContext +from runtime.planner import PlannedStep, Planner +from runtime.planner_config import PlannerConfig, load_config +from runtime.planner_prompts import PLANNER_SYSTEM_PROMPT, planner_user_prompt +from runtime.tool_calling_client import ToolCallingClient, build_client +from runtime.tool_specs import ALL_TOOL_SPECS + +if TYPE_CHECKING: + from world.models import WorldState + +FINISH_TASK_TOOL = "finish_task" + + +class AIPlanner(Planner): + def __init__( + self, + *, + client: ToolCallingClient | None = None, + config: PlannerConfig | None = None, + ) -> None: + self.config = config or load_config() + self.client = client or build_client(self.config) + + def plan( + self, + *, + goal: str, + scene: Scene, + context: TaskContext, + world: "WorldState | None" = None, + screenshot: bytes | None = None, + ) -> list[PlannedStep]: + decision = self.client.decide( + system_prompt=PLANNER_SYSTEM_PROMPT, + user_prompt=planner_user_prompt( + goal=goal, + scene_json=scene.to_dict(), + history_summary=_history_summary(world), + ), + screenshot=screenshot, + tools=ALL_TOOL_SPECS, + timeout=self.config.timeout, + ) + + if decision.tool_name == FINISH_TASK_TOOL: + if decision.arguments.get("success"): + return [] + raise TaskFailedError(decision.arguments.get("reason") or "task failed") + + return [ + PlannedStep( + action=decision.tool_name, + description=f"AI planner: {decision.tool_name}({decision.arguments})", + args=dict(decision.arguments), + ) + ] + + def goal_reached(self, *, goal: str, scene: Scene, context: TaskContext) -> bool: + # Completion is signaled exclusively via the finish_task tool call + # (mapped to an empty plan above), never via this hook. + return False + + +def _history_summary(world: "WorldState | None") -> list[dict[str, Any]]: + if world is None: + return [] + return [event.to_dict() for event in world.history] diff --git a/runtime/planner.py b/runtime/planner.py index 901b52a..cc0b01f 100644 --- a/runtime/planner.py +++ b/runtime/planner.py @@ -26,6 +26,7 @@ class Planner: scene: Scene, context: TaskContext, world: "WorldState | None" = None, + screenshot: bytes | None = None, ) -> list[PlannedStep]: if context.step_results: return [] diff --git a/runtime/planner_config.py b/runtime/planner_config.py new file mode 100644 index 0000000..e291486 --- /dev/null +++ b/runtime/planner_config.py @@ -0,0 +1,63 @@ +from __future__ import annotations + +import os +from collections.abc import Mapping +from dataclasses import dataclass + +DEFAULT_PROVIDER = "anthropic" +DEFAULT_MODEL_BY_PROVIDER = { + "anthropic": "claude-sonnet-5", + "openai": "gpt-5.6", +} +DEFAULT_TIMEOUT_SECONDS = 30.0 + +ENABLED_ENV = "AI_PLANNER_ENABLED" +PROVIDER_ENV = "AI_PLANNER_PROVIDER" +MODEL_ENV = "AI_PLANNER_MODEL" +TIMEOUT_ENV = "AI_PLANNER_TIMEOUT_SECONDS" + +SUPPORTED_PROVIDERS = frozenset(DEFAULT_MODEL_BY_PROVIDER) + + +@dataclass(frozen=True) +class PlannerConfig: + enabled: bool = False + provider: str = DEFAULT_PROVIDER + model: str = "" + timeout: float = DEFAULT_TIMEOUT_SECONDS + + def resolved_model(self) -> str: + return self.model or DEFAULT_MODEL_BY_PROVIDER[self.provider] + + +def load_config(env: Mapping[str, str] | None = None) -> PlannerConfig: + values = env or os.environ + return PlannerConfig( + enabled=_parse_bool(values.get(ENABLED_ENV), default=False), + provider=_parse_provider(values.get(PROVIDER_ENV)), + model=values.get(MODEL_ENV) or "", + timeout=_parse_timeout(values.get(TIMEOUT_ENV)), + ) + + +def _parse_bool(value: str | None, *, default: bool) -> bool: + if value is None: + return default + return value.strip().lower() in {"1", "true", "yes", "on", "enabled"} + + +def _parse_provider(value: str | None) -> str: + if value is None: + return DEFAULT_PROVIDER + provider = value.strip().lower() + return provider if provider in SUPPORTED_PROVIDERS else DEFAULT_PROVIDER + + +def _parse_timeout(value: str | None) -> float: + if value is None: + return DEFAULT_TIMEOUT_SECONDS + try: + timeout = float(value) + except ValueError: + return DEFAULT_TIMEOUT_SECONDS + return timeout if timeout > 0 else DEFAULT_TIMEOUT_SECONDS diff --git a/runtime/planner_prompts.py b/runtime/planner_prompts.py new file mode 100644 index 0000000..38eb7c7 --- /dev/null +++ b/runtime/planner_prompts.py @@ -0,0 +1,42 @@ +from __future__ import annotations + +import json +from typing import Any + +PLANNER_SYSTEM_PROMPT = """You are the planning brain of a mobile device automation agent. + +Each turn you are given a goal, the current screen as a structured Scene (a +list of UI elements with id, type, text, and pixel bounds), and — when +available — a screenshot of the same screen and a short history of recent +actions and their outcomes. + +You must call exactly one tool per turn: +- One of `tap`, `swipe`, `input_text`, `launch_app`, `terminate_app` to make + progress toward the goal. +- `finish_task` when the goal has been reached, or when it cannot be reached + and no further action would help. + +Ground every coordinate you choose in the Scene element bounds (and the +screenshot, if provided) for the current turn only — never reuse coordinates +from history, since the screen may have changed. Only call `finish_task` with +`success=True` when the current Scene shows the goal has actually been +reached. Call it with `success=False` and a clear `reason` if you are stuck, +repeating the same action without progress, or the goal is not achievable. +""" + + +def planner_user_prompt( + *, + goal: str, + scene_json: dict[str, Any], + history_summary: list[dict[str, Any]], +) -> str: + return ( + "Goal:\n" + f"{goal}\n\n" + "Current Scene (JSON):\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." + ) diff --git a/runtime/task.py b/runtime/task.py index 380d55f..2269c08 100644 --- a/runtime/task.py +++ b/runtime/task.py @@ -6,9 +6,11 @@ from dataclasses import dataclass, replace from inspect import Parameter, signature from core.models import Scene, Task, utc_now +from runtime.ai_planner import AIPlanner from runtime.context import TaskContext from runtime.executor import Executor from runtime.planner import PlannedStep, Planner +from runtime.planner_config import PlannerConfig, load_config as load_planner_config from semantic.models import SemanticScene from skills_learning.config import ( SkillAuthoringConfig, @@ -56,8 +58,10 @@ class TaskRunner: skill_authoring_config: SkillAuthoringConfig | None = None, skill_store: SkillStore | None = None, skill_embedding_client: EmbeddingClient | None = None, + planner_config: PlannerConfig | None = None, ) -> None: - self.planner = planner or Planner() + self.planner_config = planner_config or load_planner_config() + self.planner = planner or self._default_planner() self.executor = executor or Executor() self.timeline = timeline self.metadata_store = metadata_store @@ -93,9 +97,20 @@ class TaskRunner: self._update_task(task, status="running") for _ in range(self.config.max_steps): - scene = self.observer(task.device_id) - context.add_scene(scene) - steps = self._plan(task.goal, scene, context) + try: + scene = self.observer(task.device_id) + context.add_scene(scene) + screenshot = self._planning_screenshot(task.device_id) + steps = self._plan(task.goal, scene, context, screenshot=screenshot) + except Exception as exc: + reason = f"{type(exc).__name__}: {exc}" if str(exc) else type(exc).__name__ + self._update_task( + task, + status="failed", + completed=True, + failure_reason=reason, + ) + return task if not steps or self.planner.goal_reached( goal=task.goal, scene=scene, @@ -195,31 +210,41 @@ class TaskRunner: model_name=self.skill_authoring_config.embedding_model, ) + def _default_planner(self) -> Planner: + if self.planner_config.enabled: + return AIPlanner(config=self.planner_config) + return Planner() + def _plan( self, goal: str, scene: Scene, context: TaskContext, + screenshot: bytes | None = None, ) -> list[PlannedStep]: - if context.world is not None and self._planner_accepts_world(): - return self.planner.plan( - goal=goal, - scene=scene, - context=context, - world=context.world, - ) - return self.planner.plan(goal=goal, scene=scene, context=context) + kwargs: dict[str, object] = {} + if context.world is not None and self._planner_accepts("world"): + kwargs["world"] = context.world + if screenshot is not None and self._planner_accepts("screenshot"): + kwargs["screenshot"] = screenshot + return self.planner.plan(goal=goal, scene=scene, context=context, **kwargs) - def _planner_accepts_world(self) -> bool: + def _planner_accepts(self, name: str) -> bool: try: parameters = signature(self.planner.plan).parameters except (TypeError, ValueError): return True - return "world" in parameters or any( + return name in parameters or any( parameter.kind is Parameter.VAR_KEYWORD for parameter in parameters.values() ) + def _planning_screenshot(self, device_id: str) -> bytes | None: + try: + return self.screenshot_provider(device_id) + except Exception: + return None + def _update_world( self, world_handle: TaskWorldView | None, @@ -255,10 +280,7 @@ class TaskRunner: ) -> None: if not self.timeline: return - try: - screenshot = self.screenshot_provider(task.device_id) - except Exception: - screenshot = None + screenshot = self._planning_screenshot(task.device_id) self.timeline.append( task_id=task.id, scene=scene.to_dict(), diff --git a/runtime/tool_calling_client.py b/runtime/tool_calling_client.py new file mode 100644 index 0000000..7e99255 --- /dev/null +++ b/runtime/tool_calling_client.py @@ -0,0 +1,298 @@ +from __future__ import annotations + +import base64 +import json +from dataclasses import dataclass +from typing import Any, Protocol + +from runtime.planner_config import PlannerConfig +from runtime.tool_specs import ToolSpec + + +class ToolCallUnavailable(Exception): + """Internal signal for expected tool-calling transport/response failures.""" + + +@dataclass(frozen=True) +class ToolCallDecision: + tool_name: str + arguments: dict[str, Any] + + +class ToolCallingClient(Protocol): + def decide( + self, + *, + system_prompt: str, + user_prompt: str, + screenshot: bytes | None, + tools: list[ToolSpec], + timeout: float, + ) -> ToolCallDecision: ... + + +class AnthropicToolCallingClient: + def __init__( + self, + *, + model: str, + transport: Any | None = None, + max_tokens: int = 1024, + ) -> None: + self.model = model + self._transport = transport + self.max_tokens = max_tokens + + def decide( + self, + *, + system_prompt: str, + user_prompt: str, + screenshot: bytes | None, + tools: list[ToolSpec], + timeout: float, + ) -> ToolCallDecision: + try: + response = self._create_message( + system_prompt, + user_prompt, + screenshot, + tools, + timeout=timeout, + ) + return _decision_from_anthropic_response(response) + except ToolCallUnavailable: + raise + except Exception as exc: + raise ToolCallUnavailable(str(exc)) from exc + + def _create_message( + self, + system_prompt: str, + user_prompt: str, + screenshot: bytes | None, + tools: list[ToolSpec], + *, + timeout: float, + ) -> Any: + client = self._client() + kwargs = { + "model": self.model, + "max_tokens": self.max_tokens, + "timeout": timeout, + "system": [ + { + "type": "text", + "text": system_prompt, + "cache_control": {"type": "ephemeral"}, + } + ], + "messages": [ + { + "role": "user", + "content": _anthropic_content(user_prompt, screenshot), + } + ], + "tools": [_anthropic_tool(spec) for spec in tools], + "tool_choice": {"type": "any", "disable_parallel_tool_use": True}, + } + messages = getattr(client, "messages", None) + if messages is not None: + return messages.create(**kwargs) + return client.create(**kwargs) + + def _client(self) -> Any: + if self._transport is not None: + return self._transport + + try: + import anthropic + except Exception as exc: + raise ToolCallUnavailable("anthropic SDK is unavailable") from exc + + self._transport = anthropic.Anthropic() + return self._transport + + +class OpenAIToolCallingClient: + def __init__( + self, + *, + model: str, + transport: Any | None = None, + max_tokens: int = 1024, + ) -> None: + self.model = model + self._transport = transport + self.max_tokens = max_tokens + + def decide( + self, + *, + system_prompt: str, + user_prompt: str, + screenshot: bytes | None, + tools: list[ToolSpec], + timeout: float, + ) -> ToolCallDecision: + try: + response = self._create_completion( + system_prompt, + user_prompt, + screenshot, + tools, + timeout=timeout, + ) + return _decision_from_openai_response(response) + except ToolCallUnavailable: + raise + except Exception as exc: + raise ToolCallUnavailable(str(exc)) from exc + + def _create_completion( + self, + system_prompt: str, + user_prompt: str, + screenshot: bytes | None, + tools: list[ToolSpec], + *, + timeout: float, + ) -> Any: + client = self._client() + kwargs = { + "model": self.model, + "max_completion_tokens": self.max_tokens, + "timeout": timeout, + "messages": [ + {"role": "system", "content": system_prompt}, + {"role": "user", "content": _openai_content(user_prompt, screenshot)}, + ], + "tools": [_openai_tool(spec) for spec in tools], + "tool_choice": "required", + "parallel_tool_calls": False, + } + chat = getattr(client, "chat", None) + if chat is not None: + return chat.completions.create(**kwargs) + return client.create(**kwargs) + + def _client(self) -> Any: + if self._transport is not None: + return self._transport + + try: + from openai import OpenAI + except Exception as exc: + raise ToolCallUnavailable("openai SDK is unavailable") from exc + + self._transport = OpenAI() + return self._transport + + +def build_client(config: PlannerConfig) -> ToolCallingClient: + model = config.resolved_model() + if config.provider == "openai": + return OpenAIToolCallingClient(model=model) + return AnthropicToolCallingClient(model=model) + + +def _anthropic_content( + user_prompt: str, + screenshot: bytes | None, +) -> list[dict[str, Any]]: + content: list[dict[str, Any]] = [] + if screenshot is not None: + content.append( + { + "type": "image", + "source": { + "type": "base64", + "media_type": "image/png", + "data": base64.b64encode(screenshot).decode("ascii"), + }, + } + ) + content.append({"type": "text", "text": user_prompt}) + return content + + +def _anthropic_tool(spec: ToolSpec) -> dict[str, Any]: + return { + "name": spec.name, + "description": spec.description, + "input_schema": spec.parameters, + } + + +def _decision_from_anthropic_response(response: Any) -> ToolCallDecision: + content = _value(response, "content") + if not isinstance(content, list): + raise ValueError("anthropic tool-call response missing content list") + for block in content: + if _value(block, "type") != "tool_use": + continue + name = _value(block, "name") + arguments = _value(block, "input") + if isinstance(name, str) and isinstance(arguments, dict): + return ToolCallDecision(tool_name=name, arguments=arguments) + raise ValueError("anthropic response did not include a tool_use block") + + +def _openai_content( + user_prompt: str, + screenshot: bytes | None, +) -> str | list[dict[str, Any]]: + if screenshot is None: + return user_prompt + encoded = base64.b64encode(screenshot).decode("ascii") + return [ + {"type": "text", "text": user_prompt}, + { + "type": "image_url", + "image_url": {"url": f"data:image/png;base64,{encoded}"}, + }, + ] + + +def _openai_tool(spec: ToolSpec) -> dict[str, Any]: + return { + "type": "function", + "function": { + "name": spec.name, + "description": spec.description, + "parameters": spec.parameters, + }, + } + + +def _decision_from_openai_response(response: Any) -> ToolCallDecision: + choices = _value(response, "choices") + if not isinstance(choices, list) or not choices: + raise ValueError("openai tool-call response missing choices") + message = _value(choices[0], "message") + tool_calls = _value(message, "tool_calls") + if not isinstance(tool_calls, list) or not tool_calls: + raise ValueError("openai response did not include a tool call") + function = _value(tool_calls[0], "function") + name = _value(function, "name") + if not isinstance(name, str): + raise ValueError("openai tool call missing a function name") + arguments = _decode_openai_arguments(_value(function, "arguments")) + return ToolCallDecision(tool_name=name, arguments=arguments) + + +def _decode_openai_arguments(raw_arguments: Any) -> dict[str, Any]: + if isinstance(raw_arguments, dict): + return raw_arguments + if isinstance(raw_arguments, str): + decoded = json.loads(raw_arguments) + if isinstance(decoded, dict): + return decoded + raise ValueError("openai tool call arguments must decode to a JSON object") + + +def _value(source: Any, key: str) -> Any: + if isinstance(source, dict): + return source.get(key) + value = getattr(source, key, None) + return None if callable(value) else value diff --git a/runtime/tool_specs.py b/runtime/tool_specs.py new file mode 100644 index 0000000..5565ec7 --- /dev/null +++ b/runtime/tool_specs.py @@ -0,0 +1,129 @@ +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any + + +@dataclass(frozen=True) +class ToolSpec: + name: str + description: str + parameters: dict[str, Any] + + +TAP_SPEC = ToolSpec( + name="tap", + description="Tap a point on the screen, given in Scene pixel coordinates.", + parameters={ + "type": "object", + "additionalProperties": False, + "required": ["x", "y"], + "properties": { + "x": {"type": "number", "description": "X coordinate in Scene pixel space."}, + "y": {"type": "number", "description": "Y coordinate in Scene pixel space."}, + }, + }, +) + +SWIPE_SPEC = ToolSpec( + name="swipe", + description=( + "Swipe from a start point to an end point on the screen, given in " + "Scene pixel coordinates." + ), + parameters={ + "type": "object", + "additionalProperties": False, + "required": ["start_x", "start_y", "end_x", "end_y"], + "properties": { + "start_x": {"type": "number", "description": "Start X coordinate."}, + "start_y": {"type": "number", "description": "Start Y coordinate."}, + "end_x": {"type": "number", "description": "End X coordinate."}, + "end_y": {"type": "number", "description": "End Y coordinate."}, + "duration_ms": { + "type": "integer", + "description": "Swipe duration in milliseconds.", + "default": 500, + }, + }, + }, +) + +INPUT_TEXT_SPEC = ToolSpec( + name="input_text", + description="Type text into the currently focused input field.", + parameters={ + "type": "object", + "additionalProperties": False, + "required": ["text"], + "properties": { + "text": {"type": "string", "description": "Text to type."}, + }, + }, +) + +LAUNCH_APP_SPEC = ToolSpec( + name="launch_app", + description="Launch (foreground) an app by its bundle/package identifier.", + parameters={ + "type": "object", + "additionalProperties": False, + "required": ["app_id"], + "properties": { + "app_id": { + "type": "string", + "description": "App bundle/package identifier.", + }, + }, + }, +) + +TERMINATE_APP_SPEC = ToolSpec( + name="terminate_app", + description="Terminate a running app by its bundle/package identifier.", + parameters={ + "type": "object", + "additionalProperties": False, + "required": ["app_id"], + "properties": { + "app_id": { + "type": "string", + "description": "App bundle/package identifier.", + }, + }, + }, +) + +FINISH_TASK_SPEC = ToolSpec( + name="finish_task", + description=( + "Signal that the task is finished: either the goal has been reached, " + "or it cannot be reached and no further steps should be attempted. " + "Call this instead of any other tool once you are done." + ), + parameters={ + "type": "object", + "additionalProperties": False, + "required": ["success", "reason"], + "properties": { + "success": { + "type": "boolean", + "description": "True if the goal was reached, False otherwise.", + }, + "reason": { + "type": "string", + "description": "Short explanation of why the task is finished.", + }, + }, + }, +) + +ACTION_TOOL_SPECS: list[ToolSpec] = [ + TAP_SPEC, + SWIPE_SPEC, + INPUT_TEXT_SPEC, + LAUNCH_APP_SPEC, + TERMINATE_APP_SPEC, +] + +ALL_TOOL_SPECS: list[ToolSpec] = [*ACTION_TOOL_SPECS, FINISH_TASK_SPEC] diff --git a/tests/test_ai_planner.py b/tests/test_ai_planner.py new file mode 100644 index 0000000..a31db7e --- /dev/null +++ b/tests/test_ai_planner.py @@ -0,0 +1,116 @@ +from __future__ import annotations + +from typing import Any + +import pytest + +from core.errors import TaskFailedError +from core.models import Bounds, Scene, SceneElement +from runtime.ai_planner import AIPlanner +from runtime.context import TaskContext +from runtime.planner import PlannedStep +from runtime.planner_config import PlannerConfig +from runtime.tool_calling_client import ToolCallDecision +from runtime.tool_specs import ALL_TOOL_SPECS + + +class FakeToolCallingClient: + def __init__(self, decision: ToolCallDecision) -> None: + self.decision = decision + self.calls: list[dict[str, Any]] = [] + + def decide( + self, + *, + system_prompt: str, + user_prompt: str, + screenshot: bytes | None, + tools: list[Any], + timeout: float, + ) -> ToolCallDecision: + self.calls.append( + { + "system_prompt": system_prompt, + "user_prompt": user_prompt, + "screenshot": screenshot, + "tools": tools, + "timeout": timeout, + } + ) + return self.decision + + +def _scene() -> Scene: + return Scene( + width=10, + height=20, + elements=[SceneElement(id="send", type="button", text="Send", bounds=Bounds(1, 2, 3, 4))], + ) + + +def _context() -> TaskContext: + return TaskContext(task_id="task-1", goal="send a message") + + +def test_ai_planner_returns_single_planned_step_for_action_decision() -> None: + client = FakeToolCallingClient(ToolCallDecision(tool_name="tap", arguments={"x": 1, "y": 2})) + planner = AIPlanner(client=client) + + steps = planner.plan(goal="send a message", scene=_scene(), context=_context()) + + assert steps == [ + PlannedStep( + action="tap", + description="AI planner: tap({'x': 1, 'y': 2})", + args={"x": 1, "y": 2}, + ) + ] + + +def test_ai_planner_finish_task_success_returns_empty_plan() -> None: + client = FakeToolCallingClient( + ToolCallDecision(tool_name="finish_task", arguments={"success": True, "reason": "done"}) + ) + planner = AIPlanner(client=client) + + steps = planner.plan(goal="send a message", scene=_scene(), context=_context()) + + assert steps == [] + + +def test_ai_planner_finish_task_failure_raises_task_failed_error_with_reason() -> None: + client = FakeToolCallingClient( + ToolCallDecision(tool_name="finish_task", arguments={"success": False, "reason": "stuck on login"}) + ) + planner = AIPlanner(client=client) + + with pytest.raises(TaskFailedError, match="stuck on login"): + planner.plan(goal="send a message", scene=_scene(), context=_context()) + + +def test_ai_planner_finish_task_failure_without_reason_uses_default_message() -> None: + client = FakeToolCallingClient(ToolCallDecision(tool_name="finish_task", arguments={"success": False})) + planner = AIPlanner(client=client) + + with pytest.raises(TaskFailedError, match="task failed"): + planner.plan(goal="send a message", scene=_scene(), context=_context()) + + +def test_ai_planner_goal_reached_is_always_false() -> None: + client = FakeToolCallingClient(ToolCallDecision(tool_name="tap", arguments={"x": 1, "y": 2})) + planner = AIPlanner(client=client) + + assert planner.goal_reached(goal="anything", scene=_scene(), context=_context()) is False + + +def test_ai_planner_forwards_tools_screenshot_and_timeout_to_client() -> None: + client = FakeToolCallingClient(ToolCallDecision(tool_name="tap", arguments={"x": 1, "y": 2})) + planner = AIPlanner(client=client, config=PlannerConfig(timeout=12.5)) + + planner.plan(goal="send a message", scene=_scene(), context=_context(), screenshot=b"fake-bytes") + + call = client.calls[0] + assert call["tools"] == ALL_TOOL_SPECS + assert call["screenshot"] == b"fake-bytes" + assert call["timeout"] == 12.5 + assert "send a message" in call["user_prompt"] diff --git a/tests/test_ai_planner_integration.py b/tests/test_ai_planner_integration.py new file mode 100644 index 0000000..8233483 --- /dev/null +++ b/tests/test_ai_planner_integration.py @@ -0,0 +1,56 @@ +from __future__ import annotations + +import os + +import pytest + +from core.models import Bounds, Scene, SceneElement +from runtime.ai_planner import AIPlanner +from runtime.context import TaskContext +from runtime.planner_config import PlannerConfig + + +def _scene() -> Scene: + return Scene( + width=390, + height=844, + elements=[ + SceneElement(id="send", type="button", text="Send", bounds=Bounds(300, 800, 60, 30)), + ], + ) + + +@pytest.mark.integration +def test_real_anthropic_ai_planner_selects_a_tool() -> None: + if not os.environ.get("ANTHROPIC_API_KEY"): + pytest.skip("ANTHROPIC_API_KEY is required for AI planner integration test") + try: + import anthropic # noqa: F401 + except ImportError: + pytest.skip("anthropic SDK is not installed") + + planner = AIPlanner(config=PlannerConfig(enabled=True, provider="anthropic", timeout=15.0)) + context = TaskContext(task_id="task", goal="tap the send button") + + steps = planner.plan(goal="tap the send button", scene=_scene(), context=context) + + assert isinstance(steps, list) + assert len(steps) <= 1 + + +@pytest.mark.integration +def test_real_openai_ai_planner_selects_a_tool() -> None: + if not os.environ.get("OPENAI_API_KEY"): + pytest.skip("OPENAI_API_KEY is required for AI planner integration test") + try: + import openai # noqa: F401 + except ImportError: + pytest.skip("openai SDK is not installed") + + planner = AIPlanner(config=PlannerConfig(enabled=True, provider="openai", timeout=15.0)) + context = TaskContext(task_id="task", goal="tap the send button") + + steps = planner.plan(goal="tap the send button", scene=_scene(), context=context) + + assert isinstance(steps, list) + assert len(steps) <= 1 diff --git a/tests/test_ai_planner_task_runner.py b/tests/test_ai_planner_task_runner.py new file mode 100644 index 0000000..ce77353 --- /dev/null +++ b/tests/test_ai_planner_task_runner.py @@ -0,0 +1,132 @@ +from __future__ import annotations + +from core.models import Bounds, Scene, SceneElement, Task +from runtime.ai_planner import AIPlanner +from runtime.executor import Executor, ExecutorConfig +from runtime.planner import PlannedStep, Planner +from runtime.planner_config import PlannerConfig +from runtime.task import TaskRunner, TaskRunnerConfig +from tests.fakes import PNG_10X20 + + +class RaisingPlanner(Planner): + def __init__(self, error: Exception) -> None: + self.error = error + self.calls = 0 + + def plan(self, *, goal, scene, context): + self.calls += 1 + raise self.error + + def goal_reached(self, *, goal, scene, context): + return False + + +class NarrowSignaturePlanner(Planner): + """Predates the `screenshot` parameter added to the base Planner.plan().""" + + def __init__(self) -> None: + self.calls = 0 + + def plan(self, *, goal, scene, context): + self.calls += 1 + if context.step_results: + return [] + return [PlannedStep(action="tap", description="tap")] + + def goal_reached(self, *, goal, scene, context): + return bool(context.step_results) + + +class ScreenshotRecordingPlanner(Planner): + def __init__(self) -> None: + self.screenshots: list[bytes | None] = [] + + def plan(self, *, goal, scene, context, screenshot=None): + self.screenshots.append(screenshot) + if context.step_results: + return [] + return [PlannedStep(action="tap", description="tap")] + + def goal_reached(self, *, goal, scene, context): + return bool(context.step_results) + + +def _scene() -> Scene: + return Scene( + width=10, + height=20, + elements=[SceneElement(id="send", type="button", text="Send", bounds=Bounds(1, 2, 3, 4))], + ) + + +def _runner(*, planner=None, planner_config=None, observer=None) -> TaskRunner: + return TaskRunner( + planner=planner, + planner_config=planner_config, + executor=Executor( + tools={"tap": lambda **kwargs: {"ok": True}}, + config=ExecutorConfig(max_retries=1, backoff_seconds=0), + ), + config=TaskRunnerConfig(max_steps=5), + observer=observer or (lambda device_id: _scene()), + screenshot_provider=lambda device_id: PNG_10X20, + ) + + +def test_task_runner_marks_task_failed_when_planner_raises() -> None: + planner = RaisingPlanner(RuntimeError("boom")) + runner = _runner(planner=planner) + + result = runner.run(Task(goal="inspect", device_id="phone")) + + assert result.status == "failed" + assert result.failure_reason == "RuntimeError: boom" + assert planner.calls == 1 + + +def test_task_runner_marks_task_failed_when_observer_raises() -> None: + def failing_observer(device_id: str) -> Scene: + raise RuntimeError("no device") + + runner = _runner(planner=Planner(), observer=failing_observer) + + result = runner.run(Task(goal="inspect", device_id="phone")) + + assert result.status == "failed" + assert result.failure_reason == "RuntimeError: no device" + + +def test_task_runner_omits_screenshot_kwarg_for_narrow_signature_planner() -> None: + planner = NarrowSignaturePlanner() + runner = _runner(planner=planner) + + result = runner.run(Task(goal="inspect", device_id="phone")) + + assert result.status == "completed" + assert planner.calls == 2 + + +def test_task_runner_passes_screenshot_to_planner_that_declares_it() -> None: + planner = ScreenshotRecordingPlanner() + runner = _runner(planner=planner) + + result = runner.run(Task(goal="inspect", device_id="phone")) + + assert result.status == "completed" + assert planner.screenshots == [PNG_10X20, PNG_10X20] + + +def test_task_runner_default_planner_is_stub_when_ai_planner_disabled() -> None: + runner = _runner(planner=None, planner_config=PlannerConfig(enabled=False)) + + assert type(runner.planner) is Planner + + +def test_task_runner_default_planner_is_ai_planner_when_enabled() -> None: + runner = _runner( + planner=None, + planner_config=PlannerConfig(enabled=True, provider="anthropic", model="test-model"), + ) + + assert isinstance(runner.planner, AIPlanner) diff --git a/tests/test_planner_config.py b/tests/test_planner_config.py new file mode 100644 index 0000000..e699586 --- /dev/null +++ b/tests/test_planner_config.py @@ -0,0 +1,72 @@ +from __future__ import annotations + +from runtime.planner_config import ( + DEFAULT_MODEL_BY_PROVIDER, + DEFAULT_PROVIDER, + DEFAULT_TIMEOUT_SECONDS, + PlannerConfig, + load_config, +) + +_NO_RELEVANT_VARS = {"UNRELATED": "1"} + + +def test_load_config_defaults_when_unset() -> None: + config = load_config(_NO_RELEVANT_VARS) + + assert config == PlannerConfig( + enabled=False, + provider=DEFAULT_PROVIDER, + model="", + timeout=DEFAULT_TIMEOUT_SECONDS, + ) + assert config.resolved_model() == DEFAULT_MODEL_BY_PROVIDER[DEFAULT_PROVIDER] + + +def test_load_config_parses_enabled_truthy_values() -> None: + for value in ["1", "true", "True", "yes", "on", "enabled"]: + assert load_config({"AI_PLANNER_ENABLED": value}).enabled is True + + +def test_load_config_parses_enabled_falsy_values() -> None: + for value in ["0", "false", "no", "off", ""]: + assert load_config({"AI_PLANNER_ENABLED": value}).enabled is False + + +def test_load_config_selects_provider_and_resolves_default_model() -> None: + config = load_config({"AI_PLANNER_PROVIDER": "openai"}) + + assert config.provider == "openai" + assert config.resolved_model() == "gpt-5.6" + + +def test_load_config_anthropic_default_model() -> None: + config = load_config({"AI_PLANNER_PROVIDER": "anthropic"}) + + assert config.resolved_model() == "claude-sonnet-5" + + +def test_load_config_falls_back_to_default_provider_when_unsupported() -> None: + config = load_config({"AI_PLANNER_PROVIDER": "not-a-real-provider"}) + + assert config.provider == DEFAULT_PROVIDER + + +def test_load_config_model_override_wins_regardless_of_provider() -> None: + config = load_config( + {"AI_PLANNER_PROVIDER": "openai", "AI_PLANNER_MODEL": "custom-model"} + ) + + assert config.resolved_model() == "custom-model" + + +def test_load_config_parses_valid_timeout() -> None: + config = load_config({"AI_PLANNER_TIMEOUT_SECONDS": "12.5"}) + + assert config.timeout == 12.5 + + +def test_load_config_falls_back_to_default_timeout_when_invalid_or_non_positive() -> None: + for value in ["not-a-number", "0", "-5"]: + config = load_config({"AI_PLANNER_TIMEOUT_SECONDS": value, **_NO_RELEVANT_VARS}) + assert config.timeout == DEFAULT_TIMEOUT_SECONDS diff --git a/tests/test_tool_calling_client.py b/tests/test_tool_calling_client.py new file mode 100644 index 0000000..122701e --- /dev/null +++ b/tests/test_tool_calling_client.py @@ -0,0 +1,296 @@ +from __future__ import annotations + +import base64 +from typing import Any + +import pytest + +from runtime.planner_config import PlannerConfig +from runtime.tool_calling_client import ( + AnthropicToolCallingClient, + OpenAIToolCallingClient, + ToolCallDecision, + ToolCallUnavailable, + build_client, +) +from runtime.tool_specs import FINISH_TASK_SPEC, TAP_SPEC +from tests.fakes import PNG_10X20 + + +class FakeMessages: + def __init__(self, *, response: object | None = None, error: Exception | None = None) -> None: + self.response = response + self.error = error + self.calls: list[dict[str, Any]] = [] + + def create(self, **kwargs: Any) -> object: + self.calls.append(kwargs) + if self.error: + raise self.error + return self.response + + +class FakeTransport: + def __init__(self, messages: FakeMessages) -> None: + self.messages = messages + + +class FakeCompletions: + def __init__(self, *, response: object | None = None, error: Exception | None = None) -> None: + self.response = response + self.error = error + self.calls: list[dict[str, Any]] = [] + + def create(self, **kwargs: Any) -> object: + self.calls.append(kwargs) + if self.error: + raise self.error + return self.response + + +class FakeChat: + def __init__(self, completions: FakeCompletions) -> None: + self.completions = completions + + +class FakeOpenAITransport: + def __init__(self, completions: FakeCompletions) -> None: + self.chat = FakeChat(completions) + + +# --- Anthropic --------------------------------------------------------- + + +def test_anthropic_tool_calling_client_sends_forced_single_tool_call_request() -> None: + messages = FakeMessages( + response={"content": [{"type": "tool_use", "name": "tap", "input": {"x": 1, "y": 2}}]} + ) + client = AnthropicToolCallingClient(model="test-model", transport=FakeTransport(messages)) + + decision = client.decide( + system_prompt="system", + user_prompt="user", + screenshot=None, + tools=[TAP_SPEC, FINISH_TASK_SPEC], + timeout=2.5, + ) + + assert decision == ToolCallDecision(tool_name="tap", arguments={"x": 1, "y": 2}) + assert len(messages.calls) == 1 + call = messages.calls[0] + assert call["model"] == "test-model" + assert call["timeout"] == 2.5 + assert call["tool_choice"] == {"type": "any", "disable_parallel_tool_use": True} + assert call["tools"] == [ + {"name": "tap", "description": TAP_SPEC.description, "input_schema": TAP_SPEC.parameters}, + { + "name": "finish_task", + "description": FINISH_TASK_SPEC.description, + "input_schema": FINISH_TASK_SPEC.parameters, + }, + ] + assert call["system"][0]["text"] == "system" + assert call["system"][0]["cache_control"] == {"type": "ephemeral"} + assert call["messages"] == [{"role": "user", "content": [{"type": "text", "text": "user"}]}] + + +def test_anthropic_tool_calling_client_includes_image_block_when_screenshot_present() -> None: + messages = FakeMessages( + response={ + "content": [ + {"type": "tool_use", "name": "finish_task", "input": {"success": True, "reason": "done"}} + ] + } + ) + client = AnthropicToolCallingClient(model="test-model", transport=FakeTransport(messages)) + + client.decide( + system_prompt="system", + user_prompt="user", + screenshot=PNG_10X20, + tools=[FINISH_TASK_SPEC], + timeout=1, + ) + + content = messages.calls[0]["messages"][0]["content"] + assert content[0] == { + "type": "image", + "source": { + "type": "base64", + "media_type": "image/png", + "data": base64.b64encode(PNG_10X20).decode("ascii"), + }, + } + assert content[1] == {"type": "text", "text": "user"} + + +def test_anthropic_tool_calling_client_wraps_transport_errors() -> None: + messages = FakeMessages(error=TimeoutError("timed out")) + client = AnthropicToolCallingClient(model="test-model", transport=FakeTransport(messages)) + + with pytest.raises(ToolCallUnavailable): + client.decide(system_prompt="s", user_prompt="u", screenshot=None, tools=[TAP_SPEC], timeout=1) + + +@pytest.mark.parametrize( + "response", + [ + {"content": []}, + {"content": [{"type": "text", "text": "no tool call"}]}, + {"content": [{"type": "tool_use", "name": "tap", "input": "not-a-dict"}]}, + ], +) +def test_anthropic_tool_calling_client_wraps_malformed_responses(response: object) -> None: + messages = FakeMessages(response=response) + client = AnthropicToolCallingClient(model="test-model", transport=FakeTransport(messages)) + + with pytest.raises(ToolCallUnavailable): + client.decide(system_prompt="s", user_prompt="u", screenshot=None, tools=[TAP_SPEC], timeout=1) + + +# --- OpenAI -------------------------------------------------------------- + + +def test_openai_tool_calling_client_sends_forced_single_tool_call_request() -> None: + completions = FakeCompletions( + response={ + "choices": [ + {"message": {"tool_calls": [{"function": {"name": "tap", "arguments": '{"x": 1, "y": 2}'}}]}} + ] + } + ) + client = OpenAIToolCallingClient(model="test-model", transport=FakeOpenAITransport(completions)) + + decision = client.decide( + system_prompt="system", + user_prompt="user", + screenshot=None, + tools=[TAP_SPEC, FINISH_TASK_SPEC], + timeout=2.5, + ) + + assert decision == ToolCallDecision(tool_name="tap", arguments={"x": 1, "y": 2}) + assert len(completions.calls) == 1 + call = completions.calls[0] + assert call["model"] == "test-model" + assert call["timeout"] == 2.5 + assert call["max_completion_tokens"] == 1024 + assert "max_tokens" not in call + assert call["tool_choice"] == "required" + assert call["parallel_tool_calls"] is False + assert call["tools"] == [ + { + "type": "function", + "function": { + "name": "tap", + "description": TAP_SPEC.description, + "parameters": TAP_SPEC.parameters, + }, + }, + { + "type": "function", + "function": { + "name": "finish_task", + "description": FINISH_TASK_SPEC.description, + "parameters": FINISH_TASK_SPEC.parameters, + }, + }, + ] + assert call["messages"] == [ + {"role": "system", "content": "system"}, + {"role": "user", "content": "user"}, + ] + + +def test_openai_tool_calling_client_includes_image_block_when_screenshot_present() -> None: + completions = FakeCompletions( + response={ + "choices": [ + { + "message": { + "tool_calls": [ + { + "function": { + "name": "finish_task", + "arguments": '{"success": true, "reason": "done"}', + } + } + ] + } + } + ] + } + ) + client = OpenAIToolCallingClient(model="test-model", transport=FakeOpenAITransport(completions)) + + client.decide( + system_prompt="system", + user_prompt="user", + screenshot=PNG_10X20, + tools=[FINISH_TASK_SPEC], + timeout=1, + ) + + user_message = completions.calls[0]["messages"][1] + assert user_message["role"] == "user" + assert user_message["content"][0] == {"type": "text", "text": "user"} + encoded = base64.b64encode(PNG_10X20).decode("ascii") + assert user_message["content"][1] == { + "type": "image_url", + "image_url": {"url": f"data:image/png;base64,{encoded}"}, + } + + +def test_openai_tool_calling_client_accepts_arguments_already_as_dict() -> None: + completions = FakeCompletions( + response={ + "choices": [{"message": {"tool_calls": [{"function": {"name": "tap", "arguments": {"x": 1, "y": 2}}}]}}] + } + ) + client = OpenAIToolCallingClient(model="test-model", transport=FakeOpenAITransport(completions)) + + decision = client.decide(system_prompt="s", user_prompt="u", screenshot=None, tools=[TAP_SPEC], timeout=1) + + assert decision == ToolCallDecision(tool_name="tap", arguments={"x": 1, "y": 2}) + + +def test_openai_tool_calling_client_wraps_transport_errors() -> None: + completions = FakeCompletions(error=TimeoutError("timed out")) + client = OpenAIToolCallingClient(model="test-model", transport=FakeOpenAITransport(completions)) + + with pytest.raises(ToolCallUnavailable): + client.decide(system_prompt="s", user_prompt="u", screenshot=None, tools=[TAP_SPEC], timeout=1) + + +@pytest.mark.parametrize( + "response", + [ + {"choices": []}, + {"choices": [{"message": {"tool_calls": []}}]}, + {"choices": [{"message": {"tool_calls": [{"function": {"name": "tap", "arguments": "not-json"}}]}}]}, + ], +) +def test_openai_tool_calling_client_wraps_malformed_responses(response: object) -> None: + completions = FakeCompletions(response=response) + client = OpenAIToolCallingClient(model="test-model", transport=FakeOpenAITransport(completions)) + + with pytest.raises(ToolCallUnavailable): + client.decide(system_prompt="s", user_prompt="u", screenshot=None, tools=[TAP_SPEC], timeout=1) + + +# --- build_client ---------------------------------------------------------- + + +def test_build_client_selects_provider_and_resolves_default_model() -> None: + anthropic_client = build_client(PlannerConfig(provider="anthropic", model="")) + assert isinstance(anthropic_client, AnthropicToolCallingClient) + assert anthropic_client.model == "claude-sonnet-5" + + openai_client = build_client(PlannerConfig(provider="openai", model="")) + assert isinstance(openai_client, OpenAIToolCallingClient) + assert openai_client.model == "gpt-5.6" + + +def test_build_client_honors_explicit_model_override() -> None: + client = build_client(PlannerConfig(provider="openai", model="gpt-5.6-custom")) + assert client.model == "gpt-5.6-custom" diff --git a/tests/test_tool_specs.py b/tests/test_tool_specs.py new file mode 100644 index 0000000..1d6b1cc --- /dev/null +++ b/tests/test_tool_specs.py @@ -0,0 +1,72 @@ +from __future__ import annotations + +from runtime.tool_specs import ( + ACTION_TOOL_SPECS, + ALL_TOOL_SPECS, + FINISH_TASK_SPEC, + INPUT_TEXT_SPEC, + LAUNCH_APP_SPEC, + SWIPE_SPEC, + TAP_SPEC, + TERMINATE_APP_SPEC, + ToolSpec, +) + + +def test_action_tool_specs_has_five_entries_and_all_tool_specs_adds_finish_task() -> None: + assert len(ACTION_TOOL_SPECS) == 5 + assert len(ALL_TOOL_SPECS) == 6 + assert ALL_TOOL_SPECS == [*ACTION_TOOL_SPECS, FINISH_TASK_SPEC] + assert FINISH_TASK_SPEC not in ACTION_TOOL_SPECS + + +def test_all_tool_spec_names_are_unique() -> None: + names = [spec.name for spec in ALL_TOOL_SPECS] + assert len(names) == len(set(names)) + + +def test_every_tool_spec_schema_forbids_additional_properties() -> None: + for spec in ALL_TOOL_SPECS: + assert isinstance(spec, ToolSpec) + assert spec.parameters["type"] == "object" + assert spec.parameters["additionalProperties"] is False + + +def test_tap_spec_requires_x_and_y() -> None: + assert TAP_SPEC.parameters["required"] == ["x", "y"] + assert set(TAP_SPEC.parameters["properties"]) == {"x", "y"} + + +def test_swipe_spec_requires_coordinates_and_makes_duration_optional() -> None: + assert SWIPE_SPEC.parameters["required"] == ["start_x", "start_y", "end_x", "end_y"] + assert set(SWIPE_SPEC.parameters["properties"]) == { + "start_x", + "start_y", + "end_x", + "end_y", + "duration_ms", + } + assert "duration_ms" not in SWIPE_SPEC.parameters["required"] + assert SWIPE_SPEC.parameters["properties"]["duration_ms"]["default"] == 500 + + +def test_input_text_spec_requires_text() -> None: + assert INPUT_TEXT_SPEC.parameters["required"] == ["text"] + assert set(INPUT_TEXT_SPEC.parameters["properties"]) == {"text"} + + +def test_launch_and_terminate_app_specs_require_app_id() -> None: + for spec in (LAUNCH_APP_SPEC, TERMINATE_APP_SPEC): + assert spec.parameters["required"] == ["app_id"] + assert set(spec.parameters["properties"]) == {"app_id"} + + +def test_finish_task_spec_requires_success_and_reason() -> None: + assert FINISH_TASK_SPEC.parameters["required"] == ["success", "reason"] + assert set(FINISH_TASK_SPEC.parameters["properties"]) == {"success", "reason"} + assert FINISH_TASK_SPEC.parameters["properties"]["success"]["type"] == "boolean" + + +def test_no_tool_spec_declares_device_id() -> None: + for spec in ALL_TOOL_SPECS: + assert "device_id" not in spec.parameters["properties"]