From 80affc7ee39f8873d042f15aa10343bc1d23e89b Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 15:37:58 -0700 Subject: [PATCH] =?UTF-8?q?refactor(computer=5Fuse):=20doctor=20=E2=80=94?= =?UTF-8?q?=20reuse=20permissions.=5Fchild=5Fenv,=20split=20fallback/repor?= =?UTF-8?q?t=20rendering=20into=20helpers?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tools/computer_use/doctor.py | 469 ++++++++++++++--------------------- 1 file changed, 180 insertions(+), 289 deletions(-) diff --git a/tools/computer_use/doctor.py b/tools/computer_use/doctor.py index ee220b24cc..aa4f1c5eb5 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,28 @@ _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`." +) 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 _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,21 +57,36 @@ 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 = ((completed.stdout or "") + (completed.stderr or "")).strip() + if completed.returncode != 0 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" + version = m.group(1) if m else (text.splitlines()[0] if text else "unknown") + return ("fail" if completed.returncode != 0 else "pass"), version -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: + completed = _run_cli(binary, "doctor", timeout=timeout) + except (OSError, subprocess.TimeoutExpired): + return None + return ((completed.stdout or "") + (completed.stderr or "")).strip() or None def _build_identity(binary: str, report: Dict[str, Any]) -> Dict[str, Any]: """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) + cli_tok, report_tok = token(cli), token(report_v) return { "resolved_binary": binary, "cli_version": cli or None, @@ -111,55 +95,59 @@ def _build_identity(binary: str, report: 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" in payload + and "overall" in payload and isinstance(payload.get("checks"), list)) + +def _text_items(result: Dict[str, Any]) -> 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: + """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]: """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. - - 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(), - ) + """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()) +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]: """Write one JSON-RPC request and read one response line.""" @@ -171,17 +159,8 @@ def _mcp_rpc(proc: subprocess.Popen, msg_id: int, method: str, params: Any = Non 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}. " + f"stderr tail: {_stderr_tail(proc) or '(empty)'}") try: resp = json.loads(line) except (ValueError, TypeError) as e: @@ -190,13 +169,11 @@ 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 {} - @contextmanager def _mcp_session(binary: str, timeout: float) -> Iterator[subprocess.Popen]: """Spawn `` mcp`` and always close stdin / wait / kill it on exit.""" @@ -204,25 +181,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). - """ + """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,31 +203,17 @@ 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[Dict[str, Any]], 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}" - - 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 - + 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 def _drive_fallback_probes(binary: str, *, timeout: float = 12.0) -> Dict[str, Any]: """Call working MCP tools (check_permissions, list_apps) in one session. @@ -264,54 +221,39 @@ def _drive_fallback_probes(binary: str, *, timeout: float = 12.0) -> Dict[str, A 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: Dict[str, Any] = 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") + 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") - def _check(name: str, status: str, message: str, **extra: Any) -> Dict[str, Any]: """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]]: """tcc_accessibility + tcc_screen_recording checks from check_permissions output.""" if perms is None: @@ -323,15 +265,13 @@ def _tcc_checks(perms: Optional[Dict[str, Any]], perm_err: Optional[str], plat: 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 +285,12 @@ 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]: """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,82 +301,58 @@ 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: + """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``. - """ + """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 + 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]] = [ _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" - - # 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", + "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]: """Prefer real health_report; on denial/non-schema, synthesize via probes.""" @@ -450,86 +362,74 @@ 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]: """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`." - ) + check.update(status="fail", message=_ZERO_DISPLAY_MSG, hint=_ZERO_DISPLAY_HINT) if report.get("overall") == "ok": report["overall"] = "degraded" return report +# ── Rendering ──────────────────────────────────────────────────────────────── + +def _check_lines(check: Dict[str, Any], 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 ()): + rendered = json.dumps(value) if isinstance(value, (dict, list)) else value + lines.append(f" {dim}{key}={rendered}{reset}") + return lines + 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). 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", "?") + 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: @@ -541,10 +441,8 @@ def run_doctor(driver_cmd: Optional[str] = None, *, include: Sequence[str] = (), """ # 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) @@ -553,7 +451,6 @@ def run_doctor(driver_cmd: Optional[str] = None, *, include: Sequence[str] = (), print(f"cua-driver: not installed (looked for {looked_for!r}).") print(" Run: hermes computer-use install") return 2 - try: report = _drive_health_report_or_fallback(binary, include=include, skip=skip) except RuntimeError as e: @@ -561,18 +458,12 @@ def run_doctor(driver_cmd: Optional[str] = None, *, include: Sequence[str] = (), return 2 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) - + _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