diff --git a/tools/computer_use/__init__.py b/tools/computer_use/__init__.py index 6d18e7af16..eb2a9714b1 100644 --- a/tools/computer_use/__init__.py +++ b/tools/computer_use/__init__.py @@ -1,26 +1,22 @@ """Computer use toolset — universal (any-model) desktop control via cua-driver. -Drives apps through cua-driver's background computer-use primitive (focus- -without-raise + pid-scoped event posting): it does NOT steal the user's cursor, -keyboard focus, or Space. The schema is plain OpenAI function-calling so every -tool-capable model can drive it; vision models get SOM captures (numbered -overlays + AX tree) and click by element index, non-vision models use the AX -tree alone. +Drives apps through cua-driver's background primitive (focus-without-raise + +pid-scoped event posting): it does NOT steal the user's cursor, keyboard focus, +or Space. Plain OpenAI function-calling schema; vision models get SOM captures +(numbered overlays + AX tree) and click by index, non-vision models use the AX +tree alone. Model-facing guidance lives in the schema description and each +action result's `verdict`. * `tool.py` — `computer_use` handler, approval gate, response shaping. * `backend.py` — abstract `ComputerUseBackend` + result dataclasses. * `cua_backend.py` — default backend (MCP over stdio to `cua-driver`), with `cua_backend_parse` / `_session` / `_daemon` siblings. * `schema.py` — the model-facing schema (byte-frozen). - -Model-facing guidance (workflow, background-first, escalate ladder, safety) -lives in the schema description and each action result's `verdict`. """ from __future__ import annotations -# Re-export the public surface so `from tools.computer_use import ...` works. -from tools.computer_use.tool import ( # noqa: F401 +from tools.computer_use.tool import ( # noqa: F401 (public re-exports) handle_computer_use, release_computer_use_session, set_approval_callback, diff --git a/tools/computer_use/backend.py b/tools/computer_use/backend.py index adf067efe3..ee255ef7a3 100644 --- a/tools/computer_use/backend.py +++ b/tools/computer_use/backend.py @@ -1,8 +1,8 @@ """Abstract backend interface for computer use. Any implementation (cua-driver over MCP, pyautogui, noop, future Linux/Windows) -must return the shape described below. All methods synchronous; async is -handled inside the backend implementation if needed. +returns the shapes below. All methods are synchronous; async is handled inside +the backend implementation if needed. """ from __future__ import annotations @@ -12,10 +12,7 @@ from abc import ABC, abstractmethod from dataclasses import dataclass, field from typing import Any, Dict, List, Optional, Tuple -_JPEG_SOF_MARKERS = frozenset({ - 0xC0, 0xC1, 0xC2, 0xC3, 0xC5, 0xC6, 0xC7, - 0xC9, 0xCA, 0xCB, 0xCD, 0xCE, 0xCF, -}) +_JPEG_SOF_MARKERS = frozenset({0xC0, 0xC1, 0xC2, 0xC3, 0xC5, 0xC6, 0xC7, 0xC9, 0xCA, 0xCB, 0xCD, 0xCE, 0xCF}) def image_dimensions_from_bytes(raw: bytes) -> Optional[Tuple[int, int]]: @@ -43,9 +40,7 @@ def image_dimensions_from_bytes(raw: bytes) -> Optional[Tuple[int, int]]: i += 1 if marker in {0xD8, 0xD9}: continue - if marker == 0xDA: - break - if i + 2 > len(raw): + if marker == 0xDA or i + 2 > len(raw): break segment_len = int.from_bytes(raw[i:i + 2], "big") if segment_len < 2 or i + segment_len > len(raw): @@ -70,12 +65,9 @@ class UIElement: pid: int = 0 # owning process PID window_id: int = 0 # SkyLight / CG window ID attributes: Dict[str, Any] = field(default_factory=dict) - # Opaque per-snapshot element handle from cua-driver - # (trycua/cua#1961 — Surface 6 of NousResearch/hermes-agent#47072). - # When set, downstream calls can pass it alongside `index` for - # explicit stale-detection: a stale token returns an error from - # cua-driver rather than silently re-resolving to a different - # element. None for pre-#1961 drivers that didn't carry the field. + # Opaque per-snapshot handle from cua-driver. Passed alongside `index` for explicit + # stale-detection: a stale token errors instead of silently re-resolving to a + # different element. None for older drivers that lack the field. element_token: Optional[str] = None def center(self) -> Tuple[int, int]: @@ -88,11 +80,9 @@ class CaptureResult: """Result of a screen capture call. At least one of png_b64 / elements is populated depending on capture mode: - * mode="vision" → png_b64 only - * mode="ax" → elements only - * mode="som" → both (default): PNG already has numbered overlays - drawn by the backend, and `elements` holds the - matching index → element mapping. + mode="vision" → png_b64 only; mode="ax" → elements only; mode="som" (default) + → both: the PNG already carries numbered overlays drawn by the backend and + `elements` holds the matching index → element mapping. """ mode: str @@ -100,20 +90,14 @@ class CaptureResult: height: int png_b64: Optional[str] = None elements: List[UIElement] = field(default_factory=list) - # Optional: the target app/window the elements were captured for. - app: str = "" + app: str = "" # target app/window the elements were captured for window_title: str = "" - # Raw bytes we sent to Anthropic, for token estimation. - png_bytes_len: int = 0 - # Explicit MIME type for `png_b64` when the backend supplied it - # (cua-driver-rs emits `mimeType` on every image part as of - # trycua/cua#1961 — Surface 7 of NousResearch/hermes-agent#47072). - # When None, downstream consumers fall back to base64-prefix - # sniffing for back-compat with older drivers. + png_bytes_len: int = 0 # raw bytes sent to Anthropic, for token estimation + # MIME type of `png_b64` when the backend supplied it (cua-driver-rs emits `mimeType` + # on every image part). None → consumers fall back to base64-prefix sniffing (older drivers). image_mime_type: Optional[str] = None - # Optional guidance appended to the human-readable summary — used by - # capture lanes that intentionally return no elements (e.g. full-screen - # composited grabs) to tell the model how to reach an interactive lane. + # Guidance appended to the summary by capture lanes that intentionally return no + # elements (e.g. full-screen composited grabs) to point the model at an interactive lane. note: str = "" @@ -121,45 +105,36 @@ class CaptureResult: class ActionResult: """Result of any action (click / type / scroll / drag / key / wait). - Beyond the transport-level ``ok`` flag, this carries cua-driver's - structured action verdict so the model can follow the documented - verify → escalate ladder (NousResearch/hermes-agent#67052). ``ok`` stays - tool/transport success only — it is NOT the semantic verdict. Read - ``effect`` / ``escalation`` to decide the next rung. All structured - fields are optional and additive: an older driver that omits - ``structuredContent`` leaves them ``None`` and behavior is unchanged. + ``ok`` is tool/transport success only — NOT the semantic verdict. Read + ``effect`` / ``escalation`` (cua-driver's structured verdict) to decide the + next rung of the verify → escalate ladder. All structured fields are optional + and additive: an older driver that omits ``structuredContent`` leaves them + ``None`` and behavior is unchanged. """ ok: bool action: str message: str = "" # human-readable summary - # Optional trailing screenshot — set when the caller asked for a - # post-action capture or the backend always returns one. - capture: Optional[CaptureResult] = None - # Arbitrary extra fields for debugging / telemetry. - meta: Dict[str, Any] = field(default_factory=dict) + capture: Optional[CaptureResult] = None # trailing screenshot, when requested / always-on + meta: Dict[str, Any] = field(default_factory=dict) # debugging / telemetry extras # ── cua-driver structured verdict (additive; None on old drivers) ── - # AX read-back verification: True = driver read the effect back, - # False = ran but unconfirmed, None = tool doesn't carry the field. - verified: Optional[bool] = None - # Confidence signal: "confirmed" | "unverifiable" | "suspected_noop". - effect: Optional[str] = None - # Machine-readable next-rung hint: {"recommended": "px"|"foreground"|"page", - # "reason": str} — present only when the driver recommends climbing. + verified: Optional[bool] = None # AX read-back: True confirmed, False unconfirmed, None n/a + effect: Optional[str] = None # "confirmed" | "unverifiable" | "suspected_noop" + # {"recommended": "px"|"foreground"|"page", "reason": str} — only when driver recommends climbing escalation: Optional[Dict[str, Any]] = None - # Delivery rung that actually ran (e.g. "ax", "x11_pixel", "cgevent_fg"). - path: Optional[str] = None - # True when an AX walk found no actionable elements (act by px instead). - degraded: Optional[bool] = None - # The delivery_mode the caller requested for this action, echoed back. - delivery_mode: Optional[str] = None - # A structured refusal code (e.g. "background_unavailable", - # "foreground_unsupported", "desktop_scope_disabled") when present. - code: Optional[str] = None + path: Optional[str] = None # delivery rung that ran (e.g. "ax", "x11_pixel", "cgevent_fg") + degraded: Optional[bool] = None # AX walk found no actionable elements (act by px instead) + delivery_mode: Optional[str] = None # the delivery_mode the caller requested, echoed back + code: Optional[str] = None # refusal code, e.g. "background_unavailable", "desktop_scope_disabled" class ComputerUseBackend(ABC): - """Lifecycle: `start()` before first use, `stop()` at shutdown.""" + """Lifecycle: `start()` before first use, `stop()` at shutdown. + + Pointer/keyboard actions take ``delivery_mode`` (background (default) | foreground) + and ``bring_to_front``; ``button`` is left | right | middle; ``modifiers`` a list of + key names. ``element`` args are 1-based SOM indices from a prior capture. + """ @abstractmethod def start(self) -> None: ... @@ -169,63 +144,30 @@ class ComputerUseBackend(ABC): @abstractmethod def is_available(self) -> bool: - """Return True if the backend can be used on this host right now. - - Used by check_fn gating and by the post-setup wizard. - """ + """True if the backend can be used on this host right now (check_fn gating, setup wizard).""" # ── Capture ───────────────────────────────────────────────────── @abstractmethod - def capture( - self, - mode: str = "som", - app: Optional[str] = None, - pid: Optional[int] = None, - window_id: Optional[int] = None, - ) -> CaptureResult: ... + def capture(self, mode: str = "som", app: Optional[str] = None, pid: Optional[int] = None, + window_id: Optional[int] = None) -> CaptureResult: ... # ── Pointer actions ───────────────────────────────────────────── @abstractmethod - def click( - self, - *, - element: Optional[int] = None, - x: Optional[int] = None, - y: Optional[int] = None, - button: str = "left", # left | right | middle - click_count: int = 1, - modifiers: Optional[List[str]] = None, - delivery_mode: Optional[str] = None, # background (default) | foreground - bring_to_front: bool = False, - ) -> ActionResult: ... + def click(self, *, element: Optional[int] = None, x: Optional[int] = None, y: Optional[int] = None, + button: str = "left", click_count: int = 1, modifiers: Optional[List[str]] = None, + delivery_mode: Optional[str] = None, bring_to_front: bool = False) -> ActionResult: ... @abstractmethod - def drag( - self, - *, - from_element: Optional[int] = None, - to_element: Optional[int] = None, - from_xy: Optional[Tuple[int, int]] = None, - to_xy: Optional[Tuple[int, int]] = None, - button: str = "left", - modifiers: Optional[List[str]] = None, - delivery_mode: Optional[str] = None, - bring_to_front: bool = False, - ) -> ActionResult: ... + def drag(self, *, from_element: Optional[int] = None, to_element: Optional[int] = None, + from_xy: Optional[Tuple[int, int]] = None, to_xy: Optional[Tuple[int, int]] = None, + button: str = "left", modifiers: Optional[List[str]] = None, + delivery_mode: Optional[str] = None, bring_to_front: bool = False) -> ActionResult: ... @abstractmethod - def scroll( - self, - *, - direction: str, # up | down | left | right - amount: int = 3, # wheel ticks - element: Optional[int] = None, - x: Optional[int] = None, - y: Optional[int] = None, - modifiers: Optional[List[str]] = None, - delivery_mode: Optional[str] = None, - bring_to_front: bool = False, - ) -> ActionResult: ... + def scroll(self, *, direction: str, amount: int = 3, element: Optional[int] = None, + x: Optional[int] = None, y: Optional[int] = None, modifiers: Optional[List[str]] = None, + delivery_mode: Optional[str] = None, bring_to_front: bool = False) -> ActionResult: + """`direction` is up | down | left | right; `amount` is wheel ticks.""" # ── Keyboard ──────────────────────────────────────────────────── @abstractmethod @@ -243,26 +185,18 @@ class ComputerUseBackend(ABC): """Return running apps with bundle IDs, PIDs, window counts.""" def list_windows(self) -> List[Dict[str, Any]]: - """Return visible native windows with PID and window identifiers. - - Optional compatibility hook: backends that predate window discovery - remain instantiable and simply report no windows. - """ + """Visible native windows with PID and window identifiers. Optional compatibility + hook: backends that predate window discovery stay instantiable and report none.""" return [] @abstractmethod def focus_app(self, app: str, raise_window: bool = False) -> ActionResult: """Route input to `app` (by name or bundle ID). Default: focus without raise.""" - # ── Native-value mutation ──────────────────────────────────────── @abstractmethod def set_value(self, value: str, element: Optional[int] = None) -> ActionResult: - """Set a native value on an element (e.g. AXPopUpButton selection). + """Set a native value on an element (e.g. AXPopUpButton selection).""" - `element` is the 1-based SOM index returned by a prior capture call. - """ - - # ── Timing ────────────────────────────────────────────────────── def wait(self, seconds: float) -> ActionResult: """Default implementation: time.sleep.""" import time diff --git a/tools/computer_use/doctor.py b/tools/computer_use/doctor.py index aa4f1c5eb5..b3393adb7a 100644 --- a/tools/computer_use/doctor.py +++ b/tools/computer_use/doctor.py @@ -1,10 +1,10 @@ """`hermes computer-use doctor` — thin client for cua-driver's `health_report` MCP tool. cua-driver owns the health model; we drive the stdio JSON-RPC handshake, call -`health_report` and render the stable ``schema_version="1"`` payload. cua-driver -0.10.x marks `health_report` risk-unclassified (isError=true, structuredContent -``{"exit_code": 1}``) — we detect that and synthesize a composite report from -working probes (check_permissions, list_apps, CLI --version). +`health_report` and render the stable ``schema_version="1"`` payload. cua-driver 0.10.x +marks `health_report` risk-unclassified (isError=true, structuredContent ``{"exit_code": 1}``) +— we detect that and synthesize a composite report from working probes +(check_permissions, list_apps, CLI --version). Exit codes: 0 overall=="ok"; 1 degraded/failed; 2 binary missing / protocol error. """ @@ -29,11 +29,10 @@ _OVERALL_GLYPH = {"ok": "✅", "degraded": "⚠️", "failed": "❌"} _SUPPORTED_PLATFORMS = ("darwin", "linux", "windows") _TCC_HINT = "Grant {} to CuaDriver in System Settings → Privacy & Security." _ZERO_DISPLAY_MSG = "ScreenCaptureKit reachable but 0 shareable display(s) — every capture will return 0x0." -_ZERO_DISPLAY_HINT = ( - "Wake the built-in display, connect a monitor or HDMI dummy dongle (e.g. Headless Ghost), " - "or enable a virtual display (Screen Sharing/VNC, BetterDisplay). " - "Verify with `system_profiler SPDisplaysDataType`." -) +_ZERO_DISPLAY_HINT = ("Wake the built-in display, connect a monitor or HDMI dummy dongle (e.g. Headless Ghost), " + "or enable a virtual display (Screen Sharing/VNC, BetterDisplay). " + "Verify with `system_profiler SPDisplaysDataType`.") +Report = Dict[str, Any] class HealthReportUnavailable(RuntimeError): @@ -78,7 +77,7 @@ def _cli_doctor_snippet(binary: str, timeout: float = 8.0) -> Optional[str]: return None return ((completed.stdout or "") + (completed.stderr or "")).strip() or None -def _build_identity(binary: str, report: Dict[str, Any]) -> Dict[str, Any]: +def _build_identity(binary: str, report: Report) -> Report: """Hermes-side identity block comparing resolved binary vs health_report.""" def token(text: str) -> str: # dotted version-ish token out of a free-form string m = text and re.search(r"(\d+\.\d+(?:\.\d+)?(?:[-+][\w.]+)?)", text) @@ -87,12 +86,8 @@ def _build_identity(binary: str, report: Dict[str, Any]) -> Dict[str, Any]: cli = _read_cli_version(binary) or "" report_v = str(report.get("driver_version") or "") cli_tok, report_tok = token(cli), token(report_v) - return { - "resolved_binary": binary, - "cli_version": cli or None, - "health_report_driver_version": report_v or None, - "version_mismatch": bool(cli_tok and report_tok and cli_tok != report_tok), - } + return {"resolved_binary": binary, "cli_version": cli or None, "health_report_driver_version": report_v or None, + "version_mismatch": bool(cli_tok and report_tok and cli_tok != report_tok)} # ── MCP transport ──────────────────────────────────────────────────────────── @@ -102,22 +97,22 @@ def _is_valid_health_report(payload: Any) -> bool: return (isinstance(payload, dict) and "schema_version" in payload and "overall" in payload and isinstance(payload.get("checks"), list)) -def _text_items(result: Dict[str, Any]) -> Iterator[str]: +def _text_items(result: Report) -> Iterator[str]: """Text of every ``{"type": "text"}`` content item of an MCP tools/call result.""" for item in result.get("content") or []: if isinstance(item, dict) and item.get("type") == "text": yield item.get("text") or "" -def _first_text(result: Dict[str, Any], default: str) -> str: +def _first_text(result: Report, default: str) -> str: """First non-empty text content item, else *default*.""" return next((t.strip() for t in _text_items(result) if t.strip()), default) -def _extract_health_report_from_result(result: Dict[str, Any]) -> Dict[str, Any]: +def _extract_health_report_from_result(result: Report) -> Report: """Pull a schema_version=1 report out of an MCP tools/call result. - Raises ``HealthReportUnavailable`` when the tool denied the call (isError) or - the payload is not a real report (0.10's ``{"exit_code": 1}``); ``RuntimeError`` - when the response carries no content at all. + Raises ``HealthReportUnavailable`` when the tool denied the call (isError) or the + payload is not a real report (0.10's ``{"exit_code": 1}``); ``RuntimeError`` when + the response carries no content at all. """ if result.get("isError") is True: raise HealthReportUnavailable(_first_text(result, "health_report returned isError=true")) @@ -149,10 +144,10 @@ def _stderr_tail(proc: subprocess.Popen) -> List[str]: return [str(x) for x in (proc.stderr.read() or "").strip().splitlines()[-3:]] return [] -def _mcp_rpc(proc: subprocess.Popen, msg_id: int, method: str, params: Any = None) -> Dict[str, Any]: +def _mcp_rpc(proc: subprocess.Popen, msg_id: int, method: str, params: Any = None) -> Report: """Write one JSON-RPC request and read one response line.""" assert proc.stdin is not None and proc.stdout is not None - payload: Dict[str, Any] = {"jsonrpc": "2.0", "id": msg_id, "method": method} + payload: Report = {"jsonrpc": "2.0", "id": msg_id, "method": method} if params is not None: payload["params"] = params proc.stdin.write(json.dumps(payload) + "\n") @@ -171,8 +166,7 @@ def _mcp_rpc(proc: subprocess.Popen, msg_id: int, method: str, params: Any = Non def _call_tool(proc: subprocess.Popen, msg_id: int, name: str, arguments: Any = None) -> Any: """tools/call *name* and return the raw ``result`` value (``{}`` when absent).""" - resp = _mcp_rpc(proc, msg_id, "tools/call", {"name": name, "arguments": arguments or {}}) - return resp.get("result") or {} + return _mcp_rpc(proc, msg_id, "tools/call", {"name": name, "arguments": arguments or {}}).get("result") or {} @contextmanager def _mcp_session(binary: str, timeout: float) -> Iterator[subprocess.Popen]: @@ -191,7 +185,7 @@ def _mcp_session(binary: str, timeout: float) -> Iterator[subprocess.Popen]: proc.wait() def _drive_health_report(binary: str, *, include: Sequence[str] = (), skip: Sequence[str] = (), - timeout: float = 12.0) -> Dict[str, Any]: + timeout: float = 12.0) -> Report: """Handshake + `health_report` → parsed report. Raises HealthReportUnavailable (denied / non-schema — caller falls back) or RuntimeError (protocol failure).""" args = {k: list(v) for k, v in (("include", include), ("skip", skip)) if v} @@ -205,7 +199,7 @@ def _drive_health_report(binary: str, *, include: Sequence[str] = (), skip: Sequ # ── 0.10 fallback: compose a report from working probes ────────────────────── -def _probe_tool(proc: subprocess.Popen, msg_id: int, name: str) -> Tuple[Optional[Dict[str, Any]], Optional[str]]: +def _probe_tool(proc: subprocess.Popen, msg_id: int, name: str) -> Tuple[Optional[Report], Optional[str]]: """``(result, None)`` on success; ``(None, error_text)`` on isError or RPC failure.""" try: result = _call_tool(proc, msg_id, name) @@ -215,21 +209,20 @@ def _probe_tool(proc: subprocess.Popen, msg_id: int, name: str) -> Tuple[Optiona return None, _first_text(result, f"{name} isError") return result, None -def _drive_fallback_probes(binary: str, *, timeout: float = 12.0) -> Dict[str, Any]: +def _drive_fallback_probes(binary: str, *, timeout: float = 12.0) -> Report: """Call working MCP tools (check_permissions, list_apps) in one session. Returns init_version (initialize serverInfo), permissions (structuredContent dict | None), permissions_error, list_apps_ok, list_apps_error, list_apps_count. """ - out: Dict[str, Any] = dict.fromkeys(("init_version", "permissions", "permissions_error", - "list_apps_ok", "list_apps_error", "list_apps_count")) + out: Report = dict.fromkeys(("init_version", "permissions", "permissions_error", + "list_apps_ok", "list_apps_error", "list_apps_count")) with _mcp_session(binary, timeout) as proc: init_resp = _mcp_rpc(proc, 1, "initialize", {}) server_info = ((init_resp.get("result") or {}).get("serverInfo") or {}) if isinstance(server_info, dict): out["init_version"] = server_info.get("version") - # check_permissions — primary TCC signal on 0.10 - perms, err = _probe_tool(proc, 2, "check_permissions") + perms, err = _probe_tool(proc, 2, "check_permissions") # primary TCC signal on 0.10 if perms is None: out["permissions_error"] = err else: @@ -250,17 +243,15 @@ def _platform_name() -> str: sysname = (_platform_mod.system() or "").lower() return sysname if sysname in _SUPPORTED_PLATFORMS else (sysname or "unknown") -def _check(name: str, status: str, message: str, **extra: Any) -> Dict[str, Any]: +def _check(name: str, status: str, message: str, **extra: Any) -> Report: """Build one health check dict (``hint`` / ``data`` only when given).""" return {"name": name, "status": status, "message": message, **extra} -def _tcc_checks(perms: Optional[Dict[str, Any]], perm_err: Optional[str], plat: str) -> List[Dict[str, Any]]: +def _tcc_checks(perms: Optional[Report], perm_err: Optional[str], plat: str) -> List[Report]: """tcc_accessibility + tcc_screen_recording checks from check_permissions output.""" if perms is None: - status = "fail" if perm_err else "skip" - msg = perm_err or "check_permissions unavailable" + status, msg = ("fail" if perm_err else "skip"), perm_err or "check_permissions unavailable" return [_check("tcc_accessibility", status, msg), _check("tcc_screen_recording", status, msg)] - # Only real booleans select a branch; anything else (missing/odd) is the "absent" row. ax, scr, capturable = (perms.get(k) for k in ("accessibility", "screen_recording", "screen_recording_capturable")) ax = ax if isinstance(ax, bool) else None @@ -288,7 +279,7 @@ def _tcc_checks(perms: Optional[Dict[str, Any]], perm_err: Optional[str], plat: return [_check("tcc_accessibility", ax_status, ax_msg, **ax_extra), _check("tcc_screen_recording", scr_status, scr_msg, **scr_extra)] -def _ax_capability_check(probes: Dict[str, Any], ax_granted: bool) -> Dict[str, Any]: +def _ax_capability_check(probes: Report, ax_granted: bool) -> Report: """ax_capability — inferred from list_apps success or the accessibility grant.""" list_ok, list_count = probes.get("list_apps_ok"), probes.get("list_apps_count") if list_ok is True: @@ -301,7 +292,7 @@ def _ax_capability_check(probes: Dict[str, Any], ax_granted: bool) -> Dict[str, return _check("ax_capability", "pass", "inferred from accessibility grant (list_apps not probed)") return _check("ax_capability", "skip", "not probed") -def _overall_from(checks: List[Dict[str, Any]]) -> str: +def _overall_from(checks: List[Report]) -> str: """failed if binary missing/bad; ok if accessibility fine and nothing failed; otherwise degraded (screen recording or accessibility problems).""" by_name = {c.get("name"): c.get("status") for c in checks} @@ -310,7 +301,7 @@ def _overall_from(checks: List[Dict[str, Any]]) -> str: ax_ok = by_name.get("tcc_accessibility") in ("pass", "skip", None) return "ok" if ax_ok and not any(c.get("status") == "fail" for c in checks) else "degraded" -def _compose_fallback_report(binary: str, *, reason: str = "", timeout: float = 12.0) -> Dict[str, Any]: +def _compose_fallback_report(binary: str, *, reason: str = "", timeout: float = 12.0) -> Report: """Build a schema_version=1 report from CLI + working MCP probes when ``health_report`` is denied (0.10) or non-schema. Renders via ``_print_text_report``.""" plat = _platform_name() @@ -322,13 +313,12 @@ def _compose_fallback_report(binary: str, *, reason: str = "", timeout: float = else: driver_version = ver_value if ver_status == "pass" else (ver_value or "?") ver_msg = f"cua-driver {ver_value}" if ver_status == "pass" else (ver_value or "version unknown") - supported = plat in _SUPPORTED_PLATFORMS perms = probes.get("permissions") if isinstance(probes.get("permissions"), dict) else None reason_short = (reason or "health_report unavailable").strip() if len(reason_short) > 160: reason_short = reason_short[:157] + "..." - checks: List[Dict[str, Any]] = [ + checks: List[Report] = [ _check("binary_version", ver_status, ver_msg), _check("platform_supported", "pass" if supported else "fail", f"platform={plat}" + ("" if supported else " (unsupported)")), @@ -347,14 +337,12 @@ def _compose_fallback_report(binary: str, *, reason: str = "", timeout: float = for c in checks: # normalize any accidental non-vocab status values if c.get("status") not in ("pass", "fail", "skip"): c["status"] = "fail" - return { - "schema_version": "1", "platform": plat, "driver_version": str(driver_version), - "overall": _overall_from(checks), "checks": checks, - "fallback": True, "fallback_reason": reason or "health_report unavailable", - } + return {"schema_version": "1", "platform": plat, "driver_version": str(driver_version), + "overall": _overall_from(checks), "checks": checks, + "fallback": True, "fallback_reason": reason or "health_report unavailable"} def _drive_health_report_or_fallback(binary: str, *, include: Sequence[str] = (), skip: Sequence[str] = (), - timeout: float = 12.0) -> Dict[str, Any]: + timeout: float = 12.0) -> Report: """Prefer real health_report; on denial/non-schema, synthesize via probes.""" try: report = _drive_health_report(binary, include=include, skip=skip, timeout=timeout) @@ -362,22 +350,21 @@ def _drive_health_report_or_fallback(binary: str, *, include: Sequence[str] = () report = _compose_fallback_report(binary, reason=str(e), timeout=timeout) return _apply_display_count_guard(report) -def _apply_display_count_guard(report: Dict[str, Any]) -> Dict[str, Any]: +def _apply_display_count_guard(report: Report) -> Report: """Downgrade an 'ok' report whose screen capture has zero displays. - macOS ScreenCaptureKit reports ``display_count=0`` on headless Macs and when - the built-in panel is asleep — TCC grants are fine, health_report can still - say pass/ok, but every capture comes back 0x0. Failing the check turns a - silent failure into an actionable one. Applied at the report seam so both - the real and the composed fallback path get it. + macOS ScreenCaptureKit reports ``display_count=0`` on headless Macs and when the + built-in panel is asleep — TCC grants are fine, health_report can still say + pass/ok, but every capture comes back 0x0. Failing the check turns a silent + failure into an actionable one. Applied at the report seam so both the real and + the composed fallback path get it. """ checks = report.get("checks") for check in checks if isinstance(checks, list) else (): if not isinstance(check, dict) or check.get("name") != "screen_capture_capability": continue data = check.get("data") - count = data.get("display_count") if isinstance(data, dict) else None - if count == 0 and check.get("status") == "pass": + if (data.get("display_count") if isinstance(data, dict) else None) == 0 and check.get("status") == "pass": check.update(status="fail", message=_ZERO_DISPLAY_MSG, hint=_ZERO_DISPLAY_HINT) if report.get("overall") == "ok": report["overall"] = "degraded" @@ -386,7 +373,7 @@ def _apply_display_count_guard(report: Dict[str, Any]) -> Dict[str, Any]: # ── Rendering ──────────────────────────────────────────────────────────────── -def _check_lines(check: Dict[str, Any], status_cols: Dict[str, str], reset: str, dim: str) -> List[str]: +def _check_lines(check: Report, status_cols: Dict[str, str], reset: str, dim: str) -> List[str]: """One line per check, plus indented hint and ``data`` rows (structured payload some checks attach — bundle id, AX state, version triple — support staff need it).""" status = check.get("status", "?") @@ -396,21 +383,17 @@ def _check_lines(check: Dict[str, Any], status_cols: Dict[str, str], reset: str, lines.append(f" → {dim}{check['hint']}{reset}") data = check.get("data") for key, value in (data.items() if isinstance(data, dict) else ()): - rendered = json.dumps(value) if isinstance(value, (dict, list)) else value - lines.append(f" {dim}{key}={rendered}{reset}") + lines.append(f" {dim}{key}={json.dumps(value) if isinstance(value, (dict, list)) else value}{reset}") return lines -def _print_text_report(report: Dict[str, Any], color: bool, *, identity: Optional[Dict[str, Any]] = None) -> None: +def _print_text_report(report: Report, color: bool, *, identity: Optional[Report] = None) -> None: """Render the report like `cua-driver call health_report` (one line per check). - With *identity* (resolved binary + ``--version``) the header prefers the CLI - version over health_report's ``driver_version`` and prints an identity block. - """ + version over health_report's ``driver_version`` and prints an identity block.""" platform, report_v, overall = (report.get(k, "?") for k in ("platform", "driver_version", "overall")) identity = identity or {} cli_v = identity.get("cli_version") or "" header_v = cli_v or report_v # binary's own --version wins when health_report is stale - # No external color library — inline ANSI keeps doctor self-contained. # Colors only apply when overall is a known vocabulary value. ansi = ("\033[31m", "\033[33m", "\033[32m", "\033[0m", "\033[2m") @@ -435,9 +418,9 @@ def run_doctor(driver_cmd: Optional[str] = None, *, include: Sequence[str] = (), json_output: bool = False, color: Optional[bool] = None) -> int: """Resolve the cua-driver binary, call `health_report`, render the result. - Honors `HERMES_CUA_DRIVER_CMD` via the shared runtime resolver, so doctor - diagnoses what `computer_use` will actually invoke. On 0.10.x (health_report - denied) it synthesizes a report from check_permissions / list_apps / CLI probes. + Honors `HERMES_CUA_DRIVER_CMD` via the shared runtime resolver, so doctor diagnoses + what `computer_use` will actually invoke. On 0.10.x (health_report denied) it + synthesizes a report from check_permissions / list_apps / CLI probes. """ # Windows' locale codec (cp1252, cp936, ...) cannot encode the ✅ ❌ ⚠️ ⏭️ glyphs — force UTF-8. for stream in (sys.stdout, sys.stderr): @@ -447,8 +430,7 @@ def run_doctor(driver_cmd: Optional[str] = None, *, include: Sequence[str] = (), binary = resolve_cua_driver_cmd(driver_cmd) if not binary: - looked_for = driver_cmd or "cua-driver (PATH and canonical install paths)" - print(f"cua-driver: not installed (looked for {looked_for!r}).") + print(f"cua-driver: not installed (looked for {driver_cmd or 'cua-driver (PATH and canonical install paths)'!r}).") print(" Run: hermes computer-use install") return 2 try: @@ -456,7 +438,6 @@ def run_doctor(driver_cmd: Optional[str] = None, *, include: Sequence[str] = (), except RuntimeError as e: print(f"cua-driver health_report failed: {e}", file=sys.stderr) return 2 - identity = _build_identity(binary, report) if json_output: # Additive envelope: upstream health_report keys preserved, Hermes identity @@ -465,5 +446,4 @@ def run_doctor(driver_cmd: Optional[str] = None, *, include: Sequence[str] = (), sys.stdout.write("\n") else: _print_text_report(report, color=sys.stdout.isatty() if color is None else bool(color), identity=identity) - # Unknown / missing overall after fallback must not look like success. - return 0 if report.get("overall") == "ok" else 1 + return 0 if report.get("overall") == "ok" else 1 # unknown/missing overall must not look like success diff --git a/tools/computer_use/permissions.py b/tools/computer_use/permissions.py index 11bb56d785..5c69d63e4e 100644 --- a/tools/computer_use/permissions.py +++ b/tools/computer_use/permissions.py @@ -1,18 +1,15 @@ -""" -Cross-platform Computer Use readiness + macOS permission helpers. +"""Cross-platform Computer Use readiness + macOS permission helpers. -"Ready to drive" differs per platform: - * macOS — explicit TCC grants (Accessibility + Screen Recording), reported / - requested via cua-driver ``permissions status`` / ``permissions grant``. The - grants attach to cua-driver's OWN identity (``com.trycua.driver``), not - Hermes, so ``grant`` launches CuaDriver via LaunchServices for correct - dialog attribution. - * Windows / Linux — no TCC toggles; readiness == driver health. +"Ready to drive" differs per platform: macOS needs explicit TCC grants +(Accessibility + Screen Recording) reported/requested via cua-driver +``permissions status`` / ``permissions grant``; Windows/Linux have no TCC +toggles, so readiness == driver health. The grants attach to cua-driver's OWN +identity (``com.trycua.driver``), not Hermes, so ``grant`` launches CuaDriver +via LaunchServices for correct dialog attribution. ``cua-driver doctor --json`` is the universal signal; ``computer_use_status`` -folds it with the macOS permission detail into one payload for the desktop -card, the ``hermes computer-use permissions`` CLI, and -``/api/tools/computer-use/status``. +folds it with the macOS detail into one payload for the desktop card, the +``hermes computer-use permissions`` CLI and ``/api/tools/computer-use/status``. """ from __future__ import annotations @@ -21,7 +18,7 @@ import json import os import subprocess import sys -from typing import Any, Dict, List, Optional +from typing import Any, Dict, Optional from hermes_cli._subprocess_compat import windows_hide_flags @@ -49,15 +46,9 @@ def _child_env() -> Dict[str, str]: def _run(binary: str, *args: str, timeout: float) -> subprocess.CompletedProcess: - return subprocess.run( - [binary, *args], - capture_output=True, - text=True, encoding='utf-8', errors='replace', - timeout=timeout, - env=_child_env(), - stdin=subprocess.DEVNULL, - creationflags=windows_hide_flags(), - ) + return subprocess.run([binary, *args], capture_output=True, text=True, encoding='utf-8', + errors='replace', timeout=timeout, env=_child_env(), + stdin=subprocess.DEVNULL, creationflags=windows_hide_flags()) def _json_out(binary: str, *args: str, timeout: float) -> Any: @@ -74,11 +65,8 @@ def _doctor(binary: str) -> Optional[Dict[str, Any]]: return None if not isinstance(data, dict): return None - checks: List[Dict[str, str]] = [ - {k: str(p.get(k, "")) for k in ("label", "status", "message")} - for p in data.get("probes", []) - if isinstance(p, dict) - ] + checks = [{k: str(p.get(k, "")) for k in ("label", "status", "message")} + for p in data.get("probes", []) if isinstance(p, dict)] return {"ok": bool(data.get("ok")), "checks": checks} @@ -115,16 +103,13 @@ def computer_use_status(driver_cmd: Optional[str] = None) -> Dict[str, Any]: } 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: @@ -143,17 +128,13 @@ def request_permissions_grant(driver_cmd: Optional[str] = None) -> int: if sys.platform != "darwin": print("Computer Use permissions are a macOS concept; nothing to grant here.") return 64 - binary = _resolve_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." - ) + 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) diff --git a/tools/computer_use/vision_routing.py b/tools/computer_use/vision_routing.py index 9dd546641b..7b60986f8d 100644 --- a/tools/computer_use/vision_routing.py +++ b/tools/computer_use/vision_routing.py @@ -1,26 +1,24 @@ """Vision-routing decisions for ``computer_use`` capture results. -``computer_use(action='capture', mode='som'|'vision')`` returns a ``_multimodal`` -envelope with the screenshot, delivered to the active session model as the tool -result. A text-only main model, or a provider that rejects multimodal content in -tool results, turns that into a hard 400/404 tool failure — even when a working -``auxiliary.vision`` model sits in config. This module decides: return the -screenshot as multimodal content, or pre-analyse it via aux vision so the main -model only ever sees text? +``capture`` (mode som|vision) returns a ``_multimodal`` screenshot envelope as the +tool result. A text-only main model, or a provider that rejects multimodal tool +results, turns that into a hard 400/404 — even with a working ``auxiliary.vision`` +model in config. This module decides: multimodal envelope, or pre-analyse via aux +vision so the main model only ever sees text? Decision order (mirrors ``vision_analyze``): 1. ``auxiliary.vision`` explicitly configured (provider not ""/"auto", or model / base_url set) → aux routing; users who pay for a vision model want it used. -2. User declared ``supports_vision`` for the active route (config escape hatch - for custom/local VLMs absent from models.dev) → honour it (True → multimodal). +2. User-declared ``supports_vision`` for the active route (escape hatch for + custom/local VLMs absent from models.dev) → honour it (True → multimodal). 3. Provider+model carries images inside tool-result messages AND models.dev says ``supports_vision=True`` → multimodal. 4. Everything else (non-vision model, provider rejecting multimodal tool results, lookup failure) → aux routing. -The decision fails *closed* toward aux routing when metadata is missing or -ambiguous: a screenshot sent to a model that cannot read it is a hard failure, -while aux routing costs one extra LLM call and yields a usable description. +Fails *closed* toward aux routing when metadata is missing or ambiguous: a +screenshot sent to a model that cannot read it is a hard failure, while aux +routing costs one extra LLM call and yields a usable description. """ from __future__ import annotations @@ -38,9 +36,7 @@ def _explicit_aux_vision_override(cfg: Optional[Dict[str, Any]]) -> bool: path and the user-attached-image path agree. ``provider: "auto"``, blank values, or a missing block all count as *not* explicit. """ - if not isinstance(cfg, dict): - return False - aux = cfg.get("auxiliary") or {} + aux = cfg.get("auxiliary") if isinstance(cfg, dict) else None vision = aux.get("vision") if isinstance(aux, dict) else None if not isinstance(vision, dict): return False @@ -50,12 +46,8 @@ def _explicit_aux_vision_override(cfg: Optional[Dict[str, Any]]) -> bool: return not (provider in ("", "auto") and not model and not base_url) -def _lookup_user_declared_supports_vision( - provider: str, - model: str, - cfg: Optional[Dict[str, Any]], -) -> Optional[bool]: - """Return config-declared ``supports_vision`` for the active route (None on failure).""" +def _lookup_user_declared_supports_vision(provider: str, model: str, cfg: Optional[Dict[str, Any]]) -> Optional[bool]: + """Config-declared ``supports_vision`` for the active route (None on failure).""" try: from agent.image_routing import _supports_vision_override @@ -65,12 +57,8 @@ def _lookup_user_declared_supports_vision( return None -def _lookup_supports_vision( - provider: str, - model: str, - cfg: Optional[Dict[str, Any]] = None, -) -> Optional[bool]: - """Return config/models.dev ``supports_vision`` for *(provider, model)*. +def _lookup_supports_vision(provider: str, model: str, cfg: Optional[Dict[str, Any]] = None) -> Optional[bool]: + """Config/models.dev ``supports_vision`` for *(provider, model)*. Prefers ``agent.image_routing._lookup_supports_vision``; falls back to raw models.dev capabilities only when that import is unavailable. Any lookup @@ -89,10 +77,7 @@ def _lookup_supports_vision( caps = get_model_capabilities(provider, model) except Exception as exc: # pragma: no cover - defensive - logger.debug( - "computer_use vision_routing: caps lookup failed for %s:%s — %s", - provider, model, exc, - ) + logger.debug("computer_use vision_routing: caps lookup failed for %s:%s — %s", provider, model, exc) return None return None if caps is None else bool(getattr(caps, "supports_vision", False)) @@ -100,9 +85,9 @@ def _lookup_supports_vision( def _provider_accepts_multimodal_tool_result(provider: str, model: str) -> Optional[bool]: """Whether *provider*+*model* carries images inside tool-result messages. - Reuses ``tools.vision_tools._supports_media_in_tool_results`` so this stays - in lockstep with the ``vision_analyze`` native fast path. Returns None on - import failure so callers fall back to aux routing rather than guessing. + Reuses ``tools.vision_tools._supports_media_in_tool_results`` to stay in + lockstep with the ``vision_analyze`` native fast path. None on import + failure so callers fall back to aux routing rather than guessing. """ if not provider: return None @@ -114,11 +99,7 @@ def _provider_accepts_multimodal_tool_result(provider: str, model: str) -> Optio return bool(_supports_media_in_tool_results(provider, model)) -def should_route_capture_to_aux_vision( - provider: str, - model: str, - cfg: Optional[Dict[str, Any]], -) -> bool: +def should_route_capture_to_aux_vision(provider: str, model: str, cfg: Optional[Dict[str, Any]]) -> bool: """True iff the captured screenshot should be pre-analysed via aux vision. *provider* is the lower-case canonical id, *model* the slug as sent to the @@ -127,19 +108,14 @@ def should_route_capture_to_aux_vision( """ if _explicit_aux_vision_override(cfg): return True - user_declared = _lookup_user_declared_supports_vision(provider, model, cfg) if user_declared is True: return False if user_declared is False: return True - if not _provider_accepts_multimodal_tool_result(provider, model): return True - return _lookup_supports_vision(provider, model, cfg) is not True -__all__ = [ - "should_route_capture_to_aux_vision", -] +__all__ = ["should_route_capture_to_aux_vision"]