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"]