From 6462968df0092adb5d504be5927ba9b52b299194 Mon Sep 17 00:00:00 2001 From: Vishnoo Rath Date: Tue, 6 Oct 2026 17:36:54 +0530 Subject: [PATCH] fix(android): use clipboard paste and verify text entry in input_text input_text() always returned True, and its clipboard tier was gated on self._ui_adb_client, which is only populated in cloud mode (drivers/factory.py). On a normal local run that attribute is None, so the clipboard branch was skipped entirely and every entry fell through to `adb shell input text`. Because the function reported success unconditionally, an entry that did not land was indistinguishable from one that did. The agent would re-type, see an unchanged field, and retry indefinitely - observed on a Flutter app (com.nitara.farmer.uat) where `input keyevent 67` does nothing, so the field never cleared either. - Build a clipboard-capable client lazily via create_screen_client() so the clipboard tier is live locally. It routes through the Artemis Accessibility Helper, whose setClipboard can write the system clipboard at every API level (only reads are restricted since Android 10). - Prefer clipboard + KEYCODE_PASTE; keep `input text` as a fallback. - Clear through the helper's clear_text RPC when available instead of 21 synthesized backspaces, which Flutter ignores. - Verify the result by reading the focused editable node back, and return an honest success/failure so a silent failure surfaces instead of looping. Masked password fields and un-inspectable nodes stay optimistic. - Retry once after warming the backend, since the first RPC after a cold start can lose the race with the helper attach and spuriously return False. Verified on an Android 12 device: entering a phone number and a 4-box OTP code into the Flutter app now completes in a single pass instead of looping. --- artemis/drivers/android/adb_driver.py | 210 ++++++++++++++++++++++---- 1 file changed, 181 insertions(+), 29 deletions(-) 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