From a4c28865f2fcb6d20f732727a11c2a91690bcf0b Mon Sep 17 00:00:00 2001 From: somew1nd Date: Tue, 1 Sep 2026 18:07:45 -0700 Subject: [PATCH] refactor(core): consolidate action and plan definitions --- artemis/context.py | 1 - artemis/llm/__init__.py | 6 +- artemis/llm/router.py | 78 -- artemis/services/llm.py | 12 +- artemis/tools/mobile/exec_tools.py | 683 ------------------ artemis/utils/notes.py | 31 +- artemis/utils/task_tree.py | 6 +- .../action_surfaces/adb_server_manifest.json | 223 ++++++ tests/unit/mcp/test_action_specs.py | 11 +- tests/unit/mcp/test_adb_server_contract.py | 227 ++++++ ..._model_router.py => test_llm_grounding.py} | 59 +- tests/unit/test_tools.py | 70 -- tests/unit/tools/test_click_tool.py | 148 ---- tests/unit/tools/test_input_text_tool.py | 172 ----- tests/unit/tools/test_long_press_tool.py | 152 ---- tests/unit/tools/test_press_key_tool.py | 164 ----- tests/unit/tools/test_swipe_tool.py | 316 -------- 17 files changed, 481 insertions(+), 1878 deletions(-) delete mode 100644 artemis/tools/mobile/exec_tools.py create mode 100644 tests/fixtures/action_surfaces/adb_server_manifest.json create mode 100644 tests/unit/mcp/test_adb_server_contract.py rename tests/unit/{test_model_router.py => test_llm_grounding.py} (75%) delete mode 100644 tests/unit/test_tools.py delete mode 100644 tests/unit/tools/test_click_tool.py delete mode 100644 tests/unit/tools/test_input_text_tool.py delete mode 100644 tests/unit/tools/test_long_press_tool.py delete mode 100644 tests/unit/tools/test_press_key_tool.py delete mode 100644 tests/unit/tools/test_swipe_tool.py diff --git a/artemis/context.py b/artemis/context.py index a82d297..996b63d 100644 --- a/artemis/context.py +++ b/artemis/context.py @@ -179,7 +179,6 @@ class ArtemisContext(BaseModel): device: DeviceContext llm_config: LLMConfig | None = None - model_router: Any | None = None agent_config: Any = None adb_client: Any | None = None ui_adb_client: Any | None = None diff --git a/artemis/llm/__init__.py b/artemis/llm/__init__.py index 626456d..6cd97ee 100644 --- a/artemis/llm/__init__.py +++ b/artemis/llm/__init__.py @@ -12,18 +12,14 @@ # See the License for the specific language governing permissions and # limitations under the License. -"""Dynamic LLM Routing and Provider Integration layer.""" +"""LLM provider integration layer (endpoint descriptors and model factory).""" from artemis.llm.router import ( ModelEndpoint, ModelProvider, - ModelRouter, - default_router, ) __all__ = [ - "ModelRouter", "ModelEndpoint", "ModelProvider", - "default_router", ] diff --git a/artemis/llm/router.py b/artemis/llm/router.py index 471f820..5a48a6e 100644 --- a/artemis/llm/router.py +++ b/artemis/llm/router.py @@ -367,81 +367,3 @@ class ModelFactory: raise ValueError(f"Unsupported model provider: {provider}") -class ModelRouter: - """Dynamic router mapping agent roles to primary and fallback model endpoints.""" - - def __init__(self, default_model: str = "gemini-2.5-flash"): - self.default_endpoint = ModelEndpoint( - provider=ModelProvider.GOOGLE, - model_name=default_model, - ) - self._role_routes: dict[str, ModelEndpoint] = {} - self._fallback_routes: dict[str, list[ModelEndpoint]] = {} - self._setup_default_routes() - - def _setup_default_routes(self) -> None: - """Initializes recommended role mappings.""" - self._role_routes["planner"] = ModelEndpoint( - provider=ModelProvider.GOOGLE, - model_name="gemini-2.5-pro", - temperature=0.2, - ) - self._role_routes["operator"] = ModelEndpoint( - provider=ModelProvider.GOOGLE, - model_name="gemini-2.5-flash", - temperature=0.0, - ) - self._role_routes["validator"] = ModelEndpoint( - provider=ModelProvider.GOOGLE, - model_name="gemini-2.5-pro", - temperature=0.1, - ) - self._role_routes["summarizer"] = ModelEndpoint( - provider=ModelProvider.GOOGLE, - model_name="gemini-2.5-flash", - temperature=0.0, - ) - self._fallback_routes["planner"] = [ - ModelEndpoint(provider=ModelProvider.GOOGLE, model_name="gemini-2.5-flash"), - ] - - def set_role_route( - self, - role: str, - endpoint: ModelEndpoint | str, - fallbacks: list[ModelEndpoint] | None = None, - ) -> None: - """Configures or overrides a route for a specific agent role.""" - if isinstance(endpoint, str): - endpoint = ModelEndpoint(model_name=endpoint) - self._role_routes[role.lower()] = endpoint - if fallbacks is not None: - self._fallback_routes[role.lower()] = fallbacks - - def get_endpoint(self, role: str) -> ModelEndpoint: - """Retrieves the configured primary model endpoint for a role.""" - return self._role_routes.get(role.lower(), self.default_endpoint) - - def get_fallbacks(self, role: str) -> list[ModelEndpoint]: - """Retrieves fallback endpoints for a role.""" - return self._fallback_routes.get(role.lower(), []) - - def instantiate_model(self, endpoint: ModelEndpoint) -> BaseChatModel: - """Instantiates a concrete LangChain chat model via the unified ModelFactory.""" - return ModelFactory.create_model(endpoint) - - def create_chat_model(self, role: str) -> BaseChatModel: - """Creates a chat model with configured fallbacks for the specified role.""" - primary_endpoint = self.get_endpoint(role) - primary_model = self.instantiate_model(primary_endpoint) - - fallbacks = self.get_fallbacks(role) - if fallbacks: - fallback_models = [self.instantiate_model(ep) for ep in fallbacks] - return primary_model.with_fallbacks(fallback_models) - - return primary_model - - -# Global singleton router instance -default_router = ModelRouter() diff --git a/artemis/services/llm.py b/artemis/services/llm.py index 59664c3..a89d87f 100644 --- a/artemis/services/llm.py +++ b/artemis/services/llm.py @@ -1025,17 +1025,7 @@ def _resolve_endpoint( is_utils: bool = False, use_fallback: bool = False, ) -> ModelEndpoint: - """Cleanly resolves a ModelEndpoint from context router or llm_config.""" - # 1. Prefer dynamic ModelRouter if attached to context - if getattr(ctx, "model_router", None) is not None: - role_name = str(name).lower() - if use_fallback: - fallbacks = ctx.model_router.get_fallbacks(role_name) - if fallbacks: - return fallbacks[0] - return ctx.model_router.get_endpoint(role_name) - - # 2. Fall back to context llm_config + """Cleanly resolves a ModelEndpoint from context llm_config.""" if getattr(ctx, "llm_config", None) is None: try: ctx.llm_config = get_default_llm_config() diff --git a/artemis/tools/mobile/exec_tools.py b/artemis/tools/mobile/exec_tools.py deleted file mode 100644 index 1c06858..0000000 --- a/artemis/tools/mobile/exec_tools.py +++ /dev/null @@ -1,683 +0,0 @@ -# Copyright 2026 Google LLC -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -import asyncio -from typing import Any, Literal - -from langchain_core.messages import ToolMessage -from langchain_core.tools import BaseTool -from pydantic import BaseModel, Field - -from artemis.context import ArtemisContext -from artemis.controllers.unified_controller import UnifiedMobileController -from artemis.data_engine.trace import trace_langchain_tool -from artemis.drivers.base import BaseDeviceDriver -from artemis.graph.state import State -from artemis.tools.base import ArtemisTool, ToolCategory, ToolRegistry -from artemis.tools.tool_wrapper import ToolWrapper -from artemis.tools.types import CyFunctionDetector -from artemis.utils.coordinates import ( - compute_smart_swipe_coordinates, - parse_swipe_parameters, -) -from artemis.utils.logger import get_logger - -logger = get_logger(__name__) - - -class ClickArgs(BaseModel): - """Arguments schema for clicking on normalized coordinates.""" - - model_config = {"ignored_types": (CyFunctionDetector,)} - target: list[int] = Field( - ..., - description=("Click target normalized coordinates [x, y] in 0-1000 scale."), - ) - - -class LongPressArgs(BaseModel): - """Arguments schema for long pressing on normalized coordinates.""" - - model_config = {"ignored_types": (CyFunctionDetector,)} - target: list[int] = Field( - ..., - description=("Long press target normalized coordinates [x, y] in 0-1000 scale."), - ) - duration: int = Field(1000, description="Long press duration in milliseconds (default 1000).") - - -class InputTextArgs(BaseModel): - """Arguments schema for inputting text into normalized coordinates.""" - - model_config = {"ignored_types": (CyFunctionDetector,)} - text: str = Field( - ..., - description="The text content to input. Supports multi-line strings with '\\n'.", - ) - target: list[int] = Field( - ..., - description=("Input target field normalized coordinates [x, y] in 0-1000 scale."), - ) - clear_exist: bool = Field( - True, - description=( - "Whether to clear existing text before typing (True to replace all" - " text, False to append at the end of existing content)." - ), - ) - - -class SwipeArgs(BaseModel): - """Arguments schema for performing swipe gestures.""" - - model_config = {"ignored_types": (CyFunctionDetector,)} - direction: Literal["up", "down", "left", "right"] | None = Field( - None, - description=( - "Direction for scrolling and swiping: 'up' (drags bottom-to-top, scrolling down to reveal content below)," - " 'down' (drags top-to-bottom, scrolling up to reveal content above)," - " 'left' (drags right-to-left, scrolling right)," - " 'right' (drags left-to-right, scrolling left)." - ), - ) - start: list[int] | None = Field( - None, - description=( - "Start normalized coordinates [start_x, start_y] in 0-1000 scale for precise," - " local interactions (e.g. adjusting sliders, SeekBars, fine range selection, or drag-and-drop)." - ), - ) - end: list[int] | None = Field( - None, - description=( - "End normalized coordinates [end_x, end_y] in 0-1000 scale for precise," - " local interactions (e.g. adjusting sliders, SeekBars, fine range selection, or drag-and-drop)." - ), - ) - target: int | list[int] | str | None = Field( - None, - description=( - "Optional target element index (e.g. 2) or container bounds [left, top, right, bottom] to scope the directional swipe within." - ), - ) - action: Literal["up", "down", "left", "right"] | list[int] | None = Field( - None, - description=( - "Backward-compatible swipe gesture: smart direction string ('up', 'down', 'left', 'right')" - " OR precise custom coordinates [start_x, start_y, end_x, end_y] in 0-1000 scale." - ), - ) - duration: int = Field( - 800, - description=( - "Optional swipe/drag duration in milliseconds (default 800). For drag-and-drop," - " list reordering, or sliding/adjusting sliders (e.g., volume, brightness, SeekBars)," - " set duration >= 1000 (e.g. 1500). If omitted for directional swipe, duration is computed automatically." - ), - ) - - -class PressKeyArgs(BaseModel): - """Arguments schema for pressing system keys.""" - - model_config = {"ignored_types": (CyFunctionDetector,)} - key: Literal["ENTER", "BACK", "HOME", "APP_SWITCH"] = Field( - ..., - description=("Standard Android system button name (ENTER, BACK, HOME, APP_SWITCH)."), - ) - - -CLICK_DOCSTRING = "[ACTION] Click on the target normalized coordinates on the screen." - - -class ClickTool(ArtemisTool): - """Universal tool for clicking on target normalized coordinates on the screen.""" - - def __init__(self, category: ToolCategory = "action"): - super().__init__( - name="click", - description=CLICK_DOCSTRING, - args_schema=ClickArgs, - category=category, - ) - - # pylint: disable=too-many-arguments,too-many-positional-arguments,too-many-locals - async def execute( - self, - driver: BaseDeviceDriver | None = None, - ctx: ArtemisContext | None = None, - target: list[int] | None = None, - tool_call_id: str | None = None, - state: State | None = None, - **kwargs: Any, - ) -> Any: - tgt = target if target is not None else (kwargs.get("target") or kwargs.get("Target") or []) - tcid = tool_call_id if tool_call_id is not None else kwargs.get("tool_call_id") - st = state if state is not None else kwargs.get("state") - - width = 1080 - height = 2400 - if ctx and hasattr(ctx, "device") and ctx.device: - width = getattr(ctx.device, "device_width", 1080) - height = getattr(ctx.device, "device_height", 2400) - elif driver and hasattr(driver, "screen_size"): - width, height = driver.screen_size - - try: - if not tgt or len(tgt) < 2: - raise ValueError(f"Invalid target coordinates: {tgt}") - nx, ny = tgt[0], tgt[1] - x = int(max(0, min(width - 1, nx * width / 1000))) - y = int(max(0, min(height - 1, ny * height / 1000))) - - if driver is not None and hasattr(driver, "tap"): - success = await driver.tap(x, y) - outcome = ( - f"Clicked at [{nx}, {ny}] (normalized) successfully." - if success - else f"Click failed at {tgt}" - ) - elif ctx is not None: - controller = UnifiedMobileController(ctx) - result = await controller.tap_at(x, y) - success = result.error is None - outcome = ( - f"Clicked at [{nx}, {ny}] (normalized) successfully." - if success - else f"Failed to click: {result.error}" - ) - else: - success = False - outcome = "Error during click: No driver or context provided." - except Exception as e: # pylint: disable=broad-exception-caught - success = False - outcome = f"Error during click: {e}" - - if st is not None: - return ToolMessage( - tool_call_id=tcid or "", - content=outcome, - status="success" if success else "error", - ) - - return outcome - - -# Universal tool instance & aliases -click = ClickTool() -Click = ClickTool -ToolRegistry.register(click) - - -def get_click_tool(ctx: ArtemisContext) -> BaseTool: - """Exports click as a LangChain BaseTool.""" - return trace_langchain_tool(click.to_langchain_tool(ctx), ctx) - - -LONG_PRESS_DOCSTRING = "[ACTION] Long press on the target normalized coordinates on the screen." - - -class LongPressTool(ArtemisTool): - """Universal tool for long pressing on target normalized coordinates on the screen.""" - - def __init__(self, category: ToolCategory = "action"): - super().__init__( - name="long_press", - description=LONG_PRESS_DOCSTRING, - args_schema=LongPressArgs, - category=category, - ) - - # pylint: disable=too-many-arguments,too-many-positional-arguments,too-many-locals - async def execute( - self, - driver: BaseDeviceDriver | None = None, - ctx: ArtemisContext | None = None, - target: list[int] | None = None, - duration: int = 1000, - tool_call_id: str | None = None, - state: State | None = None, - **kwargs: Any, - ) -> Any: - tgt = target if target is not None else (kwargs.get("target") or kwargs.get("Target") or []) - dur = ( - duration - if duration is not None - else (kwargs.get("duration") or kwargs.get("Duration") or 1000) - ) - tcid = tool_call_id if tool_call_id is not None else kwargs.get("tool_call_id") - st = state if state is not None else kwargs.get("state") - - width = 1080 - height = 2400 - if ctx and hasattr(ctx, "device") and ctx.device: - width = getattr(ctx.device, "device_width", 1080) - height = getattr(ctx.device, "device_height", 2400) - elif driver and hasattr(driver, "screen_size"): - width, height = driver.screen_size - - try: - if not tgt or len(tgt) < 2: - raise ValueError(f"Invalid target coordinates: {tgt}") - nx, ny = tgt[0], tgt[1] - x = int(max(0, min(width - 1, nx * width / 1000))) - y = int(max(0, min(height - 1, ny * height / 1000))) - - if driver is not None and hasattr(driver, "long_press"): - success = await driver.long_press(x, y, duration_ms=dur) - outcome = ( - f"Long pressed at [{nx}, {ny}] (normalized) successfully." - if success - else f"Failed to long press: Long press failed at {tgt}" - ) - elif ctx is not None: - controller = UnifiedMobileController(ctx) - result = await controller.tap_at(x, y, long_press=True, long_press_duration=dur) - success = result.error is None - outcome = ( - f"Long pressed at [{nx}, {ny}] (normalized) successfully." - if success - else f"Failed to long press: {result.error}" - ) - else: - success = False - outcome = "Error during long press: No driver or context provided." - except Exception as e: # pylint: disable=broad-exception-caught - success = False - outcome = f"Error during long press: {e}" - - if st is not None: - return ToolMessage( - tool_call_id=tcid or "", - content=outcome, - status="success" if success else "error", - ) - - return outcome - - -# Universal tool instance & aliases -long_press = LongPressTool() -LongPress = LongPressTool -ToolRegistry.register(long_press) - - -def get_long_press_tool(ctx: ArtemisContext) -> BaseTool: - """Exports long_press as a LangChain BaseTool.""" - return trace_langchain_tool(long_press.to_langchain_tool(ctx), ctx) - - -INPUT_TEXT_DOCSTRING = "[ACTION] Type text into the target input field coordinates." - - -class InputTextTool(ArtemisTool): - """Universal tool for typing text into target input field coordinates.""" - - def __init__(self, category: ToolCategory = "action"): - super().__init__( - name="input_text", - description=INPUT_TEXT_DOCSTRING, - args_schema=InputTextArgs, - category=category, - ) - - # pylint: disable=too-many-arguments,too-many-positional-arguments,too-many-locals - async def execute( - self, - driver: BaseDeviceDriver | None = None, - ctx: ArtemisContext | None = None, - text: str | None = None, - target: list[int] | None = None, - clear_exist: bool = True, - tool_call_id: str | None = None, - state: State | None = None, - **kwargs: Any, - ) -> Any: - txt = text if text is not None else (kwargs.get("text") or kwargs.get("Text") or "") - tgt = target if target is not None else (kwargs.get("target") or kwargs.get("Target") or []) - clr = ( - clear_exist - if clear_exist is not None - else kwargs.get("clear_exist", kwargs.get("ClearExist", True)) - ) - tcid = tool_call_id if tool_call_id is not None else kwargs.get("tool_call_id") - st = state if state is not None else kwargs.get("state") - - width = 1080 - height = 2400 - if ctx and hasattr(ctx, "device") and ctx.device: - width = getattr(ctx.device, "device_width", 1080) - height = getattr(ctx.device, "device_height", 2400) - elif driver and hasattr(driver, "screen_size"): - width, height = driver.screen_size - - try: - if not tgt or len(tgt) < 2: - raise ValueError(f"Invalid target coordinates: {tgt}") - nx, ny = tgt[0], tgt[1] - x = int(max(0, min(width - 1, nx * width / 1000))) - y = int(max(0, min(height - 1, ny * height / 1000))) - - if driver is not None and hasattr(driver, "input_text"): - await driver.tap(x, y) - await asyncio.sleep(0.2) - success = await driver.input_text(txt, clear_existing=clr) - outcome = ( - f"Typed '{txt}' successfully." - if success - else f"Failed to input text into {tgt}" - ) - elif ctx is not None: - controller = UnifiedMobileController(ctx) - tap_result = await controller.tap_at(x, y) - if tap_result.error: - raise RuntimeError(f"Failed to focus input field: {tap_result.error}") - - if clr: - await controller.erase_text() - else: - await controller.press_key("123") # Move to end - - success = await controller.type_text(txt, clear_existing=False) - outcome = f"Typed '{txt}' successfully." if success else "Failed to type text." - else: - success = False - outcome = "Error during input text: No driver or context provided." - except Exception as e: # pylint: disable=broad-exception-caught - success = False - outcome = f"Error during input text: {e}" - - if st is not None: - return ToolMessage( - tool_call_id=tcid or "", - content=outcome, - status="success" if success else "error", - ) - - return outcome - - -# Universal tool instance & aliases -input_text = InputTextTool() -InputText = InputTextTool -ToolRegistry.register(input_text) - - -def get_input_text_tool(ctx: ArtemisContext) -> BaseTool: - """Exports input_text as a LangChain BaseTool.""" - return trace_langchain_tool(input_text.to_langchain_tool(ctx), ctx) - - -SWIPE_DOCSTRING = ( - "[ACTION] Perform a swipe, drag, or slider-adjustment gesture on the" - " screen.\n\n" - "• Directional Scrolling ('direction'): Recommended for general" - " browsing and standard page scrolling in most scenarios. Automatically" - " computes safe swipe vectors and adaptive duration, retains a ~40% visual overlap" - " anchor for zero-omission traversal, and prevents inertial flings. Supports" - " scoping to a sub-container via 'target'. If it fails on certain custom layouts," - " fall back to specifying exact coordinates ('start' and 'end') directly.\n" - "• Precise Coordinate Gestures ('start', 'end'): Best for local," - " fine-grained interactions such as adjusting sliders/SeekBars (e.g.," - " volume, brightness, progress bars), drag-and-drop / list reordering," - " or as a reliable fallback when directional scrolling fails on specific" - " containers. Always drag slightly PAST the target position to overcome" - " touch slop and reliably trigger the update. When setting a slider to" - " Maximum (100%) or Minimum (0%), swipe fully to the extreme boundary." -) - - -class SwipeTool(ArtemisTool): - """Universal tool for performing swipe gestures on the screen.""" - - def __init__(self, category: ToolCategory = "action"): - super().__init__( - name="swipe", - description=SWIPE_DOCSTRING, - args_schema=SwipeArgs, - category=category, - ) - - # pylint: disable=too-many-arguments,too-many-positional-arguments,too-many-branches,too-many-locals,too-many-statements - async def execute( - self, - driver: BaseDeviceDriver | None = None, - ctx: ArtemisContext | None = None, - action: Literal["up", "down", "left", "right"] | list[int] | str | None = None, - direction: Literal["up", "down", "left", "right"] | str | None = None, - start: list[int] | None = None, - end: list[int] | None = None, - duration: int = 800, - tool_call_id: str | None = None, - state: State | None = None, - **kwargs: Any, - ) -> Any: - params = dict(kwargs) - if action is not None: - params["action"] = action - if direction is not None: - params["direction"] = direction - if start is not None: - params["start"] = start - if end is not None: - params["end"] = end - if duration is not None: - params["duration"] = duration - - kind, target, dur = parse_swipe_parameters(params, default_duration=duration or 800) - - tcid = tool_call_id if tool_call_id is not None else kwargs.get("tool_call_id") - st = state if state is not None else kwargs.get("state") - - width = 1080 - height = 2400 - if ctx and hasattr(ctx, "device") and ctx.device: - width = getattr(ctx.device, "device_width", 1080) - height = getattr(ctx.device, "device_height", 2400) - elif driver and hasattr(driver, "screen_size"): - width, height = driver.screen_size - - try: - if kind is None or target is None: - raise ValueError( - f"Invalid swipe action parameter: {action or direction or (start, end) or kwargs}" - ) - - if ( - driver is not None - and hasattr(driver, "swipe_direction") - and hasattr(driver, "swipe") - ): - if kind == "direction" and isinstance(target, str): - success = await driver.swipe_direction(target, duration_ms=dur) - outcome = ( - "Swipe completed successfully." if success else f"Swipe {target} failed" - ) - elif kind == "coords" and isinstance(target, list) and len(target) == 4: - start_x = int(max(0, min(width - 1, target[0] * width / 1000))) - start_y = int(max(0, min(height - 1, target[1] * height / 1000))) - end_x = int(max(0, min(width - 1, target[2] * width / 1000))) - end_y = int(max(0, min(height - 1, target[3] * height / 1000))) - success = await driver.swipe(start_x, start_y, end_x, end_y, duration_ms=dur) - outcome = ( - "Swipe completed successfully." - if success - else f"Swipe coordinates {target} failed" - ) - else: - raise ValueError(f"Invalid swipe action parameter: {target}") - elif ctx is not None: - controller = UnifiedMobileController(ctx) - - if kind == "direction" and isinstance(target, str): - indexed_elems = getattr(state, "indexed_elements", None) if state else None - ui_hier = getattr(state, "latest_ui_hierarchy", None) if state else None - start_x, start_y, end_x, end_y, smart_dur = compute_smart_swipe_coordinates( - direction=target, - target=params.get("target"), - indexed_elements=indexed_elems, - ui_hierarchy=ui_hier, - width=width, - height=height, - duration=dur, - ) - dur = smart_dur - elif kind == "coords" and isinstance(target, list) and len(target) == 4: - nx1, ny1, nx2, ny2 = target - start_x = int(max(0, min(width - 1, nx1 * width / 1000))) - start_y = int(max(0, min(height - 1, ny1 * height / 1000))) - end_x = int(max(0, min(width - 1, nx2 * width / 1000))) - end_y = int(max(0, min(height - 1, ny2 * height / 1000))) - else: - raise ValueError(f"Invalid swipe action parameter: {target}") - - error = await controller.swipe_coords(start_x, start_y, end_x, end_y, dur) - success = error is None - outcome = "Swipe completed successfully." if success else f"Swipe failed: {error}" - else: - success = False - outcome = "Error during swipe: No driver or context provided." - except Exception as e: # pylint: disable=broad-exception-caught - success = False - outcome = f"Error during swipe: {e}" - - if st is not None: - return ToolMessage( - tool_call_id=tcid or "", - content=outcome, - status="success" if success else "error", - ) - - return outcome - - -# Universal tool instance & aliases -swipe = SwipeTool() -Swipe = SwipeTool -ToolRegistry.register(swipe) - - -def get_swipe_tool(ctx: ArtemisContext) -> BaseTool: - """Exports swipe as a LangChain BaseTool.""" - return trace_langchain_tool(swipe.to_langchain_tool(ctx), ctx) - - -PRESS_KEY_DOCSTRING = "[ACTION] Press a physical or virtual system button." - - -class PressKeyTool(ArtemisTool): - """Universal tool for pressing a physical or virtual system button.""" - - def __init__(self, category: ToolCategory = "action"): - super().__init__( - name="press_key", - description=PRESS_KEY_DOCSTRING, - args_schema=PressKeyArgs, - category=category, - ) - - # pylint: disable=too-many-arguments,too-many-positional-arguments - async def execute( - self, - driver: BaseDeviceDriver | None = None, - ctx: ArtemisContext | None = None, - key: Literal["ENTER", "BACK", "HOME", "APP_SWITCH"] | str | None = None, - tool_call_id: str | None = None, - state: State | None = None, - **kwargs: Any, - ) -> Any: - k = key if key is not None else (kwargs.get("key") or kwargs.get("Key") or "") - tcid = tool_call_id if tool_call_id is not None else kwargs.get("tool_call_id") - st = state if state is not None else kwargs.get("state") - - try: - if not k: - raise ValueError("Key parameter is required.") - - if driver is not None and hasattr(driver, "press_key"): - success = await driver.press_key(k) - outcome = ( - f"Pressed key '{k}' successfully." if success else f"Failed to press key '{k}'." - ) - elif ctx is not None: - controller = UnifiedMobileController(ctx) - success = False - if k == "ENTER": - success = await controller.press_enter() - elif k == "BACK": - success = await controller.go_back() - elif k == "HOME": - success = await controller.go_home() - elif k == "APP_SWITCH": - success = await controller.press_key("KEYCODE_APP_SWITCH") - else: - success = await controller.press_key(k) - - outcome = ( - f"Pressed key '{k}' successfully." if success else f"Failed to press key '{k}'." - ) - else: - success = False - outcome = "Error during press key: No driver or context provided." - except Exception as e: # pylint: disable=broad-exception-caught - success = False - outcome = f"Error during press key: {e}" - - if st is not None: - return ToolMessage( - tool_call_id=tcid or "", - content=outcome, - status="success" if success else "error", - ) - - return outcome - - -# Universal tool instance & aliases -press_key = PressKeyTool() -PressKey = PressKeyTool -ToolRegistry.register(press_key) - - -def get_press_key_tool(ctx: ArtemisContext) -> BaseTool: - """Exports press_key as a LangChain BaseTool.""" - return trace_langchain_tool(press_key.to_langchain_tool(ctx), ctx) - - -# Wrappers to expose -click_wrapper = ToolWrapper( - tool_fn_getter=get_click_tool, - on_success_fn=lambda *args, **kwargs: "Success", - on_failure_fn=lambda *args, **kwargs: "Failure", -) -long_press_wrapper = ToolWrapper( - tool_fn_getter=get_long_press_tool, - on_success_fn=lambda *args, **kwargs: "Success", - on_failure_fn=lambda *args, **kwargs: "Failure", -) -input_text_wrapper = ToolWrapper( - tool_fn_getter=get_input_text_tool, - on_success_fn=lambda *args, **kwargs: "Success", - on_failure_fn=lambda *args, **kwargs: "Failure", -) -swipe_wrapper = ToolWrapper( - tool_fn_getter=get_swipe_tool, - on_success_fn=lambda *args, **kwargs: "Success", - on_failure_fn=lambda *args, **kwargs: "Failure", -) -press_key_wrapper = ToolWrapper( - tool_fn_getter=get_press_key_tool, - on_success_fn=lambda *args, **kwargs: "Success", - on_failure_fn=lambda *args, **kwargs: "Failure", -) diff --git a/artemis/utils/notes.py b/artemis/utils/notes.py index 71b0e5f..69aa8e9 100644 --- a/artemis/utils/notes.py +++ b/artemis/utils/notes.py @@ -13,7 +13,6 @@ # limitations under the License. import difflib -import hashlib import json import os from pathlib import Path @@ -21,6 +20,7 @@ import re from artemis.data_engine.engine import _CURRENT_DATA_ENGINE from artemis.utils.logger import get_logger +from artemis.utils.plan_grammar import parse_plan logger = get_logger(__name__) @@ -411,23 +411,24 @@ def record_subgoal_hash_chain( if not content_before or not content_after: return - # Helper to get the active subgoal hash from task_plan string + # Helper to get the active subgoal hash from task_plan string. + # Parsing goes through the plan-grammar single source; the hash is + # ``PlanItem.key`` (= ``plan_grammar.subgoal_hash``, md5 of the stripped + # subgoal text) -- byte-identical to the historical inline md5, so anchors + # already persisted in subgoal_hash_chain.json stay valid. def get_active_hash(task_plan: str) -> str | None: try: - lines = task_plan.splitlines() - for line in reversed(lines): - if line.strip().startswith("- [/]"): - text = line.strip()[5:].strip() - return hashlib.md5(text.encode("utf-8")).hexdigest() - - # Fallback to first pending [ ] if no [/] - for line in lines: - if line.strip().startswith("- [ ]"): - text = line.strip()[5:].strip() - return hashlib.md5(text.encode("utf-8")).hexdigest() + snapshot = parse_plan(task_plan) + # Last active [/] item wins (its own key, not the parent's) -- + # matches the historical reversed-scan behavior of this helper. + item = snapshot.last_active() + if item is None: + # Fallback to the first pending [ ] item if no [/]. + item = next((i for i in snapshot.items if i.is_pending), None) + return item.key if item is not None else None except Exception as exc: - # Pure string/hash operations; a failure here is unexpected and - # silently disables subgoal rename tracking. + # A failure here is unexpected and silently disables subgoal + # rename tracking. logger.warning("Failed to compute active subgoal hash: %s", exc) return None diff --git a/artemis/utils/task_tree.py b/artemis/utils/task_tree.py index 7a8bddc..1326f6a 100644 --- a/artemis/utils/task_tree.py +++ b/artemis/utils/task_tree.py @@ -820,10 +820,8 @@ def build_plan_and_history( # 1. Plan Checklist Section output_parts = [] - # Check if task_plan has actual subgoals (e.g. starts with '- [') - has_subgoals = task_plan and any( - line.strip().startswith("- [") for line in task_plan.split("\n") - ) + # Check if task_plan has actual subgoals (single-source checkbox grammar). + has_subgoals = bool(task_plan) and bool(parse_plan(task_plan).items) if has_subgoals: output_parts.append("--- Task Plan ---") diff --git a/tests/fixtures/action_surfaces/adb_server_manifest.json b/tests/fixtures/action_surfaces/adb_server_manifest.json new file mode 100644 index 0000000..119d75e --- /dev/null +++ b/tests/fixtures/action_surfaces/adb_server_manifest.json @@ -0,0 +1,223 @@ +{ + "tap": { + "description": "Taps on the screen at coordinates.\n\n'coordinates' is a list [x, y].\n'times' is the number of consecutive clicks (default 1).\n'delay_ms' is the delay in milliseconds between consecutive clicks (default\n100).\n", + "inputSchema": { + "properties": { + "coordinates": { + "items": { + "type": "integer" + }, + "title": "Coordinates", + "type": "array" + }, + "times": { + "default": 1, + "title": "Times", + "type": "integer" + }, + "delay_ms": { + "default": 100, + "title": "Delay Ms", + "type": "integer" + } + }, + "required": [ + "coordinates" + ], + "title": "tapArguments", + "type": "object" + } + }, + "long_press_on": { + "description": "Long presses on the screen at coordinates.\n\n'coordinates' is a list [x, y].\n'duration' is in milliseconds (default 1000).\n", + "inputSchema": { + "properties": { + "coordinates": { + "items": { + "type": "integer" + }, + "title": "Coordinates", + "type": "array" + }, + "duration": { + "default": 1000, + "title": "Duration", + "type": "integer" + } + }, + "required": [ + "coordinates" + ], + "title": "long_press_onArguments", + "type": "object" + } + }, + "swipe": { + "description": "Swipes from start coordinates to end coordinates.\n\n'coordinates' is a list [start_x, start_y, end_x, end_y].\n'duration' is in milliseconds (default 400).\n\nSet duration >= 1000 to drag-and-drop. Drag slightly past the target\nposition to trigger reordering.\n", + "inputSchema": { + "properties": { + "coordinates": { + "items": { + "type": "integer" + }, + "title": "Coordinates", + "type": "array" + }, + "duration": { + "default": 400, + "title": "Duration", + "type": "integer" + } + }, + "required": [ + "coordinates" + ], + "title": "swipeArguments", + "type": "object" + } + }, + "back": { + "description": "Simulates pressing the system back button.", + "inputSchema": { + "properties": {}, + "title": "backArguments", + "type": "object" + } + }, + "launch_app": { + "description": "Launches an application by its Android package name with retries and smart polling.", + "inputSchema": { + "properties": { + "package_name": { + "title": "Package Name", + "type": "string" + } + }, + "required": [ + "package_name" + ], + "title": "launch_appArguments", + "type": "object" + } + }, + "stop_app": { + "description": "Force stops an application by its Android package name.", + "inputSchema": { + "properties": { + "package_name": { + "title": "Package Name", + "type": "string" + } + }, + "required": [ + "package_name" + ], + "title": "stop_appArguments", + "type": "object" + } + }, + "open_link": { + "description": "Opens a URL or deep link on the device.", + "inputSchema": { + "properties": { + "url": { + "title": "Url", + "type": "string" + } + }, + "required": [ + "url" + ], + "title": "open_linkArguments", + "type": "object" + } + }, + "focus_and_input_text": { + "description": "Focuses on a UI element at coordinates and inputs text.\n\n'coordinates' is a list [x, y] to tap first to gain focus.\n'clear_before_input' if True, clears all existing text before typing. If False, appends text at the end of existing content.\n'text' supports multi-line content with '\\n'.\n", + "inputSchema": { + "properties": { + "coordinates": { + "items": { + "type": "integer" + }, + "title": "Coordinates", + "type": "array" + }, + "text": { + "title": "Text", + "type": "string" + }, + "clear_before_input": { + "default": false, + "title": "Clear Before Input", + "type": "boolean" + } + }, + "required": [ + "coordinates", + "text" + ], + "title": "focus_and_input_textArguments", + "type": "object" + } + }, + "focus_and_clear_text": { + "description": "Focuses on a UI element at coordinates and clears its text content.\n\n'coordinates' is a list [x, y] to tap first to gain focus.\n", + "inputSchema": { + "properties": { + "coordinates": { + "items": { + "type": "integer" + }, + "title": "Coordinates", + "type": "array" + } + }, + "required": [ + "coordinates" + ], + "title": "focus_and_clear_textArguments", + "type": "object" + } + }, + "erase_one_char": { + "description": "Erases a single character (simulates Backspace).", + "inputSchema": { + "properties": {}, + "title": "erase_one_charArguments", + "type": "object" + } + }, + "press_key": { + "description": "Presses a specific Android key event (e.g., KEYCODE_ENTER, KEYCODE_HOME).", + "inputSchema": { + "properties": { + "keycode": { + "title": "Keycode", + "type": "string" + } + }, + "required": [ + "keycode" + ], + "title": "press_keyArguments", + "type": "object" + } + }, + "take_screenshot": { + "description": "Takes a screenshot of the device screen.\n\nReturns the screenshot as a base64 encoded JPEG string.\n", + "inputSchema": { + "properties": {}, + "title": "take_screenshotArguments", + "type": "object" + } + }, + "get_ui_hierarchy": { + "description": "Retrieves the current UI elements hierarchy from the device.", + "inputSchema": { + "properties": {}, + "title": "get_ui_hierarchyArguments", + "type": "object" + } + } +} diff --git a/tests/unit/mcp/test_action_specs.py b/tests/unit/mcp/test_action_specs.py index 9fead48..5ad23c4 100644 --- a/tests/unit/mcp/test_action_specs.py +++ b/tests/unit/mcp/test_action_specs.py @@ -210,15 +210,14 @@ def test_operator_verbs_lower_onto_manifest_actions(): def test_registry_wait_for_delay_seconds_drift_stays_dead(): - """Guards the removal of the duplicate device-action registration. + """Guards against re-growing a shadow device-action registration. ``artemis/tools/actions/device_actions.py`` used to shadow-register a second - ``wait_for_delay`` taking *seconds* while every prompt teaches milliseconds. The - registry (a direct-execution SDK surface, artemis/tools/mobile/exec_tools.py) - must never again carry a ``wait_for_delay`` whose unit disagrees with the - manifest's ``time_in_ms``. + ``wait_for_delay`` taking *seconds* while every prompt teaches milliseconds + (the last such surface, ``artemis/tools/mobile/exec_tools.py``, is deleted). + The ToolRegistry must never again carry a ``wait_for_delay`` whose unit + disagrees with the manifest's ``time_in_ms``. """ - import artemis.tools.mobile.exec_tools # noqa: F401 - registration side effect from artemis.tools.base import ToolRegistry tool = ToolRegistry.get("wait_for_delay") diff --git a/tests/unit/mcp/test_adb_server_contract.py b/tests/unit/mcp/test_adb_server_contract.py new file mode 100644 index 0000000..286748b --- /dev/null +++ b/tests/unit/mcp/test_adb_server_contract.py @@ -0,0 +1,227 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Consistency contract between the stdio ADB MCP server and the action manifest. + +``artemis/mcp/adb_server.py`` deliberately keeps its historical *pixel-space* +contract for external MCP clients (absolute pixel coordinates, combined +``coordinates`` list parameters, ``KEYCODE_*`` vocabulary), so its FastMCP tools +are hand-written rather than generated from ``artemis.mcp.action_specs``. This +suite is the compensating contract: + +* **Fixture pin** -- the full generated schema of every adb_server tool is + frozen; any signature or docstring change must regenerate the fixture in the + same commit (external clients see this schema directly). +* **Projection check** -- every device-action tool is mapped onto its canonical + ``ActionSpec`` wire dialect. Parameters that mirror a wire parameter must + agree on JSON type, requiredness, and default. Deliberate legacy divergences + are declared in ``DECLARED_DIVERGENCES`` and are *ratcheted*: if the two + surfaces converge, the stale declaration fails too, so the table always + matches reality. + +Together the two guards make silent drift between the second declaration +surface and the canonical manifest impossible in either direction. +""" + +import json +from pathlib import Path + +import pytest + +from artemis.mcp.action_specs import ACTION_SPECS, ParamSpec + +FIXTURES = Path(__file__).resolve().parents[2] / "fixtures" / "action_surfaces" + +FIXTURE_HINT = ( + "The generated schema differs from the pinned fixture. If this change is" + " intentional, regenerate the fixture in the same commit and explain the schema" + " change in the commit message; external MCP clients see this schema directly." +) + + +# --- The declared projection ---------------------------------------------------------- + +#: adb_server tool name -> (canonical action name, {adb param -> wire param}). +#: Params absent from the mapping are legacy packaging with no one-to-one wire +#: counterpart (the combined ``coordinates`` lists); their shape is asserted +#: separately below and pinned by the fixture. +ADB_TOOL_PROJECTION: dict[str, tuple[str, dict[str, str]]] = { + "tap": ("click", {"times": "times", "delay_ms": "delay_ms"}), + "long_press_on": ("long_press", {"duration": "duration_ms"}), + "swipe": ("swipe", {"duration": "duration_ms"}), + "focus_and_input_text": ("input_text", {"text": "text", "clear_before_input": "clear_exist"}), + "press_key": ("press_key", {"keycode": "key"}), + "open_link": ("open_link", {"url": "url"}), + "erase_one_char": ("erase_one_char", {}), + "focus_and_clear_text": ("focus_and_clear_text", {}), +} + +#: Combined-coordinates legacy packaging: adb tool -> (marker in docstring, arity). +#: ``coordinates`` packs the wire dialect's coordinate pair(s) into one list. +COORDINATE_PACKING: dict[str, tuple[str, int]] = { + "tap": ("[x, y]", 2), + "long_press_on": ("[x, y]", 2), + "swipe": ("[start_x, start_y, end_x, end_y]", 4), + "focus_and_input_text": ("[x, y]", 2), + "focus_and_clear_text": ("[x, y]", 2), +} + +#: Deliberate legacy divergences, (adb tool, adb param) -> reason. Each entry is +#: ratcheted: the test fails if the surfaces actually agree, so this table can +#: never go stale. +DECLARED_DIVERGENCES: dict[tuple[str, str], str] = { + ("swipe", "duration"): ( + "legacy stdio default is 400ms; the canonical wire duration_ms defaults" + " to 800ms (adaptive-duration smart swipes only exist in agent dialects)" + ), + ("focus_and_input_text", "clear_before_input"): ( + "legacy stdio appends by default (False); the canonical wire clear_exist" + " defaults to True (replace)" + ), +} + +_JSON_TYPE: dict[type, str] = {int: "integer", str: "string", bool: "boolean"} + + +def _wire_param(action: str, name: str) -> ParamSpec: + spec = ACTION_SPECS[action] + assert spec.wire is not None, f"'{action}' has no wire dialect." + for p in spec.wire.params: + if p.name == name: + return p + raise AssertionError(f"wire dialect of '{action}' has no param '{name}'") + + +async def _adb_server_tools() -> dict: + from artemis.mcp import adb_server + + tools = await adb_server.mcp.list_tools() + return {t.name: t for t in tools} + + +def _prop(tool, name: str) -> dict: + props = tool.inputSchema.get("properties", {}) + assert name in props, f"tool '{tool.name}' lost parameter '{name}'" + return props[name] + + +# --- Fixture pin ----------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_adb_server_manifest_matches_fixture(): + expected = json.loads( + (FIXTURES / "adb_server_manifest.json").read_text(encoding="utf-8") + ) + tools = await _adb_server_tools() + generated = { + name: {"description": t.description, "inputSchema": t.inputSchema} + for name, t in tools.items() + } + assert set(generated) == set(expected), FIXTURE_HINT + for name in expected: + assert generated[name] == expected[name], f"adb_server tool '{name}': {FIXTURE_HINT}" + + +# --- Projection consistency ------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_projection_covers_all_wire_backed_adb_tools(): + """Every adb_server tool that mirrors a canonical action is in the projection.""" + tools = await _adb_server_tools() + missing = set(ADB_TOOL_PROJECTION) - set(tools) + assert not missing, f"projection references adb_server tools that no longer exist: {missing}" + for adb_name, (canonical, _) in ADB_TOOL_PROJECTION.items(): + assert canonical in ACTION_SPECS, ( + f"adb_server tool '{adb_name}' maps to '{canonical}', which the canonical" + " manifest does not know." + ) + assert ACTION_SPECS[canonical].wire is not None, ( + f"adb_server tool '{adb_name}' maps to '{canonical}', which has no wire dialect." + ) + + +@pytest.mark.asyncio +async def test_mirrored_params_agree_with_wire_dialect(): + """Type/requiredness/default of every mirrored parameter matches the wire spec, + except where a divergence is declared -- and declared divergences must be real.""" + tools = await _adb_server_tools() + for adb_name, (canonical, param_map) in ADB_TOOL_PROJECTION.items(): + tool = tools[adb_name] + required = set(tool.inputSchema.get("required", [])) + for adb_param, wire_name in param_map.items(): + wire = _wire_param(canonical, wire_name) + prop = _prop(tool, adb_param) + expected_type = _JSON_TYPE.get(wire.annotation) + if expected_type is not None: + assert prop.get("type") == expected_type, ( + f"'{adb_name}.{adb_param}': schema type {prop.get('type')!r} !=" + f" canonical wire '{canonical}.{wire_name}' type {expected_type!r}" + ) + adb_required = adb_param in required + adb_default = prop.get("default") + diverged_key = (adb_name, adb_param) + agrees = adb_required == wire.required and ( + wire.required or adb_default == wire.default + ) + if diverged_key in DECLARED_DIVERGENCES: + assert not agrees, ( + f"'{adb_name}.{adb_param}' is declared divergent from" + f" '{canonical}.{wire_name}' but the surfaces now agree -- remove" + f" the stale entry from DECLARED_DIVERGENCES:" + f" {DECLARED_DIVERGENCES[diverged_key]}" + ) + else: + assert agrees, ( + f"'{adb_name}.{adb_param}' drifted from canonical wire" + f" '{canonical}.{wire_name}': required {adb_required} vs" + f" {wire.required}, default {adb_default!r} vs {wire.default!r}." + " Either re-align, or declare the divergence in" + " DECLARED_DIVERGENCES with a reason." + ) + + +@pytest.mark.asyncio +async def test_coordinate_packing_shape_is_documented(): + """The combined `coordinates` legacy parameter keeps its documented arity.""" + tools = await _adb_server_tools() + for adb_name, (marker, _arity) in COORDINATE_PACKING.items(): + tool = tools[adb_name] + prop = _prop(tool, "coordinates") + assert prop.get("type") == "array", f"'{adb_name}.coordinates' must be an array" + assert prop.get("items", {}).get("type") == "integer", ( + f"'{adb_name}.coordinates' must be a list of integers" + ) + assert "coordinates" in set(tool.inputSchema.get("required", [])), ( + f"'{adb_name}.coordinates' must stay required" + ) + assert marker in (tool.description or ""), ( + f"'{adb_name}' docstring no longer documents the coordinate shape" + f" '{marker}'; external clients rely on this wording." + ) + + +@pytest.mark.asyncio +async def test_press_key_vocabulary_divergence_is_documented(): + """adb_server's press_key takes raw KEYCODE_* strings; the canonical wire takes + bare lowercase words. The canonical spec documents that divergence in prose.""" + tools = await _adb_server_tools() + desc = tools["press_key"].description or "" + assert "KEYCODE_" in desc, ( + "adb_server press_key docstring must keep teaching KEYCODE_* vocabulary" + ) + assert "key" in ACTION_SPECS["press_key"].differences, ( + "the canonical manifest no longer documents the press_key vocabulary split" + ) diff --git a/tests/unit/test_model_router.py b/tests/unit/test_llm_grounding.py similarity index 75% rename from tests/unit/test_model_router.py rename to tests/unit/test_llm_grounding.py index b1097b8..c99b01d 100644 --- a/tests/unit/test_model_router.py +++ b/tests/unit/test_llm_grounding.py @@ -12,80 +12,33 @@ # See the License for the specific language governing permissions and # limitations under the License. -"""Unit tests for Dynamic Model Router.""" +"""Unit tests for ModelFactory instantiation and grounding tool wiring.""" -from artemis.llm.router import ModelEndpoint, ModelProvider, ModelRouter +from artemis.llm.router import ModelEndpoint, ModelFactory, ModelProvider -def test_model_router_default_routes(): - """Verify default role routing returns expected models.""" - router = ModelRouter() - - # Planner route - planner_ep = router.get_endpoint("planner") - assert planner_ep.provider == ModelProvider.GEMINI - assert "pro" in planner_ep.model_name - - # Operator route - operator_ep = router.get_endpoint("operator") - assert operator_ep.provider == ModelProvider.GEMINI - assert "flash" in operator_ep.model_name - - -def test_model_router_custom_route_override(): - """Verify dynamic role route override.""" - router = ModelRouter() - custom_ep = ModelEndpoint( - provider=ModelProvider.OPENAI, - model_name="gpt-4o", - api_base="https://custom.api.endpoint/v1", - ) - router.set_role_route("operator", custom_ep) - - ep = router.get_endpoint("operator") - assert ep.provider == ModelProvider.OPENAI - assert ep.model_name == "gpt-4o" - assert ep.api_base == "https://custom.api.endpoint/v1" - - -def test_model_router_fallbacks(): - """Verify fallback endpoint configuration.""" - router = ModelRouter() - fallbacks = [ - ModelEndpoint(provider=ModelProvider.GEMINI, model_name="gemini-2.5-flash"), - ModelEndpoint(provider=ModelProvider.OPENAI, model_name="gpt-4o-mini"), - ] - router.set_role_route("custom_worker", "gemini-2.5-pro", fallbacks=fallbacks) - fb = router.get_fallbacks("custom_worker") - assert len(fb) == 2 - assert fb[0].model_name == "gemini-2.5-flash" - assert fb[1].model_name == "gpt-4o-mini" - - -def test_model_router_anthropic_instantiation(): +def test_model_factory_anthropic_instantiation(): """Verify Anthropic Claude model instantiation with thinking budget.""" - router = ModelRouter() anth_ep = ModelEndpoint( provider=ModelProvider.ANTHROPIC, model_name="claude-3-7-sonnet-20250219", api_key="sk-ant-test-key", reasoning_effort="high", ) - model = router.instantiate_model(anth_ep) + model = ModelFactory.create_model(anth_ep) assert model.model == "claude-3-7-sonnet-20250219" assert getattr(model, "thinking", None) == {"type": "enabled", "budget_tokens": 32768} -def test_model_router_openai_instantiation(): +def test_model_factory_openai_instantiation(): """Verify OpenAI model instantiation with reasoning effort.""" - router = ModelRouter() oai_ep = ModelEndpoint( provider=ModelProvider.OPENAI, model_name="o3-mini", api_key="sk-test-key", reasoning_effort="medium", ) - model = router.instantiate_model(oai_ep) + model = ModelFactory.create_model(oai_ep) assert model.model_name == "o3-mini" assert getattr(model, "reasoning_effort", None) == "medium" diff --git a/tests/unit/test_tools.py b/tests/unit/test_tools.py deleted file mode 100644 index d6c08b5..0000000 --- a/tests/unit/test_tools.py +++ /dev/null @@ -1,70 +0,0 @@ -# Copyright 2026 Google LLC -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -"""Unit tests for Universal Tool Protocol and Actions.""" - -import pytest -from artemis.drivers.mock.mock_driver import MockDeviceDriver -from artemis.tools.base import ToolRegistry -import artemis.tools.mobile.exec_tools # noqa: F401 - Register action tools - - -@pytest.mark.asyncio -async def test_tool_registry_registration(): - """Verify standard device actions are registered in ToolRegistry.""" - click_tool = ToolRegistry.get("click") - assert click_tool is not None - assert click_tool.name == "click" - assert click_tool.category == "action" - - swipe_tool = ToolRegistry.get("swipe") - assert swipe_tool is not None - assert swipe_tool.name == "swipe" - - -@pytest.mark.asyncio -async def test_tool_execution_with_driver(): - """Verify tool execution correctly dispatches to MockDeviceDriver.""" - driver = MockDeviceDriver(width=1000, height=2000) - - # Click normalized [500, 500] -> pixel (500, 1000) - result = await ToolRegistry.execute("click", {"target": [500, 500]}, driver=driver, ctx=None) - assert "Click" in result or "Clicked" in result - assert driver.action_history[-1]["action"] == "tap" - assert driver.action_history[-1]["x"] == 500 - assert driver.action_history[-1]["y"] == 1000 - - # Swipe cardinal "up" - result = await ToolRegistry.execute("swipe", {"action": "up"}, driver=driver, ctx=None) - assert "Swipe" in result or "Swiped" in result - assert driver.action_history[-1]["action"] == "swipe_direction" - - -def test_tool_to_genai_declaration(): - """Verify tools export to Google GenAI FunctionDeclaration properly.""" - declarations = ToolRegistry.get_genai_declarations(["click", "swipe", "input_text"]) - assert len(declarations) == 3 - - decl_names = [d.name for d in declarations] - assert "click" in decl_names - assert "swipe" in decl_names - assert "input_text" in decl_names - - -def test_tool_to_langchain_tool(): - """Verify tools export to LangChain BaseTool.""" - lc_tools = ToolRegistry.get_langchain_tools(ctx=None, names=["click", "press_key"]) - assert len(lc_tools) == 2 - assert lc_tools[0].name == "click" - assert lc_tools[1].name == "press_key" diff --git a/tests/unit/tools/test_click_tool.py b/tests/unit/tools/test_click_tool.py deleted file mode 100644 index 396fa69..0000000 --- a/tests/unit/tools/test_click_tool.py +++ /dev/null @@ -1,148 +0,0 @@ -# Copyright 2026 Google LLC -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -from unittest.mock import AsyncMock, MagicMock, patch - -from artemis.context import ArtemisContext -from artemis.drivers.base import BaseDeviceDriver -from artemis.graph.state import State -from artemis.tools.base import ArtemisTool, ToolRegistry -from artemis.tools.mobile.exec_tools import ( - Click, - ClickArgs, - ClickTool, - click, - click_wrapper, - get_click_tool, -) -from langchain_core.messages import ToolMessage -import pytest - - -@pytest.fixture -def mock_ctx(): - ctx = MagicMock(spec=ArtemisContext) - ctx.device = MagicMock() - ctx.device.device_width = 1080 - ctx.device.device_height = 2400 - return ctx - - -@pytest.fixture -def mock_driver(): - driver = MagicMock(spec=BaseDeviceDriver) - driver.screen_size = (1080, 2400) - driver.tap = AsyncMock(return_value=True) - return driver - - -@pytest.fixture -def mock_state(): - return MagicMock(spec=State) - - -def test_click_tool_subclass_and_registry(): - """Verify ClickTool is a subclass of ArtemisTool and properly registered.""" - assert issubclass(ClickTool, ArtemisTool) - assert issubclass(Click, ArtemisTool) - assert isinstance(click, ArtemisTool) - assert isinstance(click, ClickTool) - - assert click.name == "click" - assert click.category == "action" - assert click.args_schema == ClickArgs - - # Registry lookup - reg_tool = ToolRegistry.get("click") - assert reg_tool is not None - assert isinstance(reg_tool, ClickTool) - - # GenAI FunctionDeclaration export - declaration = click.to_genai_declaration() - assert declaration.name == "click" - assert "target" in declaration.parameters.properties - - # Wrapper check - assert click_wrapper is not None - assert click_wrapper.tool_fn_getter == get_click_tool - - -@pytest.mark.asyncio -async def test_click_direct_execution_with_driver(mock_driver): - """Verify direct execution of ClickTool with BaseDeviceDriver.""" - result = await click.execute(driver=mock_driver, target=[500, 500]) - mock_driver.tap.assert_called_once_with(540, 1200) - assert "Clicked at [500, 500] (normalized) successfully." in result - - -@pytest.mark.asyncio -async def test_click_direct_execution_with_ctx(mock_ctx): - """Verify direct execution of ClickTool with ArtemisContext.""" - with patch("artemis.tools.mobile.exec_tools.UnifiedMobileController") as mock_controller_cls: - mock_controller_inst = MagicMock() - mock_controller_inst.tap_at = AsyncMock(return_value=MagicMock(error=None)) - mock_controller_cls.return_value = mock_controller_inst - - result = await click.execute(ctx=mock_ctx, target=[250, 750]) - mock_controller_inst.tap_at.assert_called_once_with(270, 1800) - assert "Clicked at [250, 750] (normalized) successfully." in result - - -@pytest.mark.asyncio -async def test_click_with_state_command(mock_ctx, mock_state): - """Verify ClickTool returns ToolMessage when state is provided.""" - with patch("artemis.tools.mobile.exec_tools.UnifiedMobileController") as mock_controller_cls: - mock_controller_inst = MagicMock() - mock_controller_inst.tap_at = AsyncMock(return_value=MagicMock(error=None)) - mock_controller_cls.return_value = mock_controller_inst - - cmd = await click.execute( - ctx=mock_ctx, - target=[100, 200], - tool_call_id="call_click_1", - state=mock_state, - ) - assert isinstance(cmd, ToolMessage) - assert cmd.tool_call_id == "call_click_1" - assert cmd.status == "success" - assert "Clicked at [100, 200]" in cmd.content - - -@pytest.mark.asyncio -async def test_click_callable_execution(mock_driver): - """Verify invoking click directly as a callable.""" - result = await click(driver=mock_driver, target=[100, 100]) - assert "Clicked at [100, 100] (normalized) successfully." in result - - -@pytest.mark.asyncio -async def test_click_invalid_target(mock_ctx): - """Verify error handling on invalid target coordinates.""" - result = await click.execute(ctx=mock_ctx, target=[]) - assert "Error during click: Invalid target coordinates" in result - - -@pytest.mark.asyncio -async def test_get_click_tool_langchain_ainvoke(mock_ctx): - """Verify get_click_tool exports a LangChain tool that works with ainvoke.""" - click_tool = get_click_tool(mock_ctx) - assert click_tool.name == "click" - - with patch("artemis.tools.mobile.exec_tools.UnifiedMobileController") as mock_controller_cls: - mock_controller_inst = MagicMock() - mock_controller_inst.tap_at = AsyncMock(return_value=MagicMock(error=None)) - mock_controller_cls.return_value = mock_controller_inst - - result = await click_tool.ainvoke({"target": [500, 500]}) - assert "Clicked at [500, 500] (normalized) successfully." in result diff --git a/tests/unit/tools/test_input_text_tool.py b/tests/unit/tools/test_input_text_tool.py deleted file mode 100644 index be7e648..0000000 --- a/tests/unit/tools/test_input_text_tool.py +++ /dev/null @@ -1,172 +0,0 @@ -# Copyright 2026 Google LLC -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -from unittest.mock import AsyncMock, MagicMock, patch - -from artemis.context import ArtemisContext -from artemis.drivers.base import BaseDeviceDriver -from artemis.graph.state import State -from artemis.tools.base import ArtemisTool, ToolRegistry -from artemis.tools.mobile.exec_tools import ( - InputText, - InputTextArgs, - InputTextTool, - get_input_text_tool, - input_text, - input_text_wrapper, -) -from langchain_core.messages import ToolMessage -import pytest - - -@pytest.fixture -def mock_ctx(): - ctx = MagicMock(spec=ArtemisContext) - ctx.device = MagicMock() - ctx.device.device_width = 1080 - ctx.device.device_height = 2400 - return ctx - - -@pytest.fixture -def mock_driver(): - driver = MagicMock(spec=BaseDeviceDriver) - driver.screen_size = (1080, 2400) - driver.tap = AsyncMock(return_value=True) - driver.input_text = AsyncMock(return_value=True) - return driver - - -@pytest.fixture -def mock_state(): - return MagicMock(spec=State) - - -def test_input_text_tool_subclass_and_registry(): - """Verify InputTextTool is a subclass of ArtemisTool and properly registered.""" - assert issubclass(InputTextTool, ArtemisTool) - assert issubclass(InputText, ArtemisTool) - assert isinstance(input_text, ArtemisTool) - assert isinstance(input_text, InputTextTool) - - assert input_text.name == "input_text" - assert input_text.category == "action" - assert input_text.args_schema == InputTextArgs - - # Registry lookup - reg_tool = ToolRegistry.get("input_text") - assert reg_tool is not None - assert isinstance(reg_tool, InputTextTool) - - # GenAI FunctionDeclaration export - declaration = input_text.to_genai_declaration() - assert declaration.name == "input_text" - assert "text" in declaration.parameters.properties - assert "target" in declaration.parameters.properties - assert "clear_exist" in declaration.parameters.properties - - # Wrapper check - assert input_text_wrapper is not None - assert input_text_wrapper.tool_fn_getter == get_input_text_tool - - -@pytest.mark.asyncio -async def test_input_text_direct_execution_with_driver(mock_driver): - """Verify direct execution of InputTextTool with BaseDeviceDriver.""" - result = await input_text.execute( - driver=mock_driver, text="hello world", target=[500, 500], clear_exist=True - ) - mock_driver.tap.assert_called_once_with(540, 1200) - mock_driver.input_text.assert_called_once_with("hello world", clear_existing=True) - assert "Typed 'hello world' successfully." in result - - -@pytest.mark.asyncio -async def test_input_text_direct_execution_with_ctx(mock_ctx): - """Verify direct execution of InputTextTool with ArtemisContext.""" - with patch("artemis.tools.mobile.exec_tools.UnifiedMobileController") as mock_controller_cls: - mock_controller_inst = MagicMock() - mock_controller_inst.tap_at = AsyncMock(return_value=MagicMock(error=None)) - mock_controller_inst.erase_text = AsyncMock(return_value=True) - mock_controller_inst.type_text = AsyncMock(return_value=True) - mock_controller_cls.return_value = mock_controller_inst - - result = await input_text.execute( - ctx=mock_ctx, text="test input", target=[250, 750], clear_exist=True - ) - mock_controller_inst.tap_at.assert_called_once_with(270, 1800) - mock_controller_inst.erase_text.assert_called_once() - mock_controller_inst.type_text.assert_called_once_with("test input", clear_existing=False) - assert "Typed 'test input' successfully." in result - - -@pytest.mark.asyncio -async def test_input_text_with_state_command(mock_ctx, mock_state): - """Verify InputTextTool returns ToolMessage when state is provided.""" - with patch("artemis.tools.mobile.exec_tools.UnifiedMobileController") as mock_controller_cls: - mock_controller_inst = MagicMock() - mock_controller_inst.tap_at = AsyncMock(return_value=MagicMock(error=None)) - mock_controller_inst.erase_text = AsyncMock(return_value=True) - mock_controller_inst.type_text = AsyncMock(return_value=True) - mock_controller_cls.return_value = mock_controller_inst - - cmd = await input_text.execute( - ctx=mock_ctx, - text="query", - target=[100, 200], - tool_call_id="call_it_1", - state=mock_state, - ) - assert isinstance(cmd, ToolMessage) - assert cmd.tool_call_id == "call_it_1" - assert cmd.status == "success" - assert "Typed 'query'" in cmd.content - - -@pytest.mark.asyncio -async def test_input_text_callable_execution(mock_driver): - """Verify invoking input_text directly as a callable.""" - result = await input_text(driver=mock_driver, text="call text", target=[100, 100]) - assert "Typed 'call text' successfully." in result - - -@pytest.mark.asyncio -async def test_input_text_invalid_target(mock_ctx): - """Verify error handling on invalid target coordinates.""" - result = await input_text.execute(ctx=mock_ctx, text="bad", target=[]) - assert "Error during input text: Invalid target coordinates" in result - - -@pytest.mark.asyncio -async def test_get_input_text_tool_langchain_ainvoke(mock_ctx): - """Verify get_input_text_tool exports a LangChain tool that works with ainvoke.""" - it_tool = get_input_text_tool(mock_ctx) - assert it_tool.name == "input_text" - - with patch("artemis.tools.mobile.exec_tools.UnifiedMobileController") as mock_controller_cls: - mock_controller_inst = MagicMock() - mock_controller_inst.tap_at = AsyncMock(return_value=MagicMock(error=None)) - mock_controller_inst.press_key = AsyncMock(return_value=True) - mock_controller_inst.type_text = AsyncMock(return_value=True) - mock_controller_cls.return_value = mock_controller_inst - - result = await it_tool.ainvoke( - { - "text": "abc", - "target": [500, 500], - "clear_exist": False, - } - ) - mock_controller_inst.press_key.assert_called_once_with("123") - assert "Typed 'abc' successfully." in result diff --git a/tests/unit/tools/test_long_press_tool.py b/tests/unit/tools/test_long_press_tool.py deleted file mode 100644 index fa0cde9..0000000 --- a/tests/unit/tools/test_long_press_tool.py +++ /dev/null @@ -1,152 +0,0 @@ -# Copyright 2026 Google LLC -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -from unittest.mock import AsyncMock, MagicMock, patch - -from artemis.context import ArtemisContext -from artemis.drivers.base import BaseDeviceDriver -from artemis.graph.state import State -from artemis.tools.base import ArtemisTool, ToolRegistry -from artemis.tools.mobile.exec_tools import ( - LongPress, - LongPressArgs, - LongPressTool, - get_long_press_tool, - long_press, - long_press_wrapper, -) -from langchain_core.messages import ToolMessage -import pytest - - -@pytest.fixture -def mock_ctx(): - ctx = MagicMock(spec=ArtemisContext) - ctx.device = MagicMock() - ctx.device.device_width = 1080 - ctx.device.device_height = 2400 - return ctx - - -@pytest.fixture -def mock_driver(): - driver = MagicMock(spec=BaseDeviceDriver) - driver.screen_size = (1080, 2400) - driver.long_press = AsyncMock(return_value=True) - return driver - - -@pytest.fixture -def mock_state(): - return MagicMock(spec=State) - - -def test_long_press_tool_subclass_and_registry(): - """Verify LongPressTool is a subclass of ArtemisTool and properly registered.""" - assert issubclass(LongPressTool, ArtemisTool) - assert issubclass(LongPress, ArtemisTool) - assert isinstance(long_press, ArtemisTool) - assert isinstance(long_press, LongPressTool) - - assert long_press.name == "long_press" - assert long_press.category == "action" - assert long_press.args_schema == LongPressArgs - - # Registry lookup - reg_tool = ToolRegistry.get("long_press") - assert reg_tool is not None - assert isinstance(reg_tool, LongPressTool) - - # GenAI FunctionDeclaration export - declaration = long_press.to_genai_declaration() - assert declaration.name == "long_press" - assert "target" in declaration.parameters.properties - assert "duration" in declaration.parameters.properties - - # Wrapper check - assert long_press_wrapper is not None - assert long_press_wrapper.tool_fn_getter == get_long_press_tool - - -@pytest.mark.asyncio -async def test_long_press_direct_execution_with_driver(mock_driver): - """Verify direct execution of LongPressTool with BaseDeviceDriver.""" - result = await long_press.execute(driver=mock_driver, target=[500, 500], duration=1500) - mock_driver.long_press.assert_called_once_with(540, 1200, duration_ms=1500) - assert "Long pressed at [500, 500] (normalized) successfully." in result - - -@pytest.mark.asyncio -async def test_long_press_direct_execution_with_ctx(mock_ctx): - """Verify direct execution of LongPressTool with ArtemisContext.""" - with patch("artemis.tools.mobile.exec_tools.UnifiedMobileController") as mock_controller_cls: - mock_controller_inst = MagicMock() - mock_controller_inst.tap_at = AsyncMock(return_value=MagicMock(error=None)) - mock_controller_cls.return_value = mock_controller_inst - - result = await long_press.execute(ctx=mock_ctx, target=[250, 750], duration=2000) - mock_controller_inst.tap_at.assert_called_once_with( - 270, 1800, long_press=True, long_press_duration=2000 - ) - assert "Long pressed at [250, 750] (normalized) successfully." in result - - -@pytest.mark.asyncio -async def test_long_press_with_state_command(mock_ctx, mock_state): - """Verify LongPressTool returns ToolMessage when state is provided.""" - with patch("artemis.tools.mobile.exec_tools.UnifiedMobileController") as mock_controller_cls: - mock_controller_inst = MagicMock() - mock_controller_inst.tap_at = AsyncMock(return_value=MagicMock(error=None)) - mock_controller_cls.return_value = mock_controller_inst - - cmd = await long_press.execute( - ctx=mock_ctx, - target=[100, 200], - duration=1200, - tool_call_id="call_lp_1", - state=mock_state, - ) - assert isinstance(cmd, ToolMessage) - assert cmd.tool_call_id == "call_lp_1" - assert cmd.status == "success" - assert "Long pressed at [100, 200]" in cmd.content - - -@pytest.mark.asyncio -async def test_long_press_callable_execution(mock_driver): - """Verify invoking long_press directly as a callable.""" - result = await long_press(driver=mock_driver, target=[100, 100]) - assert "Long pressed at [100, 100] (normalized) successfully." in result - - -@pytest.mark.asyncio -async def test_long_press_invalid_target(mock_ctx): - """Verify error handling on invalid target coordinates.""" - result = await long_press.execute(ctx=mock_ctx, target=[]) - assert "Error during long press: Invalid target coordinates" in result - - -@pytest.mark.asyncio -async def test_get_long_press_tool_langchain_ainvoke(mock_ctx): - """Verify get_long_press_tool exports a LangChain tool that works with ainvoke.""" - lp_tool = get_long_press_tool(mock_ctx) - assert lp_tool.name == "long_press" - - with patch("artemis.tools.mobile.exec_tools.UnifiedMobileController") as mock_controller_cls: - mock_controller_inst = MagicMock() - mock_controller_inst.tap_at = AsyncMock(return_value=MagicMock(error=None)) - mock_controller_cls.return_value = mock_controller_inst - - result = await lp_tool.ainvoke({"target": [500, 500], "duration": 1500}) - assert "Long pressed at [500, 500] (normalized) successfully." in result diff --git a/tests/unit/tools/test_press_key_tool.py b/tests/unit/tools/test_press_key_tool.py deleted file mode 100644 index 78b3fe0..0000000 --- a/tests/unit/tools/test_press_key_tool.py +++ /dev/null @@ -1,164 +0,0 @@ -# Copyright 2026 Google LLC -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -from unittest.mock import AsyncMock, MagicMock, patch - -from artemis.context import ArtemisContext -from artemis.drivers.base import BaseDeviceDriver -from artemis.graph.state import State -from artemis.tools.base import ArtemisTool, ToolRegistry -from artemis.tools.mobile.exec_tools import ( - PressKey, - PressKeyArgs, - PressKeyTool, - get_press_key_tool, - press_key, - press_key_wrapper, -) -from langchain_core.messages import ToolMessage -import pytest - - -@pytest.fixture -def mock_ctx(): - ctx = MagicMock(spec=ArtemisContext) - ctx.device = MagicMock() - return ctx - - -@pytest.fixture -def mock_driver(): - driver = MagicMock(spec=BaseDeviceDriver) - driver.press_key = AsyncMock(return_value=True) - return driver - - -@pytest.fixture -def mock_state(): - return MagicMock(spec=State) - - -def test_press_key_tool_subclass_and_registry(): - """Verify PressKeyTool is a subclass of ArtemisTool and properly registered.""" - assert issubclass(PressKeyTool, ArtemisTool) - assert issubclass(PressKey, ArtemisTool) - assert isinstance(press_key, ArtemisTool) - assert isinstance(press_key, PressKeyTool) - - assert press_key.name == "press_key" - assert press_key.category == "action" - assert press_key.args_schema == PressKeyArgs - - # Registry lookup - reg_tool = ToolRegistry.get("press_key") - assert reg_tool is not None - assert isinstance(reg_tool, PressKeyTool) - - # GenAI FunctionDeclaration export - declaration = press_key.to_genai_declaration() - assert declaration.name == "press_key" - assert "key" in declaration.parameters.properties - - # Wrapper check - assert press_key_wrapper is not None - assert press_key_wrapper.tool_fn_getter == get_press_key_tool - - -@pytest.mark.asyncio -async def test_press_key_direct_execution_with_driver(mock_driver): - """Verify direct execution with BaseDeviceDriver.""" - result = await press_key.execute(driver=mock_driver, key="BACK") - mock_driver.press_key.assert_called_once_with("BACK") - assert "Pressed key 'BACK' successfully." in result - - -@pytest.mark.asyncio -async def test_press_key_direct_execution_with_ctx(mock_ctx): - """Verify direct execution with ArtemisContext for various keys.""" - with patch("artemis.tools.mobile.exec_tools.UnifiedMobileController") as mock_controller_cls: - mock_controller_inst = MagicMock() - mock_controller_inst.press_enter = AsyncMock(return_value=True) - mock_controller_inst.go_back = AsyncMock(return_value=True) - mock_controller_inst.go_home = AsyncMock(return_value=True) - mock_controller_inst.press_key = AsyncMock(return_value=True) - mock_controller_cls.return_value = mock_controller_inst - - # Test ENTER - res_enter = await press_key.execute(ctx=mock_ctx, key="ENTER") - mock_controller_inst.press_enter.assert_called_once() - assert "Pressed key 'ENTER' successfully." in res_enter - - # Test BACK - res_back = await press_key.execute(ctx=mock_ctx, key="BACK") - mock_controller_inst.go_back.assert_called_once() - assert "Pressed key 'BACK' successfully." in res_back - - # Test HOME - res_home = await press_key.execute(ctx=mock_ctx, key="HOME") - mock_controller_inst.go_home.assert_called_once() - assert "Pressed key 'HOME' successfully." in res_home - - # Test APP_SWITCH - res_app_switch = await press_key.execute(ctx=mock_ctx, key="APP_SWITCH") - mock_controller_inst.press_key.assert_called_with("KEYCODE_APP_SWITCH") - assert "Pressed key 'APP_SWITCH' successfully." in res_app_switch - - -@pytest.mark.asyncio -async def test_press_key_with_state_command(mock_ctx, mock_state): - """Verify PressKeyTool returns ToolMessage when state is provided.""" - with patch("artemis.tools.mobile.exec_tools.UnifiedMobileController") as mock_controller_cls: - mock_controller_inst = MagicMock() - mock_controller_inst.go_home = AsyncMock(return_value=True) - mock_controller_cls.return_value = mock_controller_inst - - cmd = await press_key.execute( - ctx=mock_ctx, - key="HOME", - tool_call_id="call_pk_1", - state=mock_state, - ) - assert isinstance(cmd, ToolMessage) - assert cmd.tool_call_id == "call_pk_1" - assert cmd.status == "success" - assert "Pressed key 'HOME' successfully." in cmd.content - - -@pytest.mark.asyncio -async def test_press_key_callable_execution(mock_driver): - """Verify invoking press_key directly as a callable.""" - result = await press_key(driver=mock_driver, key="ENTER") - assert "Pressed key 'ENTER' successfully." in result - - -@pytest.mark.asyncio -async def test_press_key_missing_key(mock_ctx): - """Verify error handling when key is empty or missing.""" - result = await press_key.execute(ctx=mock_ctx, key="") - assert "Error during press key: Key parameter is required." in result - - -@pytest.mark.asyncio -async def test_get_press_key_tool_langchain_ainvoke(mock_ctx): - """Verify get_press_key_tool exports a LangChain tool that works with ainvoke.""" - pk_tool = get_press_key_tool(mock_ctx) - assert pk_tool.name == "press_key" - - with patch("artemis.tools.mobile.exec_tools.UnifiedMobileController") as mock_controller_cls: - mock_controller_inst = MagicMock() - mock_controller_inst.go_back = AsyncMock(return_value=True) - mock_controller_cls.return_value = mock_controller_inst - - result = await pk_tool.ainvoke({"key": "BACK"}) - assert "Pressed key 'BACK' successfully." in result diff --git a/tests/unit/tools/test_swipe_tool.py b/tests/unit/tools/test_swipe_tool.py deleted file mode 100644 index d1208f5..0000000 --- a/tests/unit/tools/test_swipe_tool.py +++ /dev/null @@ -1,316 +0,0 @@ -# Copyright 2026 Google LLC -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -from unittest.mock import AsyncMock, MagicMock, patch - -from artemis.context import ArtemisContext -from artemis.drivers.base import BaseDeviceDriver -from artemis.graph.state import State -from artemis.tools.base import ArtemisTool, ToolRegistry -from artemis.tools.mobile.exec_tools import ( - Swipe, - SwipeArgs, - SwipeTool, - get_swipe_tool, - swipe, - swipe_wrapper, -) -from langchain_core.messages import ToolMessage -import pytest - - -@pytest.fixture -def mock_ctx(): - ctx = MagicMock(spec=ArtemisContext) - ctx.device = MagicMock() - ctx.device.device_width = 1080 - ctx.device.device_height = 2400 - return ctx - - -@pytest.fixture -def mock_driver(): - driver = MagicMock(spec=BaseDeviceDriver) - driver.screen_size = (1080, 2400) - driver.swipe_direction = AsyncMock(return_value=True) - driver.swipe = AsyncMock(return_value=True) - return driver - - -@pytest.fixture -def mock_state(): - return MagicMock(spec=State) - - -def test_swipe_tool_subclass_and_registry(): - """Verify SwipeTool is a subclass of ArtemisTool and properly registered.""" - assert issubclass(SwipeTool, ArtemisTool) - assert issubclass(Swipe, ArtemisTool) - assert isinstance(swipe, ArtemisTool) - assert isinstance(swipe, SwipeTool) - - assert swipe.name == "swipe" - assert swipe.category == "action" - assert swipe.args_schema == SwipeArgs - - # Registry lookup - reg_tool = ToolRegistry.get("swipe") - assert reg_tool is not None - assert isinstance(reg_tool, ArtemisTool) - - # GenAI FunctionDeclaration export - declaration = swipe.to_genai_declaration() - assert declaration.name == "swipe" - assert "action" in declaration.parameters.properties - assert "duration" in declaration.parameters.properties - - # Wrapper check - assert swipe_wrapper is not None - assert swipe_wrapper.tool_fn_getter == get_swipe_tool - - -@pytest.mark.asyncio -async def test_swipe_direct_execution_with_driver_direction(mock_driver): - """Verify direct execution with direction on BaseDeviceDriver.""" - result = await swipe.execute(driver=mock_driver, action="up", duration=500) - mock_driver.swipe_direction.assert_called_once_with("up", duration_ms=500) - assert "Swipe completed successfully." in result - - -@pytest.mark.asyncio -async def test_swipe_direct_execution_with_driver_coords(mock_driver): - """Verify direct execution with coordinates on BaseDeviceDriver.""" - result = await swipe.execute(driver=mock_driver, action=[100, 200, 300, 400], duration=600) - mock_driver.swipe.assert_called_once_with(108, 480, 324, 960, duration_ms=600) - assert "Swipe completed successfully." in result - - -@pytest.mark.asyncio -async def test_swipe_direct_execution_with_ctx_direction(mock_ctx): - """Verify direct execution with direction on ArtemisContext.""" - with patch("artemis.tools.mobile.exec_tools.UnifiedMobileController") as mock_controller_cls: - mock_controller_inst = MagicMock() - mock_controller_inst.swipe_coords = AsyncMock(return_value=None) - mock_controller_cls.return_value = mock_controller_inst - - result = await swipe.execute(ctx=mock_ctx, action="down", duration=400) - mock_controller_inst.swipe_coords.assert_called_once_with(648, 720, 648, 1680, 400) - assert "Swipe completed successfully." in result - - -@pytest.mark.asyncio -async def test_swipe_direct_execution_with_ctx_coords(mock_ctx): - """Verify direct execution with coordinates on ArtemisContext.""" - with patch("artemis.tools.mobile.exec_tools.UnifiedMobileController") as mock_controller_cls: - mock_controller_inst = MagicMock() - mock_controller_inst.swipe_coords = AsyncMock(return_value=None) - mock_controller_cls.return_value = mock_controller_inst - - result = await swipe.execute(ctx=mock_ctx, action=[200, 300, 800, 900], duration=800) - mock_controller_inst.swipe_coords.assert_called_once_with(216, 720, 864, 2160, 800) - assert "Swipe completed successfully." in result - - -@pytest.mark.asyncio -async def test_swipe_with_state_command(mock_ctx, mock_state): - """Verify SwipeTool returns ToolMessage when state is provided.""" - with patch("artemis.tools.mobile.exec_tools.UnifiedMobileController") as mock_controller_cls: - mock_controller_inst = MagicMock() - mock_controller_inst.swipe_coords = AsyncMock(return_value=None) - mock_controller_cls.return_value = mock_controller_inst - - cmd = await swipe.execute( - ctx=mock_ctx, - action="left", - tool_call_id="call_sw_1", - state=mock_state, - ) - assert isinstance(cmd, ToolMessage) - assert cmd.tool_call_id == "call_sw_1" - assert cmd.status == "success" - assert "Swipe completed successfully." in cmd.content - - -@pytest.mark.asyncio -async def test_swipe_callable_execution(mock_driver): - """Verify invoking swipe directly as a callable.""" - result = await swipe(driver=mock_driver, action="right") - assert "Swipe completed successfully." in result - - -@pytest.mark.asyncio -async def test_swipe_invalid_action(mock_ctx): - """Verify error handling on invalid action.""" - with patch("artemis.tools.mobile.exec_tools.UnifiedMobileController"): - result = await swipe.execute(ctx=mock_ctx, action="diagonal") - assert "Error during swipe: Invalid" in result - - -@pytest.mark.asyncio -async def test_get_swipe_tool_langchain_ainvoke(mock_ctx): - """Verify get_swipe_tool exports a LangChain tool that works with ainvoke.""" - sw_tool = get_swipe_tool(mock_ctx) - assert sw_tool.name == "swipe" - - with patch("artemis.tools.mobile.exec_tools.UnifiedMobileController") as mock_controller_cls: - mock_controller_inst = MagicMock() - mock_controller_inst.swipe_coords = AsyncMock(return_value=None) - mock_controller_cls.return_value = mock_controller_inst - - result = await sw_tool.ainvoke({"action": "up"}) - assert "Swipe completed successfully." in result - - -@pytest.mark.asyncio -async def test_swipe_with_direction_and_start_end(mock_driver): - """Verify swipe works with explicit direction and start/end parameters.""" - # Direction parameter - result_dir = await swipe.execute(driver=mock_driver, direction="left", duration=500) - mock_driver.swipe_direction.assert_called_with("left", duration_ms=500) - assert "Swipe completed successfully." in result_dir - - # Start and End parameters - result_drag = await swipe.execute( - driver=mock_driver, start=[100, 200], end=[300, 400], duration=1500 - ) - mock_driver.swipe.assert_called_with(108, 480, 324, 960, duration_ms=1500) - assert "Swipe completed successfully." in result_drag - - -@pytest.mark.asyncio -async def test_swipe_with_coordinates_alias(mock_driver): - """Verify swipe works with coordinates alias parameter.""" - result = await swipe.execute(driver=mock_driver, coordinates=[200, 300, 400, 500], duration=800) - mock_driver.swipe.assert_called_with(216, 720, 432, 1200, duration_ms=800) - assert "Swipe completed successfully." in result - - -@pytest.mark.asyncio -async def test_swipe_with_string_coordinates(mock_driver): - """Verify swipe works with string-formatted coordinates.""" - # List format in string - result1 = await swipe.execute(driver=mock_driver, action="[920, 290, 920, 180]", duration=1000) - mock_driver.swipe.assert_called_with(993, 696, 993, 432, duration_ms=1000) - assert "Swipe completed successfully." in result1 - - # Space-separated format in string - result2 = await swipe.execute(driver=mock_driver, action="920 290 920 180", duration=1000) - assert "Swipe completed successfully." in result2 - - -def test_parse_swipe_parameters_direct(): - """Verify parse_swipe_parameters with various parameter formats.""" - from artemis.utils.coordinates import parse_swipe_parameters - - # List - k, t, d = parse_swipe_parameters([100, 200, 300, 400]) - assert k == "coords" - assert t == [100, 200, 300, 400] - - # String list - k, t, d = parse_swipe_parameters("[920, 290, 920, 180]") - assert k == "coords" - assert t == [920, 290, 920, 180] - - # String space separated - k, t, d = parse_swipe_parameters("920 290 920 180") - assert k == "coords" - assert t == [920, 290, 920, 180] - - # Dict with action string - k, t, d = parse_swipe_parameters({"action": "[920, 290, 920, 180]", "duration": 1000}) - assert k == "coords" - assert t == [920, 290, 920, 180] - assert d == 1000 - - # Dict with start and end - k, t, d = parse_swipe_parameters({"start": [100, 200], "end": [300, 400]}) - assert k == "coords" - assert t == [100, 200, 300, 400] - - # Direction string - k, t, d = parse_swipe_parameters({"action": "up"}) - assert k == "direction" - assert t == "up" - - -def test_compute_smart_swipe_coordinates_fallback(): - """Verify fallback smart swipe coordinates on full screen.""" - from artemis.utils.coordinates import compute_smart_swipe_coordinates - - sx, sy, ex, ey, dur = compute_smart_swipe_coordinates("up", width=1080, height=2400) - assert sx == 648 - assert ex == 648 - assert sy == 1680 - assert ey == 720 - assert 700 <= dur <= 900 - - -def test_compute_smart_swipe_coordinates_with_target_element(): - """Verify container-aware coordinates when target element index is specified.""" - from artemis.utils.coordinates import compute_smart_swipe_coordinates - - indexed_elements = [ - {"bounds": "[0, 200][1080, 1200]"}, # Element 1 - {"bounds": "[100, 500][900, 1500]"}, # Element 2 (width=800, height=1000) - ] - # Swipe up within element 2 - sx, sy, ex, ey, dur = compute_smart_swipe_coordinates( - direction="up", - target=2, - indexed_elements=indexed_elements, - width=1080, - height=2400, - ) - # c_left=100, c_top=500, c_width=800, c_height=1000 - # sx = ex = 100 + 800 * 0.6 = 580 - # sy = 500 + 1000 * 0.7 = 1200 - # ey = 500 + 1000 * 0.3 = 800 - assert sx == 580 - assert ex == 580 - assert sy == 1200 - assert ey == 800 - - -def test_compute_smart_swipe_coordinates_with_scrollable_hierarchy(): - """Verify auto-detection of scrollable container in UI hierarchy.""" - from artemis.utils.coordinates import compute_smart_swipe_coordinates - - ui_hierarchy = [ - { - "class": "android.widget.FrameLayout", - "bounds": "[0, 0][1080, 2400]", - "children": [ - { - "class": "androidx.recyclerview.widget.RecyclerView", - "scrollable": True, - "bounds": "[50, 300][1030, 2100]", # width=980, height=1800 - } - ], - } - ] - sx, sy, ex, ey, dur = compute_smart_swipe_coordinates( - direction="down", - ui_hierarchy=ui_hierarchy, - width=1080, - height=2400, - ) - # c_left=50, c_top=300, c_width=980, c_height=1800 - # sx = ex = 50 + 980 * 0.6 = 638 - # sy = 300 + 1800 * 0.3 = 840 - # ey = 300 + 1800 * 0.7 = 1560 - assert sx == 638 - assert ex == 638 - assert sy == 840 - assert ey == 1560