Skip to content

fix(android): use clipboard paste and verify text entry in input_text #171

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
vishnoorath wants to merge 1 commit into
base: main
Choose a base branch
from
+181 −29
Open

fix(android): use clipboard paste and verify text entry in input_text #171

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
210 changes: 181 additions & 29 deletions artemis/drivers/android/adb_driver.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@

import asyncio
import base64
import re
from io import BytesIO
from pathlib import Path
from typing import Any, Literal
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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"<node\b[^>]*>", 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:
Expand All @@ -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
Expand Down
Loading