feat(agent-runtime): add LLM-driven AI Planner with dual-provider tool calling
Replaces the stub Planner's fixed describe_screen/[] behavior with a real decision-maker: AIPlanner uses native tool/function calling (Anthropic or OpenAI, pluggable via AI_PLANNER_PROVIDER) to select exactly one grounded action per turn, with an explicit finish_task(success, reason) tool for completion/failure instead of an ambiguous "no tool call" signal. Default disabled (AI_PLANNER_ENABLED=false) and additive; TaskRunner falls back to the existing stub Planner unchanged when disabled. Amends CONSTITUTION.md's Perception Boundary with one narrow exception: only the AI Planner may receive the current step's raw screenshot bytes alongside Scene, for vision-grounded coordinate grounding. Also fixes a latent gap in TaskRunner.run(): observe/plan exceptions are now caught per iteration and turned into a failed task with a failure_reason, instead of propagating uncaught. openspec change: ai-planner-runtime. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Claude Sonnet 5
@@ -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.
|
||||
|
||||
|
||||
@@ -0,0 +1,2 @@
|
||||
schema: spec-driven
|
||||
created: 2026-07-12
|
||||
@@ -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.
|
||||
@@ -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.
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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]
|
||||
@@ -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 []
|
||||
|
||||
@@ -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
|
||||
@@ -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."
|
||||
)
|
||||
+40
-18
@@ -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(),
|
||||
|
||||
@@ -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
|
||||
@@ -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]
|
||||
@@ -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"]
|
||||
@@ -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
|
||||
@@ -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)
|
||||
@@ -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
|
||||
@@ -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"
|
||||
@@ -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"]
|
||||
Reference in New Issue
Block a user