mirrors/hermes-agent
1
0
Fork 0
mirror of https://github.com/NousResearch/hermes-agent.git synced 2026-07-20 15:33:54 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 7

Merge remote-tracking branch 'origin/main' into pr-50994

Browse source 
# Conflicts:
#	tools/computer_use/cua_backend.py
This commit is contained in:
Brooklyn Nicholson 2026-06-22 18:48:07 -05:00
commit 833710d33e
53 changed files with 3438 additions and 576 deletions
Download patch file Download diff file Expand all files Collapse all files

142
tools/computer_use/cua_backend.py
View file

@ -746,6 +746,28 @@ class _CuaDriverSession:
return capability in self._capabilities.get(tool, set())
return any(capability in caps for caps in self._capabilities.values())
def _has_tool(self, name: str) -> bool:
"""Return True when ``tools/list`` advertised a tool by this name.
Used to route capture(): cua-driver dropped the standalone
``screenshot`` tool and folded full-window PNG capture into
``get_window_state`` (whose own description notes it "Also captures
a PNG screenshot of the specified window"). Older drivers that still
expose ``screenshot`` keep using it; newer ones fall through to
``get_window_state``.
Returns False when discovery hasn't populated the map yet — callers
treat that as "unknown" and probe defensively rather than trusting it.
"""
return name in self._capabilities
@property
def capabilities_discovered(self) -> bool:
"""True once ``tools/list`` populated the per-tool map. When False,
``_has_tool`` answers are not trustworthy (discovery failed or the
session hasn't started) and capture() should probe defensively."""
return bool(self._capabilities)
@property
def capability_version(self) -> str:
"""Driver-advertised capability vocabulary version (empty string
@ -848,6 +870,45 @@ def _extract_tool_result(mcp_result: Any) -> Dict[str, Any]:
}
def _image_from_tool_result(out: Dict[str, Any]) -> tuple[Optional[str], Optional[str]]:
"""Pull a (png_b64, mime_type) pair out of a flattened tool result.
cua-driver delivers window screenshots in two shapes depending on tool +
transport:
* As an MCP ``image`` content part — surfaced by ``_extract_tool_result``
in ``out["images"]`` with a parallel ``image_mime_types`` entry. This
is what ``get_window_state`` emits over the stdio MCP transport.
* As a base64 field inside ``structuredContent`` —
``screenshot_png_b64`` (+ ``screenshot_mime_type``). This is what
``get_window_state`` returns when its structured payload carries the
image instead of a content part (newer driver builds; also the shape
seen via the ``cua-driver call`` CLI surface).
Checking both makes capture() robust to either delivery shape, so the
image never silently drops just because the driver moved it between the
content list and structuredContent. Returns ``(None, None)`` when neither
location carries an image.
"""
images = out.get("images") or []
if images and images[0]:
mimes = out.get("image_mime_types") or []
mime = mimes[0] if mimes and mimes[0] else None
return images[0], mime
structured = out.get("structuredContent") or {}
b64 = structured.get("screenshot_png_b64") or structured.get("png_b64")
if b64:
mime = (
structured.get("screenshot_mime_type")
or structured.get("mime_type")
or None
)
return b64, mime
return None, None
# ---------------------------------------------------------------------------
# The backend itself
# ---------------------------------------------------------------------------
@ -1062,28 +1123,61 @@ class CuaDriverBackend(ComputerUseBackend):
window_title = ""
if mode == "vision":
# Newer cua-driver releases no longer expose a standalone
# `screenshot` MCP tool. Request a screenshot-only capture via
# get_window_state instead; this keeps vision mode working while
# avoiding the AX walk used by som/ax captures.
sc_out = self._session.call_tool(
"get_window_state",
{
"pid": self._active_pid,
"window_id": self._active_window_id,
"capture_mode": "vision",
"session": self._session_id,
},
# Plain screenshot, no AX walk. cua-driver dropped the standalone
# `screenshot` tool (≥0.5.x) and folded full-window PNG capture
# into `get_window_state`. Route accordingly:
# * Driver advertises `screenshot` (older builds) → use it; it's
# the cheapest path (no AX tree walked server-side).
# * Otherwise (current drivers) → call `get_window_state` but
# DISCARD the AX tree/elements, returning only the PNG. Vision
# mode's whole contract is "just the pixels, no element noise",
# so we drop everything but the image.
# When capability discovery hasn't run (empty map), we don't trust
# a negative `_has_tool` answer — we still try `screenshot` first
# and fall back if the driver rejects it, so the path self-heals on
# any driver version.
use_screenshot = (
self._session._has_tool("screenshot")
or not self._session.capabilities_discovered
)
if sc_out["images"]:
png_b64 = sc_out["images"][0]
# Pick up the explicit mimeType cua-driver attaches to image
# parts (Surface 7). Empty string means the driver didn't
# carry one — callers will fall back to magic-byte sniffing.
mimes = sc_out.get("image_mime_types") or []
image_mime_type = mimes[0] if mimes and mimes[0] else None
sc_out: Optional[Dict[str, Any]] = None
if use_screenshot:
sc_out = self._session.call_tool(
"screenshot",
{
"window_id": self._active_window_id,
"format": "jpeg",
"quality": 85,
"session": self._session_id,
},
)
png_b64, image_mime_type = _image_from_tool_result(sc_out)
if not png_b64:
# Driver had no usable `screenshot` (e.g. "Unknown tool:
# screenshot" on ≥0.5.x, or an empty image part). Fall
# through to the get_window_state path below.
sc_out = None
if sc_out is None:
gws_out = self._session.call_tool(
"get_window_state",
{
"pid": self._active_pid,
"window_id": self._active_window_id,
"session": self._session_id,
},
)
png_b64, image_mime_type = _image_from_tool_result(gws_out)
# Still grab the window title — it's cheap and useful in the
# vision response — but deliberately leave `elements` empty so
# vision stays free of AX-tree noise.
text = gws_out["data"] if isinstance(gws_out["data"], str) else ""
_, tree = _split_tree_text(text)
wt = re.search(r'AXWindow\s+"([^"]+)"', tree)
if wt:
window_title = wt.group(1)
else:
# get_window_state: AX tree + optional screenshot.
# get_window_state: AX tree + screenshot.
gws_out = self._session.call_tool(
"get_window_state",
{
@ -1120,10 +1214,10 @@ class CuaDriverBackend(ComputerUseBackend):
if e.element_token
}
if gws_out["images"]:
png_b64 = gws_out["images"][0]
mimes = gws_out.get("image_mime_types") or []
image_mime_type = mimes[0] if mimes and mimes[0] else None
# Image may arrive as an MCP image part or inside
# structuredContent (screenshot_png_b64) depending on the driver
# build — _image_from_tool_result handles both.
png_b64, image_mime_type = _image_from_tool_result(gws_out)
# Extract window title from the AX tree first AXWindow line.
wt = re.search(r'AXWindow\s+"([^"]+)"', tree)

189
tools/computer_use/permissions.py Normal file
View file

@ -0,0 +1,189 @@
"""
Cross-platform Computer Use readiness + macOS permission helpers.
cua-driver runs on macOS, Windows, and Linux, but "ready to drive" means
something different on each:
* macOS — explicit TCC grants (Accessibility + Screen Recording). cua-driver
reports/requests them via ``permissions status`` / ``permissions grant``.
The grants attach to cua-driver's OWN identity (``com.trycua.driver`` /
the installed ``CuaDriver.app``), NOT Hermes — so no Hermes entitlement is
involved, and ``grant`` launches CuaDriver via LaunchServices so the macOS
dialog is attributed correctly.
* Windows — no TCC toggles; the UIAccess worker (``cua-driver-uia.exe``) may
trip a SmartScreen prompt on first run. Readiness == driver health.
* Linux — assistive control via the X11/XWayland stack. Readiness == driver
health.
The universal signal on every platform is ``cua-driver doctor --json`` (binary
integrity + platform support). ``computer_use_status`` folds that together with
the macOS permission detail into one payload for the desktop card, the
``hermes computer-use permissions`` CLI, and ``/api/tools/computer-use/status``.
"""
from __future__ import annotations
import json
import os
import shutil
import subprocess
import sys
from typing import Any, Dict, List, Optional
# Platforms with a cua-driver runtime backend (mirrors the toolset platform_gate).
_RUNTIME_PLATFORMS = frozenset({"darwin", "win32", "linux"})
_BOOLS = ("accessibility", "screen_recording", "screen_recording_capturable")
def _driver_cmd(override: Optional[str]) -> str:
if override:
return override
try:
from hermes_cli.tools_config import _cua_driver_cmd
return _cua_driver_cmd()
except Exception:
return os.environ.get("HERMES_CUA_DRIVER_CMD", "").strip() or "cua-driver"
def _child_env() -> Dict[str, str]:
"""cua-driver child env honoring the Hermes telemetry opt-in policy."""
try:
from tools.computer_use.cua_backend import cua_driver_child_env
return cua_driver_child_env()
except Exception:
return dict(os.environ)
def _run(binary: str, *args: str, timeout: float) -> subprocess.CompletedProcess:
return subprocess.run(
[binary, *args],
capture_output=True,
text=True,
timeout=timeout,
env=_child_env(),
stdin=subprocess.DEVNULL,
)
def _json_out(binary: str, *args: str, timeout: float) -> Any:
"""Run ``binary args`` and parse stdout as JSON, or ``None`` on any failure."""
raw = (_run(binary, *args, timeout=timeout).stdout or "").strip()
return json.loads(raw) if raw else None
def _doctor(binary: str) -> Optional[Dict[str, Any]]:
"""``cua-driver doctor --json`` → ``{ok, checks:[{label,status,message}]}``."""
try:
data = _json_out(binary, "doctor", "--json", timeout=12)
except Exception:
return None
if not isinstance(data, dict):
return None
checks: List[Dict[str, str]] = [
{
"label": str(p.get("label", "")),
"status": str(p.get("status", "")),
"message": str(p.get("message", "")),
}
for p in data.get("probes", [])
if isinstance(p, dict)
]
return {"ok": bool(data.get("ok")), "checks": checks}
def _mac_permissions(binary: str, out: Dict[str, Any]) -> None:
"""Fold ``cua-driver permissions status --json`` booleans into ``out``."""
try:
data = _json_out(binary, "permissions", "status", "--json", timeout=10)
except subprocess.TimeoutExpired:
out["error"] = "cua-driver permissions status timed out"
return
except Exception as exc: # spawn failure or malformed JSON
out["error"] = f"cua-driver permissions status failed: {exc}"
return
if isinstance(data, dict):
out.update({k: data[k] for k in _BOOLS if isinstance(data.get(k), bool)})
if isinstance(data.get("source"), dict):
out["source"] = data["source"]
def computer_use_status(driver_cmd: Optional[str] = None) -> Dict[str, Any]:
"""Unified, OS-aware Computer Use readiness for the desktop card.
``ready`` is the single signal the UI keys off: on macOS it's both TCC
grants; elsewhere it's driver health (no TCC model). ``None`` means
unknown (binary missing / probe failed). ``can_grant`` is macOS-only.
"""
plat = sys.platform
binary = shutil.which(_driver_cmd(driver_cmd))
out: Dict[str, Any] = {
"platform": plat,
"platform_supported": plat in _RUNTIME_PLATFORMS,
"installed": bool(binary),
"version": None,
"ready": None,
"can_grant": plat == "darwin",
"checks": [],
"source": None,
"error": None,
**{k: None for k in _BOOLS},
}
if not binary:
return out
try:
out["version"] = (_run(binary, "--version", timeout=5).stdout or "").strip() or None
except Exception:
pass
doctor = _doctor(binary)
if doctor is not None:
out["checks"] = doctor["checks"]
if plat == "darwin":
_mac_permissions(binary, out)
if out["error"] is None:
out["ready"] = out["accessibility"] is True and out["screen_recording"] is True
elif doctor is not None:
# No TCC model off macOS — readiness is driver health.
out["ready"] = doctor["ok"]
return out
def request_permissions_grant(driver_cmd: Optional[str] = None) -> int:
"""Run ``cua-driver permissions grant`` (macOS); stream its output.
Launches CuaDriver via LaunchServices so the TCC dialog is attributed to
``com.trycua.driver``, then waits for the grant. Returns the driver's exit
code (0 ok), 2 if the binary is missing, 64 on a non-macOS platform (which
has no TCC permission model to grant).
"""
if sys.platform != "darwin":
print("Computer Use permissions are a macOS concept; nothing to grant here.")
return 64
binary = shutil.which(_driver_cmd(driver_cmd))
if not binary:
print("cua-driver: not installed. Run: hermes computer-use install")
return 2
print(
"Requesting Accessibility + Screen Recording for CuaDriver.\n"
"macOS will show a dialog attributed to CuaDriver (com.trycua.driver) — "
"approve it, then return here."
)
try:
return int(
subprocess.run(
[binary, "permissions", "grant"],
env=_child_env(),
stdin=subprocess.DEVNULL,
).returncode
)
except KeyboardInterrupt: # pragma: no cover - interactive
return 130
except Exception as exc: # pragma: no cover - defensive
print(f"cua-driver permissions grant failed: {exc}", file=sys.stderr)
return 2