do.py

"""cua do — one-shot VM automation commands for agents.

Output is always a single line starting with ✅ (success) or ❌ (failure),
followed by a context line:  💻 vm-name\t🪟 Window Title  (or Desktop).

Target VM is persisted in ~/.cua/do_target.json.
Zoom state (bbox + display scale) is also persisted there.
"""

from __future__ import annotations

import argparse
import base64
import io
import json
import os
import sys
import tempfile
from datetime import datetime
from pathlib import Path

# Ensure stdout/stderr can handle Unicode emojis on Windows
if sys.platform == "win32":
    import io as _io

    if hasattr(sys.stdout, "buffer"):
        sys.stdout = _io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", errors="replace")
    if hasattr(sys.stderr, "buffer"):
        sys.stderr = _io.TextIOWrapper(sys.stderr.buffer, encoding="utf-8", errors="replace")

MAX_LENGTH = 1200  # max pixel dimension for screenshot output

# ── state ─────────────────────────────────────────────────────────────────────

_STATE_FILE = Path.home() / ".cua" / "do_target.json"
_HOST_CONSENT_FILE = Path.home() / ".cua" / "host_consented"

# Window titles that are internal OS/browser helper windows and should be hidden
_SKIP_WINDOW_TITLES = {
    "Chrome Legacy Window",
}

PROVIDERS = ("cloud", "cloudv2", "local", "lume", "lumier", "docker", "winsandbox", "host")
_REMOTE_PROVIDERS = ("cloud", "cloudv2", "lume", "lumier", "docker", "winsandbox")


def _load_state() -> dict:
    if _STATE_FILE.exists():
        try:
            return json.loads(_STATE_FILE.read_text())
        except Exception:
            pass
    return {}


def _save_state(state: dict) -> None:
    _STATE_FILE.parent.mkdir(parents=True, exist_ok=True)
    _STATE_FILE.write_text(json.dumps(state, indent=2))


def _update_state(**kwargs) -> dict:
    state = _load_state()
    state.update(kwargs)
    _save_state(state)
    return state


def _host_consented() -> bool:
    return _HOST_CONSENT_FILE.exists()


# ── output helpers ────────────────────────────────────────────────────────────


def _ok(msg: str) -> int:
    print(f"✅ {msg}")
    return 0


def _fail(msg: str) -> int:
    print(f"❌ {msg}", file=sys.stderr)
    return 1


class _PtyUnavailable(Exception):
    """Raised when the remote PTY endpoint is unreachable; triggers run_command fallback."""


# ── provider / connection helpers ─────────────────────────────────────────────


async def _get_api_url(provider_type: str, name: str) -> str:
    """Resolve the computer-server API URL for the target sandbox."""
    if provider_type == "host":
        return "http://localhost:8000"

    if provider_type == "local":
        from cua_sandbox import sandbox_state

        state = sandbox_state.load(name)
        if not state:
            raise ValueError(f"Local sandbox '{name}' not found. Check ~/.cua/sandboxes/")
        return f"http://{state['host']}:{state['api_port']}"

    if provider_type in ("cloud", "cloudv2"):
        from cua_cli.auth.store import get_api_key
        from cua_sandbox.transport.cloud import CloudTransport, cloud_get_vm

        api_key = get_api_key()
        if not api_key:
            raise ValueError("Not authenticated. Run 'cua auth login' first")
        vm = await cloud_get_vm(name, api_key=api_key)
        if not vm:
            raise ValueError(f"VM not found: {name}")
        return CloudTransport._resolve_endpoint(vm)

    # Legacy local provider types — map to sandbox state file lookup
    if provider_type in ("lume", "lumier", "docker", "winsandbox"):
        from cua_sandbox import sandbox_state

        state = sandbox_state.load(name)
        if state:
            return f"http://{state['host']}:{state['api_port']}"
        return "http://localhost:8000"

    raise ValueError(f"Unknown provider: {provider_type}")


async def _host_dispatch(command: str, params: dict) -> dict:
    """Dispatch a computer-server command to the local host via cua_auto."""
    try:
        import cua_auto.keyboard as _kb
        import cua_auto.mouse as _mouse
        import cua_auto.screen as _screen
        import cua_auto.shell as _shell
        import cua_auto.window as _win
    except ImportError as e:
        return {
            "success": False,
            "error": f"cua-auto not installed: {e}. Run: pip install cua-auto",
        }

    try:
        if command == "screenshot":
            b64 = _screen.screenshot_b64()
            return {"success": True, "image_data": b64}

        elif command == "get_screen_size":
            w, h = _screen.screen_size()
            return {"success": True, "size": {"width": w, "height": h}}

        elif command == "get_cursor_position":
            x, y = _screen.cursor_position()
            return {"success": True, "position": {"x": x, "y": y}}

        elif command == "get_current_window_id":
            handle = _win.get_active_window_handle()
            if not handle:
                return {"success": False, "error": "No active window"}
            return {"success": True, "window_id": handle}

        elif command == "get_window_name":
            title = _win.get_window_name(params.get("window_id", ""))
            if title is None:
                return {"success": False, "error": "Window not found"}
            return {"success": True, "name": title}

        elif command == "get_application_windows":
            handles = _win.get_windows_with_title(params.get("app", ""))
            return {"success": True, "windows": handles}

        elif command == "get_window_size":
            result = _win.get_window_size(params.get("window_id", ""))
            if result is None:
                return {"success": False, "error": "Window not found"}
            return {"success": True, "size": [result[0], result[1]]}

        elif command == "get_window_position":
            result = _win.get_window_position(params.get("window_id", ""))
            if result is None:
                return {"success": False, "error": "Window not found"}
            return {"success": True, "position": [result[0], result[1]]}

        elif command == "left_click":
            _mouse.click(int(params["x"]), int(params["y"]))
            return {"success": True}

        elif command == "right_click":
            _mouse.right_click(int(params["x"]), int(params["y"]))
            return {"success": True}

        elif command == "middle_click":
            _mouse.click(int(params["x"]), int(params["y"]), "middle")
            return {"success": True}

        elif command == "double_click":
            _mouse.double_click(int(params["x"]), int(params["y"]))
            return {"success": True}

        elif command == "move_cursor":
            _mouse.move_to(int(params["x"]), int(params["y"]))
            return {"success": True}

        elif command == "mouse_down":
            x = params.get("x")
            y = params.get("y")
            _mouse.mouse_down(
                int(x) if x is not None else None,
                int(y) if y is not None else None,
                params.get("button", "left"),
            )
            return {"success": True}

        elif command == "mouse_up":
            x = params.get("x")
            y = params.get("y")
            _mouse.mouse_up(
                int(x) if x is not None else None,
                int(y) if y is not None else None,
                params.get("button", "left"),
            )
            return {"success": True}

        elif command == "scroll_direction":
            direction = params.get("direction", "down")
            clicks = int(params.get("clicks", 3))
            if direction == "up":
                _mouse.scroll_up(clicks)
            elif direction == "down":
                _mouse.scroll_down(clicks)
            elif direction == "left":
                _mouse.scroll_left(clicks)
            elif direction == "right":
                _mouse.scroll_right(clicks)
            return {"success": True}

        elif command == "drag_to":
            _mouse.drag(
                int(params["start_x"]),
                int(params["start_y"]),
                int(params["end_x"]),
                int(params["end_y"]),
            )
            return {"success": True}

        elif command == "type_text":
            _kb.type_text(params.get("text", ""))
            return {"success": True}

        elif command == "press_key":
            _kb.press_key(params.get("key", ""))
            return {"success": True}

        elif command == "key_down":
            _kb.key_down(params.get("key", ""))
            return {"success": True}

        elif command == "key_up":
            _kb.key_up(params.get("key", ""))
            return {"success": True}

        elif command == "hotkey":
            keys = params.get("keys", [])
            if isinstance(keys, str):
                keys = [k.strip() for k in keys.replace("-", "+").split("+") if k.strip()]
            _kb.hotkey(keys)
            return {"success": True}

        elif command == "run_command":
            result = _shell.run(params.get("command", ""))
            return {
                "success": True,
                "stdout": result.stdout,
                "stderr": result.stderr,
                "returncode": result.returncode,
            }

        elif command == "open":
            _win.open(params.get("path") or params.get("target", ""))
            return {"success": True}

        elif command == "launch":
            pid = _win.launch(params.get("app", ""), params.get("args"))
            return {"success": True, "pid": pid}

        elif command == "activate_window":
            ok = _win.activate_window(params.get("window_id", ""))
            return {"success": bool(ok)}

        elif command == "minimize_window":
            ok = _win.minimize_window(params.get("window_id", ""))
            return {"success": bool(ok)}

        elif command == "maximize_window":
            ok = _win.maximize_window(params.get("window_id", ""))
            return {"success": bool(ok)}

        elif command == "close_window":
            ok = _win.close_window(params.get("window_id", ""))
            return {"success": bool(ok)}

        elif command == "set_window_size":
            ok = _win.set_window_size(
                params["window_id"], int(params["width"]), int(params["height"])
            )
            return {"success": bool(ok)}

        elif command == "set_window_position":
            ok = _win.set_window_position(params["window_id"], int(params["x"]), int(params["y"]))
            return {"success": bool(ok)}

        elif command == "deactivate_window":
            return {"success": False, "error": "deactivate_window not supported on host"}

        else:
            return {"success": False, "error": f"Unknown host command: {command}"}

    except Exception as e:
        return {"success": False, "error": str(e)}


async def _send(provider_type: str, name: str, command: str, params: dict) -> dict:
    """Send a command to the computer-server and return the parsed response."""
    if provider_type == "host":
        return await _host_dispatch(command, params)

    import aiohttp
    from core.http import cua_version_headers

    api_url = await _get_api_url(provider_type, name)
    headers = {"Content-Type": "application/json", **cua_version_headers()}

    if provider_type in ("cloud", "cloudv2"):
        from cua_cli.auth.store import get_api_key

        api_key = get_api_key()
        if api_key:
            headers["X-API-Key"] = api_key
        if name:
            headers["X-Container-Name"] = name

    async with aiohttp.ClientSession() as session:
        async with session.post(
            f"{api_url}/cmd",
            json={"command": command, "params": params},
            headers=headers,
            timeout=aiohttp.ClientTimeout(total=30),
        ) as resp:
            text = await resp.text()
            for line in text.splitlines():
                if line.startswith("data: "):
                    return json.loads(line[6:])
            try:
                return json.loads(text)
            except Exception:
                return {"success": False, "error": f"Unexpected response: {text[:200]}"}


# ── zoom / screenshot helpers ─────────────────────────────────────────────────


async def _resolve_zoom_bbox_by_id(provider_type: str, name: str, window_id: str) -> dict | None:
    """Get the bounding box of a window by its native handle/id."""
    pos_r = await _send(provider_type, name, "get_window_position", {"window_id": window_id})
    size_r = await _send(provider_type, name, "get_window_size", {"window_id": window_id})

    pos = pos_r.get("position") or pos_r.get("data")
    size = size_r.get("size") or size_r.get("data")
    if not pos or not size:
        return None

    if isinstance(pos, (list, tuple)):
        x, y = int(pos[0]), int(pos[1])
    else:
        x, y = int(pos.get("x", 0)), int(pos.get("y", 0))

    if isinstance(size, (list, tuple)):
        w, h = int(size[0]), int(size[1])
    else:
        w, h = int(size.get("width", 0)), int(size.get("height", 0))

    return {"x": x, "y": y, "width": w, "height": h}


async def _resolve_zoom_bbox(provider_type: str, name: str, window_name: str) -> dict | None:
    """Get the bounding box of a window by app/window name.

    Filters out internal helper windows (e.g. 'Chrome Legacy Window') so the
    correct top-level window is always selected.
    """
    wins_r = await _send(provider_type, name, "get_application_windows", {"app": window_name})
    windows = wins_r.get("windows") or wins_r.get("data") or []
    if not windows:
        return None

    # Pick the first window whose title is not an internal helper
    window_id = None
    for wid in windows:
        try:
            name_r = await _send(provider_type, name, "get_window_name", {"window_id": wid})
            title = name_r.get("name") or ""
            if title not in _SKIP_WINDOW_TITLES:
                window_id = wid
                break
        except Exception:
            pass
    if window_id is None:
        window_id = windows[0]

    return await _resolve_zoom_bbox_by_id(provider_type, name, window_id)


async def _take_screenshot_data(
    provider_type: str, name: str, state: dict
) -> tuple[bytes, float, dict | None]:
    """Take screenshot, apply zoom crop and max-length scaling.

    Returns (image_bytes, display_scale, zoom_bbox).
    display_scale < 1.0 means the image was shrunk; bbox is None if not zoomed.
    """
    from PIL import Image

    result = await _send(provider_type, name, "screenshot", {})
    img_b64 = result.get("image_data") or result.get("data")
    if not img_b64:
        raise RuntimeError(result.get("error", "no image data returned"))
    img = Image.open(io.BytesIO(base64.b64decode(img_b64)))

    zoom_window = state.get("zoom_window")
    zoom_window_id = state.get("zoom_window_id")
    bbox: dict | None = None

    if zoom_window:
        try:
            if zoom_window_id:
                bbox = await _resolve_zoom_bbox_by_id(provider_type, name, str(zoom_window_id))
            if not bbox:
                bbox = await _resolve_zoom_bbox(provider_type, name, zoom_window)
            if bbox:
                img = img.crop(
                    (
                        bbox["x"],
                        bbox["y"],
                        bbox["x"] + bbox["width"],
                        bbox["y"] + bbox["height"],
                    )
                )
        except Exception:
            bbox = None  # fall back to full screen

    w, h = img.size
    scale = 1.0
    if max(w, h) > MAX_LENGTH:
        scale = MAX_LENGTH / max(w, h)
        img = img.resize((int(w * scale), int(h * scale)), Image.LANCZOS)

    buf = io.BytesIO()
    img.save(buf, format="PNG")
    return buf.getvalue(), scale, bbox


def _coords(x: int, y: int, state: dict) -> tuple[int, int]:
    """Translate image-space (x, y) to screen-space coordinates.

    Accounts for:
    - display_scale: image was scaled for output (divide to get original pixels)
    - zoom_bbox: image was cropped, add offset to get screen coords
    """
    zoom_scale: float = state.get("zoom_scale", 1.0) or 1.0
    zoom_bbox: dict | None = state.get("zoom_bbox")

    sx = round(x / zoom_scale)
    sy = round(y / zoom_scale)

    if zoom_bbox:
        sx += zoom_bbox["x"]
        sy += zoom_bbox["y"]

    return sx, sy


async def _maybe_focus_zoom(provider_type: str, name: str, state: dict) -> None:
    """If a zoom window is tracked, bring it to focus before performing an action."""
    window_id = state.get("zoom_window_id")
    if window_id:
        try:
            await _send(provider_type, name, "activate_window", {"window_id": window_id})
        except Exception:
            pass


async def _print_context(provider_type: str, name: str, state: dict | None = None) -> None:
    """Print the VM + zoom context line after a command."""
    vm_label = name if name else provider_type
    if state is None:
        state = _load_state()
    zoom_window = state.get("zoom_window")
    zoom_window_id = state.get("zoom_window_id")
    if zoom_window:
        zoom_info = (
            f"zoom: {zoom_window} ({zoom_window_id})" if zoom_window_id else f"zoom: {zoom_window}"
        )
    else:
        zoom_info = "zoom: off"
    print(f"💻 {vm_label}\t🔍 {zoom_info}")


async def _take_screenshot_for_recording(
    provider_type: str, name: str, state: dict
) -> bytes | None:
    """Silently take a screenshot for trajectory recording. Returns None on failure."""
    try:
        img_bytes, _, _ = await _take_screenshot_data(provider_type, name, state)
        return img_bytes
    except Exception:
        return None


def _maybe_record_turn(
    args: argparse.Namespace,
    state: dict,
    action_type: str,
    action_params: dict,
    screenshot_bytes: bytes | None = None,
) -> None:
    """Record a trajectory turn if recording is enabled. Never raises."""
    if getattr(args, "no_record", False):
        return
    try:
        from cua_cli.utils.trajectory_recorder import ensure_session, record_turn

        session_dir = ensure_session(state)
        record_turn(session_dir, action_type, action_params, screenshot_bytes)
    except Exception:
        pass  # Never interfere with the primary command


def _require_target() -> dict | None:
    state = _load_state()
    if not state.get("provider"):
        _fail("No VM selected. Run: cua do switch <provider> [name]")
        return None
    return state


# ── subcommand handlers ───────────────────────────────────────────────────────


def _cmd_switch(args: argparse.Namespace) -> int:
    provider = args.provider.lower()
    old_state = _load_state()
    had_zoom = bool(old_state.get("zoom_window"))

    if provider == "host":
        if not _host_consented():
            print(
                "❌ Warning: you are about to allow an AI to control your host PC directly.\n"
                "   This grants full keyboard, mouse, and screen access to your local desktop.\n"
                "   To continue, please run: cua do-host-consent",
                file=sys.stderr,
            )
            return 1
        _save_state(
            {
                "provider": "host",
                "name": "",
                "zoom_window": None,
                "zoom_window_id": None,
                "zoom_bbox": None,
                "zoom_scale": 1.0,
            }
        )
        try:
            from cua_cli.utils.trajectory_recorder import reset_session

            reset_session(_load_state())
        except Exception:
            pass
        msg = "Switched to host (local PC)"
        if had_zoom:
            msg += " — zoom reset"
        return _ok(msg)

    if provider not in PROVIDERS:
        return _fail(f"Unknown provider '{provider}'. Choose from: {', '.join(PROVIDERS)}")

    name = args.name or ""
    _save_state(
        {
            "provider": provider,
            "name": name,
            "zoom_window": None,
            "zoom_window_id": None,
            "zoom_bbox": None,
            "zoom_scale": 1.0,
        }
    )
    try:
        from cua_cli.utils.trajectory_recorder import reset_session

        reset_session(_load_state())
    except Exception:
        pass
    label = f"{provider}/{name}" if name else provider
    msg = f"Switched to {label}"
    if had_zoom:
        msg += " — zoom reset"
    return _ok(msg)


def _cmd_status(args: argparse.Namespace) -> int:
    state = _load_state()
    provider = state.get("provider")
    if not provider:
        return _fail("No VM selected. Run: cua do switch <provider> [name]")
    name = state.get("name", "")
    label = f"{provider}/{name}" if name else provider
    zoom = state.get("zoom_window")
    zoom_info = f" [zoomed to '{zoom}']" if zoom else ""
    return _ok(f"Current target: {label}{zoom_info}")


def _cmd_ls(args: argparse.Namespace) -> int:
    from cua_cli.utils.async_utils import run_async

    explicit_provider = getattr(args, "provider", None)

    async def _list_all() -> int:
        from cua_cli.auth.store import get_api_key
        from cua_sandbox import Sandbox

        print("  host  [local]")

        try:
            local_sandboxes = await Sandbox.list(local=True)
            for s in local_sandboxes:
                print(f"  {s.name}  [{s.status}]  {s.source}")
        except Exception:
            pass

        try:
            api_key = get_api_key()
            if api_key:
                cloud_sandboxes = await Sandbox.list(local=False, api_key=api_key)
                for s in cloud_sandboxes:
                    print(f"  {s.name}  [{s.status}]  cloud")
        except Exception:
            pass

        return 0

    async def _list_one(ptype: str) -> int:
        from cua_cli.auth.store import get_api_key
        from cua_sandbox import Sandbox

        if ptype == "host":
            print("  host  [local]")
            return 0

        local = ptype in ("local", "lume", "lumier", "docker", "winsandbox")
        try:
            api_key = None if local else get_api_key()
            sandboxes = await Sandbox.list(local=local, api_key=api_key)
            if not sandboxes:
                print(f"No sandboxes found for provider '{ptype}'")
                return 0
            for s in sandboxes:
                print(f"  {s.name}  [{s.status}]")
        except Exception as e:
            return _fail(str(e))
        return 0

    if explicit_provider:
        return run_async(_list_one(explicit_provider.lower()))
    else:
        return run_async(_list_all())


def _cmd_zoom(args: argparse.Namespace) -> int:
    from cua_cli.utils.async_utils import run_async

    t = _require_target()
    if not t:
        return 1
    window_name = args.window_name

    async def _run() -> int:
        state = _load_state()
        p, n = state["provider"], state.get("name", "")
        try:
            wins_r = await _send(p, n, "get_application_windows", {"app": window_name})
            windows = wins_r.get("windows") or wins_r.get("data") or []
            if not windows:
                return _fail(f"No windows found matching '{window_name}'")
            # Fetch titles and filter out internal helper windows
            candidates = []
            for wid in windows:
                name_r = await _send(p, n, "get_window_name", {"window_id": wid})
                title = name_r.get("name") or str(wid)
                if title not in _SKIP_WINDOW_TITLES:
                    candidates.append((wid, title))
            if not candidates:
                return _fail(f"No windows found matching '{window_name}'")
            if len(candidates) > 1:
                labels = [f"{wid} ({title})" for wid, title in candidates]
                return _fail(
                    f"Multiple windows matched '{window_name}' — be more specific. "
                    f"Found: {', '.join(labels)}"
                )
            wid, matched_title = candidates[0]
        except Exception:
            wid, matched_title = None, window_name  # resolution failed, still set zoom

        _update_state(zoom_window=window_name, zoom_window_id=wid, zoom_bbox=None, zoom_scale=1.0)
        id_suffix = f" ({wid})" if wid else ""
        return _ok(f"Zoomed to '{matched_title}'{id_suffix}")

    return run_async(_run())


def _cmd_unzoom(args: argparse.Namespace) -> int:
    _update_state(zoom_window=None, zoom_window_id=None, zoom_bbox=None, zoom_scale=1.0)
    return _ok("Unzoomed — full screen")


def _cmd_screenshot(args: argparse.Namespace) -> int:
    from cua_cli.utils.async_utils import run_async

    t = _require_target()
    if not t:
        return 1

    async def _run() -> int:
        state = _load_state()
        try:
            img_bytes, scale, bbox = await _take_screenshot_data(
                state["provider"], state.get("name", ""), state
            )
        except Exception as e:
            await _print_context(state["provider"], state.get("name", ""), state)
            return _fail(str(e))

        # Persist zoom state for coordinate translation
        _update_state(zoom_scale=scale, zoom_bbox=bbox)

        save_path = getattr(args, "save", None)
        if not save_path:
            ts = datetime.now().strftime("%Y%m%d_%H%M%S")
            save_path = f"{tempfile.gettempdir()}/cua_screenshot_{ts}.png"

        with open(save_path, "wb") as f:
            f.write(img_bytes)

        _maybe_record_turn(args, state, "screenshot", {}, img_bytes)

        rc = _ok(f"screenshot saved to {save_path}")
        await _print_context(state["provider"], state.get("name", ""), state)
        return rc

    return run_async(_run())


def _cmd_snapshot(args: argparse.Namespace) -> int:
    from cua_cli.utils.async_utils import run_async

    t = _require_target()
    if not t:
        return 1

    extra = " ".join(args.instructions) if args.instructions else ""

    async def _run() -> int:
        state = _load_state()
        try:
            img_bytes, scale, bbox = await _take_screenshot_data(
                state["provider"], state.get("name", ""), state
            )
        except Exception as e:
            await _print_context(state["provider"], state.get("name", ""), state)
            return _fail(str(e))

        _update_state(zoom_scale=scale, zoom_bbox=bbox)

        ts = datetime.now().strftime("%Y%m%d_%H%M%S")
        save_path = f"{tempfile.gettempdir()}/cua_snapshot_{ts}.png"
        with open(save_path, "wb") as f:
            f.write(img_bytes)

        api_key = os.environ.get("ANTHROPIC_API_KEY")
        if not api_key:
            await _print_context(state["provider"], state.get("name", ""), state)
            return _fail("ANTHROPIC_API_KEY not set")

        try:
            import anthropic as _anthropic

            client = _anthropic.Anthropic(api_key=api_key)
            prompt = (
                "You are analyzing a screenshot for an AI agent.\n"
                "1. Write a 1-2 sentence summary of what is currently on screen.\n"
                "2. List every interactive element visible (buttons, links, inputs, "
                "menus, checkboxes, dropdowns, etc.) with its center coordinates "
                "in image pixels (origin = top-left). Be precise."
            )
            if extra:
                prompt += f"\nAdditional instructions: {extra}"

            screenshot_analysis_tool = {
                "name": "screenshot_analysis",
                "description": "Report the analysis of the screenshot, including a summary and all interactive elements.",
                "input_schema": {
                    "type": "object",
                    "properties": {
                        "summary": {
                            "type": "string",
                            "description": "1-2 sentence summary of what is currently on screen",
                        },
                        "elements": {
                            "type": "array",
                            "items": {
                                "type": "object",
                                "properties": {
                                    "name": {"type": "string", "description": "Element label or text"},
                                    "type": {"type": "string", "description": "Element type (button, link, input, etc.)"},
                                    "x": {"type": "integer", "description": "Center X coordinate in pixels"},
                                    "y": {"type": "integer", "description": "Center Y coordinate in pixels"},
                                },
                                "required": ["name", "type", "x", "y"],
                            },
                            "description": "All interactive elements visible on screen",
                        },
                    },
                    "required": ["summary", "elements"],
                },
            }

            img_b64 = base64.b64encode(img_bytes).decode()
            response = client.messages.create(
                model="claude-haiku-4-5-20251001",
                max_tokens=1024,
                tools=[screenshot_analysis_tool],
                tool_choice={"type": "tool", "name": "screenshot_analysis"},
                messages=[
                    {
                        "role": "user",
                        "content": [
                            {
                                "type": "image",
                                "source": {
                                    "type": "base64",
                                    "media_type": "image/png",
                                    "data": img_b64,
                                },
                            },
                            {"type": "text", "text": prompt},
                        ],
                    }
                ],
            )

            # Extract the tool use result — guaranteed valid JSON matching the schema
            tool_block = next(b for b in response.content if b.type == "tool_use")
            parsed = tool_block.input
            summary = parsed.get("summary", "")
            elements = parsed.get("elements", [])
            print(f"✅ snapshot — {save_path}")
            print()
            print(summary)
            if elements:
                print()
                print("Interactive elements:")
                for el in elements:
                    print(
                        f"  • {el.get('name', '?')} [{el.get('type', '?')}]  ({el.get('x', '?')}, {el.get('y', '?')})"
                    )

        except ImportError:
            await _print_context(state["provider"], state.get("name", ""), state)
            return _fail("'anthropic' package not installed. Run: pip install anthropic")
        except Exception as e:
            await _print_context(state["provider"], state.get("name", ""), state)
            return _fail(f"AI analysis failed: {e}")

        await _print_context(state["provider"], state.get("name", ""), state)
        return 0

    return run_async(_run())


def _cmd_click(args: argparse.Namespace) -> int:
    from cua_cli.utils.async_utils import run_async

    t = _require_target()
    if not t:
        return 1

    button = getattr(args, "button", "left") or "left"
    cmd_map = {"left": "left_click", "right": "right_click", "middle": "middle_click"}
    cmd = cmd_map.get(button, "left_click")

    async def _run() -> int:
        state = _load_state()
        p, n = state["provider"], state.get("name", "")
        await _maybe_focus_zoom(p, n, state)
        sx, sy = _coords(args.x, args.y, state)
        try:
            result = await _send(p, n, cmd, {"x": sx, "y": sy})
        except Exception as e:
            await _print_context(p, n, state)
            return _fail(str(e))
        if not result.get("success", True):
            await _print_context(p, n, state)
            return _fail(result.get("error", "click failed"))
        _scr = await _take_screenshot_for_recording(p, n, state)
        _maybe_record_turn(args, state, "click", {"x": args.x, "y": args.y}, _scr)
        rc = _ok(f"clicked ({args.x}, {args.y}) [{button}]")
        await _print_context(p, n, state)
        return rc

    return run_async(_run())


def _cmd_dclick(args: argparse.Namespace) -> int:
    from cua_cli.utils.async_utils import run_async

    t = _require_target()
    if not t:
        return 1

    async def _run() -> int:
        state = _load_state()
        p, n = state["provider"], state.get("name", "")
        await _maybe_focus_zoom(p, n, state)
        sx, sy = _coords(args.x, args.y, state)
        try:
            result = await _send(p, n, "double_click", {"x": sx, "y": sy})
        except Exception as e:
            await _print_context(p, n, state)
            return _fail(str(e))
        if not result.get("success", True):
            await _print_context(p, n, state)
            return _fail(result.get("error", "double-click failed"))
        _scr = await _take_screenshot_for_recording(p, n, state)
        _maybe_record_turn(args, state, "double_click", {"x": args.x, "y": args.y}, _scr)
        rc = _ok(f"double-clicked ({args.x}, {args.y})")
        await _print_context(p, n, state)
        return rc

    return run_async(_run())


def _cmd_move(args: argparse.Namespace) -> int:
    from cua_cli.utils.async_utils import run_async

    t = _require_target()
    if not t:
        return 1

    async def _run() -> int:
        state = _load_state()
        p, n = state["provider"], state.get("name", "")
        await _maybe_focus_zoom(p, n, state)
        sx, sy = _coords(args.x, args.y, state)
        try:
            result = await _send(p, n, "move_cursor", {"x": sx, "y": sy})
        except Exception as e:
            await _print_context(p, n, state)
            return _fail(str(e))
        if not result.get("success", True):
            await _print_context(p, n, state)
            return _fail(result.get("error", "move failed"))
        _maybe_record_turn(args, state, "move", {"x": args.x, "y": args.y})
        rc = _ok(f"cursor moved to ({args.x}, {args.y})")
        await _print_context(p, n, state)
        return rc

    return run_async(_run())


def _cmd_type(args: argparse.Namespace) -> int:
    from cua_cli.utils.async_utils import run_async

    t = _require_target()
    if not t:
        return 1

    async def _run() -> int:
        state = _load_state()
        p, n = state["provider"], state.get("name", "")
        await _maybe_focus_zoom(p, n, state)
        try:
            result = await _send(p, n, "type_text", {"text": args.text})
        except Exception as e:
            await _print_context(p, n, state)
            return _fail(str(e))
        if not result.get("success", True):
            await _print_context(p, n, state)
            return _fail(result.get("error", "type failed"))
        _scr = await _take_screenshot_for_recording(p, n, state)
        _maybe_record_turn(args, state, "type", {"text": args.text}, _scr)
        preview = args.text[:40] + ("…" if len(args.text) > 40 else "")
        rc = _ok(f"typed: {preview!r}")
        await _print_context(p, n, state)
        return rc

    return run_async(_run())


def _cmd_key(args: argparse.Namespace) -> int:
    from cua_cli.utils.async_utils import run_async

    t = _require_target()
    if not t:
        return 1

    async def _run() -> int:
        state = _load_state()
        p, n = state["provider"], state.get("name", "")
        await _maybe_focus_zoom(p, n, state)
        try:
            result = await _send(p, n, "press_key", {"key": args.key})
        except Exception as e:
            await _print_context(p, n, state)
            return _fail(str(e))
        if not result.get("success", True):
            await _print_context(p, n, state)
            return _fail(result.get("error", "key press failed"))
        _scr = await _take_screenshot_for_recording(p, n, state)
        _maybe_record_turn(args, state, "keypress", {"keys": [args.key]}, _scr)
        rc = _ok(f"pressed key: {args.key}")
        await _print_context(p, n, state)
        return rc

    return run_async(_run())


def _cmd_hotkey(args: argparse.Namespace) -> int:
    from cua_cli.utils.async_utils import run_async

    t = _require_target()
    if not t:
        return 1

    keys = [k.strip() for k in args.keys.replace("-", "+").split("+") if k.strip()]

    async def _run() -> int:
        state = _load_state()
        p, n = state["provider"], state.get("name", "")
        await _maybe_focus_zoom(p, n, state)
        try:
            result = await _send(p, n, "hotkey", {"keys": keys})
        except Exception as e:
            await _print_context(p, n, state)
            return _fail(str(e))
        if not result.get("success", True):
            await _print_context(p, n, state)
            return _fail(result.get("error", "hotkey failed"))
        _scr = await _take_screenshot_for_recording(p, n, state)
        _maybe_record_turn(args, state, "hotkey", {"keys": keys}, _scr)
        rc = _ok(f"hotkey: {'+'.join(keys)}")
        await _print_context(p, n, state)
        return rc

    return run_async(_run())


def _cmd_scroll(args: argparse.Namespace) -> int:
    from cua_cli.utils.async_utils import run_async

    t = _require_target()
    if not t:
        return 1

    async def _run() -> int:
        state = _load_state()
        p, n = state["provider"], state.get("name", "")
        await _maybe_focus_zoom(p, n, state)
        try:
            result = await _send(
                p, n, "scroll_direction", {"direction": args.direction, "clicks": args.amount}
            )
        except Exception as e:
            await _print_context(p, n, state)
            return _fail(str(e))
        if not result.get("success", True):
            await _print_context(p, n, state)
            return _fail(result.get("error", "scroll failed"))
        _scr = await _take_screenshot_for_recording(p, n, state)
        _maybe_record_turn(
            args,
            state,
            "scroll",
            {"scroll_direction": args.direction, "scroll_amount": args.amount},
            _scr,
        )
        rc = _ok(f"scrolled {args.direction} {args.amount}x")
        await _print_context(p, n, state)
        return rc

    return run_async(_run())


def _cmd_drag(args: argparse.Namespace) -> int:
    from cua_cli.utils.async_utils import run_async

    t = _require_target()
    if not t:
        return 1

    async def _run() -> int:
        state = _load_state()
        p, n = state["provider"], state.get("name", "")
        await _maybe_focus_zoom(p, n, state)
        sx1, sy1 = _coords(args.x1, args.y1, state)
        sx2, sy2 = _coords(args.x2, args.y2, state)
        try:
            result = await _send(
                p,
                n,
                "drag_to",
                {"start_x": sx1, "start_y": sy1, "end_x": sx2, "end_y": sy2},
            )
        except Exception as e:
            await _print_context(p, n, state)
            return _fail(str(e))
        if not result.get("success", True):
            await _print_context(p, n, state)
            return _fail(result.get("error", "drag failed"))
        _scr = await _take_screenshot_for_recording(p, n, state)
        _maybe_record_turn(
            args,
            state,
            "drag",
            {"start_x": args.x1, "start_y": args.y1, "end_x": args.x2, "end_y": args.y2},
            _scr,
        )
        rc = _ok(f"dragged ({args.x1},{args.y1}) → ({args.x2},{args.y2})")
        await _print_context(p, n, state)
        return rc

    return run_async(_run())


def _default_shell() -> str:
    return "powershell" if sys.platform == "win32" else "bash"


def _cmd_shell_noninteractive(command: str | None, args: argparse.Namespace) -> int:
    """Run a non-interactive shell command via run_command (original behaviour)."""
    from cua_cli.utils.async_utils import run_async

    if not command or not command.strip():
        return _fail("No command provided")

    async def _run() -> int:
        state = _load_state()
        try:
            result = await _send(
                state["provider"], state.get("name", ""), "run_command", {"command": command}
            )
        except Exception as e:
            await _print_context(state["provider"], state.get("name", ""), state)
            return _fail(str(e))
        if not result.get("success", True):
            await _print_context(state["provider"], state.get("name", ""), state)
            return _fail(result.get("error", "shell command failed"))
        stdout = result.get("stdout", "").strip()
        rc_code = result.get("returncode", 0)
        if rc_code != 0:
            stderr = result.get("stderr", "").strip()
            await _print_context(state["provider"], state.get("name", ""), state)
            return _fail(f"exit {rc_code}: {stderr or stdout}")
        _maybe_record_turn(args, state, "shell", {"command": command})
        rc = _ok(stdout if stdout else "done")
        await _print_context(state["provider"], state.get("name", ""), state)
        return rc

    return run_async(_run())


def _shell_host_pty(command: str | None, cols: int | None = None, rows: int | None = None) -> int:
    """Interactive PTY session on the local host via cua_auto.terminal."""
    import shutil
    import signal
    import threading

    try:
        import cua_auto.terminal as _term
    except ImportError as e:
        return _fail(f"cua-auto not installed: {e}")

    _auto_cols, _auto_rows = shutil.get_terminal_size((80, 24))
    cols = cols if cols is not None else _auto_cols
    rows = rows if rows is not None else _auto_rows

    def _on_data(data: bytes) -> None:
        try:
            sys.stdout.buffer.write(data)
            sys.stdout.buffer.flush()
        except Exception:
            pass

    session = _term.terminal.create(
        command=command or _default_shell(),
        cols=cols,
        rows=rows,
        on_data=_on_data,
    )

    if sys.platform == "win32":
        # Windows: use msvcrt for raw stdin reading
        import msvcrt

        def _stdin_loop() -> None:
            while True:
                try:
                    ch = msvcrt.getch()
                    if not ch:
                        break
                    _term.terminal.send_stdin(session.pid, ch)
                except Exception:
                    break

        stdin_thread = threading.Thread(target=_stdin_loop, daemon=True)
        stdin_thread.start()
        exit_code = _term.terminal.wait(session.pid)

    else:
        # Unix/macOS: set raw mode, SIGWINCH for resize
        import termios
        import tty

        old_settings = termios.tcgetattr(sys.stdin.fileno())

        def _resize(_sig=None, _frame=None) -> None:
            c, r = shutil.get_terminal_size((80, 24))
            _term.terminal.resize(session.pid, c, r)

        signal.signal(signal.SIGWINCH, _resize)
        tty.setraw(sys.stdin.fileno())

        def _stdin_loop() -> None:
            while True:
                try:
                    data = os.read(sys.stdin.fileno(), 1024)
                    if not data:
                        break
                    _term.terminal.send_stdin(session.pid, data)
                except Exception:
                    break

        stdin_thread = threading.Thread(target=_stdin_loop, daemon=True)
        try:
            stdin_thread.start()
            exit_code = _term.terminal.wait(session.pid)
        finally:
            try:
                termios.tcsetattr(sys.stdin.fileno(), termios.TCSADRAIN, old_settings)
            except Exception:
                pass
            signal.signal(signal.SIGWINCH, signal.SIG_DFL)

    return exit_code or 0


async def _shell_remote_pty(
    provider: str, name: str, command: str | None, cols: int | None = None, rows: int | None = None
) -> int:
    """Interactive PTY session via WebSocket to a remote computer-server."""
    import asyncio
    import shutil
    import signal
    import threading

    import aiohttp

    _auto_cols, _auto_rows = shutil.get_terminal_size((80, 24))
    cols = cols if cols is not None else _auto_cols
    rows = rows if rows is not None else _auto_rows

    try:
        api_url = await _get_api_url(provider, name)
    except Exception as e:
        return _fail(str(e))

    ws_url = api_url.replace("https://", "wss://").replace("http://", "ws://")

    # Build auth headers (for the initial POST) and query params (for the WS).
    headers: dict = {}
    ws_params: dict = {}
    if provider in ("cloud", "cloudv2"):
        from cua_cli.auth.store import get_api_key

        api_key = get_api_key()
        if api_key:
            headers["X-API-Key"] = api_key
            ws_params["api_key"] = api_key
        if name:
            headers["X-Container-Name"] = name
            ws_params["container_name"] = name

    # Create PTY session
    try:
        async with aiohttp.ClientSession() as http:
            async with http.post(
                f"{api_url}/pty",
                json={"command": command or _default_shell(), "cols": cols, "rows": rows},
                headers={**headers, "Content-Type": "application/json"},
                timeout=aiohttp.ClientTimeout(total=15),
            ) as resp:
                resp.raise_for_status()
                data = await resp.json()
    except aiohttp.ClientResponseError as e:
        raise _PtyUnavailable(f"{e.status} {e.message} ({e.request_info.real_url})") from e
    except Exception as e:
        raise _PtyUnavailable(str(e)) from e

    pid: int = data["pid"]

    exit_code_cell: list[int] = [0]
    done_event = asyncio.Event()

    if sys.platform == "win32":
        # Windows: use msvcrt for raw stdin
        import msvcrt

        async def _run_ws() -> None:
            async with aiohttp.ClientSession() as http:
                async with http.ws_connect(f"{ws_url}/pty/{pid}/ws", params=ws_params) as ws:

                    def _stdin_loop() -> None:
                        while not done_event.is_set():
                            try:
                                ch = msvcrt.getch()
                                if ch:
                                    encoded = base64.b64encode(ch).decode()
                                    asyncio.run_coroutine_threadsafe(
                                        ws.send_str(json.dumps({"type": "stdin", "data": encoded})),
                                        asyncio.get_event_loop(),
                                    )
                            except Exception:
                                break

                    t = threading.Thread(target=_stdin_loop, daemon=True)
                    t.start()

                    async for raw_msg in ws:
                        if raw_msg.type == aiohttp.WSMsgType.TEXT:
                            try:
                                msg = json.loads(raw_msg.data)
                            except Exception:
                                continue
                            if msg.get("type") == "output":
                                chunk = base64.b64decode(msg["data"])
                                sys.stdout.buffer.write(chunk)
                                sys.stdout.buffer.flush()
                            elif msg.get("type") == "exit":
                                exit_code_cell[0] = int(msg.get("code", 0))
                                break
                        elif raw_msg.type in (aiohttp.WSMsgType.CLOSED, aiohttp.WSMsgType.ERROR):
                            break
                    done_event.set()

        await _run_ws()

    else:
        import termios
        import tty

        old_settings = termios.tcgetattr(sys.stdin.fileno())
        tty.setraw(sys.stdin.fileno())

        loop = asyncio.get_event_loop()

        async def _run_ws() -> None:
            async with aiohttp.ClientSession() as http:
                async with http.ws_connect(f"{ws_url}/pty/{pid}/ws", params=ws_params) as ws:

                    def _resize(_sig=None, _frame=None) -> None:
                        c, r = shutil.get_terminal_size((80, 24))
                        asyncio.run_coroutine_threadsafe(
                            ws.send_str(json.dumps({"type": "resize", "cols": c, "rows": r})),
                            loop,
                        )

                    signal.signal(signal.SIGWINCH, _resize)

                    def _stdin_loop() -> None:
                        while not done_event.is_set():
                            try:
                                data = os.read(sys.stdin.fileno(), 1024)
                                if not data:
                                    break
                                encoded = base64.b64encode(data).decode()
                                asyncio.run_coroutine_threadsafe(
                                    ws.send_str(json.dumps({"type": "stdin", "data": encoded})),
                                    loop,
                                )
                            except Exception:
                                break

                    t = threading.Thread(target=_stdin_loop, daemon=True)
                    t.start()

                    async for raw_msg in ws:
                        if raw_msg.type == aiohttp.WSMsgType.TEXT:
                            try:
                                msg = json.loads(raw_msg.data)
                            except Exception:
                                continue
                            if msg.get("type") == "output":
                                chunk = base64.b64decode(msg["data"])
                                sys.stdout.buffer.write(chunk)
                                sys.stdout.buffer.flush()
                            elif msg.get("type") == "exit":
                                exit_code_cell[0] = int(msg.get("code", 0))
                                break
                        elif raw_msg.type in (aiohttp.WSMsgType.CLOSED, aiohttp.WSMsgType.ERROR):
                            break

                    done_event.set()

        try:
            await _run_ws()
        finally:
            try:
                termios.tcsetattr(sys.stdin, termios.TCSADRAIN, old_settings)
            except Exception:
                pass
            signal.signal(signal.SIGWINCH, signal.SIG_DFL)

    return exit_code_cell[0]


def _cmd_shell(args: argparse.Namespace) -> int:
    command_parts = getattr(args, "shell_command", [])
    command = " ".join(command_parts).strip() if command_parts else None

    t = _require_target()
    if not t:
        return 1

    # Non-interactive (piped / scripted): keep old run_command behaviour
    if not sys.stdin.isatty():
        return _cmd_shell_noninteractive(command, args)

    # Interactive PTY mode
    state = _load_state()
    provider = state["provider"]
    name = state.get("name", "")

    cols: int | None = getattr(args, "cols", None)
    rows: int | None = getattr(args, "rows", None)

    if provider == "host":
        return _shell_host_pty(command, cols=cols, rows=rows)

    from cua_cli.utils.async_utils import run_async

    try:
        return run_async(_shell_remote_pty(provider, name, command, cols=cols, rows=rows))
    except _PtyUnavailable as e:
        if not command:
            return _fail(f"PTY unavailable: {e}")
        print(f"⚠️  PTY unavailable ({e}), falling back to run_command", file=sys.stderr)
        return _cmd_shell_noninteractive(command, args)


def _cmd_open(args: argparse.Namespace) -> int:
    from cua_cli.utils.async_utils import run_async

    t = _require_target()
    if not t:
        return 1

    async def _run() -> int:
        state = _load_state()
        try:
            result = await _send(
                state["provider"], state.get("name", ""), "open", {"path": args.path}
            )
        except Exception as e:
            await _print_context(state["provider"], state.get("name", ""), state)
            return _fail(str(e))
        if not result.get("success", True):
            await _print_context(state["provider"], state.get("name", ""), state)
            return _fail(result.get("error", "open failed"))
        _maybe_record_turn(args, state, "open", {"path": args.path})
        rc = _ok(f"opened: {args.path}")
        await _print_context(state["provider"], state.get("name", ""), state)
        return rc

    return run_async(_run())


# ── window subcommands ────────────────────────────────────────────────────────


def _cmd_window(args: argparse.Namespace) -> int:
    from cua_cli.utils.async_utils import run_async

    t = _require_target()
    if not t:
        return 1

    action = args.window_action

    async def _run() -> int:
        state = _load_state()
        p, n = state["provider"], state.get("name", "")

        if action == "ls":
            app = getattr(args, "app", "") or ""
            try:
                result = await _send(p, n, "get_application_windows", {"app": app})
            except Exception as e:
                return _fail(str(e))
            windows = result.get("windows") or result.get("data") or []
            if not isinstance(windows, list):
                print(result)
                return 0
            for wid in windows:
                try:
                    name_r = await _send(p, n, "get_window_name", {"window_id": wid})
                    title = name_r.get("name") or "?"
                    if title in _SKIP_WINDOW_TITLES:
                        continue
                    pos_r = await _send(p, n, "get_window_position", {"window_id": wid})
                    size_r = await _send(p, n, "get_window_size", {"window_id": wid})
                    pos = pos_r.get("position") or pos_r.get("data")
                    size = size_r.get("size") or size_r.get("data")
                    if pos and size:
                        if isinstance(pos, (list, tuple)):
                            x, y = int(pos[0]), int(pos[1])
                        else:
                            x, y = int(pos.get("x", 0)), int(pos.get("y", 0))
                        if isinstance(size, (list, tuple)):
                            w, h = int(size[0]), int(size[1])
                        else:
                            w, h = int(size.get("width", 0)), int(size.get("height", 0))
                        bbox = f"{x},{y},{w},{h}"
                    else:
                        bbox = "?"
                    print(f"  {wid}  {title}  [{bbox}]")
                except Exception:
                    print(f"  {wid}")
            return 0

        if action == "unfocus":
            try:
                result = await _send(p, n, "deactivate_window", {})
            except Exception:
                result = {"success": False}
            if not result.get("success", True):
                try:
                    result = await _send(p, n, "press_key", {"key": "escape"})
                except Exception as e:
                    await _print_context(p, n)
                    return _fail(str(e))
            rc = _ok("unfocused window")
            await _print_context(p, n)
            return rc

        # Commands requiring window_id
        wid = args.window_id

        cmd_map = {
            "focus": ("activate_window", {"window_id": wid}),
            "activate": ("activate_window", {"window_id": wid}),
            "minimize": ("minimize_window", {"window_id": wid}),
            "maximize": ("maximize_window", {"window_id": wid}),
            "close": ("close_window", {"window_id": wid}),
        }

        if action in cmd_map:
            cmd, params = cmd_map[action]
            try:
                result = await _send(p, n, cmd, params)
            except Exception as e:
                await _print_context(p, n)
                return _fail(str(e))
            if not result.get("success", True):
                await _print_context(p, n)
                return _fail(result.get("error", f"{action} failed"))
            rc = _ok(f"{action}d window {wid}")
            await _print_context(p, n)
            return rc

        if action == "resize":
            try:
                result = await _send(
                    p,
                    n,
                    "set_window_size",
                    {"window_id": wid, "width": args.width, "height": args.height},
                )
            except Exception as e:
                await _print_context(p, n)
                return _fail(str(e))
            if not result.get("success", True):
                await _print_context(p, n)
                return _fail(result.get("error", "resize failed"))
            rc = _ok(f"resized window {wid} to {args.width}x{args.height}")
            await _print_context(p, n)
            return rc

        if action == "move":
            try:
                result = await _send(
                    p,
                    n,
                    "set_window_position",
                    {"window_id": wid, "x": args.x, "y": args.y},
                )
            except Exception as e:
                await _print_context(p, n)
                return _fail(str(e))
            if not result.get("success", True):
                await _print_context(p, n)
                return _fail(result.get("error", "move failed"))
            rc = _ok(f"moved window {wid} to ({args.x},{args.y})")
            await _print_context(p, n)
            return rc

        if action == "info":
            try:
                size_r = await _send(p, n, "get_window_size", {"window_id": wid})
                pos_r = await _send(p, n, "get_window_position", {"window_id": wid})
                name_r = await _send(p, n, "get_window_name", {"window_id": wid})
            except Exception as e:
                return _fail(str(e))
            print(
                json.dumps(
                    {
                        "id": wid,
                        "title": name_r.get("name") or name_r.get("data") or "",
                        "size": size_r.get("size") or size_r.get("data"),
                        "position": pos_r.get("position") or pos_r.get("data"),
                    },
                    indent=2,
                )
            )
            return 0

        return _fail(f"Unknown window action: {action}")

    return run_async(_run())


# ── host consent command (registered separately as cua do-host-consent) ───────


def register_host_consent_parser(subparsers: argparse._SubParsersAction) -> None:
    subparsers.add_parser(
        "do-host-consent",
        help="Grant consent to control the host PC with cua do",
        description=(
            "Grants permission for 'cua do switch host', allowing AI agents to control\n"
            "your local keyboard, mouse, and screen via a local computer-server."
        ),
    )


def execute_host_consent(args: argparse.Namespace) -> int:
    _HOST_CONSENT_FILE.parent.mkdir(parents=True, exist_ok=True)
    _HOST_CONSENT_FILE.write_text("consented")
    _update_state(provider="host", name="")
    return _ok("consented — target switched to host")


# ── parser registration ───────────────────────────────────────────────────────


def register_parser(subparsers: argparse._SubParsersAction) -> None:
    p = subparsers.add_parser(
        "do",
        help="Send one-shot automation commands to a VM",
        description=(
            "Send automation commands to a target VM. "
            "Output is a single line starting with ✅ or ❌.\n"
            "A second line shows the current VM + focused window context.\n"
            "Coordinates are in screenshot-image space; zoom/max-length translation is applied automatically."
        ),
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  cua do switch docker my-container   Select target VM
  cua do switch host                  Control the local PC (requires consent)
  cua do-host-consent                 Grant consent for host control
  cua do ls                           List VMs for current provider
  cua do ls docker
  cua do status
  cua do zoom "Chrome"                Crop all screenshots to Chrome window
  cua do unzoom
  cua do screenshot
  cua do screenshot --save /tmp/s.png
  cua do snapshot                     Screenshot + AI summary (needs ANTHROPIC_API_KEY)
  cua do snapshot "focus on the form"
  cua do click 100 200
  cua do click 100 200 right
  cua do dclick 100 200
  cua do move 500 300
  cua do type "hello world"
  cua do key enter
  cua do hotkey cmd+c
  cua do scroll down 5
  cua do drag 100 200 400 500
  cua do shell "ls -la"
  cua do open https://example.com
  cua do window ls
  cua do window ls Terminal
  cua do window focus <id>
  cua do window unfocus
  cua do window minimize/maximize/close <id>
  cua do window resize <id> 1280 800
  cua do window move <id> 0 0
  cua do window info <id>
""",
    )

    p.add_argument(
        "--no-record",
        action="store_true",
        default=False,
        help="Disable trajectory recording for this command",
    )

    sub = p.add_subparsers(dest="do_action", metavar="action")
    sub.required = True

    # switch
    sw = sub.add_parser("switch", help="Select target VM (use 'host' for local PC)")
    sw.add_argument("provider", choices=PROVIDERS)
    sw.add_argument("name", nargs="?", default="")

    # status
    sub.add_parser("status", help="Show current target and zoom state")

    # ls
    ls = sub.add_parser("ls", help="List VMs for a provider")
    ls.add_argument("provider", nargs="?", choices=PROVIDERS)

    # zoom / unzoom
    zm = sub.add_parser("zoom", help="Crop screenshots + translate coords to a window")
    zm.add_argument("window_name", help="App/window name to zoom to")
    sub.add_parser("unzoom", help="Return to full-screen screenshots")

    # screenshot
    ss = sub.add_parser("screenshot", help="Take a screenshot (saved to temp dir)")
    ss.add_argument("--save", "-s", metavar="PATH")

    # snapshot
    sn = sub.add_parser(
        "snapshot",
        help="Screenshot + AI screen summary and interactive elements (needs ANTHROPIC_API_KEY)",
    )
    sn.add_argument("instructions", nargs="*", help="Extra instructions for the AI")

    # click
    cl = sub.add_parser("click", help="Click at coordinates (image space)")
    cl.add_argument("x", type=int)
    cl.add_argument("y", type=int)
    cl.add_argument("button", nargs="?", default="left", choices=["left", "right", "middle"])

    # dclick
    dc = sub.add_parser("dclick", help="Double-click at coordinates (image space)")
    dc.add_argument("x", type=int)
    dc.add_argument("y", type=int)

    # move
    mv = sub.add_parser("move", help="Move cursor (image space)")
    mv.add_argument("x", type=int)
    mv.add_argument("y", type=int)

    # type
    ty = sub.add_parser("type", help="Type text")
    ty.add_argument("text")

    # key
    ky = sub.add_parser("key", help="Press a key (enter, escape, tab, …)")
    ky.add_argument("key")

    # hotkey
    hk = sub.add_parser("hotkey", help="Keyboard shortcut (e.g. cmd+c, ctrl+shift+s)")
    hk.add_argument("keys")

    # scroll
    sc = sub.add_parser("scroll", help="Scroll in a direction")
    sc.add_argument("direction", choices=["up", "down", "left", "right"])
    sc.add_argument("amount", nargs="?", type=int, default=3)

    # drag
    dr = sub.add_parser("drag", help="Drag (image space)")
    dr.add_argument("x1", type=int)
    dr.add_argument("y1", type=int)
    dr.add_argument("x2", type=int)
    dr.add_argument("y2", type=int)

    # shell
    sh = sub.add_parser(
        "shell",
        help="Run a shell command (or open an interactive terminal) in the VM",
    )
    sh.add_argument("shell_command", nargs=argparse.REMAINDER)
    sh.add_argument("--cols", type=int, default=None, help="Terminal width (default: auto-detect)")
    sh.add_argument("--rows", type=int, default=None, help="Terminal height (default: auto-detect)")

    # open
    op = sub.add_parser("open", help="Open a file or URL")
    op.add_argument("path")

    # window
    win = sub.add_parser("window", help="Window management")
    win_sub = win.add_subparsers(dest="window_action", metavar="action")
    win_sub.required = True

    wls = win_sub.add_parser("ls", help="List windows")
    wls.add_argument("app", nargs="?", default="")

    win_sub.add_parser("unfocus", help="Remove focus from the current window")

    for act in ("focus", "activate", "minimize", "maximize", "close"):
        wp = win_sub.add_parser(act)
        wp.add_argument("window_id")

    wr = win_sub.add_parser("resize")
    wr.add_argument("window_id")
    wr.add_argument("width", type=int)
    wr.add_argument("height", type=int)

    wm = win_sub.add_parser("move")
    wm.add_argument("window_id")
    wm.add_argument("x", type=int)
    wm.add_argument("y", type=int)

    wi = win_sub.add_parser("info")
    wi.add_argument("window_id")


# ── dispatch ──────────────────────────────────────────────────────────────────


def execute(args: argparse.Namespace) -> int:
    dispatch = {
        "switch": _cmd_switch,
        "status": _cmd_status,
        "ls": _cmd_ls,
        "zoom": _cmd_zoom,
        "unzoom": _cmd_unzoom,
        "screenshot": _cmd_screenshot,
        "snapshot": _cmd_snapshot,
        "click": _cmd_click,
        "dclick": _cmd_dclick,
        "move": _cmd_move,
        "type": _cmd_type,
        "key": _cmd_key,
        "hotkey": _cmd_hotkey,
        "scroll": _cmd_scroll,
        "drag": _cmd_drag,
        "shell": _cmd_shell,
        "open": _cmd_open,
        "window": _cmd_window,
    }
    handler = dispatch.get(args.do_action)
    if not handler:
        return _fail(f"Unknown action: {args.do_action}")
    return handler(args)