diff --git a/artemis/drivers/android/adb_driver.py b/artemis/drivers/android/adb_driver.py index 360aa4862..b69bdecf1 100644 --- a/artemis/drivers/android/adb_driver.py +++ b/artemis/drivers/android/adb_driver.py @@ -16,6 +16,7 @@ import asyncio import base64 +import re from io import BytesIO from pathlib import Path from typing import Any, Literal @@ -83,6 +84,11 @@ def __init__( self._device_id = device_id self._adb_client = adb_client self._ui_adb_client = ui_adb_client + # Lazily built clipboard-capable client. ``_ui_adb_client`` is only + # supplied by the MCP/cloud paths; on a plain local run it is None, and + # the clipboard tier in input_text() would silently be skipped so every + # entry fell through to `input text`. This is created on first use. + self._input_client: Any = None self._width = width self._height = height self._device: AdbDevice | None = None @@ -304,42 +310,187 @@ async def swipe_direction( ) return False - async def input_text(self, text: str, clear_existing: bool = True) -> bool: + async def _get_input_client(self) -> Any: + """Return a client that can set the clipboard / set text, or None. + + Prefers the injected UI client, then any client already resolved by the + helper/uiautomator factory. The factory client routes through the + Artemis Accessibility Helper, which can write the system clipboard on + every API level (reads are the restricted operation since Android 10). + """ + if self._ui_adb_client is not None: + return self._ui_adb_client + if self._input_client is not None: + return self._input_client try: - if clear_existing: - # Safe & robust clearing: Move to End -> Shift+Home selection -> Delete -> Fallback backspaces - clear_cmd = ( - "input keyevent 123 && " - "input keyevent --meta 1 122 && " - "input keyevent 67 && " - "input keyevent 67 67 67 67 67 67 67 67 67 67 67 67 67 67 67 67 67 67 67 67" - ) - await asyncio.to_thread(self.device.shell, clear_cmd) - else: - # Append mode: move cursor to the very end of existing text - await asyncio.to_thread( - self.device.shell, - "input keyevent 123", # KEYCODE_MOVE_END + from artemis.clients.screen_client_factory import create_screen_client + + self._input_client = await asyncio.to_thread( + create_screen_client, self._device_id + ) + except Exception as e: + logger.debug(f"Could not build a clipboard-capable client: {e}") + self._input_client = None + return self._input_client + + async def _clipboard_paste(self, text: str) -> bool: + """Set the clipboard and paste with KEYCODE_PASTE. True on confirmed paste.""" + client = await self._get_input_client() + if client is None: + return False + + # The first RPC after a cold start can lose the race with the helper's + # backend attach / forward setup and return False even though the + # helper is healthy. Retry once after a beat before giving up, otherwise + # a transient miss silently drops us onto the input-text path. + set_ok = False + for attempt in range(2): + try: + ok = False + if hasattr(client, "set_clipboard"): + res = client.set_clipboard(text) + if asyncio.iscoroutine(res): + res = await res + ok = bool(res) + elif hasattr(client, "_device") and getattr(client, "_device", None): + res = client._device.set_clipboard(text) + if asyncio.iscoroutine(res): + res = await res + ok = True + set_ok = ok + except Exception as e: + logger.debug(f"set_clipboard attempt {attempt + 1} failed: {e}") + set_ok = False + if set_ok: + break + if attempt == 0: + # Nudge the backend to resolve/attach before retrying. + try: + if hasattr(client, "get_hierarchy"): + res = client.get_hierarchy() + if asyncio.iscoroutine(res): + await res + except Exception: + pass + await asyncio.sleep(1.0) + + if not set_ok: + return False + + try: + await asyncio.to_thread(self.device.shell, "input keyevent 279") + except Exception as e: + logger.debug(f"KEYCODE_PASTE failed: {e}") + return False + # Give the IME a beat to drain the paste before we inspect the field. + await asyncio.sleep(0.35) + return await self._verify_focused_text(text) + + async def _verify_focused_text(self, expected: str) -> bool: + """Best-effort check that ``expected`` is present in the focused field. + + Returns True when the text is confirmed, and also when the field cannot + be inspected (masked password fields, custom WebViews, no editable node + in the dump) - we only fail when we can positively see that nothing + landed. That keeps masked fields working while still stopping the + silent-failure loop that made the agent retry the same entry forever. + """ + try: + client = await self._get_input_client() + if client is None or not hasattr(client, "get_hierarchy"): + return True + xml = client.get_hierarchy() + if asyncio.iscoroutine(xml): + xml = await xml + if not isinstance(xml, str) or not xml: + return True + + # Collect every focused editable node; React Native and Compose + # routinely emit several, and only one of them is the real field. + editable: list[str] = [] + for match in re.finditer(r"]*>", xml): + node = match.group(0) + if 'focused="true"' not in node: + continue + is_editable = ( + "EditText" in node + or 'class="android.widget.EditText"' in node + or 'editable="true"' in node ) + if not is_editable: + continue + text_attr = re.search(r'\stext="([^"]*)"', node) + editable.append(text_attr.group(1) if text_attr else "") + + if not editable: + # No focused editable node found -> cannot judge; assume success. + return True + + probe = expected.replace("\n", "").strip() + if not probe: + return True + + for current in editable: + if not current: + # An empty focused field proves nothing either way; keep + # looking rather than matching "" as a substring. + continue + # password variants render dots; a mismatch there is not evidence. + if set(current) <= {"•", "*", "●", "."}: + return True + if probe in current or current in probe: + return True + # Every focused editable field was empty while we expected content. + return False + except Exception as e: + logger.debug(f"Focused-text verification skipped: {e}") + return True + async def input_text(self, text: str, clear_existing: bool = True) -> bool: + try: # Normalize literal escaped newlines from LLM / tool call serialization norm_text = text.replace(r"\r\n", "\n").replace(r"\n", "\n").replace(r"\r", "\n") - # 1. Tier 1: Try clipboard injection + KEYCODE_PASTE (Zero IME interference, preserves multiline, works for all charsets) - if self._ui_adb_client: + client = await self._get_input_client() + + # 0. Clear through the helper when available - one RPC is more + # reliable than 21 synthesized backspaces on custom keyboards. + cleared = False + if client is not None and hasattr(client, "clear_text"): try: - set_clip_ok = False - if hasattr(self._ui_adb_client, "set_clipboard"): - set_clip_ok = self._ui_adb_client.set_clipboard(norm_text) - elif hasattr(self._ui_adb_client, "_device") and self._ui_adb_client._device: - self._ui_adb_client._device.set_clipboard(norm_text) - set_clip_ok = True - - if set_clip_ok: - await asyncio.to_thread(self.device.shell, "input keyevent 279") - return True + res = client.clear_text() + if asyncio.iscoroutine(res): + res = await res + cleared = bool(res) except Exception as e: - logger.debug(f"Clipboard paste fallback to ADB input: {e}") + logger.debug(f"Helper clear_text failed: {e}") + + if not cleared: + if clear_existing: + # Safe & robust clearing: Move to End -> Shift+Home selection -> Delete -> Fallback backspaces + clear_cmd = ( + "input keyevent 123 && " + "input keyevent --meta 1 122 && " + "input keyevent 67 && " + "input keyevent 67 67 67 67 67 67 67 67 67 67 67 67 67 67 67 67 67 67 67 67" + ) + await asyncio.to_thread(self.device.shell, clear_cmd) + else: + # Append mode: move cursor to the very end of existing text + await asyncio.to_thread( + self.device.shell, + "input keyevent 123", # KEYCODE_MOVE_END + ) + + # 1. Tier 1: clipboard injection + KEYCODE_PASTE. This is the + # IME-free path and is what apps with a custom keypad (e.g. a + # phone-number field that ignores synthesized input events) + # actually accept. + if await self._clipboard_paste(norm_text): + return True + logger.debug( + "Clipboard paste did not take; falling back to native input text." + ) # 2. Tier 2: Check if ADBKeyboard is currently active try: @@ -364,7 +515,8 @@ async def input_text(self, text: str, clear_existing: bool = True) -> bool: if line: escaped = _escape_for_adb_text(line) await asyncio.to_thread(self.device.shell, f"input text {escaped}") - return True + await asyncio.sleep(0.35) + return await self._verify_focused_text(norm_text) except Exception as e: logger.error(f"Input text failed for '{text}': {e}") return False