From c6fc5fd1698d8b79bc12e4dc3d9f9ae5d70e9259 Mon Sep 17 00:00:00 2001 From: Jack <72348727+Jack-GitHub12@users.noreply.github.com> Date: Tue, 26 May 2026 00:07:52 -0500 Subject: [PATCH] =?UTF-8?q?Smart=20macros=20=E2=80=94=20record=20once,=20r?= =?UTF-8?q?eplay=20many=20(literal=20+=20LLM-adaptive)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds the macro recording/replay system from goal/009: - mobile_use/record_replay.py: `_ui_fingerprint(helpers)` digest (top-N labels, focused element, app bundle id, ~200 bytes), `annotate(intent)` ctx mgr that attaches intent + fingerprint to journal entries, sidecar .jsonl alongside the .py script, `replay_smart(script, helpers, llm)` literal-first / LLM-fallback engine, `MacroStepFailed` with rich context for skip-vs-raise recovery. - mobile_use/agent_loop.py: `retarget_action(intent, recorded_fp, current_ui, llm)` reuses agent prompt scaffolding, mockable callable LLM. - mobile_use/macro.py + cli.py: `mobile-use macro {record,replay,list,show}` CLI. Flags: --smart/--literal (default literal), --intent for record, --dir (default ~/.mobile-use/macros/). - docs/macros.md: end-to-end walkthrough. - README + SKILL: smart-macro example + when-to-use vs agent loop guidance. Tests: 68 new (record_replay + smart_replay + macro_cli). Full suite 611 pass. --- README.md | 13 +- SKILL.md | 30 +++ docs/macros.md | 147 ++++++++++++++ mobile_use/agent_loop.py | 118 +++++++++++ mobile_use/cli.py | 5 + mobile_use/macro.py | 252 ++++++++++++++++++++++++ mobile_use/record_replay.py | 356 ++++++++++++++++++++++++++++++++- tests/test_macro_cli.py | 139 +++++++++++++ tests/test_record_replay.py | 230 ++++++++++++++++++++++ tests/test_smart_replay.py | 378 ++++++++++++++++++++++++++++++++++++ 10 files changed, 1663 insertions(+), 5 deletions(-) create mode 100644 docs/macros.md create mode 100644 mobile_use/macro.py create mode 100644 tests/test_macro_cli.py create mode 100644 tests/test_smart_replay.py diff --git a/README.md b/README.md index c34c014..a64d1c9 100644 --- a/README.md +++ b/README.md @@ -91,15 +91,26 @@ def run_script(): record_screen(duration=10) # save mp4 to /tmp (XCUITest + UIAutomator2) -# record/replay a tap sequence: +# record/replay a tap sequence (dumb — literal replay): from mobile_use import record_replay import iphone_harness.helpers as h record_replay.start_recording("flow.py", helpers=h) # ... your taps/swipes/typing ... record_replay.stop_recording() # writes runnable flow.py record_replay.replay("flow.py") # play it back + +# smart macro — annotate intent + LLM re-targets when the UI shifts: +with record_replay.recording("compose.py", helpers=h): + with record_replay.annotate("open compose screen"): + h.tap(h.find(label="Compose")) + with record_replay.annotate("type message body"): + h.type_text("hello") +# replay_smart re-finds buttons via your LLM when labels / layout move +record_replay.replay_smart("compose.py", helpers=h, llm=my_llm_callable) ``` +CLI equivalent — `mobile-use macro record ` opens a REPL with helpers + recording active; `mobile-use macro replay --smart` adapts steps when the UI shifts. See [docs/macros.md](docs/macros.md) for the full walkthrough. + ### Manual setup (skip if `mobile-use bootstrap` worked) ```bash diff --git a/SKILL.md b/SKILL.md index 94cd777..3f3b4c6 100644 --- a/SKILL.md +++ b/SKILL.md @@ -113,6 +113,36 @@ If you struggle with a generic mechanic, look in the skills directories: - **Permission dialogs:** block the app. Use `grant_permission()` or `alert_accept()`. - **Screen locked:** `unlock()` first. PIN/pattern → surface to user. +## Macros — record once, replay many + +Use macros for repetitive flows the agent runs verbatim (smoke tests, demos, +deterministic setup). Two tiers: + +- **Literal** — `record_replay.replay("flow.py", helpers=h)` re-executes the exact + recorded calls. Fastest, cheapest, brittle to any UI change. +- **Smart** — `record_replay.replay_smart("flow.py", helpers=h, llm=client)` checks + the UI fingerprint before each annotated block; if it diverged from record + time, asks the LLM to re-target the action. Costs one LLM call per shifted + step (zero when UI is unchanged). + +Record with intent so smart replay knows what each segment was for: + +```python +with record_replay.recording("compose.py", helpers=h): + with record_replay.annotate("open compose screen"): + h.tap(h.find(label="Compose")) + with record_replay.annotate("send message body"): + h.type_text("hi") + h.tap(h.find(label="Send")) +``` + +CLI: `mobile-use macro record ` opens a REPL with helpers + recording +active; `mobile-use macro replay --smart` runs intent-aware replay. + +**When to prefer macros over the agent loop:** flow is known, deterministic, and +will be re-run many times. **Stick with the agent loop** when you're exploring, +the screen state is unpredictable, or you need branching logic per response. + ## Domain skills (opt-in) When enabled (`IPH_DOMAIN_SKILLS=1` or `ANH_DOMAIN_SKILLS=1`), call `domain_skills(id)` after launching: diff --git a/docs/macros.md b/docs/macros.md new file mode 100644 index 0000000..7c46da0 --- /dev/null +++ b/docs/macros.md @@ -0,0 +1,147 @@ +# Macros — record / replay device action sequences + +mobile_use macros let you capture a sequence of helper calls and re-execute +them later, either **literally** (same calls, same args, byte-for-byte) or +**smart** (intent-aware, LLM re-targets steps when the UI shifts). + +## Two tiers + +| Tier | Method | Cost | Brittleness | When to use | +|-------------|--------------------------------|-------------|----------------------|------------------------------| +| Literal | `record_replay.replay(...)` | free | breaks on any UI change | smoke tests, demos, deterministic flows | +| Smart | `record_replay.replay_smart(...)` | 0–1 LLM call per shifted step | recovers from rename/relayout | flows that need to survive app updates | + +Both tiers share the same recording — annotate with intent at record time and +you get both options at replay time for free. + +## Quick start + +### 1. Record (Python API) + +```python +from mobile_use import record_replay +import iphone_harness.helpers as h + +with record_replay.recording("compose.py", helpers=h): + with record_replay.annotate("open compose screen"): + h.tap(h.find(label="Compose")) + with record_replay.annotate("type message body"): + h.type_text("hello world") + h.tap(h.find(label="Send")) +``` + +Result: `compose.py` (runnable script) + `compose.py.jsonl` (sidecar with +intent + UI fingerprint per step). + +### 2. Record (CLI) + +```bash +mobile-use macro record compose --intent "open compose screen" +# Drops you into a Python REPL with `h` (helpers) pre-imported. +# Make calls, then Ctrl+D to stop. The initial intent annotates the first block. +>>> h.tap(h.find(label="Compose")) +>>> h.type_text("hello") +>>> exit() +Saved → /Users/you/.mobile-use/macros/compose.py +``` + +### 3. Replay literal + +```bash +mobile-use macro replay compose # default: literal +# or +python3 -c 'from mobile_use import record_replay; \ + import iphone_harness.helpers as h; \ + record_replay.replay("compose.py", helpers=h)' +``` + +### 4. Replay smart + +```bash +export MU_MACRO_LLM="my_pkg.llm_client:complete" # callable(prompt)->str +mobile-use macro replay compose --smart +``` + +Or programmatically: + +```python +def my_llm(prompt: str) -> str: + response = anthropic_client.messages.create(...) + return response.content[0].text + +record_replay.replay_smart("compose.py", helpers=h, llm=my_llm) +``` + +## How smart replay decides + +For each entry in the journal: + +1. If the entry has no intent → run literal. +2. If the entry has intent + a recorded fingerprint: + - Capture the current UI fingerprint. + - If recorded fingerprint **matches** current (same app, ≥50% label overlap) + → run literal. + - If recorded fingerprint **diverged** + `llm` provided + → call `agent_loop.retarget_action(intent, recorded_fp, current_ui, ...)`, + execute the adapted call. + - If diverged + no `llm` → warn to stderr, run literal anyway. + +The fingerprint is captured **once** at each `annotate()` entry, not per call — +so the cost of recording is one `ui_tree()` per intent block, not per tap. + +## Storage layout + +Default macros directory: `~/.mobile-use/macros/`. Override with `--dir ` +or `MU_MACRO_DIR=`. + +``` +~/.mobile-use/macros/ + compose.py # runnable literal-replay script + compose.py.jsonl # sidecar with intent + fingerprint per step (only present if annotated) + smoke-test.py +``` + +## CLI reference + +``` +mobile-use macro record [--intent ] [--dir ] [--ios|--android] +mobile-use macro replay [--smart|--literal] [--on-failure raise|skip] [--dir ] [--ios|--android] +mobile-use macro list [--dir ] +mobile-use macro show [--dir ] +``` + +## Recovery: when smart replay can't re-target + +When `--smart` is on and the LLM returns `{"skip": true}` or unparseable output +for a step, the engine raises `record_replay.MacroStepFailed`. Override with +`--on-failure skip` to log the failure and continue with the next step. + +```python +try: + record_replay.replay_smart("flow.py", helpers=h, llm=my_llm, on_failure="raise") +except record_replay.MacroStepFailed as e: + print(f"step {e.step_index} ({e.recorded_fn}) intent={e.intent!r}: {e.reason}") +``` + +## Non-goals (v1) + +- **No cross-platform replay** — an iOS macro can't run on Android (different UI + tree shapes, different helper names). +- **No vision-only re-targeting** — smart replay reads accessibility labels + + text only; no screenshot OCR fallback yet. +- **No macro editor / GUI** — record + replay via CLI / Python API is the whole + surface. Edit the `.py` script directly if you need to. +- **No cloud / shared library** — macros live in your local + `~/.mobile-use/macros/` only. + +## Choosing between macros and the agent loop + +| Use macros when | Use the agent loop when | +|---------------------------------------------|---------------------------------------| +| Flow is known and deterministic | Exploring an unfamiliar app | +| You'll re-run it many times | Each run needs different decisions | +| You want offline reproducibility (no LLM) | Branching / conditional logic needed | +| You want a self-contained `.py` artifact | You want a continuous perceive-act loop| + +Smart macros sit in the middle: pre-recorded happy path + LLM repair when +the UI shifts. Cheap when nothing changed, robust when it did. diff --git a/mobile_use/agent_loop.py b/mobile_use/agent_loop.py index e95e552..447f205 100644 --- a/mobile_use/agent_loop.py +++ b/mobile_use/agent_loop.py @@ -12,6 +12,7 @@ - Auto skill authoring (writes .md files for discoveries) - Action history with rollback support """ +import json import os import sys import time @@ -230,6 +231,123 @@ def run_interactive(self): self.session.save() +_RETARGET_PROMPT = """You are a UI re-targeting assistant for replaying a recorded mobile-app action. + +Recorded intent: {intent!r} +At record time, the UI looked like: + app: {recorded_app} + visible labels (top {n_recorded}): {recorded_labels} + focused element: {recorded_focused} + +Originally invoked: {recorded_call} + +The UI is now different: + app: {current_app} + visible labels (top {n_current}): {current_labels} + focused element: {current_focused} + +Full current UI tree (compact, top 30): +{current_ui_json} + +Pick the SINGLE best adapted action that fulfills the intent on the current UI. +Reply with strict JSON, one of: + {{"fn": "", "args": [...], "kwargs": {{...}}}} + {{"skip": true, "reason": ""}} + +Rules: +- Prefer label/text-based selectors over xy coordinates when possible. +- Reuse the recorded helper-function name when the same op type still applies. +- Do NOT invent helper functions; stick to names that plausibly exist on the helpers module. +- Return ONLY the JSON object. No markdown, no commentary. +""" + + +def retarget_action(intent, recorded_fp, current_ui, recorded_call, llm, + current_app=None, current_focused=None): + """Ask an LLM to adapt a recorded action when the UI fingerprint has shifted. + + Args: + intent: human label for the recorded segment (from annotate()). + recorded_fp: fingerprint dict captured at record time + ({"app","labels","focused","count"}). + current_ui: list of element dicts from helpers.ui_tree(visible_only=True). + recorded_call: dict {fn, args, kwargs} of the literal recorded action. + llm: callable(prompt: str) -> str returning strict JSON. + Wrap your Anthropic / OpenAI / etc. client to fit. + current_app: optional current app bundle id; omitted from prompt if None. + current_focused: optional currently-focused element label. + + Returns: + dict with same shape as recorded_call (adapted), or + None if the LLM signals skip / returns unparseable output / raises. + """ + if not callable(llm): + raise TypeError("retarget_action: llm must be callable(prompt) -> str") + + rfp = recorded_fp or {} + current_labels = [] + seen = set() + for el in (current_ui or []): + if not isinstance(el, dict): + continue + lbl = (el.get("label") or el.get("name") + or el.get("text") or el.get("content_desc") or "") + if lbl and lbl not in seen: + seen.add(lbl) + current_labels.append(lbl) + current_labels = sorted(current_labels)[:20] + + try: + ui_json = json.dumps((current_ui or [])[:30], default=str) + except (TypeError, ValueError): + ui_json = "[]" + + prompt = _RETARGET_PROMPT.format( + intent=intent, + recorded_app=rfp.get("app", ""), + n_recorded=len(rfp.get("labels", []) or []), + recorded_labels=rfp.get("labels", []), + recorded_focused=rfp.get("focused"), + recorded_call=json.dumps(recorded_call, default=str), + current_app=current_app or "", + n_current=len(current_labels), + current_labels=current_labels, + current_focused=current_focused, + current_ui_json=ui_json, + ) + + try: + raw = llm(prompt) + except Exception: + return None + if not isinstance(raw, str): + return None + + raw = raw.strip() + if raw.startswith("```"): + raw = raw.strip("`") + if raw.startswith("json"): + raw = raw[4:] + raw = raw.strip() + + try: + parsed = json.loads(raw) + except json.JSONDecodeError: + return None + if not isinstance(parsed, dict): + return None + if parsed.get("skip") is True: + return None + if not isinstance(parsed.get("fn"), str): + return None + + return { + "fn": parsed["fn"], + "args": list(parsed.get("args") or []), + "kwargs": dict(parsed.get("kwargs") or {}), + } + + def run_agent(platform=None, args=None): """Entry point from CLI.""" if platform is None: diff --git a/mobile_use/cli.py b/mobile_use/cli.py index e832157..8db4275 100644 --- a/mobile_use/cli.py +++ b/mobile_use/cli.py @@ -365,6 +365,11 @@ def main(): sys.exit(ios_wda.build_main(remaining[2:])) sys.exit(f"Unknown `mobile-use ios` action: {remaining[1]!r}. Try `mobile-use ios --help`.") + # macro — record/replay action sequences (literal + smart) + if remaining and remaining[0] == "macro": + from . import macro + sys.exit(macro.main(remaining[1:], platform=platform)) + # Training data commands if remaining and remaining[0] == "export-training": from .collector import export_training_data diff --git a/mobile_use/macro.py b/mobile_use/macro.py new file mode 100644 index 0000000..bad9ad2 --- /dev/null +++ b/mobile_use/macro.py @@ -0,0 +1,252 @@ +"""mobile-use macro — record / replay device action sequences. + +Subcommands: + record [--intent ] Open Python REPL with helpers pre-imported + and recording active. Ctrl+D / exit() to stop. + `--intent` opens an initial annotate(...) block. + replay [--smart] [--literal] + Default: literal replay (re-runs the .py script). + `--smart` enables fingerprint-aware LLM re-targeting + (requires MU_MACRO_LLM env to point at a callable). + list Show saved macros with mtime + step count. + show Print the recorded .py script. + +Common flags: + --dir Macro storage directory. Default: ~/.mobile-use/macros/ + (override globally via MU_MACRO_DIR env). + --ios / --android Platform selector. Auto-detects when one device connected. + --on-failure raise|skip + For --smart replay: what to do when LLM can't re-target a step. + Default: raise (abort on first failure). +""" +import argparse +import datetime +import importlib +import os +import sys +from pathlib import Path + +from . import record_replay + + +def _default_dir() -> Path: + return Path(os.environ.get( + "MU_MACRO_DIR", str(Path.home() / ".mobile-use" / "macros") + )) + + +def _resolve_path(directory: Path, name: str) -> Path: + if not name.endswith(".py"): + name = name + ".py" + return directory / name + + +def _load_helpers(platform: str): + if platform == "ios": + import iphone_harness.helpers as h + return h + if platform == "android": + import android_harness.helpers as h + return h + raise ValueError(f"unknown platform: {platform!r}") + + +def _resolve_llm(): + """Look up MU_MACRO_LLM env: dotted import path 'module:callable' or 'module'. + + Returns the callable or None. Callable must accept(prompt: str) -> str. + """ + spec = os.environ.get("MU_MACRO_LLM") + if not spec: + return None + mod_name, _, attr = spec.partition(":") + if not mod_name: + return None + try: + mod = importlib.import_module(mod_name) + except ImportError: + return None + target = getattr(mod, attr, None) if attr else getattr(mod, "llm", None) + return target if callable(target) else None + + +def _count_steps(path: Path) -> int: + n = 0 + try: + for line in path.read_text().splitlines(): + if line.strip().startswith("h."): + n += 1 + except OSError: + return -1 + return n + + +def cmd_record(parsed, platform): + directory = Path(parsed.dir or str(_default_dir())) + directory.mkdir(parents=True, exist_ok=True) + out = _resolve_path(directory, parsed.name) + + helpers = _load_helpers(platform) + + print(f"Recording → {out}") + print("Helpers pre-imported as `h`. Make calls; Ctrl+D / exit() to stop.") + if parsed.intent: + print(f"Initial intent: {parsed.intent!r}") + + record_replay.start_recording(str(out), helpers=helpers) + initial_block = None + if parsed.intent: + initial_block = record_replay.annotate(parsed.intent) + initial_block.__enter__() + try: + ns = { + "h": helpers, "helpers": helpers, + "record_replay": record_replay, + "annotate": record_replay.annotate, + } + import code + code.interact(local=ns, banner="", exitmsg="") + finally: + if initial_block is not None: + try: + initial_block.__exit__(None, None, None) + except Exception: + pass + path = record_replay.stop_recording() + print(f"Saved → {path}") + return 0 + + +def cmd_replay(parsed, platform): + directory = Path(parsed.dir or str(_default_dir())) + path = _resolve_path(directory, parsed.name) + if not path.exists(): + print(f"Macro not found: {path}", file=sys.stderr) + return 2 + + helpers = _load_helpers(platform) + + if parsed.smart and parsed.literal: + print("Pass either --smart or --literal, not both.", file=sys.stderr) + return 2 + + if parsed.smart: + llm = _resolve_llm() + if llm is None: + print( + "[record_replay] --smart enabled but MU_MACRO_LLM is unset or " + "points to a non-callable — running literal-with-warnings.", + file=sys.stderr, + ) + results = record_replay.replay_smart( + str(path), helpers, llm=llm, on_failure=parsed.on_failure, + ) + for r in results: + extra = "" + if r.get("intent"): + extra = f" intent={r['intent']!r}" + print(f" step {r['step']:>3} {r['fn']:<20} → {r['outcome']}{extra}") + else: + record_replay.replay(str(path), helpers=helpers) + return 0 + + +def cmd_list(parsed): + directory = Path(parsed.dir or str(_default_dir())) + if not directory.exists(): + print(f"No macros directory at {directory}") + return 0 + macros = sorted(directory.glob("*.py")) + if not macros: + print(f"No macros in {directory}") + return 0 + print(f"Macros in {directory}:") + for p in macros: + mt = datetime.datetime.fromtimestamp(p.stat().st_mtime).strftime("%Y-%m-%d %H:%M") + steps = _count_steps(p) + sidecar = p.with_suffix(".py.jsonl") + annotated = " (annotated)" if sidecar.exists() else "" + print(f" {p.stem:30} {mt} {steps} steps{annotated}") + return 0 + + +def cmd_show(parsed): + directory = Path(parsed.dir or str(_default_dir())) + path = _resolve_path(directory, parsed.name) + if not path.exists(): + print(f"Macro not found: {path}", file=sys.stderr) + return 2 + print(path.read_text()) + return 0 + + +def _build_parser(): + p = argparse.ArgumentParser(prog="mobile-use macro", add_help=False) + p.add_argument("subcommand", nargs="?", default=None, + choices=[None, "record", "replay", "list", "show"]) + p.add_argument("name", nargs="?", default=None) + p.add_argument("--smart", action="store_true") + p.add_argument("--literal", action="store_true") + p.add_argument("--intent", default=None) + p.add_argument("--dir", default=None) + p.add_argument("--on-failure", dest="on_failure", + choices=["raise", "skip"], default="raise") + p.add_argument("-h", "--help", action="store_true") + return p + + +def main(args=None, platform=None): + args = list(args) if args is not None else sys.argv[1:] + parser = _build_parser() + + # argparse can't gracefully handle "no subcommand" with choices=[None, ...], + # so peek manually for help/empty. + if not args or args[0] in {"-h", "--help"}: + print(__doc__) + return 0 + + parsed = parser.parse_args(args) + + if parsed.help: + print(__doc__) + return 0 + + sub = parsed.subcommand + if sub is None: + print(__doc__) + return 0 + + if sub == "list": + return cmd_list(parsed) + if sub == "show": + if not parsed.name: + print("mobile-use macro show: requires ", file=sys.stderr) + return 2 + return cmd_show(parsed) + + # record / replay need a name + platform + if not parsed.name: + print(f"mobile-use macro {sub}: requires ", file=sys.stderr) + return 2 + + if platform is None: + from .cli import _detect_platform + platform = _detect_platform() + if platform is None: + print( + "Cannot detect platform. Pass --ios or --android.", + file=sys.stderr, + ) + return 2 + + if sub == "record": + return cmd_record(parsed, platform) + if sub == "replay": + return cmd_replay(parsed, platform) + + print(f"Unknown subcommand: {sub}", file=sys.stderr) + return 2 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/mobile_use/record_replay.py b/mobile_use/record_replay.py index 4141161..5f2e6c5 100644 --- a/mobile_use/record_replay.py +++ b/mobile_use/record_replay.py @@ -24,6 +24,7 @@ import functools import json +import sys import time from pathlib import Path from typing import Any @@ -47,9 +48,81 @@ "helpers_module": None, "originals": {}, "started_at": None, + "current_intent": None, + "current_fingerprint": None, } +# Max distinct labels kept in a fingerprint. Keeps the snapshot ~200 bytes. +_FP_TOP_N = 20 + + +def _ui_fingerprint(helpers) -> dict: + """Capture a lightweight snapshot of the current UI for replay-time diffing. + + Returns a dict with stable keys regardless of platform / helper availability: + { + "app": str current app bundle id ('' if unknown) + "labels": list[str] sorted unique visible labels (top _FP_TOP_N) + "focused": str|None label/text of focused element (None if unknown) + "count": int total visible elements (0 if tree unavailable) + } + + Designed to be safe to call even when helpers don't expose ui_tree/active_app + (e.g. mock helpers in tests, or a stripped-down replay env) — degrades to an + empty fingerprint instead of raising. + + Works cross-platform: reads `label`/`name` (iOS), `text`/`content_desc` + (Android). Focused detection uses both the boolean `focused` field (Android) + and the `traits` string containing 'Focused' (iOS). + """ + fp: dict[str, Any] = {"app": "", "labels": [], "focused": None, "count": 0} + + if hasattr(helpers, "active_app"): + try: + app = helpers.active_app() + if isinstance(app, str): + fp["app"] = app + except Exception: + pass + + if hasattr(helpers, "ui_tree"): + try: + tree = helpers.ui_tree(visible_only=True) + except TypeError: + try: + tree = helpers.ui_tree() + except Exception: + tree = None + except Exception: + tree = None + + if isinstance(tree, list) and tree: + seen: set[str] = set() + labels: list[str] = [] + for el in tree: + if not isinstance(el, dict): + continue + lbl = ( + el.get("label") + or el.get("name") + or el.get("text") + or el.get("content_desc") + or "" + ) + lbl = lbl.strip() if isinstance(lbl, str) else "" + if lbl and lbl not in seen: + seen.add(lbl) + labels.append(lbl) + if fp["focused"] is None: + if el.get("focused") is True or "Focused" in str(el.get("traits") or ""): + fp["focused"] = lbl or None + fp["labels"] = sorted(labels)[:_FP_TOP_N] + fp["count"] = len(tree) + + return fp + + def is_recording() -> bool: return _state["recording"] @@ -57,12 +130,17 @@ def is_recording() -> bool: def _make_recorder(name: str, original): @functools.wraps(original) def recorded(*args, **kwargs): - _state["journal"].append({ + entry = { "t": round(time.time() - _state["started_at"], 3), "fn": name, "args": list(args), "kwargs": dict(kwargs), - }) + } + if _state["current_intent"] is not None: + entry["intent"] = _state["current_intent"] + if _state["current_fingerprint"] is not None: + entry["fingerprint"] = _state["current_fingerprint"] + _state["journal"].append(entry) return original(*args, **kwargs) return recorded @@ -113,7 +191,13 @@ def start_recording(output_path: str, helpers, fn_names: tuple[str, ...] | None def stop_recording() -> str: - """End recording, restore helpers, write the script. Returns the path.""" + """End recording, restore helpers, write the script. Returns the path. + + Writes a sidecar ``.jsonl`` with structured journal entries + when any entry carries an intent (added via ``annotate()``). Smart-replay + reads the sidecar to recover intent + fingerprint metadata that won't fit + in the runnable .py script. + """ if not _state["recording"]: raise RuntimeError("record_replay: not recording.") @@ -121,16 +205,73 @@ def stop_recording() -> str: for name, original in _state["originals"].items(): setattr(helpers, name, original) - script = _generate_script(_state["journal"], helpers.__name__) + journal = _state["journal"] + script = _generate_script(journal, helpers.__name__) out = _state["output_path"] Path(out).write_text(script) + if any("intent" in e for e in journal): + sidecar = Path(out).with_suffix(Path(out).suffix + ".jsonl") + with sidecar.open("w") as f: + for entry in journal: + f.write(json.dumps(entry) + "\n") + _state["recording"] = False _state["journal"] = [] _state["originals"] = {} + _state["current_intent"] = None + _state["current_fingerprint"] = None return out +class annotate: + """Tag recorded calls inside this block with an intent + captured fingerprint. + + Only meaningful while record_replay is actively recording. Outside of + ``start_recording``/``stop_recording``, the annotate block is a no-op + (recording is the source of truth; annotate is metadata for whatever's + already being captured). + + Example:: + + with recording("flow.py", helpers=h): + with annotate("open compose screen"): + h.tap(find(label="Compose")) + with annotate("type message body"): + h.type_text("hi") + + Each annotate block snapshots the UI fingerprint at ``__enter__`` — that + fingerprint is the "expected state" smart-replay diffs against when + replaying the segment. Nested annotate blocks restore the outer intent + + fingerprint on inner exit. + """ + + def __init__(self, intent: str): + if not isinstance(intent, str) or not intent.strip(): + raise ValueError("annotate(intent=...) requires a non-empty string") + self.intent = intent + self._prev_intent = None + self._prev_fp = None + + def __enter__(self): + self._prev_intent = _state["current_intent"] + self._prev_fp = _state["current_fingerprint"] + _state["current_intent"] = self.intent + if _state["recording"] and _state["helpers_module"] is not None: + try: + _state["current_fingerprint"] = _ui_fingerprint(_state["helpers_module"]) + except Exception: + _state["current_fingerprint"] = None + else: + _state["current_fingerprint"] = None + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + _state["current_intent"] = self._prev_intent + _state["current_fingerprint"] = self._prev_fp + return False + + def _format_arg(v) -> str: if isinstance(v, str): return json.dumps(v) @@ -223,3 +364,210 @@ def replay(script_path: str, helpers=None): if helpers is not None: ns["h"] = helpers exec(compile(code, str(path), "exec"), ns) + + +# ---------- Smart replay ---------- + +# Fingerprint match threshold: fraction of recorded labels still present in +# current UI. Below this → mismatch → escalate to LLM (or warn if no LLM). +_FP_MATCH_THRESHOLD = 0.5 + + +class MacroStepFailed(Exception): + """Raised by replay_smart when a step cannot be replayed. + + Attributes set on the instance: + step_index int 0-based index of failing entry in the journal + intent str | None annotated intent (if any) of the failing step + recorded_fn str recorded helper function name + reason str human-readable failure category + fingerprint dict | None fingerprint captured at record time (if any) + """ + + def __init__(self, step_index, intent, recorded_fn, reason, fingerprint=None): + self.step_index = step_index + self.intent = intent + self.recorded_fn = recorded_fn + self.reason = reason + self.fingerprint = fingerprint + super().__init__( + f"macro step {step_index} ({recorded_fn!r}) failed: {reason}" + + (f" — intent: {intent!r}" if intent else "") + ) + + +def _fingerprint_matches(recorded_fp, current_fp) -> bool: + """Heuristic: same app + ≥_FP_MATCH_THRESHOLD label overlap. + + Empty recorded fingerprint → always matches (no expectation to compare). + """ + if not recorded_fp: + return True + recorded_labels = set(recorded_fp.get("labels") or []) + if not recorded_labels: + return True + if recorded_fp.get("app") and current_fp.get("app") \ + and recorded_fp["app"] != current_fp["app"]: + return False + current_labels = set(current_fp.get("labels") or []) + overlap = recorded_labels & current_labels + return (len(overlap) / len(recorded_labels)) >= _FP_MATCH_THRESHOLD + + +def _load_journal(script_path: str) -> list[dict]: + """Read the sidecar .jsonl if present, else return [].""" + sidecar = Path(script_path).with_suffix(Path(script_path).suffix + ".jsonl") + if not sidecar.exists(): + return [] + out = [] + for line in sidecar.read_text().splitlines(): + line = line.strip() + if not line: + continue + try: + out.append(json.loads(line)) + except json.JSONDecodeError: + continue + return out + + +def replay_smart(script_path, helpers, llm=None, on_failure="raise"): + """Replay a recorded macro with intent-aware re-targeting. + + Each entry in the sidecar journal carries (fn, args, kwargs) plus optional + intent + recorded fingerprint. For each entry: + + 1. Probe the current UI fingerprint via helpers. + 2. If the entry has no intent OR the recorded fingerprint still matches + the current UI → run the recorded call literally. + 3. If the fingerprint diverged AND an llm callable is provided → ask + retarget_action for an adapted call and execute that instead. + 4. If the fingerprint diverged but no llm was provided → log a warning + to stderr and fall back to the literal call (best-effort). + 5. Any execution failure routes through `on_failure`: + - "raise" (default) → raise MacroStepFailed + - "skip" → log + continue with next step + + When no sidecar exists, falls back to plain literal replay (the .py script). + + Args: + script_path: path to the .py file produced by stop_recording. The + sidecar `.jsonl` is read alongside. + helpers: helpers module (iphone_harness.helpers or + android_harness.helpers, or a compatible mock). + llm: optional callable(prompt) -> str. Required to actually + re-target on UI shifts; without it smart replay is + "literal with warnings". + on_failure: "raise" | "skip" + + Returns: + list of dicts, one per step: + {"step": int, "fn": str, "intent": str|None, + "outcome": "literal"|"retargeted"|"skipped"|"failed", + "result": , + "error": } + """ + if on_failure not in ("raise", "skip"): + raise ValueError("on_failure must be 'raise' or 'skip'") + + path = Path(script_path) + if not path.exists(): + raise FileNotFoundError(f"replay script not found: {script_path}") + + journal = _load_journal(script_path) + if not journal: + # No sidecar — degrade to dumb replay. + replay(script_path, helpers=helpers) + return [] + + from . import agent_loop # lazy import — avoids cycle if user only uses dumb replay + + results = [] + for i, entry in enumerate(journal): + fn_name = entry.get("fn") + args = list(entry.get("args") or []) + kwargs = dict(entry.get("kwargs") or {}) + intent = entry.get("intent") + recorded_fp = entry.get("fingerprint") + + outcome = "literal" + adapted = None + if intent and recorded_fp is not None: + try: + current_fp = _ui_fingerprint(helpers) + except Exception: + current_fp = {"app": "", "labels": [], "focused": None, "count": 0} + + if not _fingerprint_matches(recorded_fp, current_fp): + if llm is None: + sys.stderr.write( + f"[record_replay] step {i} ({fn_name}): UI fingerprint diverged " + f"but no llm provided — falling back to literal replay\n" + ) + else: + try: + current_ui = helpers.ui_tree(visible_only=True) \ + if hasattr(helpers, "ui_tree") else [] + except Exception: + current_ui = [] + adapted = agent_loop.retarget_action( + intent, recorded_fp, current_ui, + {"fn": fn_name, "args": args, "kwargs": kwargs}, + llm=llm, + current_app=current_fp.get("app"), + current_focused=current_fp.get("focused"), + ) + if adapted is None: + err = MacroStepFailed( + i, intent, fn_name, + "LLM declined to re-target or returned no adapted action", + fingerprint=recorded_fp, + ) + if on_failure == "raise": + raise err + results.append({ + "step": i, "fn": fn_name, "intent": intent, + "outcome": "skipped", "error": str(err), + }) + continue + fn_name = adapted["fn"] + args = adapted["args"] + kwargs = adapted["kwargs"] + outcome = "retargeted" + + fn = getattr(helpers, fn_name, None) + if fn is None or not callable(fn): + err = MacroStepFailed( + i, intent, entry.get("fn"), + f"helper {fn_name!r} not found on helpers module", + fingerprint=recorded_fp, + ) + if on_failure == "raise": + raise err + results.append({ + "step": i, "fn": fn_name, "intent": intent, + "outcome": "failed", "error": str(err), + }) + continue + + try: + result = fn(*args, **kwargs) + except Exception as e: + err = MacroStepFailed( + i, intent, fn_name, f"helper raised: {e}", + fingerprint=recorded_fp, + ) + if on_failure == "raise": + raise err + results.append({ + "step": i, "fn": fn_name, "intent": intent, + "outcome": "failed", "error": str(err), + }) + continue + + results.append({ + "step": i, "fn": fn_name, "intent": intent, + "outcome": outcome, "result": result, + }) + + return results diff --git a/tests/test_macro_cli.py b/tests/test_macro_cli.py new file mode 100644 index 0000000..13b7538 --- /dev/null +++ b/tests/test_macro_cli.py @@ -0,0 +1,139 @@ +"""Tests for the `mobile-use macro` CLI subcommand.""" +import json +import subprocess +import sys +from pathlib import Path + +import pytest + + +def _run(*args, env=None): + """Invoke the cli module as a subprocess. Returns (rc, stdout, stderr).""" + cmd = [sys.executable, "-m", "mobile_use.cli", "macro", *args] + real_env = None + if env: + import os + real_env = {**os.environ, **env} + proc = subprocess.run(cmd, capture_output=True, text=True, env=real_env) + return proc.returncode, proc.stdout, proc.stderr + + +# ---------- help / dispatch ---------- + +def test_macro_help_returns_zero(): + rc, out, _ = _run("--help") + assert rc == 0 + assert "record" in out and "replay" in out and "list" in out and "show" in out + + +def test_macro_no_subcommand_prints_help(): + rc, out, _ = _run() + assert rc == 0 + assert "Subcommands" in out + + +def test_macro_show_requires_name(tmp_path): + rc, _, err = _run("show", "--dir", str(tmp_path)) + assert rc == 2 + assert "requires " in err + + +def test_macro_show_missing_file(tmp_path): + rc, _, err = _run("show", "nope", "--dir", str(tmp_path)) + assert rc == 2 + assert "not found" in err.lower() + + +# ---------- list ---------- + +def test_macro_list_empty_dir_does_not_create(tmp_path): + empty = tmp_path / "nope" + rc, out, _ = _run("list", "--dir", str(empty)) + assert rc == 0 + assert "No macros directory" in out + assert not empty.exists() # list doesn't create + + +def test_macro_list_shows_saved_macros(tmp_path): + """Macros recorded into a dir should be listed with step counts.""" + # Synthesize a macro file directly (don't need a device) + out = tmp_path / "flow_a.py" + out.write_text( + '"""recorded"""\n' + 'import fake as h\n' + 'h.tap_at_xy(1, 2)\n' + 'h.type_text("hi")\n' + ) + out2 = tmp_path / "flow_b.py" + out2.write_text('"""x"""\nimport fake as h\nh.swipe(0, 0, 0, 100)\n') + + rc, std, _ = _run("list", "--dir", str(tmp_path)) + assert rc == 0 + assert "flow_a" in std + assert "flow_b" in std + assert "2 steps" in std + assert "1 steps" in std + + +def test_macro_list_annotated_marker(tmp_path): + """Macros with a .py.jsonl sidecar should show '(annotated)' marker.""" + p = tmp_path / "ann.py" + p.write_text("import fake as h\nh.tap_at_xy(1, 2)\n") + p.with_suffix(".py.jsonl").write_text(json.dumps({ + "t": 0.0, "fn": "tap_at_xy", "args": [1, 2], "kwargs": {}, + "intent": "test", + }) + "\n") + + rc, std, _ = _run("list", "--dir", str(tmp_path)) + assert rc == 0 + assert "(annotated)" in std + + +# ---------- show ---------- + +def test_macro_show_prints_script(tmp_path): + p = tmp_path / "x.py" + body = "import fake as h\nh.tap_at_xy(10, 20)\n" + p.write_text(body) + rc, std, _ = _run("show", "x", "--dir", str(tmp_path)) + assert rc == 0 + assert "tap_at_xy(10, 20)" in std + + +def test_macro_show_appends_py_extension(tmp_path): + """`macro show flow` should find flow.py.""" + (tmp_path / "flow.py").write_text("# ok\nimport fake as h\nh.tap_at_xy(1, 2)\n") + rc, std, _ = _run("show", "flow", "--dir", str(tmp_path)) + assert rc == 0 + assert "tap_at_xy" in std + + +# ---------- replay (literal, in-process) ---------- + +def test_replay_literal_missing_file(tmp_path): + rc, _, err = _run("replay", "missing", "--literal", "--dir", str(tmp_path), "--ios") + assert rc == 2 + assert "not found" in err.lower() + + +def test_replay_smart_and_literal_mutually_exclusive(tmp_path): + p = tmp_path / "x.py" + p.write_text("import fake as h\nh.tap_at_xy(1, 2)\n") + rc, _, err = _run("replay", "x", "--smart", "--literal", + "--dir", str(tmp_path), "--ios") + # Note: --ios pins platform so cmd_replay runs; the smart+literal check happens + # before any helpers import. + assert rc == 2 + assert "either --smart or --literal" in err.lower() + + +# ---------- dispatch via top-level cli ---------- + +def test_cli_dispatches_macro_subcommand(): + """`mobile-use macro --help` via the top-level cli should not fail.""" + proc = subprocess.run( + [sys.executable, "-m", "mobile_use.cli", "macro", "--help"], + capture_output=True, text=True, + ) + assert proc.returncode == 0 + assert "record" in proc.stdout diff --git a/tests/test_record_replay.py b/tests/test_record_replay.py index ffc5cba..7a311dc 100644 --- a/tests/test_record_replay.py +++ b/tests/test_record_replay.py @@ -1,4 +1,5 @@ """Tests for record_replay — captures helper calls and replays them.""" +import json import types from pathlib import Path @@ -170,3 +171,232 @@ def test_records_with_android_harness_helpers(): record_replay.start_recording(path, helpers=anh, fn_names=("tap_at_xy",)) record_replay.stop_recording() Path(path).unlink(missing_ok=True) + + +# ---------- UI fingerprint helper ---------- + +def _fake_helpers_with_ui(tree, app="com.example.app"): + mod = types.ModuleType("fake_with_ui") + mod.active_app = lambda: app + mod.ui_tree = lambda visible_only=False, compact=False: list(tree) + return mod + + +def test_fingerprint_empty_when_helpers_lack_ui(): + """Helpers without ui_tree/active_app yield empty fingerprint, no exception.""" + h = _make_fake_helpers() + fp = record_replay._ui_fingerprint(h) + assert fp == {"app": "", "labels": [], "focused": None, "count": 0} + + +def test_fingerprint_collects_visible_labels_ios_shape(): + tree = [ + {"type": "Button", "label": "Send", "name": "sendBtn", "traits": ""}, + {"type": "Button", "label": "Cancel", "name": "cancelBtn", "traits": ""}, + {"type": "TextField", "label": "", "name": "composeField", "traits": "Focused"}, + ] + h = _make_fake_helpers() + h.active_app = lambda: "com.apple.mobilesms" + h.ui_tree = lambda visible_only=False, compact=False: tree + fp = record_replay._ui_fingerprint(h) + assert fp["app"] == "com.apple.mobilesms" + assert fp["labels"] == ["Cancel", "Send", "composeField"] + assert fp["focused"] == "composeField" + assert fp["count"] == 3 + + +def test_fingerprint_collects_text_android_shape(): + tree = [ + {"type": "Button", "text": "Reply", "content_desc": "", "focused": False}, + {"type": "EditText", "text": "", "content_desc": "Message box", "focused": True}, + ] + h = _make_fake_helpers() + h.active_app = lambda: "com.android.messaging" + h.ui_tree = lambda visible_only=False, compact=False: tree + fp = record_replay._ui_fingerprint(h) + assert fp["app"] == "com.android.messaging" + assert "Reply" in fp["labels"] + assert "Message box" in fp["labels"] + assert fp["focused"] == "Message box" + + +def test_fingerprint_caps_labels_at_top_n(): + tree = [{"type": "X", "label": f"item-{i:03d}"} for i in range(50)] + h = _make_fake_helpers() + h.ui_tree = lambda visible_only=False, compact=False: tree + fp = record_replay._ui_fingerprint(h) + assert len(fp["labels"]) == record_replay._FP_TOP_N + # Sorted-stable cap: top of sorted set + assert fp["labels"] == sorted([f"item-{i:03d}" for i in range(50)])[: record_replay._FP_TOP_N] + + +def test_fingerprint_dedupes_repeated_labels(): + tree = [ + {"type": "Cell", "label": "Row"}, + {"type": "Cell", "label": "Row"}, + {"type": "Cell", "label": "Row"}, + {"type": "Cell", "label": "Other"}, + ] + h = _make_fake_helpers() + h.ui_tree = lambda visible_only=False, compact=False: tree + fp = record_replay._ui_fingerprint(h) + assert fp["labels"] == ["Other", "Row"] + assert fp["count"] == 4 # count counts all visible, not unique + + +def test_fingerprint_swallows_active_app_errors(): + h = _make_fake_helpers() + def boom(): + raise RuntimeError("WDA disconnected") + h.active_app = boom + h.ui_tree = lambda visible_only=False, compact=False: [] + fp = record_replay._ui_fingerprint(h) + assert fp["app"] == "" # gracefully empty + + +def test_fingerprint_swallows_ui_tree_errors(): + h = _make_fake_helpers() + h.active_app = lambda: "x" + def boom(visible_only=False, compact=False): + raise RuntimeError("appium gone") + h.ui_tree = boom + fp = record_replay._ui_fingerprint(h) + assert fp["labels"] == [] + assert fp["count"] == 0 + assert fp["app"] == "x" + + +def test_fingerprint_handles_ui_tree_without_kwargs(): + """Some mock helpers may not accept visible_only=. Fall back to no-arg call.""" + h = _make_fake_helpers() + h.ui_tree = lambda: [{"type": "X", "label": "Hello"}] + fp = record_replay._ui_fingerprint(h) + assert fp["labels"] == ["Hello"] + + +def test_fingerprint_size_is_small(): + """Realistic-size fingerprint should serialize to ≤ 1KB JSON (target ~200 bytes).""" + import json + tree = [{"type": "Button", "label": f"Btn {i}"} for i in range(record_replay._FP_TOP_N)] + h = _make_fake_helpers() + h.active_app = lambda: "com.test.app" + h.ui_tree = lambda visible_only=False, compact=False: tree + fp = record_replay._ui_fingerprint(h) + serialized = json.dumps(fp) + assert len(serialized) < 1024, f"fingerprint too large: {len(serialized)} bytes" + + +# ---------- annotate / intent ---------- + +def test_annotate_exists(): + assert hasattr(record_replay, "annotate") + + +def test_annotate_rejects_empty_intent(): + with pytest.raises(ValueError): + record_replay.annotate("") + with pytest.raises(ValueError): + record_replay.annotate(" ") + with pytest.raises(ValueError): + record_replay.annotate(None) # type: ignore + + +def test_annotate_outside_recording_is_noop(): + """annotate() block outside start_recording does not raise and does not journal.""" + with record_replay.annotate("noop intent"): + pass # no recording active → just sets/restores thread state + + +def test_annotate_tags_recorded_calls(tmp_path): + """Calls inside annotate block carry intent in the journal.""" + h = _make_fake_helpers() + h.active_app = lambda: "com.test.app" + h.ui_tree = lambda visible_only=False, compact=False: [ + {"type": "Btn", "label": "Compose"}, + ] + out = tmp_path / "rec.py" + record_replay.start_recording(str(out), helpers=h) + try: + with record_replay.annotate("open compose"): + h.tap_at_xy(10, 20) + h.type_text("hi") + # Outside the block → no intent tagged + h.tap_at_xy(30, 40) + finally: + record_replay.stop_recording() + + sidecar = out.with_suffix(".py.jsonl") + assert sidecar.exists(), "sidecar .jsonl should be written when any entry has intent" + lines = [json.loads(l) for l in sidecar.read_text().splitlines() if l.strip()] + assert len(lines) == 3 + assert lines[0]["intent"] == "open compose" + assert lines[1]["intent"] == "open compose" + assert "intent" not in lines[2] + # Fingerprint captured at annotate __enter__ + assert "fingerprint" in lines[0] + assert lines[0]["fingerprint"]["app"] == "com.test.app" + assert "Compose" in lines[0]["fingerprint"]["labels"] + + +def test_annotate_no_sidecar_when_no_intents(tmp_path): + """No annotate used → no .jsonl sidecar written (backward compat).""" + h = _make_fake_helpers() + out = tmp_path / "rec.py" + record_replay.start_recording(str(out), helpers=h) + h.tap_at_xy(1, 2) + record_replay.stop_recording() + assert not out.with_suffix(".py.jsonl").exists() + + +def test_annotate_restores_prev_intent_on_exit(tmp_path): + """After annotate block exits, recording state returns to prior intent.""" + h = _make_fake_helpers() + out = tmp_path / "rec.py" + record_replay.start_recording(str(out), helpers=h) + try: + assert record_replay._state["current_intent"] is None + with record_replay.annotate("outer"): + assert record_replay._state["current_intent"] == "outer" + with record_replay.annotate("inner"): + assert record_replay._state["current_intent"] == "inner" + assert record_replay._state["current_intent"] == "outer" + assert record_replay._state["current_intent"] is None + finally: + record_replay.stop_recording() + + +def test_annotate_clears_state_on_exception(tmp_path): + """If body raises inside annotate, state still restored.""" + h = _make_fake_helpers() + out = tmp_path / "rec.py" + record_replay.start_recording(str(out), helpers=h) + try: + with pytest.raises(RuntimeError): + with record_replay.annotate("crashy"): + raise RuntimeError("boom") + assert record_replay._state["current_intent"] is None + finally: + record_replay.stop_recording() + + +def test_annotate_fingerprint_captured_at_entry(tmp_path): + """Fingerprint is snapshotted once at __enter__, not re-fetched per call.""" + h = _make_fake_helpers() + tree_state = [[{"type": "B", "label": "First"}]] # mutable, so we can mutate between calls + h.active_app = lambda: "app" + h.ui_tree = lambda visible_only=False, compact=False: tree_state[0] + out = tmp_path / "rec.py" + record_replay.start_recording(str(out), helpers=h) + try: + with record_replay.annotate("step"): + h.tap_at_xy(1, 2) + # Change UI mid-block — the journal entries should still all reference + # the fingerprint captured at __enter__. + tree_state[0] = [{"type": "B", "label": "AfterChange"}] + h.tap_at_xy(3, 4) + finally: + record_replay.stop_recording() + + lines = [json.loads(l) for l in out.with_suffix(".py.jsonl").read_text().splitlines() if l.strip()] + assert lines[0]["fingerprint"]["labels"] == ["First"] + assert lines[1]["fingerprint"]["labels"] == ["First"] # not "AfterChange" diff --git a/tests/test_smart_replay.py b/tests/test_smart_replay.py new file mode 100644 index 0000000..920909f --- /dev/null +++ b/tests/test_smart_replay.py @@ -0,0 +1,378 @@ +"""Tests for smart replay — LLM re-targets recorded actions when UI shifts. + +Covers: + D3: agent_loop.retarget_action (this file) + D4: record_replay.replay_smart (added later, kept in same file) + D6: MacroStepFailed (added later) +""" +import json +import types +from pathlib import Path + +import pytest + +from mobile_use import record_replay +from mobile_use import agent_loop + + +# ---------- helpers ---------- + +def _mk_llm(reply): + """Return a callable(prompt) -> str that ignores prompt and returns reply. + + Pass a string for canned reply, or a list for one-per-call replies. + """ + if isinstance(reply, list): + i = {"n": 0} + def call(prompt): + r = reply[i["n"]] + i["n"] += 1 + return r + return call + return lambda prompt: reply + + +def _fake_helpers(): + mod = types.ModuleType("fake_smart_helpers") + calls = [] + def tap_at_xy(x, y): + calls.append(("tap_at_xy", (x, y), {})) + def tap(el): + calls.append(("tap", (el,), {})) + def type_text(text): + calls.append(("type_text", (text,), {})) + def find(label=None, text=None): + if label or text: + return {"label": label or text, "cx": 100, "cy": 200} + return None + mod.tap_at_xy = tap_at_xy + mod.tap = tap + mod.type_text = type_text + mod.find = find + mod._calls = calls + return mod + + +# ---------- D3: retarget_action ---------- + +def test_retarget_action_exists(): + assert callable(agent_loop.retarget_action) + + +def test_retarget_requires_callable_llm(): + with pytest.raises(TypeError): + agent_loop.retarget_action("intent", {}, [], {"fn": "tap"}, llm="not-callable") + + +def test_retarget_returns_adapted_action_from_llm_json(): + llm = _mk_llm('{"fn": "tap", "args": [], "kwargs": {"el": "ComposeButton"}}') + recorded_fp = {"app": "x", "labels": ["Old"], "focused": None, "count": 1} + current_ui = [{"type": "Button", "label": "Compose"}] + recorded_call = {"fn": "tap_at_xy", "args": [10, 20], "kwargs": {}} + + out = agent_loop.retarget_action( + "open compose", recorded_fp, current_ui, recorded_call, llm=llm + ) + assert out == {"fn": "tap", "args": [], "kwargs": {"el": "ComposeButton"}} + + +def test_retarget_returns_none_on_skip_signal(): + llm = _mk_llm('{"skip": true, "reason": "no matching element"}') + out = agent_loop.retarget_action("x", {}, [], {"fn": "tap"}, llm=llm) + assert out is None + + +def test_retarget_returns_none_on_invalid_json(): + llm = _mk_llm("not valid json at all") + out = agent_loop.retarget_action("x", {}, [], {"fn": "tap"}, llm=llm) + assert out is None + + +def test_retarget_strips_markdown_fences(): + llm = _mk_llm('```json\n{"fn": "tap", "args": [1, 2]}\n```') + out = agent_loop.retarget_action("x", {}, [], {"fn": "tap"}, llm=llm) + assert out == {"fn": "tap", "args": [1, 2], "kwargs": {}} + + +def test_retarget_returns_none_on_llm_exception(): + def boom(prompt): + raise RuntimeError("rate limited") + out = agent_loop.retarget_action("x", {}, [], {"fn": "tap"}, llm=boom) + assert out is None + + +def test_retarget_returns_none_when_fn_missing_from_response(): + llm = _mk_llm('{"args": [1, 2], "kwargs": {}}') + out = agent_loop.retarget_action("x", {}, [], {"fn": "tap"}, llm=llm) + assert out is None + + +def test_retarget_returns_none_on_non_dict_response(): + llm = _mk_llm('["unexpected", "list"]') + out = agent_loop.retarget_action("x", {}, [], {"fn": "tap"}, llm=llm) + assert out is None + + +def test_retarget_normalizes_missing_args_kwargs(): + llm = _mk_llm('{"fn": "press_home"}') + out = agent_loop.retarget_action("x", {}, [], {"fn": "press_home"}, llm=llm) + assert out == {"fn": "press_home", "args": [], "kwargs": {}} + + +def test_retarget_prompt_includes_intent_and_fingerprint(): + captured = {} + def llm(prompt): + captured["prompt"] = prompt + return '{"fn": "tap"}' + + fp = {"app": "com.test", "labels": ["Old", "Btn"], "focused": "Old", "count": 2} + agent_loop.retarget_action( + "tap the new thing", + fp, + [{"type": "Button", "label": "New"}], + {"fn": "tap_at_xy", "args": [10, 20]}, + llm=llm, + current_app="com.test", + current_focused="New", + ) + + assert "tap the new thing" in captured["prompt"] + assert "com.test" in captured["prompt"] + assert "Old" in captured["prompt"] + assert "New" in captured["prompt"] + assert "tap_at_xy" in captured["prompt"] + + +def test_retarget_handles_non_string_llm_response(): + llm = lambda p: 12345 # ints, not strings + out = agent_loop.retarget_action("x", {}, [], {"fn": "tap"}, llm=llm) + assert out is None + + +def test_retarget_handles_empty_recorded_fp(): + llm = _mk_llm('{"fn": "tap"}') + out = agent_loop.retarget_action("x", None, None, {"fn": "tap"}, llm=llm) + assert out == {"fn": "tap", "args": [], "kwargs": {}} + + +# ---------- D4: replay_smart engine ---------- + +def _record_smart_macro(tmp_path, intent_blocks): + """Build a recorded macro with annotated blocks. Returns script_path. + + intent_blocks: list of (intent_str, ui_tree, calls_fn) tuples. + intent_str : str — annotate intent label, or None for unannotated + ui_tree : list[dict] — what helpers.ui_tree should return at record time + calls_fn : callable(helpers) → makes helper calls under the annotate block + """ + h = _fake_helpers() + # active_app fixed throughout to keep test stable + h.active_app = lambda: "com.test.smart" + out = tmp_path / "macro.py" + record_replay.start_recording(str(out), helpers=h) + try: + for intent, ui_tree, calls_fn in intent_blocks: + h.ui_tree = lambda visible_only=False, compact=False, _t=ui_tree: _t + if intent is None: + calls_fn(h) + else: + with record_replay.annotate(intent): + calls_fn(h) + finally: + record_replay.stop_recording() + return str(out) + + +def test_replay_smart_exists(): + assert callable(record_replay.replay_smart) + + +def test_replay_smart_rejects_unknown_on_failure(tmp_path): + h = _fake_helpers() + out = tmp_path / "x.py" + out.write_text("") + with pytest.raises(ValueError): + record_replay.replay_smart(str(out), h, on_failure="explode") + + +def test_replay_smart_missing_script_raises(tmp_path): + with pytest.raises(FileNotFoundError): + record_replay.replay_smart(str(tmp_path / "missing.py"), _fake_helpers()) + + +def test_replay_smart_no_sidecar_falls_back_to_literal(tmp_path): + """When no .jsonl sidecar exists, smart replay degrades to dumb replay.""" + h = _fake_helpers() + out = tmp_path / "rec.py" + h.__name__ = "fake_smart_helpers" + record_replay.start_recording(str(out), helpers=h) + h.tap_at_xy(5, 6) + record_replay.stop_recording() + assert not out.with_suffix(".py.jsonl").exists() # confirm no sidecar + + fresh = _fake_helpers() + fresh.__name__ = "fake_smart_helpers" + import sys as _sys + _sys.modules["fake_smart_helpers"] = fresh + try: + results = record_replay.replay_smart(str(out), fresh) + assert results == [] # signal: dumb-replay fallback + assert fresh._calls == [("tap_at_xy", (5, 6), {})] + finally: + del _sys.modules["fake_smart_helpers"] + + +def test_replay_smart_literal_when_fingerprint_matches(tmp_path): + """Annotated step where current UI matches recorded → run literal call.""" + tree = [{"type": "Button", "label": "Compose"}] + out = _record_smart_macro(tmp_path, [ + ("open compose", tree, lambda h: h.tap_at_xy(10, 20)), + ]) + + fresh = _fake_helpers() + fresh.active_app = lambda: "com.test.smart" + fresh.ui_tree = lambda visible_only=False, compact=False: tree + # No llm provided — should be fine because fingerprint matches + results = record_replay.replay_smart(out, fresh, llm=None) + assert len(results) == 1 + assert results[0]["outcome"] == "literal" + assert fresh._calls == [("tap_at_xy", (10, 20), {})] + + +def test_replay_smart_warns_when_mismatch_without_llm(tmp_path, capsys): + """Fingerprint diverged + no llm → warn to stderr, run literal fallback.""" + out = _record_smart_macro(tmp_path, [ + ("open compose", [{"type": "Button", "label": "Compose"}], + lambda h: h.tap_at_xy(10, 20)), + ]) + + fresh = _fake_helpers() + fresh.active_app = lambda: "com.test.smart" + fresh.ui_tree = lambda visible_only=False, compact=False: [ + {"type": "Button", "label": "Completely Different"}, + ] + results = record_replay.replay_smart(out, fresh, llm=None) + captured = capsys.readouterr() + assert "diverged" in captured.err + assert len(results) == 1 + assert results[0]["outcome"] == "literal" + + +def test_replay_smart_retargets_via_llm_on_mismatch(tmp_path): + """Fingerprint diverged + llm → calls retarget_action, executes adapted call.""" + out = _record_smart_macro(tmp_path, [ + ("tap compose", [{"type": "Button", "label": "Compose"}], + lambda h: h.tap_at_xy(50, 60)), + ]) + + fresh = _fake_helpers() + fresh.active_app = lambda: "com.test.smart" + fresh.ui_tree = lambda visible_only=False, compact=False: [ + {"type": "Button", "label": "NewCompose"}, + ] + llm = _mk_llm('{"fn": "tap_at_xy", "args": [99, 88]}') + + results = record_replay.replay_smart(out, fresh, llm=llm) + assert results[0]["outcome"] == "retargeted" + assert fresh._calls == [("tap_at_xy", (99, 88), {})] + + +def test_replay_smart_unannotated_steps_run_literal(tmp_path): + """Unannotated + annotated mixed: unannotated runs literal regardless of UI.""" + tree = [{"type": "Button", "label": "Compose"}] + out = _record_smart_macro(tmp_path, [ + # mix one annotated step (forces sidecar) with one unannotated step + ("compose intent", tree, lambda h: h.tap_at_xy(10, 20)), + (None, tree, lambda h: h.tap_at_xy(99, 88)), + ]) + # Sanity: sidecar exists because at least one step is annotated + assert Path(out).with_suffix(".py.jsonl").exists() + + fresh = _fake_helpers() + fresh.active_app = lambda: "com.test.smart" + fresh.ui_tree = lambda visible_only=False, compact=False: tree # match + results = record_replay.replay_smart(out, fresh, llm=None) + assert len(results) == 2 + assert all(r["outcome"] == "literal" for r in results) + assert fresh._calls == [("tap_at_xy", (10, 20), {}), ("tap_at_xy", (99, 88), {})] + + +# ---------- D6: MacroStepFailed + recovery ---------- + +def test_macro_step_failed_exception_class_exists(): + assert issubclass(record_replay.MacroStepFailed, Exception) + + +def test_macro_step_failed_attributes_set(): + err = record_replay.MacroStepFailed( + step_index=3, intent="x", recorded_fn="tap", + reason="testing", fingerprint={"app": "y"}, + ) + assert err.step_index == 3 + assert err.intent == "x" + assert err.recorded_fn == "tap" + assert err.reason == "testing" + assert err.fingerprint == {"app": "y"} + assert "3" in str(err) + assert "tap" in str(err) + assert "testing" in str(err) + + +def test_recovery_raises_when_llm_declines_and_on_failure_raise(tmp_path): + out = _record_smart_macro(tmp_path, [ + ("step", [{"type": "B", "label": "X"}], lambda h: h.tap_at_xy(1, 2)), + ]) + + fresh = _fake_helpers() + fresh.active_app = lambda: "com.test.smart" + fresh.ui_tree = lambda visible_only=False, compact=False: [ + {"type": "Other", "label": "Z"}, + ] + llm = _mk_llm('{"skip": true, "reason": "no target"}') + + with pytest.raises(record_replay.MacroStepFailed) as exc: + record_replay.replay_smart(out, fresh, llm=llm, on_failure="raise") + assert exc.value.step_index == 0 + assert exc.value.intent == "step" + + +def test_recovery_skip_continues_to_next_step(tmp_path): + out = _record_smart_macro(tmp_path, [ + ("a", [{"type": "B", "label": "X"}], lambda h: h.tap_at_xy(1, 2)), + ("b", [{"type": "B", "label": "X"}], lambda h: h.tap_at_xy(3, 4)), + ]) + + fresh = _fake_helpers() + fresh.active_app = lambda: "com.test.smart" + fresh.ui_tree = lambda visible_only=False, compact=False: [ + {"type": "Other", "label": "Z"}, + ] + llm = _mk_llm(['{"skip": true, "reason": "no a"}', + '{"fn": "tap_at_xy", "args": [9, 9]}']) + + results = record_replay.replay_smart(out, fresh, llm=llm, on_failure="skip") + assert len(results) == 2 + assert results[0]["outcome"] == "skipped" + assert results[1]["outcome"] == "retargeted" + assert fresh._calls == [("tap_at_xy", (9, 9), {})] + + +def test_recovery_failed_helper_routed_per_on_failure(tmp_path): + out = _record_smart_macro(tmp_path, [ + (None, [], lambda h: h.tap_at_xy(1, 2)), + ]) + # Tamper with sidecar to reference a nonexistent helper + sidecar = Path(out).with_suffix(".py.jsonl") + sidecar.write_text(json.dumps({ + "t": 0.0, "fn": "no_such_helper", "args": [], "kwargs": {}, + "intent": "x", # force smart path + "fingerprint": {"app": "", "labels": [], "focused": None, "count": 0}, + }) + "\n") + + fresh = _fake_helpers() + with pytest.raises(record_replay.MacroStepFailed): + record_replay.replay_smart(out, fresh, on_failure="raise") + + results = record_replay.replay_smart(out, fresh, on_failure="skip") + assert results[0]["outcome"] == "failed" + assert "not found" in results[0]["error"]