Smart macros — record once, replay many (literal + LLM-adaptive)

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 <name>` opens a REPL with helpers + recording active; `mobile-use macro replay <name> --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

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 <name>` opens a REPL with helpers + recording
+active; `mobile-use macro replay <name> --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:

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 <path>`
+or `MU_MACRO_DIR=<path>`.
+
+```
+~/.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 <name> [--intent <txt>] [--dir <path>] [--ios|--android]
+mobile-use macro replay <name> [--smart|--literal] [--on-failure raise|skip] [--dir <path>] [--ios|--android]
+mobile-use macro list [--dir <path>]
+mobile-use macro show <name> [--dir <path>]
+```
+
+## 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.

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": "<helper-fn>", "args": [...], "kwargs": {{...}}}}
+  {{"skip": true, "reason": "<short 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:

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 <subcmd> — 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