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 1/4] =?UTF-8?q?refactor(computer=5Fuse):=20doctor=20?=
 =?UTF-8?q?=E2=80=94=20reuse=20permissions.=5Fchild=5Fenv,=20split=20fallb?=
 =?UTF-8?q?ack/report=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 ``<binary> 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 ``<binary> 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 ``<binary> 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 ``<binary> 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 `<binary> 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

From 04de344f41d43079787be6e6ca3ba6c253f9740e Mon Sep 17 00:00:00 2001
From: Teknium <127238744+teknium1@users.noreply.github.com>
Date: Wed, 2 Sep 2026 15:45:55 -0700
Subject: [PATCH 2/4] refactor(computer_use): compact
 backend/permissions/vision_routing/__init__ docs; tighten doctor helpers

---
 tools/computer_use/__init__.py       |  18 ++-
 tools/computer_use/backend.py        | 172 +++++++++------------------
 tools/computer_use/doctor.py         | 126 +++++++++-----------
 tools/computer_use/permissions.py    |  55 +++------
 tools/computer_use/vision_routing.py |  66 ++++------
 5 files changed, 152 insertions(+), 285 deletions(-)

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

From 6417d5dfbcf2129fdc698d9fe8280184d8329a28 Mon Sep 17 00:00:00 2001
From: Teknium <127238744+teknium1@users.noreply.github.com>
Date: Wed, 2 Sep 2026 15:51:02 -0700
Subject: [PATCH 3/4] =?UTF-8?q?refactor(computer=5Fuse):=20doctor=20?=
 =?UTF-8?q?=E2=80=94=20inline=20fallback=20seam=20into=20run=5Fdoctor,=20s?=
 =?UTF-8?q?hared=20CLI=20output=20helper?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 tools/computer_use/doctor.py | 42 +++++++++++++++---------------------
 1 file changed, 17 insertions(+), 25 deletions(-)

diff --git a/tools/computer_use/doctor.py b/tools/computer_use/doctor.py
index b3393adb7a..2d55e733c0 100644
--- a/tools/computer_use/doctor.py
+++ b/tools/computer_use/doctor.py
@@ -46,6 +46,9 @@ def _run_cli(binary: str, *args: str, timeout: float) -> subprocess.CompletedPro
     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]:
     """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."""
@@ -62,20 +65,18 @@ def _cli_driver_version(binary: str, timeout: float = 5.0) -> Tuple[str, Optiona
         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:
+    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"
-    version = m.group(1) if m else (text.splitlines()[0] if text else "unknown")
-    return ("fail" if completed.returncode != 0 else "pass"), version
+    return ("fail" if failed else "pass"), m.group(1) if m else (text.splitlines()[0] if text else "unknown")
 
 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)
+        return _combined_output(_run_cli(binary, "doctor", timeout=timeout)) or None
     except (OSError, subprocess.TimeoutExpired):
         return None
-    return ((completed.stdout or "") + (completed.stderr or "")).strip() or None
 
 def _build_identity(binary: str, report: Report) -> Report:
     """Hermes-side identity block comparing resolved binary vs health_report."""
@@ -94,8 +95,7 @@ def _build_identity(binary: str, report: Report) -> Report:
 
 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))
+    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."""
@@ -154,8 +154,7 @@ def _mcp_rpc(proc: subprocess.Popen, msg_id: int, method: str, params: Any = Non
     proc.stdin.flush()
     line = proc.stdout.readline()
     if not line:
-        raise RuntimeError(f"cua-driver mcp produced no response for {method!r}. "
-                           f"stderr tail: {_stderr_tail(proc) 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:
@@ -240,8 +239,7 @@ def _drive_fallback_probes(binary: str, *, timeout: float = 12.0) -> Report:
     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) -> Report:
     """Build one health check dict (``hint`` / ``data`` only when given)."""
@@ -276,8 +274,7 @@ def _tcc_checks(perms: Optional[Report], perm_err: Optional[str], plat: str) ->
     }
     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: Report, ax_granted: bool) -> Report:
     """ax_capability — inferred from list_apps success or the accessibility grant."""
@@ -341,15 +338,6 @@ def _compose_fallback_report(binary: str, *, reason: str = "", timeout: float =
             "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) -> Report:
-    """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: Report) -> Report:
     """Downgrade an 'ok' report whose screen capture has zero displays.
 
@@ -433,11 +421,15 @@ def run_doctor(driver_cmd: Optional[str] = None, *, include: Sequence[str] = (),
         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

From 7e939af627410faf8a09154813b8bb88a095f31e Mon Sep 17 00:00:00 2001
From: Teknium <127238744+teknium1@users.noreply.github.com>
Date: Wed, 2 Sep 2026 15:54:54 -0700
Subject: [PATCH 4/4] refactor(computer_use): compact
 permissions/vision_routing/backend/__init__ docstrings and spacing

---
 tools/computer_use/__init__.py       | 19 +++++-------
 tools/computer_use/backend.py        | 27 +++++++----------
 tools/computer_use/permissions.py    | 26 +++++-----------
 tools/computer_use/vision_routing.py | 44 ++++++++--------------------
 4 files changed, 38 insertions(+), 78 deletions(-)

diff --git a/tools/computer_use/__init__.py b/tools/computer_use/__init__.py
index eb2a9714b1..4de6c2bb6a 100644
--- a/tools/computer_use/__init__.py
+++ b/tools/computer_use/__init__.py
@@ -1,17 +1,14 @@
 """Computer use toolset — universal (any-model) desktop control via cua-driver.
 
-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`.
+Drives apps through cua-driver's background primitive (focus-without-raise + pid-scoped
+event posting): it does NOT steal the user's cursor, keyboard focus, or Space. Plain
+OpenAI function-calling schema; vision models get SOM captures (numbered overlays + AX
+tree) and click by index, non-vision models use the AX tree alone. Model-facing guidance
+lives in the schema description and each action result's `verdict`.
 
-* `tool.py`        — `computer_use` handler, approval gate, response shaping.
-* `backend.py`     — abstract `ComputerUseBackend` + result dataclasses.
-* `cua_backend.py` — default backend (MCP over stdio to `cua-driver`), with
-                     `cua_backend_parse` / `_session` / `_daemon` siblings.
-* `schema.py`      — the model-facing schema (byte-frozen).
+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
diff --git a/tools/computer_use/backend.py b/tools/computer_use/backend.py
index ee255ef7a3..b2b334a507 100644
--- a/tools/computer_use/backend.py
+++ b/tools/computer_use/backend.py
@@ -16,11 +16,9 @@ _JPEG_SOF_MARKERS = frozenset({0xC0, 0xC1, 0xC2, 0xC3, 0xC5, 0xC6, 0xC7, 0xC9, 0
 
 
 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])
@@ -77,13 +75,9 @@ 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" (default)
-    → both: the PNG already carries 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)
@@ -105,11 +99,10 @@ class CaptureResult:
 class ActionResult:
     """Result of any action (click / type / scroll / drag / key / wait).
 
-    ``ok`` is tool/transport success only — NOT the semantic verdict. Read
-    ``effect`` / ``escalation`` (cua-driver's structured verdict) to decide the
-    next rung of the verify → escalate ladder. All structured fields are optional
-    and additive: an older driver that omits ``structuredContent`` leaves them
-    ``None`` and behavior is unchanged.
+    ``ok`` 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
diff --git a/tools/computer_use/permissions.py b/tools/computer_use/permissions.py
index 5c69d63e4e..3521af89f8 100644
--- a/tools/computer_use/permissions.py
+++ b/tools/computer_use/permissions.py
@@ -1,15 +1,12 @@
 """Cross-platform Computer Use readiness + macOS permission helpers.
 
-"Ready to drive" differs per platform: macOS needs explicit TCC grants
-(Accessibility + Screen Recording) reported/requested via cua-driver
-``permissions status`` / ``permissions grant``; Windows/Linux have no TCC
-toggles, so readiness == driver health. The grants attach to cua-driver's OWN
-identity (``com.trycua.driver``), not Hermes, so ``grant`` launches CuaDriver
-via LaunchServices for correct dialog attribution.
-
-``cua-driver doctor --json`` is the universal signal; ``computer_use_status``
-folds it with the macOS 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
@@ -30,46 +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())
 
-
 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 = [{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:
diff --git a/tools/computer_use/vision_routing.py b/tools/computer_use/vision_routing.py
index 7b60986f8d..009489544f 100644
--- a/tools/computer_use/vision_routing.py
+++ b/tools/computer_use/vision_routing.py
@@ -28,14 +28,10 @@ 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.
-    """
+    """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):
@@ -45,25 +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]:
     """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]:
-    """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).
-    """
+    """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:
@@ -74,21 +64,16 @@ def _lookup_supports_vision(provider: str, model: str, cfg: Optional[Dict[str, A
         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)
         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`` to stay in
-    lockstep with the ``vision_analyze`` native fast path. None on import
-    failure so callers fall back to aux routing rather than guessing.
-    """
+    """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:
@@ -98,14 +83,10 @@ 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).
-    """
+    """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)
@@ -117,5 +98,4 @@ def should_route_capture_to_aux_vision(provider: str, model: str, cfg: Optional[
         return True
     return _lookup_supports_vision(provider, model, cfg) is not True
 
-
 __all__ = ["should_route_capture_to_aux_vision"]