hermes-admin/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 33 Packages Projects Releases Wiki Activity

refactor(computer_use): compact cua_backend origin (best-effort helper, trimmed re-exports)

Browse Source
This commit is contained in:
Teknium
2026-09-02 15:54:29 -07:00
parent b083383770
commit 337267f00e
1 changed files with 55 additions and 92 deletions
Download Patch File Download Diff File Expand all files Collapse all files

147
tools/computer_use/cua_backend.py
View File

@@ -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"))