diff --git a/tests/computer_use/test_doctor.py b/tests/computer_use/test_doctor.py index a912176138e..abb1abe6263 100644 --- a/tests/computer_use/test_doctor.py +++ b/tests/computer_use/test_doctor.py @@ -322,9 +322,10 @@ class TestDriverCmdResolution: ) with patch("shutil.which", return_value="/env/path/cua-driver") as which_mock, \ patch("subprocess.Popen", return_value=proc), \ - patch("sys.stdout", new_callable=StringIO): + patch("sys.stdout", new_callable=StringIO), \ + patch("hermes_cli.tools_config._cua_driver_cmd", side_effect=Exception("force env")): + # Force env-var resolution path inside run_doctor. doctor.run_doctor() - # First (and only) which call should have used the env var. which_mock.assert_called_with("/env/path/cua-driver") @pytest.mark.skipif(sys.platform == "win32", reason="POSIX user-local path regression") @@ -345,4 +346,238 @@ class TestDriverCmdResolution: patch("sys.stdout", new_callable=StringIO): assert doctor.run_doctor() == 0 - health.assert_called_once_with(str(driver), include=(), skip=()) + health.assert_called_once_with(str(driver), include=(), skip=(), timeout=12.0) + + +# ── cua-driver 0.10 unclassified health_report fallback ──────────────────── + + +def _unclassified_health_result() -> dict: + """MCP tools/call result shape from cua-driver 0.10.x denial.""" + return { + "isError": True, + "content": [ + { + "type": "text", + "text": ( + "Permission denied: tool 'health_report' has no " + "reviewed risk classification" + ), + } + ], + "structuredContent": {"exit_code": 1}, + } + + +def _perms_ok_result() -> dict: + return { + "isError": False, + "content": [{"type": "text", "text": "ok"}], + "structuredContent": { + "accessibility": True, + "screen_recording": True, + "screen_recording_capturable": True, + }, + } + + +def _list_apps_ok_result() -> dict: + return { + "isError": False, + "content": [{"type": "text", "text": "Found 1 app"}], + "structuredContent": { + "apps": [{"name": "Finder", "pid": 1, "running": True}], + }, + } + + +class TestHealthReportFallback: + """cua-driver 0.10 marks health_report risk-unclassified → isError. + + Doctor must NOT treat structuredContent={exit_code:1} as a real report + (that produced '• cua-driver ? on ? — ?'). It synthesizes schema_version=1 + via check_permissions / list_apps / CLI --version instead. + """ + + def test_isError_unclassified_uses_fallback_overall_ok(self): + from tools.computer_use import doctor + + health_proc = _fake_proc_with_responses( + { + "jsonrpc": "2.0", + "id": 1, + "result": {"serverInfo": {"name": "cua-driver", "version": "0.10.0"}}, + }, + {"jsonrpc": "2.0", "id": 2, "result": _unclassified_health_result()}, + ) + probe_proc = _fake_proc_with_responses( + { + "jsonrpc": "2.0", + "id": 1, + "result": {"serverInfo": {"name": "cua-driver", "version": "0.10.0"}}, + }, + {"jsonrpc": "2.0", "id": 2, "result": _perms_ok_result()}, + {"jsonrpc": "2.0", "id": 3, "result": _list_apps_ok_result()}, + ) + procs = iter([health_proc, probe_proc]) + run_mock = MagicMock( + return_value=MagicMock(returncode=0, stdout="cua-driver 0.10.0\n", stderr=""), + ) + + with patch("shutil.which", return_value="/fake/cua-driver"), \ + patch("subprocess.Popen", side_effect=lambda *a, **k: next(procs)), \ + patch("subprocess.run", run_mock), \ + patch("sys.stdout", new_callable=StringIO) as out: + code = doctor.run_doctor(color=False) + + assert code == 0 + text = out.getvalue() + assert "ok" in text + assert "0.10.0" in text + # Fallback path must be visible in the check list + assert "health_report_path" in text + assert "fallback composite" in text + assert "tcc_accessibility" in text + assert "binary_version" in text + + def test_isError_unclassified_json_payload_has_schema_and_fallback_flag(self): + from tools.computer_use import doctor + + health_proc = _fake_proc_with_responses( + {"jsonrpc": "2.0", "id": 1, "result": {"serverInfo": {"version": "0.10.0"}}}, + {"jsonrpc": "2.0", "id": 2, "result": _unclassified_health_result()}, + ) + probe_proc = _fake_proc_with_responses( + {"jsonrpc": "2.0", "id": 1, "result": {"serverInfo": {"version": "0.10.0"}}}, + {"jsonrpc": "2.0", "id": 2, "result": _perms_ok_result()}, + {"jsonrpc": "2.0", "id": 3, "result": _list_apps_ok_result()}, + ) + procs = iter([health_proc, probe_proc]) + run_mock = MagicMock( + return_value=MagicMock(returncode=0, stdout="cua-driver 0.10.0\n", stderr=""), + ) + + with patch("shutil.which", return_value="/fake/cua-driver"), \ + patch("subprocess.Popen", side_effect=lambda *a, **k: next(procs)), \ + patch("subprocess.run", run_mock), \ + patch("sys.stdout", new_callable=StringIO) as out: + code = doctor.run_doctor(json_output=True) + + assert code == 0 + parsed = json.loads(out.getvalue()) + assert parsed["schema_version"] == "1" + assert parsed["overall"] == "ok" + assert parsed.get("fallback") is True + names = [c["name"] for c in parsed["checks"]] + assert "binary_version" in names + assert "tcc_accessibility" in names + assert "tcc_screen_recording" in names + assert "ax_capability" in names + assert "health_report_path" in names + # Must not be the raw denial payload + assert "exit_code" not in parsed + + def test_structuredContent_exit_code_only_triggers_fallback(self): + """Even without isError, bare {exit_code:1} is not a valid report.""" + from tools.computer_use import doctor + + # Some gateways might drop isError but still ship exit_code-only SC. + health_proc = _fake_proc_with_responses( + {"jsonrpc": "2.0", "id": 1, "result": {}}, + { + "jsonrpc": "2.0", + "id": 2, + "result": { + "isError": False, + "structuredContent": {"exit_code": 1}, + "content": [{"type": "text", "text": "Permission denied"}], + }, + }, + ) + probe_proc = _fake_proc_with_responses( + {"jsonrpc": "2.0", "id": 1, "result": {"serverInfo": {"version": "0.10.0"}}}, + {"jsonrpc": "2.0", "id": 2, "result": _perms_ok_result()}, + {"jsonrpc": "2.0", "id": 3, "result": _list_apps_ok_result()}, + ) + procs = iter([health_proc, probe_proc]) + run_mock = MagicMock( + return_value=MagicMock(returncode=0, stdout="cua-driver 0.10.0\n", stderr=""), + ) + + with patch("shutil.which", return_value="/fake/cua-driver"), \ + patch("subprocess.Popen", side_effect=lambda *a, **k: next(procs)), \ + patch("subprocess.run", run_mock), \ + patch("sys.stdout", new_callable=StringIO) as out: + code = doctor.run_doctor(json_output=True) + + assert code == 0 + parsed = json.loads(out.getvalue()) + assert parsed["schema_version"] == "1" + assert parsed.get("fallback") is True + + def test_real_schema_version_1_preferred_over_fallback(self): + """When health_report returns a real schema_version=1 payload, use it + and never call the composite fallback path.""" + from tools.computer_use import doctor + + proc = _fake_proc_with_responses( + {"jsonrpc": "2.0", "id": 1, "result": {}}, + {"jsonrpc": "2.0", "id": 2, "result": {"structuredContent": _ok_report()}}, + ) + with patch("shutil.which", return_value="/fake/cua-driver"), \ + patch("subprocess.Popen", return_value=proc), \ + patch.object(doctor, "_compose_fallback_report") as fallback_mock, \ + patch("sys.stdout", new_callable=StringIO) as out: + code = doctor.run_doctor(json_output=True) + + assert code == 0 + fallback_mock.assert_not_called() + parsed = json.loads(out.getvalue()) + assert parsed == _ok_report() + assert "fallback" not in parsed + + def test_extract_raises_health_report_unavailable_on_isError(self): + from tools.computer_use import doctor + + with __import__("pytest").raises(doctor.HealthReportUnavailable) as ei: + doctor._extract_health_report_from_result(_unclassified_health_result()) + assert "Permission denied" in str(ei.value) or "unclassified" in str(ei.value).lower() or "risk" in str(ei.value).lower() + + def test_fallback_degraded_when_accessibility_denied(self): + from tools.computer_use import doctor + + health_proc = _fake_proc_with_responses( + {"jsonrpc": "2.0", "id": 1, "result": {"serverInfo": {"version": "0.10.0"}}}, + {"jsonrpc": "2.0", "id": 2, "result": _unclassified_health_result()}, + ) + denied_perms = { + "isError": False, + "structuredContent": { + "accessibility": False, + "screen_recording": False, + }, + } + probe_proc = _fake_proc_with_responses( + {"jsonrpc": "2.0", "id": 1, "result": {"serverInfo": {"version": "0.10.0"}}}, + {"jsonrpc": "2.0", "id": 2, "result": denied_perms}, + { + "jsonrpc": "2.0", + "id": 3, + "result": {"isError": True, "content": [{"type": "text", "text": "no ax"}]}, + }, + ) + procs = iter([health_proc, probe_proc]) + run_mock = MagicMock( + return_value=MagicMock(returncode=0, stdout="cua-driver 0.10.0\n", stderr=""), + ) + + with patch("shutil.which", return_value="/fake/cua-driver"), \ + patch("subprocess.Popen", side_effect=lambda *a, **k: next(procs)), \ + patch("subprocess.run", run_mock), \ + patch("sys.stdout", new_callable=StringIO) as out: + code = doctor.run_doctor(json_output=True) + + assert code == 1 + parsed = json.loads(out.getvalue()) + assert parsed["overall"] == "degraded" + assert parsed.get("fallback") is True diff --git a/tools/computer_use/doctor.py b/tools/computer_use/doctor.py index 54fad066684..028bb91ae83 100644 --- a/tools/computer_use/doctor.py +++ b/tools/computer_use/doctor.py @@ -7,6 +7,12 @@ renders the structured response. When the driver gets new checks, they flow through here without code changes on the Hermes side — the only contract is the stable `schema_version="1"` payload shape. +cua-driver 0.10.x marks `health_report` with risk.class='unclassified', so +MCP tools/call returns isError=true ("Permission denied: ... no reviewed +risk classification") with structuredContent ``{"exit_code": 1}``. That is +NOT a schema_version=1 report — we detect it and synthesize a composite +report via working probes (check_permissions, list_apps, CLI --version). + Exit code conventions: - 0: overall == "ok" - 1: overall in ("degraded", "failed") @@ -17,9 +23,12 @@ from __future__ import annotations import json import os +import platform as _platform_mod +import re +import shutil import subprocess import sys -from typing import Any, Dict, List, Optional, Sequence +from typing import Any, Dict, List, Optional, Sequence, Tuple # Match the ALLOWED_STATUS_VALUES + ALLOWED_OVERALL_VALUES the cua-driver @@ -36,6 +45,14 @@ _OVERALL_GLYPH = { } +class HealthReportUnavailable(RuntimeError): + """health_report MCP tool denied or returned a non-schema payload. + + Raised so ``run_doctor`` can fall back to composite probes that work on + cua-driver builds where ``health_report`` is risk-unclassified (0.10.x). + """ + + def _cua_child_env() -> Dict[str, str]: """cua-driver child env with the Hermes telemetry policy applied. @@ -68,33 +85,77 @@ def _sanitized_cua_env() -> Dict[str, str]: return env -def _drive_health_report( - binary: str, - *, - include: Sequence[str] = (), - skip: Sequence[str] = (), - timeout: float = 12.0, -) -> Dict[str, Any]: - """Spawn ` mcp`, perform the JSON-RPC handshake, call - `health_report`, and return the parsed `structuredContent` dict. +def _is_valid_health_report(payload: Any) -> bool: + """True when *payload* looks like a schema_version=1 health_report.""" + if not isinstance(payload, dict): + return False + if "schema_version" not in payload: + return False + if "overall" not in payload: + return False + if not isinstance(payload.get("checks"), list): + return False + return True - Raises `RuntimeError` on a protocol-level failure (binary crash, - malformed response, JSON-RPC error). Never raises on a `health_report` - that has failing checks — the tool's contract is to always return a - well-formed report with `overall` set, never to set `isError`. + +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 (e.g. 0.10's + ``{"exit_code": 1}`` structuredContent on unclassified denial). + Raises ``RuntimeError`` when the response shape is unusable for other + reasons (no content at all). """ - args: Dict[str, Any] = {} - if include: - args["include"] = list(include) - if skip: - args["skip"] = list(skip) + if result.get("isError") is True: + # Prefer the human text; fall back to a generic denial message. + denial = "health_report returned isError=true" + for item in result.get("content") or []: + if isinstance(item, dict) and item.get("type") == "text": + text = (item.get("text") or "").strip() + if text: + denial = text + break + raise HealthReportUnavailable(denial) + 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 + text = item.get("text", "") + try: + parsed = json.loads(text) + except (ValueError, TypeError): + continue + if _is_valid_health_report(parsed): + return parsed + + # structuredContent present but not a real report (the 0.10 unclassified + # path ships {"exit_code": 1}) — treat as 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())}" + ) + + +def _open_mcp(binary: str) -> subprocess.Popen: + """Spawn `` mcp`` with UTF-8 + sanitized env.""" # cua-driver emits UTF-8 (containing emoji in check messages on macOS # and arbitrary file paths on Windows). The Python default # text-mode encoding follows the system locale — `cp1252` on a # default Windows install — which raises UnicodeDecodeError on the # first non-ASCII byte. Pin the codec. - proc = subprocess.Popen( + return subprocess.Popen( [binary, "mcp"], stdin=subprocess.PIPE, stdout=subprocess.PIPE, @@ -105,74 +166,487 @@ def _drive_health_report( bufsize=1, env=_sanitized_cua_env(), ) + + +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.""" + assert proc.stdin is not None and proc.stdout is not None + payload: Dict[str, Any] = {"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)'}" + ) + try: + resp = json.loads(line) + except (ValueError, TypeError) as e: + raise RuntimeError(f"{method} response was not valid JSON: {e}\nraw: {line[:200]}") + if "error" in resp: + raise RuntimeError(f"{method} JSON-RPC error: {resp['error']}") + return resp + + +def _close_mcp(proc: subprocess.Popen, timeout: float) -> None: + try: + 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`, perform the JSON-RPC handshake, call + `health_report`, and return the parsed schema_version=1 report. + + Raises: + HealthReportUnavailable: tool denied (isError) or non-schema payload + (cua-driver 0.10 unclassified). Caller should fall back. + RuntimeError: protocol-level failure (binary crash, malformed JSON, + JSON-RPC error, empty content). + """ + args: Dict[str, Any] = {} + if include: + args["include"] = list(include) + if skip: + args["skip"] = list(skip) + + proc = _open_mcp(binary) try: # 1. initialize - proc.stdin.write(json.dumps({ - "jsonrpc": "2.0", "id": 1, - "method": "initialize", "params": {}, - }) + "\n") - proc.stdin.flush() - init_line = proc.stdout.readline() - if not init_line: - stderr_tail = (proc.stderr.read() or "").strip().splitlines()[-3:] - raise RuntimeError( - f"cua-driver mcp produced no initialize response. " - f"stderr tail: {stderr_tail or '(empty)'}" - ) + init_resp = _mcp_rpc(proc, 1, "initialize", {}) + _ = init_resp # handshake only # 2. tools/call health_report - proc.stdin.write(json.dumps({ - "jsonrpc": "2.0", "id": 2, - "method": "tools/call", - "params": {"name": "health_report", "arguments": args}, - }) + "\n") - proc.stdin.flush() - call_line = proc.stdout.readline() - if not call_line: - raise RuntimeError("cua-driver mcp closed stdout without responding to health_report.") + call_resp = _mcp_rpc( + proc, + 2, + "tools/call", + {"name": "health_report", "arguments": args}, + ) finally: - try: - proc.stdin.close() - except Exception: - pass - try: - proc.wait(timeout=timeout) - except subprocess.TimeoutExpired: - proc.kill() - proc.wait() + _close_mcp(proc, timeout) + result = call_resp.get("result") or {} + if not isinstance(result, dict): + raise RuntimeError(f"health_report result was not an object: {type(result).__name__}") + 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``.""" try: - resp = json.loads(call_line) - except (ValueError, TypeError) as e: - raise RuntimeError(f"health_report response was not valid JSON: {e}\nraw: {call_line[:200]}") + completed = subprocess.run( + [binary, "--version"], + capture_output=True, + text=True, + encoding="utf-8", + errors="replace", + timeout=timeout, + env=_sanitized_cua_env(), + ) + except (OSError, subprocess.TimeoutExpired) as e: + return "fail", f"--version failed: {e}" - if "error" in resp: - raise RuntimeError(f"health_report JSON-RPC error: {resp['error']}") + text = ((completed.stdout or "") + (completed.stderr or "")).strip() + if completed.returncode != 0 and not text: + return "fail", f"--version exited {completed.returncode}" - result = resp.get("result") or {} + # 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") + if completed.returncode != 0: + return "fail", version + return "pass", version - # Preferred: structuredContent (cua-driver-rs always emits it on the - # health_report response). Fall back to parsing the first text item - # as JSON for older cua-driver builds that didn't carry structuredContent. - sc = result.get("structuredContent") - if isinstance(sc, dict): - return sc - for item in result.get("content", []): - if item.get("type") == "text": - text = item.get("text", "") - try: - # Many health_report payloads ship JSON in the text item too. - parsed = json.loads(text) - if isinstance(parsed, dict) and "schema_version" in parsed: - return parsed - except (ValueError, TypeError): - pass +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 = subprocess.run( + [binary, "doctor"], + capture_output=True, + text=True, + encoding="utf-8", + errors="replace", + timeout=timeout, + env=_sanitized_cua_env(), + ) + except (OSError, subprocess.TimeoutExpired): + return None + out = ((completed.stdout or "") + (completed.stderr or "")).strip() + return out or None - raise RuntimeError( - "health_report response carried neither structuredContent nor a parseable " - f"JSON text block. Result keys: {list(result.keys())}" - ) + +def _drive_fallback_probes( + binary: str, + *, + timeout: float = 12.0, +) -> Dict[str, Any]: + """Call working MCP tools (check_permissions, list_apps) in one session. + + Returns a dict with keys: + - init_version: str | None (from initialize serverInfo) + - permissions: structuredContent dict | None + - permissions_error: str | None + - list_apps_ok: bool | None + - list_apps_error: str | None + - list_apps_count: int | None + """ + out: Dict[str, Any] = { + "init_version": None, + "permissions": None, + "permissions_error": None, + "list_apps_ok": None, + "list_apps_error": None, + "list_apps_count": None, + } + proc = _open_mcp(binary) + try: + 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_resp = _mcp_rpc( + proc, 2, "tools/call", {"name": "check_permissions", "arguments": {}} + ) + perm_result = perm_resp.get("result") or {} + if perm_result.get("isError") is True: + msg = "check_permissions isError" + for item in perm_result.get("content") or []: + if isinstance(item, dict) and item.get("type") == "text": + t = (item.get("text") or "").strip() + if t: + msg = t + break + out["permissions_error"] = msg + else: + sc = perm_result.get("structuredContent") + out["permissions"] = sc if isinstance(sc, dict) else {} + except RuntimeError as e: + out["permissions_error"] = str(e) + + # list_apps — light AX capability probe + try: + apps_resp = _mcp_rpc( + proc, 3, "tools/call", {"name": "list_apps", "arguments": {}} + ) + apps_result = apps_resp.get("result") or {} + if apps_result.get("isError") is True: + msg = "list_apps isError" + for item in apps_result.get("content") or []: + if isinstance(item, dict) and item.get("type") == "text": + t = (item.get("text") or "").strip() + if t: + msg = t + break + out["list_apps_ok"] = False + out["list_apps_error"] = msg + else: + sc = apps_result.get("structuredContent") or {} + apps = sc.get("apps") if isinstance(sc, dict) else None + if isinstance(apps, list): + out["list_apps_ok"] = True + out["list_apps_count"] = len(apps) + else: + # text-only success still counts as AX working + out["list_apps_ok"] = True + out["list_apps_count"] = None + except RuntimeError as e: + out["list_apps_ok"] = False + out["list_apps_error"] = str(e) + finally: + _close_mcp(proc, timeout) + + return out + + +def _platform_name() -> str: + sysname = (_platform_mod.system() or "").lower() + if sysname == "darwin": + return "darwin" + if sysname == "windows": + return "windows" + if sysname == "linux": + return "linux" + return sysname or "unknown" + + +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``. + """ + plat = _platform_name() + checks: List[Dict[str, Any]] = [] + + 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" + ver_msg = f"cua-driver {driver_version}" + else: + ver_msg = ( + f"cua-driver {ver_value}" if ver_status == "pass" else (ver_value or "version unknown") + ) + + checks.append({ + "name": "binary_version", + "status": ver_status, + "message": ver_msg, + }) + + # platform_supported — doctor runs wherever the binary runs + supported = plat in ("darwin", "linux", "windows") + checks.append({ + "name": "platform_supported", + "status": "pass" if supported else "fail", + "message": f"platform={plat}" + ("" if supported else " (unsupported)"), + }) + + # session_active — we don't start a session in doctor; mark skip + checks.append({ + "name": "session_active", + "status": "skip", + "message": "not probed (doctor does not open a cua session)", + }) + + perms = probes.get("permissions") if isinstance(probes.get("permissions"), dict) else None + perm_err = probes.get("permissions_error") + + if perms is not None: + ax = perms.get("accessibility") + scr = perms.get("screen_recording") + capturable = perms.get("screen_recording_capturable") + + if ax is True: + checks.append({ + "name": "tcc_accessibility", + "status": "pass", + "message": "Accessibility is granted.", + "data": {"accessibility": True}, + }) + elif ax is False: + checks.append({ + "name": "tcc_accessibility", + "status": "fail", + "message": "Accessibility is not granted.", + "hint": "Grant Accessibility to CuaDriver in System Settings → Privacy & Security.", + "data": {"accessibility": False}, + }) + else: + checks.append({ + "name": "tcc_accessibility", + "status": "skip", + "message": "accessibility field absent from check_permissions", + }) + + if scr is True and capturable is False: + checks.append({ + "name": "tcc_screen_recording", + "status": "fail", + "message": "Screen Recording granted but not capturable.", + "hint": ( + "Screen Recording permission may need a restart of CuaDriver " + "or a re-grant in System Settings." + ), + "data": { + "screen_recording": True, + "screen_recording_capturable": False, + }, + }) + elif scr is True: + checks.append({ + "name": "tcc_screen_recording", + "status": "pass", + "message": "Screen Recording is granted.", + "data": { + "screen_recording": True, + "screen_recording_capturable": capturable, + }, + }) + elif scr is False: + checks.append({ + "name": "tcc_screen_recording", + "status": "fail", + "message": "Screen Recording is not granted.", + "hint": "Grant Screen Recording to CuaDriver in System Settings → Privacy & Security.", + "data": {"screen_recording": False}, + }) + else: + # Non-macOS or field absent + if plat == "darwin": + checks.append({ + "name": "tcc_screen_recording", + "status": "skip", + "message": "screen_recording field absent from check_permissions", + }) + else: + checks.append({ + "name": "tcc_screen_recording", + "status": "skip", + "message": f"not applicable on {plat}", + }) + else: + checks.append({ + "name": "tcc_accessibility", + "status": "fail" if perm_err else "skip", + "message": perm_err or "check_permissions unavailable", + }) + checks.append({ + "name": "tcc_screen_recording", + "status": "fail" if perm_err else "skip", + "message": perm_err or "check_permissions unavailable", + }) + + # ax_capability — infer from list_apps success or accessibility grant + list_ok = probes.get("list_apps_ok") + list_err = probes.get("list_apps_error") + list_count = probes.get("list_apps_count") + ax_granted = bool(perms and perms.get("accessibility") is True) + if list_ok is True: + count_msg = f" ({list_count} apps)" if isinstance(list_count, int) else "" + checks.append({ + "name": "ax_capability", + "status": "pass", + "message": f"list_apps succeeded{count_msg}", + }) + elif list_ok is False: + checks.append({ + "name": "ax_capability", + "status": "fail", + "message": ( + list_err + or ( + "list_apps failed despite accessibility grant" + if ax_granted + else "list_apps failed" + ) + ), + }) + elif ax_granted: + checks.append({ + "name": "ax_capability", + "status": "pass", + "message": "inferred from accessibility grant (list_apps not probed)", + }) + else: + checks.append({ + "name": "ax_capability", + "status": "skip", + "message": "not probed", + }) + + # 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({ + "name": "health_report_path", + "status": "skip", + "message": ( + "fallback composite (cua-driver 0.10 unclassified health_report); " + f"cause: {reason_short}" + ), + }) + + # Optional CLI doctor text (best-effort) + doctor_txt = _cli_doctor_snippet(binary) + if doctor_txt: + first = doctor_txt.splitlines()[0].strip() + cli_ok = "[ok" in doctor_txt.lower() or "ok ]" in doctor_txt + checks.append({ + "name": "cli_doctor", + "status": "pass" if cli_ok else "skip", + "message": first, + "data": {"snippet": doctor_txt[:2000]}, + }) + + # Normalize any accidental non-vocab status values + for c in checks: + if c.get("status") not in ("pass", "fail", "skip"): + c["status"] = "fail" + + # overall: ok if TCC+binary ok; degraded if partial; failed if binary missing/bad + status_by_name = {c.get("name"): c.get("status") for c in checks} + binary_ok = status_by_name.get("binary_version") == "pass" + tcc_ax_status = status_by_name.get("tcc_accessibility") + tcc_ok = tcc_ax_status in ("pass", "skip", None) + fail_count = sum(1 for c in checks if c.get("status") == "fail") + + if not binary_ok: + overall = "failed" + elif tcc_ok and fail_count == 0: + overall = "ok" + elif tcc_ok and fail_count > 0: + # Binary + accessibility fine, but something else failed (e.g. screen + # recording) → degraded rather than failed. + overall = "degraded" + else: + # Accessibility denied or broken — computer-use is partially/fully blocked. + 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: + return _drive_health_report( + binary, include=include, skip=skip, timeout=timeout, + ) + except HealthReportUnavailable as e: + return _compose_fallback_report( + binary, reason=str(e), timeout=timeout, + ) def _print_text_report(report: Dict[str, Any], color: bool) -> None: @@ -241,6 +715,11 @@ def run_doctor( Honors `HERMES_CUA_DRIVER_CMD` via the shared runtime resolver, so the doctor diagnoses what your `computer_use` toolset will actually invoke. + + On cua-driver 0.10.x, ``health_report`` may be risk-unclassified and + denied; doctor then synthesizes a schema_version=1 report from + check_permissions / list_apps / CLI probes instead of printing + ``• cua-driver ? on ? — ?``. """ # Windows ships stdout/stderr wrapped with the system ANSI codec # (`cp1252` on a US locale, `cp936` on zh-CN, etc.). The check-matrix @@ -263,7 +742,9 @@ def run_doctor( return 2 try: - report = _drive_health_report(binary, include=include, skip=skip) + report = _drive_health_report_or_fallback( + binary, include=include, skip=skip, + ) except RuntimeError as e: print(f"cua-driver health_report failed: {e}", file=sys.stderr) return 2 @@ -279,4 +760,7 @@ def run_doctor( overall = report.get("overall") if overall in ("degraded", "failed"): return 1 + if overall != "ok": + # Unknown / missing overall after fallback should not look like success. + return 1 return 0