diff --git a/tools/computer_use/__init__.py b/tools/computer_use/__init__.py index 6d18e7af16..4de6c2bb6a 100644 --- a/tools/computer_use/__init__.py +++ b/tools/computer_use/__init__.py @@ -1,26 +1,19 @@ """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. - -* `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) +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`. + +Modules: `tool.py` (handler, approval gate, response shaping), `backend.py` (abstract +`ComputerUseBackend` + result dataclasses), `cua_backend.py` (default MCP-over-stdio +backend + `cua_backend_parse`/`_session`/`_daemon` siblings), `schema.py` (byte-frozen). """ 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..b2b334a507 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,18 +12,13 @@ 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]]: - """Return (width, height) for PNG / JPEG bytes, or None when unreadable. - - PNG: IHDR. JPEG: walk segments (skipping 0xFF fill bytes) to the first SOF - marker; stop at SOS. Used by the tool layer's provider min-size guard. - """ + """(width, height) for PNG / JPEG bytes, or None when unreadable. PNG: IHDR. JPEG: walk + segments (skipping 0xFF fill bytes) to the first SOF marker; stop at SOS. Used by the + tool layer's provider min-size guard.""" if raw.startswith(b"\x89PNG\r\n\x1a\n") and len(raw) >= 24: try: width, height = struct.unpack(">II", raw[16:24]) @@ -43,9 +38,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 +63,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]: @@ -85,35 +75,23 @@ class UIElement: @dataclass 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. - """ + """Result of a screen capture call. 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 width: int # screenshot width (logical px, pre-Anthropic-scale) 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 +99,35 @@ 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 pick the next rung of the + verify → escalate ladder. Structured fields are optional and additive: an older + driver that omits ``structuredContent`` leaves them ``None``, behavior 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 +137,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 +178,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 ee220b24cc..2d55e733c0 100644 --- a/tools/computer_use/doctor.py +++ b/tools/computer_use/doctor.py @@ -1,11 +1,10 @@ """`hermes computer-use doctor` — thin client for cua-driver's `health_report` MCP tool. -cua-driver owns the health model; this module drives the stdio JSON-RPC handshake, -calls `health_report`, and renders the response. The only contract is the stable -`schema_version="1"` payload shape. 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). +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). Exit codes: 0 overall=="ok"; 1 degraded/failed; 2 binary missing / protocol error. """ @@ -13,16 +12,15 @@ Exit codes: 0 overall=="ok"; 1 degraded/failed; 2 binary missing / protocol erro from __future__ import annotations import json -import os import platform as _platform_mod import re import subprocess import sys -from contextlib import contextmanager +from contextlib import contextmanager, suppress from typing import Any, Dict, Iterator, List, Optional, Sequence, Tuple from hermes_cli._subprocess_compat import windows_hide_flags - +from tools.computer_use.permissions import _child_env as _sanitized_cua_env # Match the ALLOWED_STATUS_VALUES + ALLOWED_OVERALL_VALUES the cua-driver # integration test pins. If health_report widens its vocabulary, add here. @@ -30,57 +28,30 @@ _STATUS_GLYPH = {"pass": "✅", "fail": "❌", "skip": "⏭️"} _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`.") +Report = Dict[str, Any] class HealthReportUnavailable(RuntimeError): """health_report denied or non-schema payload — ``run_doctor`` falls back to probes.""" -def _sanitized_cua_env() -> Dict[str, str]: - """cua-driver child env (telemetry policy + provider secrets stripped); - degrades to ``os.environ`` on import error so doctor keeps working.""" - try: - from tools.computer_use.cua_backend import sanitized_cua_driver_env - - return sanitized_cua_driver_env() - except Exception: - return dict(os.environ) - - -def _is_valid_health_report(payload: Any) -> bool: - """True when *payload* looks like a schema_version=1 health_report.""" - return ( - isinstance(payload, dict) - and "schema_version" in payload - and "overall" in payload - and isinstance(payload.get("checks"), list) - ) - +# ── CLI probes ─────────────────────────────────────────────────────────────── def _run_cli(binary: str, *args: str, timeout: float) -> subprocess.CompletedProcess: """Run `` args`` with UTF-8 capture + sanitized env (raises on failure).""" - return subprocess.run( - [binary, *args], capture_output=True, text=True, encoding="utf-8", - errors="replace", timeout=timeout, env=_sanitized_cua_env(), - ) - - -def _first_text(result: Dict[str, Any], default: str) -> str: - """First non-empty text content item of an MCP tools/call result, else *default*.""" - for item in result.get("content") or []: - if isinstance(item, dict) and item.get("type") == "text": - text = (item.get("text") or "").strip() - if text: - return text - return default + return subprocess.run([binary, *args], capture_output=True, text=True, encoding="utf-8", + errors="replace", timeout=timeout, env=_sanitized_cua_env()) +def _combined_output(completed: subprocess.CompletedProcess) -> str: + return ((completed.stdout or "") + (completed.stderr or "")).strip() def _read_cli_version(binary: str, *, timeout: float = 5.0) -> Optional[str]: - """Return ``cua-driver --version`` first line (stripped), or None on failure. - - health_report's ``driver_version`` can disagree with the actual binary - (observed on Windows); doctor surfaces both so operators are not misled. - """ + """First line of ``cua-driver --version`` or None. health_report's ``driver_version`` + can disagree with the real binary (seen on Windows); doctor surfaces both.""" try: completed = _run_cli(binary, "--version", timeout=timeout) except (OSError, subprocess.TimeoutExpired, ValueError, TypeError): @@ -88,100 +59,102 @@ def _read_cli_version(binary: str, *, timeout: float = 5.0) -> Optional[str]: text = (completed.stdout or completed.stderr or "").strip() return text.splitlines()[0].strip() if text else None +def _cli_driver_version(binary: str, timeout: float = 5.0) -> Tuple[str, Optional[str]]: + """Return (status, version_or_message) from ``cua-driver --version``.""" + try: + completed = _run_cli(binary, "--version", timeout=timeout) + except (OSError, subprocess.TimeoutExpired) as e: + return "fail", f"--version failed: {e}" + text, failed = _combined_output(completed), completed.returncode != 0 + if failed and not text: + return "fail", f"--version exited {completed.returncode}" + m = re.search(r"(\d+\.\d+\.\d+(?:[-+][\w.]+)?)", text) # typical: "cua-driver 0.10.0" + return ("fail" if failed else "pass"), m.group(1) if m else (text.splitlines()[0] if text else "unknown") -def _normalize_version_token(text: str) -> str: - """Pull a dotted version-ish token out of a free-form version string.""" - if not text: - return "" - m = re.search(r"(\d+\.\d+(?:\.\d+)?(?:[-+][\w.]+)?)", text) - return m.group(1) if m else text.strip().lower() +def _cli_doctor_snippet(binary: str, timeout: float = 8.0) -> Optional[str]: + """Optional one-shot ``cua-driver doctor`` text (best-effort, never fatal).""" + try: + return _combined_output(_run_cli(binary, "doctor", timeout=timeout)) or None + except (OSError, subprocess.TimeoutExpired): + return 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) + return m.group(1) if m else text.strip().lower() + cli = _read_cli_version(binary) or "" report_v = str(report.get("driver_version") or "") - cli_tok = _normalize_version_token(cli) - report_tok = _normalize_version_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), - } + 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)} -def _extract_health_report_from_result(result: Dict[str, Any]) -> Dict[str, Any]: +# ── MCP transport ──────────────────────────────────────────────────────────── + +def _is_valid_health_report(payload: Any) -> bool: + """True when *payload* looks like a schema_version=1 health_report.""" + return isinstance(payload, dict) and {"schema_version", "overall"} <= payload.keys() and isinstance(payload.get("checks"), list) + +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: 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: 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 health 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")) - sc = result.get("structuredContent") if _is_valid_health_report(sc): return sc # type: ignore[return-value] - - # Older builds: JSON text block with schema_version. - for item in result.get("content") or []: - if not isinstance(item, dict) or item.get("type") != "text": - continue - try: - parsed = json.loads(item.get("text", "")) - except (ValueError, TypeError): - continue - if _is_valid_health_report(parsed): - return parsed - - # structuredContent present but not a real report — unavailable, not fatal protocol. - if isinstance(sc, dict): - raise HealthReportUnavailable( - "health_report structuredContent lacks schema_version/overall/checks " - f"(keys={sorted(sc.keys())})" - ) - raise RuntimeError( - "health_report response carried neither structuredContent nor a parseable " - f"JSON text block. Result keys: {list(result.keys())}" - ) - + for text in _text_items(result): # older builds: JSON text block with schema_version + with suppress(ValueError, TypeError): + parsed = json.loads(text) + if _is_valid_health_report(parsed): + return parsed + if isinstance(sc, dict): # present but not a real report — unavailable, not fatal + raise HealthReportUnavailable("health_report structuredContent lacks schema_version/overall/checks " + f"(keys={sorted(sc.keys())})") + raise RuntimeError("health_report response carried neither structuredContent nor a parseable " + f"JSON text block. Result keys: {list(result.keys())}") def _open_mcp(binary: str) -> subprocess.Popen: - """Spawn `` mcp`` with UTF-8 + sanitized env. + """Spawn `` mcp``. cua-driver emits UTF-8 (emoji, arbitrary paths); the + locale default (`cp1252` on Windows) would raise UnicodeDecodeError, so pin the codec.""" + return subprocess.Popen([binary, "mcp"], stdin=subprocess.PIPE, stdout=subprocess.PIPE, + stderr=subprocess.PIPE, text=True, encoding="utf-8", errors="replace", + bufsize=1, creationflags=windows_hide_flags(), env=_sanitized_cua_env()) - cua-driver emits UTF-8 (emoji, arbitrary paths); the locale default - (`cp1252` on Windows) would raise UnicodeDecodeError, so pin the codec. - """ - return subprocess.Popen( - [binary, "mcp"], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, - text=True, encoding="utf-8", errors="replace", bufsize=1, - creationflags=windows_hide_flags(), env=_sanitized_cua_env(), - ) +def _stderr_tail(proc: subprocess.Popen) -> List[str]: + """Last 3 stderr lines of *proc* (best-effort, ``[]`` when unreadable).""" + with suppress(Exception): + if proc.stderr is not None: + 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") proc.stdin.flush() line = proc.stdout.readline() if not line: - stderr_tail: List[str] = [] - if proc.stderr is not None: - try: - raw_err = proc.stderr.read() or "" - stderr_tail = [str(x) for x in raw_err.strip().splitlines()[-3:]] - except Exception: - pass - raise RuntimeError( - f"cua-driver mcp produced no response for {method!r}. " - f"stderr tail: {stderr_tail or '(empty)'}" - ) + raise RuntimeError(f"cua-driver mcp produced no response for {method!r}. stderr tail: {_stderr_tail(proc) or '(empty)'}") try: resp = json.loads(line) except (ValueError, TypeError) as e: @@ -190,12 +163,9 @@ def _mcp_rpc(proc: subprocess.Popen, msg_id: int, method: str, params: Any = Non raise RuntimeError(f"{method} JSON-RPC error: {resp['error']}") return resp - 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]: @@ -204,25 +174,19 @@ def _mcp_session(binary: str, timeout: float) -> Iterator[subprocess.Popen]: try: yield proc finally: - try: + with suppress(Exception): if proc.stdin is not None: proc.stdin.close() - except Exception: - pass try: proc.wait(timeout=timeout) except subprocess.TimeoutExpired: proc.kill() proc.wait() - def _drive_health_report(binary: str, *, include: Sequence[str] = (), skip: Sequence[str] = (), - timeout: float = 12.0) -> Dict[str, Any]: - """Spawn ` mcp`, handshake, call `health_report`, return the parsed report. - - Raises HealthReportUnavailable (denied / non-schema — caller falls back) or - RuntimeError (protocol-level failure). - """ + 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} with _mcp_session(binary, timeout) as proc: _mcp_rpc(proc, 1, "initialize", {}) @@ -232,106 +196,71 @@ def _drive_health_report(binary: str, *, include: Sequence[str] = (), skip: Sequ return _extract_health_report_from_result(result) -def _cli_driver_version(binary: str, timeout: float = 5.0) -> Tuple[str, Optional[str]]: - """Return (status, version_or_message) from ``cua-driver --version``.""" +# ── 0.10 fallback: compose a report from working probes ────────────────────── + +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: - completed = _run_cli(binary, "--version", timeout=timeout) - except (OSError, subprocess.TimeoutExpired) as e: - return "fail", f"--version failed: {e}" + result = _call_tool(proc, msg_id, name) + except RuntimeError as e: + return None, str(e) + if result.get("isError") is True: + return None, _first_text(result, f"{name} isError") + return result, None - text = ((completed.stdout or "") + (completed.stderr or "")).strip() - if completed.returncode != 0 and not text: - return "fail", f"--version exited {completed.returncode}" - - # Typical: "cua-driver 0.10.0" - m = re.search(r"(\d+\.\d+\.\d+(?:[-+][\w.]+)?)", text) - version = m.group(1) if m else (text.splitlines()[0] if text else "unknown") - return ("fail" if completed.returncode != 0 else "pass"), version - - -def _cli_doctor_snippet(binary: str, timeout: float = 8.0) -> Optional[str]: - """Optional one-shot ``cua-driver doctor`` text (best-effort, never fatal).""" - try: - completed = _run_cli(binary, "doctor", timeout=timeout) - except (OSError, subprocess.TimeoutExpired): - return None - return ((completed.stdout or "") + (completed.stderr or "")).strip() or 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 - try: - perm_result = _call_tool(proc, 2, "check_permissions") - if perm_result.get("isError") is True: - out["permissions_error"] = _first_text(perm_result, "check_permissions isError") - else: - sc = perm_result.get("structuredContent") - out["permissions"] = sc if isinstance(sc, dict) else {} - except RuntimeError as e: - out["permissions_error"] = str(e) - + perms, err = _probe_tool(proc, 2, "check_permissions") # primary TCC signal on 0.10 + if perms is None: + out["permissions_error"] = err + else: + sc = perms.get("structuredContent") + out["permissions"] = sc if isinstance(sc, dict) else {} # list_apps — light AX capability probe; text-only success still counts as AX working - try: - apps_result = _call_tool(proc, 3, "list_apps") - if apps_result.get("isError") is True: - out["list_apps_ok"] = False - out["list_apps_error"] = _first_text(apps_result, "list_apps isError") - else: - sc = apps_result.get("structuredContent") or {} - apps = sc.get("apps") if isinstance(sc, dict) else None - out["list_apps_ok"] = True - out["list_apps_count"] = len(apps) if isinstance(apps, list) else None - except RuntimeError as e: - out["list_apps_ok"] = False - out["list_apps_error"] = str(e) + apps, err = _probe_tool(proc, 3, "list_apps") + out["list_apps_ok"] = apps is not None + if apps is None: + out["list_apps_error"] = err + else: + sc = apps.get("structuredContent") or {} + app_list = sc.get("apps") if isinstance(sc, dict) else None + out["list_apps_count"] = len(app_list) if isinstance(app_list, list) else None return out - def _platform_name() -> str: - sysname = (_platform_mod.system() or "").lower() - return sysname if sysname in _SUPPORTED_PLATFORMS else (sysname or "unknown") + return (_platform_mod.system() or "").lower() 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 scr = scr if isinstance(scr, bool) else None - ax_rows = { True: ("pass", "Accessibility is granted.", {"data": {"accessibility": True}}), False: ("fail", "Accessibility is not granted.", {"hint": _TCC_HINT.format("Accessibility"), "data": {"accessibility": False}}), None: ("skip", "accessibility field absent from check_permissions", {}), } - scr_rows = { - # (scr, capturable is False) — the granted-but-not-capturable row wins first. + scr_rows = { # (scr, capturable is False) — the granted-but-not-capturable row wins first. (True, True): ("fail", "Screen Recording granted but not capturable.", {"hint": "Screen Recording permission may need a restart of CuaDriver " "or a re-grant in System Settings.", @@ -345,16 +274,11 @@ def _tcc_checks(perms: Optional[Dict[str, Any]], perm_err: Optional[str], plat: } ax_status, ax_msg, ax_extra = ax_rows[ax] scr_status, scr_msg, scr_extra = scr_rows[(scr, capturable is False and scr is True)] - return [ - _check("tcc_accessibility", ax_status, ax_msg, **ax_extra), - _check("tcc_screen_recording", scr_status, scr_msg, **scr_extra), - ] + 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 = probes.get("list_apps_ok") - list_count = probes.get("list_apps_count") + list_ok, list_count = probes.get("list_apps_ok"), probes.get("list_apps_count") if list_ok is True: count_msg = f" ({list_count} apps)" if isinstance(list_count, int) else "" return _check("ax_capability", "pass", f"list_apps succeeded{count_msg}") @@ -365,214 +289,153 @@ 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[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} + if by_name.get("binary_version") != "pass": + return "failed" + 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]: - """Build a schema_version=1 report from CLI + working MCP probes. - - Used when ``health_report`` is denied (unclassified risk on 0.10) or - returns a non-schema payload. Compatible with ``_print_text_report``. - """ +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() - ver_status, ver_value = _cli_driver_version(binary) - driver_version = ver_value if ver_status == "pass" else (ver_value or "?") - # Prefer MCP initialize version when CLI parse is messy probes = _drive_fallback_probes(binary, timeout=timeout) - if probes.get("init_version"): - driver_version = str(probes["init_version"]) - ver_status = "pass" + if probes.get("init_version"): # MCP initialize version beats a messy CLI parse + ver_status, driver_version = "pass", str(probes["init_version"]) ver_msg = f"cua-driver {driver_version}" 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 - checks: List[Dict[str, Any]] = [ + 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[Report] = [ _check("binary_version", ver_status, ver_msg), _check("platform_supported", "pass" if supported else "fail", f"platform={plat}" + ("" if supported else " (unsupported)")), # doctor does not start a session, so session_active is never probed _check("session_active", "skip", "not probed (doctor does not open a cua session)"), + *_tcc_checks(perms, probes.get("permissions_error"), plat), + _ax_capability_check(probes, bool(perms and perms.get("accessibility") is True)), + _check("health_report_path", "skip", + f"fallback composite (cua-driver 0.10 unclassified health_report); cause: {reason_short}"), ] - - perms = probes.get("permissions") if isinstance(probes.get("permissions"), dict) else None - checks += _tcc_checks(perms, probes.get("permissions_error"), plat) - checks.append(_ax_capability_check(probes, bool(perms and perms.get("accessibility") is True))) - - # Annotate that we used the fallback path - reason_short = (reason or "health_report unavailable").strip() - if len(reason_short) > 160: - reason_short = reason_short[:157] + "..." - checks.append(_check( - "health_report_path", "skip", - f"fallback composite (cua-driver 0.10 unclassified health_report); cause: {reason_short}", - )) - - # Optional CLI doctor text (best-effort) - doctor_txt = _cli_doctor_snippet(binary) + doctor_txt = _cli_doctor_snippet(binary) # optional CLI doctor text (best-effort) if doctor_txt: cli_ok = "[ok" in doctor_txt.lower() or "ok ]" in doctor_txt checks.append(_check("cli_doctor", "pass" if cli_ok else "skip", doctor_txt.splitlines()[0].strip(), data={"snippet": doctor_txt[:2000]})) - - # Normalize any accidental non-vocab status values - for c in checks: + 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"} - # overall: failed if binary missing/bad; ok if accessibility fine and nothing - # failed; otherwise degraded (screen recording or accessibility problems). - status_by_name = {c.get("name"): c.get("status") for c in checks} - fail_count = sum(1 for c in checks if c.get("status") == "fail") - if status_by_name.get("binary_version") != "pass": - overall = "failed" - elif status_by_name.get("tcc_accessibility") in ("pass", "skip", None) and fail_count == 0: - overall = "ok" - else: - overall = "degraded" - - return { - "schema_version": "1", - "platform": plat, - "driver_version": str(driver_version), - "overall": overall, - "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]: - """Prefer real health_report; on denial/non-schema, synthesize via probes.""" - try: - report = _drive_health_report(binary, include=include, skip=skip, timeout=timeout) - except HealthReportUnavailable as e: - 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 will come back 0x0. Marking the - check failed 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") - if not isinstance(checks, list): - return report - for check in 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": - check["status"] = "fail" - check["message"] = "ScreenCaptureKit reachable but 0 shareable display(s) — every capture will return 0x0." - check["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`." - ) + 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" return report -def _print_text_report(report: Dict[str, Any], color: bool, *, identity: Optional[Dict[str, Any]] = None) -> None: - """Render the report like `cua-driver call health_report` (one line per check). +# ── Rendering ──────────────────────────────────────────────────────────────── +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", "?") + lines = [f" {_STATUS_GLYPH.get(status, '•')} {status_cols.get(status, '')}{check.get('name', '?')}{reset}: " + f"{check.get('message') or ''}"] + if check.get("hint"): + lines.append(f" → {dim}{check['hint']}{reset}") + data = check.get("data") + for key, value in (data.items() if isinstance(data, dict) else ()): + lines.append(f" {dim}{key}={json.dumps(value) if isinstance(value, (dict, list)) else value}{reset}") + return lines + +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. - """ - platform = report.get("platform", "?") - report_v = report.get("driver_version", "?") - overall = report.get("overall", "?") + 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 "" - mismatch = bool(identity.get("version_mismatch")) header_v = cli_v or report_v # binary's own --version wins when health_report is stale - - # No external color library — keep ANSI inline so doctor stays self-contained. + # 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") red, yellow, green, reset, dim = ansi if color and overall in _OVERALL_GLYPH else ("",) * 5 col_for = {"failed": red, "degraded": yellow, "ok": green}.get(overall, "") status_cols = {"pass": green, "fail": red, "skip": dim} - print(f"{_OVERALL_GLYPH.get(overall, '•')} cua-driver {header_v} on {platform} — {col_for}{overall}{reset}") + lines = [f"{_OVERALL_GLYPH.get(overall, '•')} cua-driver {header_v} on {platform} — {col_for}{overall}{reset}"] if identity.get("resolved_binary"): - print(f" {dim}binary: {identity['resolved_binary']}{reset}") + lines.append(f" {dim}binary: {identity['resolved_binary']}{reset}") if cli_v and report_v and str(report_v) not in str(cli_v) and str(cli_v) not in str(report_v): # Only annotate when the free-form strings clearly differ. - print(f" {dim}--version: {cli_v}{reset}") - print(f" {dim}health_report.driver_version: {report_v}{reset}") - if mismatch: - print(f" {yellow}⚠️ version mismatch: health_report says {report_v!r} but binary --version is {cli_v!r}{reset}") - print(f" {dim}→ trust --version / packages/current for debugging; health_report's binary_version check can lag on Windows{reset}") - + lines += [f" {dim}--version: {cli_v}{reset}", f" {dim}health_report.driver_version: {report_v}{reset}"] + if identity.get("version_mismatch"): + lines += [f" {yellow}⚠️ version mismatch: health_report says {report_v!r} but binary --version is {cli_v!r}{reset}", + f" {dim}→ trust --version / packages/current for debugging; health_report's binary_version check can lag on Windows{reset}"] for check in report.get("checks", []): - name = check.get("name", "?") - status = check.get("status", "?") - glyph = _STATUS_GLYPH.get(status, "•") - message = check.get("message") or "" - print(f" {glyph} {status_cols.get(status, '')}{name}{reset}: {message}") - hint = check.get("hint") - if hint: - print(f" → {dim}{hint}{reset}") - # `data` is the structured payload some checks attach (bundle id, AX - # state, version triple) — users / support staff frequently need it. - data = check.get("data") - if isinstance(data, dict) and data: - for key, value in data.items(): - rendered = json.dumps(value) if isinstance(value, (dict, list)) else value - print(f" {dim}{key}={rendered}{reset}") - + lines += _check_lines(check, status_cols, reset, dim) + print("\n".join(lines)) def run_doctor(driver_cmd: Optional[str] = None, *, include: Sequence[str] = (), skip: 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): - try: + with suppress(AttributeError, OSError): stream.reconfigure(encoding="utf-8", errors="replace") # type: ignore[union-attr] - except (AttributeError, OSError): - pass from tools.computer_use.cua_backend import resolve_cua_driver_cmd 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: - report = _drive_health_report_or_fallback(binary, include=include, skip=skip) + try: # prefer real health_report; on denial/non-schema, synthesize via probes + try: + report = _drive_health_report(binary, include=include, skip=skip, timeout=12.0) + except HealthReportUnavailable as e: + report = _compose_fallback_report(binary, reason=str(e), timeout=12.0) except RuntimeError as e: print(f"cua-driver health_report failed: {e}", file=sys.stderr) return 2 - + report = _apply_display_count_guard(report) identity = _build_identity(binary, report) - if json_output: # Additive envelope: upstream health_report keys preserved, Hermes identity # under hermes_identity so parsers that only read overall/checks keep working. - payload = dict(report) - payload["hermes_identity"] = identity - json.dump(payload, sys.stdout, indent=2, sort_keys=True) + json.dump({**report, "hermes_identity": identity}, sys.stdout, indent=2, sort_keys=True) sys.stdout.write("\n") else: - if color is None: - color = sys.stdout.isatty() - _print_text_report(report, color=bool(color), identity=identity) - - # Unknown / missing overall after fallback must not look like success. - return 0 if report.get("overall") == "ok" else 1 + _print_text_report(report, color=sys.stdout.isatty() if color is None else bool(color), identity=identity) + 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..3521af89f8 100644 --- a/tools/computer_use/permissions.py +++ b/tools/computer_use/permissions.py @@ -1,18 +1,12 @@ -""" -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. - -``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``. +"Ready to drive" differs per platform: macOS needs explicit TCC grants (Accessibility + +Screen Recording) 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 detail into one +payload for the desktop card, the ``permissions`` CLI and ``/api/tools/computer-use/status``. """ from __future__ import annotations @@ -21,7 +15,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 @@ -33,55 +27,39 @@ _BOOLS = ("accessibility", "screen_recording", "screen_recording_capturable") def _resolve_driver_cmd(override: Optional[str]) -> Optional[str]: """Use the runtime resolver for UI status and permission commands too.""" from tools.computer_use.cua_backend import resolve_cua_driver_cmd - return resolve_cua_driver_cmd(override) - def _child_env() -> Dict[str, str]: """cua-driver child env (telemetry policy + provider secrets stripped); degrades to ``os.environ`` on import error so probes never break.""" try: from tools.computer_use.cua_backend import sanitized_cua_driver_env - return sanitized_cua_driver_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, 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: """Run ``binary args`` and parse stdout as JSON (``None`` on empty output).""" 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 + data = 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} - def _mac_permissions(binary: str, out: Dict[str, Any]) -> None: """Fold ``cua-driver permissions status --json`` booleans into ``out``.""" try: @@ -115,16 +93,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 +118,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..009489544f 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 @@ -30,17 +28,11 @@ from typing import Any, Dict, Optional logger = logging.getLogger(__name__) - def _explicit_aux_vision_override(cfg: Optional[Dict[str, Any]]) -> bool: - """True when ``auxiliary.vision`` carries a non-default user override. - - Mirrors ``agent.image_routing._explicit_aux_vision_override`` so the capture - 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 {} + """True when ``auxiliary.vision`` carries a non-default user override. Mirrors + ``agent.image_routing._explicit_aux_vision_override`` so the capture path and the + user-attached-image path agree; ``provider: "auto"``, blanks or a missing block are *not* explicit.""" + 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 @@ -49,33 +41,19 @@ def _explicit_aux_vision_override(cfg: Optional[Dict[str, Any]]) -> bool: base_url = str(vision.get("base_url") or "").strip() 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 - return _supports_vision_override(cfg, provider, model) except Exception as exc: # pragma: no cover - defensive logger.debug("computer_use vision_routing: config override lookup failed: %s", exc) 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)*. - - Prefers ``agent.image_routing._lookup_supports_vision``; falls back to raw - models.dev capabilities only when that import is unavailable. Any lookup - error yields None (caller fails closed toward aux routing). - """ +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 error → None (caller fails closed to aux).""" if not provider or not model: return None try: @@ -86,24 +64,16 @@ def _lookup_supports_vision( if _lookup_image_supports is not None: return _lookup_image_supports(provider, model, cfg) from agent.models_dev import get_model_capabilities - 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)) - 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. - """ + """Whether *provider*+*model* carries images inside tool-result messages. Reuses + ``tools.vision_tools._supports_media_in_tool_results`` to stay in lockstep with the + ``vision_analyze`` fast path; None on import failure so callers fall back to aux, not guess.""" if not provider: return None try: @@ -113,33 +83,19 @@ def _provider_accepts_multimodal_tool_result(provider: str, model: str) -> Optio return None 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: - """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 - provider, *cfg* the loaded ``config.yaml`` dict (or None). False means keep - the multimodal envelope (main model handles vision natively). - """ +def should_route_capture_to_aux_vision(provider: str, model: str, cfg: Optional[Dict[str, Any]]) -> bool: + """True iff the screenshot should be pre-analysed via aux vision; False keeps the + multimodal envelope. *provider* is the lower-case canonical id, *model* the slug as + sent to the provider, *cfg* the loaded ``config.yaml`` dict (or None).""" 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"]