Merge branch 'simp/r2-cua-misc' into simp/integration2
This commit is contained in:
Teknium
@@ -1,26 +1,19 @@
|
||||
"""Computer use toolset — universal (any-model) desktop control via cua-driver.
|
||||
|
||||
Drives apps through cua-driver's background computer-use primitive (focus-
|
||||
without-raise + pid-scoped event posting): it does NOT steal the user's cursor,
|
||||
keyboard focus, or Space. The schema is plain OpenAI function-calling so every
|
||||
tool-capable model can drive it; vision models get SOM captures (numbered
|
||||
overlays + AX tree) and click by element index, non-vision models use the AX
|
||||
tree alone.
|
||||
|
||||
* `tool.py` — `computer_use` handler, approval gate, response shaping.
|
||||
* `backend.py` — abstract `ComputerUseBackend` + result dataclasses.
|
||||
* `cua_backend.py` — default backend (MCP over stdio to `cua-driver`), with
|
||||
`cua_backend_parse` / `_session` / `_daemon` siblings.
|
||||
* `schema.py` — the model-facing schema (byte-frozen).
|
||||
|
||||
Model-facing guidance (workflow, background-first, escalate ladder, safety)
|
||||
Drives apps through cua-driver's background primitive (focus-without-raise + pid-scoped
|
||||
event posting): it does NOT steal the user's cursor, keyboard focus, or Space. Plain
|
||||
OpenAI function-calling schema; vision models get SOM captures (numbered overlays + AX
|
||||
tree) and click by index, non-vision models use the AX tree alone. Model-facing guidance
|
||||
lives in the schema description and each action result's `verdict`.
|
||||
|
||||
Modules: `tool.py` (handler, approval gate, response shaping), `backend.py` (abstract
|
||||
`ComputerUseBackend` + result dataclasses), `cua_backend.py` (default MCP-over-stdio
|
||||
backend + `cua_backend_parse`/`_session`/`_daemon` siblings), `schema.py` (byte-frozen).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
# Re-export the public surface so `from tools.computer_use import ...` works.
|
||||
from tools.computer_use.tool import ( # noqa: F401
|
||||
from tools.computer_use.tool import ( # noqa: F401 (public re-exports)
|
||||
handle_computer_use,
|
||||
release_computer_use_session,
|
||||
set_approval_callback,
|
||||
|
||||
+55
-128
@@ -1,8 +1,8 @@
|
||||
"""Abstract backend interface for computer use.
|
||||
|
||||
Any implementation (cua-driver over MCP, pyautogui, noop, future Linux/Windows)
|
||||
must return the shape described below. All methods synchronous; async is
|
||||
handled inside the backend implementation if needed.
|
||||
returns the shapes below. All methods are synchronous; async is handled inside
|
||||
the backend implementation if needed.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@@ -12,18 +12,13 @@ from abc import ABC, abstractmethod
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Any, Dict, List, Optional, Tuple
|
||||
|
||||
_JPEG_SOF_MARKERS = frozenset({
|
||||
0xC0, 0xC1, 0xC2, 0xC3, 0xC5, 0xC6, 0xC7,
|
||||
0xC9, 0xCA, 0xCB, 0xCD, 0xCE, 0xCF,
|
||||
})
|
||||
_JPEG_SOF_MARKERS = frozenset({0xC0, 0xC1, 0xC2, 0xC3, 0xC5, 0xC6, 0xC7, 0xC9, 0xCA, 0xCB, 0xCD, 0xCE, 0xCF})
|
||||
|
||||
|
||||
def image_dimensions_from_bytes(raw: bytes) -> Optional[Tuple[int, int]]:
|
||||
"""Return (width, height) for PNG / JPEG bytes, or None when unreadable.
|
||||
|
||||
PNG: IHDR. JPEG: walk segments (skipping 0xFF fill bytes) to the first SOF
|
||||
marker; stop at SOS. Used by the tool layer's provider min-size guard.
|
||||
"""
|
||||
"""(width, height) for PNG / JPEG bytes, or None when unreadable. PNG: IHDR. JPEG: walk
|
||||
segments (skipping 0xFF fill bytes) to the first SOF marker; stop at SOS. Used by the
|
||||
tool layer's provider min-size guard."""
|
||||
if raw.startswith(b"\x89PNG\r\n\x1a\n") and len(raw) >= 24:
|
||||
try:
|
||||
width, height = struct.unpack(">II", raw[16:24])
|
||||
@@ -43,9 +38,7 @@ def image_dimensions_from_bytes(raw: bytes) -> Optional[Tuple[int, int]]:
|
||||
i += 1
|
||||
if marker in {0xD8, 0xD9}:
|
||||
continue
|
||||
if marker == 0xDA:
|
||||
break
|
||||
if i + 2 > len(raw):
|
||||
if marker == 0xDA or i + 2 > len(raw):
|
||||
break
|
||||
segment_len = int.from_bytes(raw[i:i + 2], "big")
|
||||
if segment_len < 2 or i + segment_len > len(raw):
|
||||
@@ -70,12 +63,9 @@ class UIElement:
|
||||
pid: int = 0 # owning process PID
|
||||
window_id: int = 0 # SkyLight / CG window ID
|
||||
attributes: Dict[str, Any] = field(default_factory=dict)
|
||||
# Opaque per-snapshot element handle from cua-driver
|
||||
# (trycua/cua#1961 — Surface 6 of NousResearch/hermes-agent#47072).
|
||||
# When set, downstream calls can pass it alongside `index` for
|
||||
# explicit stale-detection: a stale token returns an error from
|
||||
# cua-driver rather than silently re-resolving to a different
|
||||
# element. None for pre-#1961 drivers that didn't carry the field.
|
||||
# Opaque per-snapshot handle from cua-driver. Passed alongside `index` for explicit
|
||||
# stale-detection: a stale token errors instead of silently re-resolving to a
|
||||
# different element. None for older drivers that lack the field.
|
||||
element_token: Optional[str] = None
|
||||
|
||||
def center(self) -> Tuple[int, int]:
|
||||
@@ -85,35 +75,23 @@ class UIElement:
|
||||
|
||||
@dataclass
|
||||
class CaptureResult:
|
||||
"""Result of a screen capture call.
|
||||
|
||||
At least one of png_b64 / elements is populated depending on capture mode:
|
||||
* mode="vision" → png_b64 only
|
||||
* mode="ax" → elements only
|
||||
* mode="som" → both (default): PNG already has numbered overlays
|
||||
drawn by the backend, and `elements` holds the
|
||||
matching index → element mapping.
|
||||
"""
|
||||
"""Result of a screen capture call. mode="vision" → png_b64 only; mode="ax" → elements
|
||||
only; mode="som" (default) → both: the PNG already carries numbered overlays drawn by
|
||||
the backend and `elements` holds the matching index → element mapping."""
|
||||
|
||||
mode: str
|
||||
width: int # screenshot width (logical px, pre-Anthropic-scale)
|
||||
height: int
|
||||
png_b64: Optional[str] = None
|
||||
elements: List[UIElement] = field(default_factory=list)
|
||||
# Optional: the target app/window the elements were captured for.
|
||||
app: str = ""
|
||||
app: str = "" # target app/window the elements were captured for
|
||||
window_title: str = ""
|
||||
# Raw bytes we sent to Anthropic, for token estimation.
|
||||
png_bytes_len: int = 0
|
||||
# Explicit MIME type for `png_b64` when the backend supplied it
|
||||
# (cua-driver-rs emits `mimeType` on every image part as of
|
||||
# trycua/cua#1961 — Surface 7 of NousResearch/hermes-agent#47072).
|
||||
# When None, downstream consumers fall back to base64-prefix
|
||||
# sniffing for back-compat with older drivers.
|
||||
png_bytes_len: int = 0 # raw bytes sent to Anthropic, for token estimation
|
||||
# MIME type of `png_b64` when the backend supplied it (cua-driver-rs emits `mimeType`
|
||||
# on every image part). None → consumers fall back to base64-prefix sniffing (older drivers).
|
||||
image_mime_type: Optional[str] = None
|
||||
# Optional guidance appended to the human-readable summary — used by
|
||||
# capture lanes that intentionally return no elements (e.g. full-screen
|
||||
# composited grabs) to tell the model how to reach an interactive lane.
|
||||
# Guidance appended to the summary by capture lanes that intentionally return no
|
||||
# elements (e.g. full-screen composited grabs) to point the model at an interactive lane.
|
||||
note: str = ""
|
||||
|
||||
|
||||
@@ -121,45 +99,35 @@ class CaptureResult:
|
||||
class ActionResult:
|
||||
"""Result of any action (click / type / scroll / drag / key / wait).
|
||||
|
||||
Beyond the transport-level ``ok`` flag, this carries cua-driver's
|
||||
structured action verdict so the model can follow the documented
|
||||
verify → escalate ladder (NousResearch/hermes-agent#67052). ``ok`` stays
|
||||
tool/transport success only — it is NOT the semantic verdict. Read
|
||||
``effect`` / ``escalation`` to decide the next rung. All structured
|
||||
fields are optional and additive: an older driver that omits
|
||||
``structuredContent`` leaves them ``None`` and behavior is unchanged.
|
||||
``ok`` is tool/transport success only — NOT the semantic verdict; read ``effect`` /
|
||||
``escalation`` (cua-driver's structured verdict) to pick the next rung of the
|
||||
verify → escalate ladder. Structured fields are optional and additive: an older
|
||||
driver that omits ``structuredContent`` leaves them ``None``, behavior unchanged.
|
||||
"""
|
||||
|
||||
ok: bool
|
||||
action: str
|
||||
message: str = "" # human-readable summary
|
||||
# Optional trailing screenshot — set when the caller asked for a
|
||||
# post-action capture or the backend always returns one.
|
||||
capture: Optional[CaptureResult] = None
|
||||
# Arbitrary extra fields for debugging / telemetry.
|
||||
meta: Dict[str, Any] = field(default_factory=dict)
|
||||
capture: Optional[CaptureResult] = None # trailing screenshot, when requested / always-on
|
||||
meta: Dict[str, Any] = field(default_factory=dict) # debugging / telemetry extras
|
||||
# ── cua-driver structured verdict (additive; None on old drivers) ──
|
||||
# AX read-back verification: True = driver read the effect back,
|
||||
# False = ran but unconfirmed, None = tool doesn't carry the field.
|
||||
verified: Optional[bool] = None
|
||||
# Confidence signal: "confirmed" | "unverifiable" | "suspected_noop".
|
||||
effect: Optional[str] = None
|
||||
# Machine-readable next-rung hint: {"recommended": "px"|"foreground"|"page",
|
||||
# "reason": str} — present only when the driver recommends climbing.
|
||||
verified: Optional[bool] = None # AX read-back: True confirmed, False unconfirmed, None n/a
|
||||
effect: Optional[str] = None # "confirmed" | "unverifiable" | "suspected_noop"
|
||||
# {"recommended": "px"|"foreground"|"page", "reason": str} — only when driver recommends climbing
|
||||
escalation: Optional[Dict[str, Any]] = None
|
||||
# Delivery rung that actually ran (e.g. "ax", "x11_pixel", "cgevent_fg").
|
||||
path: Optional[str] = None
|
||||
# True when an AX walk found no actionable elements (act by px instead).
|
||||
degraded: Optional[bool] = None
|
||||
# The delivery_mode the caller requested for this action, echoed back.
|
||||
delivery_mode: Optional[str] = None
|
||||
# A structured refusal code (e.g. "background_unavailable",
|
||||
# "foreground_unsupported", "desktop_scope_disabled") when present.
|
||||
code: Optional[str] = None
|
||||
path: Optional[str] = None # delivery rung that ran (e.g. "ax", "x11_pixel", "cgevent_fg")
|
||||
degraded: Optional[bool] = None # AX walk found no actionable elements (act by px instead)
|
||||
delivery_mode: Optional[str] = None # the delivery_mode the caller requested, echoed back
|
||||
code: Optional[str] = None # refusal code, e.g. "background_unavailable", "desktop_scope_disabled"
|
||||
|
||||
|
||||
class ComputerUseBackend(ABC):
|
||||
"""Lifecycle: `start()` before first use, `stop()` at shutdown."""
|
||||
"""Lifecycle: `start()` before first use, `stop()` at shutdown.
|
||||
|
||||
Pointer/keyboard actions take ``delivery_mode`` (background (default) | foreground)
|
||||
and ``bring_to_front``; ``button`` is left | right | middle; ``modifiers`` a list of
|
||||
key names. ``element`` args are 1-based SOM indices from a prior capture.
|
||||
"""
|
||||
|
||||
@abstractmethod
|
||||
def start(self) -> None: ...
|
||||
@@ -169,63 +137,30 @@ class ComputerUseBackend(ABC):
|
||||
|
||||
@abstractmethod
|
||||
def is_available(self) -> bool:
|
||||
"""Return True if the backend can be used on this host right now.
|
||||
|
||||
Used by check_fn gating and by the post-setup wizard.
|
||||
"""
|
||||
"""True if the backend can be used on this host right now (check_fn gating, setup wizard)."""
|
||||
|
||||
# ── Capture ─────────────────────────────────────────────────────
|
||||
@abstractmethod
|
||||
def capture(
|
||||
self,
|
||||
mode: str = "som",
|
||||
app: Optional[str] = None,
|
||||
pid: Optional[int] = None,
|
||||
window_id: Optional[int] = None,
|
||||
) -> CaptureResult: ...
|
||||
def capture(self, mode: str = "som", app: Optional[str] = None, pid: Optional[int] = None,
|
||||
window_id: Optional[int] = None) -> CaptureResult: ...
|
||||
|
||||
# ── Pointer actions ─────────────────────────────────────────────
|
||||
@abstractmethod
|
||||
def click(
|
||||
self,
|
||||
*,
|
||||
element: Optional[int] = None,
|
||||
x: Optional[int] = None,
|
||||
y: Optional[int] = None,
|
||||
button: str = "left", # left | right | middle
|
||||
click_count: int = 1,
|
||||
modifiers: Optional[List[str]] = None,
|
||||
delivery_mode: Optional[str] = None, # background (default) | foreground
|
||||
bring_to_front: bool = False,
|
||||
) -> ActionResult: ...
|
||||
def click(self, *, element: Optional[int] = None, x: Optional[int] = None, y: Optional[int] = None,
|
||||
button: str = "left", click_count: int = 1, modifiers: Optional[List[str]] = None,
|
||||
delivery_mode: Optional[str] = None, bring_to_front: bool = False) -> ActionResult: ...
|
||||
|
||||
@abstractmethod
|
||||
def drag(
|
||||
self,
|
||||
*,
|
||||
from_element: Optional[int] = None,
|
||||
to_element: Optional[int] = None,
|
||||
from_xy: Optional[Tuple[int, int]] = None,
|
||||
to_xy: Optional[Tuple[int, int]] = None,
|
||||
button: str = "left",
|
||||
modifiers: Optional[List[str]] = None,
|
||||
delivery_mode: Optional[str] = None,
|
||||
bring_to_front: bool = False,
|
||||
) -> ActionResult: ...
|
||||
def drag(self, *, from_element: Optional[int] = None, to_element: Optional[int] = None,
|
||||
from_xy: Optional[Tuple[int, int]] = None, to_xy: Optional[Tuple[int, int]] = None,
|
||||
button: str = "left", modifiers: Optional[List[str]] = None,
|
||||
delivery_mode: Optional[str] = None, bring_to_front: bool = False) -> ActionResult: ...
|
||||
|
||||
@abstractmethod
|
||||
def scroll(
|
||||
self,
|
||||
*,
|
||||
direction: str, # up | down | left | right
|
||||
amount: int = 3, # wheel ticks
|
||||
element: Optional[int] = None,
|
||||
x: Optional[int] = None,
|
||||
y: Optional[int] = None,
|
||||
modifiers: Optional[List[str]] = None,
|
||||
delivery_mode: Optional[str] = None,
|
||||
bring_to_front: bool = False,
|
||||
) -> ActionResult: ...
|
||||
def scroll(self, *, direction: str, amount: int = 3, element: Optional[int] = None,
|
||||
x: Optional[int] = None, y: Optional[int] = None, modifiers: Optional[List[str]] = None,
|
||||
delivery_mode: Optional[str] = None, bring_to_front: bool = False) -> ActionResult:
|
||||
"""`direction` is up | down | left | right; `amount` is wheel ticks."""
|
||||
|
||||
# ── Keyboard ────────────────────────────────────────────────────
|
||||
@abstractmethod
|
||||
@@ -243,26 +178,18 @@ class ComputerUseBackend(ABC):
|
||||
"""Return running apps with bundle IDs, PIDs, window counts."""
|
||||
|
||||
def list_windows(self) -> List[Dict[str, Any]]:
|
||||
"""Return visible native windows with PID and window identifiers.
|
||||
|
||||
Optional compatibility hook: backends that predate window discovery
|
||||
remain instantiable and simply report no windows.
|
||||
"""
|
||||
"""Visible native windows with PID and window identifiers. Optional compatibility
|
||||
hook: backends that predate window discovery stay instantiable and report none."""
|
||||
return []
|
||||
|
||||
@abstractmethod
|
||||
def focus_app(self, app: str, raise_window: bool = False) -> ActionResult:
|
||||
"""Route input to `app` (by name or bundle ID). Default: focus without raise."""
|
||||
|
||||
# ── Native-value mutation ────────────────────────────────────────
|
||||
@abstractmethod
|
||||
def set_value(self, value: str, element: Optional[int] = None) -> ActionResult:
|
||||
"""Set a native value on an element (e.g. AXPopUpButton selection).
|
||||
"""Set a native value on an element (e.g. AXPopUpButton selection)."""
|
||||
|
||||
`element` is the 1-based SOM index returned by a prior capture call.
|
||||
"""
|
||||
|
||||
# ── Timing ──────────────────────────────────────────────────────
|
||||
def wait(self, seconds: float) -> ActionResult:
|
||||
"""Default implementation: time.sleep."""
|
||||
import time
|
||||
|
||||
+206
-343
@@ -1,11 +1,10 @@
|
||||
"""`hermes computer-use doctor` — thin client for cua-driver's `health_report` MCP tool.
|
||||
|
||||
cua-driver owns the health model; this module drives the stdio JSON-RPC handshake,
|
||||
calls `health_report`, and renders the response. The only contract is the stable
|
||||
`schema_version="1"` payload shape. cua-driver 0.10.x marks `health_report`
|
||||
risk-unclassified (isError=true, structuredContent ``{"exit_code": 1}``) — we detect
|
||||
that and synthesize a composite report from working probes (check_permissions,
|
||||
list_apps, CLI --version).
|
||||
cua-driver owns the health model; we drive the stdio JSON-RPC handshake, call
|
||||
`health_report` and render the stable ``schema_version="1"`` payload. cua-driver 0.10.x
|
||||
marks `health_report` risk-unclassified (isError=true, structuredContent ``{"exit_code": 1}``)
|
||||
— we detect that and synthesize a composite report from working probes
|
||||
(check_permissions, list_apps, CLI --version).
|
||||
|
||||
Exit codes: 0 overall=="ok"; 1 degraded/failed; 2 binary missing / protocol error.
|
||||
"""
|
||||
@@ -13,16 +12,15 @@ Exit codes: 0 overall=="ok"; 1 degraded/failed; 2 binary missing / protocol erro
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import platform as _platform_mod
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
from contextlib import contextmanager
|
||||
from contextlib import contextmanager, suppress
|
||||
from typing import Any, Dict, Iterator, List, Optional, Sequence, Tuple
|
||||
|
||||
from hermes_cli._subprocess_compat import windows_hide_flags
|
||||
|
||||
from tools.computer_use.permissions import _child_env as _sanitized_cua_env
|
||||
|
||||
# Match the ALLOWED_STATUS_VALUES + ALLOWED_OVERALL_VALUES the cua-driver
|
||||
# integration test pins. If health_report widens its vocabulary, add here.
|
||||
@@ -30,57 +28,30 @@ _STATUS_GLYPH = {"pass": "✅", "fail": "❌", "skip": "⏭️"}
|
||||
_OVERALL_GLYPH = {"ok": "✅", "degraded": "⚠️", "failed": "❌"}
|
||||
_SUPPORTED_PLATFORMS = ("darwin", "linux", "windows")
|
||||
_TCC_HINT = "Grant {} to CuaDriver in System Settings → Privacy & Security."
|
||||
_ZERO_DISPLAY_MSG = "ScreenCaptureKit reachable but 0 shareable display(s) — every capture will return 0x0."
|
||||
_ZERO_DISPLAY_HINT = ("Wake the built-in display, connect a monitor or HDMI dummy dongle (e.g. Headless Ghost), "
|
||||
"or enable a virtual display (Screen Sharing/VNC, BetterDisplay). "
|
||||
"Verify with `system_profiler SPDisplaysDataType`.")
|
||||
Report = Dict[str, Any]
|
||||
|
||||
|
||||
class HealthReportUnavailable(RuntimeError):
|
||||
"""health_report denied or non-schema payload — ``run_doctor`` falls back to probes."""
|
||||
|
||||
|
||||
def _sanitized_cua_env() -> Dict[str, str]:
|
||||
"""cua-driver child env (telemetry policy + provider secrets stripped);
|
||||
degrades to ``os.environ`` on import error so doctor keeps working."""
|
||||
try:
|
||||
from tools.computer_use.cua_backend import sanitized_cua_driver_env
|
||||
|
||||
return sanitized_cua_driver_env()
|
||||
except Exception:
|
||||
return dict(os.environ)
|
||||
|
||||
|
||||
def _is_valid_health_report(payload: Any) -> bool:
|
||||
"""True when *payload* looks like a schema_version=1 health_report."""
|
||||
return (
|
||||
isinstance(payload, dict)
|
||||
and "schema_version" in payload
|
||||
and "overall" in payload
|
||||
and isinstance(payload.get("checks"), list)
|
||||
)
|
||||
|
||||
# ── CLI probes ───────────────────────────────────────────────────────────────
|
||||
|
||||
def _run_cli(binary: str, *args: str, timeout: float) -> subprocess.CompletedProcess:
|
||||
"""Run ``<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 _combined_output(completed: subprocess.CompletedProcess) -> str:
|
||||
return ((completed.stdout or "") + (completed.stderr or "")).strip()
|
||||
|
||||
def _read_cli_version(binary: str, *, timeout: float = 5.0) -> Optional[str]:
|
||||
"""Return ``cua-driver --version`` first line (stripped), or None on failure.
|
||||
|
||||
health_report's ``driver_version`` can disagree with the actual binary
|
||||
(observed on Windows); doctor surfaces both so operators are not misled.
|
||||
"""
|
||||
"""First line of ``cua-driver --version`` or None. health_report's ``driver_version``
|
||||
can disagree with the real binary (seen on Windows); doctor surfaces both."""
|
||||
try:
|
||||
completed = _run_cli(binary, "--version", timeout=timeout)
|
||||
except (OSError, subprocess.TimeoutExpired, ValueError, TypeError):
|
||||
@@ -88,100 +59,102 @@ def _read_cli_version(binary: str, *, timeout: float = 5.0) -> Optional[str]:
|
||||
text = (completed.stdout or completed.stderr or "").strip()
|
||||
return text.splitlines()[0].strip() if text else None
|
||||
|
||||
def _cli_driver_version(binary: str, timeout: float = 5.0) -> Tuple[str, Optional[str]]:
|
||||
"""Return (status, version_or_message) from ``cua-driver --version``."""
|
||||
try:
|
||||
completed = _run_cli(binary, "--version", timeout=timeout)
|
||||
except (OSError, subprocess.TimeoutExpired) as e:
|
||||
return "fail", f"--version failed: {e}"
|
||||
text, failed = _combined_output(completed), completed.returncode != 0
|
||||
if failed and not text:
|
||||
return "fail", f"--version exited {completed.returncode}"
|
||||
m = re.search(r"(\d+\.\d+\.\d+(?:[-+][\w.]+)?)", text) # typical: "cua-driver 0.10.0"
|
||||
return ("fail" if failed else "pass"), m.group(1) if m else (text.splitlines()[0] if text else "unknown")
|
||||
|
||||
def _normalize_version_token(text: str) -> str:
|
||||
"""Pull a dotted version-ish token out of a free-form version string."""
|
||||
if not text:
|
||||
return ""
|
||||
m = re.search(r"(\d+\.\d+(?:\.\d+)?(?:[-+][\w.]+)?)", text)
|
||||
return m.group(1) if m else text.strip().lower()
|
||||
def _cli_doctor_snippet(binary: str, timeout: float = 8.0) -> Optional[str]:
|
||||
"""Optional one-shot ``cua-driver doctor`` text (best-effort, never fatal)."""
|
||||
try:
|
||||
return _combined_output(_run_cli(binary, "doctor", timeout=timeout)) or None
|
||||
except (OSError, subprocess.TimeoutExpired):
|
||||
return None
|
||||
|
||||
|
||||
def _build_identity(binary: str, report: Dict[str, Any]) -> Dict[str, Any]:
|
||||
def _build_identity(binary: str, report: Report) -> Report:
|
||||
"""Hermes-side identity block comparing resolved binary vs health_report."""
|
||||
def token(text: str) -> str: # dotted version-ish token out of a free-form string
|
||||
m = text and re.search(r"(\d+\.\d+(?:\.\d+)?(?:[-+][\w.]+)?)", text)
|
||||
return m.group(1) if m else text.strip().lower()
|
||||
|
||||
cli = _read_cli_version(binary) or ""
|
||||
report_v = str(report.get("driver_version") or "")
|
||||
cli_tok = _normalize_version_token(cli)
|
||||
report_tok = _normalize_version_token(report_v)
|
||||
return {
|
||||
"resolved_binary": binary,
|
||||
"cli_version": cli or None,
|
||||
"health_report_driver_version": report_v or None,
|
||||
"version_mismatch": bool(cli_tok and report_tok and cli_tok != report_tok),
|
||||
}
|
||||
cli_tok, report_tok = token(cli), token(report_v)
|
||||
return {"resolved_binary": binary, "cli_version": cli or None, "health_report_driver_version": report_v or None,
|
||||
"version_mismatch": bool(cli_tok and report_tok and cli_tok != report_tok)}
|
||||
|
||||
|
||||
def _extract_health_report_from_result(result: Dict[str, Any]) -> Dict[str, Any]:
|
||||
# ── MCP transport ────────────────────────────────────────────────────────────
|
||||
|
||||
def _is_valid_health_report(payload: Any) -> bool:
|
||||
"""True when *payload* looks like a schema_version=1 health_report."""
|
||||
return isinstance(payload, dict) and {"schema_version", "overall"} <= payload.keys() and isinstance(payload.get("checks"), list)
|
||||
|
||||
def _text_items(result: Report) -> Iterator[str]:
|
||||
"""Text of every ``{"type": "text"}`` content item of an MCP tools/call result."""
|
||||
for item in result.get("content") or []:
|
||||
if isinstance(item, dict) and item.get("type") == "text":
|
||||
yield item.get("text") or ""
|
||||
|
||||
def _first_text(result: Report, default: str) -> str:
|
||||
"""First non-empty text content item, else *default*."""
|
||||
return next((t.strip() for t in _text_items(result) if t.strip()), default)
|
||||
|
||||
def _extract_health_report_from_result(result: Report) -> Report:
|
||||
"""Pull a schema_version=1 report out of an MCP tools/call result.
|
||||
|
||||
Raises ``HealthReportUnavailable`` when the tool denied the call (isError)
|
||||
or the payload is not a real health report (0.10's ``{"exit_code": 1}``);
|
||||
``RuntimeError`` when the response carries no content at all.
|
||||
Raises ``HealthReportUnavailable`` when the tool denied the call (isError) or the
|
||||
payload is not a real report (0.10's ``{"exit_code": 1}``); ``RuntimeError`` when
|
||||
the response carries no content at all.
|
||||
"""
|
||||
if result.get("isError") is True:
|
||||
raise HealthReportUnavailable(_first_text(result, "health_report returned isError=true"))
|
||||
|
||||
sc = result.get("structuredContent")
|
||||
if _is_valid_health_report(sc):
|
||||
return sc # type: ignore[return-value]
|
||||
|
||||
# Older builds: JSON text block with schema_version.
|
||||
for item in result.get("content") or []:
|
||||
if not isinstance(item, dict) or item.get("type") != "text":
|
||||
continue
|
||||
try:
|
||||
parsed = json.loads(item.get("text", ""))
|
||||
except (ValueError, TypeError):
|
||||
continue
|
||||
if _is_valid_health_report(parsed):
|
||||
return parsed
|
||||
|
||||
# structuredContent present but not a real report — unavailable, not fatal protocol.
|
||||
if isinstance(sc, dict):
|
||||
raise HealthReportUnavailable(
|
||||
"health_report structuredContent lacks schema_version/overall/checks "
|
||||
f"(keys={sorted(sc.keys())})"
|
||||
)
|
||||
raise RuntimeError(
|
||||
"health_report response carried neither structuredContent nor a parseable "
|
||||
f"JSON text block. Result keys: {list(result.keys())}"
|
||||
)
|
||||
|
||||
for text in _text_items(result): # older builds: JSON text block with schema_version
|
||||
with suppress(ValueError, TypeError):
|
||||
parsed = json.loads(text)
|
||||
if _is_valid_health_report(parsed):
|
||||
return parsed
|
||||
if isinstance(sc, dict): # present but not a real report — unavailable, not fatal
|
||||
raise HealthReportUnavailable("health_report structuredContent lacks schema_version/overall/checks "
|
||||
f"(keys={sorted(sc.keys())})")
|
||||
raise RuntimeError("health_report response carried neither structuredContent nor a parseable "
|
||||
f"JSON text block. Result keys: {list(result.keys())}")
|
||||
|
||||
def _open_mcp(binary: str) -> subprocess.Popen:
|
||||
"""Spawn ``<binary> mcp`` with UTF-8 + sanitized 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())
|
||||
|
||||
cua-driver emits UTF-8 (emoji, arbitrary paths); the locale default
|
||||
(`cp1252` on Windows) would raise UnicodeDecodeError, so pin the codec.
|
||||
"""
|
||||
return subprocess.Popen(
|
||||
[binary, "mcp"], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
|
||||
text=True, encoding="utf-8", errors="replace", bufsize=1,
|
||||
creationflags=windows_hide_flags(), env=_sanitized_cua_env(),
|
||||
)
|
||||
def _stderr_tail(proc: subprocess.Popen) -> List[str]:
|
||||
"""Last 3 stderr lines of *proc* (best-effort, ``[]`` when unreadable)."""
|
||||
with suppress(Exception):
|
||||
if proc.stderr is not None:
|
||||
return [str(x) for x in (proc.stderr.read() or "").strip().splitlines()[-3:]]
|
||||
return []
|
||||
|
||||
|
||||
def _mcp_rpc(proc: subprocess.Popen, msg_id: int, method: str, params: Any = None) -> Dict[str, Any]:
|
||||
def _mcp_rpc(proc: subprocess.Popen, msg_id: int, method: str, params: Any = None) -> Report:
|
||||
"""Write one JSON-RPC request and read one response line."""
|
||||
assert proc.stdin is not None and proc.stdout is not None
|
||||
payload: Dict[str, Any] = {"jsonrpc": "2.0", "id": msg_id, "method": method}
|
||||
payload: Report = {"jsonrpc": "2.0", "id": msg_id, "method": method}
|
||||
if params is not None:
|
||||
payload["params"] = params
|
||||
proc.stdin.write(json.dumps(payload) + "\n")
|
||||
proc.stdin.flush()
|
||||
line = proc.stdout.readline()
|
||||
if not line:
|
||||
stderr_tail: List[str] = []
|
||||
if proc.stderr is not None:
|
||||
try:
|
||||
raw_err = proc.stderr.read() or ""
|
||||
stderr_tail = [str(x) for x in raw_err.strip().splitlines()[-3:]]
|
||||
except Exception:
|
||||
pass
|
||||
raise RuntimeError(
|
||||
f"cua-driver mcp produced no response for {method!r}. "
|
||||
f"stderr tail: {stderr_tail or '(empty)'}"
|
||||
)
|
||||
raise RuntimeError(f"cua-driver mcp produced no response for {method!r}. stderr tail: {_stderr_tail(proc) or '(empty)'}")
|
||||
try:
|
||||
resp = json.loads(line)
|
||||
except (ValueError, TypeError) as e:
|
||||
@@ -190,12 +163,9 @@ def _mcp_rpc(proc: subprocess.Popen, msg_id: int, method: str, params: Any = Non
|
||||
raise RuntimeError(f"{method} JSON-RPC error: {resp['error']}")
|
||||
return resp
|
||||
|
||||
|
||||
def _call_tool(proc: subprocess.Popen, msg_id: int, name: str, arguments: Any = None) -> Any:
|
||||
"""tools/call *name* and return the raw ``result`` value (``{}`` when absent)."""
|
||||
resp = _mcp_rpc(proc, msg_id, "tools/call", {"name": name, "arguments": arguments or {}})
|
||||
return resp.get("result") or {}
|
||||
|
||||
return _mcp_rpc(proc, msg_id, "tools/call", {"name": name, "arguments": arguments or {}}).get("result") or {}
|
||||
|
||||
@contextmanager
|
||||
def _mcp_session(binary: str, timeout: float) -> Iterator[subprocess.Popen]:
|
||||
@@ -204,25 +174,19 @@ def _mcp_session(binary: str, timeout: float) -> Iterator[subprocess.Popen]:
|
||||
try:
|
||||
yield proc
|
||||
finally:
|
||||
try:
|
||||
with suppress(Exception):
|
||||
if proc.stdin is not None:
|
||||
proc.stdin.close()
|
||||
except Exception:
|
||||
pass
|
||||
try:
|
||||
proc.wait(timeout=timeout)
|
||||
except subprocess.TimeoutExpired:
|
||||
proc.kill()
|
||||
proc.wait()
|
||||
|
||||
|
||||
def _drive_health_report(binary: str, *, include: Sequence[str] = (), skip: Sequence[str] = (),
|
||||
timeout: float = 12.0) -> Dict[str, Any]:
|
||||
"""Spawn `<binary> mcp`, handshake, call `health_report`, return the parsed report.
|
||||
|
||||
Raises HealthReportUnavailable (denied / non-schema — caller falls back) or
|
||||
RuntimeError (protocol-level failure).
|
||||
"""
|
||||
timeout: float = 12.0) -> Report:
|
||||
"""Handshake + `health_report` → parsed report. Raises HealthReportUnavailable
|
||||
(denied / non-schema — caller falls back) or RuntimeError (protocol failure)."""
|
||||
args = {k: list(v) for k, v in (("include", include), ("skip", skip)) if v}
|
||||
with _mcp_session(binary, timeout) as proc:
|
||||
_mcp_rpc(proc, 1, "initialize", {})
|
||||
@@ -232,106 +196,71 @@ def _drive_health_report(binary: str, *, include: Sequence[str] = (), skip: Sequ
|
||||
return _extract_health_report_from_result(result)
|
||||
|
||||
|
||||
def _cli_driver_version(binary: str, timeout: float = 5.0) -> Tuple[str, Optional[str]]:
|
||||
"""Return (status, version_or_message) from ``cua-driver --version``."""
|
||||
# ── 0.10 fallback: compose a report from working probes ──────────────────────
|
||||
|
||||
def _probe_tool(proc: subprocess.Popen, msg_id: int, name: str) -> Tuple[Optional[Report], Optional[str]]:
|
||||
"""``(result, None)`` on success; ``(None, error_text)`` on isError or RPC failure."""
|
||||
try:
|
||||
completed = _run_cli(binary, "--version", timeout=timeout)
|
||||
except (OSError, subprocess.TimeoutExpired) as e:
|
||||
return "fail", f"--version failed: {e}"
|
||||
result = _call_tool(proc, msg_id, name)
|
||||
except RuntimeError as e:
|
||||
return None, str(e)
|
||||
if result.get("isError") is True:
|
||||
return None, _first_text(result, f"{name} isError")
|
||||
return result, None
|
||||
|
||||
text = ((completed.stdout or "") + (completed.stderr or "")).strip()
|
||||
if completed.returncode != 0 and not text:
|
||||
return "fail", f"--version exited {completed.returncode}"
|
||||
|
||||
# Typical: "cua-driver 0.10.0"
|
||||
m = re.search(r"(\d+\.\d+\.\d+(?:[-+][\w.]+)?)", text)
|
||||
version = m.group(1) if m else (text.splitlines()[0] if text else "unknown")
|
||||
return ("fail" if completed.returncode != 0 else "pass"), version
|
||||
|
||||
|
||||
def _cli_doctor_snippet(binary: str, timeout: float = 8.0) -> Optional[str]:
|
||||
"""Optional one-shot ``cua-driver doctor`` text (best-effort, never fatal)."""
|
||||
try:
|
||||
completed = _run_cli(binary, "doctor", timeout=timeout)
|
||||
except (OSError, subprocess.TimeoutExpired):
|
||||
return None
|
||||
return ((completed.stdout or "") + (completed.stderr or "")).strip() or None
|
||||
|
||||
|
||||
def _drive_fallback_probes(binary: str, *, timeout: float = 12.0) -> Dict[str, Any]:
|
||||
def _drive_fallback_probes(binary: str, *, timeout: float = 12.0) -> Report:
|
||||
"""Call working MCP tools (check_permissions, list_apps) in one session.
|
||||
|
||||
Returns init_version (initialize serverInfo), permissions (structuredContent
|
||||
dict | None), permissions_error, list_apps_ok, list_apps_error, list_apps_count.
|
||||
"""
|
||||
out: Dict[str, Any] = dict.fromkeys((
|
||||
"init_version", "permissions", "permissions_error",
|
||||
"list_apps_ok", "list_apps_error", "list_apps_count",
|
||||
))
|
||||
out: Report = dict.fromkeys(("init_version", "permissions", "permissions_error",
|
||||
"list_apps_ok", "list_apps_error", "list_apps_count"))
|
||||
with _mcp_session(binary, timeout) as proc:
|
||||
init_resp = _mcp_rpc(proc, 1, "initialize", {})
|
||||
server_info = ((init_resp.get("result") or {}).get("serverInfo") or {})
|
||||
if isinstance(server_info, dict):
|
||||
out["init_version"] = server_info.get("version")
|
||||
|
||||
# check_permissions — primary TCC signal on 0.10
|
||||
try:
|
||||
perm_result = _call_tool(proc, 2, "check_permissions")
|
||||
if perm_result.get("isError") is True:
|
||||
out["permissions_error"] = _first_text(perm_result, "check_permissions isError")
|
||||
else:
|
||||
sc = perm_result.get("structuredContent")
|
||||
out["permissions"] = sc if isinstance(sc, dict) else {}
|
||||
except RuntimeError as e:
|
||||
out["permissions_error"] = str(e)
|
||||
|
||||
perms, err = _probe_tool(proc, 2, "check_permissions") # primary TCC signal on 0.10
|
||||
if perms is None:
|
||||
out["permissions_error"] = err
|
||||
else:
|
||||
sc = perms.get("structuredContent")
|
||||
out["permissions"] = sc if isinstance(sc, dict) else {}
|
||||
# list_apps — light AX capability probe; text-only success still counts as AX working
|
||||
try:
|
||||
apps_result = _call_tool(proc, 3, "list_apps")
|
||||
if apps_result.get("isError") is True:
|
||||
out["list_apps_ok"] = False
|
||||
out["list_apps_error"] = _first_text(apps_result, "list_apps isError")
|
||||
else:
|
||||
sc = apps_result.get("structuredContent") or {}
|
||||
apps = sc.get("apps") if isinstance(sc, dict) else None
|
||||
out["list_apps_ok"] = True
|
||||
out["list_apps_count"] = len(apps) if isinstance(apps, list) else None
|
||||
except RuntimeError as e:
|
||||
out["list_apps_ok"] = False
|
||||
out["list_apps_error"] = str(e)
|
||||
apps, err = _probe_tool(proc, 3, "list_apps")
|
||||
out["list_apps_ok"] = apps is not None
|
||||
if apps is None:
|
||||
out["list_apps_error"] = err
|
||||
else:
|
||||
sc = apps.get("structuredContent") or {}
|
||||
app_list = sc.get("apps") if isinstance(sc, dict) else None
|
||||
out["list_apps_count"] = len(app_list) if isinstance(app_list, list) else None
|
||||
return out
|
||||
|
||||
|
||||
def _platform_name() -> str:
|
||||
sysname = (_platform_mod.system() or "").lower()
|
||||
return sysname if sysname in _SUPPORTED_PLATFORMS else (sysname or "unknown")
|
||||
return (_platform_mod.system() or "").lower() or "unknown"
|
||||
|
||||
|
||||
def _check(name: str, status: str, message: str, **extra: Any) -> Dict[str, Any]:
|
||||
def _check(name: str, status: str, message: str, **extra: Any) -> Report:
|
||||
"""Build one health check dict (``hint`` / ``data`` only when given)."""
|
||||
return {"name": name, "status": status, "message": message, **extra}
|
||||
|
||||
|
||||
def _tcc_checks(perms: Optional[Dict[str, Any]], perm_err: Optional[str], plat: str) -> List[Dict[str, Any]]:
|
||||
def _tcc_checks(perms: Optional[Report], perm_err: Optional[str], plat: str) -> List[Report]:
|
||||
"""tcc_accessibility + tcc_screen_recording checks from check_permissions output."""
|
||||
if perms is None:
|
||||
status = "fail" if perm_err else "skip"
|
||||
msg = perm_err or "check_permissions unavailable"
|
||||
status, msg = ("fail" if perm_err else "skip"), perm_err or "check_permissions unavailable"
|
||||
return [_check("tcc_accessibility", status, msg), _check("tcc_screen_recording", status, msg)]
|
||||
|
||||
# Only real booleans select a branch; anything else (missing/odd) is the "absent" row.
|
||||
ax, scr, capturable = (perms.get(k) for k in ("accessibility", "screen_recording", "screen_recording_capturable"))
|
||||
ax = ax if isinstance(ax, bool) else None
|
||||
scr = scr if isinstance(scr, bool) else None
|
||||
|
||||
ax_rows = {
|
||||
True: ("pass", "Accessibility is granted.", {"data": {"accessibility": True}}),
|
||||
False: ("fail", "Accessibility is not granted.",
|
||||
{"hint": _TCC_HINT.format("Accessibility"), "data": {"accessibility": False}}),
|
||||
None: ("skip", "accessibility field absent from check_permissions", {}),
|
||||
}
|
||||
scr_rows = {
|
||||
# (scr, capturable is False) — the granted-but-not-capturable row wins first.
|
||||
scr_rows = { # (scr, capturable is False) — the granted-but-not-capturable row wins first.
|
||||
(True, True): ("fail", "Screen Recording granted but not capturable.",
|
||||
{"hint": "Screen Recording permission may need a restart of CuaDriver "
|
||||
"or a re-grant in System Settings.",
|
||||
@@ -345,16 +274,11 @@ def _tcc_checks(perms: Optional[Dict[str, Any]], perm_err: Optional[str], plat:
|
||||
}
|
||||
ax_status, ax_msg, ax_extra = ax_rows[ax]
|
||||
scr_status, scr_msg, scr_extra = scr_rows[(scr, capturable is False and scr is True)]
|
||||
return [
|
||||
_check("tcc_accessibility", ax_status, ax_msg, **ax_extra),
|
||||
_check("tcc_screen_recording", scr_status, scr_msg, **scr_extra),
|
||||
]
|
||||
return [_check("tcc_accessibility", ax_status, ax_msg, **ax_extra), _check("tcc_screen_recording", scr_status, scr_msg, **scr_extra)]
|
||||
|
||||
|
||||
def _ax_capability_check(probes: Dict[str, Any], ax_granted: bool) -> Dict[str, Any]:
|
||||
def _ax_capability_check(probes: Report, ax_granted: bool) -> Report:
|
||||
"""ax_capability — inferred from list_apps success or the accessibility grant."""
|
||||
list_ok = probes.get("list_apps_ok")
|
||||
list_count = probes.get("list_apps_count")
|
||||
list_ok, list_count = probes.get("list_apps_ok"), probes.get("list_apps_count")
|
||||
if list_ok is True:
|
||||
count_msg = f" ({list_count} apps)" if isinstance(list_count, int) else ""
|
||||
return _check("ax_capability", "pass", f"list_apps succeeded{count_msg}")
|
||||
@@ -365,214 +289,153 @@ def _ax_capability_check(probes: Dict[str, Any], ax_granted: bool) -> Dict[str,
|
||||
return _check("ax_capability", "pass", "inferred from accessibility grant (list_apps not probed)")
|
||||
return _check("ax_capability", "skip", "not probed")
|
||||
|
||||
def _overall_from(checks: List[Report]) -> str:
|
||||
"""failed if binary missing/bad; ok if accessibility fine and nothing failed;
|
||||
otherwise degraded (screen recording or accessibility problems)."""
|
||||
by_name = {c.get("name"): c.get("status") for c in checks}
|
||||
if by_name.get("binary_version") != "pass":
|
||||
return "failed"
|
||||
ax_ok = by_name.get("tcc_accessibility") in ("pass", "skip", None)
|
||||
return "ok" if ax_ok and not any(c.get("status") == "fail" for c in checks) else "degraded"
|
||||
|
||||
def _compose_fallback_report(binary: str, *, reason: str = "", timeout: float = 12.0) -> Dict[str, Any]:
|
||||
"""Build a schema_version=1 report from CLI + working MCP probes.
|
||||
|
||||
Used when ``health_report`` is denied (unclassified risk on 0.10) or
|
||||
returns a non-schema payload. Compatible with ``_print_text_report``.
|
||||
"""
|
||||
def _compose_fallback_report(binary: str, *, reason: str = "", timeout: float = 12.0) -> Report:
|
||||
"""Build a schema_version=1 report from CLI + working MCP probes when
|
||||
``health_report`` is denied (0.10) or non-schema. Renders via ``_print_text_report``."""
|
||||
plat = _platform_name()
|
||||
|
||||
ver_status, ver_value = _cli_driver_version(binary)
|
||||
driver_version = ver_value if ver_status == "pass" else (ver_value or "?")
|
||||
# Prefer MCP initialize version when CLI parse is messy
|
||||
probes = _drive_fallback_probes(binary, timeout=timeout)
|
||||
if probes.get("init_version"):
|
||||
driver_version = str(probes["init_version"])
|
||||
ver_status = "pass"
|
||||
if probes.get("init_version"): # MCP initialize version beats a messy CLI parse
|
||||
ver_status, driver_version = "pass", str(probes["init_version"])
|
||||
ver_msg = f"cua-driver {driver_version}"
|
||||
else:
|
||||
driver_version = ver_value if ver_status == "pass" else (ver_value or "?")
|
||||
ver_msg = f"cua-driver {ver_value}" if ver_status == "pass" else (ver_value or "version unknown")
|
||||
|
||||
supported = plat in _SUPPORTED_PLATFORMS
|
||||
checks: List[Dict[str, Any]] = [
|
||||
perms = probes.get("permissions") if isinstance(probes.get("permissions"), dict) else None
|
||||
reason_short = (reason or "health_report unavailable").strip()
|
||||
if len(reason_short) > 160:
|
||||
reason_short = reason_short[:157] + "..."
|
||||
checks: List[Report] = [
|
||||
_check("binary_version", ver_status, ver_msg),
|
||||
_check("platform_supported", "pass" if supported else "fail",
|
||||
f"platform={plat}" + ("" if supported else " (unsupported)")),
|
||||
# doctor does not start a session, so session_active is never probed
|
||||
_check("session_active", "skip", "not probed (doctor does not open a cua session)"),
|
||||
*_tcc_checks(perms, probes.get("permissions_error"), plat),
|
||||
_ax_capability_check(probes, bool(perms and perms.get("accessibility") is True)),
|
||||
_check("health_report_path", "skip",
|
||||
f"fallback composite (cua-driver 0.10 unclassified health_report); cause: {reason_short}"),
|
||||
]
|
||||
|
||||
perms = probes.get("permissions") if isinstance(probes.get("permissions"), dict) else None
|
||||
checks += _tcc_checks(perms, probes.get("permissions_error"), plat)
|
||||
checks.append(_ax_capability_check(probes, bool(perms and perms.get("accessibility") is True)))
|
||||
|
||||
# Annotate that we used the fallback path
|
||||
reason_short = (reason or "health_report unavailable").strip()
|
||||
if len(reason_short) > 160:
|
||||
reason_short = reason_short[:157] + "..."
|
||||
checks.append(_check(
|
||||
"health_report_path", "skip",
|
||||
f"fallback composite (cua-driver 0.10 unclassified health_report); cause: {reason_short}",
|
||||
))
|
||||
|
||||
# Optional CLI doctor text (best-effort)
|
||||
doctor_txt = _cli_doctor_snippet(binary)
|
||||
doctor_txt = _cli_doctor_snippet(binary) # optional CLI doctor text (best-effort)
|
||||
if doctor_txt:
|
||||
cli_ok = "[ok" in doctor_txt.lower() or "ok ]" in doctor_txt
|
||||
checks.append(_check("cli_doctor", "pass" if cli_ok else "skip",
|
||||
doctor_txt.splitlines()[0].strip(), data={"snippet": doctor_txt[:2000]}))
|
||||
|
||||
# Normalize any accidental non-vocab status values
|
||||
for c in checks:
|
||||
for c in checks: # normalize any accidental non-vocab status values
|
||||
if c.get("status") not in ("pass", "fail", "skip"):
|
||||
c["status"] = "fail"
|
||||
return {"schema_version": "1", "platform": plat, "driver_version": str(driver_version),
|
||||
"overall": _overall_from(checks), "checks": checks,
|
||||
"fallback": True, "fallback_reason": reason or "health_report unavailable"}
|
||||
|
||||
# overall: failed if binary missing/bad; ok if accessibility fine and nothing
|
||||
# failed; otherwise degraded (screen recording or accessibility problems).
|
||||
status_by_name = {c.get("name"): c.get("status") for c in checks}
|
||||
fail_count = sum(1 for c in checks if c.get("status") == "fail")
|
||||
if status_by_name.get("binary_version") != "pass":
|
||||
overall = "failed"
|
||||
elif status_by_name.get("tcc_accessibility") in ("pass", "skip", None) and fail_count == 0:
|
||||
overall = "ok"
|
||||
else:
|
||||
overall = "degraded"
|
||||
|
||||
return {
|
||||
"schema_version": "1",
|
||||
"platform": plat,
|
||||
"driver_version": str(driver_version),
|
||||
"overall": overall,
|
||||
"checks": checks,
|
||||
"fallback": True,
|
||||
"fallback_reason": reason or "health_report unavailable",
|
||||
}
|
||||
|
||||
|
||||
def _drive_health_report_or_fallback(binary: str, *, include: Sequence[str] = (), skip: Sequence[str] = (),
|
||||
timeout: float = 12.0) -> Dict[str, Any]:
|
||||
"""Prefer real health_report; on denial/non-schema, synthesize via probes."""
|
||||
try:
|
||||
report = _drive_health_report(binary, include=include, skip=skip, timeout=timeout)
|
||||
except HealthReportUnavailable as e:
|
||||
report = _compose_fallback_report(binary, reason=str(e), timeout=timeout)
|
||||
return _apply_display_count_guard(report)
|
||||
|
||||
|
||||
def _apply_display_count_guard(report: Dict[str, Any]) -> Dict[str, Any]:
|
||||
def _apply_display_count_guard(report: Report) -> Report:
|
||||
"""Downgrade an 'ok' report whose screen capture has zero displays.
|
||||
|
||||
macOS ScreenCaptureKit reports ``display_count=0`` on headless Macs and
|
||||
when the built-in panel is asleep — TCC grants are fine, health_report
|
||||
can still say pass/ok, but every capture will come back 0x0. Marking the
|
||||
check failed turns a silent failure into an actionable one. Applied at
|
||||
the report seam so both the real and the composed fallback path get it.
|
||||
macOS ScreenCaptureKit reports ``display_count=0`` on headless Macs and when the
|
||||
built-in panel is asleep — TCC grants are fine, health_report can still say
|
||||
pass/ok, but every capture comes back 0x0. Failing the check turns a silent
|
||||
failure into an actionable one. Applied at the report seam so both the real and
|
||||
the composed fallback path get it.
|
||||
"""
|
||||
checks = report.get("checks")
|
||||
if not isinstance(checks, list):
|
||||
return report
|
||||
for check in checks:
|
||||
for check in checks if isinstance(checks, list) else ():
|
||||
if not isinstance(check, dict) or check.get("name") != "screen_capture_capability":
|
||||
continue
|
||||
data = check.get("data")
|
||||
count = data.get("display_count") if isinstance(data, dict) else None
|
||||
if count == 0 and check.get("status") == "pass":
|
||||
check["status"] = "fail"
|
||||
check["message"] = "ScreenCaptureKit reachable but 0 shareable display(s) — every capture will return 0x0."
|
||||
check["hint"] = (
|
||||
"Wake the built-in display, connect a monitor or HDMI dummy dongle (e.g. Headless Ghost), "
|
||||
"or enable a virtual display (Screen Sharing/VNC, BetterDisplay). "
|
||||
"Verify with `system_profiler SPDisplaysDataType`."
|
||||
)
|
||||
if (data.get("display_count") if isinstance(data, dict) else None) == 0 and check.get("status") == "pass":
|
||||
check.update(status="fail", message=_ZERO_DISPLAY_MSG, hint=_ZERO_DISPLAY_HINT)
|
||||
if report.get("overall") == "ok":
|
||||
report["overall"] = "degraded"
|
||||
return report
|
||||
|
||||
|
||||
def _print_text_report(report: Dict[str, Any], color: bool, *, identity: Optional[Dict[str, Any]] = None) -> None:
|
||||
"""Render the report like `cua-driver call health_report` (one line per check).
|
||||
# ── Rendering ────────────────────────────────────────────────────────────────
|
||||
|
||||
def _check_lines(check: Report, status_cols: Dict[str, str], reset: str, dim: str) -> List[str]:
|
||||
"""One line per check, plus indented hint and ``data`` rows (structured payload
|
||||
some checks attach — bundle id, AX state, version triple — support staff need it)."""
|
||||
status = check.get("status", "?")
|
||||
lines = [f" {_STATUS_GLYPH.get(status, '•')} {status_cols.get(status, '')}{check.get('name', '?')}{reset}: "
|
||||
f"{check.get('message') or ''}"]
|
||||
if check.get("hint"):
|
||||
lines.append(f" → {dim}{check['hint']}{reset}")
|
||||
data = check.get("data")
|
||||
for key, value in (data.items() if isinstance(data, dict) else ()):
|
||||
lines.append(f" {dim}{key}={json.dumps(value) if isinstance(value, (dict, list)) else value}{reset}")
|
||||
return lines
|
||||
|
||||
def _print_text_report(report: Report, color: bool, *, identity: Optional[Report] = None) -> None:
|
||||
"""Render the report like `cua-driver call health_report` (one line per check).
|
||||
With *identity* (resolved binary + ``--version``) the header prefers the CLI
|
||||
version over health_report's ``driver_version`` and prints an identity block.
|
||||
"""
|
||||
platform = report.get("platform", "?")
|
||||
report_v = report.get("driver_version", "?")
|
||||
overall = report.get("overall", "?")
|
||||
version over health_report's ``driver_version`` and prints an identity block."""
|
||||
platform, report_v, overall = (report.get(k, "?") for k in ("platform", "driver_version", "overall"))
|
||||
identity = identity or {}
|
||||
cli_v = identity.get("cli_version") or ""
|
||||
mismatch = bool(identity.get("version_mismatch"))
|
||||
header_v = cli_v or report_v # binary's own --version wins when health_report is stale
|
||||
|
||||
# No external color library — keep ANSI inline so doctor stays self-contained.
|
||||
# No external color library — inline ANSI keeps doctor self-contained.
|
||||
# Colors only apply when overall is a known vocabulary value.
|
||||
ansi = ("\033[31m", "\033[33m", "\033[32m", "\033[0m", "\033[2m")
|
||||
red, yellow, green, reset, dim = ansi if color and overall in _OVERALL_GLYPH else ("",) * 5
|
||||
col_for = {"failed": red, "degraded": yellow, "ok": green}.get(overall, "")
|
||||
status_cols = {"pass": green, "fail": red, "skip": dim}
|
||||
|
||||
print(f"{_OVERALL_GLYPH.get(overall, '•')} cua-driver {header_v} on {platform} — {col_for}{overall}{reset}")
|
||||
lines = [f"{_OVERALL_GLYPH.get(overall, '•')} cua-driver {header_v} on {platform} — {col_for}{overall}{reset}"]
|
||||
if identity.get("resolved_binary"):
|
||||
print(f" {dim}binary: {identity['resolved_binary']}{reset}")
|
||||
lines.append(f" {dim}binary: {identity['resolved_binary']}{reset}")
|
||||
if cli_v and report_v and str(report_v) not in str(cli_v) and str(cli_v) not in str(report_v):
|
||||
# Only annotate when the free-form strings clearly differ.
|
||||
print(f" {dim}--version: {cli_v}{reset}")
|
||||
print(f" {dim}health_report.driver_version: {report_v}{reset}")
|
||||
if mismatch:
|
||||
print(f" {yellow}⚠️ version mismatch: health_report says {report_v!r} but binary --version is {cli_v!r}{reset}")
|
||||
print(f" {dim}→ trust --version / packages/current for debugging; health_report's binary_version check can lag on Windows{reset}")
|
||||
|
||||
lines += [f" {dim}--version: {cli_v}{reset}", f" {dim}health_report.driver_version: {report_v}{reset}"]
|
||||
if identity.get("version_mismatch"):
|
||||
lines += [f" {yellow}⚠️ version mismatch: health_report says {report_v!r} but binary --version is {cli_v!r}{reset}",
|
||||
f" {dim}→ trust --version / packages/current for debugging; health_report's binary_version check can lag on Windows{reset}"]
|
||||
for check in report.get("checks", []):
|
||||
name = check.get("name", "?")
|
||||
status = check.get("status", "?")
|
||||
glyph = _STATUS_GLYPH.get(status, "•")
|
||||
message = check.get("message") or ""
|
||||
print(f" {glyph} {status_cols.get(status, '')}{name}{reset}: {message}")
|
||||
hint = check.get("hint")
|
||||
if hint:
|
||||
print(f" → {dim}{hint}{reset}")
|
||||
# `data` is the structured payload some checks attach (bundle id, AX
|
||||
# state, version triple) — users / support staff frequently need it.
|
||||
data = check.get("data")
|
||||
if isinstance(data, dict) and data:
|
||||
for key, value in data.items():
|
||||
rendered = json.dumps(value) if isinstance(value, (dict, list)) else value
|
||||
print(f" {dim}{key}={rendered}{reset}")
|
||||
|
||||
lines += _check_lines(check, status_cols, reset, dim)
|
||||
print("\n".join(lines))
|
||||
|
||||
def run_doctor(driver_cmd: Optional[str] = None, *, include: Sequence[str] = (), skip: Sequence[str] = (),
|
||||
json_output: bool = False, color: Optional[bool] = None) -> int:
|
||||
"""Resolve the cua-driver binary, call `health_report`, render the result.
|
||||
|
||||
Honors `HERMES_CUA_DRIVER_CMD` via the shared runtime resolver, so doctor
|
||||
diagnoses what `computer_use` will actually invoke. On 0.10.x (health_report
|
||||
denied) it synthesizes a report from check_permissions / list_apps / CLI probes.
|
||||
Honors `HERMES_CUA_DRIVER_CMD` via the shared runtime resolver, so doctor diagnoses
|
||||
what `computer_use` will actually invoke. On 0.10.x (health_report denied) it
|
||||
synthesizes a report from check_permissions / list_apps / CLI probes.
|
||||
"""
|
||||
# Windows' locale codec (cp1252, cp936, ...) cannot encode the ✅ ❌ ⚠️ ⏭️ glyphs — force UTF-8.
|
||||
for stream in (sys.stdout, sys.stderr):
|
||||
try:
|
||||
with suppress(AttributeError, OSError):
|
||||
stream.reconfigure(encoding="utf-8", errors="replace") # type: ignore[union-attr]
|
||||
except (AttributeError, OSError):
|
||||
pass
|
||||
from tools.computer_use.cua_backend import resolve_cua_driver_cmd
|
||||
|
||||
binary = resolve_cua_driver_cmd(driver_cmd)
|
||||
if not binary:
|
||||
looked_for = driver_cmd or "cua-driver (PATH and canonical install paths)"
|
||||
print(f"cua-driver: not installed (looked for {looked_for!r}).")
|
||||
print(f"cua-driver: not installed (looked for {driver_cmd or 'cua-driver (PATH and canonical install paths)'!r}).")
|
||||
print(" Run: hermes computer-use install")
|
||||
return 2
|
||||
|
||||
try:
|
||||
report = _drive_health_report_or_fallback(binary, include=include, skip=skip)
|
||||
try: # prefer real health_report; on denial/non-schema, synthesize via probes
|
||||
try:
|
||||
report = _drive_health_report(binary, include=include, skip=skip, timeout=12.0)
|
||||
except HealthReportUnavailable as e:
|
||||
report = _compose_fallback_report(binary, reason=str(e), timeout=12.0)
|
||||
except RuntimeError as e:
|
||||
print(f"cua-driver health_report failed: {e}", file=sys.stderr)
|
||||
return 2
|
||||
|
||||
report = _apply_display_count_guard(report)
|
||||
identity = _build_identity(binary, report)
|
||||
|
||||
if json_output:
|
||||
# Additive envelope: upstream health_report keys preserved, Hermes identity
|
||||
# under hermes_identity so parsers that only read overall/checks keep working.
|
||||
payload = dict(report)
|
||||
payload["hermes_identity"] = identity
|
||||
json.dump(payload, sys.stdout, indent=2, sort_keys=True)
|
||||
json.dump({**report, "hermes_identity": identity}, sys.stdout, indent=2, sort_keys=True)
|
||||
sys.stdout.write("\n")
|
||||
else:
|
||||
if color is None:
|
||||
color = sys.stdout.isatty()
|
||||
_print_text_report(report, color=bool(color), identity=identity)
|
||||
|
||||
# Unknown / missing overall after fallback must not look like success.
|
||||
return 0 if report.get("overall") == "ok" else 1
|
||||
_print_text_report(report, color=sys.stdout.isatty() if color is None else bool(color), identity=identity)
|
||||
return 0 if report.get("overall") == "ok" else 1 # unknown/missing overall must not look like success
|
||||
|
||||
@@ -1,18 +1,12 @@
|
||||
"""
|
||||
Cross-platform Computer Use readiness + macOS permission helpers.
|
||||
"""Cross-platform Computer Use readiness + macOS permission helpers.
|
||||
|
||||
"Ready to drive" differs per platform:
|
||||
* macOS — explicit TCC grants (Accessibility + Screen Recording), reported /
|
||||
requested via cua-driver ``permissions status`` / ``permissions grant``. The
|
||||
grants attach to cua-driver's OWN identity (``com.trycua.driver``), not
|
||||
Hermes, so ``grant`` launches CuaDriver via LaunchServices for correct
|
||||
dialog attribution.
|
||||
* Windows / Linux — no TCC toggles; readiness == driver health.
|
||||
|
||||
``cua-driver doctor --json`` is the universal signal; ``computer_use_status``
|
||||
folds it with the macOS permission detail into one payload for the desktop
|
||||
card, the ``hermes computer-use permissions`` CLI, and
|
||||
``/api/tools/computer-use/status``.
|
||||
"Ready to drive" differs per platform: macOS needs explicit TCC grants (Accessibility +
|
||||
Screen Recording) via cua-driver ``permissions status`` / ``permissions grant``;
|
||||
Windows/Linux have no TCC toggles, so readiness == driver health. The grants attach to
|
||||
cua-driver's OWN identity (``com.trycua.driver``), not Hermes, so ``grant`` launches
|
||||
CuaDriver via LaunchServices for correct dialog attribution. ``cua-driver doctor --json``
|
||||
is the universal signal; ``computer_use_status`` folds it with the macOS detail into one
|
||||
payload for the desktop card, the ``permissions`` CLI and ``/api/tools/computer-use/status``.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@@ -21,7 +15,7 @@ import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
from typing import Any, Dict, List, Optional
|
||||
from typing import Any, Dict, Optional
|
||||
|
||||
from hermes_cli._subprocess_compat import windows_hide_flags
|
||||
|
||||
@@ -33,55 +27,39 @@ _BOOLS = ("accessibility", "screen_recording", "screen_recording_capturable")
|
||||
def _resolve_driver_cmd(override: Optional[str]) -> Optional[str]:
|
||||
"""Use the runtime resolver for UI status and permission commands too."""
|
||||
from tools.computer_use.cua_backend import resolve_cua_driver_cmd
|
||||
|
||||
return resolve_cua_driver_cmd(override)
|
||||
|
||||
|
||||
def _child_env() -> Dict[str, str]:
|
||||
"""cua-driver child env (telemetry policy + provider secrets stripped);
|
||||
degrades to ``os.environ`` on import error so probes never break."""
|
||||
try:
|
||||
from tools.computer_use.cua_backend import sanitized_cua_driver_env
|
||||
|
||||
return sanitized_cua_driver_env()
|
||||
except Exception:
|
||||
return dict(os.environ)
|
||||
|
||||
|
||||
def _run(binary: str, *args: str, timeout: float) -> subprocess.CompletedProcess:
|
||||
return subprocess.run(
|
||||
[binary, *args],
|
||||
capture_output=True,
|
||||
text=True, encoding='utf-8', errors='replace',
|
||||
timeout=timeout,
|
||||
env=_child_env(),
|
||||
stdin=subprocess.DEVNULL,
|
||||
creationflags=windows_hide_flags(),
|
||||
)
|
||||
|
||||
return subprocess.run([binary, *args], capture_output=True, text=True, encoding='utf-8',
|
||||
errors='replace', timeout=timeout, env=_child_env(),
|
||||
stdin=subprocess.DEVNULL, creationflags=windows_hide_flags())
|
||||
|
||||
def _json_out(binary: str, *args: str, timeout: float) -> Any:
|
||||
"""Run ``binary args`` and parse stdout as JSON (``None`` on empty output)."""
|
||||
raw = (_run(binary, *args, timeout=timeout).stdout or "").strip()
|
||||
return json.loads(raw) if raw else None
|
||||
|
||||
|
||||
def _doctor(binary: str) -> Optional[Dict[str, Any]]:
|
||||
"""``cua-driver doctor --json`` → ``{ok, checks:[{label,status,message}]}``."""
|
||||
try:
|
||||
data = _json_out(binary, "doctor", "--json", timeout=12)
|
||||
except Exception:
|
||||
return None
|
||||
data = None
|
||||
if not isinstance(data, dict):
|
||||
return None
|
||||
checks: List[Dict[str, str]] = [
|
||||
{k: str(p.get(k, "")) for k in ("label", "status", "message")}
|
||||
for p in data.get("probes", [])
|
||||
if isinstance(p, dict)
|
||||
]
|
||||
checks = [{k: str(p.get(k, "")) for k in ("label", "status", "message")}
|
||||
for p in data.get("probes", []) if isinstance(p, dict)]
|
||||
return {"ok": bool(data.get("ok")), "checks": checks}
|
||||
|
||||
|
||||
def _mac_permissions(binary: str, out: Dict[str, Any]) -> None:
|
||||
"""Fold ``cua-driver permissions status --json`` booleans into ``out``."""
|
||||
try:
|
||||
@@ -115,16 +93,13 @@ def computer_use_status(driver_cmd: Optional[str] = None) -> Dict[str, Any]:
|
||||
}
|
||||
if not binary:
|
||||
return out
|
||||
|
||||
try:
|
||||
out["version"] = (_run(binary, "--version", timeout=5).stdout or "").strip() or None
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
doctor = _doctor(binary)
|
||||
if doctor is not None:
|
||||
out["checks"] = doctor["checks"]
|
||||
|
||||
if plat == "darwin":
|
||||
_mac_permissions(binary, out)
|
||||
if out["error"] is None:
|
||||
@@ -143,17 +118,13 @@ def request_permissions_grant(driver_cmd: Optional[str] = None) -> int:
|
||||
if sys.platform != "darwin":
|
||||
print("Computer Use permissions are a macOS concept; nothing to grant here.")
|
||||
return 64
|
||||
|
||||
binary = _resolve_driver_cmd(driver_cmd)
|
||||
if not binary:
|
||||
print("cua-driver: not installed. Run: hermes computer-use install")
|
||||
return 2
|
||||
|
||||
print(
|
||||
"Requesting Accessibility + Screen Recording for CuaDriver.\n"
|
||||
"macOS will show a dialog attributed to CuaDriver (com.trycua.driver) — "
|
||||
"approve it, then return here."
|
||||
)
|
||||
print("Requesting Accessibility + Screen Recording for CuaDriver.\n"
|
||||
"macOS will show a dialog attributed to CuaDriver (com.trycua.driver) — "
|
||||
"approve it, then return here.")
|
||||
try:
|
||||
return int(subprocess.run([binary, "permissions", "grant"], env=_child_env(),
|
||||
stdin=subprocess.DEVNULL).returncode)
|
||||
|
||||
@@ -1,26 +1,24 @@
|
||||
"""Vision-routing decisions for ``computer_use`` capture results.
|
||||
|
||||
``computer_use(action='capture', mode='som'|'vision')`` returns a ``_multimodal``
|
||||
envelope with the screenshot, delivered to the active session model as the tool
|
||||
result. A text-only main model, or a provider that rejects multimodal content in
|
||||
tool results, turns that into a hard 400/404 tool failure — even when a working
|
||||
``auxiliary.vision`` model sits in config. This module decides: return the
|
||||
screenshot as multimodal content, or pre-analyse it via aux vision so the main
|
||||
model only ever sees text?
|
||||
``capture`` (mode som|vision) returns a ``_multimodal`` screenshot envelope as the
|
||||
tool result. A text-only main model, or a provider that rejects multimodal tool
|
||||
results, turns that into a hard 400/404 — even with a working ``auxiliary.vision``
|
||||
model in config. This module decides: multimodal envelope, or pre-analyse via aux
|
||||
vision so the main model only ever sees text?
|
||||
|
||||
Decision order (mirrors ``vision_analyze``):
|
||||
1. ``auxiliary.vision`` explicitly configured (provider not ""/"auto", or model /
|
||||
base_url set) → aux routing; users who pay for a vision model want it used.
|
||||
2. User declared ``supports_vision`` for the active route (config escape hatch
|
||||
for custom/local VLMs absent from models.dev) → honour it (True → multimodal).
|
||||
2. User-declared ``supports_vision`` for the active route (escape hatch for
|
||||
custom/local VLMs absent from models.dev) → honour it (True → multimodal).
|
||||
3. Provider+model carries images inside tool-result messages AND models.dev says
|
||||
``supports_vision=True`` → multimodal.
|
||||
4. Everything else (non-vision model, provider rejecting multimodal tool results,
|
||||
lookup failure) → aux routing.
|
||||
|
||||
The decision fails *closed* toward aux routing when metadata is missing or
|
||||
ambiguous: a screenshot sent to a model that cannot read it is a hard failure,
|
||||
while aux routing costs one extra LLM call and yields a usable description.
|
||||
Fails *closed* toward aux routing when metadata is missing or ambiguous: a
|
||||
screenshot sent to a model that cannot read it is a hard failure, while aux
|
||||
routing costs one extra LLM call and yields a usable description.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@@ -30,17 +28,11 @@ from typing import Any, Dict, Optional
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def _explicit_aux_vision_override(cfg: Optional[Dict[str, Any]]) -> bool:
|
||||
"""True when ``auxiliary.vision`` carries a non-default user override.
|
||||
|
||||
Mirrors ``agent.image_routing._explicit_aux_vision_override`` so the capture
|
||||
path and the user-attached-image path agree. ``provider: "auto"``, blank
|
||||
values, or a missing block all count as *not* explicit.
|
||||
"""
|
||||
if not isinstance(cfg, dict):
|
||||
return False
|
||||
aux = cfg.get("auxiliary") or {}
|
||||
"""True when ``auxiliary.vision`` carries a non-default user override. Mirrors
|
||||
``agent.image_routing._explicit_aux_vision_override`` so the capture path and the
|
||||
user-attached-image path agree; ``provider: "auto"``, blanks or a missing block are *not* explicit."""
|
||||
aux = cfg.get("auxiliary") if isinstance(cfg, dict) else None
|
||||
vision = aux.get("vision") if isinstance(aux, dict) else None
|
||||
if not isinstance(vision, dict):
|
||||
return False
|
||||
@@ -49,33 +41,19 @@ def _explicit_aux_vision_override(cfg: Optional[Dict[str, Any]]) -> bool:
|
||||
base_url = str(vision.get("base_url") or "").strip()
|
||||
return not (provider in ("", "auto") and not model and not base_url)
|
||||
|
||||
|
||||
def _lookup_user_declared_supports_vision(
|
||||
provider: str,
|
||||
model: str,
|
||||
cfg: Optional[Dict[str, Any]],
|
||||
) -> Optional[bool]:
|
||||
"""Return config-declared ``supports_vision`` for the active route (None on failure)."""
|
||||
def _lookup_user_declared_supports_vision(provider: str, model: str, cfg: Optional[Dict[str, Any]]) -> Optional[bool]:
|
||||
"""Config-declared ``supports_vision`` for the active route (None on failure)."""
|
||||
try:
|
||||
from agent.image_routing import _supports_vision_override
|
||||
|
||||
return _supports_vision_override(cfg, provider, model)
|
||||
except Exception as exc: # pragma: no cover - defensive
|
||||
logger.debug("computer_use vision_routing: config override lookup failed: %s", exc)
|
||||
return None
|
||||
|
||||
|
||||
def _lookup_supports_vision(
|
||||
provider: str,
|
||||
model: str,
|
||||
cfg: Optional[Dict[str, Any]] = None,
|
||||
) -> Optional[bool]:
|
||||
"""Return config/models.dev ``supports_vision`` for *(provider, model)*.
|
||||
|
||||
Prefers ``agent.image_routing._lookup_supports_vision``; falls back to raw
|
||||
models.dev capabilities only when that import is unavailable. Any lookup
|
||||
error yields None (caller fails closed toward aux routing).
|
||||
"""
|
||||
def _lookup_supports_vision(provider: str, model: str, cfg: Optional[Dict[str, Any]] = None) -> Optional[bool]:
|
||||
"""Config/models.dev ``supports_vision`` for *(provider, model)*. Prefers
|
||||
``agent.image_routing._lookup_supports_vision``; falls back to raw models.dev capabilities
|
||||
only when that import is unavailable. Any lookup error → None (caller fails closed to aux)."""
|
||||
if not provider or not model:
|
||||
return None
|
||||
try:
|
||||
@@ -86,24 +64,16 @@ def _lookup_supports_vision(
|
||||
if _lookup_image_supports is not None:
|
||||
return _lookup_image_supports(provider, model, cfg)
|
||||
from agent.models_dev import get_model_capabilities
|
||||
|
||||
caps = get_model_capabilities(provider, model)
|
||||
except Exception as exc: # pragma: no cover - defensive
|
||||
logger.debug(
|
||||
"computer_use vision_routing: caps lookup failed for %s:%s — %s",
|
||||
provider, model, exc,
|
||||
)
|
||||
logger.debug("computer_use vision_routing: caps lookup failed for %s:%s — %s", provider, model, exc)
|
||||
return None
|
||||
return None if caps is None else bool(getattr(caps, "supports_vision", False))
|
||||
|
||||
|
||||
def _provider_accepts_multimodal_tool_result(provider: str, model: str) -> Optional[bool]:
|
||||
"""Whether *provider*+*model* carries images inside tool-result messages.
|
||||
|
||||
Reuses ``tools.vision_tools._supports_media_in_tool_results`` so this stays
|
||||
in lockstep with the ``vision_analyze`` native fast path. Returns None on
|
||||
import failure so callers fall back to aux routing rather than guessing.
|
||||
"""
|
||||
"""Whether *provider*+*model* carries images inside tool-result messages. Reuses
|
||||
``tools.vision_tools._supports_media_in_tool_results`` to stay in lockstep with the
|
||||
``vision_analyze`` fast path; None on import failure so callers fall back to aux, not guess."""
|
||||
if not provider:
|
||||
return None
|
||||
try:
|
||||
@@ -113,33 +83,19 @@ def _provider_accepts_multimodal_tool_result(provider: str, model: str) -> Optio
|
||||
return None
|
||||
return bool(_supports_media_in_tool_results(provider, model))
|
||||
|
||||
|
||||
def should_route_capture_to_aux_vision(
|
||||
provider: str,
|
||||
model: str,
|
||||
cfg: Optional[Dict[str, Any]],
|
||||
) -> bool:
|
||||
"""True iff the captured screenshot should be pre-analysed via aux vision.
|
||||
|
||||
*provider* is the lower-case canonical id, *model* the slug as sent to the
|
||||
provider, *cfg* the loaded ``config.yaml`` dict (or None). False means keep
|
||||
the multimodal envelope (main model handles vision natively).
|
||||
"""
|
||||
def should_route_capture_to_aux_vision(provider: str, model: str, cfg: Optional[Dict[str, Any]]) -> bool:
|
||||
"""True iff the screenshot should be pre-analysed via aux vision; False keeps the
|
||||
multimodal envelope. *provider* is the lower-case canonical id, *model* the slug as
|
||||
sent to the provider, *cfg* the loaded ``config.yaml`` dict (or None)."""
|
||||
if _explicit_aux_vision_override(cfg):
|
||||
return True
|
||||
|
||||
user_declared = _lookup_user_declared_supports_vision(provider, model, cfg)
|
||||
if user_declared is True:
|
||||
return False
|
||||
if user_declared is False:
|
||||
return True
|
||||
|
||||
if not _provider_accepts_multimodal_tool_result(provider, model):
|
||||
return True
|
||||
|
||||
return _lookup_supports_vision(provider, model, cfg) is not True
|
||||
|
||||
|
||||
__all__ = [
|
||||
"should_route_capture_to_aux_vision",
|
||||
]
|
||||
__all__ = ["should_route_capture_to_aux_vision"]
|
||||
|
||||
Reference in New Issue
Block a user