d6e84ff513
* fix(sessions): use latestConnState.client to avoid stale reference after reconnect Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01JKGMRpgihA4io2frodqLjt * fix(cua): lru_cache get_screen_size, remove redundant if-matches guard, drop inner import re Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01JKGMRpgihA4io2frodqLjt * fix(cua): use center-x comparator for send button, defer screen_w fetch Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01JKGMRpgihA4io2frodqLjt --------- Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>
750 lines
29 KiB
Python
Executable File
750 lines
29 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
Android Computer-Use Agent (CUA) smoke test for OpenCode Mobile.
|
|
|
|
Drives an Android emulator via ADB using an LLM vision loop:
|
|
screenshot → vision model → action → repeat
|
|
|
|
Inspired by:
|
|
- openai/openai-cua-sample-app (browser CUA pattern)
|
|
- X-PLUG/MobileAgent (ADB + VLM loop)
|
|
- TencentQQGYLab/AppAgent (multimodal smartphone agent)
|
|
|
|
Requirements:
|
|
pip install openai
|
|
ADB in PATH with a connected device/emulator.
|
|
|
|
Usage:
|
|
# OpenAI
|
|
export OPENAI_API_KEY=sk-...
|
|
python scripts/android-cua-smoke.py
|
|
|
|
# Azure OpenAI (with deployment)
|
|
export OPENAI_API_KEY=<key>
|
|
export OPENAI_BASE_URL=https://<resource>.openai.azure.com/openai/deployments/<deployment>/
|
|
python scripts/android-cua-smoke.py --model gpt-4o
|
|
|
|
# Google Gemini (via OpenAI-compat)
|
|
export GEMINI_API_KEY=AIza...
|
|
python scripts/android-cua-smoke.py
|
|
|
|
# xAI Grok
|
|
export XAI_API_KEY=xai-...
|
|
python scripts/android-cua-smoke.py
|
|
|
|
# Any OpenAI-compatible endpoint (LiteLLM, Ollama, etc.)
|
|
export OPENAI_API_KEY=dummy
|
|
export OPENAI_BASE_URL=http://localhost:4000/v1
|
|
python scripts/android-cua-smoke.py --model gpt-4o
|
|
|
|
# Custom goal
|
|
python scripts/android-cua-smoke.py --goal "Open settings and toggle dark mode"
|
|
"""
|
|
|
|
import argparse
|
|
import base64
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
import threading
|
|
import re
|
|
import xml.etree.ElementTree as ET
|
|
from functools import lru_cache
|
|
from pathlib import Path
|
|
|
|
try:
|
|
from openai import OpenAI, AzureOpenAI
|
|
except ImportError:
|
|
sys.exit("openai package required: pip install openai")
|
|
|
|
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ADB helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def adb(*args: str) -> str:
|
|
"""Run an adb command and return stdout."""
|
|
result = subprocess.run(
|
|
["adb", *args],
|
|
capture_output=True, text=True, timeout=30,
|
|
)
|
|
if result.returncode != 0 and "Error" in result.stderr:
|
|
raise RuntimeError(f"adb {' '.join(args)} failed: {result.stderr.strip()}")
|
|
return result.stdout.strip()
|
|
|
|
|
|
_step_counter = 0
|
|
APP_PACKAGE = "cc.agentlabs.opencode"
|
|
|
|
|
|
def _bounds_center(bounds: str) -> tuple[int, int] | None:
|
|
match = re.match(r"\[(\d+),(\d+)\]\[(\d+),(\d+)\]", bounds or "")
|
|
if not match:
|
|
return None
|
|
x1, y1, x2, y2 = map(int, match.groups())
|
|
return ((x1 + x2) // 2, (y1 + y2) // 2)
|
|
|
|
|
|
def current_foreground_package() -> str:
|
|
"""Return resumed foreground package name when available."""
|
|
out = adb("shell", "dumpsys", "activity", "activities")
|
|
for line in out.splitlines():
|
|
if "mResumedActivity" not in line:
|
|
continue
|
|
match = re.search(r"\s([a-zA-Z0-9_\.]+)/", line)
|
|
if match:
|
|
return match.group(1)
|
|
return ""
|
|
|
|
|
|
def ensure_app_foreground(package: str = APP_PACKAGE, retries: int = 3,
|
|
verbose: bool = True) -> bool:
|
|
"""Bring app to foreground before scenario start."""
|
|
for attempt in range(retries):
|
|
current = current_foreground_package()
|
|
if current == package:
|
|
return True
|
|
|
|
adb("shell", "monkey", "-p", package, "-c", "android.intent.category.LAUNCHER", "1")
|
|
time.sleep(2.0)
|
|
|
|
if verbose:
|
|
seen = current or "unknown"
|
|
print(f" [prep] foreground package was '{seen}', launched '{package}' (attempt {attempt + 1}/{retries})")
|
|
|
|
return current_foreground_package() == package
|
|
|
|
|
|
def maybe_dismiss_telemetry_consent(package: str = APP_PACKAGE,
|
|
verbose: bool = True) -> bool:
|
|
"""Dismiss first-launch telemetry consent modal when present."""
|
|
xml = ui_dump()
|
|
if not xml:
|
|
return False
|
|
|
|
try:
|
|
root = ET.fromstring(xml)
|
|
except ET.ParseError:
|
|
return False
|
|
|
|
consent_markers = (
|
|
"help improve opencode",
|
|
"share anonymous crash reports",
|
|
)
|
|
dismiss_markers = (
|
|
"not now", "no thanks", "decline", "skip", "later",
|
|
"don't allow", "dont allow", "deny", "continue without",
|
|
"opt out", "cancel",
|
|
)
|
|
|
|
page_text = " ".join(
|
|
" ".join(filter(None, [
|
|
node.attrib.get("text", ""),
|
|
node.attrib.get("content-desc", ""),
|
|
])).lower()
|
|
for node in root.iter()
|
|
)
|
|
|
|
if not any(marker in page_text for marker in consent_markers):
|
|
return False
|
|
|
|
candidates = []
|
|
for node in root.iter():
|
|
clickable = node.attrib.get("clickable") == "true"
|
|
if not clickable:
|
|
continue
|
|
|
|
label = " ".join(filter(None, [
|
|
node.attrib.get("text", ""),
|
|
node.attrib.get("content-desc", ""),
|
|
node.attrib.get("resource-id", ""),
|
|
])).strip().lower()
|
|
center = _bounds_center(node.attrib.get("bounds", ""))
|
|
if not center:
|
|
continue
|
|
candidates.append((label, center))
|
|
|
|
for label, (x, y) in candidates:
|
|
if any(marker in label for marker in dismiss_markers):
|
|
adb("shell", "input", "tap", str(x), str(y))
|
|
time.sleep(1.0)
|
|
if verbose:
|
|
print(f" [prep] dismissed telemetry consent via '{label or 'button'}' at ({x}, {y})")
|
|
return True
|
|
|
|
if verbose:
|
|
print(" [prep] telemetry consent detected but dismiss button was not found")
|
|
return False
|
|
|
|
|
|
def screenshot_b64() -> str:
|
|
"""Capture emulator screenshot and return as base64 PNG. Retries on timeout."""
|
|
global _step_counter
|
|
_step_counter += 1
|
|
for attempt in range(3):
|
|
try:
|
|
result = subprocess.run(
|
|
["adb", "exec-out", "screencap", "-p"],
|
|
capture_output=True, timeout=30,
|
|
)
|
|
if result.returncode == 0 and len(result.stdout) > 100:
|
|
# Save for debugging
|
|
debug_path = f"/tmp/cua_step_{_step_counter:03d}.png"
|
|
Path(debug_path).write_bytes(result.stdout)
|
|
return base64.b64encode(result.stdout).decode()
|
|
except subprocess.TimeoutExpired:
|
|
if attempt < 2:
|
|
time.sleep(3)
|
|
continue
|
|
raise
|
|
# Fallback: screencap on device then pull
|
|
with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as f:
|
|
path = f.name
|
|
try:
|
|
subprocess.run(["adb", "shell", "screencap", "-p", "/sdcard/_cua_screen.png"],
|
|
capture_output=True, timeout=30)
|
|
subprocess.run(["adb", "pull", "/sdcard/_cua_screen.png", path],
|
|
capture_output=True, timeout=10)
|
|
data = Path(path).read_bytes()
|
|
Path(f"/tmp/cua_step_{_step_counter:03d}.png").write_bytes(data)
|
|
return base64.b64encode(data).decode()
|
|
finally:
|
|
Path(path).unlink(missing_ok=True)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Screen recording
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def start_screen_recording(scenario_name: str) -> tuple:
|
|
"""Start ADB screen recording. Returns (thread, stop_event, remote_path)."""
|
|
remote_path = f"/sdcard/cua_{scenario_name}.mp4"
|
|
stop_event = threading.Event()
|
|
|
|
def _record():
|
|
# --time-limit 180 caps at 3 min; we stop it early via SIGTERM via adb
|
|
try:
|
|
subprocess.run(
|
|
["adb", "shell", f"screenrecord --time-limit 180 {remote_path}"],
|
|
capture_output=True, timeout=200,
|
|
)
|
|
except Exception:
|
|
pass
|
|
|
|
thread = threading.Thread(target=_record, daemon=True)
|
|
thread.start()
|
|
time.sleep(1.0) # let recorder spin up
|
|
return thread, stop_event, remote_path
|
|
|
|
|
|
def stop_screen_recording(thread: threading.Thread, remote_path: str,
|
|
local_path: str) -> bool:
|
|
"""Stop recorder, pull video to local_path. Returns True on success."""
|
|
# Stop screenrecord via pkill (SIGINT flushes MP4 moov atom)
|
|
subprocess.run(
|
|
["adb", "shell", "pkill", "-2", "screenrecord"],
|
|
capture_output=True, timeout=10,
|
|
)
|
|
time.sleep(2.0) # let MP4 finalise
|
|
thread.join(timeout=5)
|
|
|
|
result = subprocess.run(
|
|
["adb", "pull", remote_path, local_path],
|
|
capture_output=True, timeout=30,
|
|
)
|
|
if result.returncode == 0 and Path(local_path).exists():
|
|
print(f" [recording] saved to {local_path}")
|
|
return True
|
|
print(f" [recording] pull failed: {result.stderr.decode(errors='replace').strip()}")
|
|
return False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ArchiveBox upload
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def upload_to_archivebox(video_path: str, scenario_name: str) -> bool:
|
|
"""Upload video to ArchiveBox if ARCHIVEBOX_URL is configured.
|
|
|
|
Reads ARCHIVEBOX_URL and ARCHIVEBOX_API_KEY from environment.
|
|
Returns True if uploaded, False/skipped otherwise.
|
|
"""
|
|
url = os.environ.get("ARCHIVEBOX_URL", "").rstrip("/")
|
|
api_key = os.environ.get("ARCHIVEBOX_API_KEY", "")
|
|
if not url:
|
|
print(" [archivebox] ARCHIVEBOX_URL not set — skipping upload")
|
|
return False
|
|
|
|
try:
|
|
import urllib.request
|
|
import urllib.parse
|
|
|
|
video_data = Path(video_path).read_bytes()
|
|
boundary = "----CUAUploadBoundary"
|
|
body_parts = []
|
|
body_parts.append(f"--{boundary}\r\nContent-Disposition: form-data; name=\"url\"\r\n\r\nfile://{scenario_name}.mp4".encode())
|
|
body_parts.append(
|
|
f"--{boundary}\r\nContent-Disposition: form-data; name=\"file\"; filename=\"{scenario_name}.mp4\"\r\nContent-Type: video/mp4\r\n\r\n".encode()
|
|
+ video_data
|
|
)
|
|
body_parts.append(f"--{boundary}--\r\n".encode())
|
|
body = b"\r\n".join(body_parts)
|
|
|
|
headers = {
|
|
"Content-Type": f"multipart/form-data; boundary={boundary}",
|
|
"Content-Length": str(len(body)),
|
|
}
|
|
if api_key:
|
|
headers["X-API-Key"] = api_key
|
|
|
|
req = urllib.request.Request(f"{url}/api/v1/add", data=body, headers=headers, method="POST")
|
|
with urllib.request.urlopen(req, timeout=60) as resp:
|
|
resp_data = resp.read().decode(errors="replace")
|
|
print(f" [archivebox] uploaded {scenario_name}.mp4 → {url} ({resp.status})")
|
|
return True
|
|
except Exception as exc:
|
|
print(f" [archivebox] upload failed: {exc}")
|
|
return False
|
|
|
|
|
|
def ui_dump() -> str:
|
|
"""Dump UI hierarchy XML and return as string (optional context for LLM)."""
|
|
adb("shell", "uiautomator", "dump", "/sdcard/_cua_ui.xml")
|
|
result = subprocess.run(
|
|
["adb", "pull", "/sdcard/_cua_ui.xml", "/tmp/_cua_ui.xml"],
|
|
capture_output=True, timeout=10,
|
|
)
|
|
if result.returncode == 0:
|
|
return Path("/tmp/_cua_ui.xml").read_text(errors="replace")
|
|
return ""
|
|
|
|
|
|
def execute_action(action: dict) -> str:
|
|
"""Execute an action dict returned by the LLM. Returns status string."""
|
|
act = action.get("type", "")
|
|
|
|
if act == "tap":
|
|
x, y = int(action["x"]), int(action["y"])
|
|
adb("shell", "input", "tap", str(x), str(y))
|
|
return f"tapped ({x}, {y})"
|
|
|
|
elif act == "type":
|
|
text = action.get("text", "")
|
|
# ADB input text needs escaping
|
|
escaped = text.replace(" ", "%s").replace("&", "\\&").replace(";", "\\;")
|
|
adb("shell", "input", "text", escaped)
|
|
return f"typed '{text}'"
|
|
|
|
elif act == "key":
|
|
key = action.get("key", "")
|
|
key_map = {
|
|
"enter": "66", "back": "4", "home": "3",
|
|
"delete": "67", "tab": "61",
|
|
}
|
|
code = key_map.get(key.lower(), key)
|
|
adb("shell", "input", "keyevent", code)
|
|
return f"pressed key {key}"
|
|
|
|
elif act == "swipe":
|
|
x1, y1 = int(action["x1"]), int(action["y1"])
|
|
x2, y2 = int(action["x2"]), int(action["y2"])
|
|
duration = int(action.get("duration", 300))
|
|
adb("shell", "input", "swipe", str(x1), str(y1), str(x2), str(y2), str(duration))
|
|
return f"swiped ({x1},{y1})->({x2},{y2})"
|
|
|
|
elif act == "send":
|
|
# Auto-locate send button: rightmost clickable ViewGroup in the bottom input bar.
|
|
# Threshold is screen-relative (bottom 25%) so it works on any emulator
|
|
# resolution — API 30 default profile is 1080x1920, not the 2400-tall pixel
|
|
# we previously hardcoded against.
|
|
_, screen_h = get_screen_size()
|
|
bottom_threshold = int(screen_h * 0.75)
|
|
xml = ui_dump()
|
|
matches = re.findall(r'clickable="true"[^>]*bounds="\[(\d+),(\d+)\]\[(\d+),(\d+)\]"', xml)
|
|
bottom_buttons = [(int(x1), int(y1), int(x2), int(y2)) for x1, y1, x2, y2 in matches if int(y1) > bottom_threshold]
|
|
if bottom_buttons:
|
|
# Rightmost = highest center-x (not left-edge x1, which misidentifies wide buttons)
|
|
send_btn = max(bottom_buttons, key=lambda b: (b[0] + b[2]) // 2)
|
|
cx = (send_btn[0] + send_btn[2]) // 2
|
|
cy = (send_btn[1] + send_btn[3]) // 2
|
|
adb("shell", "input", "tap", str(cx), str(cy))
|
|
return f"send button tapped ({cx}, {cy})"
|
|
# Fallback: tap bottom-right corner of the screen, offset slightly inward
|
|
screen_w, _ = get_screen_size()
|
|
fx = screen_w - 80
|
|
fy = screen_h - 120
|
|
adb("shell", "input", "tap", str(fx), str(fy))
|
|
return f"send button tapped (fallback {fx}, {fy})"
|
|
|
|
elif act == "wait":
|
|
secs = float(action.get("seconds", 2))
|
|
time.sleep(secs)
|
|
return f"waited {secs}s"
|
|
|
|
elif act == "done":
|
|
return "DONE"
|
|
|
|
elif act == "fail":
|
|
return "FAIL: " + action.get("reason", "unknown")
|
|
|
|
else:
|
|
return f"unknown action: {act}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# LLM CUA loop
|
|
# ---------------------------------------------------------------------------
|
|
|
|
SYSTEM_PROMPT = """\
|
|
You are an Android phone automation agent. You control the device by issuing actions.
|
|
|
|
On each turn you receive a screenshot of the current screen.
|
|
Respond with a JSON object for ONE action to take next.
|
|
|
|
Available actions:
|
|
{"type": "tap", "x": <int>, "y": <int>}
|
|
{"type": "type", "text": "<string>"}
|
|
{"type": "key", "key": "enter|back|home|delete|tab"}
|
|
{"type": "swipe", "x1": <int>, "y1": <int>, "x2": <int>, "y2": <int>, "duration": <ms>}
|
|
{"type": "send"} -- tap the send button (auto-locates via UI hierarchy)
|
|
{"type": "wait", "seconds": <float>}
|
|
{"type": "done", "summary": "<what was accomplished>"}
|
|
{"type": "fail", "reason": "<why the goal cannot be achieved>"}
|
|
|
|
Rules:
|
|
- Issue exactly ONE action per turn as a JSON object. No markdown, no explanation outside JSON.
|
|
- Coordinates are in pixels relative to the screenshot dimensions.
|
|
- IMPORTANT: In this app, pressing "enter" inserts a newline, it does NOT send the message.
|
|
To send a message, use the {"type": "send"} action which auto-locates and taps the send button.
|
|
After typing, dismiss the keyboard by pressing "back", then use {"type": "send"}.
|
|
- When the goal is fully achieved, respond with {"type": "done", ...}.
|
|
- If stuck after several attempts, respond with {"type": "fail", ...}.
|
|
"""
|
|
|
|
|
|
def call_llm(client, model: str, system: str, history: list) -> str:
|
|
"""Call LLM via OpenAI-compatible API with retry on rate limit."""
|
|
for attempt in range(3):
|
|
try:
|
|
response = client.chat.completions.create(
|
|
model=model,
|
|
messages=[{"role": "system", "content": system}] + history,
|
|
max_completion_tokens=300,
|
|
temperature=0,
|
|
)
|
|
return response.choices[0].message.content.strip()
|
|
except Exception as e:
|
|
if "429" in str(e) and attempt < 2:
|
|
wait = 15 * (attempt + 1)
|
|
print(f" [rate limited, retrying in {wait}s...]")
|
|
time.sleep(wait)
|
|
continue
|
|
raise
|
|
|
|
|
|
def make_client(model: str):
|
|
"""Create OpenAI client. Supports AZURE_OPENAI_*, AZURE_DEV_AI_*, OPENAI_API_KEY, GEMINI_API_KEY, XAI_API_KEY."""
|
|
if os.environ.get("AZURE_OPENAI_API_KEY"):
|
|
azure_model = os.environ.get("AZURE_OPENAI_MODEL", "gpt-5.4")
|
|
return AzureOpenAI(
|
|
api_key=os.environ["AZURE_OPENAI_API_KEY"],
|
|
azure_endpoint=os.environ["AZURE_OPENAI_ENDPOINT"],
|
|
api_version=os.environ.get("AZURE_OPENAI_API_VERSION", "2024-08-01-preview"),
|
|
), azure_model
|
|
# Azure AI Foundry endpoint (cognitiveservices.azure.com/openai/v1) — use OpenAI client
|
|
if os.environ.get("AZURE_DEV_AI_API_KEY"):
|
|
base_url = os.environ.get("AZURE_DEV_AI_BASE_URL", "https://vibe-dev-ai.cognitiveservices.azure.com/openai/v1")
|
|
azure_model = os.environ.get("AZURE_DEV_AI_MODEL", "gpt-4o-2024-11-20")
|
|
return OpenAI(api_key=os.environ["AZURE_DEV_AI_API_KEY"], base_url=base_url), azure_model
|
|
if os.environ.get("OPENAI_API_KEY"):
|
|
base = os.environ.get("OPENAI_BASE_URL")
|
|
return OpenAI(base_url=base) if base else OpenAI(), model
|
|
if os.environ.get("XAI_API_KEY"):
|
|
return OpenAI(
|
|
api_key=os.environ["XAI_API_KEY"],
|
|
base_url="https://api.x.ai/v1",
|
|
), "grok-2-vision-1212"
|
|
if os.environ.get("GEMINI_API_KEY"):
|
|
return OpenAI(
|
|
api_key=os.environ["GEMINI_API_KEY"],
|
|
base_url="https://generativelanguage.googleapis.com/v1beta/openai/",
|
|
), "gemini-2.0-flash"
|
|
sys.exit("Set AZURE_OPENAI_API_KEY, AZURE_DEV_AI_API_KEY, OPENAI_API_KEY, XAI_API_KEY, or GEMINI_API_KEY")
|
|
|
|
|
|
@lru_cache(maxsize=1)
|
|
def get_screen_size() -> tuple[int, int]:
|
|
"""Return (width, height) of the connected device screen.
|
|
|
|
Cached after the first call — screen dimensions are stable for the
|
|
lifetime of a test run, and caching avoids a redundant ADB round-trip
|
|
on every `send` action.
|
|
"""
|
|
try:
|
|
out = adb("shell", "wm", "size")
|
|
# "Physical size: 1080x1920" or "Override size: 1080x1920"
|
|
for line in out.splitlines():
|
|
if "size:" in line.lower():
|
|
dims = line.split(":")[-1].strip()
|
|
w, h = dims.split("x")
|
|
return int(w), int(h)
|
|
except Exception:
|
|
pass
|
|
return 1080, 1920
|
|
|
|
|
|
def run_cua(goal: str, max_steps: int = 30, model: str = "gpt-4o",
|
|
include_ui_xml: bool = False, verbose: bool = True) -> dict:
|
|
"""Run the CUA loop until done/fail/max_steps."""
|
|
client, model = make_client(model)
|
|
history = []
|
|
screen_w, screen_h = get_screen_size()
|
|
|
|
for step in range(1, max_steps + 1):
|
|
# Capture screenshot
|
|
img_b64 = screenshot_b64()
|
|
|
|
# Build user message with screenshot
|
|
content = [
|
|
{"type": "text", "text": f"Step {step}. Screen is {screen_w}x{screen_h} pixels. Goal: {goal}\nWhat action should I take next?"},
|
|
{"type": "image_url", "image_url": {"url": f"data:image/png;base64,{img_b64}", "detail": "high"}},
|
|
]
|
|
|
|
# Optionally include UI XML for better element identification
|
|
if include_ui_xml:
|
|
xml = ui_dump()
|
|
if xml:
|
|
# Truncate to avoid token explosion
|
|
content.append({"type": "text", "text": f"UI hierarchy (truncated):\n{xml[:4000]}"})
|
|
|
|
history.append({"role": "user", "content": content})
|
|
|
|
# Call LLM
|
|
reply = call_llm(client, model, SYSTEM_PROMPT, history)
|
|
history.append({"role": "assistant", "content": reply})
|
|
|
|
# Parse action
|
|
try:
|
|
clean = reply.strip()
|
|
# Strip markdown code fences
|
|
if clean.startswith("```"):
|
|
clean = clean.split("\n", 1)[1].rsplit("```", 1)[0].strip()
|
|
# If model returned multiple JSON objects, take the first
|
|
m = re.search(r'\{[^{}]*\}', clean)
|
|
if m:
|
|
action = json.loads(m.group(0))
|
|
else:
|
|
action = json.loads(clean)
|
|
except json.JSONDecodeError:
|
|
if verbose:
|
|
print(f" [step {step}] Failed to parse: {reply[:100]}")
|
|
continue
|
|
|
|
# Execute
|
|
result = execute_action(action)
|
|
if verbose:
|
|
print(f" [step {step}] {action.get('type', '?')} -> {result}")
|
|
|
|
if result == "DONE":
|
|
return {"status": "success", "steps": step, "summary": action.get("summary", "")}
|
|
if result.startswith("FAIL"):
|
|
return {"status": "fail", "steps": step, "reason": action.get("reason", "")}
|
|
|
|
# Brief pause between actions for UI to settle
|
|
time.sleep(1.0)
|
|
|
|
# Keep history manageable: only last 6 turns (12 messages) + system
|
|
if len(history) > 12:
|
|
history = history[-12:]
|
|
|
|
return {"status": "timeout", "steps": max_steps}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Smoke test scenarios
|
|
# ---------------------------------------------------------------------------
|
|
|
|
SMOKE_SCENARIOS = [
|
|
{
|
|
"name": "send_message",
|
|
"goal": (
|
|
"You see the OpenCode mobile app. Tap the '+' button (top-right) to create a new session. "
|
|
"Tap the text input at the bottom. Type 'ping'. Press back to dismiss keyboard. "
|
|
"Use the send action. Wait 5 seconds, then take another screenshot. "
|
|
"If you don't yet see an assistant reply, wait another 10 seconds and re-check (assistant replies can take 15+ seconds). "
|
|
"If still no assistant bubble, wait another 15 seconds and re-check one more time. "
|
|
"Report success if you see both a 'You' bubble and an 'Assistant' bubble. "
|
|
"Report failure only after at least 30 seconds of total waiting with no assistant bubble."
|
|
),
|
|
},
|
|
{
|
|
"name": "multi_turn",
|
|
"goal": (
|
|
"You see the OpenCode mobile app. Tap '+' (top-right) to create a new session. "
|
|
"Tap the text input. Type 'what is 2+2'. Press back. Use send action. "
|
|
"Wait up to 30 seconds for an assistant reply (re-check every 10 seconds). "
|
|
"Then tap the text input again, type 'and 3+3?'. Press back. Use send action. "
|
|
"Wait up to 30 seconds for the second assistant reply (re-check every 10 seconds). "
|
|
"Report success if you see two assistant reply bubbles."
|
|
),
|
|
},
|
|
{
|
|
"name": "verify_session_list",
|
|
"goal": (
|
|
"You see the OpenCode mobile app. Tap the '+' button (top-right) to create a new session. "
|
|
"Wait 2 seconds for the session to be created. "
|
|
"Navigate back to the sessions list by tapping the bottom-left 'Sessions' tab or pressing the back button. "
|
|
"Wait 3 seconds for the session list to load. "
|
|
"Report success if you can see at least one session entry in the list. "
|
|
"Report failure if the sessions list appears empty or shows an error message."
|
|
),
|
|
},
|
|
]
|
|
|
|
# Extended scenarios requiring an external OpenCode server.
|
|
# Run with: python scripts/android-cua-smoke.py --opencode-url http://<host>:<port>
|
|
def _connect_and_verify_sessions_goal(url: str) -> str:
|
|
return (
|
|
f"You see the OpenCode mobile app. "
|
|
"Go to the Connections tab (bottom navigation bar). "
|
|
"If a connection to the server already exists, tap it to make it active and skip to the next step. "
|
|
"Otherwise tap '+' or 'Add Connection', "
|
|
f"enter the URL '{url}', leave username/password blank, tap Save or Connect. "
|
|
"Wait 3 seconds. "
|
|
"Now navigate to the Sessions tab (bottom navigation bar). "
|
|
"Wait 5 seconds for sessions to load. "
|
|
"If the sessions list is empty or shows 'No sessions yet', tap the '+' button "
|
|
"(top-right) to create a new session, wait 3 seconds, then navigate back to the "
|
|
"Sessions tab and wait 3 seconds for the list to refresh. "
|
|
"Report SUCCESS if you see at least one session listed (a session title is visible). "
|
|
"Report FAILURE if the sessions list is still empty, shows 'No sessions yet', or shows an error."
|
|
)
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="Android CUA smoke test")
|
|
parser.add_argument("--goal", help="Custom goal (overrides built-in scenarios)")
|
|
parser.add_argument("--model", default="gpt-4o", help="Vision model to use")
|
|
parser.add_argument("--max-steps", type=int, default=30)
|
|
parser.add_argument("--include-xml", action="store_true", help="Include UI XML in context")
|
|
parser.add_argument("--quiet", action="store_true")
|
|
parser.add_argument(
|
|
"--opencode-url",
|
|
help="OpenCode server URL (e.g. http://100.108.64.76:4096). "
|
|
"Used by the default connect-and-verify regression scenario.",
|
|
)
|
|
parser.add_argument(
|
|
"--skip-connect-scenario",
|
|
action="store_true",
|
|
help="Skip the default connect-and-verify regression scenario.",
|
|
)
|
|
parser.add_argument(
|
|
"--only-connect-scenario",
|
|
action="store_true",
|
|
help="Run ONLY the connect-and-verify-sessions scenario. Use in CI with a "
|
|
"local opencode server for a deterministic true-E2E (no model backend needed).",
|
|
)
|
|
parser.add_argument(
|
|
"--scenarios",
|
|
help="Comma-separated explicit scenario set to run, e.g. "
|
|
"'connect_and_verify_sessions,send_message,verify_session_list'. "
|
|
"Valid names: connect_and_verify_sessions, send_message, multi_turn, "
|
|
"verify_session_list. Overrides --only-connect-scenario and the default set.",
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
# Verify ADB
|
|
try:
|
|
devices = adb("devices")
|
|
if "device" not in devices.split("\n", 1)[-1]:
|
|
sys.exit("No ADB device connected. Start emulator first.")
|
|
except FileNotFoundError:
|
|
sys.exit("adb not found in PATH")
|
|
|
|
connect_url = args.opencode_url or os.environ.get("OPENCODE_URL") or "http://100.108.64.76:4096"
|
|
connect_scenario = {
|
|
"name": "connect_and_verify_sessions",
|
|
"goal": _connect_and_verify_sessions_goal(connect_url),
|
|
}
|
|
|
|
if args.scenarios:
|
|
# Explicit named set (CI widened gate). Look up by name across the full catalog.
|
|
catalog = {connect_scenario["name"]: connect_scenario}
|
|
for s in SMOKE_SCENARIOS:
|
|
catalog[s["name"]] = s
|
|
requested = [n.strip() for n in args.scenarios.split(",") if n.strip()]
|
|
unknown = [n for n in requested if n not in catalog]
|
|
if unknown:
|
|
sys.exit(f"Unknown scenario(s): {', '.join(unknown)}. "
|
|
f"Valid: {', '.join(catalog.keys())}")
|
|
scenarios = [catalog[n] for n in requested]
|
|
elif args.only_connect_scenario:
|
|
# CI true-E2E: just connect to the local opencode server and verify the list.
|
|
scenarios = [connect_scenario]
|
|
else:
|
|
scenarios = [{"name": "custom", "goal": args.goal}] if args.goal else list(SMOKE_SCENARIOS)
|
|
# Keep connect-and-verify in the default smoke path so regressions are exercised.
|
|
if not args.goal and not args.skip_connect_scenario:
|
|
scenarios.append(connect_scenario)
|
|
|
|
results = []
|
|
for scenario in scenarios:
|
|
if not args.quiet:
|
|
print(f"\n{'='*60}")
|
|
print(f"Scenario: {scenario['name']}")
|
|
print(f"Goal: {scenario['goal'][:80]}...")
|
|
print(f"{'='*60}")
|
|
|
|
# Start screen recording
|
|
rec_thread, _stop_ev, remote_path = start_screen_recording(scenario["name"])
|
|
local_video = f"/tmp/cua_{scenario['name']}.mp4"
|
|
|
|
try:
|
|
if not ensure_app_foreground(verbose=not args.quiet):
|
|
print(f" [prep] warning: could not confirm {APP_PACKAGE} in foreground")
|
|
maybe_dismiss_telemetry_consent(verbose=not args.quiet)
|
|
ensure_app_foreground(verbose=not args.quiet)
|
|
|
|
result = run_cua(
|
|
goal=scenario["goal"],
|
|
max_steps=args.max_steps,
|
|
model=args.model,
|
|
include_ui_xml=args.include_xml,
|
|
verbose=not args.quiet,
|
|
)
|
|
finally:
|
|
# Always stop and pull the recording
|
|
stop_screen_recording(rec_thread, remote_path, local_video)
|
|
upload_to_archivebox(local_video, scenario["name"])
|
|
|
|
result["scenario"] = scenario["name"]
|
|
result["video"] = local_video if Path(local_video).exists() else None
|
|
results.append(result)
|
|
|
|
if not args.quiet:
|
|
print(f"\nResult: {result['status']} in {result['steps']} steps")
|
|
if result.get("video"):
|
|
print(f"Video: {result['video']}")
|
|
if result.get("summary"):
|
|
print(f"Summary: {result['summary']}")
|
|
if result.get("reason"):
|
|
print(f"Reason: {result['reason']}")
|
|
|
|
# Exit code: 0 if all passed
|
|
failed = [r for r in results if r["status"] != "success"]
|
|
if failed:
|
|
print(f"\n{'!'*60}")
|
|
print(f"FAILED: {len(failed)}/{len(results)} scenarios")
|
|
sys.exit(1)
|
|
else:
|
|
print(f"\nAll {len(results)} scenarios passed.")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|