diff --git a/tools/computer_use/cua_backend.py b/tools/computer_use/cua_backend.py index 135f70cc3b..9ce43e4115 100644 --- a/tools/computer_use/cua_backend.py +++ b/tools/computer_use/cua_backend.py @@ -16,6 +16,8 @@ keeps working; siblings look policy helpers up lazily through this module. from __future__ import annotations +import contextlib +import importlib import logging import os import shutil # noqa: F401 (tests patch cua_backend.shutil / .subprocess / .threading) @@ -27,9 +29,7 @@ from typing import Any, Dict, List, Optional from hermes_cli._subprocess_compat import windows_hide_flags from tools.computer_use.backend import ActionResult, ComputerUseBackend -from tools.computer_use.cua_backend_capture import ( # noqa: F401 - _CaptureMixin, _linux_x11_active_window_id, _select_capture_target, -) +from tools.computer_use.cua_backend_capture import _CaptureMixin, _select_capture_target # noqa: F401 from tools.computer_use.cua_backend_daemon import ( # noqa: F401 _EmbeddedCuaDaemon, _embedded_daemon_spawn_command, _resolve_cua_driver_app_path, _validate_cua_driver_app_signature, @@ -44,7 +44,7 @@ from tools.computer_use.cua_backend_input import _InputMixin from tools.computer_use.cua_backend_parse import ( # noqa: F401 _action_result_from, _extract_tool_result, _image_dimensions_from_bytes, _ingest_windows, _is_placeholder_id, _parse_elements_from_structured, _parse_elements_from_tree, - _parse_key_combo, _parse_xprop_net_active_window, _windows_from_tool_result, + _parse_xprop_net_active_window, _windows_from_tool_result, ) from tools.computer_use.cua_backend_session import _AsyncBridge, _CuaDriverSession # noqa: F401 @@ -77,10 +77,8 @@ def _cua_no_overlay() -> bool: val = _computer_use_cfg().get("no_overlay") if val is not None: return bool(val) - if sys.platform == "darwin": - return True if sys.platform != "linux": - return False + return sys.platform == "darwin" if not os.environ.get("DISPLAY"): return True try: @@ -104,12 +102,6 @@ def _cua_configured_permission_mode() -> str: raw = str(_computer_use_cfg().get("permission_mode", "standard") or "").strip().lower() return raw if raw in {"standard", "bounded"} else "standard" -def _cua_capability_manifest() -> Optional[str]: - """``computer_use.capability_manifest`` path, or None. Existence is - validated by ``_EmbeddedCuaDaemon`` so a missing file fails loudly.""" - raw = _computer_use_cfg().get("capability_manifest") - return raw.strip() if isinstance(raw, str) and raw.strip() else None - 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 @@ -157,17 +149,24 @@ def sanitized_cua_driver_env() -> Dict[str, str]: except Exception: return env -def _run_driver(driver_cmd: str, *args: str, timeout: float) -> subprocess.CompletedProcess: - """Run a short cua-driver verb with the sanitized env, hidden window and - stdin=DEVNULL (older drivers fall into a stdin-reading mode on unknown - verbs; EOF makes them exit fast instead of blocking until the timeout).""" - return subprocess.run( - [driver_cmd, *args], - capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=timeout, - stdin=subprocess.DEVNULL, - creationflags=windows_hide_flags(), - env=sanitized_cua_driver_env(), - ) +def _run_quiet(argv: List[str], *, timeout: float, swallow: Any = (), + **kw: Any) -> Any: + """``subprocess.run`` for short probe verbs: text mode, stdin=DEVNULL (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.""" + if "stdout" not in kw: + kw["capture_output"] = True + try: + return subprocess.run(argv, text=True, timeout=timeout, stdin=subprocess.DEVNULL, **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()) # --------------------------------------------------------------------------- @@ -182,13 +181,8 @@ def _linux_session_locked() -> Optional[bool]: systemd-logind, probe failure).""" if sys.platform != "linux": return None - - def _loginctl(*args: str) -> subprocess.CompletedProcess: - return subprocess.run(["loginctl", *args], capture_output=True, text=True, - timeout=2.0, stdin=subprocess.DEVNULL) - try: - proc = _loginctl("list-sessions", "--no-legend") + proc = _run_quiet(["loginctl", "list-sessions", "--no-legend"], timeout=2.0) if proc.returncode != 0: return None any_seat = False @@ -197,35 +191,32 @@ def _linux_session_locked() -> Optional[bool]: if len(parts) < 2 or "seat" not in line: continue any_seat = True - if "LockedHint=no" in _loginctl("show-session", parts[0], "-p", "LockedHint").stdout: + hint = _run_quiet(["loginctl", "show-session", parts[0], "-p", "LockedHint"], timeout=2.0) + if "LockedHint=no" in hint.stdout: return False return True if any_seat else None except Exception: return None +# (predicate, reason) in priority order; the default applies when none match. +_EMPTY_DISCOVERY_REASONS = ( + (lambda: _linux_session_locked() is True, + "the desktop session is LOCKED (loginctl LockedHint=yes) — unlock the screen; " + "a locked compositor hides windows and freezes app renderers"), + (lambda: sys.platform == "linux" and not os.environ.get("DISPLAY"), + "no DISPLAY is set — X11/XWayland is not reachable from this process"), + # Headless Mac / asleep panel: ScreenCaptureKit has 0 shareable displays while TCC grants look fine. + (lambda: sys.platform == "darwin", + "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`"), +) + 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 grants look 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)" - ) + return next((reason for applies, reason in _EMPTY_DISCOVERY_REASONS if applies()), + "window discovery returned no windows; run `hermes computer-use doctor` " + "(display reachability, AX capability)") # --------------------------------------------------------------------------- @@ -245,19 +236,12 @@ def _maybe_repair_runtime_contract(contract: Dict[str, Any]) -> Dict[str, Any]: explicit ``HERMES_CUA_DRIVER_CMD`` override is authoritative even when broken, and a missing binary means installation was never requested.""" global _contract_repair_attempted - if ( - contract.get("ready") - or _contract_repair_attempted - or os.environ.get(_CUA_DRIVER_CMD_ENV, "").strip() - or not contract.get("binary") - ): + if (contract.get("ready") or _contract_repair_attempted + or os.environ.get(_CUA_DRIVER_CMD_ENV, "").strip() or not contract.get("binary")): return contract _contract_repair_attempted = True - logger.info( - "computer_use: installed cua-driver is not usable (%s); " - "attempting automatic repair", - contract.get("reason") or "runtime contract is incomplete", - ) + logger.info("computer_use: installed cua-driver is not usable (%s); attempting automatic repair", + contract.get("reason") or "runtime contract is incomplete") try: from hermes_cli.tools_config import install_cua_driver @@ -306,11 +290,10 @@ class CuaDriverBackend(_CaptureMixin, _InputMixin, ComputerUseBackend): # The manifest is mandatory for bounded (the daemon validates it) # and optional for unrestricted, where it still caps what an # approval-bypassed run may touch. + raw = _computer_use_cfg().get("capability_manifest") self._embedded_daemon = _EmbeddedCuaDaemon( - resolve_cua_driver_cmd() or "", - permission_mode, - capability_manifest=_cua_capability_manifest(), - ) + resolve_cua_driver_cmd() or "", 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. @@ -339,35 +322,31 @@ class CuaDriverBackend(_CaptureMixin, _InputMixin, ComputerUseBackend): if not contract.get("ready"): contract = _maybe_repair_runtime_contract(contract) if not contract.get("ready"): - reason = contract.get("reason") or "runtime contract is incomplete" - repair = ( - "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." - ) - raise RuntimeError(f"cua-driver is not ready: {reason}. {repair}") + repair = ("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.") + raise RuntimeError(f"cua-driver is not ready: " + f"{contract.get('reason') or 'runtime contract is incomplete'}. {repair}") _maybe_nudge_update() # `mcp` is an optional extra: lazy-install on first use (gated by # `security.allow_lazy_installs`); failure raises FeatureUnavailable # with the exact `uv pip install` hint. from tools.lazy_deps import ensure as _lazy_ensure _lazy_ensure("tool.computer_use", prompt=False) - import importlib importlib.invalidate_caches() # a just-installed package may not be importable yet + daemon = self._embedded_daemon try: - if self._embedded_daemon is not None: - self._embedded_daemon.start() + if daemon is not None: + daemon.start() self._session.start() except Exception: - if self._embedded_daemon is not None: - self._embedded_daemon.stop() + if daemon is not None: + daemon.stop() raise - # Declare this run's identity. Non-fatal: cua-driver accepts anonymous # calls (the cursor just won't render), so degrade rather than abort. 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: @@ -388,14 +367,13 @@ class CuaDriverBackend(_CaptureMixin, _InputMixin, ComputerUseBackend): if self._session._started: self._best_effort("end_session failed (continuing teardown)", self._session.call_tool, "end_session", {"session": self._session_id}) - try: - self._session.stop() - finally: - try: - self._bridge.stop() - finally: - if self._embedded_daemon is not None: - self._embedded_daemon.stop() + # Every teardown step runs even if an earlier one raised (LIFO callbacks: + # session, then bridge, then the private daemon). + with contextlib.ExitStack() as teardown: + if self._embedded_daemon is not None: + 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: @@ -412,10 +390,7 @@ class CuaDriverBackend(_CaptureMixin, _InputMixin, ComputerUseBackend): # ── Target state ─────────────────────────────────────────────── 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._active_pid = self._active_window_id = self._last_app = self._last_target = None self._snapshot_tokens = {} def _set_active_target(self, target: Dict[str, Any]) -> None: @@ -427,15 +402,9 @@ class CuaDriverBackend(_CaptureMixin, _InputMixin, ComputerUseBackend): self._last_target = {"pid": self._active_pid, "window_id": self._active_window_id} # ── App lifecycle / focus ───────────────────────────────────────── - 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]: + 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.""" @@ -460,8 +429,7 @@ class CuaDriverBackend(_CaptureMixin, _InputMixin, ComputerUseBackend): return self._action("bring_to_front", args, inject_session=False) # ── Agent cursor / config ──────────────────────────────────────── - def set_agent_cursor_enabled(self, enabled: bool, *, - cursor_id: Optional[str] = None) -> ActionResult: + 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: @@ -484,20 +452,16 @@ class CuaDriverBackend(_CaptureMixin, _InputMixin, ComputerUseBackend): return self._session.call_tool(name, payload, timeout=timeout) # ── Internal ─────────────────────────────────────────────────── - def _maybe_attach_element_token(self, tool: str, args: Dict[str, Any]) -> None: - """Attach the snapshot's ``element_token`` to an ``element_index`` call so - a superseded snapshot yields an explicit 'stale' error. Gated on the - per-tool capability: older drivers (``additionalProperties: false``) - must never see the field.""" + 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. Gated on the + # per-tool capability: older drivers (`additionalProperties: false`) + # must never see the field. idx = args.get("element_index") token = self._snapshot_tokens.get(idx) if isinstance(idx, int) else None - if token and self._session.supports_capability("accessibility.element_tokens", tool=tool): + if token and self._session.supports_capability("accessibility.element_tokens", tool=name): args["element_token"] = token - - def _action(self, name: str, args: Dict[str, Any], *, inject_session: bool = True) -> ActionResult: - self._maybe_attach_element_token(name, args) - # setdefault preserves any explicit session a caller already supplied. - if inject_session: + 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) @@ -511,9 +475,6 @@ class CuaDriverBackend(_CaptureMixin, _InputMixin, ComputerUseBackend): message = str(structured.get("message", "")) # Merge data + structuredContent into meta, structured winning on # overlap (it is the canonical verdict surface). - meta: Dict[str, Any] = {} - for part in (data, structured): - if isinstance(part, dict): - meta.update(part) + 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")) diff --git a/tools/computer_use/cua_backend_daemon.py b/tools/computer_use/cua_backend_daemon.py index b42980e371..c45f3e3a62 100644 --- a/tools/computer_use/cua_backend_daemon.py +++ b/tools/computer_use/cua_backend_daemon.py @@ -24,6 +24,13 @@ logger = logging.getLogger("tools.computer_use.cua_backend") # official releases. Exact matches only: a suffixed identifier or other team is an impostor. _CUA_DRIVER_BUNDLE_ID = "com.trycua.driver" _CUA_DRIVER_TEAM_IDS = ("4YEC26S9KF", "YCK386LBJ7") +_QUIET_ERRORS = (OSError, subprocess.SubprocessError) + +def _cb(): + """Origin module, looked up lazily so ``patch("tools.computer_use.cua_backend.X")`` applies.""" + from tools.computer_use import cua_backend + + return cua_backend def _resolve_cua_driver_app_path(driver_cmd: str) -> Optional[str]: """Return the CuaDriver.app bundle that CARRIES *driver_cmd*, if any. Derived from the @@ -43,13 +50,11 @@ def _validate_cua_driver_app_signature(app_path: str) -> None: ``Identifier=com.trycua.driver`` and an expected TeamIdentifier. ``TeamIdentifier=not set`` (ad-hoc dev builds) is allowed only with ``computer_use.allow_unsigned_driver: true``. Raises RuntimeError on any mismatch or when codesign is unavailable/fails.""" - from tools.computer_use import cua_backend as _cb - codesign = shutil.which("codesign") if not codesign: raise RuntimeError("codesign is required to verify CuaDriver.app before launching it.") try: - proc = subprocess.run([codesign, "-dv", app_path], capture_output=True, text=True, timeout=15) + proc = _cb()._run_quiet([codesign, "-dv", app_path], timeout=15) except (OSError, subprocess.TimeoutExpired) as exc: raise RuntimeError(f"could not verify CuaDriver.app signature: {exc}") from exc if proc.returncode != 0: @@ -64,7 +69,7 @@ def _validate_cua_driver_app_signature(app_path: str) -> None: raise RuntimeError(f"CuaDriver.app at {app_path} has identifier {identifier!r}, " f"expected {_CUA_DRIVER_BUNDLE_ID!r}; refusing to launch it.") if team in _CUA_DRIVER_TEAM_IDS or ( - team in ("", "not set") and _cb._computer_use_cfg().get("allow_unsigned_driver") is True): + team in ("", "not set") and _cb()._computer_use_cfg().get("allow_unsigned_driver") is True): return raise RuntimeError( f"CuaDriver.app at {app_path} is signed by team {team!r}, expected one of " @@ -116,11 +121,8 @@ class _EmbeddedCuaDaemon: _START_TIMEOUT_SECONDS = 15.0 def __init__(self, driver_cmd: str, permission_mode: str, capability_manifest: Optional[str] = None) -> None: - from tools.computer_use import cua_backend as _cb - if permission_mode not in {"unrestricted", "bounded"}: raise ValueError("embedded permission override supports unrestricted or bounded only") - self.capability_manifest: Optional[str] = None manifest = str(capability_manifest or "").strip() if not manifest and permission_mode == "bounded": raise ValueError("bounded permission mode requires computer_use.capability_manifest") @@ -128,19 +130,19 @@ class _EmbeddedCuaDaemon: manifest = os.path.abspath(os.path.expanduser(manifest)) if not os.path.isfile(manifest): raise ValueError(f"capability manifest not found: {manifest}") - self.capability_manifest = manifest + self.capability_manifest: Optional[str] = manifest or None # bounded always forwards (the driver validates it); other modes accept only a v3 # manifest — a legacy one would abort startup instead. - self.manifest_applies = bool(self.capability_manifest) and ( - permission_mode == "bounded" or _cb._manifest_is_mode_independent(str(self.capability_manifest))) - if self.capability_manifest and not self.manifest_applies: + self.manifest_applies = bool(manifest) and ( + permission_mode == "bounded" or _cb()._manifest_is_mode_independent(manifest)) + if manifest and not self.manifest_applies: logger.warning("computer_use.capability_manifest is a legacy (v1/v2) manifest, " "which cua-driver only accepts in bounded mode — it will NOT " "bound this %s session. Migrate the manifest to version 3 to " "keep a ceiling on approval-bypassed runs.", permission_mode) self.permission_mode = permission_mode self._driver_cmd = self._command = driver_cmd - self._mcp_args: List[str] = list(_cb._CUA_DRIVER_ARGS) + self._mcp_args: List[str] = list(_cb()._CUA_DRIVER_ARGS) self._process: Any = None self._owns_runtime = self._running = self._launch_via_app = False self._stderr_tail: deque[str] = deque(maxlen=20) @@ -150,14 +152,17 @@ class _EmbeddedCuaDaemon: else os.path.join(tempfile.gettempdir(), f"hc-{token}.sock")) def child_env(self) -> Dict[str, str]: - from tools.computer_use import cua_backend as _cb - - env = _cb.cua_driver_child_env() + env = _cb().cua_driver_child_env() env["CUA_DRIVER_PERMISSION_MODE"] = self.permission_mode if self.permission_mode == "unrestricted": env["CUA_DRIVER_DANGEROUSLY_BYPASS_APPROVALS"] = "1" return env + def _sanitized_env(self) -> Dict[str, str]: + from tools.environments.local import _sanitize_subprocess_env + + return _sanitize_subprocess_env(self.child_env()) + def _drain_stderr(self, process: Any) -> None: try: for line in getattr(process, "stderr", None) or (): @@ -169,8 +174,6 @@ class _EmbeddedCuaDaemon: pass def _serve_args(self) -> List[str]: - from tools.computer_use import cua_backend as _cb - serve_args = ["serve", "--embedded", "--socket", self.socket_path, "--no-permissions-gate", "--permission-mode", self.permission_mode] if self.permission_mode == "unrestricted": @@ -180,19 +183,16 @@ class _EmbeddedCuaDaemon: # The private daemon owns the cursor overlay, so the overlay policy must apply to this # long-lived serve process, not only its MCP proxy. Appended BEFORE the macOS app-launch # wrapping so the flag travels inside `open ... --args` with the rest of the serve args. - return _cb._mcp_args_with_overlay_flag(serve_args, driver_cmd=self._command) + return _cb()._mcp_args_with_overlay_flag(serve_args, driver_cmd=self._command) def start(self) -> None: if self._running: return - from tools.computer_use import cua_backend as _cb - from tools.environments.local import _sanitize_subprocess_env - - self._driver_cmd = self._driver_cmd or _cb.resolve_cua_driver_cmd() or "" + self._driver_cmd = self._driver_cmd or _cb().resolve_cua_driver_cmd() or "" if not self._driver_cmd: - raise RuntimeError(_cb.cua_driver_install_hint()) - self._command, self._mcp_args = _cb._resolve_mcp_invocation(self._driver_cmd) - env = _sanitize_subprocess_env(self.child_env()) + raise RuntimeError(_cb().cua_driver_install_hint()) + self._command, self._mcp_args = _cb()._resolve_mcp_invocation(self._driver_cmd) + env = self._sanitized_env() self._launch_via_app = sys.platform == "darwin" command = _embedded_daemon_spawn_command(self._command, self._serve_args(), platform=sys.platform) self._process = subprocess.Popen(command, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, @@ -219,13 +219,9 @@ class _EmbeddedCuaDaemon: def _socket_ready(self, env: Dict[str, str]) -> bool: """``cua-driver status --socket`` exits 0 once the private daemon accepts connections.""" - 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): - return False - return probe.returncode == 0 + probe = _cb()._run_quiet([self._command, "status", "--socket", self.socket_path], + timeout=2.0, env=env, swallow=_QUIET_ERRORS) + return probe is not None and probe.returncode == 0 def proxy_invocation(self) -> Tuple[str, List[str]]: if not self._running: @@ -237,14 +233,9 @@ class _EmbeddedCuaDaemon: owns_runtime, self._owns_runtime = self._owns_runtime, False self._running = False if owns_runtime: - 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 + _cb()._run_quiet([self._command, "stop", "--socket", self.socket_path], timeout=3.0, + stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, + env=self._sanitized_env(), swallow=_QUIET_ERRORS) if process is not None: _wait_or_kill(process) if sys.platform != "win32" and os.path.exists(self.socket_path): diff --git a/tools/computer_use/cua_backend_driver.py b/tools/computer_use/cua_backend_driver.py index 6ae4908799..81a4b656c4 100644 --- a/tools/computer_use/cua_backend_driver.py +++ b/tools/computer_use/cua_backend_driver.py @@ -36,6 +36,7 @@ _CUA_DRIVER_RUNTIME_CONTRACT_ARGS = { "stop": {"--socket"}, } _SEMVER_RE = re.compile(r"v?(\d+)\.(\d+)\.(\d+)(?:[-+].*)?") +_UPSTREAM_SCRIPTS = "https://raw.githubusercontent.com/trycua/cua/main/libs/cua-driver/scripts" def _cb(): """Origin module, looked up lazily so ``patch("tools.computer_use.cua_backend.X")`` applies.""" @@ -47,11 +48,8 @@ def _driver_json(driver_cmd: str, *args: str, timeout: float, require_ok: bool) """Run a driver verb and parse its stdout as a JSON object; None on spawn failure, empty stdout (older drivers print usage to stderr), unparseable or non-object output — and, with ``require_ok``, on a non-zero exit.""" - try: - proc = _cb()._run_driver(driver_cmd, *args, timeout=timeout) - except Exception: - return None - out = (proc.stdout or "").strip() + proc = _cb()._run_driver(driver_cmd, *args, timeout=timeout, swallow=Exception) + out = (proc.stdout or "").strip() if proc is not None else "" if not out or (require_ok and proc.returncode != 0): return None try: @@ -60,6 +58,11 @@ def _driver_json(driver_cmd: str, *args: str, timeout: float, require_ok: bool) return None return data if isinstance(data, dict) else None +def _valid_mcp_args(invocation: Any) -> Optional[List[str]]: + """``mcp_invocation.args`` when it is a list of strings (possibly empty), else None.""" + args = invocation.get("args") if isinstance(invocation, dict) else None + return args if isinstance(args, list) and all(isinstance(a, str) for a in args) else None + # --------------------------------------------------------------------------- # Binary resolution @@ -100,18 +103,12 @@ def _candidate_cua_driver_commands(override: Optional[str] = None) -> List[str]: home = os.path.expanduser("~") if sys.platform == "win32": local_app_data = os.environ.get("LOCALAPPDATA") or os.path.join(home, "AppData", "Local") - installed = [ - os.path.join(local_app_data, "Programs", "Cua", "cua-driver", "bin", "cua-driver.exe"), - os.path.join(home, ".local", "bin", "cua-driver.exe"), - os.path.join(home, ".local", "bin", "cua-driver"), - ] + installed = [os.path.join(local_app_data, "Programs", "Cua", "cua-driver", "bin", "cua-driver.exe"), + os.path.join(home, ".local", "bin", "cua-driver.exe"), + os.path.join(home, ".local", "bin", "cua-driver")] else: - installed = [ - 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", - ] + installed = [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 [_CUA_DRIVER_DEFAULT_CMD, *installed] def resolve_cua_driver_cmd(override: Optional[str] = None) -> Optional[str]: @@ -129,28 +126,20 @@ def cua_driver_binary_available() -> bool: return _cb().resolve_cua_driver_cmd() is not None def cua_driver_install_hint() -> str: - scripts = "https://raw.githubusercontent.com/trycua/cua/main/libs/cua-driver/scripts" - if sys.platform == "win32": - installer = f" irm {scripts}/install.ps1 | iex" - else: - installer = f' /bin/bash -c "$(curl -fsSL {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." - ) + installer = (f" irm {_UPSTREAM_SCRIPTS}/install.ps1 | iex" if sys.platform == "win32" + else f' /bin/bash -c "$(curl -fsSL {_UPSTREAM_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.") # --------------------------------------------------------------------------- # MCP invocation # --------------------------------------------------------------------------- -def _mcp_args_with_overlay_flag( - args: List[str], - driver_cmd: str = _CUA_DRIVER_DEFAULT_CMD, -) -> List[str]: +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 _cb()._cua_no_overlay() and _cb()._cua_driver_supports_no_overlay(driver_cmd): return [*args, "--no-overlay"] @@ -174,12 +163,9 @@ def _resolve_mcp_invocation(driver_cmd: str, *, timeout: float = 6.0) -> Tuple[s start over a failed discovery hop. ``--no-overlay`` appended when allowed.""" manifest = _driver_json(driver_cmd, "manifest", timeout=timeout, require_ok=True) or {} invocation = manifest.get("mcp_invocation") - invocation = invocation if isinstance(invocation, dict) else {} - args = invocation.get("args") - valid_args = isinstance(args, list) and all(isinstance(a, str) for a in args) - if not valid_args: - args = list(_CUA_DRIVER_ARGS) - command = invocation.get("command") if valid_args else None + args = _valid_mcp_args(invocation) + command = invocation.get("command") if args is not None and isinstance(invocation, dict) else None + args = list(_CUA_DRIVER_ARGS) if args is None else args if isinstance(command, str) and command: # Translate a Windows ``C:\...`` command for WSL BEFORE the separator # check (backslash is not a separator on POSIX). A generic ``cua-driver`` @@ -199,58 +185,53 @@ def _resolve_mcp_invocation(driver_cmd: str, *, timeout: float = 6.0) -> Tuple[s # cua-driver's native `check-update` verb compares the installed binary against # the latest GitHub release (cached ~20h); we prefer it over a hardcoded floor. -def cua_driver_runtime_contract_status(binary: Optional[str] = None) -> Dict[str, Any]: - """Report whether a local driver can host Hermes' 0.20 integration.""" - resolved = binary or _cb().resolve_cua_driver_cmd() - - def _not_ready(reason: str, version: Optional[str] = None) -> Dict[str, Any]: - return {"ready": False, "binary": resolved, "version": version, "reason": reason} - - if not resolved: - return _not_ready("cua-driver is not installed") - try: - result = _cb()._run_driver(resolved, "manifest", timeout=15.0 if sys.platform == "win32" else 5.0) - except (OSError, subprocess.SubprocessError) as exc: - return _not_ready(f"manifest check failed: {exc}") - if result.returncode != 0: - detail = (result.stderr or result.stdout or "manifest command failed").strip() - return _not_ready(detail.splitlines()[-1][:200]) - try: - manifest = json.loads(result.stdout or "") - except (TypeError, ValueError): - manifest = None - if not isinstance(manifest, dict): - return _not_ready("driver manifest is missing or invalid") - - raw_version = str(manifest.get("binary_version") or "").strip() - match = _SEMVER_RE.fullmatch(raw_version) - if not match: - return _not_ready("driver manifest does not report a semantic version", raw_version or None) - if tuple(int(part) for part in match.groups()) < _CUA_DRIVER_RUNTIME_CONTRACT_MIN: - return _not_ready("Hermes computer use requires cua-driver 0.20.0 or newer", raw_version) - - invocation = manifest.get("mcp_invocation") - invocation_args = invocation.get("args") if isinstance(invocation, dict) else None - if not (invocation_args and isinstance(invocation_args, list) - and all(isinstance(arg, str) for arg in invocation_args)): - return _not_ready("driver manifest does not provide an MCP launch command", raw_version) - +def _manifest_contract_gaps(manifest: Dict[str, Any]) -> List[str]: + """``" "`` entries the driver's advertised subcommands lack.""" advertised: Dict[str, set[str]] = { - command["name"]: { - arg["name"] for arg in command.get("args") or [] - if isinstance(arg, dict) and isinstance(arg.get("name"), str) - } + command["name"]: {arg["name"] for arg in command.get("args") or [] + if isinstance(arg, dict) and isinstance(arg.get("name"), str)} for command in manifest.get("subcommands") or [] if isinstance(command, dict) and isinstance(command.get("name"), str) } - missing = [ - f"{command} {arg}" - for command, required_args in _CUA_DRIVER_RUNTIME_CONTRACT_ARGS.items() - for arg in sorted(required_args - advertised.get(command, set())) - ] - if missing: - return _not_ready("driver manifest is missing: " + ", ".join(missing), raw_version) - return {"ready": True, "binary": resolved, "version": raw_version, "reason": ""} + return [f"{command} {arg}" for command, required in _CUA_DRIVER_RUNTIME_CONTRACT_ARGS.items() + for arg in sorted(required - advertised.get(command, set()))] + +def cua_driver_runtime_contract_status(binary: Optional[str] = None) -> Dict[str, Any]: + """Report whether a local driver can host Hermes' 0.20 integration.""" + resolved = binary or _cb().resolve_cua_driver_cmd() + version: Optional[str] = None + reason = "cua-driver is not installed" + if resolved: + try: + result = _cb()._run_driver(resolved, "manifest", timeout=15.0 if sys.platform == "win32" else 5.0) + except (OSError, subprocess.SubprocessError) as exc: + result, reason = None, f"manifest check failed: {exc}" + if result is not None and result.returncode != 0: + detail = (result.stderr or result.stdout or "manifest command failed").strip() + result, reason = None, detail.splitlines()[-1][:200] + if result is not None: + try: + manifest = json.loads(result.stdout or "") + except (TypeError, ValueError): + manifest = None + reason = _manifest_contract_reason(manifest if isinstance(manifest, dict) else None) + if isinstance(manifest, dict): + version = str(manifest.get("binary_version") or "").strip() or None + return {"ready": not reason, "binary": resolved, "version": version, "reason": reason} + +def _manifest_contract_reason(manifest: Optional[Dict[str, Any]]) -> str: + """Why a parsed manifest fails the 0.20 contract, or ``""`` when it passes.""" + if manifest is None: + return "driver manifest is missing or invalid" + match = _SEMVER_RE.fullmatch(str(manifest.get("binary_version") or "").strip()) + if not match: + return "driver manifest does not report a semantic version" + if tuple(int(part) for part in match.groups()) < _CUA_DRIVER_RUNTIME_CONTRACT_MIN: + return "Hermes computer use requires cua-driver 0.20.0 or newer" + if not _valid_mcp_args(manifest.get("mcp_invocation")): + return "driver manifest does not provide an MCP launch command" + missing = _manifest_contract_gaps(manifest) + return "driver manifest is missing: " + ", ".join(missing) if missing else "" def cua_driver_update_check(*, timeout: Optional[float] = None) -> Optional[Dict[str, Any]]: """``cua-driver check-update --json`` payload (``{current_version, @@ -274,9 +255,6 @@ def cua_driver_update_nudge() -> Optional[str]: state = _cb().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`." - ) + return (f"cua-driver {state.get('latest_version') or '?'} is available " + f"(you have {state.get('current_version') or '?'}); " + f"update with `hermes computer-use install --upgrade`.")