a5bd246865
* refactor(plugins): remove the Sep 2026 decomposition compat layer on schedule The PLUGIN-COMPAT layer (2776813df3+d63e380324+0a5164cebe) kept pre-#102117 import paths alive for external plugins until 2026-09-14. That window closed two weeks ago; since then the loader has already been skipping plugins that use the old paths. This removes the layer itself: - 328 appended `PLUGIN-COMPAT` blocks (lazy `__getattr__` pointer tables, re-exported third-party names, restored dead definitions) and the three re-export stub modules (gateway/startup_watchdog, hermes_cli/observability/relay_runtime, tools/environments/modal_utils) - COMPAT_MANIFEST.md, compat_manifest.json, scripts/check_compat_pointers.py and its lint step - the reporting surfaces: CLI banner notice, `hermes plugins compat`, the `hermes doctor` section, the post-update notice, the Desktop one-time dialog, the loader's pre-import skip and the `plugins.allow_deprecated_imports` escape hatch An external plugin that still imports an old path now fails to load with its ImportError as the reason in `hermes plugins list`, the same path as any broken plugin. hermes_cli/plugin_compat.py stays as three inert stubs (compat_report, removal_in_effect, summary_lines): an already-running pre-removal `hermes update` lazy-imports them after the checkout swap (tests/compat/old_updater_surface.json). In-tree fallout, both already dead: hermes_cli/setup.py::_check_espeak_ng (no callers; its `shutil` came from a compat block) and gateway/config.py::SessionResetPolicy ("retained solely for the scheduled plugin-compat window"). Two test_run_agent patches targeted the removed `run_agent.handle_function_call` pointer; they now patch `model_tools.handle_function_call`, the seam production reads, like every sibling test in that file. * chore: retrigger CI (zero-job startup_failure phantom) * test: drop resolution allowlist rows for the two deleted which() sites hermes_cli/setup.py::_check_espeak_ng (dead) and tools/skillevaluator_scan.py::scanner_available (a restored definition inside a PLUGIN-COMPAT block) no longer exist; the stale-row gate requires their allowlist entries go with them.
431 lines
28 KiB
Python
431 lines
28 KiB
Python
"""Cua-driver backend (macOS, Windows, Linux): MCP over stdio to `cua-driver`. The async `mcp` SDK runs on a
|
|
background loop (``cua_backend_session``); the same tool surface works on all three platforms, and per-host gaps
|
|
(no DISPLAY, missing AT-SPI, TCC) surface via `hermes computer-use doctor` instead of failing silently. Install
|
|
with `hermes computer-use install`. The macOS path uses private SkyLight SPIs that can break on OS updates.
|
|
Siblings: ``cua_backend_driver`` (binary/contract), ``cua_backend_capture`` + ``cua_backend_input``
|
|
(mixins), ``cua_backend_parse``, ``cua_backend_session`` (bridge + session + CLI fallback), ``cua_backend_daemon``
|
|
(private daemon + macOS app identity). Siblings look this module's config/policy helpers up lazily."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import contextlib
|
|
import logging
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import uuid
|
|
from pathlib import PureWindowsPath
|
|
from typing import Any, Dict, List, Optional, Tuple
|
|
|
|
from hermes_cli._subprocess_compat import windows_hide_flags
|
|
from hermes_platform.host.runtime import is_wsl
|
|
from tools.computer_use.backend import ActionResult, ComputerUseBackend
|
|
from tools.computer_use.cua_backend_capture import _CaptureMixin
|
|
from tools.computer_use.cua_backend_daemon import _EmbeddedCuaDaemon
|
|
from tools.computer_use.cua_backend_driver import ( # noqa: F401 — resolve_cua_driver_cmd: frozen updater surface
|
|
_CUA_DRIVER_CMD_ENV, cua_driver_binary_available, cua_driver_runtime_contract_status,
|
|
resolve_cua_driver_cmd)
|
|
from tools.computer_use.cua_backend_input import _InputMixin
|
|
from tools.computer_use.cua_backend_parse import _action_result_from
|
|
from tools.computer_use.cua_backend_session import _AsyncBridge, _CuaDriverSession
|
|
|
|
logger = logging.getLogger(__name__)
|
|
# cua-driver's anonymous PostHog telemetry gate ("0" disables; absent => ON upstream).
|
|
_CUA_TELEMETRY_ENV_VAR = "CUA_DRIVER_RS_TELEMETRY_ENABLED"
|
|
_CUA_NATIVE_WAYLAND_ENV_VAR = "CUA_DRIVER_RS_ENABLE_WAYLAND"
|
|
|
|
|
|
def _computer_use_cfg() -> Dict[str, Any]:
|
|
"""The ``computer_use`` config block, or ``{}`` when config is unreadable."""
|
|
with contextlib.suppress(Exception):
|
|
from hermes_cli.config import load_config
|
|
return (load_config() or {}).get("computer_use") or {}
|
|
return {}
|
|
|
|
def _cua_no_overlay() -> bool:
|
|
"""Pass ``--no-overlay``? ``computer_use.no_overlay`` overrides; else off on macOS (cursor-overlay redraw
|
|
loop can peg a core after a session), headless Linux / WSL2 / containers, and Linux X11 (the overlay is a
|
|
fullscreen always-on-top all-workspaces window with no compositor-owned lifecycle, so an unclean session
|
|
end can leave it wedged over every app); on for Windows and Linux Wayland (compositor owns the surface).
|
|
|
|
Explicit ``True`` / ``False`` overrides auto-detection. See #28152, #47032.
|
|
"""
|
|
val = _computer_use_cfg().get("no_overlay")
|
|
if val is not None or sys.platform != "linux":
|
|
return bool(val) if val is not None else sys.platform == "darwin"
|
|
wsl = is_wsl()
|
|
return wsl or not os.environ.get("DISPLAY") or (
|
|
# Linux/X11: the cursor overlay is a fullscreen, always-on-top, all-workspaces X11 window
|
|
# (save-unders path). An unclean session end (agent interrupted mid-capture, stale target window)
|
|
# can leave it stuck above every app on every workspace, wedging desktop input until the app
|
|
# restarts — the same failure class as the HUD window on Mutter/X11 (#83473). There is no
|
|
# compositor-owned surface to tear down with the client connection, so default the overlay off on
|
|
# X11 too; set computer_use.no_overlay: false to keep the cursor. Wayland keeps it: the compositor
|
|
# owns the overlay surface lifecycle there.
|
|
os.environ.get("XDG_SESSION_TYPE") != "wayland" and not os.environ.get("WAYLAND_DISPLAY"))
|
|
|
|
|
|
def _cua_telemetry_disabled() -> bool:
|
|
"""True unless ``computer_use.cua_telemetry`` opts in (unreadable config fails SAFE toward disabling)."""
|
|
return not bool(_computer_use_cfg().get("cua_telemetry", False))
|
|
|
|
def _cua_configured_permission_mode() -> str:
|
|
"""``computer_use.permission_mode``: ``standard`` (default) or ``bounded``; unknown values fall closed to
|
|
``standard``. ``unrestricted`` is deliberately NOT a config value — it stays tied to the per-session YOLO
|
|
toggle so a stale config line can never silently bypass approvals."""
|
|
raw = str(_computer_use_cfg().get("permission_mode", "standard") or "").strip().lower()
|
|
return "bounded" if raw == "bounded" else "standard"
|
|
|
|
# ``computer_use.ax_max_elements``: bound on the DRIVER's accessibility-tree walk per capture. The
|
|
# visible-element cap in tool.py (_DEFAULT_MAX_ELEMENTS) trims the RESPONSE only, so without this an
|
|
# unbounded walk pays for nodes the model never sees; 0 disables the bound (driver default: 2,000
|
|
# elements / depth 25).
|
|
_DEFAULT_AX_MAX_ELEMENTS = 200
|
|
|
|
def _cua_configured_ax_max_elements() -> int:
|
|
"""Bound on ``get_window_state``'s AX walk; 0 = no bound (driver default). Unreadable config fails to
|
|
the default, never to unbounded. Measured on macOS (cua-driver 0.28.2, M-series): a 1,444-node Chrome
|
|
window 540 ms -> 83 ms and a 456-node Finder window 6.9 s -> 0.6 s at 200, on the CLI transport. The
|
|
bounded result is a prefix of the unbounded walk, so the elements the model sees are unchanged. Raise
|
|
it toward 400 to keep the full first 100 visible elements on a pathological tree (~1.4 s on Finder);
|
|
cost grows with the bound, and a bound in the thousands buys nothing the response cap keeps. This
|
|
bounds the nodes COLLECTED, not the walk's wall clock: a target whose accessibility surface exceeds the
|
|
driver's own 20 s walk timeout still fails at every bound (and every depth bound), so a timeout error is
|
|
never an argument for a smaller or larger value here."""
|
|
raw = _computer_use_cfg().get("ax_max_elements", _DEFAULT_AX_MAX_ELEMENTS)
|
|
try:
|
|
return max(0, int(raw))
|
|
except (TypeError, ValueError):
|
|
return _DEFAULT_AX_MAX_ELEMENTS
|
|
|
|
def _manifest_is_mode_independent(path: str) -> bool:
|
|
"""True when this manifest may accompany any permission mode: v1/v2 declare ``mode: bounded`` and abort
|
|
startup under an unrestricted runtime; v3 has no mode and is the ceiling the driver accepts alongside any
|
|
mode. Unreadable / unparseable -> False (forwarding one would turn a working session into a hard startup
|
|
failure; bounded forwards unconditionally anyway)."""
|
|
try:
|
|
import hermes_yaml as yaml
|
|
|
|
with open(path, "r", encoding="utf-8-sig") as handle:
|
|
parsed = yaml.safe_load(handle)
|
|
except Exception:
|
|
logger.debug("could not read capability manifest %s", path, exc_info=True)
|
|
return False
|
|
version = parsed.get("version") if isinstance(parsed, dict) else None
|
|
return isinstance(version, int) and not isinstance(version, bool) and version >= 3
|
|
|
|
def _computer_use_max_image_dimension() -> Optional[int]:
|
|
"""``computer_use.max_image_dimension`` longest-edge cap (default 1456 = aux-vision downscale); ``0``/negative -> None."""
|
|
try:
|
|
dim = int(_computer_use_cfg().get("max_image_dimension", 1456))
|
|
except (TypeError, ValueError):
|
|
dim = 1456
|
|
return dim if dim > 0 else None
|
|
|
|
def desktop_identity(env: Optional[Dict[str, str]] = None) -> str:
|
|
"""The screen a backend spawned from ``env`` acts on: its DISPLAY (``''`` when none). Recorded next to the
|
|
cached backend so a Bot Desktop that starts (or restarts on another number) AFTER the backend was cached is
|
|
noticed — the cached cua-driver still points at the old seat or at no display at all."""
|
|
return str((cua_driver_child_env(env) if env is None else env).get("DISPLAY") or "")
|
|
|
|
|
|
def backend_display_stale(recorded: str, current: str) -> bool:
|
|
"""True when a cached backend's recorded display identity no longer matches the one a fresh spawn would get."""
|
|
return (recorded or "") != (current or "")
|
|
|
|
|
|
def cua_driver_child_env(base_env: Optional[Dict[str, str]] = None) -> Dict[str, str]:
|
|
"""Env for spawning cua-driver: ``base_env`` (default ``os.environ``) plus ``CUA_DRIVER_RS_TELEMETRY_ENABLED=0``
|
|
unless the user opted in, plus the native-Wayland bridge (``computer_use.native_wayland`` config opt-in, only when
|
|
the child has a Wayland display). Used by every spawn site (MCP, status, doctor, install) so CLI and gateway
|
|
runtimes share one policy."""
|
|
env = dict(os.environ if base_env is None else base_env)
|
|
# A running Bot Desktop for this profile owns the agent's screen: DISPLAY/XAUTHORITY/DBUS point there so
|
|
# cua-driver never acts on a seat the human is sitting at (#90374 class) and headless hosts get a display.
|
|
from tools.bot_desktop.runtime import desktop_env as _bot_desktop_env
|
|
env = _bot_desktop_env(env)
|
|
if _cua_telemetry_disabled():
|
|
env[_CUA_TELEMETRY_ENV_VAR] = "0"
|
|
if sys.platform == "linux" and env.get("WAYLAND_DISPLAY") and bool(_computer_use_cfg().get("native_wayland", False)):
|
|
env[_CUA_NATIVE_WAYLAND_ENV_VAR] = "1"
|
|
return env
|
|
|
|
def sandbox_mcp_invocation() -> Optional[Tuple[Tuple[str, List[str]], Dict[str, str]]]:
|
|
"""``((command, args), child_env)`` spawning ``cua-driver mcp`` INSIDE the terminal backend when the Bot
|
|
Desktop is placed there (the driver in the sandbox image drives the sandbox's own screen); None on a
|
|
gateway-hosted desktop, where the local driver is used. Placement is the authority: a ``terminal``
|
|
placement gets its screen started here and a ``refused`` one raises — the host driver is never the
|
|
fallback for a sandbox whose screen is down."""
|
|
from tools.bot_desktop import placement, runtime as _bd_runtime
|
|
if _bd_runtime.tool_placement() == placement.GATEWAY:
|
|
return None
|
|
published = _bd_runtime.published_env()
|
|
if not published.get("DISPLAY"):
|
|
raise RuntimeError("the screen inside the terminal backend's sandbox is gone; start it again")
|
|
from tools.bot_desktop import sandbox_host
|
|
env = _bd_runtime._sandbox_env(create=True)
|
|
if env is None:
|
|
raise RuntimeError("the terminal backend's sandbox is not running, so there is nowhere to run cua-driver")
|
|
command, args = sandbox_host.cua_mcp_invocation(env, _bd_runtime._profile_name(),
|
|
{**published, _CUA_TELEMETRY_ENV_VAR: "0"})
|
|
_bd_runtime.touch_activity()
|
|
return (command, args), {"PATH": os.environ.get("PATH", "")}
|
|
|
|
|
|
def sanitized_cua_driver_env() -> Dict[str, str]:
|
|
"""``cua_driver_child_env()`` with Hermes provider secrets stripped — cua-driver is a third-party binary and must
|
|
never inherit API keys. Falls back to the unsanitized telemetry env if the sanitizer can't import."""
|
|
env = cua_driver_child_env()
|
|
with contextlib.suppress(Exception):
|
|
# 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
|
|
return _sanitize_subprocess_env(env)
|
|
return env
|
|
|
|
def _run_quiet(argv: List[str], *, timeout: float, swallow: Any = (), **kw: Any) -> Any:
|
|
"""``subprocess.run`` for short probe verbs: text mode, stdin=DEVNULL unless overridden (older drivers fall into a
|
|
stdin-reading mode on unknown verbs; EOF makes them exit fast instead of blocking until the timeout), output
|
|
captured unless the caller redirects it. Exceptions in ``swallow`` return None; others raise."""
|
|
kw.setdefault("stdin", subprocess.DEVNULL)
|
|
kw.setdefault("encoding", "utf-8")
|
|
kw.setdefault("errors", "replace")
|
|
"stdout" in kw or kw.setdefault("capture_output", True)
|
|
try:
|
|
return subprocess.run(argv, text=True, timeout=timeout, stdin=kw.pop("stdin"), encoding=kw.pop("encoding"),
|
|
errors=kw.pop("errors"), **kw)
|
|
except swallow:
|
|
return None
|
|
|
|
def _run_driver(driver_cmd: str, *args: str, timeout: float, swallow: Any = ()) -> Any:
|
|
"""Run a short cua-driver verb with the sanitized env and hidden window."""
|
|
return _run_quiet([driver_cmd, *args], timeout=timeout, swallow=swallow, encoding="utf-8",
|
|
errors="replace", creationflags=windows_hide_flags(), env=sanitized_cua_driver_env())
|
|
|
|
def cua_daemon_listening(driver_cmd: str, socket_path: Optional[str] = None, *, timeout: float = 3.0) -> Optional[bool]:
|
|
"""Socket-level liveness of a ``cua-driver serve`` daemon: ``cua-driver status`` connects to the daemon
|
|
socket (the driver's default, or ``socket_path``) and exits 0 only when a daemon answers. False when the
|
|
CLI reports the daemon is not running, None when the probe itself failed (unknown). Never raises.
|
|
The binary-level runtime contract (``manifest``) cannot see this — a dead daemon looks healthy there (#114748)."""
|
|
args = ("status", "--socket", socket_path) if socket_path else ("status",)
|
|
proc = _run_driver(driver_cmd, *args, timeout=timeout, swallow=(OSError, subprocess.SubprocessError))
|
|
if proc is None:
|
|
return None
|
|
if proc.returncode == 0:
|
|
return True
|
|
return False if "not running" in f"{proc.stdout}\n{proc.stderr}".lower() else None
|
|
|
|
def _linux_session_locked() -> Optional[bool]:
|
|
"""Is the graphical session locked? (Linux; best-effort.) A locked KDE/GNOME session freezes renderers and
|
|
half-disables the AX tree, so discovery legitimately returns nothing — which otherwise reads as a driver bug.
|
|
True/False when loginctl answers, None when unavailable (non-Linux, no systemd-logind, probe failure)."""
|
|
# 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 != "linux":
|
|
return None
|
|
try:
|
|
proc = _run_quiet(["loginctl", "list-sessions", "--no-legend"], timeout=2.0)
|
|
seats = [line.split()[0] for line in proc.stdout.splitlines() if len(line.split()) >= 2 and "seat" in line]
|
|
if proc.returncode != 0 or not seats:
|
|
return None
|
|
return not any("LockedHint=no" in _run_quiet(["loginctl", "show-session", s, "-p", "LockedHint"], timeout=2.0).stdout
|
|
for s in seats)
|
|
except Exception:
|
|
return None
|
|
|
|
def _empty_discovery_reason() -> str:
|
|
"""One-line diagnosis for 'window discovery found nothing'."""
|
|
if _linux_session_locked() is True:
|
|
return ("the desktop session is LOCKED (loginctl LockedHint=yes) — unlock the screen; "
|
|
"a locked compositor hides windows and freezes app renderers")
|
|
if sys.platform == "linux" and not os.environ.get("DISPLAY"):
|
|
return "no DISPLAY is set — X11/XWayland is not reachable from this process"
|
|
if sys.platform == "darwin": # headless Mac / asleep panel: ScreenCaptureKit has 0 shareable displays while TCC looks fine
|
|
return ("window discovery returned no windows; on macOS this usually means no shareable display (headless Mac or "
|
|
"panel asleep) — wake the display or attach a monitor/HDMI dummy, then run `hermes computer-use doctor`")
|
|
return "window discovery returned no windows; run `hermes computer-use doctor` (display reachability, AX capability)"
|
|
|
|
class CuaDriverBackend(_CaptureMixin, _InputMixin, 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", "bounded", "unrestricted"}:
|
|
raise ValueError(f"unsupported cua-driver permission mode: {permission_mode}")
|
|
self.permission_mode = permission_mode
|
|
self._embedded_daemon: Optional[_EmbeddedCuaDaemon] = None
|
|
if permission_mode != "standard":
|
|
# Manifest: mandatory for bounded (the daemon validates it), optional for unrestricted where it still
|
|
# caps what an approval-bypassed run may touch.
|
|
raw = _computer_use_cfg().get("capability_manifest")
|
|
# Resolve at daemon launch, after start() reconciles the PM pin.
|
|
self._embedded_daemon = _EmbeddedCuaDaemon(
|
|
"", permission_mode,
|
|
capability_manifest=raw.strip() if isinstance(raw, str) and raw.strip() else None)
|
|
self._bridge = _AsyncBridge()
|
|
self._session = _CuaDriverSession(self._bridge, self._embedded_daemon)
|
|
# Sticky target (set by capture()/focus_app(), used by actions): `_active_pid`, `_active_window_id`, `_last_app`,
|
|
# `_last_target` (exact identity for capture_after — Linux app names may be generic, e.g. several unrelated Qt
|
|
# windows all say Qt6Application), `_snapshot_tokens` (element_index -> element_token, attached to actions so
|
|
# cua-driver reports "stale" instead of silently re-resolving).
|
|
self._clear_active_target()
|
|
# Public session label (one per Hermes run) sent as `session` on every call: owns the cursor color and
|
|
# gives config/recording state a stable owner across transport restarts. Part of the 0.20 runtime contract.
|
|
self._session_id: str = f"hermes-{uuid.uuid4().hex[:12]}"
|
|
self._session.set_transport_reset_callback(self._handle_transport_reset)
|
|
|
|
def _handle_transport_reset(self) -> None:
|
|
"""Invalidate every capability minted by the replaced transport."""
|
|
self._clear_active_target()
|
|
|
|
def start(self) -> None:
|
|
# Driver inside the terminal backend: the sandbox image pins its own cua-driver; the host
|
|
# binary (if any) is not the one that will run, so neither its acquisition nor its contract
|
|
# matters. On the host, runtime acquisition is on-demand, never the explicit install command
|
|
# (which may elevate for host setup and bypass the lazy-install gate).
|
|
if sandbox_mcp_invocation() is not None:
|
|
contract = {"ready": True}
|
|
else:
|
|
if not os.environ.get(_CUA_DRIVER_CMD_ENV, "").strip():
|
|
from pm import ensure
|
|
ensure("cua-driver")
|
|
contract = cua_driver_runtime_contract_status()
|
|
if not contract.get("ready"):
|
|
raise RuntimeError(f"cua-driver is not ready: {contract.get('reason') or 'runtime contract is incomplete'}. "
|
|
+ ("Update the binary selected by HERMES_CUA_DRIVER_CMD or remove that override."
|
|
if os.environ.get(_CUA_DRIVER_CMD_ENV, "").strip() else "Run `hermes computer-use install` to repair it."))
|
|
|
|
# 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_import() raises
|
|
# InstallError explaining why PM could not enable the SDK,
|
|
# which surfaces via the backend-unavailable path in tool.py.
|
|
from pm import ensure_import
|
|
ensure_import("computer-use")
|
|
# A just-installed package may not be importable until the import
|
|
# machinery's caches are refreshed within this process.
|
|
import importlib
|
|
importlib.invalidate_caches()
|
|
with contextlib.ExitStack() as rollback: # a failed start stops the private daemon, then re-raises
|
|
if self._embedded_daemon is not None:
|
|
rollback.callback(self._embedded_daemon.stop) and self._embedded_daemon.start()
|
|
self._session.start()
|
|
rollback.pop_all()
|
|
# Declare this run's identity. Non-fatal: cua-driver accepts anonymous calls (cursor won't render), so degrade.
|
|
self._best_effort("start_session failed (continuing anonymous)",
|
|
self._session.call_tool, "start_session", {"session": self._session_id})
|
|
# Post-handshake tuning guards on `_started`: before the handshake flips it, call_tool would re-enter
|
|
# session.start() (stubbed start() recurses).
|
|
if self._session._started:
|
|
max_dim = _computer_use_max_image_dimension()
|
|
if max_dim: # smaller screenshots cost less over the daemon socket and per turn
|
|
self._best_effort("set_config(max_image_dimension) failed",
|
|
self.set_config, max_image_dimension=max_dim)
|
|
if _cua_no_overlay(): # belt-and-suspenders when --no-overlay is unsupported or ignored
|
|
self._best_effort("set_agent_cursor_enabled failed",
|
|
self.set_agent_cursor_enabled, False, cursor_id=self._session_id)
|
|
|
|
def stop(self) -> None:
|
|
# Best-effort end_session so the driver cleans per-session state (cursor overlay, recording ownership,
|
|
# config overrides); the connection drop below releases daemon-side state regardless.
|
|
if self._session._started:
|
|
self._best_effort("end_session failed (continuing teardown)",
|
|
self._session.call_tool, "end_session", {"session": self._session_id})
|
|
with contextlib.ExitStack() as teardown: # every step runs even if one raised (LIFO: session, bridge, daemon)
|
|
self._embedded_daemon is None or teardown.callback(self._embedded_daemon.stop)
|
|
teardown.callback(self._bridge.stop)
|
|
teardown.callback(self._session.stop)
|
|
|
|
@staticmethod
|
|
def _best_effort(what: str, fn, *args: Any, **kwargs: Any) -> None:
|
|
"""Run a non-fatal driver call, logging (debug) instead of raising."""
|
|
try:
|
|
fn(*args, **kwargs)
|
|
except Exception as e:
|
|
logger.debug("cua-driver %s: %s", what, e)
|
|
|
|
def is_available(self) -> bool:
|
|
return sys.platform in ("darwin", "win32", "linux") and cua_driver_binary_available() # other Unix-likes untested E2E
|
|
|
|
def _clear_active_target(self) -> None:
|
|
"""Forget a capture/focus target so a failed lookup cannot misroute input."""
|
|
self._active_pid = self._active_window_id = self._last_app = self._last_target = 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] = {}
|
|
|
|
def _set_active_target(self, target: Dict[str, Any]) -> None:
|
|
self._active_pid = target["pid"]
|
|
self._active_window_id = target["window_id"]
|
|
self._snapshot_tokens = {} # prior snapshot's tokens: disarm before any capture so an exception can't pair them
|
|
self._last_target = {"pid": self._active_pid, "window_id": self._active_window_id}
|
|
|
|
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 returning ``{pid, bundle_id, name, windows[]}``. ``creates_new_application_instance=True``
|
|
forces a fresh instance so concurrent runs touching the same app get isolated windows."""
|
|
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, **{k: v for k, v in (
|
|
("bundle_id", bundle_id), ("name", name), ("urls", urls and list(urls)),
|
|
("additional_arguments", additional_arguments and list(additional_arguments)),
|
|
("creates_new_application_instance", creates_new_application_instance or None)) if v}}
|
|
out = self._session.call_tool("launch_app", args)
|
|
return out["structuredContent"] or {"data": out["data"]}
|
|
|
|
def bring_to_front(self, *, pid: int, window_id: Optional[int] = None) -> ActionResult:
|
|
"""Activate a window so subsequent foreground-dispatched input lands on it."""
|
|
args: Dict[str, Any] = {"pid": int(pid), **({} if window_id is None else {"window_id": int(window_id)})}
|
|
# Strict live schema with no session property: a standalone native focus op, not a session-scoped input action.
|
|
return self._action("bring_to_front", args, inject_session=False)
|
|
|
|
def set_agent_cursor_enabled(self, enabled: bool, *, cursor_id: Optional[str] = None) -> ActionResult:
|
|
"""Toggle the agent cursor overlay's visibility for this run."""
|
|
return self._action("set_agent_cursor_enabled",
|
|
{"enabled": bool(enabled), **({"cursor_id": cursor_id} if cursor_id else {})})
|
|
|
|
def set_config(self, **config) -> ActionResult:
|
|
"""Set cua-driver config keys (e.g. ``max_image_dimension``); unknown keys pass through — cua-driver validates."""
|
|
return self._action("set_config", dict(config))
|
|
|
|
def call_tool(self, name: str, args: Optional[Dict[str, Any]] = None, *, timeout: float = 30.0) -> Dict[str, Any]:
|
|
"""Generic escape hatch: call any cua-driver MCP tool by name. ``session`` is injected via setdefault, so
|
|
this is the supported path for tools the wrapper does not type-wrap (preferred over ``self._session.call_tool``)."""
|
|
payload = dict(args) if args else {}
|
|
payload.setdefault("session", self._session_id)
|
|
return self._session.call_tool(name, payload, timeout=timeout)
|
|
|
|
def _action(self, name: str, args: Dict[str, Any], *, inject_session: bool = True) -> ActionResult:
|
|
# Attach the snapshot's `element_token` to an `element_index` call so a superseded snapshot yields an explicit
|
|
# 'stale' error. Two ways to establish support, the live input schema first: cua-driver 0.21+ stopped
|
|
# publishing per-tool `capabilities[]` while still accepting `element_token` in its schema, and it REFUSES a
|
|
# bare `element_index` (`snapshot_id_required`) — gating on the capability alone broke EVERY element click and
|
|
# left only pixel clicks working. The capability check stays so older drivers that shipped the vocabulary keep
|
|
# working; drivers advertising neither (`additionalProperties: false`) must never see the property.
|
|
idx = args.get("element_index")
|
|
token = self._snapshot_tokens.get(idx) if isinstance(idx, int) else None
|
|
if token and (self._session.supports_input_property(name, "element_token")
|
|
or self._session.supports_capability("accessibility.element_tokens", tool=name)):
|
|
args["element_token"] = token
|
|
if inject_session: # setdefault preserves any explicit session a caller already supplied
|
|
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}")
|
|
data = out["data"]
|
|
structured = out.get("structuredContent") or {}
|
|
message = (str(data.get("message", "")) if isinstance(data, dict) else data if isinstance(data, str) else "") \
|
|
or (str(structured.get("message", "")) if isinstance(structured, dict) else "")
|
|
# Merge data + structuredContent into meta, structured winning on overlap (canonical verdict surface).
|
|
meta = {k: v for part in (data, structured) if isinstance(part, dict) for k, v in part.items()}
|
|
return _action_result_from(name, not out["isError"], message, meta, structured,
|
|
requested_delivery=args.get("delivery_mode"))
|