chenmingxuan
4ea7939d9c
Add optional ensure_state, gesture region and final value feedback shared by Flash and Pro. Preserve completed inputs when verification fails and stop repeated ineffective strategies. Regenerate pinned tool schemas for the new optional arguments.
926 lines
40 KiB
Python
926 lines
40 KiB
Python
# 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.
|
|
|
|
"""ADB-backed actuator: the reference implementation of the actuator contract.
|
|
|
|
Outcome messages describe the dispatched command, such as
|
|
"Tapped at [x, y] (normalized)." The agent verifies the effect from subsequent
|
|
observations.
|
|
|
|
Error convention: device-level refusals the controller *reports* (an ``error`` field,
|
|
a falsy success flag) come back as ``ActionResult(ok=False)`` with the historical
|
|
message. Unexpected exceptions are allowed to propagate -- the caller (executor or MCP
|
|
server layer) owns the ``"Error during X: {e}"`` wrapping, exactly as the original
|
|
executor's ``try/except`` did.
|
|
"""
|
|
|
|
import asyncio
|
|
import inspect
|
|
import re
|
|
import time
|
|
from typing import Any
|
|
|
|
from artemis.context import ArtemisContext
|
|
from artemis.controllers.unified_controller import UnifiedMobileController
|
|
from artemis.mcp.action_manifest import DEVICE_ACTIONS, ExtensionTool
|
|
from artemis.mcp.action_types import ActionCode, ActionResult
|
|
from artemis.utils.logger import get_logger
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
__all__ = ["AdbActuator", "find_element_at_coords", "ensure_focus_at_coords"]
|
|
|
|
|
|
def find_element_at_coords(elements: list[dict], x: int, y: int) -> dict | None:
|
|
"""Finds the smallest (leaf-most) focusable element containing pixel [x, y].
|
|
|
|
Moved from ``artemis.mcp.adb_server`` so both the stdio server and in-process
|
|
actuator share one implementation.
|
|
"""
|
|
matching_element = None
|
|
min_area = float("inf")
|
|
|
|
for elem in elements:
|
|
is_focusable = (
|
|
elem.get("focusable") == "true"
|
|
or elem.get("clickable") == "true"
|
|
or "EditText" in str(elem.get("class", ""))
|
|
)
|
|
if not is_focusable:
|
|
continue
|
|
|
|
bounds_str = elem.get("bounds")
|
|
if bounds_str and isinstance(bounds_str, str):
|
|
match = re.match(r"\[(\-?\d+),(\-?\d+)\]\[(\-?\d+),(\-?\d+)\]", bounds_str)
|
|
if match:
|
|
x1, y1, x2, y2 = map(int, match.groups())
|
|
if x1 <= x <= x2 and y1 <= y <= y2:
|
|
area = (x2 - x1) * (y2 - y1)
|
|
if area < min_area:
|
|
min_area = area
|
|
matching_element = elem
|
|
|
|
return matching_element
|
|
|
|
|
|
async def ensure_focus_at_coords(controller, x: int, y: int) -> str | None:
|
|
"""Ensures the element at pixel [x, y] is focused, tapping only if needed.
|
|
|
|
Returns an error string on failure, ``None`` on success. Moved from
|
|
``artemis.mcp.adb_server``.
|
|
"""
|
|
try:
|
|
elements = await controller.get_ui_elements()
|
|
elem = find_element_at_coords(elements, x, y)
|
|
if elem and elem.get("focused") == "true":
|
|
logger.info(f"Element under [{x}, {y}] is already focused. Skipping tap.")
|
|
return None
|
|
except Exception as e:
|
|
logger.warning(f"Failed to check focus status: {e}. Falling back to unconditional tap.")
|
|
|
|
result = await controller.tap_at(x=x, y=y)
|
|
if hasattr(result, "error") and result.error:
|
|
return result.error
|
|
|
|
# Wait for the UI to settle and keyboard to pop up after tapping
|
|
await asyncio.sleep(1.0)
|
|
return None
|
|
|
|
|
|
class AdbActuator:
|
|
"""Drives an Android device through ``UnifiedMobileController`` over ADB."""
|
|
|
|
def __init__(
|
|
self,
|
|
ctx: ArtemisContext,
|
|
controller: UnifiedMobileController | None = None,
|
|
):
|
|
self.ctx = ctx
|
|
self.controller = controller or UnifiedMobileController(ctx)
|
|
from artemis.mcp.value_feedback import ValueFeedbackRunner
|
|
|
|
self._value_feedback = ValueFeedbackRunner()
|
|
|
|
# --- Contract --------------------------------------------------------------------
|
|
|
|
def capabilities(self) -> frozenset[str]:
|
|
from artemis.mcp.native_actions import NATIVE_ACTIONS, native_tools_for_capabilities
|
|
|
|
native = self.controller.driver.native_capabilities
|
|
return (DEVICE_ACTIONS - NATIVE_ACTIONS) | native_tools_for_capabilities(native)
|
|
|
|
def extensions(self) -> list[ExtensionTool]:
|
|
return []
|
|
|
|
# --- Coordinate helpers ----------------------------------------------------------
|
|
|
|
def _dims(self) -> tuple[int, int]:
|
|
width = getattr(self.ctx.device, "device_width", 1080) if self.ctx.device else 1080
|
|
height = getattr(self.ctx.device, "device_height", 2400) if self.ctx.device else 2400
|
|
return width or 1080, height or 2400
|
|
|
|
def _to_px(self, nx: int, ny: int) -> tuple[int, int]:
|
|
width, height = self._dims()
|
|
x = int(max(0, min(width - 1, int(nx) * width / 1000)))
|
|
y = int(max(0, min(height - 1, int(ny) * height / 1000)))
|
|
return x, y
|
|
|
|
# --- Required action -------------------------------------------------------------
|
|
|
|
async def _native_action(self, name: str, args: dict[str, Any]) -> ActionResult:
|
|
from artemis.mcp.native_actions import READ_ONLY_ACTIONS, validate_native_action
|
|
from artemis.mcp.native_executor import execute_native
|
|
|
|
try:
|
|
request = validate_native_action(name, args, *self._dims())
|
|
except ValueError as exc:
|
|
return ActionResult.failure(
|
|
name,
|
|
str(exc),
|
|
code=ActionCode.INVALID_ARGS,
|
|
metadata={"execution": "not_started", "side_effect": "none"},
|
|
)
|
|
if name not in self.capabilities():
|
|
return ActionResult.failure(
|
|
name,
|
|
"The connected backend does not advertise this native action.",
|
|
code=ActionCode.UNSUPPORTED,
|
|
metadata={"execution": "not_started", "side_effect": "none"},
|
|
)
|
|
if resource_id := getattr(request, "require_visible", None):
|
|
try:
|
|
guard = await self._visibility_guard(resource_id)
|
|
except (OSError, RuntimeError, ValueError) as exc:
|
|
return ActionResult.failure(
|
|
name,
|
|
f"Visibility check failed before input: {exc}",
|
|
metadata={"execution": "not_started", "side_effect": "none"},
|
|
)
|
|
if not guard["visible"]:
|
|
return ActionResult.failure(
|
|
name,
|
|
"Required native control is not visible; no input executed.",
|
|
code=ActionCode.TARGET_NOT_FOUND,
|
|
metadata={
|
|
"execution": "not_started",
|
|
"side_effect": "none",
|
|
"visibility_guard": guard,
|
|
},
|
|
)
|
|
receipt = await execute_native(name, request, *self._dims(), self.controller.native_request)
|
|
receipt["read_only"] = name in READ_ONLY_ACTIONS
|
|
receipt["side_effect"] = (
|
|
"none"
|
|
if name in READ_ONLY_ACTIONS or receipt.get("execution") in {"not_started", "observed"}
|
|
else "possible"
|
|
)
|
|
if receipt.get("success"):
|
|
if name in {"pinch", "touch_gesture"} and receipt.get("execution") == "completed":
|
|
self._value_feedback.history.clear()
|
|
return ActionResult.success(
|
|
name,
|
|
"Native operation completed. Inspect the receipt and current app state; business verification remains separate.",
|
|
metadata=receipt,
|
|
)
|
|
codes = {
|
|
"UNSUPPORTED": ActionCode.UNSUPPORTED,
|
|
"NOT_FOUND": ActionCode.TARGET_NOT_FOUND,
|
|
"INVALID_ARGUMENT": ActionCode.INVALID_ARGS,
|
|
"TIMEOUT": ActionCode.TIMEOUT,
|
|
}
|
|
return ActionResult.failure(
|
|
name,
|
|
str(receipt.get("error") or receipt.get("code") or "Native action failed."),
|
|
code=codes.get(receipt.get("code"), ActionCode.DEVICE_ERROR),
|
|
metadata=receipt,
|
|
)
|
|
|
|
async def pinch(self, **kwargs: Any) -> ActionResult:
|
|
return await self._native_action("pinch", kwargs)
|
|
|
|
async def touch_gesture(self, **kwargs: Any) -> ActionResult:
|
|
return await self._native_action("touch_gesture", kwargs)
|
|
|
|
async def read_control(self, **kwargs: Any) -> ActionResult:
|
|
return await self._native_action("read_control", kwargs)
|
|
|
|
async def control_action(self, **kwargs: Any) -> ActionResult:
|
|
return await self._native_action("control_action", kwargs)
|
|
|
|
async def wait_until(self, **kwargs: Any) -> ActionResult:
|
|
return await self._native_action("wait_until", kwargs)
|
|
|
|
async def scroll_to(self, **kwargs: Any) -> ActionResult:
|
|
return await self._native_action("scroll_to", kwargs)
|
|
|
|
async def drag(self, **kwargs: Any) -> ActionResult:
|
|
"""Dispatch exact pixel geometry without a normalized-coordinate round trip."""
|
|
from artemis.mcp.gestures import DragRequest, plan_drag
|
|
|
|
try:
|
|
request = DragRequest.model_validate(kwargs)
|
|
planned = plan_drag(request, *self._dims())
|
|
except ValueError as exc:
|
|
return ActionResult.failure("drag", str(exc), code=ActionCode.INVALID_ARGS)
|
|
if request.feedback is not None and "read_control" not in self.capabilities():
|
|
return ActionResult.failure(
|
|
"drag",
|
|
"Value feedback requires native read_control; no input executed.",
|
|
code=ActionCode.UNSUPPORTED,
|
|
metadata={"execution": "not_started", "side_effect": "none"},
|
|
)
|
|
return await self._value_feedback.run(
|
|
request.feedback,
|
|
[planned],
|
|
"drag",
|
|
self.controller.native_request,
|
|
lambda: self._dispatch_drag(planned),
|
|
)
|
|
|
|
async def _dispatch_drag(self, planned: dict) -> ActionResult:
|
|
metadata = dict(planned)
|
|
metadata.pop("action")
|
|
if resource_id := planned.get("require_visible"):
|
|
try:
|
|
guard = await self._visibility_guard(resource_id)
|
|
except (OSError, RuntimeError) as exc:
|
|
return ActionResult.failure(
|
|
"drag",
|
|
f"Visibility check failed; no input executed: {exc}",
|
|
metadata={**metadata, "execution": "not_started", "side_effect": "none"},
|
|
)
|
|
metadata["visibility_guard"] = guard
|
|
if not guard["visible"]:
|
|
return ActionResult.failure(
|
|
"drag",
|
|
"TARGET_NOT_FOUND: required panel control has disappeared; no drag executed. Re-establish the panel and operate it in one guarded action_sequence.",
|
|
code=ActionCode.TARGET_NOT_FOUND,
|
|
metadata={**metadata, "execution": "not_started", "side_effect": "none"},
|
|
)
|
|
error = await self.controller.swipe_coords(*planned["coordinates"], planned["duration_ms"])
|
|
if error:
|
|
return ActionResult.failure("drag", str(error), metadata=metadata)
|
|
return ActionResult.success(
|
|
"drag",
|
|
f"Dragged from {planned['coordinates'][:2]} to {planned['coordinates'][2:]}"
|
|
f" (pixel) over {planned['duration_ms']}ms. Verify the scale/value in the next observation.",
|
|
duration_ms=planned["duration_ms"],
|
|
metadata=metadata,
|
|
)
|
|
|
|
async def _live_ui_elements(self) -> list[dict]:
|
|
elements = await self.controller.get_ui_elements()
|
|
if not isinstance(elements, list) or not elements:
|
|
raise RuntimeError("No live UI hierarchy available for the visibility guard.")
|
|
return elements
|
|
|
|
async def _visibility_guard(self, resource_id: str, elements: list[dict] | None = None) -> dict:
|
|
"""Read live UI locally; never treat failed/empty hierarchy as absence."""
|
|
if elements is None:
|
|
elements = await self._live_ui_elements()
|
|
visible = any(
|
|
(e.get("resource_id") or e.get("resource-id")) == resource_id
|
|
and e.get("visible-to-user", True) not in (False, "false")
|
|
for e in elements
|
|
if isinstance(e, dict)
|
|
)
|
|
return {"resource_id": resource_id, "visible": visible, "checked_at": time.time()}
|
|
|
|
def _live_target_coordinates(
|
|
self, resource_id: str, elements: list[dict], expected_point: tuple[int, int] | None = None
|
|
) -> list[int]:
|
|
from artemis.utils.visualization import parse_bounds
|
|
|
|
width, height = self._dims()
|
|
matches = []
|
|
for element in elements:
|
|
if (
|
|
element.get("resource_id") or element.get("resource-id")
|
|
) != resource_id or element.get("visible-to-user", True) in (False, "false"):
|
|
continue
|
|
raw_bounds = element.get("bounds")
|
|
bounds = parse_bounds(raw_bounds) if isinstance(raw_bounds, str) else raw_bounds
|
|
if not (
|
|
isinstance(bounds, (list, tuple))
|
|
and len(bounds) == 4
|
|
and all(isinstance(v, int) and not isinstance(v, bool) for v in bounds)
|
|
):
|
|
continue
|
|
if bounds:
|
|
left, top, right, bottom = bounds
|
|
left, top = max(0, left), max(0, top)
|
|
right, bottom = min(width, right), min(height, bottom)
|
|
if expected_point and not (
|
|
left <= expected_point[0] < right and top <= expected_point[1] < bottom
|
|
):
|
|
continue
|
|
if left < right and top < bottom:
|
|
matches.append([int((left + right) / 2), int((top + bottom) / 2)])
|
|
if len(matches) != 1:
|
|
raise LookupError(
|
|
f"TARGET_NOT_FOUND: {resource_id} has {len(matches)} visible matches; no further tap executed. Use a unique observed resource_id or re-observed coordinates."
|
|
)
|
|
return matches[0]
|
|
|
|
async def _tap_native_target(
|
|
self,
|
|
resource_id: str,
|
|
receipt: dict,
|
|
*,
|
|
times: int = 1,
|
|
delay_ms: int = 100,
|
|
elements: list[dict] | None = None,
|
|
expected_point: tuple[int, int] | None = None,
|
|
) -> str | None:
|
|
"""Revalidate every repeated native tap, retaining partial input receipts."""
|
|
receipt.update(completed_taps=0, execution="not_started")
|
|
for index in range(times):
|
|
if index:
|
|
await asyncio.sleep(max(0, delay_ms) / 1000)
|
|
elements = None
|
|
if elements is None:
|
|
elements = await self._live_ui_elements()
|
|
point = self._live_target_coordinates(resource_id, elements, expected_point)
|
|
coordinate_key = "pixel_coordinates" if expected_point is not None else "coordinates"
|
|
point = list(expected_point) if expected_point is not None else point
|
|
receipt[coordinate_key] = point
|
|
receipt["execution"] = "unknown"
|
|
result = await self.controller.tap_at(*point)
|
|
if error := getattr(result, "error", None):
|
|
return str(error)
|
|
receipt["completed_taps"] += 1
|
|
return None
|
|
|
|
async def _ensure_visible_target(self, step: dict, receipt: dict) -> None:
|
|
"""Prepare explicit UI once; poll only reads, never retry the opener."""
|
|
receipt.update(execution="not_started", completed_taps=0, visibility_attempts=0)
|
|
elements = await self._live_ui_elements()
|
|
guard = await self._visibility_guard(step["target"], elements)
|
|
receipt["visibility_guard"] = guard
|
|
receipt["visibility_attempts"] += 1
|
|
if guard["visible"]:
|
|
self._live_target_coordinates(step["target"], elements)
|
|
receipt["execution"] = "skipped"
|
|
return
|
|
error = await self._tap_native_target(step["opener"], receipt, elements=elements)
|
|
if error:
|
|
raise RuntimeError(error)
|
|
try:
|
|
async with asyncio.timeout(step["timeout_ms"] / 1000):
|
|
while True:
|
|
elements = await self._live_ui_elements()
|
|
guard = await self._visibility_guard(step["target"], elements)
|
|
receipt["visibility_guard"] = guard
|
|
receipt["visibility_attempts"] += 1
|
|
if guard["visible"]:
|
|
self._live_target_coordinates(step["target"], elements)
|
|
receipt["execution"] = "completed"
|
|
return
|
|
await asyncio.sleep(0.1)
|
|
except TimeoutError as exc:
|
|
raise TimeoutError(
|
|
f"Panel target {step['target']} did not become visible within {step['timeout_ms']}ms after one opener tap. No opener retry; re-observe before continuing."
|
|
) from exc
|
|
|
|
async def action_sequence(
|
|
self, steps: list[Any], feedback: dict[str, Any] | None = None
|
|
) -> ActionResult:
|
|
"""Execute once; optional visibility guards read live UI without model turns."""
|
|
from artemis.config.agent import load_agent_config
|
|
from artemis.mcp.gestures import plan_sequence, validate_sequence
|
|
|
|
try:
|
|
limit = load_agent_config().pro.execution.max_burst_actions
|
|
request = validate_sequence({"steps": steps, "feedback": feedback}, limit)
|
|
planned = plan_sequence(
|
|
request,
|
|
*self._dims(),
|
|
limit=limit,
|
|
)
|
|
except ValueError as exc:
|
|
return ActionResult.failure("action_sequence", str(exc), code=ActionCode.INVALID_ARGS)
|
|
from artemis.mcp.native_actions import NATIVE_ACTIONS
|
|
|
|
unavailable = {
|
|
step["action"]
|
|
for step in planned
|
|
if step["action"] in NATIVE_ACTIONS and step["action"] not in self.capabilities()
|
|
}
|
|
if unavailable:
|
|
return ActionResult.failure(
|
|
"action_sequence",
|
|
f"Unsupported members {sorted(unavailable)}; no input executed.",
|
|
code=ActionCode.UNSUPPORTED,
|
|
metadata={"execution": "not_started", "side_effect": "none"},
|
|
)
|
|
if request.feedback is not None and "read_control" not in self.capabilities():
|
|
return ActionResult.failure(
|
|
"action_sequence",
|
|
"Value feedback requires native read_control; no input executed.",
|
|
code=ActionCode.UNSUPPORTED,
|
|
metadata={"execution": "not_started", "side_effect": "none"},
|
|
)
|
|
return await self._value_feedback.run(
|
|
request.feedback,
|
|
planned,
|
|
"action_sequence",
|
|
self.controller.native_request,
|
|
lambda: self._execute_sequence(planned),
|
|
)
|
|
|
|
async def _execute_sequence(self, planned: list[dict]) -> ActionResult:
|
|
import time
|
|
|
|
from artemis.data_engine.trace import TraceSpan
|
|
from artemis.mcp.native_actions import NATIVE_ACTIONS
|
|
from artemis.runtime.task_control import TaskControl, has_pending_instruction
|
|
|
|
started = time.monotonic()
|
|
receipts = [
|
|
{**step, "index": i + 1, "execution": "not_started"} for i, step in enumerate(planned)
|
|
]
|
|
failed = None
|
|
failed_code = ActionCode.DEVICE_ERROR
|
|
for step, receipt in zip(planned, receipts, strict=True):
|
|
control = getattr(self.ctx, "task_control", None)
|
|
if isinstance(control, TaskControl) and (
|
|
has_pending_instruction(control.directory)
|
|
or (control.directory / "pause.request").exists()
|
|
):
|
|
failed = "Sequence interrupted before the next input; re-observe before continuing."
|
|
break
|
|
with TraceSpan(
|
|
name=f"action_sequence.{step['action']}", trace_type="tool", ctx=self.ctx
|
|
) as span:
|
|
span.payload = {"args": step}
|
|
receipt["started_ms"] = round((time.monotonic() - started) * 1000, 3)
|
|
receipt["execution"] = "unknown"
|
|
member_execution = "completed"
|
|
try:
|
|
if step["action"] == "ensure_visible":
|
|
try:
|
|
await self._ensure_visible_target(step, receipt)
|
|
member_execution = receipt["execution"]
|
|
error = None
|
|
except (LookupError, TimeoutError) as exc:
|
|
error = str(exc)
|
|
member_execution = receipt["execution"]
|
|
failed_code = (
|
|
ActionCode.TIMEOUT
|
|
if isinstance(exc, TimeoutError)
|
|
else ActionCode.TARGET_NOT_FOUND
|
|
)
|
|
elif step["action"] == "click":
|
|
guard = None
|
|
elements = None
|
|
error = None
|
|
if step.get("unless_visible") or step.get("target_resource_id"):
|
|
receipt["execution"] = "not_started"
|
|
elements = await self._live_ui_elements()
|
|
if resource_id := step.get("unless_visible"):
|
|
guard = await self._visibility_guard(resource_id, elements)
|
|
receipt["visibility_guard"] = guard
|
|
if guard and guard["visible"]:
|
|
member_execution = "skipped"
|
|
error = None
|
|
else:
|
|
if resource_id := step.get("target_resource_id"):
|
|
try:
|
|
error = await self._tap_native_target(
|
|
resource_id,
|
|
receipt,
|
|
times=step.get("times", 1),
|
|
delay_ms=step.get("delay_ms", 100),
|
|
elements=elements,
|
|
)
|
|
except LookupError as exc:
|
|
error = str(exc)
|
|
member_execution = receipt["execution"]
|
|
failed_code = ActionCode.TARGET_NOT_FOUND
|
|
else:
|
|
receipt["execution"] = "unknown"
|
|
repeats = (
|
|
{"times": step["times"], "delay_ms": step["delay_ms"]}
|
|
if "times" in step
|
|
else {}
|
|
)
|
|
result = await self.controller.tap_at(
|
|
*receipt["coordinates"], **repeats
|
|
)
|
|
error = getattr(result, "error", None)
|
|
elif step["action"] in NATIVE_ACTIONS:
|
|
result = await self._native_action(step["action"], step["args"])
|
|
error = None if result.ok else result.message
|
|
receipt.update(result.metadata)
|
|
receipt["native_execution"] = result.metadata.get("execution", "unknown")
|
|
member_execution = (
|
|
"skipped"
|
|
if result.ok and result.metadata.get("skipped")
|
|
else receipt["native_execution"]
|
|
)
|
|
if not result.ok:
|
|
failed_code = result.code
|
|
elif step["action"] == "drag":
|
|
result = await self._dispatch_drag(step)
|
|
error = None if result.ok else result.message
|
|
if not result.ok:
|
|
member_execution = result.metadata.get("execution", "unknown")
|
|
failed_code = result.code
|
|
if guard_result := result.metadata.get("visibility_guard"):
|
|
receipt["visibility_guard"] = guard_result
|
|
else:
|
|
await asyncio.sleep(step["time_in_ms"] / 1000)
|
|
error = None
|
|
receipt.update(
|
|
execution=(
|
|
member_execution
|
|
if member_execution
|
|
in {"skipped", "not_started", "cancelled", "observed"}
|
|
or receipt.get("native_execution") == "completed"
|
|
or not error
|
|
else "unknown"
|
|
),
|
|
error=error,
|
|
)
|
|
if error:
|
|
failed = str(error)
|
|
span.status = "failed"
|
|
span.error = failed
|
|
except asyncio.CancelledError:
|
|
receipt.update(execution="unknown", error="Input cancelled; effect uncertain.")
|
|
raise
|
|
except (OSError, RuntimeError) as exc:
|
|
receipt.update(error=str(exc))
|
|
failed = str(exc)
|
|
span.status = "failed"
|
|
span.error = failed
|
|
finally:
|
|
receipt["ended_ms"] = round((time.monotonic() - started) * 1000, 3)
|
|
span.result = dict(receipt)
|
|
span.payload["receipt"] = dict(receipt)
|
|
if failed:
|
|
break
|
|
completed = sum(r["execution"] == "completed" for r in receipts)
|
|
skipped = sum(r["execution"] == "skipped" for r in receipts)
|
|
attempted = any(
|
|
r["action"] not in {"wait", "wait_until"}
|
|
and r["execution"] in {"completed", "unknown", "cancelled"}
|
|
for r in receipts
|
|
)
|
|
execution = "completed" if not failed else ("partial" if completed else "unknown")
|
|
if failed and not attempted:
|
|
execution = "not_started"
|
|
metadata = {
|
|
"sequence_results": receipts,
|
|
"completed_steps": completed,
|
|
"skipped_steps": skipped,
|
|
"execution": execution,
|
|
"side_effect": "possible" if attempted else "none",
|
|
}
|
|
if failed:
|
|
return ActionResult.failure(
|
|
"action_sequence",
|
|
f"Sequence stopped after {completed}/{len(planned)} completed steps: {failed}"
|
|
" Remaining steps were not executed; do not replay completed inputs.",
|
|
code=failed_code,
|
|
metadata=metadata,
|
|
)
|
|
return ActionResult.success(
|
|
"action_sequence",
|
|
f"Executed {completed} gesture steps consecutively; skipped {skipped} already satisfied steps. Verify the intended result in the final screen.",
|
|
metadata=metadata,
|
|
)
|
|
|
|
async def click_sequence(
|
|
self, points: list[tuple[int, int]], delay_ms: int = 50
|
|
) -> ActionResult:
|
|
outcomes = []
|
|
for i, (nx, ny) in enumerate(points):
|
|
x, y = self._to_px(nx, ny)
|
|
result = await self.controller.tap_at(x, y)
|
|
err = getattr(result, "error", None) if result else None
|
|
if err:
|
|
return ActionResult.failure(
|
|
"click_sequence",
|
|
f"Error executing click at step {i + 1}: {err}",
|
|
detail=str(err),
|
|
)
|
|
outcomes.append(f"[{nx}, {ny}]")
|
|
if i < len(points) - 1:
|
|
await asyncio.sleep(max(0, delay_ms) / 1000.0)
|
|
|
|
return ActionResult.success(
|
|
"click_sequence",
|
|
f"Tapped in sequence at {'; '.join(outcomes)} (normalized).",
|
|
)
|
|
|
|
# --- Optional actions ------------------------------------------------------------
|
|
|
|
async def click(
|
|
self,
|
|
nx: int,
|
|
ny: int,
|
|
times: int = 1,
|
|
delay_ms: int = 100,
|
|
*,
|
|
expected_resource_id: str | None = None,
|
|
) -> ActionResult:
|
|
x, y = self._to_px(nx, ny)
|
|
if expected_resource_id:
|
|
receipt: dict = {}
|
|
try:
|
|
error = await self._tap_native_target(
|
|
expected_resource_id,
|
|
receipt,
|
|
times=times,
|
|
delay_ms=delay_ms,
|
|
expected_point=(x, y),
|
|
)
|
|
except (LookupError, OSError, RuntimeError) as exc:
|
|
return ActionResult.failure(
|
|
"click",
|
|
f"Observed target could not be confirmed for the next tap; completed {receipt.get('completed_taps', 0)} taps: {exc}. Re-observe. For an auto-hiding panel use one guarded action_sequence containing its opener and target, using observed resource IDs.",
|
|
code=ActionCode.TARGET_NOT_FOUND
|
|
if isinstance(exc, LookupError)
|
|
else ActionCode.DEVICE_ERROR,
|
|
metadata={
|
|
**receipt,
|
|
"side_effect": "none"
|
|
if receipt.get("execution") == "not_started"
|
|
else "possible",
|
|
"expected_resource_id": expected_resource_id,
|
|
},
|
|
)
|
|
if error:
|
|
return ActionResult.failure("click", error, metadata=receipt)
|
|
return ActionResult.success(
|
|
"click",
|
|
f"Tapped at [{nx}, {ny}] (normalized).",
|
|
normalized_coordinates=[int(nx), int(ny)],
|
|
metadata={**receipt, "execution": "completed"},
|
|
)
|
|
result = await self.controller.tap_at(x, y, times=times, delay_ms=delay_ms)
|
|
err = getattr(result, "error", None) if result else None
|
|
if err is not None:
|
|
return ActionResult.failure("click", f"Error executing click: {err}", detail=str(err))
|
|
return ActionResult.success(
|
|
"click",
|
|
f"Tapped at [{nx}, {ny}] (normalized).",
|
|
normalized_coordinates=[int(nx), int(ny)],
|
|
)
|
|
|
|
async def long_press(self, nx: int, ny: int, duration_ms: int = 1000) -> ActionResult:
|
|
x, y = self._to_px(nx, ny)
|
|
result = await self.controller.tap_at(
|
|
x, y, long_press=True, long_press_duration=duration_ms
|
|
)
|
|
err = getattr(result, "error", None) if result else None
|
|
if err is not None:
|
|
return ActionResult.failure(
|
|
"long_press", f"Error executing long press: {err}", detail=str(err)
|
|
)
|
|
return ActionResult.success(
|
|
"long_press",
|
|
f"Long-pressed at [{nx}, {ny}] (normalized) for {duration_ms}ms.",
|
|
normalized_coordinates=[int(nx), int(ny)],
|
|
duration_ms=duration_ms,
|
|
)
|
|
|
|
async def input_text(
|
|
self,
|
|
text: str,
|
|
target: tuple[int, int] | None = None,
|
|
clear_exist: bool = True,
|
|
) -> ActionResult:
|
|
coords = None
|
|
if target:
|
|
nx, ny = target
|
|
x, y = self._to_px(nx, ny)
|
|
coords = [int(nx), int(ny)]
|
|
err = await ensure_focus_at_coords(self.controller, x, y)
|
|
if err:
|
|
return ActionResult.failure(
|
|
"input_text",
|
|
f"Error focusing element: {err}",
|
|
detail=str(err),
|
|
normalized_coordinates=coords,
|
|
)
|
|
if clear_exist:
|
|
if hasattr(self.controller, "erase_text") and not await self.controller.erase_text():
|
|
return ActionResult.failure(
|
|
"input_text",
|
|
"Failed to clear existing text",
|
|
normalized_coordinates=coords,
|
|
)
|
|
elif hasattr(self.controller, "press_key"):
|
|
# KEYCODE_MOVE_END: move cursor to the end for reliable append.
|
|
await self.controller.press_key("123")
|
|
if hasattr(self.controller, "type_text") and not await self.controller.type_text(
|
|
text, clear_existing=False
|
|
):
|
|
return ActionResult.failure(
|
|
"input_text",
|
|
f"Failed typing '{text}'.",
|
|
normalized_coordinates=coords,
|
|
)
|
|
|
|
return ActionResult.success(
|
|
"input_text",
|
|
f"Typed '{text}'"
|
|
+ (f" at [{coords[0]}, {coords[1]}] (normalized)." if coords else "."),
|
|
normalized_coordinates=coords,
|
|
)
|
|
|
|
async def swipe(
|
|
self,
|
|
start: tuple[int, int],
|
|
end: tuple[int, int],
|
|
duration_ms: int = 800,
|
|
) -> ActionResult:
|
|
nx1, ny1 = start
|
|
nx2, ny2 = end
|
|
px1, py1 = self._to_px(nx1, ny1)
|
|
px2, py2 = self._to_px(nx2, ny2)
|
|
err = await self.controller.swipe_coords(px1, py1, px2, py2, duration_ms)
|
|
if err:
|
|
return ActionResult.failure("swipe", f"Error dragging: {err}", detail=str(err))
|
|
return ActionResult.success(
|
|
"swipe",
|
|
f"Swiped from [{nx1}, {ny1}] to [{nx2}, {ny2}] (normalized).",
|
|
normalized_coordinates=[int(nx1), int(ny1), int(nx2), int(ny2)],
|
|
duration_ms=duration_ms,
|
|
)
|
|
|
|
async def press_key(self, key: str) -> ActionResult:
|
|
key_str = str(key).lower()
|
|
if not key_str:
|
|
return ActionResult.failure(
|
|
"press_key",
|
|
f"Error executing key press '{key}'.",
|
|
code=ActionCode.INVALID_ARGS,
|
|
)
|
|
|
|
if key_str == "back" and hasattr(self.controller, "go_back"):
|
|
await self.controller.go_back()
|
|
elif key_str == "home" and hasattr(self.controller, "go_home"):
|
|
await self.controller.go_home()
|
|
elif key_str == "enter" and hasattr(self.controller, "press_enter"):
|
|
await self.controller.press_enter()
|
|
elif hasattr(self.controller, "press_key"):
|
|
# Pass unrecognized keycodes through verbatim (KEYCODE_* names, numeric
|
|
# codes): the driver resolves or forwards them, matching the historical
|
|
# adb_server behavior of accepting any Android key event.
|
|
res = await self.controller.press_key(key)
|
|
err = getattr(res, "error", None) if res else None
|
|
if err:
|
|
return ActionResult.failure(
|
|
"press_key",
|
|
f"Error executing key press '{key}': {err}",
|
|
detail=str(err),
|
|
)
|
|
if not res:
|
|
return ActionResult.failure("press_key", f"Error executing key press '{key}'.")
|
|
|
|
return ActionResult.success("press_key", f"Pressed key '{key}'.")
|
|
|
|
async def manage_app(self, action: str, app_name: str) -> ActionResult:
|
|
# Imported lazily: launch_app pulls in tool wrappers that are costly at import.
|
|
from artemis.tools.mobile.launch_app import find_package
|
|
from artemis.utils.app_launch_utils import launch_app_with_retries
|
|
|
|
res = find_package(self.ctx, app_name, use_fallback=False)
|
|
pkg = await res if inspect.iscoroutine(res) else res
|
|
if not pkg:
|
|
return ActionResult.failure(
|
|
"manage_app",
|
|
f"Error finding package for app: {app_name}",
|
|
code=ActionCode.PACKAGE_NOT_FOUND,
|
|
)
|
|
target_pkg = pkg
|
|
|
|
if action.lower() == "launch":
|
|
res_launch = launch_app_with_retries(self.ctx, target_pkg)
|
|
if inspect.iscoroutine(res_launch):
|
|
res_launch = await res_launch
|
|
if isinstance(res_launch, tuple):
|
|
success, error_msg = res_launch
|
|
else:
|
|
success, error_msg = bool(res_launch), ""
|
|
|
|
if success:
|
|
return ActionResult.success(
|
|
"manage_app",
|
|
f"Launched app '{app_name}' ({target_pkg}); foreground confirmed.",
|
|
)
|
|
return ActionResult.failure(
|
|
"manage_app",
|
|
f"Failed to launch app '{app_name}': {error_msg}",
|
|
detail=error_msg or None,
|
|
)
|
|
elif action.lower() == "stop":
|
|
if hasattr(self.controller, "terminate_app"):
|
|
res_term = self.controller.terminate_app(target_pkg)
|
|
if inspect.iscoroutine(res_term):
|
|
await res_term
|
|
return ActionResult.success(
|
|
"manage_app", f"Force-stopped app '{app_name}' ({target_pkg})."
|
|
)
|
|
return ActionResult.failure(
|
|
"manage_app",
|
|
f"Invalid manage_app action: {action}",
|
|
code=ActionCode.INVALID_ARGS,
|
|
)
|
|
|
|
async def wait_for_delay(self, time_in_ms: int) -> ActionResult:
|
|
delay_s = max(0.0, float(time_in_ms) / 1000.0)
|
|
await asyncio.sleep(delay_s)
|
|
return ActionResult.success(
|
|
"wait_for_delay",
|
|
f"Waited {time_in_ms}ms.",
|
|
duration_ms=int(time_in_ms),
|
|
)
|
|
|
|
async def wait_for_text(
|
|
self, text: str, wait_state: str | None = None, timeout_ms: int | None = None
|
|
) -> ActionResult:
|
|
timeout_s = (timeout_ms or 5000) / 1000.0
|
|
start = time.time()
|
|
found = False
|
|
target_state = (wait_state or "appear").lower()
|
|
|
|
while time.time() - start < timeout_s:
|
|
tree = await self.controller.get_ui_elements()
|
|
text_present = text.lower() in str(tree or "").lower()
|
|
if target_state == "appear" and text_present:
|
|
found = True
|
|
break
|
|
elif target_state == "disappear" and not text_present:
|
|
found = True
|
|
break
|
|
await asyncio.sleep(0.5)
|
|
|
|
if found:
|
|
return ActionResult.success(
|
|
"wait_for_text",
|
|
f"Text '{text}' {'appeared' if target_state == 'appear' else 'disappeared'}"
|
|
f" in the UI tree after {int((time.time() - start) * 1000)}ms.",
|
|
)
|
|
return ActionResult.failure(
|
|
"wait_for_text",
|
|
f"Timed out waiting for text '{text}' to {target_state}.",
|
|
code=ActionCode.TIMEOUT,
|
|
)
|
|
|
|
async def open_link(self, url: str) -> ActionResult:
|
|
success = await self.controller.open_url(url)
|
|
if success:
|
|
return ActionResult.success("open_link", f"Opened link '{url}'.")
|
|
return ActionResult.failure("open_link", f"Failed to open link '{url}'.")
|
|
|
|
async def erase_one_char(self) -> ActionResult:
|
|
success = await self.controller.erase_text(nb_chars=1)
|
|
if success:
|
|
return ActionResult.success("erase_one_char", "Erased one character.")
|
|
return ActionResult.failure("erase_one_char", "Failed to erase one character.")
|
|
|
|
async def focus_and_clear_text(self, nx: int, ny: int) -> ActionResult:
|
|
x, y = self._to_px(nx, ny)
|
|
err = await ensure_focus_at_coords(self.controller, x, y)
|
|
if err:
|
|
return ActionResult.failure(
|
|
"focus_and_clear_text",
|
|
f"Error focusing element: {err}",
|
|
detail=str(err),
|
|
normalized_coordinates=[int(nx), int(ny)],
|
|
)
|
|
success = await self.controller.erase_text()
|
|
if success:
|
|
return ActionResult.success(
|
|
"focus_and_clear_text",
|
|
f"Cleared text at [{nx}, {ny}] (normalized).",
|
|
normalized_coordinates=[int(nx), int(ny)],
|
|
)
|
|
return ActionResult.failure("focus_and_clear_text", "Failed to erase text.")
|
|
|
|
# --- Internal observation primitives ---------------------------------------------
|
|
|
|
async def take_screenshot(self) -> str:
|
|
return await self.controller.take_screenshot()
|
|
|
|
async def get_ui_elements(self) -> Any:
|
|
return await self.controller.get_ui_elements()
|
|
|
|
async def get_screen_data(self) -> Any:
|
|
return await self.controller.get_screen_data()
|