chenmingxuan/Artemis-Eapil
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 32 Wiki Activity

merge(core): integrate routing and execution cleanup

Browse Source
This commit is contained in:
somew1nd committed 2026-09-01 18:08:36 -07:00
2 parents 12c6d98aa5 a4c28865f2
commit 4724e81741
17 files changed
+481 -1878
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

artemis/context.py
-1
View File
@@ -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
artemis/llm/__init__.py
+1 -5
View File
@@ -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",
]
artemis/llm/router.py
-78
View File
@@ -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()
artemis/services/llm.py
+1 -11
View File
@@ -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()
artemis/tools/mobile/exec_tools.py
-683
View File
@@ -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",
)
artemis/utils/notes.py
+16 -15
View File
@@ -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
artemis/utils/task_tree.py
+2 -4
View File
@@ -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 ---")
tests/fixtures/action_surfaces/adb_server_manifest.json
Vendored
+223
View File
@@ -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"
}
}
}
tests/unit/mcp/test_action_specs.py
+5 -6
View File
@@ -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")
tests/unit/mcp/test_adb_server_contract.py
+227
View File
@@ -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"
)
tests/unit/test_model_router.py → tests/unit/test_llm_grounding.py
+6 -53
View File
@@ -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"
tests/unit/test_tools.py
-70
View File
@@ -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"
tests/unit/tools/test_click_tool.py
-148
View File
@@ -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
tests/unit/tools/test_input_text_tool.py
-172
View File
@@ -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
tests/unit/tools/test_long_press_tool.py
-152
View File
@@ -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
tests/unit/tools/test_press_key_tool.py
-164
View File
@@ -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
tests/unit/tools/test_swipe_tool.py
-316
View File
@@ -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