diff --git a/tools/computer_use/cua_backend.py b/tools/computer_use/cua_backend.py index f02a9c6fa6..fe4ead2efd 100644 --- a/tools/computer_use/cua_backend.py +++ b/tools/computer_use/cua_backend.py @@ -6,10 +6,12 @@ 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_parse`` (pure parsing), ``cua_backend_session`` -(bridge + session + CLI fallback), ``cua_backend_daemon`` (private daemon + -macOS app identity). Moved names are re-imported here so -``patch("tools.computer_use.cua_backend.X")`` keeps working. +Siblings: ``cua_backend_driver`` (binary resolution, runtime contract, update +check), ``cua_backend_capture`` / ``cua_backend_input`` (backend mixins), +``cua_backend_parse`` (pure parsing), ``cua_backend_session`` (bridge + session ++ CLI fallback), ``cua_backend_daemon`` (private daemon + macOS app identity). +Moved names are re-imported here so ``patch("tools.computer_use.cua_backend.X")`` +keeps working; siblings look policy helpers up lazily through this module. """ from __future__ import annotations @@ -30,15 +32,16 @@ from tools.computer_use.cua_backend_capture import ( # noqa: F401 _linux_x11_active_window_id, _select_capture_target, ) +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, +) from tools.computer_use.cua_backend_driver import ( # noqa: F401 _CUA_DRIVER_ARGS, _CUA_DRIVER_CMD_ENV, - _CUA_DRIVER_DEFAULT_CMD, - _CUA_DRIVER_RUNTIME_CONTRACT_ARGS, - _CUA_DRIVER_RUNTIME_CONTRACT_MIN, - _candidate_cua_driver_commands, _cua_driver_supports_no_overlay, - _has_path_separator, _mcp_args_with_overlay_flag, _resolve_mcp_invocation, _wsl_windows_path_to_posix, @@ -50,41 +53,22 @@ from tools.computer_use.cua_backend_driver import ( # noqa: F401 resolve_cua_driver_cmd, ) from tools.computer_use.cua_backend_input import _InputMixin -from tools.computer_use.cua_backend_daemon import ( # noqa: F401 - _CUA_DRIVER_BUNDLE_ID, - _CUA_DRIVER_TEAM_IDS, - _EmbeddedCuaDaemon, - _embedded_daemon_spawn_command, - _resolve_cua_driver_app_path, - _validate_cua_driver_app_signature, -) from tools.computer_use.cua_backend_parse import ( # noqa: F401 - _ELEMENT_LINE_RE, - _MISSING, - _NON_APP_WINDOW_TITLE_PREFIXES, _action_result_from, - _apps_from_windows, _extract_tool_result, _image_dimensions_from_bytes, - _image_from_tool_result, _ingest_windows, _is_placeholder_id, - _is_real_app_window, - _mcp_field, _parse_elements_from_structured, _parse_elements_from_tree, _parse_key_combo, _parse_xprop_net_active_window, - _positive_int, - _split_tree_text, _windows_from_tool_result, - _z_index_uninformative, ) from tools.computer_use.cua_backend_session import _AsyncBridge, _CuaDriverSession # noqa: F401 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" @@ -153,9 +137,7 @@ 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") - if not isinstance(raw, str) or not raw.strip(): - return None - return raw.strip() + return raw.strip() if isinstance(raw, str) and raw.strip() else None def _manifest_is_mode_independent(path: str) -> bool: @@ -175,9 +157,7 @@ def _manifest_is_mode_independent(path: str) -> bool: except Exception: logger.debug("could not read capability manifest %s", path, exc_info=True) return False - if not isinstance(parsed, dict): - return False - version = parsed.get("version") + version = parsed.get("version") if isinstance(parsed, dict) else None return isinstance(version, int) and not isinstance(version, bool) and version >= 3 @@ -241,9 +221,13 @@ def _linux_session_locked() -> Optional[bool]: """ 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 = subprocess.run(["loginctl", "list-sessions", "--no-legend"], capture_output=True, - text=True, timeout=2.0, stdin=subprocess.DEVNULL) + proc = _loginctl("list-sessions", "--no-legend") if proc.returncode != 0: return None any_seat = False @@ -252,14 +236,13 @@ def _linux_session_locked() -> Optional[bool]: if len(parts) < 2 or "seat" not in line: continue any_seat = True - probe = subprocess.run(["loginctl", "show-session", parts[0], "-p", "LockedHint"], capture_output=True, - text=True, timeout=2.0, stdin=subprocess.DEVNULL) - if "LockedHint=no" in probe.stdout: + if "LockedHint=no" in _loginctl("show-session", parts[0], "-p", "LockedHint").stdout: return False return True if any_seat else None except Exception: return None + def _empty_discovery_reason() -> str: """One-line diagnosis for 'window discovery found nothing'.""" if _linux_session_locked() is True: @@ -290,13 +273,13 @@ def _empty_discovery_reason() -> str: # --------------------------------------------------------------------------- _update_checked = False - # One auto-repair attempt per process: when the runtime-contract gate fails # for something a reinstall fixes (old version, missing manifest verbs) run # the standard install path once instead of telling the user to. Guarded so a # failing installer can't loop — the second start() goes straight to the error. _contract_repair_attempted = False + def _maybe_repair_runtime_contract(contract: Dict[str, Any]) -> Dict[str, Any]: """Try one automatic driver repair; return the post-repair contract (or the original when no repair was attempted / it failed). Never raises. An @@ -329,6 +312,7 @@ def _maybe_repair_runtime_contract(contract: Dict[str, Any]) -> Dict[str, Any]: except Exception: return contract + 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.""" @@ -352,7 +336,6 @@ def _maybe_nudge_update() -> None: # The backend itself # --------------------------------------------------------------------------- - class CuaDriverBackend(_CaptureMixin, _InputMixin, ComputerUseBackend): """Default computer-use backend. Cross-platform via cua-driver MCP.""" @@ -426,10 +409,8 @@ class CuaDriverBackend(_CaptureMixin, _InputMixin, ComputerUseBackend): # Declare this run's identity. Non-fatal: cua-driver accepts anonymous # calls (the cursor just won't render), so 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) + 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() and tests that @@ -439,26 +420,20 @@ class CuaDriverBackend(_CaptureMixin, _InputMixin, ComputerUseBackend): # 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) + self._best_effort("set_config(max_image_dimension) failed", + self.set_config, max_image_dimension=max_dim) # Belt-and-suspenders when --no-overlay is unsupported or ignored. 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) + 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 first 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: - 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) + self._best_effort("end_session failed (continuing teardown)", + self._session.call_tool, "end_session", {"session": self._session_id}) try: self._session.stop() finally: @@ -468,11 +443,17 @@ class CuaDriverBackend(_CaptureMixin, _InputMixin, ComputerUseBackend): if self._embedded_daemon is not None: self._embedded_daemon.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: # 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() + return sys.platform in ("darwin", "win32", "linux") and cua_driver_binary_available() # ── Target state ─────────────────────────────────────────────── def _clear_active_target(self) -> None: @@ -491,8 +472,7 @@ class CuaDriverBackend(_CaptureMixin, _InputMixin, ComputerUseBackend): self._snapshot_tokens = {} self._last_target = {"pid": self._active_pid, "window_id": self._active_window_id} - - # ── App lifecycle ──────────────────────────────────────────────── + # ── App lifecycle / focus ───────────────────────────────────────── def launch_app( self, *, @@ -539,12 +519,12 @@ class CuaDriverBackend(_CaptureMixin, _InputMixin, ComputerUseBackend): keys pass through verbatim — cua-driver validates its own schema.""" return self._action("set_config", dict(config)) - # ── 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. ``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``).""" + """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) @@ -556,22 +536,11 @@ class CuaDriverBackend(_CaptureMixin, _InputMixin, ComputerUseBackend): when the snapshot was superseded. Gated on the per-tool capability so older drivers (``additionalProperties: false``) never see the field.""" 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 + token = self._snapshot_tokens.get(idx) if isinstance(idx, int) else None + if token and self._session.supports_capability("accessibility.element_tokens", tool=tool): + args["element_token"] = token - def _action( - self, - name: str, - args: Dict[str, Any], - *, - inject_session: bool = True, - ) -> ActionResult: + 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: @@ -581,22 +550,16 @@ class CuaDriverBackend(_CaptureMixin, _InputMixin, ComputerUseBackend): 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 + message = str(data.get("message", "")) if isinstance(data, dict) else data if isinstance(data, str) else "" if not message and isinstance(structured, dict): message = str(structured.get("message", "")) # Merge data + structuredContent into meta, structured winning on # 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, + for part in (data, structured): + if isinstance(part, dict): + meta.update(part) + return _action_result_from(name, not out["isError"], message, meta, structured, requested_delivery=args.get("delivery_mode"))