"""Cua-driver backend (macOS, Windows, Linux).

Speaks MCP over stdio to `cua-driver`. The Python `mcp` SDK is async, so we
run a dedicated asyncio event loop on a background thread and marshal sync
calls through it.

The same `cua-driver call <tool>` surface (click, type_text, hotkey, drag,
scroll, screenshot, launch_app, list_apps, list_windows, get_window_state,
move_cursor, wait) works identically across macOS, Windows, and Linux —
cua-driver's PARITY matrix marks the action tools VERIFIED on macOS and
Windows in the cross-platform Rust port (`cua-driver-rs`).

Linux is the most recent runtime (X11 today, Wayland via XWayland; pure-
Wayland progress tracked upstream). It is enabled in
`check_computer_use_requirements` alongside macOS and Windows. The plumbing
in this file is OS-agnostic; per-host gaps (no DISPLAY, missing AT-SPI,
etc.) surface as specific blocked checks via `hermes computer-use doctor`
rather than failing silently.

Install:
  - **macOS**:
      /bin/bash -c "$(curl -fsSL https://raw.githubusercontent.com/trycua/cua/main/libs/cua-driver/scripts/install.sh)"
  - **Windows** (PowerShell):
      irm https://raw.githubusercontent.com/trycua/cua/main/libs/cua-driver/scripts/install.ps1 | iex

After install, `cua-driver` is on $PATH and supports `cua-driver mcp` (stdio
transport) which is what we invoke.

The macOS path uses private SkyLight SPIs (SLEventPostToPid,
SLPSPostEventRecordTo, _AXObserverAddNotificationAndCheckRemote) that aren't
Apple-public and can break on OS updates. The Windows path in cua-driver-rs
uses stable Win32 APIs (SendInput + UI Automation) — not subject to the
same SPI breakage class.
"""

from __future__ import annotations

import asyncio
import base64
from collections import deque
import concurrent.futures
import functools
import json
import logging
import os
import re
import shutil
import subprocess
import sys
import tempfile
import threading
import time
import uuid
from pathlib import PureWindowsPath
from typing import Any, Dict, List, Optional, Tuple

from hermes_cli._subprocess_compat import windows_hide_flags
from tools.computer_use.backend import (
    ActionResult,
    CaptureResult,
    ComputerUseBackend,
    UIElement,
)
from tools.computer_use.browser_route import CuaTypedBrowserRoute

logger = logging.getLogger(__name__)


def _action_result_from(
    name: str,
    ok: bool,
    message: str,
    meta: Dict[str, Any],
    structured: Dict[str, Any],
    *,
    requested_delivery: Optional[str] = None,
) -> ActionResult:
    """Build an ActionResult, lifting cua-driver's structured verdict.

    All structured fields are additive: a driver that omits
    ``structuredContent`` (or any individual field) leaves the corresponding
    ActionResult attribute ``None``, so callers and tests see unchanged
    behavior on old drivers. See the action response shape in
    cua-driver's mcp-tool-notes and NousResearch/hermes-agent#67052.
    """
    sc = structured if isinstance(structured, dict) else {}

    def _pick(key: str) -> Any:
        # structuredContent is canonical; fall back to a flattened meta copy.
        if key in sc:
            return sc.get(key)
        return meta.get(key)

    verified = _pick("verified")
    if not isinstance(verified, bool):
        verified = None
    effect = _pick("effect")
    if not isinstance(effect, str):
        effect = None
    escalation = _pick("escalation")
    if not isinstance(escalation, dict):
        escalation = None
    path = _pick("path")
    if not isinstance(path, str):
        path = None
    degraded = _pick("degraded")
    if not isinstance(degraded, bool):
        degraded = None
    # Refusal/limitation code — drivers spell it "code" or "reason_code".
    code = _pick("code") or _pick("reason_code")
    if not isinstance(code, str):
        code = None
    # Echo the delivery mode the caller actually requested (the driver's
    # `path` records the rung that ran; this records what we asked for).
    delivery_mode = requested_delivery if isinstance(requested_delivery, str) else None

    return ActionResult(
        ok=ok,
        action=name,
        message=message,
        meta=meta,
        verified=verified,
        effect=effect,
        escalation=escalation,
        path=path,
        degraded=degraded,
        delivery_mode=delivery_mode,
        code=code,
    )



# ---------------------------------------------------------------------------
# Update checking
# ---------------------------------------------------------------------------
#
# cua-driver ships a native `check-update` verb (and a `check_for_update` MCP
# tool) that compares the installed binary against the latest GitHub release —
# the source of truth — and caches the result (~20h). We prefer that over a
# hardcoded version floor, which would rot and can't know what "latest" is.
#
# There is intentionally no version *pin* knob: the upstream installer always
# fetches the latest release, so a `HERMES_CUA_DRIVER_VERSION` env var would
# only have *looked* like it pinned. For a reproducible version, point
# `HERMES_CUA_DRIVER_CMD` at a specific binary instead.

_CUA_DRIVER_CMD_ENV = "HERMES_CUA_DRIVER_CMD"
_CUA_DRIVER_DEFAULT_CMD = "cua-driver"
_CUA_DRIVER_ARGS = ["mcp"]  # stdio MCP transport (fallback when the
                            # driver doesn't expose `manifest` — see
                            # `_resolve_mcp_invocation` below)

# Whole-screen / desktop capture. cua-driver is a window-oriented driver —
# its `get_window_state` / `screenshot` tools capture a single window (by
# pid + window_id), and there is no MCP tool that captures the entire virtual
# desktop or an arbitrary monitor as one image. But the OS shell surfaces
# themselves (the desktop backdrop and the taskbar/menu-bar) are real windows
# that show up in `list_windows`, so "show me my screen" / "click the taskbar"
# is reachable by targeting those windows. When `app` is one of these
# sentinels, capture() resolves to the desktop/shell window instead of an
# application window.
_SCREEN_CAPTURE_SENTINELS = {"screen", "desktop", "fullscreen", "full screen", "all"}

# Known shell/desktop window identifiers across platforms. Matched
# case-insensitively as a substring against both the window's app_name and
# its title (cua-driver surfaces the Win32 class name / app name here).
#   Windows: Progman / WorkerW back the desktop; Shell_TrayWnd is the taskbar.
#   macOS:   Finder owns the desktop; the menu bar / Dock are the shell.
_DESKTOP_WINDOW_NAMES = (
    "progman", "workerw", "program manager",  # Windows desktop
    "shell_traywnd", "taskbar",               # Windows taskbar
    "finder", "desktop", "dock",              # macOS desktop / shell
)

# Linux/X11 can surface GNOME Shell / desktop backdrop windows before real app
# windows and cua-driver 0.6.x currently does not assign a useful z-order for
# them. These windows are targetable X11 windows but do not produce screenshots
# through get_window_state, so default app capture must skip them.
_NON_APP_WINDOW_TITLE_PREFIXES = (
    "@!",          # GNOME Shell background/monitor helper windows
    "Desktop",
    "gnome-shell",
    "GNOME Shell",
)


# Env var cua-driver reads to gate its anonymous usage telemetry (PostHog).
# Setting it to "0" disables telemetry; absence => the binary's own default
# (telemetry ON upstream).
_CUA_TELEMETRY_ENV_VAR = "CUA_DRIVER_RS_TELEMETRY_ENABLED"


def _computer_use_cfg() -> Dict[str, Any]:
    """The ``computer_use`` config block, or ``{}`` when config is unreadable."""
    try:
        from hermes_cli.config import load_config

        return (load_config() or {}).get("computer_use") or {}
    except Exception:
        return {}


def _cua_no_overlay() -> bool:
    """True when Hermes should pass ``--no-overlay`` to cua-driver.

    Reads ``computer_use.no_overlay``. Default ``None`` (auto-detect):
    disable the overlay where idle CPU burn is a known failure mode —
    macOS (cursor-overlay vImage redraw loop, #28152/#47032), headless
    Linux / WSL2 / containers — and keep it on Windows / desktop Linux
    with a display. Explicit ``True`` / ``False`` overrides auto-detection.
    """
    val = _computer_use_cfg().get("no_overlay")
    if val is not None:
        return bool(val)
    # Auto-detect: macOS overlay can peg a core indefinitely after a
    # computer_use session (#47032). Prefer off until the driver teardown
    # is solid; set computer_use.no_overlay: false to keep the cursor.
    if sys.platform == "darwin":
        return True
    if sys.platform != "linux":
        return False
    if not os.environ.get("DISPLAY"):
        return True
    try:
        with open("/proc/version", encoding="utf-8") as f:
            if "microsoft" in f.read().lower():
                return True
    except Exception:
        pass
    return False


def _cua_telemetry_disabled() -> bool:
    """True when Hermes should disable cua-driver telemetry for this user.

    Reads ``computer_use.cua_telemetry`` (default False → telemetry off).
    Unreadable config falls SAFE toward disabling telemetry.
    """
    # opt-in flag: True => user wants telemetry => do NOT disable.
    return not bool(_computer_use_cfg().get("cua_telemetry", False))


def _computer_use_max_image_dimension() -> Optional[int]:
    """Longest-edge cap for cua-driver screenshots, or None to leave unset.

    Reads ``computer_use.max_image_dimension`` (default 1456, matching the
    aux-vision downscale). ``0`` / negative / non-numeric → None.
    """
    try:
        dim = int(_computer_use_cfg().get("max_image_dimension", 1456))
    except (TypeError, ValueError):
        return 1456
    return dim if dim > 0 else None


def cua_driver_child_env(base_env: Optional[Dict[str, str]] = None) -> Dict[str, str]:
    """Return the environment dict for spawning cua-driver.

    Starts from ``base_env`` (defaults to ``os.environ``) and, when telemetry
    is disabled (the default), injects ``CUA_DRIVER_RS_TELEMETRY_ENABLED=0``.
    When the user has opted in, the var is left untouched so cua-driver uses
    its own default. Used by every cua-driver spawn site (MCP backend, status,
    doctor, install) so the policy is applied consistently.
    """
    env = dict(base_env if base_env is not None else os.environ)
    if _cua_telemetry_disabled():
        env[_CUA_TELEMETRY_ENV_VAR] = "0"
    return env


def _z_index_uninformative(windows: List[Dict[str, Any]]) -> bool:
    """True when every window shares the same z_index (common on Linux/X11)."""
    if not windows:
        return True
    return len({w.get("z_index", 0) for w in windows}) <= 1


def _parse_xprop_net_active_window(stdout: str) -> Optional[int]:
    """Parse ``xprop -root _NET_ACTIVE_WINDOW`` stdout into a window id.

    Accepts the common ``window id # 0x...`` form and falls back to the first
    hex token. Returns None for empty/unparseable output.
    """
    text = stdout or ""
    match = re.search(r"window id # (0x[0-9a-fA-F]+)", text)
    if not match:
        match = re.search(r"(0x[0-9a-fA-F]+)", text)
    if not match:
        return None
    try:
        return int(match.group(1), 16)
    except ValueError:
        return None


def _linux_x11_active_window_id() -> Optional[int]:
    """Best-effort read of ``_NET_ACTIVE_WINDOW`` via xprop. Never raises."""
    if sys.platform != "linux" or not os.environ.get("DISPLAY"):
        return None
    try:
        proc = subprocess.run(
            ["xprop", "-root", "_NET_ACTIVE_WINDOW"],
            capture_output=True,
            text=True, encoding="utf-8", errors="replace",
            timeout=2,
            check=False,
        )
    except Exception:
        return None
    if proc.returncode != 0:
        return None
    return _parse_xprop_net_active_window(proc.stdout or "")


def _is_real_app_window(w: Dict[str, Any]) -> bool:
    """Return False for desktop/shell helper windows that capture as empty."""
    title = w.get("title", "")
    return not any(
        title.startswith(p) or title.lower().startswith(p.lower())
        for p in _NON_APP_WINDOW_TITLE_PREFIXES
    )


def _select_capture_target(
    windows: List[Dict[str, Any]],
    *,
    app_requested: bool,
    exact_target: bool = False,
) -> Dict[str, Any]:
    """Select the best window for capture from normalized list_windows output.

    Callers pass windows already sorted by ``z_index`` descending (higher =
    frontmost). When ordering is informative, keep that frontmost contract.
    For unqualified default captures (no app filter and no exact
    pid/window_id) on Linux, desktop/shell helper windows (GNOME ``ding``
    "Desktop Icons", ``@!x,y;BDHF`` backdrop helpers) are skipped first —
    they are targetable X11 windows but capture as empty. Then, when every
    remaining candidate shares the same ``z_index`` (tied or unknown, the
    common Linux/X11 case), prefer ``_NET_ACTIVE_WINDOW`` over list order
    (#58026). Exact-target captures must not pay for an ``xprop`` probe.
    """
    candidates = [w for w in windows if not w["off_screen"]]
    pool = candidates
    if not exact_target and not app_requested and sys.platform == "linux":
        real_apps = [w for w in candidates if _is_real_app_window(w)]
        if real_apps:
            pool = real_apps
        if pool and _z_index_uninformative(pool):
            active_id = _linux_x11_active_window_id()
            if active_id is not None:
                for w in pool:
                    if w.get("window_id") == active_id:
                        return w
    if pool:
        return pool[0]
    return windows[0]


def _wsl_windows_path_to_posix(path: str) -> str:
    """Translate a Windows absolute manifest command when Hermes runs in WSL.

    Windows cua-driver manifests can report ``C:\\Users\\...\\cua-driver.exe``
    even though the Hermes process uses POSIX subprocess spawning inside WSL.
    The same file is reachable through DrvFS as ``/mnt/c/Users/...``.
    Non-Windows paths and non-WSL hosts are returned unchanged.
    """
    if not re.match(r"^[A-Za-z]:[\\/]", path):
        return path
    try:
        from hermes_constants import is_wsl

        if not is_wsl():
            return path
    except Exception:
        return path
    win = PureWindowsPath(path)
    drive = (win.drive or "").rstrip(":").lower()
    if not drive:
        return path
    return os.path.join("/mnt", drive, *(str(part) for part in win.parts[1:]))


class _EmbeddedCuaDaemon:
    """Private host-owned daemon used for an explicit unrestricted session.

    Cua Driver permission mode is immutable after daemon startup.  Reusing the
    machine-wide daemon would therefore let one Hermes session's YOLO choice
    affect another session.  A private embedded daemon gives the requesting
    session its own socket, process, and launch-time risk acknowledgement.
    """

    _START_TIMEOUT_SECONDS = 15.0

    def __init__(self, driver_cmd: str, permission_mode: str) -> None:
        if permission_mode != "unrestricted":
            raise ValueError("embedded permission override supports unrestricted only")
        self.permission_mode = permission_mode
        self._driver_cmd = driver_cmd
        self._command = driver_cmd
        self._mcp_args: List[str] = list(_CUA_DRIVER_ARGS)
        self._process: Any = None
        self._stderr_tail: deque[str] = deque(maxlen=20)
        self._stderr_thread: Optional[threading.Thread] = None
        token = uuid.uuid4().hex[:12]
        if sys.platform == "win32":
            self.socket_path = rf"\\.\pipe\hermes-cua-{token}"
        else:
            self.socket_path = os.path.join(
                tempfile.gettempdir(), f"hc-{token}.sock"
            )

    def child_env(self) -> Dict[str, str]:
        env = cua_driver_child_env()
        env["CUA_DRIVER_PERMISSION_MODE"] = "unrestricted"
        env["CUA_DRIVER_DANGEROUSLY_BYPASS_APPROVALS"] = "1"
        return env

    def _drain_stderr(self, process: Any) -> None:
        stream = getattr(process, "stderr", None)
        if stream is None:
            return
        try:
            for line in stream:
                text = str(line).strip()
                if text:
                    self._stderr_tail.append(text)
                    logger.debug("embedded cua-driver: %s", text)
        except Exception:
            pass

    def start(self) -> None:
        if self._process is not None and self._process.poll() is None:
            return
        from tools.environments.local import _sanitize_subprocess_env

        if not self._driver_cmd:
            self._driver_cmd = resolve_cua_driver_cmd() or ""
        if not self._driver_cmd:
            raise RuntimeError(cua_driver_install_hint())
        self._command, self._mcp_args = _resolve_mcp_invocation(self._driver_cmd)
        env = _sanitize_subprocess_env(self.child_env())
        command = [
            self._command,
            "serve",
            "--embedded",
            "--socket",
            self.socket_path,
            "--no-permissions-gate",
            "--permission-mode",
            "unrestricted",
            "--dangerously-bypass-approvals",
        ]
        self._process = subprocess.Popen(
            command,
            stdin=subprocess.DEVNULL,
            stdout=subprocess.DEVNULL,
            stderr=subprocess.PIPE,
            text=True,
            env=env,
        )
        self._stderr_thread = threading.Thread(
            target=self._drain_stderr,
            args=(self._process,),
            name="hermes-cua-daemon-stderr",
            daemon=True,
        )
        self._stderr_thread.start()

        deadline = time.monotonic() + self._START_TIMEOUT_SECONDS
        while time.monotonic() < deadline:
            if self._process.poll() is not None:
                detail = "; ".join(self._stderr_tail) or "no diagnostic output"
                raise RuntimeError(
                    f"embedded cua-driver exited during startup: {detail}"
                )
            try:
                probe = subprocess.run(
                    [self._command, "status", "--socket", self.socket_path],
                    stdin=subprocess.DEVNULL,
                    capture_output=True,
                    text=True,
                    timeout=2.0,
                    env=env,
                )
            except (OSError, subprocess.SubprocessError):
                probe = None
            if probe is not None and probe.returncode == 0:
                return
            time.sleep(0.1)

        self.stop()
        detail = "; ".join(self._stderr_tail) or "daemon did not become ready"
        raise RuntimeError(f"embedded cua-driver startup timed out: {detail}")

    def proxy_invocation(self) -> Tuple[str, List[str]]:
        if self._process is None or self._process.poll() is not None:
            raise RuntimeError("embedded cua-driver daemon is not running")
        return self._command, [
            *self._mcp_args,
            "--embedded",
            "--socket",
            self.socket_path,
        ]

    def stop(self) -> None:
        process = self._process
        self._process = None
        if process is not None and process.poll() is None:
            from tools.environments.local import _sanitize_subprocess_env

            try:
                subprocess.run(
                    [self._command, "stop", "--socket", self.socket_path],
                    stdin=subprocess.DEVNULL,
                    stdout=subprocess.DEVNULL,
                    stderr=subprocess.DEVNULL,
                    timeout=3.0,
                    env=_sanitize_subprocess_env(self.child_env()),
                )
            except (OSError, subprocess.SubprocessError):
                pass
            try:
                process.wait(timeout=5.0)
            except subprocess.TimeoutExpired:
                process.terminate()
                try:
                    process.wait(timeout=2.0)
                except subprocess.TimeoutExpired:
                    process.kill()
                    process.wait(timeout=2.0)
        if sys.platform != "win32" and os.path.exists(self.socket_path):
            try:
                os.remove(self.socket_path)
            except OSError:
                pass


def _resolve_mcp_invocation(
    driver_cmd: str,
    *,
    timeout: float = 6.0,
) -> Tuple[str, List[str]]:
    """Return ``(command, args)`` that spawn cua-driver's stdio MCP server.

    Surface 8 of NousResearch/hermes-agent#47072: instead of hardcoding
    ``["mcp"]`` we ask the driver itself via ``cua-driver manifest``
    (trycua/cua#1961). The manifest carries a stable ``mcp_invocation``
    pointer with both ``command`` and ``args``, so a future cua-driver
    that renames or relocates the subcommand keeps working without a
    Hermes patch.

    Falls back to ``(driver_cmd, ["mcp"])`` for older drivers that don't
    expose ``manifest``, or any indeterminate failure — the wrapper must
    not refuse to start just because the discovery hop failed.

    When ``computer_use.no_overlay`` is enabled (or auto-detected on
    Linux), ``--no-overlay`` is appended to suppress the cursor overlay
    rendering loop that can consume CPU indefinitely when idle
    (#28152, #47032).  Older drivers that don't recognise the flag will
    reject it; callers should fall back to the no-overlay invocation on
    spawn failure.
    """
    try:
        from tools.environments.local import _sanitize_subprocess_env
        proc = subprocess.run(
            [driver_cmd, "manifest"],
            capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=timeout,
            stdin=subprocess.DEVNULL,
            creationflags=windows_hide_flags(),
            # cua-driver is a third-party binary — never hand it provider
            # API keys via inherited env (same policy as the MCP and CLI
            # fallback spawns below; #53503/#55709/#58889 lineage).
            env=_sanitize_subprocess_env(cua_driver_child_env()),
        )
    except Exception:
        return driver_cmd, _mcp_args_with_overlay_flag(list(_CUA_DRIVER_ARGS), driver_cmd=driver_cmd)
    out = (proc.stdout or "").strip()
    if proc.returncode != 0 or not out:
        return driver_cmd, _mcp_args_with_overlay_flag(list(_CUA_DRIVER_ARGS), driver_cmd=driver_cmd)
    try:
        manifest = json.loads(out)
    except (ValueError, TypeError):
        return driver_cmd, _mcp_args_with_overlay_flag(list(_CUA_DRIVER_ARGS), driver_cmd=driver_cmd)
    if not isinstance(manifest, dict):
        return driver_cmd, _mcp_args_with_overlay_flag(list(_CUA_DRIVER_ARGS), driver_cmd=driver_cmd)
    invocation = manifest.get("mcp_invocation")
    if not isinstance(invocation, dict):
        return driver_cmd, _mcp_args_with_overlay_flag(list(_CUA_DRIVER_ARGS), driver_cmd=driver_cmd)
    args = invocation.get("args")
    command = invocation.get("command")
    if not isinstance(args, list) or not all(isinstance(a, str) for a in args):
        return driver_cmd, _mcp_args_with_overlay_flag(list(_CUA_DRIVER_ARGS), driver_cmd=driver_cmd)
    if not isinstance(command, str) or not command:
        # The driver knows the subcommand but didn't surface its own path.
        # Keep our resolved driver_cmd; the args are still authoritative.
        return driver_cmd, _mcp_args_with_overlay_flag(args, driver_cmd=driver_cmd)
    # A Windows-installed cua-driver can hand a WSL-hosted Hermes an absolute
    # ``C:\...`` command; translate it to its DrvFS ``/mnt/<drive>/...`` form
    # BEFORE the path-separator check (backslash is not a separator on POSIX,
    # so the raw Windows string would otherwise be discarded here).
    command = _wsl_windows_path_to_posix(command)
    if not _has_path_separator(command):
        # A manifest may legitimately retain the generic ``cua-driver`` name.
        # Under a GUI's thin PATH that would lose the resolved user-local path
        # and fail at MCP spawn, so preserve the concrete command we verified.
        return driver_cmd, _mcp_args_with_overlay_flag(args, driver_cmd=driver_cmd)
    # Manifest surfaced a relocated executable — probe THAT binary for
    # `--no-overlay` support rather than the system-resolved one, so a
    # wrapper/relocation with a different feature set doesn't crash on
    # an unknown flag (or silently keep an unwanted overlay).
    return command, _mcp_args_with_overlay_flag(args, driver_cmd=command)


def _mcp_args_with_overlay_flag(
    args: List[str],
    driver_cmd: str = _CUA_DRIVER_DEFAULT_CMD,
) -> List[str]:
    """Return *args* with ``--no-overlay`` appended when configured and supported."""
    if _cua_no_overlay() and _cua_driver_supports_no_overlay(driver_cmd):
        return [*args, "--no-overlay"]
    return list(args)


@functools.lru_cache(maxsize=1)
def _cua_driver_supports_no_overlay(driver_cmd: str) -> bool:
    """True if the installed cua-driver recognises ``--no-overlay``.

    Probes ``<driver> --help`` once and caches the result.  Older
    drivers (< 0.6.x) reject unknown flags, so passing ``--no-overlay``
    would crash the MCP spawn.
    """
    try:
        # cua-driver is a third-party binary — never hand it provider
        # API keys via inherited env (same policy as the manifest probe
        # and MCP spawn; #53503/#55709/#58889 lineage).
        from tools.environments.local import _sanitize_subprocess_env
        proc = subprocess.run(
            [driver_cmd, "--help"],
            capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=3.0,
            stdin=subprocess.DEVNULL,
            creationflags=windows_hide_flags(),
            env=_sanitize_subprocess_env(cua_driver_child_env()),
        )
        help_text = (proc.stdout or "") + (proc.stderr or "")
        return "--no-overlay" in help_text
    except Exception:
        return False

# Regex to parse element lines from get_window_state AX tree markdown.
#
# cua-driver renders each actionable node as one of:
#   - [N] AXRole "label"                         (quoted label, classic)
#   - [N] AXRole = "value"                        (value form, e.g. AXStaticText/AXPopUpButton)
#   - [N] AXRole (label)                          (parenthesised label, e.g. AXButton (Dark))
#   - [N] AXRole (order) id=Label                 (order number + id= label, newer builds)
#   - [N] AXRole id=Label                         (id= label only)
#   - [N] AXRole                                  (no label)
# followed by trailing metadata like [help="..." actions=[...]].
#
# Earlier the regex only matched the quoted and id= forms, so the very common
# `(label)` and `= "value"` forms (System Settings buttons, static text, popups)
# came back with an empty label — which made label-driven clicking impossible.
# A parenthesised group that is purely digits is an ORDER index, not a label, so
# it is excluded and we fall through to the id= label.
#
# Group 1: element index   Group 2: AX role
# Groups 3-6: the label in value / quoted / paren / id= form (whichever matched)
_ELEMENT_LINE_RE = re.compile(
    r'^\s*(?:-\s+)?\[(\d+)\]\s+(\w+)'
    r'(?:'
      r'\s*=\s*"([^"]*)"'              # = "value"
      r'|\s+"([^"]*)"'                 # "value"
      r'|\s+\((?!\d+\))([^)]*)\)'      # (value) but not a pure-digit (order) number
    r')?'
    r'(?:\s+(?:\(\d+\)\s+)?id=([^\s\[\]]+))?',  # optional id=value (after an optional (order))
    re.MULTILINE,
)


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------

def _is_macos() -> bool:
    return sys.platform == "darwin"


def _has_path_separator(value: str) -> bool:
    return os.sep in value or (os.altsep is not None and os.altsep in value)


def _candidate_cua_driver_commands(override: Optional[str] = None) -> List[str]:
    """Return candidate cua-driver commands in resolution order.

    ``override`` is authoritative when supplied. Otherwise a non-empty
    ``HERMES_CUA_DRIVER_CMD`` is authoritative; only when neither is set do we
    use PATH and canonical install locations.

    Desktop apps launched from Finder/Dock often inherit a narrow PATH that
    omits user-local install directories. The upstream cua-driver installer
    commonly places the binary under ``~/.local/bin`` on POSIX systems, so a
    Hermes Desktop/TUI session can otherwise filter out the `computer_use`
    tool even though `hermes computer-use doctor` succeeds from a login shell.
    """
    configured = (override if override is not None else os.environ.get(_CUA_DRIVER_CMD_ENV, "")).strip()
    if configured:
        # An explicit override is authoritative: if it is wrong, report the
        # driver missing instead of silently picking a different binary.
        return [configured]

    candidates = [_CUA_DRIVER_DEFAULT_CMD]
    home = os.path.expanduser("~")
    if sys.platform == "win32":
        candidates.extend([
            os.path.join(home, ".local", "bin", "cua-driver.exe"),
            os.path.join(home, ".local", "bin", "cua-driver"),
        ])
    else:
        candidates.extend([
            os.path.join(home, ".local", "bin", "cua-driver"),
            os.path.join(home, ".cargo", "bin", "cua-driver"),
            "/opt/homebrew/bin/cua-driver",
            "/usr/local/bin/cua-driver",
        ])
    return candidates


def resolve_cua_driver_cmd(override: Optional[str] = None) -> Optional[str]:
    """Resolve the cua-driver executable for every runtime/status surface.

    A supplied override (or ``HERMES_CUA_DRIVER_CMD``) is never silently
    replaced by another binary. Otherwise resolve PATH first, then canonical
    user-local installation locations used by the official installer.
    """
    for candidate in _candidate_cua_driver_commands(override):
        expanded = os.path.expanduser(candidate)
        if _has_path_separator(expanded):
            if shutil.which(expanded):
                return expanded
        else:
            resolved = shutil.which(expanded)
            if resolved:
                return resolved
    return None


def cua_driver_binary_available() -> bool:
    """True if `cua-driver` resolves via env, PATH, or known install paths."""
    return resolve_cua_driver_cmd() is not None


def cua_driver_update_check(*, timeout: Optional[float] = None) -> Optional[Dict[str, Any]]:
    """Run ``cua-driver check-update --json`` and return its parsed state.

    The payload mirrors the ``check_for_update`` MCP tool:
    ``{current_version, latest_version, update_available, ...}``.

    ``timeout`` defaults to 8s on POSIX and 25s on Windows — first-spawn of
    the exe there routinely eats several seconds in Defender/SmartScreen
    scanning, and a false timeout is expensive: callers treat ``None`` as
    indeterminate, and the ``install_cua_driver(upgrade=True)`` path used to
    fall through to a full multi-minute reinstall on it.

    Returns ``None`` (callers should stay quiet) when the result is
    indeterminate: the binary is missing, the driver is too old to support
    the verb (it predates trycua/cua#1734), the GitHub check failed (an
    ``error`` field is set), or the output didn't parse. Best-effort; never
    raises.
    """
    if timeout is None:
        timeout = 25.0 if sys.platform == "win32" else 8.0
    driver_cmd = resolve_cua_driver_cmd()
    if not driver_cmd:
        return None
    try:
        from tools.environments.local import _sanitize_subprocess_env
        proc = subprocess.run(
            [driver_cmd, "check-update", "--json"],
            capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=timeout,
            # Some older drivers don't have the verb and fall through to a
            # stdin-reading mode rather than erroring — DEVNULL gives them EOF
            # so they exit fast instead of blocking until the timeout.
            stdin=subprocess.DEVNULL,
            creationflags=windows_hide_flags(),
            # Sanitized like every other cua-driver spawn: third-party
            # binary, no inherited provider keys (#53503/#55709/#58889).
            env=_sanitize_subprocess_env(cua_driver_child_env()),
        )
    except Exception:
        return None
    out = (proc.stdout or "").strip()
    if not out:
        # Older drivers don't have the verb: usage goes to stderr, stdout empty.
        return None
    try:
        data = json.loads(out)
    except (ValueError, TypeError):
        return None
    if not isinstance(data, dict) or data.get("error"):
        # A failed check (exit 1) carries its reason in `error` — indeterminate.
        return None
    return data


def cua_driver_update_nudge() -> Optional[str]:
    """One-line "an update is available" message, or ``None`` when up to date,
    indeterminate, or the driver is too old to report."""
    state = cua_driver_update_check()
    if not state or not state.get("update_available"):
        return None
    latest = state.get("latest_version") or "?"
    current = state.get("current_version") or "?"
    return (
        f"cua-driver {latest} is available (you have {current}); "
        f"update with `hermes computer-use install --upgrade`."
    )


_update_checked = False


def _maybe_nudge_update() -> None:
    """Emit an update nudge at most once per process, off-thread so the
    (cached, ~20h) GitHub poll never blocks the first computer_use action."""
    global _update_checked
    if _update_checked:
        return
    _update_checked = True

    def _run() -> None:
        try:
            msg = cua_driver_update_nudge()
        except Exception:
            return
        if msg:
            logger.info("computer_use: %s", msg)

    threading.Thread(
        target=_run, name="cua-driver-update-check", daemon=True
    ).start()


def cua_driver_install_hint() -> str:
    if sys.platform == "win32":
        installer = (
            '  irm https://raw.githubusercontent.com/trycua/cua/main/'
            'libs/cua-driver/scripts/install.ps1 | iex'
        )
    else:
        installer = (
            '  /bin/bash -c "$(curl -fsSL '
            'https://raw.githubusercontent.com/trycua/cua/main/'
            'libs/cua-driver/scripts/install.sh)"'
        )
    return (
        "cua-driver is not installed. Install with one of:\n"
        "  hermes computer-use install\n"
        "Or run the upstream installer directly:\n"
        f"{installer}\n"
        "Or run `hermes tools` and enable the Computer Use toolset to install it automatically."
    )


def _parse_elements_from_tree(markdown: str) -> List[UIElement]:
    """Parse UIElement list from get_window_state AX tree markdown.

    Last-resort fallback for cua-driver builds that don't carry the
    canonical ``structuredContent.elements`` array (see
    ``_parse_elements_from_structured`` — Surface 2 of #47072 prefers
    that path).

    Captures the label whichever form cua-driver used: ``= "value"``,
    ``"quoted"``, ``(parenthesised)``, or ``id=Label``. Bounds always
    come back ``(0, 0, 0, 0)`` because the markdown surface doesn't
    carry them — yet another reason to prefer the structured path;
    element-index clicks don't need them (the driver resolves the index
    to a frame internally).
    """
    elements = []
    for m in _ELEMENT_LINE_RE.finditer(markdown):
        # groups 3-6: value / quoted / paren / id= label (first non-None wins)
        label = m.group(3) or m.group(4) or m.group(5) or m.group(6) or ""
        elements.append(UIElement(
            index=int(m.group(1)),
            role=m.group(2),
            label=label,
            bounds=(0, 0, 0, 0),
        ))
    return elements


def _parse_elements_from_structured(raw_elements: List[Dict[str, Any]]) -> List[UIElement]:
    """Surface 2 of NousResearch/hermes-agent#47072: read the canonical
    ``structuredContent.elements`` array cua-driver-rs emits on every
    ``get_window_state`` response (trycua/cua#1961).

    Each entry has at minimum ``element_index``, ``role``, ``label``;
    ``frame`` (``{x, y, w, h}``) is included whenever the AT-SPI /
    AXFrame call returned usable bounds. Older code parsed the same
    information out of the markdown tree via a regex (lossy: bounds
    were always ``(0, 0, 0, 0)``) — this path preserves the real
    frame so downstream consumers (e.g. ``UIElement.center()``) work
    against pixel coordinates instead of just the index lookup.

    Unknown / malformed entries are skipped rather than failing the
    whole walk — the wrapper degrades to "fewer elements" rather than
    "no elements" on a bad row.
    """
    elements: List[UIElement] = []
    for raw in raw_elements:
        if not isinstance(raw, dict):
            continue
        idx = raw.get("element_index")
        if not isinstance(idx, int):
            continue
        role = raw.get("role") if isinstance(raw.get("role"), str) else ""
        label = raw.get("label") if isinstance(raw.get("label"), str) else ""
        frame = raw.get("frame") if isinstance(raw.get("frame"), dict) else None
        bounds: Tuple[int, int, int, int] = (0, 0, 0, 0)
        if frame:
            try:
                bounds = (
                    int(frame.get("x", 0)),
                    int(frame.get("y", 0)),
                    int(frame.get("w", 0)),
                    int(frame.get("h", 0)),
                )
            except (TypeError, ValueError):
                bounds = (0, 0, 0, 0)
        # Surface 6: opaque element_token. cua-driver-rs format is
        # `s{snapshot_hex}:{index}`. We treat it as a black-box string —
        # the driver owns the parse + LRU semantics.
        raw_token = raw.get("element_token")
        token = raw_token if isinstance(raw_token, str) and raw_token else None
        elements.append(UIElement(
            index=idx,
            role=role,
            label=label,
            bounds=bounds,
            element_token=token,
        ))
    return elements


def _image_dimensions_from_bytes(raw: bytes) -> Tuple[int, int]:
    """Best-effort PNG/JPEG dimension sniffing without extra dependencies."""
    if raw.startswith(b"\x89PNG\r\n\x1a\n") and len(raw) >= 24:
        width = int.from_bytes(raw[16:20], "big")
        height = int.from_bytes(raw[20:24], "big")
        if width > 0 and height > 0:
            return width, height

    if raw.startswith(b"\xff\xd8"):
        i = 2
        n = len(raw)
        while i + 9 < n:
            if raw[i] != 0xFF:
                i += 1
                continue
            marker = raw[i + 1]
            i += 2
            if marker in {0xD8, 0xD9} or 0xD0 <= marker <= 0xD7:
                continue
            if i + 2 > n:
                break
            segment_len = int.from_bytes(raw[i:i + 2], "big")
            if segment_len < 2 or i + segment_len > n:
                break
            if marker in {
                0xC0, 0xC1, 0xC2, 0xC3, 0xC5, 0xC6, 0xC7,
                0xC9, 0xCA, 0xCB, 0xCD, 0xCE, 0xCF,
            }:
                if segment_len >= 7:
                    height = int.from_bytes(raw[i + 3:i + 5], "big")
                    width = int.from_bytes(raw[i + 5:i + 7], "big")
                    if width > 0 and height > 0:
                        return width, height
                break
            i += segment_len

    return 0, 0


def _split_tree_text(full_text: str) -> Tuple[str, str]:
    """Split get_window_state text into (summary_line, tree_markdown)."""
    lines = full_text.split("\n", 1)
    summary = lines[0]
    tree = lines[1] if len(lines) > 1 else ""
    return summary, tree


def _parse_key_combo(keys: str) -> Tuple[Optional[str], List[str]]:
    """Parse a key string like 'cmd+s' into (key, modifiers).

    Returns (key, modifiers) where key is the non-modifier key and modifiers
    is a list of modifier names (cmd, shift, option, ctrl).
    """
    MODIFIER_NAMES = {"cmd", "command", "shift", "option", "alt", "ctrl", "control", "fn"}
    KEY_ALIASES = {"command": "cmd", "alt": "option", "control": "ctrl"}

    parts = [p.strip().lower() for p in re.split(r'[+\-]', keys) if p.strip()]
    modifiers = []
    key = None
    for part in parts:
        normalized = KEY_ALIASES.get(part, part)
        if normalized in MODIFIER_NAMES:
            modifiers.append(normalized)
        else:
            key = part  # last non-modifier wins
    return key, modifiers


# ---------------------------------------------------------------------------
# Asyncio bridge — one long-lived loop on a background thread
# ---------------------------------------------------------------------------

class _AsyncBridge:
    """Runs one asyncio loop on a daemon thread; marshals coroutines from the caller."""

    def __init__(self) -> None:
        self._loop: Optional[asyncio.AbstractEventLoop] = None
        self._thread: Optional[threading.Thread] = None
        self._ready = threading.Event()

    def start(self) -> None:
        if self._thread and self._thread.is_alive():
            return
        self._ready.clear()

        def _run() -> None:
            self._loop = asyncio.new_event_loop()
            asyncio.set_event_loop(self._loop)
            self._ready.set()
            try:
                self._loop.run_forever()
            finally:
                try:
                    self._loop.close()
                except Exception:
                    pass

        self._thread = threading.Thread(target=_run, daemon=True, name="cua-driver-loop")
        self._thread.start()
        if not self._ready.wait(timeout=5.0):
            raise RuntimeError("cua-driver asyncio bridge failed to start")

    def run(self, coro, timeout: Optional[float] = 30.0) -> Any:
        from agent.async_utils import safe_schedule_threadsafe
        if not self._loop or not self._thread or not self._thread.is_alive():
            if asyncio.iscoroutine(coro):
                coro.close()
            raise RuntimeError("cua-driver bridge not started")
        fut = safe_schedule_threadsafe(coro, self._loop)
        if fut is None:
            raise RuntimeError("cua-driver bridge not started")
        return fut.result(timeout=timeout)

    def stop(self) -> None:
        if self._loop and self._loop.is_running():
            self._loop.call_soon_threadsafe(self._loop.stop)
        if self._thread:
            self._thread.join(timeout=2.0)
        self._thread = None
        self._loop = None


# ---------------------------------------------------------------------------
# MCP session (lazy, shared across tool calls)
# ---------------------------------------------------------------------------

class _CuaDriverSession:
    """Holds the mcp ClientSession. Spawned lazily; re-entered on drop.

    Lifecycle ownership: a single long-running coroutine
    (`_lifecycle_coro`) opens both the stdio_client and ClientSession
    contexts, populates capabilities, sets `_ready_event`, and then waits
    on `_shutdown_event`. When shutdown is signalled the same coroutine
    closes the contexts — keeping anyio's cancel-scope task-identity
    invariant intact (the bridge schedules each `bridge.run(coro)` as a
    NEW task, so opening contexts in one and closing them in another
    raises "Attempted to exit cancel scope in a different task").
    Tool calls run in their own short-lived tasks; they only touch the
    session object, never the surrounding contexts.
    """

    def __init__(
        self,
        bridge: _AsyncBridge,
        embedded_daemon: Optional[_EmbeddedCuaDaemon] = None,
    ) -> None:
        self._bridge = bridge
        self._embedded_daemon = embedded_daemon
        self._session = None
        self._lock = threading.Lock()
        self._started = False
        # Surface 4 of NousResearch/hermes-agent#47072: per-tool
        # capability-token sets, populated from `tools/list` at session
        # init. Keys are tool names (e.g. "click", "get_window_state");
        # values are sets of capability strings (e.g.
        # "accessibility.element_tokens", "input.keyboard.type.terminal_safe").
        # Empty until the session starts; consumers should call
        # `supports_capability` rather than reading directly.
        self._capabilities: Dict[str, set] = {}
        # Raw input schemas are the compatibility source of truth for action
        # properties.  cua-driver 0.9-era builds advertise delivery_mode in
        # inputSchema while intentionally omitting the old, fabricated
        # ``input.delivery_mode`` capability token.
        self._tool_schemas: Dict[str, Dict[str, Any]] = {}
        self._capability_version: str = ""
        # Lifecycle plumbing — see class docstring above.
        self._ready_event = threading.Event()
        self._shutdown_event: Optional[asyncio.Event] = None  # created on bridge loop
        self._lifecycle_future = None  # concurrent.futures.Future
        self._setup_error: Optional[BaseException] = None
        # Stable driver-side identity declared through start_session.
        # Used to revive a logical ended-session rejection without
        # recursive call_tool re-entry or backend-owned state (#71166).
        self._declared_session_id: Optional[str] = None

    def _require_started(self) -> None:
        if not self._started:
            raise RuntimeError("cua-driver session not started")

    async def _lifecycle_coro(self) -> None:
        """Long-lived owner of the stdio MCP contexts. Opens, signals
        ready, blocks on shutdown, then cleans up. enter + exit happen
        in the SAME asyncio task, so anyio's cancel-scope invariant
        holds — fixing the "Attempted to exit cancel scope in a
        different task than it was entered in" warning emitted by the
        previous _aenter/_aexit split.
        """
        import time as _time
        from mcp import ClientSession, StdioServerParameters
        from mcp.client.stdio import stdio_client
        from tools.environments.local import _sanitize_subprocess_env

        # Build the shutdown event on the loop's thread so the asyncio
        # primitive belongs to the correct loop.
        self._shutdown_event = asyncio.Event()
        _t0 = _time.monotonic()
        # Phase marker surfaced by the ready-timeout error (issue #57025):
        # when startup wedges, the caller reports HOW FAR it got instead of
        # an opaque "never reached ready".
        self._startup_phase = "binary-check"

        try:
            driver_cmd = resolve_cua_driver_cmd()
            if not driver_cmd:
                raise RuntimeError(cua_driver_install_hint())

            # Surface 8: ask cua-driver itself which subcommand spawns
            # the MCP server, instead of hardcoding ["mcp"]. Falls back
            # transparently for older drivers / any discovery failure.
            self._startup_phase = "manifest-discovery"
            if self._embedded_daemon is not None:
                command, args = self._embedded_daemon.proxy_invocation()
                child_env = self._embedded_daemon.child_env()
            else:
                command, args = _resolve_mcp_invocation(driver_cmd)
                child_env = cua_driver_child_env()
            _t_manifest = _time.monotonic()
            params = StdioServerParameters(
                command=command,
                args=args,
                # Apply the telemetry policy first (default: disabled), then
                # sanitize Hermes-managed secrets out of the child env.
                env=_sanitize_subprocess_env(child_env),
            )

            async with stdio_client(params) as (read, write):
                self._startup_phase = "mcp-initialize"
                async with ClientSession(read, write) as session:
                    await session.initialize()
                    _t_init = _time.monotonic()
                    # Populate capabilities + capability_version BEFORE
                    # exposing the session to callers, so the first
                    # tool call already sees them.
                    self._startup_phase = "capability-discovery"
                    await self._populate_capabilities(session)
                    self._session = session
                    self._startup_phase = "ready"
                    self._ready_event.set()
                    logger.info(
                        "cua-driver session ready in %.1fs "
                        "(manifest=%.1fs, mcp_init=%.1fs)",
                        _time.monotonic() - _t0,
                        _t_manifest - _t0,
                        _t_init - _t_manifest,
                    )
                    # Hold the contexts open until stop() / restart asks
                    # us to wind down. Tool calls run as their own tasks
                    # on the same loop and touch self._session directly.
                    await self._shutdown_event.wait()
        except BaseException as e:
            # Capture both ordinary errors and anyio CancelledError.
            # The caller (start()) inspects this to surface setup
            # failures to the synchronous world.
            self._setup_error = e
            self._ready_event.set()
            raise
        finally:
            # Clearing _session before the contexts unwind would let a
            # racing call_tool see None during teardown — but the
            # outer context-manager exits AFTER this block, so set to
            # None here is fine: stop() has already flipped _started.
            self._session = None
            # Reset _started so a session that dies for ANY reason (MCP
            # connection drop, driver crash, unexpected coro exit) is
            # re-enterable: the next start()/call sees _started False and
            # rebuilds the session instead of hanging forever on a dead one
            # via _require_started(). On the normal stop() path this is a
            # harmless idempotent no-op (stop() already set it False). A
            # plain bool write is atomic in CPython, so this is safe from
            # the bridge-loop thread without taking self._lock (which stop()
            # may hold while awaiting this coro's future). See #55048 Bug 1.
            self._started = False

    async def _populate_capabilities(self, session: Any) -> None:
        """Surface 4: cache per-tool capability sets + capability_version
        from tools/list. Soft prerequisite — discovery failure leaves
        the map empty and supports_capability degrades to False."""
        self._capabilities = {}
        self._tool_schemas = {}
        self._capability_version = ""
        try:
            tools_list = await session.list_tools()
            for tool in getattr(tools_list, "tools", []) or []:
                tool_name = getattr(tool, "name", None)
                if not isinstance(tool_name, str):
                    continue
                caps = getattr(tool, "capabilities", None)
                if caps is None:
                    # Some MCP SDKs forward custom fields via
                    # `model_extra` (Pydantic v2) instead of attributes.
                    extra = getattr(tool, "model_extra", None) or {}
                    caps = extra.get("capabilities")
                if isinstance(caps, list):
                    self._capabilities[tool_name] = {
                        c for c in caps if isinstance(c, str)
                    }
                else:
                    self._capabilities[tool_name] = set()
                schema = getattr(tool, "inputSchema", None)
                if schema is None:
                    schema = (getattr(tool, "model_extra", None) or {}).get(
                        "inputSchema"
                    )
                self._tool_schemas[tool_name] = (
                    dict(schema) if isinstance(schema, dict) else {}
                )
            # capability_version is a top-level sibling of `tools` on the
            # tools/list response. cua-driver-core/src/tool.rs:354 emits
            # it; cua-driver-core/src/protocol.rs:150 leaves it OUT of
            # initialize — so we discover here, not there.
            cv = getattr(tools_list, "capability_version", None)
            if cv is None:
                extra = getattr(tools_list, "model_extra", None) or {}
                cv = extra.get("capability_version")
            if isinstance(cv, str):
                self._capability_version = cv
        except Exception as e:
            logger.debug("cua-driver tools/list capability discovery failed: %s", e)

    def start(self) -> None:
        with self._lock:
            if self._started:
                return
            self._bridge.start()
            self._start_lifecycle_locked()
            self._started = True

    def _start_lifecycle_locked(self) -> None:
        """Spawn the lifecycle owner and wait for it to reach ready.
        Caller must hold self._lock."""
        # Reset per-session state.
        self._ready_event = threading.Event()
        self._setup_error = None
        self._shutdown_event = None
        # Fire-and-forget schedule on the bridge loop. The future tracks
        # completion of the WHOLE lifecycle (open → wait → close), not
        # just the open step — start() waits on _ready_event separately.
        loop = self._bridge._loop
        if loop is None:
            raise RuntimeError("cua-driver bridge not started")
        self._lifecycle_future = asyncio.run_coroutine_threadsafe(
            self._lifecycle_coro(), loop
        )
        if not self._ready_event.wait(timeout=30.0):
            # Best-effort: signal shutdown if the future is still alive.
            self._signal_shutdown_locked()
            # Surface which startup phase wedged (issue #57025) — "doctor
            # passes but the wrapper times out" reports are undiagnosable
            # from a bare "never reached ready".
            phase = getattr(self, "_startup_phase", "unknown")
            from hermes_constants import display_hermes_home
            raise RuntimeError(
                "cua-driver session never reached ready (timeout 30s; "
                f"stuck in phase: {phase}). "
                "Run `hermes computer-use doctor` and check "
                f"{display_hermes_home()}/logs/agent.log for the phase timings."
            )
        # If setup failed, the lifecycle coroutine set _setup_error
        # before setting _ready_event. Re-raise it on the caller's thread.
        if self._setup_error is not None:
            raise RuntimeError(
                f"cua-driver session setup failed: {self._setup_error}"
            ) from self._setup_error

    def stop(self) -> None:
        with self._lock:
            if not self._started:
                return
            self._started = False
            self._stop_lifecycle_locked()

    def _stop_lifecycle_locked(self) -> None:
        """Signal shutdown + wait for the lifecycle coroutine to unwind.
        Caller must hold self._lock."""
        self._signal_shutdown_locked()
        fut = self._lifecycle_future
        if fut is None:
            return
        try:
            # 5s budget for context unwind (stdio_client teardown).
            fut.result(timeout=5.0)
        except concurrent.futures.TimeoutError:
            logger.warning("cua-driver session shutdown timed out (5s)")
        except Exception as e:
            # Real shutdown errors (not the previous cancel-scope race
            # which is now structurally impossible) still get surfaced.
            logger.warning("cua-driver shutdown error: %s", e)
        finally:
            self._lifecycle_future = None

    def _signal_shutdown_locked(self) -> None:
        """Set the asyncio shutdown event from the caller's thread."""
        loop = self._bridge._loop
        event = self._shutdown_event
        if loop is not None and event is not None and loop.is_running():
            try:
                loop.call_soon_threadsafe(event.set)
            except RuntimeError:
                # Loop closed — nothing to signal.
                pass

    async def _call_tool_async(self, name: str, args: Dict[str, Any]) -> Dict[str, Any]:
        result = await self._session.call_tool(name, args)
        return _extract_tool_result(result)

    # ── Capability detection (Surface 4 of #47072) ────────────────────
    def supports_capability(self, capability: str, tool: Optional[str] = None) -> bool:
        """Return True when the connected cua-driver advertises the given
        capability token (trycua/cua#1961 capability vocabulary).

        When ``tool`` is given, scope the check to that specific tool's
        advertised capability set. When omitted, return True if ANY tool
        advertises the capability — useful for "is this feature available
        anywhere on the driver" probes.

        Always returns False before the session is started (so consumers
        on a dead/uninitialised wrapper degrade rather than crash).
        """
        if tool is not None:
            return capability in self._capabilities.get(tool, set())
        return any(capability in caps for caps in self._capabilities.values())

    def _has_tool(self, name: str) -> bool:
        """Return True when ``tools/list`` advertised a tool by this name.

        Used to route capture(): cua-driver dropped the standalone
        ``screenshot`` tool and folded full-window PNG capture into
        ``get_window_state`` (whose own description notes it "Also captures
        a PNG screenshot of the specified window"). Older drivers that still
        expose ``screenshot`` keep using it; newer ones fall through to
        ``get_window_state``.

        Returns False when discovery hasn't populated the map yet — callers
        treat that as "unknown" and probe defensively rather than trusting it.
        """
        return name in self._capabilities

    def supports_input_property(self, tool: str, property_name: str) -> bool:
        """Return whether a live action schema accepts ``property_name``.

        This deliberately inspects tools/list rather than guessing from the
        package version or requiring a capability token the driver never
        shipped.  A missing/invalid schema fails closed.
        """
        schema = getattr(self, "_tool_schemas", {}).get(tool, {})
        properties = schema.get("properties") if isinstance(schema, dict) else None
        return isinstance(properties, dict) and property_name in properties

    @property
    def capabilities_discovered(self) -> bool:
        """True once ``tools/list`` populated the per-tool map. When False,
        ``_has_tool`` answers are not trustworthy (discovery failed or the
        session hasn't started) and capture() should probe defensively."""
        return bool(self._capabilities)

    @property
    def capability_version(self) -> str:
        """Driver-advertised capability vocabulary version (empty string
        when the driver predates the field — older builds had no version)."""
        return self._capability_version

    @staticmethod
    def _logical_error_text(result: Dict[str, Any]) -> str:
        """Flatten a logical MCP error into text for narrow classification."""
        chunks: List[str] = []
        for value in (result.get("data"), result.get("structuredContent")):
            if isinstance(value, str):
                chunks.append(value)
            elif value is not None:
                try:
                    chunks.append(json.dumps(value, sort_keys=True))
                except (TypeError, ValueError):
                    chunks.append(str(value))
        return "\n".join(chunks)

    @classmethod
    def _is_ended_session_result(cls, result: Any) -> bool:
        """Recognise cua-driver's explicit recoverable ended-session result."""
        if not isinstance(result, dict) or result.get("isError") is not True:
            return False
        message = cls._logical_error_text(result).lower()
        return (
            "session" in message
            and ("has ended" in message or "session ended" in message)
            and "start_session" in message
        )

    def _revive_declared_session_once(
        self,
        name: str,
        args: Dict[str, Any],
        first_result: Dict[str, Any],
        timeout: float,
    ) -> Dict[str, Any]:
        """Revive the stable session and replay one rejected tool call once."""
        session_id = self._declared_session_id
        if not session_id or name in self._LIFECYCLE_CALLS:
            return first_result

        logger.warning(
            "cua-driver session %s ended during %s; reviving and retrying once",
            session_id,
            name,
        )
        revive_result = self._bridge.run(
            self._call_tool_async("start_session", {"session": session_id}),
            timeout=timeout,
        )
        if revive_result.get("isError") is True:
            logger.warning(
                "cua-driver session %s could not be revived: %s",
                session_id,
                self._logical_error_text(revive_result),
            )
            return first_result

        # Return the second result as-is. A second rejection is surfaced; no loop.
        return self._bridge.run(
            self._call_tool_async(name, args),
            timeout=timeout,
        )

    @staticmethod
    def _is_closed_session_error(exc: Exception) -> bool:
        """Return True for MCP/stdio failures that are recoverable by reconnecting."""
        name = exc.__class__.__name__
        module = getattr(exc.__class__, "__module__", "")
        return (
            name in {"ClosedResourceError", "BrokenResourceError", "EndOfStream"}
            or (module.startswith("anyio") and "Resource" in name)
            or isinstance(exc, (BrokenPipeError, EOFError))
        )

    @staticmethod
    def _is_transient_daemon_error(exc: Exception) -> bool:
        """Return True for the cua-driver daemon-proxy EAGAIN congestion error.

        On macOS the ``cua-driver mcp`` bridge forwards calls to the CuaDriver
        daemon over a non-blocking unix socket. Heavier ops (notably
        ``get_window_state``, which walks the AX tree and captures a PNG) can
        come back as an ``McpError`` carrying ``Resource temporarily
        unavailable (os error 35)`` — POSIX EAGAIN — when the socket buffer is
        momentarily full. This is transient by definition: the same call
        succeeds when retried after a short pause (which is why spaced-out
        single calls work while rapid/large ones intermittently fail). Detect
        it by message so we can retry with backoff rather than surfacing an
        empty 0x0 capture to the model. See the EAGAIN diagnosis in
        references/catalog-add-troubleshooting (apple-music skill) and the
        cua-driver daemon-proxy note.
        """
        msg = str(exc)
        return (
            "Resource temporarily unavailable" in msg
            or "os error 35" in msg
            or "daemon transport error" in msg
            or "daemon proxy" in msg
        )

    def _restart_session_locked(self) -> None:
        """Recreate the MCP session after the daemon/stdin transport was closed.
        Caller must hold self._lock (the reconnect-once retry path holds it)."""
        if self._started:
            try:
                self._stop_lifecycle_locked()
            except Exception as e:
                logger.debug("cua-driver session cleanup before reconnect failed: %s", e)
        self._started = False
        # Clear stale capability state; the next start populates from scratch.
        self._capabilities = {}
        self._capability_version = ""
        self._start_lifecycle_locked()
        self._started = True

    def _call_tool_via_cli(self, name: str, args: Dict[str, Any], timeout: float) -> Dict[str, Any]:
        """Fallback transport: invoke ``cua-driver call <tool> <json>`` as a
        subprocess instead of going through the stdio MCP bridge.

        The ``cua-driver mcp`` stdio bridge can persistently fail to forward
        heavier calls (notably ``get_window_state``) to the daemon with POSIX
        EAGAIN, while the plain ``cua-driver call`` path — which talks to the
        daemon over its own socket — keeps working. When the MCP path gives up,
        we retry over the CLI and remap the JSON into the same dict shape that
        ``_extract_tool_result`` produces, so callers (capture(), _action(),
        list_windows parsing) are transport-agnostic.

        For ``get_window_state`` we route the screenshot to a temp file via
        ``screenshot_out_file`` so the daemon returns a tiny JSON body (a path)
        instead of a multi-megabyte base64 blob — the large payload is what
        congests the daemon socket and triggers EAGAIN in the first place. We
        read the PNG back from disk and base64-encode it ourselves. The CLI
        call is itself retried a few times with backoff, since the underlying
        daemon socket can still be momentarily busy.
        """
        import subprocess as _subprocess
        import tempfile as _tempfile
        import time as _time
        from tools.environments.local import _sanitize_subprocess_env

        call_args = dict(args)
        shot_file: Optional[str] = None
        if name == "get_window_state" and "screenshot_out_file" not in call_args:
            fd, shot_file = _tempfile.mkstemp(prefix="cua_shot_", suffix=".png")
            os.close(fd)
            call_args["screenshot_out_file"] = shot_file

        driver_command = resolve_cua_driver_cmd()
        if not driver_command:
            raise RuntimeError(cua_driver_install_hint())
        child_env = cua_driver_child_env()
        socket_args: List[str] = []
        embedded_daemon = getattr(self, "_embedded_daemon", None)
        if embedded_daemon is not None:
            driver_command = embedded_daemon.proxy_invocation()[0]
            child_env = embedded_daemon.child_env()
            socket_args = ["--socket", embedded_daemon.socket_path]
        cmd = [
            driver_command,
            "call",
            name,
            json.dumps(call_args),
            *socket_args,
        ]
        attempts = 4
        backoff = 0.5
        parsed: Any = None
        last_err = ""
        try:
            for attempt in range(attempts):
                try:
                    proc = _subprocess.run(
                        cmd, capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=max(15.0, timeout),
                        creationflags=windows_hide_flags(),
                        env=_sanitize_subprocess_env(child_env),
                    )
                except Exception as e:  # pragma: no cover - subprocess spawn failure
                    raise RuntimeError(f"cua-driver CLI fallback for {name} failed to spawn: {e}") from e

                out = (proc.stdout or "").strip()
                last_err = out[:200] or (proc.stderr or "")[:200]
                start = min(
                    (i for i in (out.find("{"), out.find("[")) if i != -1),
                    default=-1,
                )
                if start != -1:
                    try:
                        candidate = json.loads(out[start:])
                    except json.JSONDecodeError:
                        candidate = None
                    if candidate is not None:
                        parsed = candidate
                        break
                # No JSON (EAGAIN warning / empty) — retry with backoff.
                if attempt < attempts - 1:
                    logger.warning(
                        "cua-driver CLI fallback for %s got no JSON "
                        "(attempt %d/%d); retrying in %.1fs",
                        name, attempt + 1, attempts, backoff,
                    )
                    _time.sleep(backoff)
                    backoff *= 2

            if parsed is None:
                raise RuntimeError(
                    f"cua-driver CLI fallback for {name} returned no JSON after "
                    f"{attempts} attempts: {last_err}"
                )

            # Remap structured JSON into {data, images, structuredContent, isError}.
            images: List[str] = []
            data: Any = None
            structured: Optional[Dict] = parsed if isinstance(parsed, dict) else None
            is_error = False
            if isinstance(parsed, dict):
                # Current cua-driver CLI responses may report logical failures
                # in-band even when the subprocess itself exits successfully.
                # Preserve that bit so stateful callers can fail closed.
                is_error = parsed.get("isError") is True or parsed.get("is_error") is True
                shot = parsed.get("screenshot_png_b64")
                if not shot:
                    # Screenshot was routed to a file (ours or the daemon's choice).
                    fpath = parsed.get("screenshot_file_path") or shot_file
                    if fpath and os.path.exists(fpath):
                        try:
                            with open(fpath, "rb") as fh:
                                shot = base64.b64encode(fh.read()).decode("ascii")
                        except Exception as e:
                            logger.debug("cua-driver CLI fallback: failed reading %s: %s", fpath, e)
                if shot:
                    images.append(shot)
                tree = parsed.get("tree_markdown")
                if tree is not None:
                    ec = parsed.get("element_count")
                    summary = f"{ec} elements" if ec is not None else ""
                    data = f"{summary}\n{tree}" if summary else tree
            return {
                "data": data,
                "images": images,
                "structuredContent": structured,
                "isError": is_error,
            }
        finally:
            if shot_file and os.path.exists(shot_file):
                try:
                    os.remove(shot_file)
                except OSError:
                    pass

    # Lifecycle handshake calls issued BY start()/stop() themselves — these
    # must not trigger the auto-restart guard below, or start() would recurse
    # into start() when the session-start hasn't flipped _started yet.
    _LIFECYCLE_CALLS = frozenset({"start_session", "end_session"})

    def call_tool(self, name: str, args: Dict[str, Any], timeout: float = 30.0) -> Dict[str, Any]:
        # A prior session may have died (MCP drop / driver crash): its
        # lifecycle coro reset _started to False in its finally (#55048).
        if not self._started and name not in self._LIFECYCLE_CALLS:
            logger.warning(
                "cua-driver session not active on %s; (re)starting before call", name
            )
            self.start()
        self._require_started()

        try:
            result = self._bridge.run(
                self._call_tool_async(name, args),
                timeout=timeout,
            )
        except Exception as e:
            if self._is_transient_daemon_error(e):
                logger.warning(
                    "cua-driver MCP transport failed on %s (%s); "
                    "falling back to CLI transport", name, e,
                )
                return self._call_tool_via_cli(name, args, timeout)
            if not self._is_closed_session_error(e):
                raise
            logger.warning("cua-driver MCP session closed during %s; reconnecting once", name)
            with self._lock:
                self._restart_session_locked()
            result = self._bridge.run(
                self._call_tool_async(name, args),
                timeout=timeout,
            )

        # Remember only a successfully declared stable identity. Failed
        # start_session calls must not leave stale recovery state behind.
        if name == "start_session" and result.get("isError") is not True:
            declared_id = args.get("session")
            if isinstance(declared_id, str) and declared_id:
                self._declared_session_id = declared_id

        if self._is_ended_session_result(result):
            result = self._revive_declared_session_once(name, args, result, timeout)

        if (
            name == "end_session"
            and result.get("isError") is not True
            and args.get("session") == self._declared_session_id
        ):
            self._declared_session_id = None
        return result


def _extract_tool_result(mcp_result: Any) -> Dict[str, Any]:
    """Convert an mcp CallToolResult into a plain dict.

    cua-driver returns a mix of text parts, image parts, and structuredContent.
    We flatten into:
      {
        "data": <text or parsed json>,
        "images": [b64, ...],
        "image_mime_types": [mime, ...],   # parallel to `images`, "" when absent
        "structuredContent": <dict|None>,
        "isError": bool,
      }
    structuredContent is populated from the MCP result's structuredContent field
    (MCP spec §2024-11-05+) and takes precedence for structured data like
    list_windows window arrays.

    `image_mime_types` is the explicit `mimeType` cua-driver emits on every
    image part as of trycua/cua#1961 (Surface 7 of
    NousResearch/hermes-agent#47072). Each entry corresponds index-for-index
    with `images`; an empty string entry signals the part carried no
    mimeType (older cua-driver build), and the caller should fall back to
    base64-prefix sniffing.
    """
    data: Any = None
    images: List[str] = []
    image_mime_types: List[str] = []
    # Use identity, not truthiness: unittest mocks and proxy objects commonly
    # synthesize truthy attributes that were never present in the real result.
    is_error = getattr(mcp_result, "isError", False) is True
    structured: Optional[Dict] = getattr(mcp_result, "structuredContent", None) or None
    text_chunks: List[str] = []
    for part in getattr(mcp_result, "content", []) or []:
        ptype = getattr(part, "type", None)
        if ptype == "text":
            text_chunks.append(getattr(part, "text", "") or "")
        elif ptype == "image":
            b64 = getattr(part, "data", None)
            if b64:
                images.append(b64)
                mime = getattr(part, "mimeType", None) or ""
                image_mime_types.append(mime)
    if text_chunks:
        joined = "\n".join(t for t in text_chunks if t)
        try:
            data = json.loads(joined) if joined.strip().startswith(("{", "[")) else joined
        except json.JSONDecodeError:
            data = joined
    return {
        "data": data,
        "images": images,
        "image_mime_types": image_mime_types,
        "structuredContent": structured,
        "isError": is_error,
    }


def _image_from_tool_result(out: Dict[str, Any]) -> tuple[Optional[str], Optional[str]]:
    """Pull a (png_b64, mime_type) pair out of a flattened tool result.

    cua-driver delivers window screenshots in two shapes depending on tool +
    transport:

      * As an MCP ``image`` content part — surfaced by ``_extract_tool_result``
        in ``out["images"]`` with a parallel ``image_mime_types`` entry. This
        is what ``get_window_state`` emits over the stdio MCP transport.
      * As a base64 field inside ``structuredContent`` —
        ``screenshot_png_b64`` (+ ``screenshot_mime_type``). This is what
        ``get_window_state`` returns when its structured payload carries the
        image instead of a content part (newer driver builds; also the shape
        seen via the ``cua-driver call`` CLI surface).

    Checking both makes capture() robust to either delivery shape, so the
    image never silently drops just because the driver moved it between the
    content list and structuredContent. Returns ``(None, None)`` when neither
    location carries an image.
    """
    images = out.get("images") or []
    if images and images[0]:
        mimes = out.get("image_mime_types") or []
        mime = mimes[0] if mimes and mimes[0] else None
        return images[0], mime

    structured = out.get("structuredContent") or {}
    b64 = structured.get("screenshot_png_b64") or structured.get("png_b64")
    if b64:
        mime = (
            structured.get("screenshot_mime_type")
            or structured.get("mime_type")
            or None
        )
        return b64, mime

    return None, None


def _positive_int(value: Any) -> Optional[int]:
    """Return a positive integer, rejecting booleans and malformed values."""
    if isinstance(value, bool) or not isinstance(value, (int, str)):
        return None
    try:
        parsed = int(value)
    except ValueError:
        return None
    return parsed if parsed > 0 else None


def _ingest_windows(raw_windows: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    """Normalise cua-driver ``list_windows`` entries, dropping unusable ones.

    Every downstream operation needs both an integer ``pid`` (for
    get_window_state / action tools) and ``window_id`` (for screenshot /
    element clicks), so a window missing either is uncapturable.

    Crucially, on X11 a window's PID comes from the *optional*
    ``_NET_WM_PID`` property — the desktop root, panels, and
    override-redirect popups routinely omit it, so the driver reports
    ``pid: null`` for them. Coercing every entry unconditionally
    (``int(w["pid"])``) let one such window abort enumeration of the real,
    targetable windows. We skip the unusable entries instead so capture()
    and focus_app() still find the windows that matter.

    ``z_index`` follows CUA Driver semantics: higher = closer to front.
    Wayland may return ``z_index: null`` (undefined stacking order); we
    treat null as the lowest priority so real windows still sort above
    desktop/root windows, and the backmost never ends up selected as the
    capture target.
    """
    windows: List[Dict[str, Any]] = []
    for w in raw_windows:
        # Compatibility envelopes are untrusted input: skip non-dict members
        # instead of raising AttributeError on one malformed record.
        if not isinstance(w, dict):
            continue
        pid_int = _positive_int(w.get("pid"))
        window_id_int = _positive_int(w.get("window_id"))
        if pid_int is None or window_id_int is None:
            continue
        z_raw = w.get("z_index")
        z_index = z_raw if isinstance(z_raw, (int, float)) and not isinstance(z_raw, bool) else 0
        app_name = w.get("app_name", "")
        title = w.get("title", "")
        windows.append({
            "app_name": app_name if isinstance(app_name, str) else "",
            "pid": pid_int,
            "window_id": window_id_int,
            # cua-driver 0.6.x on Linux may return JSON null here.
            # Only explicit False means off-screen; null means unknown.
            "off_screen": w.get("is_on_screen") is False,
            "title": title if isinstance(title, str) else "",
            "z_index": z_index,
        })
    return windows


def _windows_from_tool_result(out: Dict[str, Any]) -> List[Dict[str, Any]]:
    """Return list_windows payloads across cua-driver result shapes."""
    structured = out.get("structuredContent")
    if isinstance(structured, dict):
        windows = structured.get("windows")
        if isinstance(windows, list) and windows:
            return windows

    data = out.get("data")
    if isinstance(data, dict):
        windows = data.get("windows")
        if isinstance(windows, list) and windows:
            return windows
        legacy_windows = data.get("_legacy_windows")
        if isinstance(legacy_windows, list) and legacy_windows:
            return legacy_windows

    windows = out.get("windows")
    if isinstance(windows, list) and windows:
        return windows
    legacy_windows = out.get("_legacy_windows")
    if isinstance(legacy_windows, list) and legacy_windows:
        return legacy_windows
    return []


def _apps_from_windows(windows: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    apps: List[Dict[str, Any]] = []
    seen: set[tuple[str, int]] = set()
    for summary in _ingest_windows(windows):
        name = summary["app_name"]
        if not name:
            continue
        key = (name, summary["pid"])
        if key in seen:
            continue
        seen.add(key)
        apps.append({"name": name, "pid": summary["pid"]})
    return apps


# ---------------------------------------------------------------------------
# The backend itself
# ---------------------------------------------------------------------------

class CuaDriverBackend(ComputerUseBackend):
    """Default computer-use backend. Cross-platform via cua-driver MCP."""

    def __init__(self, permission_mode: str = "standard") -> None:
        if permission_mode not in {"standard", "unrestricted"}:
            raise ValueError(f"unsupported cua-driver permission mode: {permission_mode}")
        self.permission_mode = permission_mode
        self._embedded_daemon = (
            _EmbeddedCuaDaemon(resolve_cua_driver_cmd() or "", permission_mode)
            if permission_mode == "unrestricted"
            else None
        )
        self._bridge = _AsyncBridge()
        self._session = _CuaDriverSession(self._bridge, self._embedded_daemon)
        # Sticky context — updated by capture(), used by action tools.
        self._active_pid: Optional[int] = None
        self._active_window_id: Optional[int] = None
        self._last_app: Optional[str] = None  # last app name targeted via capture/focus_app
        # Exact identity for capture_after. App names may be generic on Linux
        # (for example, multiple unrelated Qt windows can say Qt6Application).
        self._last_target: Optional[Dict[str, Optional[int]]] = None
        # Surface 6 of NousResearch/hermes-agent#47072: per-snapshot
        # `element_index -> element_token` map populated on capture().
        # Action tools (click/scroll/set_value/...) attach the matching
        # token alongside `element_index` so cua-driver detects "stale"
        # explicitly instead of silently re-resolving to a different
        # element. Cleared whenever a fresh capture overwrites the
        # snapshot context.
        self._snapshot_tokens: Dict[int, str] = {}
        # Per-instance cua-driver session id. cua-driver's MCP server
        # instructions ask every consumer to declare a stable session
        # at the start of a run (start_session) and tear it down at
        # the end (end_session). Doing so:
        #   - Gets a distinct agent-cursor color per Hermes run, with
        #     overlay rendering visualising where actions land
        #     (without moving the real OS cursor).
        #   - Isolates per-session config + recording ownership so
        #     concurrent Hermes runs / subagents don't step on each
        #     other.
        # We mint a UUID4-based id once per CuaDriverBackend instance —
        # one Hermes run = one backend = one session — and pass it as
        # `session` on every cua-driver tool call. Sessions are an
        # additive feature on the cua-driver side: when our id is
        # unknown to the driver (older builds), the tool calls
        # degrade to the anonymous / unsynced path documented in the
        # MCP server instructions.
        self._session_id: str = f"hermes-{uuid.uuid4().hex[:12]}"
        self._typed_browser = CuaTypedBrowserRoute(
            session_id=self._session_id,
            call_tool=self._session.call_tool,
            has_tool=self._session._has_tool,
        )

    def _browser_route(self) -> CuaTypedBrowserRoute:
        """Return the per-backend typed route, including test-constructed instances."""
        route = getattr(self, "_typed_browser", None)
        if route is None:
            route = CuaTypedBrowserRoute(
                session_id=self._session_id,
                call_tool=self._session.call_tool,
                has_tool=self._session._has_tool,
            )
            self._typed_browser = route
        return route

    # ── Lifecycle ──────────────────────────────────────────────────
    def start(self) -> None:
        _maybe_nudge_update()
        # The MCP client SDK (`mcp`) is an optional dependency (the
        # `computer-use` / `mcp` extras), not part of Hermes' minimal core.
        # Lazy-install it on first use — the same pattern every other optional
        # backend uses — so users never hit an opaque `No module named 'mcp'`
        # at invoke time. Auto-install is gated by `security.allow_lazy_installs`
        # (default on); when it's disabled or fails, ensure() raises
        # FeatureUnavailable carrying an actionable `uv pip install mcp==…`
        # hint, which surfaces via the backend-unavailable path in tool.py.
        from tools.lazy_deps import ensure as _lazy_ensure
        _lazy_ensure("tool.computer_use", prompt=False)
        # A just-installed package may not be importable until the import
        # machinery's caches are refreshed within this process.
        import importlib
        importlib.invalidate_caches()
        try:
            if self._embedded_daemon is not None:
                self._embedded_daemon.start()
            self._session.start()
        except Exception:
            if self._embedded_daemon is not None:
                self._embedded_daemon.stop()
            raise

        # Declare the run's session identity to cua-driver. From the
        # cua-driver server instructions: "start_session(session) once
        # at the start of a run → declares THIS run's identity (a
        # stable id you choose). Pass that same `session` on every
        # action below. It owns your agent cursor (a distinct color
        # per id) and follows the run across apps/windows." Failure
        # to start the session is non-fatal — cua-driver's tools
        # accept anonymous calls (the cursor just won't render),
        # so we degrade rather than abort.
        try:
            self._session.call_tool("start_session", {"session": self._session_id})
        except Exception as e:
            logger.debug("cua-driver start_session failed (continuing anonymous): %s", e)

        # Post-handshake session tuning. Both guard on `_started`: before the
        # handshake flips it, call_tool would re-enter session.start() (see
        # _LIFECYCLE_CALLS) and tests that stub start() would recurse.
        if self._session._started:
            # Cap screenshot size so every later get_window_state / SOM
            # capture pays less over the daemon socket and in the model turn.
            max_dim = _computer_use_max_image_dimension()
            if max_dim:
                try:
                    self.set_config(max_image_dimension=max_dim)
                except Exception as e:
                    logger.debug("cua-driver set_config(max_image_dimension) failed: %s", e)
            # Belt-and-suspenders when --no-overlay is unsupported or ignored:
            # hide the agent cursor overlay via the session API so macOS idle
            # redraw loops cannot keep burning CPU after the first action.
            if _cua_no_overlay():
                try:
                    self.set_agent_cursor_enabled(False, cursor_id=self._session_id)
                except Exception as e:
                    logger.debug("cua-driver set_agent_cursor_enabled failed: %s", e)

    def stop(self) -> None:
        # Tear the cua-driver session down before disconnecting so the
        # driver can clean up per-session state (cursor overlay, recording
        # ownership, config overrides). Best-effort — even if it fails,
        # the connection drop below releases the daemon-side state via
        # the session_end hook cua-driver registers internally.
        if self._session._started:
            try:
                self._session.call_tool("end_session", {"session": self._session_id})
            except Exception as e:
                logger.debug("cua-driver end_session failed (continuing teardown): %s", e)
        try:
            self._session.stop()
        finally:
            try:
                self._bridge.stop()
            finally:
                if self._embedded_daemon is not None:
                    self._embedded_daemon.stop()

    def is_available(self) -> bool:
        # cua-driver runs on macOS, Windows, and Linux. The Linux path is
        # the most recent addition (X11 + Wayland both supported upstream
        # as of mid-2026). Override the platform check at your own risk:
        # other Unix-likes haven't been exercised end-to-end.
        if sys.platform not in ("darwin", "win32", "linux"):
            return False
        return cua_driver_binary_available()

    def _clear_active_target(self) -> None:
        """Forget a capture/focus target so a failed lookup cannot misroute input."""
        self._active_pid = None
        self._active_window_id = None
        self._last_app = None
        self._last_target = None
        self._snapshot_tokens = {}

    def _failed_capture(self, mode: str, message: str = "") -> CaptureResult:
        """Return an empty capture after disarming any prior target context."""
        self._clear_active_target()
        return CaptureResult(
            mode=mode,
            width=0,
            height=0,
            png_b64=None,
            elements=[],
            app="",
            window_title=message,
            png_bytes_len=0,
        )

    def _call_capture_tool(self, name: str, args: Dict[str, Any]) -> Dict[str, Any]:
        """Call a capture-stage tool and disarm state on transport or logical failure."""
        try:
            out = self._session.call_tool(name, args)
        except Exception:
            self._clear_active_target()
            raise
        if out.get("isError") is True:
            message = out.get("data")
            self._clear_active_target()
            raise RuntimeError(
                f"cua-driver {name} failed"
                + (f": {message}" if isinstance(message, str) and message else "")
            )
        return out

    def _load_windows(self) -> List[Dict[str, Any]]:
        """Load normalized visible windows, with the shared CLI recovery path.

        Windows are sorted by ``z_index`` **descending**: CUA Driver
        defines higher values as closer to the front, so the frontmost
        window ends up at index 0 — which is what ``capture()`` and
        ``focus_app()`` pick as the default target.  ``_ingest_windows``
        already normalised null ``z_index`` (Wayland) to 0, so those
        windows sort to the back.
        """
        out = self._call_capture_tool(
            "list_windows",
            {"on_screen_only": True, "session": self._session_id},
        )
        windows = _ingest_windows(_windows_from_tool_result(out))
        windows.sort(key=lambda w: w["z_index"], reverse=True)
        if windows:
            return windows

        logger.warning(
            "cua-driver list_windows returned no windows over MCP; "
            "re-fetching via CLI transport",
        )
        try:
            cli_out = self._session._call_tool_via_cli(
                "list_windows",
                {"on_screen_only": True, "session": self._session_id},
                20.0,
            )
        except Exception as exc:
            logger.error("cua-driver CLI re-fetch for list_windows failed: %s", exc)
            return []
        if cli_out.get("isError") is True:
            logger.error("cua-driver CLI re-fetch for list_windows returned an error")
            self._clear_active_target()
            return []
        windows = _ingest_windows(_windows_from_tool_result(cli_out))
        windows.sort(key=lambda w: w["z_index"], reverse=True)
        return windows

    def _match_windows_for_app(
        self, windows: List[Dict[str, Any]], app: str
    ) -> List[Dict[str, Any]]:
        """Resolve ``app=`` through exact names before convenience substrings.

        Linux ``list_windows`` can omit an app name while ``list_apps`` retains
        name/bundle-ID metadata. Exact direct names and exact metadata aliases
        must win over substring matches: querying ``Code`` must not silently
        select ``Visual Studio Code`` merely because it is frontmost.
        """
        app_lower = app.strip().lower()
        if not app_lower:
            return []

        direct_exact = [
            w for w in windows
            if app_lower == str(w.get("app_name", "")).strip().lower()
        ]
        if direct_exact:
            return direct_exact

        try:
            running_apps = self.list_apps()
        except Exception as exc:
            # A title can still be the only usable identity on X11 when app
            # enumeration is unavailable, so retain the constrained title
            # fallback below instead of treating this as a hard no-match.
            logger.debug("computer_use list_apps fallback failed for %r: %s", app, exc)
            running_apps = []

        exact_pids: set[int] = set()
        partial_pids: set[int] = set()
        for raw_app in running_apps:
            if not isinstance(raw_app, dict) or raw_app.get("running") is False:
                continue
            raw_pid = raw_app.get("pid")
            if isinstance(raw_pid, bool) or not isinstance(raw_pid, (int, str)):
                continue
            try:
                pid = int(raw_pid)
            except ValueError:
                continue
            if pid <= 0:
                continue

            aliases = {
                value.strip().lower()
                for key in ("bundle_id", "bundleId", "name", "app_name", "display_name")
                if isinstance((value := raw_app.get(key)), str) and value.strip()
            }
            if app_lower in aliases:
                exact_pids.add(pid)
            elif any(app_lower in alias for alias in aliases):
                partial_pids.add(pid)

        metadata_exact = [w for w in windows if w.get("pid") in exact_pids]
        if metadata_exact:
            return metadata_exact

        direct_partial = [
            w for w in windows
            if app_lower in str(w.get("app_name", "")).lower()
        ]
        if direct_partial:
            return direct_partial

        metadata_partial = [w for w in windows if w.get("pid") in partial_pids]
        if metadata_partial:
            return metadata_partial

        # Some X11 backends expose a title but no app name. Restrict this final
        # fallback to nameless rows so a localized app name is not overridden
        # merely because its title happens to be in the caller's language.
        return [
            w for w in windows
            if not str(w.get("app_name", "")).strip()
            and app_lower in str(w.get("title", "")).lower()
        ]

    # ── Capture ────────────────────────────────────────────────────
    def capture(
        self,
        mode: str = "som",
        app: Optional[str] = None,
        pid: Optional[int] = None,
        window_id: Optional[int] = None,
    ) -> CaptureResult:
        """Capture the frontmost on-screen window or an exact known target.

        Maps hermes `capture(mode, app)` → cua-driver `list_windows` +
        `get_window_state` (ax/som) or `screenshot` (vision).
        """
        # Step 1: enumerate on-screen windows to find target pid/window_id.
        # Surface 3 of NousResearch/hermes-agent#47072: read the canonical
        # `structuredContent.windows` array directly. Pre-fix the wrapper
        # also kept a text-line regex (`_WINDOW_LINE_RE`) as a fallback for
        # cua-driver builds that predated structuredContent; the supersede
        # PR's effective minimum (trycua/cua#1961 + #1908) is well past
        # that, so the fallback is gone — the wrapper now treats the
        # structured shape as the only contract.
        # An exact pid/window pair is both the stable capture_after target and
        # the escape hatch when app/window discovery is unavailable on X11.
        if pid is not None or window_id is not None:
            if pid is None or window_id is None:
                return self._failed_capture(
                    mode, "<capture targeting requires both pid and window_id>",
                )
            target_pid = _positive_int(pid)
            target_window_id = _positive_int(window_id)
            if target_pid is None or target_window_id is None:
                return self._failed_capture(
                    mode, "<capture targeting requires positive integer pid and window_id>",
                )
            windows = [{
                "app_name": app or "",
                "pid": target_pid,
                "window_id": target_window_id,
                "off_screen": False,
                "title": "",
                "z_index": 0,
            }]
        else:
            try:
                windows = self._load_windows()
            except Exception:
                self._clear_active_target()
                raise
            if not windows:
                return self._failed_capture(mode)

        # Filter by app name (case-insensitive substring) if requested.
        # When the filter matches nothing, surface that explicitly instead of
        # silently capturing the frontmost window — on macOS the `app_name`
        # returned by list_windows is the localized name (e.g. "計算機"), so
        # `app="Calculator"` legitimately matches no windows on a non-English
        # system and the caller needs to retry with the localized name.
        if pid is None and window_id is None and app and app.strip().lower() in _SCREEN_CAPTURE_SENTINELS:
            # Whole-screen / desktop request. cua-driver has no virtual-desktop
            # capture tool, so resolve to the OS shell/desktop window (the
            # desktop backdrop or the taskbar/menu-bar), which list_windows
            # does surface. This makes "show me my screen" and "click the
            # taskbar" work; a single image still can't span multiple monitors
            # — that's a driver limitation, not a wrapper one.
            def _is_desktop_window(w: Dict[str, Any]) -> bool:
                haystack = f"{w.get('app_name', '')} {w.get('title', '')}".lower()
                return any(name in haystack for name in _DESKTOP_WINDOW_NAMES)

            desktop = [w for w in windows if _is_desktop_window(w)]
            if not desktop:
                return self._failed_capture(
                    mode,
                    (
                        f"<no desktop/shell window found for app={app!r}; "
                        f"cua-driver captures one window at a time and exposes "
                        f"no whole-virtual-desktop or per-monitor capture. "
                        f"Call list_apps / capture(app='<AppName>') to target a "
                        f"specific window instead. On Windows the taskbar is "
                        f"'Shell_TrayWnd' and the desktop is 'Progman'.>"
                    ),
                )
            # Prefer the desktop backdrop (Progman/WorkerW/Finder) over the
            # taskbar when both are present, so a bare "screen" capture shows
            # the full desktop rather than just the task strip.
            windows = sorted(
                desktop,
                key=lambda w: 0 if any(
                    n in f"{w.get('app_name', '')} {w.get('title', '')}".lower()
                    for n in ("progman", "workerw", "program manager", "finder", "desktop")
                ) else 1,
            )
        elif pid is None and window_id is None and app:
            filtered = self._match_windows_for_app(windows, app)
            if not filtered:
                return self._failed_capture(
                    mode,
                    (
                        f"<no on-screen window matched app={app!r}; "
                        f"call list_apps to see available app names or bundle IDs "
                        f"(macOS reports localized names, e.g. '計算機' "
                        f"instead of 'Calculator'; some Linux/Qt apps only "
                        f"resolve via list_apps metadata)>"
                    ),
                )
            windows = filtered

        # Pick first on-screen window (sorted by z_index / z-order above).
        # On Linux, unqualified default captures skip desktop/shell helper
        # windows and, with tied/unknown z_index, may additionally consult
        # _NET_ACTIVE_WINDOW (#58026).
        target = _select_capture_target(
            windows,
            app_requested=bool(app),
            exact_target=pid is not None or window_id is not None,
        )
        self._active_pid = target["pid"]
        self._active_window_id = target["window_id"]
        # Tokens belong to the prior window snapshot. Disarm them before any
        # capture call so an exception cannot pair old tokens with this target.
        self._snapshot_tokens = {}
        app_name = target["app_name"]
        # Record the resolved app name so capture_after= follow-ups can re-target
        # the same app rather than falling back to the frontmost window.
        if app or not self._last_app:
            self._last_app = app_name or app or ""
        self._last_target = {
            "pid": self._active_pid,
            "window_id": self._active_window_id,
        }

        # Step 2: capture.
        png_b64: Optional[str] = None
        image_mime_type: Optional[str] = None
        elements: List[UIElement] = []
        width = height = 0
        window_title = ""

        if mode == "vision":
            # Plain screenshot, no AX walk. cua-driver dropped the standalone
            # `screenshot` tool (≥0.5.x) and folded full-window PNG capture
            # into `get_window_state`. Route accordingly:
            #   * Driver advertises `screenshot` (older builds) → use it; it's
            #     the cheapest path (no AX tree walked server-side).
            #   * Otherwise (current drivers) → call `get_window_state` but
            #     DISCARD the AX tree/elements, returning only the PNG. Vision
            #     mode's whole contract is "just the pixels, no element noise",
            #     so we drop everything but the image.
            # When capability discovery hasn't run (empty map), we don't trust
            # a negative `_has_tool` answer — we still try `screenshot` first
            # and fall back if the driver rejects it, so the path self-heals on
            # any driver version.
            use_screenshot = (
                self._session._has_tool("screenshot")
                or not self._session.capabilities_discovered
            )
            sc_out: Optional[Dict[str, Any]] = None
            if use_screenshot:
                sc_out = self._call_capture_tool(
                    "screenshot",
                    {
                        "window_id": self._active_window_id,
                        "format": "jpeg",
                        "quality": 85,
                        "session": self._session_id,
                    },
                )
                png_b64, image_mime_type = _image_from_tool_result(sc_out)
                if not png_b64:
                    # Driver had no usable `screenshot` (e.g. "Unknown tool:
                    # screenshot" on ≥0.5.x, or an empty image part). Fall
                    # through to the get_window_state path below.
                    sc_out = None

            if sc_out is None:
                gws_out = self._call_capture_tool(
                    "get_window_state",
                    {
                        "pid": self._active_pid,
                        "window_id": self._active_window_id,
                        "session": self._session_id,
                    },
                )
                png_b64, image_mime_type = _image_from_tool_result(gws_out)
                # Still grab the window title — it's cheap and useful in the
                # vision response — but deliberately leave `elements` empty so
                # vision stays free of AX-tree noise.
                text = gws_out["data"] if isinstance(gws_out["data"], str) else ""
                _, tree = _split_tree_text(text)
                wt = re.search(r'AXWindow\s+"([^"]+)"', tree)
                if wt:
                    window_title = wt.group(1)

            if not png_b64:
                # Both MCP attempts came back imageless without raising (flaky
                # bridge dropping the heavy payload) — re-fetch the window
                # state over the CLI transport, which embeds a screenshot.
                logger.warning(
                    "cua-driver vision capture returned no image over MCP "
                    "(window_id=%s); re-fetching via CLI transport",
                    self._active_window_id,
                )
                try:
                    cli_out = self._session._call_tool_via_cli(
                        "get_window_state",
                        {
                            "pid": self._active_pid,
                            "window_id": self._active_window_id,
                            "session": self._session_id,
                        },
                        30.0,
                    )
                    if cli_out.get("isError") is True:
                        self._clear_active_target()
                    elif cli_out.get("images"):
                        png_b64 = cli_out["images"][0]
                        image_mime_type = "image/png"
                except Exception as cli_exc:
                    logger.error(
                        "cua-driver CLI re-fetch for vision screenshot failed: %s", cli_exc,
                    )
        else:
            # get_window_state: AX tree + screenshot.
            gws_out = self._call_capture_tool(
                "get_window_state",
                {
                    "pid": self._active_pid,
                    "window_id": self._active_window_id,
                    "session": self._session_id,
                },
            )
            # The persistent MCP session can return a degenerate result —
            # empty/partial data with NO exception — when the bridge is flaky
            # (e.g. it reconnected mid-call and dropped the heavy
            # get_window_state payload). That surfaces to the model as a silent
            # 0x0 capture. Detect "no screenshot AND no parseable tree" and
            # force a one-shot CLI-transport re-fetch, which talks to the daemon
            # over a different socket and returns the full result. This is
            # distinct from the EAGAIN McpError path (handled in call_tool);
            # here the MCP call "succeeded" but gave us nothing usable.
            def _gws_is_empty(out: Dict[str, Any]) -> bool:
                if out.get("images"):
                    return False
                sc_ = out.get("structuredContent") or {}
                # Modern drivers carry the payload in structuredContent
                # (elements array / embedded screenshot) with no markdown
                # tree — that is NOT an empty result.
                if sc_.get("elements") or sc_.get("screenshot_png_b64"):
                    return False
                txt = out.get("data") if isinstance(out.get("data"), str) else ""
                _, tr = _split_tree_text(txt or "")
                return not (tr and tr.strip())

            if _gws_is_empty(gws_out):
                logger.warning(
                    "cua-driver get_window_state returned an empty result over MCP "
                    "(pid=%s window_id=%s); re-fetching via CLI transport",
                    self._active_pid, self._active_window_id,
                )
                try:
                    cli_out = self._session._call_tool_via_cli(
                        "get_window_state",
                        {
                            "pid": self._active_pid,
                            "window_id": self._active_window_id,
                            "session": self._session_id,
                        },
                        30.0,
                    )
                    if cli_out.get("isError") is True:
                        self._clear_active_target()
                    elif not _gws_is_empty(cli_out):
                        gws_out = cli_out
                except Exception as cli_exc:
                    logger.error(
                        "cua-driver CLI re-fetch for get_window_state failed: %s", cli_exc,
                    )

            text = gws_out["data"] if isinstance(gws_out["data"], str) else ""
            summary, tree = _split_tree_text(text)

            # Surface 2 of NousResearch/hermes-agent#47072: prefer the
            # canonical structuredContent.elements array (trycua/cua#1961).
            # Falls back to markdown regex parsing for cua-driver builds
            # that didn't carry the structured shape — those bounds come
            # back (0,0,0,0); the structured path preserves real frames.
            sc_elements = (gws_out.get("structuredContent") or {}).get("elements")
            if isinstance(sc_elements, list) and sc_elements:
                elements = _parse_elements_from_structured(sc_elements)
            else:
                elements = _parse_elements_from_tree(tree) if tree else []

            # Surface 6: refresh the snapshot-token cache from this
            # capture. Tokens are tied to a specific cua-driver snapshot
            # — when a fresh capture lands, the prior snapshot's tokens
            # are stale, so we overwrite the whole map (and clear it
            # entirely when the new capture carries none).
            self._snapshot_tokens = {
                e.index: e.element_token
                for e in elements
                if e.element_token
            }

            # Image may arrive as an MCP image part or inside
            # structuredContent (screenshot_png_b64) depending on the driver
            # build — _image_from_tool_result handles both.
            png_b64, image_mime_type = _image_from_tool_result(gws_out)

            # Extract window title from the AX tree first AXWindow line.
            wt = re.search(r'AXWindow\s+"([^"]+)"', tree)
            if wt:
                window_title = wt.group(1)

        png_bytes_len = 0
        if png_b64:
            try:
                raw = base64.b64decode(png_b64, validate=False)
                png_bytes_len = len(raw)
                detected_width, detected_height = _image_dimensions_from_bytes(raw)
                if detected_width and detected_height:
                    width = detected_width
                    height = detected_height
            except Exception:
                png_bytes_len = len(png_b64) * 3 // 4

        return CaptureResult(
            mode=mode,
            width=width,
            height=height,
            png_b64=png_b64,
            elements=elements,
            app=app_name,
            window_title=window_title,
            png_bytes_len=png_bytes_len,
            image_mime_type=image_mime_type,
        )

    # ── Pointer ────────────────────────────────────────────────────
    def _apply_delivery(
        self,
        action: str,
        args: Dict[str, Any],
        delivery_mode: Optional[str],
    ) -> Optional[ActionResult]:
        """Attach delivery_mode to an input-action args dict.

        Background is the default and never needs a flag. Foreground is only
        sent when the live action schema accepts it; on an older driver that
        lacks the property we refuse with a structured
        ``foreground_unsupported`` result instead of silently downgrading to
        background (which would land the input somewhere the model didn't
        expect). Returns an ActionResult to short-circuit on refusal, or None
        to proceed. See NousResearch/hermes-agent#67052 phase B.
        """
        if not delivery_mode or delivery_mode == "background":
            return None
        if delivery_mode != "foreground":
            return ActionResult(
                ok=False, action=action, code="bad_delivery_mode",
                message=f"unknown delivery_mode {delivery_mode!r} — use background|foreground.",
            )
        # Foreground requested. Only send it if the driver understands it.
        if not self._session.supports_input_property(action, "delivery_mode"):
            return ActionResult(
                ok=False, action=action, code="foreground_unsupported",
                delivery_mode="foreground",
                message=(
                    "The connected cua-driver action schema does not accept "
                    "delivery_mode, so foreground delivery is unavailable. "
                    "Use another verified rung without assuming the reported "
                    "package version describes the live schema."
                ),
            )
        args["delivery_mode"] = "foreground"
        return None

    def _run_input_action(
        self,
        action: str,
        args: Dict[str, Any],
        delivery_mode: Optional[str],
        bring_to_front: bool,
    ) -> ActionResult:
        """Apply one delivery rung, optionally focusing via its own tool.

        ``bring_to_front`` is never an input-action property.  When explicitly
        requested, the separately approved standalone focus action runs first,
        then the original foreground input runs unchanged.
        """
        refusal = self._apply_delivery(action, args, delivery_mode)
        if refusal is not None:
            return refusal
        if bring_to_front:
            if delivery_mode != "foreground":
                return ActionResult(
                    ok=False,
                    action=action,
                    code="bring_to_front_requires_foreground",
                    message="bring_to_front requires delivery_mode='foreground'.",
                )
            if not self._session._has_tool("bring_to_front"):
                return ActionResult(
                    ok=False,
                    action=action,
                    code="bring_to_front_unsupported",
                    delivery_mode="foreground",
                    message="The connected cua-driver does not advertise the standalone bring_to_front tool.",
                )
            if self._active_pid is None or self._active_window_id is None:
                return ActionResult(
                    ok=False,
                    action=action,
                    code="bring_to_front_target_required",
                    delivery_mode="foreground",
                    message="Capture an exact target before requesting persistent foreground focus.",
                )
            focused = self.bring_to_front(
                pid=self._active_pid,
                window_id=self._active_window_id,
            )
            if not focused.ok:
                return focused
        result = self._action(action, args)
        if bring_to_front:
            result.meta["foreground_focus"] = {
                "invoked": True,
                "tool": "bring_to_front",
            }
        return result

    def click(
        self,
        *,
        element: Optional[int] = None,
        x: Optional[int] = None,
        y: Optional[int] = None,
        button: str = "left",
        click_count: int = 1,
        modifiers: Optional[List[str]] = None,
        delivery_mode: Optional[str] = None,
        bring_to_front: bool = False,
    ) -> ActionResult:
        pid = self._active_pid
        if pid is None:
            return ActionResult(ok=False, action="click",
                                message="No active window — call capture() first.")

        # Choose tool by click_count only — single-vs-double — and pass the
        # button through to `click`'s `button` enum (Surface 5 of
        # NousResearch/hermes-agent#47072). cua-driver-rs gained an explicit
        # `button: "left"|"right"|"middle"` arg on `click` in trycua/cua#1961
        # which rejects unknown buttons; before that, `middle` was silently
        # mapped to a left-click via name-routing through `right_click`.
        # `right_click`/`middle_click` MCP tools are deprecated aliases —
        # kept around but no longer invoked from here.
        button_norm = (button or "left").lower()
        if button_norm not in {"left", "right", "middle"}:
            return ActionResult(ok=False, action="click",
                                message=f"unknown button {button!r} — expected left, right, middle.")
        tool = "double_click" if click_count == 2 else "click"

        args: Dict[str, Any] = {"pid": pid, "button": button_norm}
        if element is not None:
            if self._active_window_id is None:
                return ActionResult(ok=False, action=tool,
                                    message="No active window_id for element_index click.")
            args["element_index"] = element
            args["window_id"] = self._active_window_id
        elif x is not None and y is not None:
            if self._active_window_id is None:
                return ActionResult(ok=False, action=tool,
                                    message="No active window_id for coordinate click.")
            args["x"] = x
            args["y"] = y
            args["window_id"] = self._active_window_id
        else:
            return ActionResult(ok=False, action=tool,
                                message="click requires element= or x/y.")
        if modifiers:
            args["modifier"] = modifiers

        return self._run_input_action(tool, args, delivery_mode, bring_to_front)

    def drag(
        self,
        *,
        from_element: Optional[int] = None,
        to_element: Optional[int] = None,
        from_xy: Optional[Tuple[int, int]] = None,
        to_xy: Optional[Tuple[int, int]] = None,
        button: str = "left",
        modifiers: Optional[List[str]] = None,
        delivery_mode: Optional[str] = None,
        bring_to_front: bool = False,
    ) -> ActionResult:
        pid = self._active_pid
        if pid is None:
            return ActionResult(ok=False, action="drag",
                                message="No active window — call capture() first.")
        args: Dict[str, Any] = {"pid": pid}
        if from_element is not None and to_element is not None:
            if self._active_window_id is None:
                return ActionResult(ok=False, action="drag",
                                    message="No active window_id for element-based drag.")
            args["from_element"] = from_element
            args["to_element"] = to_element
            args["window_id"] = self._active_window_id
        elif from_xy is not None and to_xy is not None:
            if self._active_window_id is None:
                return ActionResult(ok=False, action="drag",
                                    message="No active window_id for coordinate drag.")
            args["from_x"], args["from_y"] = int(from_xy[0]), int(from_xy[1])
            args["to_x"], args["to_y"] = int(to_xy[0]), int(to_xy[1])
            args["window_id"] = self._active_window_id
        else:
            return ActionResult(ok=False, action="drag",
                                message="drag requires from_element/to_element or from_coordinate/to_coordinate.")
        return self._run_input_action("drag", args, delivery_mode, bring_to_front)

    def scroll(
        self,
        *,
        direction: str,
        amount: int = 3,
        element: Optional[int] = None,
        x: Optional[int] = None,
        y: Optional[int] = None,
        modifiers: Optional[List[str]] = None,
        delivery_mode: Optional[str] = None,
        bring_to_front: bool = False,
    ) -> ActionResult:
        pid = self._active_pid
        if pid is None:
            return ActionResult(ok=False, action="scroll",
                                message="No active window — call capture() first.")
        args: Dict[str, Any] = {
            "pid": pid,
            "direction": direction,
            "amount": max(1, min(50, amount)),
        }
        if element is not None and self._active_window_id is not None:
            args["element_index"] = element
            args["window_id"] = self._active_window_id
        elif x is not None and y is not None:
            if self._active_window_id is None:
                return ActionResult(ok=False, action="scroll",
                                    message="No active window_id for coordinate scroll.")
            # CUA Driver 0.7.1 Linux schema rejects x/y on scroll. Only
            # include them when the driver explicitly advertises support
            # for coordinate scrolling; otherwise omit and let the driver
            # scroll the targeted window (window_id is still sent for
            # routing).  This is the safe default when capabilities
            # haven't been discovered yet (older drivers).
            if self._session.supports_capability(
                "input.scroll.coordinates", tool="scroll"
            ):
                args["x"] = x
                args["y"] = y
            args["window_id"] = self._active_window_id
        return self._run_input_action("scroll", args, delivery_mode, bring_to_front)

    # ── Keyboard ───────────────────────────────────────────────────
    def type_text(self, text: str, *, delivery_mode: Optional[str] = None,
                  bring_to_front: bool = False) -> ActionResult:
        pid = self._active_pid
        window_id = self._active_window_id
        if pid is None or window_id is None:
            return ActionResult(ok=False, action="type_text",
                                message="No active window — call capture() first.")
        args: Dict[str, Any] = {"pid": pid, "window_id": window_id, "text": text}
        return self._run_input_action("type_text", args, delivery_mode, bring_to_front)

    def key(self, keys: str, *, delivery_mode: Optional[str] = None,
            bring_to_front: bool = False) -> ActionResult:
        pid = self._active_pid
        window_id = self._active_window_id
        if pid is None or window_id is None:
            return ActionResult(ok=False, action="key",
                                message="No active window — call capture() first.")

        key_name, modifiers = _parse_key_combo(keys)
        if not key_name:
            return ActionResult(ok=False, action="key",
                                message=f"Could not parse key from '{keys}'.")

        if modifiers:
            # hotkey requires at least one modifier + one key.
            args: Dict[str, Any] = {"pid": pid, "window_id": window_id,
                                    "keys": modifiers + [key_name]}
            return self._run_input_action("hotkey", args, delivery_mode, bring_to_front)
        else:
            args = {"pid": pid, "window_id": window_id, "key": key_name}
            return self._run_input_action("press_key", args, delivery_mode, bring_to_front)

    # ── Value setter ────────────────────────────────────────────────
    def set_value(self, value: str, element: Optional[int] = None) -> ActionResult:
        """Set a value on an element. Handles AXPopUpButton selects natively."""
        pid = self._active_pid
        window_id = self._active_window_id
        if pid is None or window_id is None:
            return ActionResult(ok=False, action="set_value",
                                message="No active window — call capture() first.")
        if element is None:
            return ActionResult(ok=False, action="set_value",
                                message="set_value requires element= (element index).")
        args: Dict[str, Any] = {
            "pid": pid,
            "window_id": window_id,
            "element_index": element,
            "value": value,
        }
        return self._action("set_value", args)

    # ── Introspection ──────────────────────────────────────────────
    def list_apps(self) -> List[Dict[str, Any]]:
        out = self._session.call_tool("list_apps", {"session": self._session_id})
        structured = out.get("structuredContent")
        data = out.get("data")

        # structuredContent is the canonical MCP payload. Empty lists fall
        # through so a populated compatibility envelope can still recover.
        if isinstance(structured, dict):
            apps = structured.get("apps")
            if isinstance(apps, list) and apps:
                return apps
        # Older drivers and direct CLI fallbacks may put apps in data instead.
        if isinstance(data, list) and data:
            return data
        if isinstance(data, dict):
            apps = data.get("apps")
            if isinstance(apps, list) and apps:
                return apps
        apps = out.get("apps")
        if isinstance(apps, list) and apps:
            return apps

        derived = _apps_from_windows(_windows_from_tool_result(out))
        if derived:
            return derived

        # Old text-only drivers retain a small, name/PID-only fallback.
        if isinstance(data, str):
            parsed_apps = []
            for line in data.splitlines():
                m = re.search(r'(.+?)\s+\(pid\s+(\d+)\)', line)
                if m:
                    parsed_apps.append({"name": m.group(1).strip(), "pid": int(m.group(2))})
            return parsed_apps
        return []

    def list_windows(self) -> List[Dict[str, Any]]:
        return self._load_windows()

    def focus_app(self, app: str, raise_window: bool = False) -> ActionResult:
        """Target an app, optionally invoking standalone foreground focus.

        cua-driver background-automation never needs to bring a window to the
        front: capture(app=...) already selects the right window via
        list_windows. We implement focus_app as a pure window-selector —
        enumerate on-screen windows, find the best match for *app*, and store
        its pid/window_id so that subsequent click/type calls hit the right
        process.

        The default remains non-disruptive. ``raise_window=True`` is explicit,
        separately approved by the Hermes adapter, and uses cua-driver's
        standalone ``bring_to_front`` tool rather than an action property.
        """
        try:
            windows = self._load_windows()
        except Exception:
            self._clear_active_target()
            raise

        matched = self._match_windows_for_app(windows, app)
        # Don't silently fall back to the frontmost window when the filter
        # matches nothing — that hides the real failure (often a localized
        # macOS app name mismatch, e.g. caller passed "Calculator" but
        # list_windows returns "計算機").
        target = matched[0] if matched else None
        if target:
            self._active_pid = target["pid"]
            self._active_window_id = target["window_id"]
            self._snapshot_tokens = {}
            self._last_app = target["app_name"] or app  # retained for back-compat diagnostics
            self._last_target = {
                "pid": self._active_pid,
                "window_id": self._active_window_id,
            }
            if raise_window:
                if not self._session._has_tool("bring_to_front"):
                    return ActionResult(
                        ok=False,
                        action="focus_app",
                        code="bring_to_front_unsupported",
                        message="The connected cua-driver does not advertise the standalone bring_to_front tool.",
                    )
                focused = self.bring_to_front(
                    pid=self._active_pid,
                    window_id=self._active_window_id,
                )
                if not focused.ok:
                    return focused
                focused.action = "focus_app"
                focused.meta["target_selected"] = True
                return focused
            return ActionResult(
                ok=True, action="focus_app",
                message=f"Targeted {target['app_name']} (pid {self._active_pid}, "
                        f"window {self._active_window_id}) without raising window.",
            )
        self._clear_active_target()
        return ActionResult(ok=False, action="focus_app",
                            message=f"No on-screen window found for app '{app}'.")

    # ── App lifecycle ────────────────────────────────────────────────
    #
    # cua-driver exposes launch_app / kill_app / bring_to_front as a
    # complete set. focus_app() above is a *window-selector* (no
    # process state change); these methods drive the process layer.

    def launch_app(
        self,
        *,
        bundle_id: Optional[str] = None,
        name: Optional[str] = None,
        urls: Optional[List[str]] = None,
        additional_arguments: Optional[List[str]] = None,
        creates_new_application_instance: bool = False,
    ) -> Dict[str, Any]:
        """Idempotent launch. Returns ``{pid, bundle_id, name, windows[]}``
        so callers can skip an extra ``list_windows`` round-trip before
        ``get_window_state``.

        ``creates_new_application_instance=True`` forces a new instance
        even if the app is already running — use it when concurrent
        runs may touch the same app so each session gets its own
        isolated window."""
        if not bundle_id and not name:
            raise ValueError("launch_app requires either bundle_id or name")
        args: Dict[str, Any] = {"session": self._session_id}
        if bundle_id:
            args["bundle_id"] = bundle_id
        if name:
            args["name"] = name
        if urls:
            args["urls"] = list(urls)
        if additional_arguments:
            args["additional_arguments"] = list(additional_arguments)
        if creates_new_application_instance:
            args["creates_new_application_instance"] = True
        out = self._session.call_tool("launch_app", args)
        return out["structuredContent"] or {"data": out["data"]}

    def kill_app(self, *, pid: int) -> ActionResult:
        """Terminate by pid. Equivalent to ``kill -9`` on POSIX,
        ``taskkill /F`` on Windows."""
        return self._action("kill_app", {"pid": int(pid)})

    def bring_to_front(self, *, pid: int,
                       window_id: Optional[int] = None) -> ActionResult:
        """Activate a window so subsequent foreground-dispatched input
        lands on it. cua-driver's docstring notes this is the cheaper
        path than per-call SetForegroundWindow flashes."""
        args: Dict[str, Any] = {"pid": int(pid)}
        if window_id is not None:
            args["window_id"] = int(window_id)
        # The live 0.9-era schema is strict and deliberately has no session
        # property. It is a standalone native focus operation, not a
        # session-scoped input action.
        return self._action("bring_to_front", args, inject_session=False)

    # ── Typed browser (cua-driver 0.9 contract) ───────────────────
    def typed_browser_state(self, **kwargs: Any) -> Dict[str, Any]:
        """Exact-bind a native browser window or read fresh semantic state."""
        return self._browser_route().observe(**kwargs)

    def typed_browser_prepare(self, **kwargs: Any) -> Dict[str, Any]:
        """Prepare an explicitly approved driver-owned browser profile."""
        return self._browser_route().prepare(**kwargs)

    def typed_browser_action(
        self,
        driver_tool: str,
        *,
        tab_id: Optional[str] = None,
        args: Optional[Dict[str, Any]] = None,
    ) -> Dict[str, Any]:
        """Run one namespaced typed-browser mutation in this exact route."""
        return self._browser_route().mutate(driver_tool, tab_id=tab_id, args=args)

    # ── Pointer + display introspection ─────────────────────────────

    def move_cursor(self, x: int, y: int) -> ActionResult:
        """Move the agent-cursor *overlay* to a screen point. This is a
        visual hint — it does NOT move the real OS pointer (cua-driver
        explicitly avoids stealing pointer focus). The overlay glides
        smoothly to the target, so consumers use it before a click to
        give a visible "where the agent is going" cue."""
        return self._action("move_cursor", {"x": int(x), "y": int(y)})

    def get_cursor_position(self) -> Tuple[int, int]:
        """Return the *real* OS cursor position in screen points
        (origin top-left)."""
        out = self._session.call_tool(
            "get_cursor_position", {"session": self._session_id}
        )
        sc = out.get("structuredContent") or {}
        return int(sc.get("x", 0)), int(sc.get("y", 0))

    def get_screen_size(self) -> Dict[str, Any]:
        """Return the logical size of the main display in points plus
        its backing scale factor. Shape:
        ``{width, height, backing_scale_factor}``."""
        out = self._session.call_tool(
            "get_screen_size", {"session": self._session_id}
        )
        return out.get("structuredContent") or {}

    def zoom(self, *, window_id: int, x: float, y: float, w: float, h: float,
             factor: float = 1.0, format: str = "jpeg",
             quality: int = 85) -> Dict[str, Any]:
        """Return a JPEG / PNG of a sub-region of a window, optionally
        scaled. cua-driver supports zoom-to-rect for callers that need
        a higher-resolution view of a specific element."""
        return self._session.call_tool("zoom", {
            "window_id": int(window_id),
            "x": float(x), "y": float(y), "w": float(w), "h": float(h),
            "factor": float(factor),
            "format": format, "quality": int(quality),
            "session": self._session_id,
        })

    # ── Agent cursor (overlay) ──────────────────────────────────────
    #
    # Sessions (start_session/end_session, wired in start/stop) own the
    # cursor. These knobs tune its appearance + behavior per-session.
    # All accept an optional `cursor_id` to address a specific cursor
    # when the run drives multiple (rare); the default is this run's
    # session id.

    def set_agent_cursor_enabled(self, enabled: bool, *,
                                 cursor_id: Optional[str] = None) -> ActionResult:
        """Toggle the agent cursor overlay's visibility for this run."""
        args: Dict[str, Any] = {"enabled": bool(enabled)}
        if cursor_id:
            args["cursor_id"] = cursor_id
        return self._action("set_agent_cursor_enabled", args)

    def set_agent_cursor_motion(self, *,
                                glide_ms: Optional[float] = None,
                                dwell_ms: Optional[float] = None,
                                idle_hide_ms: Optional[float] = None,
                                cursor_id: Optional[str] = None) -> ActionResult:
        """Tune the overlay's motion timings — glide duration, post-click
        dwell, idle-hide delay. Each None means "leave at current value"."""
        args: Dict[str, Any] = {}
        if glide_ms is not None:
            args["glide_ms"] = float(glide_ms)
        if dwell_ms is not None:
            args["dwell_ms"] = float(dwell_ms)
        if idle_hide_ms is not None:
            args["idle_hide_ms"] = float(idle_hide_ms)
        if cursor_id:
            args["cursor_id"] = cursor_id
        return self._action("set_agent_cursor_motion", args)

    def set_agent_cursor_style(self, *,
                               gradient_colors: Optional[List[str]] = None,
                               bloom_color: Optional[str] = None,
                               image_path: Optional[str] = None,
                               cursor_id: Optional[str] = None) -> ActionResult:
        """Customise the cursor body. ``gradient_colors`` are CSS hex
        strings tip→tail; ``bloom_color`` is the radial halo; an
        ``image_path`` (.svg/.png/.ico) replaces the silhouette
        entirely. Empty values revert to the palette default."""
        args: Dict[str, Any] = {}
        if gradient_colors is not None:
            args["gradient_colors"] = list(gradient_colors)
        if bloom_color is not None:
            args["bloom_color"] = bloom_color
        if image_path is not None:
            args["image_path"] = image_path
        if cursor_id:
            args["cursor_id"] = cursor_id
        return self._action("set_agent_cursor_style", args)

    def get_agent_cursor_state(self, *,
                               cursor_id: Optional[str] = None) -> Dict[str, Any]:
        """Return ``{x, y, config: {cursor_color, cursor_icon, ...},
        enabled}`` for this run's cursor (or the named ``cursor_id``)."""
        args: Dict[str, Any] = {"session": self._session_id}
        if cursor_id:
            args["cursor_id"] = cursor_id
        out = self._session.call_tool("get_agent_cursor_state", args)
        return out.get("structuredContent") or {}

    # ── Recording / replay ──────────────────────────────────────────

    def start_recording(self, *, output_dir: str,
                        record_video: bool = False) -> Dict[str, Any]:
        """Enable trajectory recording (per-turn screenshots + action
        JSON) to ``output_dir``. ``record_video=True`` ALSO captures
        the main display to ``<output_dir>/recording.mp4`` (H.264).
        Recording ownership is keyed by this run's session id so
        concurrent runs don't fight over the recorder."""
        out = self._session.call_tool("start_recording", {
            "output_dir": output_dir,
            "record_video": bool(record_video),
            "session": self._session_id,
        })
        return out.get("structuredContent") or {}

    def stop_recording(self) -> Dict[str, Any]:
        """Disable recording and finalise the mp4 (if video was on).
        Returns the recorder's final state including ``last_video_path``."""
        out = self._session.call_tool("stop_recording", {
            "session": self._session_id,
        })
        return out.get("structuredContent") or {}

    def get_recording_state(self) -> Dict[str, Any]:
        """Return the current recorder state without changing it.
        Shape: ``{recording, enabled, output_dir, next_turn,
        last_video_path, last_error, owner, video_active}``."""
        out = self._session.call_tool(
            "get_recording_state", {"session": self._session_id}
        )
        return out.get("structuredContent") or {}

    def replay_trajectory(self, *, trajectory_dir: str,
                          dry_run: bool = False,
                          speed_factor: float = 1.0) -> Dict[str, Any]:
        """Replay a prior recording's turn stream by re-invoking each
        turn's tool call in lexical order. ``dry_run=True`` logs without
        actually firing the tools."""
        return self._session.call_tool("replay_trajectory", {
            "trajectory_dir": trajectory_dir,
            "dry_run": bool(dry_run),
            "speed_factor": float(speed_factor),
            "session": self._session_id,
        })

    def install_ffmpeg(self) -> Dict[str, Any]:
        """Bootstrap ffmpeg for ``start_recording(record_video=True)``
        on Linux / Windows. macOS records natively via ScreenCaptureKit
        and doesn't need ffmpeg."""
        return self._session.call_tool(
            "install_ffmpeg", {"session": self._session_id}
        )

    # ── Config ──────────────────────────────────────────────────────

    def get_config(self) -> Dict[str, Any]:
        """Return the current cua-driver runtime config."""
        out = self._session.call_tool(
            "get_config", {"session": self._session_id}
        )
        return out.get("structuredContent") or {}

    def set_config(self, **config) -> ActionResult:
        """Set cua-driver config keys. Common keys include
        ``max_image_dimension`` (image-output resizing), recording
        flags, etc. Unknown keys are passed through verbatim — cua-driver
        validates against its own schema."""
        return self._action("set_config", dict(config))

    # ── Lower-level introspection ───────────────────────────────────

    def get_accessibility_tree(self) -> Dict[str, Any]:
        """Return a lightweight snapshot of running regular apps +
        on-screen visible windows with bounds, z-order, owner pid.
        Roughly the data ``list_windows`` exposes, in one call. Most
        callers should prefer ``capture()`` / ``focus_app()`` which
        already use this shape internally."""
        out = self._session.call_tool(
            "get_accessibility_tree", {"session": self._session_id}
        )
        return out.get("structuredContent") or {"data": out["data"]}

    # ── Browser page tool ───────────────────────────────────────────

    def page(self, *, pid: int, action: str,
             **page_args: Any) -> Dict[str, Any]:
        """Interact with a browser page loaded in a running app (Chrome,
        Safari, Edge, ...). cua-driver routes through CDP / Apple Events
        / AX tree depending on the target. ``action`` + ``page_args``
        shape depends on the requested operation (e.g. ``action="eval"``
        takes ``js: str``); see cua-driver's ``page`` tool description
        for the full grammar."""
        args: Dict[str, Any] = {
            "pid": int(pid),
            "action": action,
            "session": self._session_id,
        }
        args.update(page_args)
        return self._session.call_tool("page", args)

    # ── Generic escape hatch ────────────────────────────────────────

    def call_tool(self, name: str, args: Optional[Dict[str, Any]] = None,
                  *, timeout: float = 30.0) -> Dict[str, Any]:
        """Call any cua-driver MCP tool by name with arbitrary args.
        ``session`` is injected (preserves the caller's explicit one
        via setdefault). For tools the wrapper doesn't already type-
        wrap, this is the supported escape hatch — preferred over
        reaching for ``self._session.call_tool`` directly because it
        keeps the session-id contract consistent with everything else."""
        payload = dict(args) if args else {}
        payload.setdefault("session", self._session_id)
        return self._session.call_tool(name, payload, timeout=timeout)

    # ── Internal ───────────────────────────────────────────────────
    def _maybe_attach_element_token(self, tool: str, args: Dict[str, Any]) -> None:
        """Surface 6: when the wrapper is about to call a token-capable
        tool with `element_index`, look up the matching `element_token`
        from the last snapshot and attach it. cua-driver-rs's contract
        for combined args is documented in trycua/cua#1961:

          "element_token takes precedence over element_index when both
           supplied. Returns an explicit 'stale' error if the snapshot
           has been superseded."

        Gated on the per-tool capability claim so we don't send the
        field to drivers that predate the surface (which would reject
        the schema with `additionalProperties: false`).
        """
        idx = args.get("element_index")
        if not isinstance(idx, int):
            return
        token = self._snapshot_tokens.get(idx)
        if not token:
            return
        if not self._session.supports_capability(
            "accessibility.element_tokens", tool=tool
        ):
            return
        args["element_token"] = token

    def _action(
        self,
        name: str,
        args: Dict[str, Any],
        *,
        inject_session: bool = True,
    ) -> ActionResult:
        # Attach the snapshot's element_token whenever the call carries
        # an element_index and the target tool advertises support.
        self._maybe_attach_element_token(name, args)
        # Carry this run's session id so the cua-driver agent cursor
        # and per-session state (config overrides, recording ownership)
        # stay tied to this run. setdefault preserves any explicit
        # session a caller already supplied.
        if inject_session:
            args.setdefault("session", self._session_id)
        try:
            out = self._session.call_tool(name, args)
        except Exception as e:
            logger.exception("cua-driver %s call failed", name)
            return ActionResult(ok=False, action=name, message=f"cua-driver error: {e}")
        ok = not out["isError"]
        data = out["data"]
        structured = out.get("structuredContent") or {}
        message = ""
        if isinstance(data, dict):
            message = str(data.get("message", ""))
        elif isinstance(data, str):
            message = data
        if not message and isinstance(structured, dict):
            message = str(structured.get("message", ""))
        # Merge data + structuredContent into meta for debugging, structured
        # winning on key overlap (it is the canonical verdict surface).
        meta: Dict[str, Any] = {}
        if isinstance(data, dict):
            meta.update(data)
        if isinstance(structured, dict):
            meta.update(structured)
        return _action_result_from(name, ok, message, meta, structured,
                                   requested_delivery=args.get("delivery_mode"))
