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

refactor(computer_use): unify cua quiet-subprocess probes via _run_quiet; table-drive discovery reasons; trim re-imports, contract gate, daemon teardown

Browse Source
This commit is contained in:
Teknium
2026-09-02 20:25:17 -07:00
parent a651429cf4
commit ea7e75f65a
3 changed files with 181 additions and 251 deletions
Download Patch File Download Diff File Expand all files Collapse all files

199
tools/computer_use/cua_backend.py
View File

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

71
tools/computer_use/cua_backend_daemon.py
View File

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

162
tools/computer_use/cua_backend_driver.py
View File

@@ -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]:
"""``"<verb> <flag>"`` 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`.")