feat: add continuous multi-touch and long-press drag gestures

artemis/agents/operator/operator.py

@@ -1252,6 +1252,20 @@ def target_fields(el: dict) -> dict:
             )
             return actions, None
 
+        elif tool_name == "perform_gesture":
+            from artemis.mcp.gestures import validate_phases
+
+            try:
+                phases = validate_phases(args.get("phases"))
+                description = args.get("target_description")
+                if not isinstance(description, str) or not description.strip():
+                    raise ValueError("target_description is required for a gesture")
+            except (ValueError, TypeError) as exc:
+                return [], f"Invalid gesture: {exc}"
+            return [
+                {"action": "perform_gesture", "phases": phases, "target_description": description}
+            ], None
+
         elif tool_name == "swipe":
             kind, target, parsed_duration = parse_swipe_parameters(args, default_duration=None)
             duration = args.get("duration")

artemis/agents/operator/prompts.py

@@ -83,6 +83,7 @@ def load_operator_prompts() -> dict[str, str]:
 _TURN_ENDING_ORDER = (
     "click",
     "swipe",
+    "perform_gesture",
     "input_text",
     "long_press",
     "press_key",

artemis/agents/validator/execution_loop.py

@@ -205,7 +205,9 @@ async def _process_action(
         if burst:
             max_local_retries = 1
         else:
-            max_local_retries = 1 if action_name == "launch_app" else 2
+            # A timed-out gesture may already be dispatched; never replay it.
+            # Keep the existing retry policy for every other action.
+            max_local_retries = 1 if action_name in ("launch_app", "perform_gesture") else 2
         outcome.success, outcome.error_msg = await _attempt_local_execution(
             node,
             session,

artemis/agents/validator/tool_declarations.py

@@ -238,6 +238,8 @@ def normalize_coordinate_target(target: Any) -> int | list[int] | Any:
 
 SWIPE_TOOL = tool_declaration("swipe")
 
+PERFORM_GESTURE_TOOL = tool_declaration("perform_gesture")
+
 PRESS_KEY_TOOL = tool_declaration("press_key")
 
 READ_NOTE_TOOL = ToolDeclaration(
@@ -324,6 +326,7 @@ def normalize_coordinate_target(target: Any) -> int | list[int] | Any:
     LONG_PRESS_TOOL,
     INPUT_TEXT_TOOL,
     SWIPE_TOOL,
+    PERFORM_GESTURE_TOOL,
     PRESS_KEY_TOOL,
     READ_NOTE_TOOL,
     LIST_NOTES_TOOL,

artemis/clients/accessibility_client.py

@@ -413,6 +413,51 @@ def press_key(self, key: str) -> bool:
     def tap(self, x: float, y: float) -> bool:
         return self._rpc("tap", {"x": x, "y": y})
 
+    def perform_gesture(self, phases: list[dict], request_id: str) -> dict[str, Any]:
+        from artemis.mcp.gestures import gesture_duration_bound_ms, validate_phases
+
+        phases = validate_phases(phases)
+        info = json.loads(self._http("/ping").decode("utf-8"))
+        if "perform_gesture" not in info.get("capabilities", []):
+            return {
+                "success": False,
+                "status": "unsupported",
+                "error": "Upgrade Accessibility Helper: perform_gesture is not supported",
+            }
+        long_press_drag = phases[0].get("kind") == "long_press_drag"
+        if long_press_drag and "gesture_long_press_drag" not in info.get("capabilities", []):
+            return {
+                "success": False,
+                "status": "unsupported",
+                "error": "Upgrade Accessibility Helper: long-press dragging is not supported",
+            }
+        if any(
+            "control_points" in phase
+            or any("control_points" in p for p in phase.get("pointers", []))
+            for phase in phases
+        ) and ("gesture_cubic_bezier" not in info.get("capabilities", [])):
+            return {
+                "success": False,
+                "status": "unsupported",
+                "error": "Upgrade Accessibility Helper: native cubic Bezier is not supported",
+            }
+        if (len(phases) > 1 or long_press_drag) and not info.get("gesture_continuation", False):
+            return {
+                "success": False,
+                "status": "unsupported",
+                "error": "Continuous gestures require Android 8.0+",
+            }
+        timeout = gesture_duration_bound_ms(phases) / 1000 + 10
+        payload = {"cmd": "perform_gesture", "request_id": request_id, "phases": phases}
+        return json.loads(self._http("/action", payload, timeout=timeout).decode("utf-8"))
+
+    def cancel_gesture(self, request_id: str) -> dict[str, Any]:
+        return json.loads(
+            self._http("/action", {"cmd": "cancel_gesture", "request_id": request_id}).decode(
+                "utf-8"
+            )
+        )
+
     def swipe(self, x1: float, y1: float, x2: float, y2: float, duration_ms: int = 300) -> bool:
         return self._rpc("swipe", {"x1": x1, "y1": y1, "x2": x2, "y2": y2, "duration": duration_ms})
 

artemis/controllers/unified_controller.py

@@ -56,6 +56,16 @@ def __init__(self, ctx: ArtemisContext):
         super().__init__(ctx, get_driver(ctx))
         self._segment_cache: dict[tuple[str, int, float, float], VideoRecordingResult] = {}
 
+    async def perform_gesture(self, phases: list[dict]) -> dict:
+        method = getattr(self._driver, "perform_gesture", None)
+        if method is None:
+            return {
+                "success": False,
+                "status": "unsupported",
+                "error": "Driver has no continuous multi-touch support",
+            }
+        return await method(phases)
+
     @staticmethod
     async def _spawn_scrcpy(command: list[str]) -> asyncio.subprocess.Process:
         kwargs: dict[str, Any] = {

artemis/drivers/android/adb_driver.py

@@ -254,6 +254,30 @@ async def tap(
     async def long_press(self, x: int, y: int, duration_ms: int = 1000) -> bool:
         return await self.tap(x=x, y=y, duration_ms=duration_ms)
 
+    async def perform_gesture(self, phases: list[dict]) -> dict:
+        from artemis.clients.accessibility_client import AccessibilityClient
+        from artemis.mcp.gestures import validate_phases
+        from uuid import uuid4
+
+        phases = validate_phases(phases)
+        client = AccessibilityClient(self.device_id, provision_on_connect=False)
+        request_id = uuid4().hex
+        task = asyncio.create_task(asyncio.to_thread(client.perform_gesture, phases, request_id))
+        try:
+            return await asyncio.shield(task)
+        except asyncio.CancelledError:
+            # The action request is never replayed. Ask the device to finish the current
+            # phase and lift its held contacts, then wait for the terminal reply.
+            try:
+                await asyncio.shield(asyncio.to_thread(client.cancel_gesture, request_id))
+                await asyncio.wait_for(asyncio.shield(task), timeout=10.0)
+            except Exception:
+                logger.warning(
+                    "Gesture cancellation could not be confirmed; device needs reconciliation"
+                )
+                task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None)
+            raise
+
     async def swipe(
         self,
         start_x: int,

artemis/drivers/mock/mock_driver.py

@@ -106,6 +106,22 @@ async def long_press(self, x: int, y: int, duration_ms: int = 1000) -> bool:
         )
         return True
 
+    async def perform_gesture(self, phases: list[dict]) -> dict:
+        from artemis.mcp.gestures import validate_phases
+
+        phases = validate_phases(phases)
+        self.action_history.append({"action": "perform_gesture", "phases": phases})
+        return {
+            "success": True,
+            "status": "completed",
+            "release_confirmed": True,
+            "phases_completed": (
+                2 + int(phases[0].get("release_delay_ms", 0) > 0)
+                if phases[0].get("kind") == "long_press_drag"
+                else len(phases)
+            ),
+        }
+
     async def swipe(
         self,
         start_x: int,

artemis/mcp/action_executor.py

@@ -351,6 +351,20 @@ def _translate(
                 recorded,
             )
 
+        if raw_name == "perform_gesture":
+            from artemis.mcp.gestures import validate_phases
+
+            try:
+                phases = validate_phases(args.get("phases"))
+            except (ValueError, TypeError) as exc:
+                raise _ArgError(f"Invalid gesture: {exc}") from exc
+            return (
+                "perform_gesture",
+                {"phases": phases},
+                None,
+                self._require_description(args, raw_name),
+            )
+
         if raw_name == "swipe":
             return self._translate_swipe(args, state, index_elements=index_elements)
 

artemis/mcp/action_manifest.py

@@ -79,6 +79,7 @@
         "long_press",
         "input_text",
         "swipe",
+        "perform_gesture",
         "press_key",
         "manage_app",
         "wait_for_delay",

artemis/mcp/action_names.py

@@ -35,6 +35,7 @@
     "tap": "click",
     "long_press_on": "long_press",
     "swipe": "swipe",
+    "perform_gesture": "perform_gesture",
     "focus_and_input_text": "input_text",
     "focus_and_clear_text": "focus_and_clear_text",
     "erase_one_char": "erase_one_char",
@@ -106,6 +107,11 @@ def to_canonical_call(
             "duration_ms": action_item.get("duration", 1000),
         }
 
+    if verb == "perform_gesture":
+        from artemis.mcp.gestures import validate_phases
+
+        return "perform_gesture", {"phases": validate_phases(action_item.get("phases"))}
+
     if verb == "swipe":
         coords = action_item.get("normalized_coordinates")
         if not (isinstance(coords, (list, tuple)) and len(coords) == 4):

artemis/mcp/action_specs.py

@@ -64,7 +64,9 @@
 
 from langchain_core.tools import StructuredTool
 from mcp.types import CallToolResult
-from pydantic import Field, create_model
+from pydantic import BaseModel, Field, create_model
+
+from artemis.mcp.gestures import GestureInput, validate_phases
 
 from artemis.core.tool_declaration import ToolDeclaration
 from artemis.mcp.action_types import ActionResult
@@ -86,6 +88,21 @@
 ]
 
 
+GESTURE_DESCRIPTION = (
+    "[ACTION] Perform continuous touch gestures that a normal swipe cannot express: "
+    "simultaneous multi-finger input, continuous holds and drags, or delayed release. "
+    "Coordinates use 0-1000 screen space. In explicit phases, pointers move together; "
+    "a one-point path holds still. Keep the same pointer IDs and join endpoints across "
+    "phases to maintain contact until the final release. Optional control_points "
+    "define a cubic Bezier path with two endpoints and two controls. "
+    "Alternatively, provide one long_press_drag entry with start and end coordinates "
+    "and optional control_points. The endpoint is freely chosen, including inside the screen. "
+    "duration_ms is movement time; "
+    "release_delay_ms is the endpoint hold before lifting (zero releases immediately). "
+    "Observe the screen afterward to verify the intended effect."
+)
+
+
 # --- Spec structure ------------------------------------------------------------------
 
 
@@ -206,6 +223,10 @@ async def _wire_input_text(actuator: Any, a: dict[str, Any]) -> ActionResult:
     return await actuator.input_text(a["text"], norm, clear_exist=a["clear_exist"])
 
 
+async def _wire_perform_gesture(actuator: Any, a: dict[str, Any]) -> ActionResult:
+    return await actuator.perform_gesture(validate_phases(a["phases"]))
+
+
 async def _wire_swipe(actuator: Any, a: dict[str, Any]) -> ActionResult:
     start, end = a["start"], a["end"]
     return await actuator.swipe(
@@ -463,6 +484,27 @@ async def _wire_focus_and_clear_text(actuator: Any, a: dict[str, Any]) -> Action
             " the wire; required only for coordinate targets."
         ),
     ),
+    ActionSpec(
+        name="perform_gesture",
+        operator=OperatorDialect(
+            description=GESTURE_DESCRIPTION
+            + " target_description states the intended surface/objects and gesture purpose, not a claimed observation.",
+            params=(
+                ParamSpec(
+                    "phases",
+                    list[GestureInput],
+                    "Explicit continuous phases or a single native long_press_drag object.",
+                ),
+                ParamSpec("target_description", str, "Target surface/objects and gesture purpose."),
+            ),
+        ),
+        wire=WireDialect(
+            description=GESTURE_DESCRIPTION,
+            params=(ParamSpec("phases", list[GestureInput]),),
+            bind=_wire_perform_gesture,
+        ),
+        differences="target_description is recorded by the agent and is not sent to the helper.",
+    ),
     ActionSpec(
         name="swipe",
         operator=OperatorDialect(
@@ -686,6 +728,7 @@ async def _wire_focus_and_clear_text(actuator: Any, a: dict[str, Any]) -> Action
     "click",
     "input_text",
     "swipe",
+    "perform_gesture",
     "press_key",
     "manage_app",
     "wait_for_delay",
@@ -765,6 +808,20 @@ def _json_schema(annotation: Any) -> dict[str, Any]:
     ``required`` flag says so); they never appear in the projected schema, so
     the declaration carries no ``null`` alternatives or ``default`` noise.
     """
+    if isinstance(annotation, type) and issubclass(annotation, BaseModel):
+        schema = annotation.model_json_schema()
+        definitions = schema.get("$defs", {})
+
+        def inline(value):
+            if isinstance(value, list):
+                return [inline(v) for v in value]
+            if not isinstance(value, dict):
+                return value
+            if "$ref" in value:
+                return inline(definitions[value["$ref"].split("/")[-1]])
+            return {k: inline(v) for k, v in value.items() if k not in ("$defs", "title")}
+
+        return inline(schema)
     origin = get_origin(annotation)
     if origin in (Union, types.UnionType):
         members = [a for a in get_args(annotation) if a is not type(None)]

artemis/mcp/actuators/adb.py

@@ -186,6 +186,32 @@ async def long_press(self, nx: int, ny: int, duration_ms: int = 1000) -> ActionR
             duration_ms=duration_ms,
         )
 
+    async def perform_gesture(self, phases: list[dict]) -> ActionResult:
+        from artemis.mcp.gestures import validate_phases
+
+        phases = validate_phases(phases)
+        result = await self.controller.perform_gesture(phases)
+        if not (
+            result.get("success") is True
+            and result.get("status") == "completed"
+            and result.get("release_confirmed") is True
+        ):
+            return ActionResult.failure(
+                "perform_gesture",
+                f"Gesture {result.get('status', 'failed')}: {result.get('error', 'input not completed')}",
+                code={
+                    "unsupported": ActionCode.UNSUPPORTED,
+                    "invalid_arguments": ActionCode.INVALID_ARGS,
+                    "timeout": ActionCode.TIMEOUT,
+                }.get(result.get("status"), ActionCode.DEVICE_ERROR),
+                detail=str(result),
+            )
+        return ActionResult.success(
+            "perform_gesture",
+            "Gesture completed. Observe the screen to verify the intended effect.",
+            detail=str(result),
+        )
+
     async def input_text(
         self,
         text: str,

artemis/mcp/actuators/base.py

@@ -75,6 +75,8 @@ async def input_text(
         clear_exist: bool = True,
     ) -> ActionResult: ...
 
+    async def perform_gesture(self, phases: list[dict]) -> ActionResult: ...
+
     async def swipe(
         self,
         start: tuple[int, int],