Initial upload: Hermes 2 Agent (part 10)
be22d4d verified 2 months ago
57.1 kB
"""Entry point for the `computer_use` tool.

Universal (any-model) desktop control across macOS, Windows, and Linux via
cua-driver's background computer-use primitive. Replaces #4562's
Anthropic-native `computer_20251124` approach — the schema here is standard
OpenAI function-calling so every tool-capable model can drive it.

Linux is the most recent runtime (X11 + Wayland, via cua-driver-rs's
AT-SPI tree path); it is enabled here alongside macOS and Windows. When a
host's display server or accessibility stack isn't reachable, cua-driver's
`health_report` (surfaced by `hermes computer-use doctor`) reports the
exact blocked check rather than the toolset silently failing.

Return contract
---------------
For text-only results (wait, key, list_apps, focus_app, failures, etc.):
  JSON string.

For captures / actions with `capture_after=True`:
  A dict wrapped as the OpenAI-style multi-part tool-message content:

      {
        "_multimodal": True,
        "content": [
            {"type": "text", "text": "<human-readable summary + SOM index>"},
            {"type": "image_url",
             "image_url": {"url": "data:image/png;base64,<b64>"}},
        ],
        "text_summary": "<text used for fallback string content>",
      }

  run_agent.py's tool-message builder inspects `_multimodal` and emits a
  list-shaped `content` for OpenAI-compatible providers. The Anthropic
  adapter splices the base64 image into a `tool_result` block (see
  `agent/anthropic_adapter.py`). Every provider that supports multi-part
  tool content gets the image; text-only providers see the summary only.
"""

from __future__ import annotations

import atexit
import base64
import json
import logging
import os
import re
import struct
import sys
import threading
from typing import Any, Dict, List, Optional, Tuple

from tools.computer_use.backend import (
    ActionResult,
    CaptureResult,
    ComputerUseBackend,
    UIElement,
)

logger = logging.getLogger(__name__)


# ---------------------------------------------------------------------------
# Approval & safety
# ---------------------------------------------------------------------------

_approval_callback = None


def set_approval_callback(cb) -> None:
    """Register a callback for computer_use approval prompts (used by CLI).

    Matches the terminal_tool._approval_callback pattern. The callback
    receives (action, args, summary) and returns one of:
      "approve_once" | "approve_session" | "always_approve" | "deny".
    """
    global _approval_callback
    _approval_callback = cb


# Actions that read, not mutate. Always allowed.
_SAFE_ACTIONS = frozenset({
    "capture", "wait", "list_apps", "list_windows", "cua_browser_state",
})

# Actions that mutate user-visible state. Go through approval.
_DESTRUCTIVE_ACTIONS = frozenset({
    "click", "double_click", "right_click", "middle_click",
    "drag", "scroll", "type", "key", "set_value", "focus_app",
    "cua_browser_prepare", "cua_browser_navigate", "cua_browser_click",
    "cua_browser_type", "cua_browser_pointer", "cua_browser_dialog",
    "cua_browser_set_input_files", "cua_browser_download",
})

# Hard-blocked key combinations. Mirrored from #4562 — these are destructive
# regardless of approval level (e.g. logout kills the session Hermes runs in).
_BLOCKED_KEY_COMBOS = {
    frozenset({"cmd", "shift", "backspace"}),   # empty trash
    frozenset({"cmd", "option", "backspace"}),   # force delete
    frozenset({"cmd", "ctrl", "q"}),             # lock screen
    frozenset({"cmd", "shift", "q"}),            # log out
    frozenset({"cmd", "option", "shift", "q"}),  # force log out
    # Windows secure/session shortcuts. The Windows driver accepts Win-key
    # combos, and Alt is canonicalized to option below, so block the
    # destructive variants before any backend sees them.
    frozenset({"win", "l"}),
    frozenset({"ctrl", "option", "delete"}),
    frozenset({"ctrl", "option", "del"}),
    frozenset({"option", "f4"}),
}

_KEY_ALIASES = {
    "command": "cmd", "control": "ctrl", "alt": "option", "⌘": "cmd", "⌥": "option",
    "windows": "win", "super": "win", "meta": "win",
}


def _canon_key_combo(keys: str) -> frozenset:
    # Split on both "+" and "-": the cua-driver backend's _parse_key_combo
    # accepts hyphen-separated combos too, so "ctrl-alt-delete" executes as
    # the real destructive shortcut. Mirror its separators here, otherwise the
    # _BLOCKED_KEY_COMBOS gate is trivially bypassed with hyphen notation.
    parts = [p.strip().lower() for p in re.split(r"\s*[+\-]\s*", keys) if p.strip()]
    parts = [_KEY_ALIASES.get(p, p) for p in parts]
    return frozenset(parts)


# Dangerous text patterns for the `type` action. Same list as #4562.
_BLOCKED_TYPE_PATTERNS = [
    re.compile(r"curl\s+[^|]*\|\s*bash", re.IGNORECASE),
    re.compile(r"curl\s+[^|]*\|\s*sh", re.IGNORECASE),
    re.compile(r"wget\s+[^|]*\|\s*bash", re.IGNORECASE),
    re.compile(r"\bsudo\s+rm\s+-[rf]", re.IGNORECASE),
    re.compile(r"\brm\s+-rf\s+/\s*$", re.IGNORECASE),
    re.compile(r":\s*\(\)\s*\{\s*:\|:\s*&\s*\}", re.IGNORECASE),  # fork bomb
]


def _is_blocked_type(text: str) -> Optional[str]:
    for pat in _BLOCKED_TYPE_PATTERNS:
        if pat.search(text):
            return pat.pattern
    return None


# ---------------------------------------------------------------------------
# Backend selection — env-swappable for tests
# ---------------------------------------------------------------------------

# Per-Hermes-session cached backends. Each backend owns its own cua-driver
# session, native target, typed-browser binding, refs, and grant namespace.
_backend_lock = threading.Lock()
# Backward-compatible empty-session injection hook used by older tests.
# Process-scoped aux-vision routing cache: (provider, model) → bool.
_AUX_VISION_ROUTE_CACHE: Dict[Tuple[str, str], bool] = {}
_backend: Optional[ComputerUseBackend] = None
_backends: Dict[str, ComputerUseBackend] = {}
_backend_call_locks: Dict[str, threading.RLock] = {}
_backend_permission_modes: Dict[str, str] = {}
# Approval state, scoped per conversation/run (keyed by session_id) so a
# gateway serving concurrent sessions can't leak one run's "always approve"
# unlock into another. Falls back to a shared "" bucket for callers that
# don't pass a session_id (e.g. the classic single-run CLI). Values:
#   _session_auto_approve[sid] -> bool   ("always_approve everything")
#   _always_allow[sid]         -> set of (action, delivery_mode) scope keys
# See NousResearch/hermes-agent#67052 gap 4.
_approval_lock = threading.Lock()
_session_auto_approve: Dict[str, bool] = {}
_always_allow: Dict[str, set] = {}


def _cua_permission_mode(session_id: str) -> str:
    """Map Hermes's explicit approval bypass onto Cua's immutable mode.

    Hermes has TWO session-identity namespaces: the tool-dispatch path passes
    the DB ``session_id`` (``agent.session_id``), while gateway ``/yolo``
    keys approval state off the gateway ``session_key`` (set per turn via the
    ``set_current_session_key`` contextvar in tools/approval.py). CLI and TUI
    use the DB id for both. Checking ONLY ``session_id`` here would make a
    gateway ``/yolo`` toggle silently invisible to computer_use (works in
    CLI, dead on messaging platforms), so we consult both namespaces —
    bypass in either means the user explicitly opted out of approvals for
    this run. Fails closed on any resolution error.
    """
    try:
        from tools.approval import (
            get_current_session_key,
            is_approval_bypass_active_for_session,
        )

        if is_approval_bypass_active_for_session(session_id):
            return "unrestricted"
        current_key = get_current_session_key(default="")
        if current_key and is_approval_bypass_active_for_session(current_key):
            return "unrestricted"
    except Exception:
        # Approval state must fail closed if it cannot be resolved.
        pass
    return "standard"


def _get_backend(session_id: str = "") -> ComputerUseBackend:
    global _backend
    sid = str(session_id or "")
    while True:
        stale_backend: Optional[ComputerUseBackend] = None
        stale_lock: Optional[threading.RLock] = None
        with _backend_lock:
            # Resolve the mode while holding the cache lock. Session YOLO
            # mutation never holds the approval lock while releasing this
            # cache, so the lock order cannot cycle.
            permission_mode = _cua_permission_mode(sid)
            if sid == "" and _backend is not None and sid not in _backends:
                # Preserve the long-standing empty-session injection hook used
                # by integrations and tests while normalizing it into the
                # session-owned cache/lifecycle path.
                _backends[sid] = _backend
                _backend_call_locks[sid] = threading.RLock()
                _backend_permission_modes[sid] = permission_mode
            cached = _backends.get(sid)
            if cached is not None:
                if _backend_permission_modes.get(sid, "standard") == permission_mode:
                    return cached
                # Cua's permission mode cannot change after daemon startup. A
                # /yolo toggle replaces only this session's backend.
                stale_backend = _backends.pop(sid)
                stale_lock = _backend_call_locks.pop(sid, None)
                _backend_permission_modes.pop(sid, None)
                if sid == "":
                    _backend = None
            else:
                backend_name = os.environ.get(
                    "HERMES_COMPUTER_USE_BACKEND", "cua"
                ).lower()
                if backend_name in {"cua", "cua-driver", ""}:
                    from tools.computer_use.cua_backend import CuaDriverBackend

                    backend = CuaDriverBackend(permission_mode=permission_mode)
                elif backend_name == "noop":  # pragma: no cover
                    backend = _NoopBackend()
                else:
                    raise RuntimeError(
                        f"Unknown HERMES_COMPUTER_USE_BACKEND={backend_name!r}"
                    )
                # Starting under the cache lock preserves the existing
                # one-backend-per-session invariant. A concurrent mode toggle
                # releases this backend before returning to its caller.
                backend.start()
                _backends[sid] = backend
                _backend_call_locks[sid] = threading.RLock()
                _backend_permission_modes[sid] = permission_mode
                if sid == "":
                    _backend = backend
                return backend

        # Stop a mismatched backend outside the global cache lock. Another
        # session can continue creating or releasing its own backend, and the
        # loop re-reads the authoritative mode before installing a replacement.
        try:
            if stale_lock is not None:
                with stale_lock:
                    stale_backend.stop()
            elif stale_backend is not None:
                stale_backend.stop()
        except Exception:
            pass


def release_computer_use_session(session_id: str) -> bool:
    """Release one session-owned computer-use backend.

    This is the production lifecycle seam for hosts and policy plugins. It
    removes the exact session backend, its call lock, and its recorded
    permission mode before stopping the backend, so new lookups cannot retain
    the stale target/ref namespace — and stops a private embedded daemon when
    Hermes YOLO selected unrestricted mode. Approval state is cleared even
    when no backend was started.

    Returns ``True`` when a backend was found and released, ``False`` when the
    session was already absent. Safe to call repeatedly.
    """
    global _backend
    sid = str(session_id or "")
    with _backend_lock:
        backend = _backends.pop(sid, None)
        call_lock = _backend_call_locks.pop(sid, None)
        _backend_permission_modes.pop(sid, None)
        # Preserve the backward-compatible empty-session injection hook:
        # older callers/tests may populate only `_backend`.
        if sid == "" and backend is None:
            backend = _backend
        if sid == "" and _backend is backend:
            _backend = None

    with _approval_lock:
        _session_auto_approve.pop(sid, None)
        _always_allow.pop(sid, None)

    if backend is None:
        return False
    try:
        # Let an in-flight action finish before ending the driver session and
        # dropping its target/ref state. Do not hold the global cache lock
        # while waiting: unrelated Hermes sessions remain independent.
        if call_lock is not None:
            with call_lock:
                backend.stop()
        else:
            backend.stop()
    except Exception:
        logger.debug(
            "computer_use backend release failed for session %s",
            sid,
            exc_info=True,
        )
    return True


def _shutdown_backend_atexit() -> None:
    """Stop all cached backends so cua-driver children don't outlive us.

    Each session backend holds a long-lived ``cua-driver`` subprocess, so
    without this a driver can survive the Hermes process that spawned it
    (#28152 item 3). #69903 kept the orphan from burning a core by disabling
    the cursor overlay; the process itself still lingered.

    Mirrors ``browser_tool``'s ``atexit.register(_emergency_cleanup_all_sessions)``
    — same spawn-and-drive-a-subprocess shape. atexit only, no signal handlers:
    a ``SystemExit`` raised from a prompt_toolkit key binding corrupts its
    coroutine state and makes the process unkillable. Never raises, since an
    exception escaping atexit prints a traceback on every exit.
    """
    global _backend
    # Drop the global lock before stop() — teardown budgets 5s and shouldn't
    # block an unrelated caller waiting to spawn.
    with _backend_lock:
        unique = {
            id(backend): (backend, _backend_call_locks.get(sid))
            for sid, backend in _backends.items()
        }
        if _backend is not None:
            unique.setdefault(
                id(_backend),
                (_backend, _backend_call_locks.get("")),
            )
        _backend = None
        _backends.clear()
        _backend_call_locks.clear()
        _backend_permission_modes.clear()

    with _approval_lock:
        _session_auto_approve.clear()
        _always_allow.clear()

    for backend, call_lock in unique.values():
        try:
            if call_lock is not None:
                with call_lock:
                    backend.stop()
            else:
                backend.stop()
        except Exception as e:
            logger.debug("cua-driver atexit teardown failed: %s", e)


atexit.register(_shutdown_backend_atexit)


def reset_backend_for_tests() -> None:  # pragma: no cover
    """Test helper — tear down the cached backend and per-session state."""
    _shutdown_backend_atexit()
    _AUX_VISION_ROUTE_CACHE.clear()


class _NoopBackend(ComputerUseBackend):  # pragma: no cover
    """Test/CI stub. Records calls; returns trivial results."""

    def __init__(self) -> None:
        self.calls: List[Tuple[str, Dict[str, Any]]] = []
        self._started = False

    def start(self) -> None: self._started = True
    def stop(self) -> None: self._started = False
    def is_available(self) -> bool: return True

    def capture(
        self,
        mode: str = "som",
        app: Optional[str] = None,
        pid: Optional[int] = None,
        window_id: Optional[int] = None,
    ) -> CaptureResult:
        self.calls.append((
            "capture",
            {"mode": mode, "app": app, "pid": pid, "window_id": window_id},
        ))
        return CaptureResult(mode=mode, width=1024, height=768, png_b64=None,
                             elements=[], app=app or "", window_title="")

    def click(self, **kw) -> ActionResult:
        self.calls.append(("click", kw))
        return ActionResult(ok=True, action="click")

    def drag(self, **kw) -> ActionResult:
        self.calls.append(("drag", kw))
        return ActionResult(ok=True, action="drag")

    def scroll(self, **kw) -> ActionResult:
        self.calls.append(("scroll", kw))
        return ActionResult(ok=True, action="scroll")

    def type_text(self, text: str, **kw) -> ActionResult:
        self.calls.append(("type", {"text": text, **kw}))
        return ActionResult(ok=True, action="type")

    def key(self, keys: str, **kw) -> ActionResult:
        self.calls.append(("key", {"keys": keys, **kw}))
        return ActionResult(ok=True, action="key")

    def list_apps(self) -> List[Dict[str, Any]]:
        self.calls.append(("list_apps", {}))
        return []

    def list_windows(self) -> List[Dict[str, Any]]:
        self.calls.append(("list_windows", {}))
        return []

    def focus_app(self, app: str, raise_window: bool = False) -> ActionResult:
        self.calls.append(("focus_app", {"app": app, "raise": raise_window}))
        return ActionResult(ok=True, action="focus_app")

    def set_value(self, value: str, element: Optional[int] = None) -> ActionResult:
        self.calls.append(("set_value", {"value": value, "element": element}))
        return ActionResult(ok=True, action="set_value")


# ---------------------------------------------------------------------------
# Dispatch
# ---------------------------------------------------------------------------

def handle_computer_use(args: Dict[str, Any], **kwargs) -> Any:
    """Main entry point — dispatched by tools.registry.

    Returns either a JSON string (text-only) or a dict marked `_multimodal`
    (image + summary) which run_agent.py wraps into the tool message.
    """
    action = (args.get("action") or "").strip().lower()
    if not action:
        return json.dumps({"error": "missing `action`"})
    # Per-run key for approval-state and daemon-mode isolation across
    # concurrent sessions.
    session_id = str(kwargs.get("session_id") or "")

    # Safety: validate actions before approval prompt.
    if action in {"type", "cua_browser_type"}:
        text = args.get("text", "")
        pat = _is_blocked_type(text)
        if pat:
            return json.dumps({
                "error": f"blocked pattern in type text: {pat!r}",
                "hint": "Dangerous shell patterns cannot be typed via computer_use.",
            })

    if action == "key":
        keys = args.get("keys", "")
        combo = _canon_key_combo(keys)
        for blocked in _BLOCKED_KEY_COMBOS:
            if blocked.issubset(combo) and len(blocked) <= len(combo):
                return json.dumps({
                    "error": f"blocked key combo: {sorted(blocked)}",
                    "hint": "Destructive system shortcuts are hard-blocked.",
                })

    if args.get("bring_to_front") and args.get("delivery_mode") != "foreground":
        return json.dumps({
            "error": "bring_to_front requires delivery_mode='foreground'",
            "code": "bring_to_front_requires_foreground",
        })

    # Approval gate (destructive actions only).
    if action in _DESTRUCTIVE_ACTIONS:
        err = _request_approval(action, args, session_id)
        if err is not None:
            return err
    # Persistent focus is a separate, visible side effect from the input
    # itself. Keep its approval scope distinct even when the input rung has
    # already been approved for this session.
    if args.get("bring_to_front") or (
        action == "focus_app" and args.get("raise_window")
    ):
        err = _request_approval("bring_to_front", args, session_id)
        if err is not None:
            return err

    # Dispatch to backend.
    try:
        backend = _get_backend(session_id=session_id)
    except Exception as e:
        return json.dumps({
            "error": f"computer_use backend unavailable: {e}",
            "hint": "If the cua-driver binary is missing, run `hermes computer-use install`. "
                    "If a Python dependency is missing, the error above shows the exact install command.",
        })

    try:
        with _backend_lock:
            call_lock = _backend_call_locks.setdefault(session_id, threading.RLock())
        with call_lock:
            return _dispatch(backend, action, args)
    except Exception as e:
        logger.exception("computer_use %s failed", action)
        return json.dumps({"error": f"{action} failed: {e}"})


def _request_approval(action: str, args: Dict[str, Any],
                      session_id: str = "") -> Optional[str]:
    """Return None if approved, or a JSON error string if denied.

    Approval is scoped by (action, delivery_mode) AND by session_id.
    Foreground delivery is a visible focus change, so a prior background
    approval — even ``approve_session`` on the same action — must NOT
    silently authorize it (NousResearch/hermes-agent#67052).
    ``always_approve`` (the blanket "auto-approve everything" unlock) still
    covers foreground, since the user explicitly opted into unattended
    operation. State is keyed on session_id so concurrent runs don't leak
    unlocks into one another.
    """
    is_foreground = args.get("delivery_mode") == "foreground"
    scope_key = (action, "foreground" if is_foreground else "background")
    with _approval_lock:
        if _session_auto_approve.get(session_id):
            return None
        if scope_key in _always_allow.get(session_id, set()):
            return None
    cb = _approval_callback
    if cb is None:
        # No CLI approval wired — default allow. Gateway approval is handled
        # one layer out via the normal tool-approval infra.
        return None
    summary = _summarize_action(action, args)
    try:
        verdict = cb(action, args, summary)
    except Exception as e:
        logger.warning("approval callback failed: %s", e)
        verdict = "deny"
    if verdict == "approve_once":
        return None
    if verdict == "approve_session" or verdict == "always_approve":
        with _approval_lock:
            _always_allow.setdefault(session_id, set()).add(scope_key)
            if verdict == "always_approve":
                _session_auto_approve[session_id] = True
        return None
    return json.dumps({"error": "denied by user", "action": action})


def _summarize_action(action: str, args: Dict[str, Any]) -> str:
    fg = " [FOREGROUND — briefly raises the window / changes focus]" \
        if args.get("delivery_mode") == "foreground" else ""
    if action in {"click", "double_click", "right_click", "middle_click"}:
        if args.get("element") is not None:
            return f"{action} element #{args['element']}{fg}"
        coord = args.get("coordinate")
        if coord:
            return f"{action} at {tuple(coord)}{fg}"
        return action + fg
    if action == "drag":
        src = args.get("from_element") or args.get("from_coordinate")
        dst = args.get("to_element") or args.get("to_coordinate")
        return f"drag {src} → {dst}{fg}"
    if action == "scroll":
        return f"scroll {args.get('direction', '?')} x{args.get('amount', 3)}{fg}"
    if action == "type":
        text = args.get("text", "")
        return f"type {text[:60]!r}" + ("..." if len(text) > 60 else "") + fg
    if action == "key":
        return f"key {args.get('keys', '')!r}{fg}"
    if action == "focus_app":
        return f"focus {args.get('app', '')!r}" + (" (raise)" if args.get("raise_window") else "")
    return action + fg


def _dispatch(backend: ComputerUseBackend, action: str, args: Dict[str, Any]) -> Any:
    capture_after = bool(args.get("capture_after"))

    if action == "capture":
        mode = str(args.get("mode", "som"))
        if mode not in {"som", "vision", "ax"}:
            return json.dumps({"error": f"bad mode {mode!r}; use som|vision|ax"})
        capture_kwargs: Dict[str, Any] = {"mode": mode, "app": args.get("app")}
        if args.get("pid") is not None or args.get("window_id") is not None:
            capture_kwargs.update({
                "pid": args.get("pid"),
                "window_id": args.get("window_id"),
            })
        cap = backend.capture(**capture_kwargs)
        return _capture_response(cap, max_elements=_coerce_max_elements(args.get("max_elements")))

    if action == "wait":
        seconds = float(args.get("seconds", 1.0))
        res = backend.wait(seconds)
        return _text_response(res)

    if action == "list_apps":
        apps = backend.list_apps()
        return json.dumps({"apps": apps, "count": len(apps)})

    if action == "list_windows":
        windows = backend.list_windows()
        return json.dumps({"windows": windows, "count": len(windows)})

    if action == "focus_app":
        app = args.get("app")
        if not app:
            return json.dumps({"error": "focus_app requires `app`"})
        res = backend.focus_app(app, raise_window=bool(args.get("raise_window")))
        return _maybe_follow_capture(backend, res, capture_after)

    # cua-driver's typed browser surface is namespaced inside the existing
    # computer_use tool so it cannot collide with native browser/MCP tools.
    # The backend owns the opaque driver session, target, tab and ref state;
    # none of those capabilities can be supplied across Hermes sessions.
    if action == "cua_browser_state":
        state_args: Dict[str, Any] = {}
        for public, internal in (
            ("pid", "pid"),
            ("window_id", "window_id"),
            ("tab_id", "tab_id"),
            ("snapshot_format", "snapshot_format"),
            ("query", "query"),
            ("scope_ref", "scope_ref"),
            ("continuation", "continuation"),
        ):
            if args.get(public) is not None:
                state_args[internal] = args[public]
        return json.dumps(backend.typed_browser_state(**state_args))

    if action == "cua_browser_prepare":
        return json.dumps(backend.typed_browser_prepare(
            pid=args.get("pid"),
            window_id=args.get("window_id"),
            profile_mode=args.get("profile_mode", "isolated_new"),
            profile_name=args.get("profile_name"),
            allow_launch=bool(args.get("allow_launch")),
        ))

    browser_tools = {
        "cua_browser_navigate": "browser_navigate",
        "cua_browser_click": "browser_click",
        "cua_browser_type": "browser_type",
        "cua_browser_pointer": "browser_pointer",
        "cua_browser_dialog": "browser_dialog",
        "cua_browser_set_input_files": "browser_set_input_files",
        "cua_browser_download": "browser_download",
    }
    driver_tool = browser_tools.get(action)
    if driver_tool is not None:
        call_args: Dict[str, Any] = {}
        allowed_fields = {
            "browser_navigate": ("url",),
            "browser_click": ("ref", "input_route", "x", "y"),
            "browser_type": ("ref", "text"),
            "browser_pointer": (
                "ref", "destination_ref", "input_route", "x", "y",
                "to_x", "to_y", "delta_x", "delta_y",
            ),
            "browser_dialog": (
                "dialog_id", "prompt_text", "delivery_mode",
            ),
            "browser_set_input_files": ("ref", "files"),
            "browser_download": ("ref", "destination_root"),
        }
        for field in allowed_fields[driver_tool]:
            if args.get(field) is not None:
                call_args[field] = args[field]
        if (
            driver_tool in {"browser_click", "browser_pointer"}
            and args.get("coordinate") is not None
        ):
            coordinate = args["coordinate"]
            if isinstance(coordinate, (list, tuple)) and len(coordinate) == 2:
                call_args["x"], call_args["y"] = coordinate
        pointer_action = args.get("browser_pointer_action")
        dialog_action = args.get("browser_dialog_action")
        # Direct adapter callers may omit the public discriminator from args;
        # retain this narrow compatibility path without making it usable to
        # override the namespaced action selected by handle_computer_use.
        nested_action = args.get("action")
        if nested_action not in browser_tools:
            if driver_tool == "browser_pointer" and pointer_action is None:
                pointer_action = nested_action
            if driver_tool == "browser_dialog" and dialog_action is None:
                dialog_action = nested_action
        if pointer_action is not None:
            call_args["action"] = pointer_action
        if dialog_action is not None:
            call_args["action"] = dialog_action
        if args.get("browser_type_mode") is not None:
            call_args["mode"] = args["browser_type_mode"]
        return json.dumps(backend.typed_browser_action(
            driver_tool,
            tab_id=args.get("tab_id"),
            args=call_args,
        ))

    # delivery_mode / bring_to_front thread through every input action so the
    # model can escalate background → foreground per cua-driver's ladder.
    delivery_mode = args.get("delivery_mode")
    bring_to_front = bool(args.get("bring_to_front"))

    if action in {"click", "double_click", "right_click", "middle_click"}:
        button = args.get("button")
        click_count = 1
        if action == "double_click":
            click_count = 2
        elif action == "right_click":
            button = "right"
        elif action == "middle_click":
            button = "middle"
        else:
            button = button or "left"
        element = args.get("element")
        coord = args.get("coordinate") or (None, None)
        x, y = (coord[0], coord[1]) if coord and coord[0] is not None else (None, None)
        res = backend.click(
            element=element if element is not None else None,
            x=x, y=y, button=button or "left", click_count=click_count,
            modifiers=args.get("modifiers"),
            delivery_mode=delivery_mode, bring_to_front=bring_to_front,
        )
        return _maybe_follow_capture(backend, res, capture_after)

    if action == "drag":
        has_elements = args.get("from_element") is not None and args.get("to_element") is not None
        has_coords = args.get("from_coordinate") and args.get("to_coordinate")
        if not has_elements and not has_coords:
            return json.dumps({
                "error": "drag requires from_coordinate/to_coordinate or from_element/to_element",
            })
        res = backend.drag(
            from_element=args.get("from_element"),
            to_element=args.get("to_element"),
            from_xy=tuple(args["from_coordinate"]) if args.get("from_coordinate") else None,
            to_xy=tuple(args["to_coordinate"]) if args.get("to_coordinate") else None,
            button=args.get("button", "left"),
            modifiers=args.get("modifiers"),
            delivery_mode=delivery_mode, bring_to_front=bring_to_front,
        )
        return _maybe_follow_capture(backend, res, capture_after)

    if action == "scroll":
        coord = args.get("coordinate") or (None, None)
        res = backend.scroll(
            direction=args.get("direction", "down"),
            amount=int(args.get("amount", 3)),
            element=args.get("element"),
            x=coord[0] if coord and coord[0] is not None else None,
            y=coord[1] if coord and coord[1] is not None else None,
            modifiers=args.get("modifiers"),
            delivery_mode=delivery_mode, bring_to_front=bring_to_front,
        )
        return _maybe_follow_capture(backend, res, capture_after)

    if action == "type":
        res = backend.type_text(args.get("text", ""),
                                delivery_mode=delivery_mode, bring_to_front=bring_to_front)
        return _maybe_follow_capture(backend, res, capture_after)

    if action == "key":
        res = backend.key(args.get("keys", ""),
                          delivery_mode=delivery_mode, bring_to_front=bring_to_front)
        return _maybe_follow_capture(backend, res, capture_after)

    if action == "set_value":
        value = args.get("value")
        if value is None:
            return json.dumps({"error": "set_value requires `value`"})
        res = backend.set_value(value=str(value), element=args.get("element"))
        return _maybe_follow_capture(backend, res, capture_after)

    return json.dumps({"error": f"unknown action {action!r}"})


# ---------------------------------------------------------------------------
# Response shaping
# ---------------------------------------------------------------------------

def _classify_action_result(res: ActionResult) -> Dict[str, Any]:
    """Choose the next ladder step from semantic evidence, in precedence order.

    An escalation recommendation is advisory. It never overrides a confirmed
    effect and it never turns an unverifiable action into permission to repeat
    input. The model must first obtain fresh evidence.
    """
    if res.effect == "confirmed" or res.verified is True:
        return {"decision": "done"}
    if res.effect == "unverifiable":
        return {"decision": "verify_fresh_state"}
    if res.effect == "suspected_noop" or not res.ok or res.code is not None:
        decision: Dict[str, Any] = {"decision": "escalate"}
        if isinstance(res.escalation, dict):
            decision["recommended"] = res.escalation.get("recommended")
        return decision
    # Transport success without semantic proof is not proof of effect.
    return {"decision": "verify_fresh_state"}


def _action_payload(res: ActionResult) -> Dict[str, Any]:
    payload: Dict[str, Any] = {"ok": res.ok, "action": res.action}
    if res.message:
        payload["message"] = res.message
    # Surface cua-driver's structured verdict additively so the model can
    # follow the verify → escalate ladder. Only include fields the driver
    # actually returned (None = old driver / not carried). ok is transport
    # success; effect/escalation are the semantic verdict.
    if res.verified is not None:
        payload["verified"] = res.verified
    if res.effect is not None:
        payload["effect"] = res.effect
    if res.escalation is not None:
        payload["escalation"] = res.escalation
    if res.path is not None:
        payload["path"] = res.path
    if res.degraded is not None:
        payload["degraded"] = res.degraded
    if res.delivery_mode is not None:
        payload["delivery_mode"] = res.delivery_mode
    if res.code is not None:
        payload["code"] = res.code
    if res.meta:
        payload["meta"] = res.meta
    payload["verdict"] = _classify_action_result(res)
    return payload


def _text_response(res: ActionResult) -> str:
    return json.dumps(_action_payload(res))


# Default cap for the AX `elements` array returned by capture. Dense UIs
# (Electron apps, Obsidian, JetBrains IDEs) can publish 500+ AX nodes, which
# can exhaust session context after a single capture. The model-facing
# `max_elements` argument lets callers raise this when they need the full tree.
_DEFAULT_MAX_ELEMENTS = 100
# Hard upper bound on caller-supplied `max_elements`. Without this, a tool
# call passing a very large integer would silently disable the safeguard and
# reintroduce the original unbounded behavior.
_MAX_ALLOWED_MAX_ELEMENTS = 1000
_MIN_PROVIDER_IMAGE_DIMENSION = 8


def _image_dimensions_from_b64(image_b64: str) -> Optional[Tuple[int, int]]:
    """Return (width, height) for common inline screenshot formats.

    Some providers reject images below 8x8 before the model sees the tool
    result. Inspecting the encoded bytes here lets computer_use fall back to
    its AX/SOM text payload instead of sending an unusable placeholder.
    """
    if not image_b64:
        return None
    try:
        raw = base64.b64decode(image_b64, validate=False)
    except Exception:
        return None

    # PNG: signature + IHDR width/height.
    if raw.startswith(b"\x89PNG\r\n\x1a\n") and len(raw) >= 24:
        try:
            width, height = struct.unpack(">II", raw[16:24])
            return int(width), int(height)
        except Exception:
            return None

    # JPEG: scan for SOF markers that carry dimensions.
    if raw.startswith(b"\xff\xd8") and len(raw) > 4:
        i = 2
        while i + 9 < len(raw):
            if raw[i] != 0xFF:
                i += 1
                continue
            marker = raw[i + 1]
            i += 2
            while marker == 0xFF and i < len(raw):
                marker = raw[i]
                i += 1
            if marker in {0xD8, 0xD9}:
                continue
            if marker == 0xDA:
                break
            if 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):
                break
            if marker in {
                0xC0, 0xC1, 0xC2, 0xC3, 0xC5, 0xC6, 0xC7,
                0xC9, 0xCA, 0xCB, 0xCD, 0xCE, 0xCF,
            } and segment_len >= 7:
                height = int.from_bytes(raw[i + 3:i + 5], "big")
                width = int.from_bytes(raw[i + 5:i + 7], "big")
                return int(width), int(height)
            i += segment_len
    return None


def _coerce_max_elements(value: Any) -> int:
    """Validate the caller-supplied ``max_elements``.

    Falls back to :data:`_DEFAULT_MAX_ELEMENTS` for missing / non-integer /
    sub-1 inputs so the cap can never be silently disabled by a malformed
    tool-call argument. Clamps oversized values to
    :data:`_MAX_ALLOWED_MAX_ELEMENTS` so a caller cannot bypass the
    safeguard by passing a very large integer.
    """
    if value is None:
        return _DEFAULT_MAX_ELEMENTS
    try:
        n = int(value)
    except (TypeError, ValueError):
        return _DEFAULT_MAX_ELEMENTS
    if n < 1:
        return _DEFAULT_MAX_ELEMENTS
    if n > _MAX_ALLOWED_MAX_ELEMENTS:
        return _MAX_ALLOWED_MAX_ELEMENTS
    return n


def _capture_response(cap: CaptureResult, max_elements: int = _DEFAULT_MAX_ELEMENTS) -> Any:
    total_elements = len(cap.elements)
    visible_elements = cap.elements[:max_elements]
    truncated_elements = max(0, total_elements - len(visible_elements))
    image_dimensions = _image_dimensions_from_b64(cap.png_b64 or "") if cap.png_b64 else None
    response_width = image_dimensions[0] if image_dimensions else cap.width
    response_height = image_dimensions[1] if image_dimensions else cap.height
    image_too_small = bool(
        image_dimensions
        and (
            image_dimensions[0] < _MIN_PROVIDER_IMAGE_DIMENSION
            or image_dimensions[1] < _MIN_PROVIDER_IMAGE_DIMENSION
        )
    )

    # Index only what's actually surfaced in the response — otherwise the
    # human-readable summary references element indices the model cannot
    # find in the JSON `elements` array (e.g. max_elements=10 vs the default
    # 40-line index window).
    element_index = _format_elements(visible_elements)
    summary_lines = [
        f"capture mode={cap.mode} {response_width}x{response_height}"
        + (f" app={cap.app}" if cap.app else "")
        + (f" window={cap.window_title!r}" if cap.window_title else ""),
        f"{total_elements} interactable element(s):",
    ]
    if element_index:
        summary_lines.extend(element_index)
    # Multimodal and AX paths both reference `summary`; build it once up-front
    # so the aux-vision routing branch (which fires before either path is
    # selected) has a valid value to hand to _route_capture_through_aux_vision.
    # The AX path appends the "truncated to N of M" note to summary_lines
    # below and rebuilds; the multimodal path keeps this version untouched.
    if image_too_small:
        summary_lines.append(
            f"  (screenshot omitted: {image_dimensions[0]}x{image_dimensions[1]} "
            f"is below the {_MIN_PROVIDER_IMAGE_DIMENSION}x{_MIN_PROVIDER_IMAGE_DIMENSION} "
            "provider minimum)"
        )
    summary = "\n".join(summary_lines)

    if cap.png_b64 and cap.mode != "ax" and not image_too_small:
        # Decide whether to hand the screenshot to the auxiliary.vision
        # pipeline (text-only result) or keep the multimodal envelope (main
        # model handles vision natively). Issue #24015: previously the
        # multimodal envelope was returned unconditionally, so non-vision
        # main models tripped HTTP 404 / 400 at the provider boundary even
        # when auxiliary.vision was explicitly configured to handle this.
        if _should_route_through_aux_vision():
            routed = _route_capture_through_aux_vision(cap, summary)
            if routed is not None:
                return routed
            # Aux routing was requested but failed (vision node down, aux call
            # raised, empty analysis, etc.). Routing being requested means the
            # main model may not be able to consume images; falling through to
            # the multimodal envelope can break the capture with a provider
            # error. Degrade to the AX/SOM text payload instead so element
            # indices remain usable while vision is unavailable.
            summary_lines.append(
                "  (vision unavailable: the auxiliary vision model could not "
                "be reached; screenshot omitted. Element-index actions still "
                "work — drive via the element list above.)"
            )
            if truncated_elements:
                summary_lines.append(
                    f"  (response truncated to {len(visible_elements)} of "
                    f"{total_elements} elements; raise max_elements or pass "
                    "app= to narrow)"
                )
            payload = {
                "mode": cap.mode,
                "width": response_width,
                "height": response_height,
                "app": cap.app,
                "window_title": cap.window_title,
                "elements": [_element_to_dict(e) for e in visible_elements],
                "total_elements": total_elements,
                "summary": "\n".join(summary_lines),
                "vision_unavailable": True,
            }
            if truncated_elements:
                payload["truncated_elements"] = truncated_elements
            return json.dumps(payload)

        # Prefer the explicit MIME type cua-driver attaches to its image
        # parts (Surface 7 of NousResearch/hermes-agent#47072 — trycua/cua#1961
        # made `mimeType` part of every MCP image-part response). Fall back
        # to base64-prefix sniffing for older cua-driver builds that didn't
        # carry the field. JPEG base64 starts with /9j/; PNG with iVBOR.
        _mime = cap.image_mime_type
        if not _mime:
            _b64_prefix = cap.png_b64[:8]
            _mime = "image/jpeg" if _b64_prefix.startswith("/9j/") else "image/png"
        # The multimodal response carries the screenshot, not the AX
        # elements array, so a "response truncated to N of M elements"
        # note would be inaccurate — skip it on this branch.
        return {
            "_multimodal": True,
            "content": [
                {"type": "text", "text": summary},
                {"type": "image_url",
                 "image_url": {"url": f"data:{_mime};base64,{cap.png_b64}"}},
            ],
            "text_summary": summary,
            "meta": {"mode": cap.mode, "width": response_width, "height": response_height,
                     "elements": total_elements, "png_bytes": cap.png_bytes_len},
        }
    # AX-only (or image-missing fallback): text path actually carries the
    # `elements` array, so the truncation note applies here.
    if truncated_elements:
        summary_lines.append(
            f"  (response truncated to {len(visible_elements)} of {total_elements} elements; "
            f"raise max_elements or pass app= to narrow)"
        )
    summary = "\n".join(summary_lines)
    payload: Dict[str, Any] = {
        "mode": cap.mode,
        "width": response_width,
        "height": response_height,
        "app": cap.app,
        "window_title": cap.window_title,
        "elements": [_element_to_dict(e) for e in visible_elements],
        "total_elements": total_elements,
        "summary": summary,
    }
    if truncated_elements:
        payload["truncated_elements"] = truncated_elements
    return json.dumps(payload)


# ---------------------------------------------------------------------------
# auxiliary.vision routing for captured screenshots (#24015)
# ---------------------------------------------------------------------------

# Longest image side handed to the aux vision model. Full-resolution desktop
# captures tokenize heavily and can overflow small local-model context windows;
# ~1456px keeps SOM badges legible while cutting per-capture vision latency.
_MAX_VISION_DIM = 1456


def _shrink_capture_for_vision(raw: bytes, ext: str,
                               max_dim: int = _MAX_VISION_DIM) -> bytes:
    """Downscale encoded image bytes so the longest side is <= max_dim.

    Returns the original bytes unchanged when the image already fits or when
    Pillow is unavailable/fails — no worse than the pre-shrink behavior.
    """
    try:
        from io import BytesIO
        from PIL import Image
        img = Image.open(BytesIO(raw))
        if max(img.size) <= max_dim:
            return raw
        img.thumbnail((max_dim, max_dim))
        out = BytesIO()
        img.save(out, format="JPEG" if ext == ".jpg" else "PNG")
        return out.getvalue()
    except Exception as exc:
        logger.debug("computer_use: vision downscale skipped: %s", exc)
        return raw

def _should_route_through_aux_vision() -> bool:
    """Return True when ``_capture_response`` should hand the PNG to aux vision.

    Reads the active main provider/model and the loaded config and asks the
    routing helper. Any failure (config import, runtime override missing,
    etc.) returns False so the existing multimodal envelope continues to be
    returned — fail open on the routing decision so a broken config can
    never silently drop the screenshot for vision-capable main models.
    """
    try:
        from agent.auxiliary_client import _read_main_model, _read_main_provider
        from hermes_cli.config import load_config
        from tools.computer_use.vision_routing import (
            should_route_capture_to_aux_vision,
        )
    except Exception as exc:  # pragma: no cover - defensive
        logger.debug("computer_use: aux-vision routing import failed: %s", exc)
        return False
    try:
        provider = _read_main_provider() or ""
        model = _read_main_model() or ""
    except Exception as exc:  # pragma: no cover - defensive
        logger.debug("computer_use: aux-vision routing config read failed: %s", exc)
        return False
    cache_key = (str(provider), str(model))
    cached = _AUX_VISION_ROUTE_CACHE.get(cache_key)
    if cached is not None:
        return cached
    try:
        cfg = load_config()
        decision = bool(should_route_capture_to_aux_vision(provider, model, cfg))
    except Exception as exc:  # pragma: no cover - defensive
        logger.debug("computer_use: aux-vision routing decision failed: %s", exc)
        return False
    _AUX_VISION_ROUTE_CACHE[cache_key] = decision
    return decision


def _capture_after_mode() -> str:
    """Mode for ``capture_after`` follow-ups. Default ``som`` (screenshot)."""
    try:
        from hermes_cli.config import load_config

        raw = ((load_config() or {}).get("computer_use") or {}).get(
            "capture_after_mode", "som"
        )
    except Exception:
        return "som"
    mode = str(raw or "som").strip().lower()
    return mode if mode in {"som", "vision", "ax"} else "som"


def _route_capture_through_aux_vision(
    cap: CaptureResult,
    summary: str,
) -> Optional[str]:
    """Pre-analyse the captured PNG via ``vision_analyze`` and return a text result.

    The captured base64 PNG is materialised to ``$HERMES_HOME/cache/vision/``
    and handed to ``vision_analyze_tool`` with a generic describe prompt.
    The resulting text description is merged into the existing AX/SOM
    summary so the main model receives a single text payload that mentions
    every interactable element AND a description of what the screenshot
    looked like.

    Returns:
      A JSON-encoded text response on success.
      ``None`` on failure (caller falls back to the multimodal envelope).
    """
    if not cap.png_b64:
        return None
    try:
        import base64 as _base64
        import os as _os
        import uuid as _uuid

        from hermes_constants import get_hermes_dir
        from model_tools import _run_async
        from tools.vision_tools import vision_analyze_tool
    except Exception as exc:  # pragma: no cover - defensive
        logger.debug("computer_use: aux-vision import failed: %s", exc)
        return None

    temp_image_path = None
    try:
        try:
            raw = _base64.b64decode(cap.png_b64, validate=False)
        except Exception as exc:
            logger.debug("computer_use: failed to decode capture base64: %s", exc)
            return None

        # Pick an extension that matches the on-disk bytes so vision_analyze's
        # MIME sniffing returns the right content-type.
        # Surface 7: prefer the explicit MIME type cua-driver supplied.
        _mime_for_ext = cap.image_mime_type or ""
        if _mime_for_ext == "image/jpeg" or (not _mime_for_ext and cap.png_b64[:8].startswith("/9j/")):
            ext = ".jpg"
        else:
            ext = ".png"
        cache_dir = get_hermes_dir("cache/vision", "temp_vision_images")
        cache_dir.mkdir(parents=True, exist_ok=True)
        temp_image_path = cache_dir / f"computer_use_{_uuid.uuid4().hex}{ext}"
        raw = _shrink_capture_for_vision(raw, ext)
        temp_image_path.write_bytes(raw)

        prompt = (
            "Describe what is visible in this desktop application screenshot in "
            "concise but specific terms. Mention the app name and window "
            "title if visible, the overall layout, any labelled buttons, "
            "menus or text fields, and any prominent text content the user "
            "would need to know about. Do not invent details that are not "
            "actually visible.\n\n"
            f"AX/SOM index for cross-reference:\n{summary}"
        )

        result_json = _run_async(
            vision_analyze_tool(str(temp_image_path), prompt)
        )
    except Exception as exc:
        logger.warning(
            "computer_use: auxiliary.vision pre-analysis failed (%s); "
            "returning to caller without aux analysis",
            exc,
        )
        return None
    finally:
        if temp_image_path is not None:
            try:
                _os.unlink(str(temp_image_path))
            except Exception:
                pass

    analysis_text = ""
    if isinstance(result_json, str):
        try:
            parsed = json.loads(result_json)
            if isinstance(parsed, dict):
                analysis_text = str(parsed.get("analysis") or "").strip()
        except (TypeError, json.JSONDecodeError):
            analysis_text = result_json.strip()

    if not analysis_text:
        return None

    return json.dumps({
        "mode": cap.mode,
        "width": cap.width,
        "height": cap.height,
        "app": cap.app,
        "window_title": cap.window_title,
        "elements": [_element_to_dict(e) for e in cap.elements],
        "summary": summary,
        "vision_analysis": analysis_text,
        "vision_analysis_routed_via": "auxiliary.vision",
    })


def _maybe_follow_capture(
    backend: ComputerUseBackend, res: ActionResult, do_capture: bool,
) -> Any:
    if not do_capture:
        return _text_response(res)
    # Skip the follow-up capture when the action itself failed: showing a
    # normal-looking screenshot after a failure misleads the model into thinking
    # the action succeeded. Return the error text instead.
    if not res.ok:
        return _text_response(res)
    try:
        # Preserve the exact selected window when possible. Linux may expose a
        # generic app name for several unrelated windows, so app-only recapture
        # can silently switch targets after a successful action.
        target = getattr(backend, "_last_target", None) or {}
        pid = target.get("pid")
        window_id = target.get("window_id")
        mode = _capture_after_mode()
        if pid is not None and window_id is not None:
            cap = backend.capture(mode=mode, pid=pid, window_id=window_id)
        else:
            cap = backend.capture(mode=mode, app=getattr(backend, "_last_app", None))
    except Exception as e:
        logger.warning("follow-up capture failed: %s", e)
        return _text_response(res)
    # Combine action summary with the capture.
    resp = _capture_response(cap)
    if isinstance(resp, dict) and resp.get("_multimodal"):
        # Keep the complete evidence/verdict contract visible when an image is
        # attached; otherwise capture_after would accidentally discard the
        # very signal that governs whether repeating input is allowed.
        prefix = json.dumps(_action_payload(res))
        resp["content"][0]["text"] = prefix + "\n\n" + resp["content"][0]["text"]
        resp["text_summary"] = prefix + "\n\n" + resp["text_summary"]
        resp["action_result"] = _action_payload(res)
        return resp
    # Fallback: action + text capture merged.
    try:
        data = json.loads(resp)
    except (TypeError, json.JSONDecodeError):
        data = {"capture": resp}
    data.update(_action_payload(res))
    return json.dumps(data)


def _format_elements(elements: List[UIElement], max_lines: int = 40) -> List[str]:
    out: List[str] = []
    for e in elements[:max_lines]:
        label = e.label.replace("\n", " ")[:60]
        out.append(f"  #{e.index} {e.role} {label!r} @ {e.bounds}"
                   + (f" [{e.app}]" if e.app else ""))
    if len(elements) > max_lines:
        out.append(f"  ... +{len(elements) - max_lines} more (call capture with app= to narrow)")
    return out


def _element_to_dict(e: UIElement) -> Dict[str, Any]:
    return {
        "index": e.index,
        "role": e.role,
        "label": e.label,
        "bounds": list(e.bounds),
        "app": e.app,
    }


# ---------------------------------------------------------------------------
# Availability check (used by the tool registry check_fn)
# ---------------------------------------------------------------------------

def check_computer_use_requirements() -> bool:
    """Return True iff computer_use can run on this host.

    Conditions: macOS, Windows, or Linux + cua-driver binary installed (or
    override via env). cua-driver runs on all three; the Linux path is
    headed/X11 today (Wayland via XWayland), pure-Wayland progress tracked
    upstream. Linux users see specific blocked checks via
    `hermes computer-use doctor` if their session is incomplete (e.g. no
    DISPLAY set).
    """
    if sys.platform not in ("darwin", "win32", "linux"):
        return False
    from tools.computer_use.cua_backend import cua_driver_binary_available
    return cua_driver_binary_available()


def get_computer_use_schema() -> Dict[str, Any]:
    from tools.computer_use.schema import COMPUTER_USE_SCHEMA
    return COMPUTER_USE_SCHEMA