From b56963344984e37cfa30281469e4ae421ce9beb6 Mon Sep 17 00:00:00 2001
From: Teknium <127238744+teknium1@users.noreply.github.com>
Date: Wed, 2 Sep 2026 23:08:33 -0700
Subject: [PATCH 01/10] refactor(computer_use): fold
 _configured_permission_mode, _listing, _input into callers; merge capture
 MIME/ext helpers

---
 tools/computer_use/tool.py | 67 ++++++++++++++++++--------------------
 1 file changed, 31 insertions(+), 36 deletions(-)

diff --git a/tools/computer_use/tool.py b/tools/computer_use/tool.py
index 84bc6a9e9b..781a853386 100644
--- a/tools/computer_use/tool.py
+++ b/tools/computer_use/tool.py
@@ -94,19 +94,17 @@ _session_auto_approve: Dict[str, bool] = {}   # sid -> "always_approve everythin
 _always_allow: Dict[str, set] = {}            # sid -> set of (action, delivery_mode) scope keys
 _escalation_warned: set = set()               # sids already warned that a bypass widened the driver mode
 
-def _configured_permission_mode() -> str:
-    """Configured cua mode (standard | bounded; "standard" if unresolvable). bounded needs a
-    computer_use.capability_manifest — the backend fails loudly without it."""
-    with contextlib.suppress(Exception):
-        from tools.computer_use.cua_backend import _cua_configured_permission_mode
-        return _cua_configured_permission_mode()
-    return "standard"
-
 def _cua_permission_mode(session_id: str) -> str:
     """Map Hermes's approval bypass onto Cua's immutable mode; fails closed. Both identity namespaces are consulted
     (DB ``session_id`` and gateway ``session_key`` contextvar) or a gateway ``/yolo`` would be invisible here.
     Warns once per session that ``-z``/``--yolo`` swapped the driver onto a private ``unrestricted`` daemon, dropping
     the configured ceiling: deliberate (``unrestricted`` is not a config value) but easy to trigger by accident."""
+    # Configured cua mode (standard | bounded; "standard" if unresolvable). bounded needs a
+    # computer_use.capability_manifest — the backend fails loudly without it.
+    configured = "standard"
+    with contextlib.suppress(Exception):
+        from tools.computer_use.cua_backend import _cua_configured_permission_mode
+        configured = _cua_configured_permission_mode()
     with contextlib.suppress(Exception):
         from tools.approval import get_current_session_key, is_approval_bypass_active_for_session
         if is_approval_bypass_active_for_session(session_id) or (
@@ -115,14 +113,13 @@ def _cua_permission_mode(session_id: str) -> str:
                 warn = (key := str(session_id or "")) not in _escalation_warned
                 _escalation_warned.add(key)
             if warn:
-                configured = _configured_permission_mode()
                 logger.warning(
                     "computer_use: approval bypass (--yolo / -z) escalated the cua-driver permission mode from the "
                     "configured '%s' to 'unrestricted' for this session. Runtime approval prompts are disabled and the "
                     "driver's residual ceilings no longer apply. Drop the bypass flag to keep '%s', or declare a "
                     "version-3 computer_use.capability_manifest to keep a ceiling on bypassed runs.", configured, configured)
             return "unrestricted"
-    return _configured_permission_mode()
+    return configured
 
 def _new_backend(permission_mode: str) -> ComputerUseBackend:
     backend_name = os.environ.get("HERMES_COMPUTER_USE_BACKEND", "cua").lower()
@@ -360,8 +357,8 @@ def _do_focus_app(backend, action, args, **_):
         return json.dumps({"error": "focus_app requires `app`"})
     return backend.focus_app(args["app"], raise_window=bool(args.get("raise_window")))
 
-def _listing(key: str, method: str):
-    return lambda backend, action, args, **_: json.dumps({key: (items := getattr(backend, method)()), "count": len(items)})
+def _do_listing(backend, action, args, key, **_):
+    return json.dumps({key: (items := getattr(backend, action)()), "count": len(items)})
 
 def _summarize_click(action: str, args: Dict[str, Any], fg: str) -> str:
     where = (f" element #{args['element']}" if args.get("element") is not None
@@ -373,30 +370,29 @@ def _summarize_click(action: str, args: Dict[str, Any], fg: str) -> str:
 # ``summarize(action, args, fg_suffix)`` renders the one-line approval prompt.
 _ActionSpec = namedtuple("_ActionSpec", "handler input destructive summarize",
                          defaults=(False, False, lambda a, args, fg: a + fg))
-
-def _input(handler, summarize=None) -> _ActionSpec:
-    return _ActionSpec(handler, True, True, **({"summarize": summarize} if summarize else {}))
+_input = partial(_ActionSpec, input=True, destructive=True)
 
 _ACTIONS: Dict[str, _ActionSpec] = {
-    "click": _input(_do_click, _summarize_click),
-    "double_click": _input(partial(_do_click, count=2), _summarize_click),
-    "right_click": _input(partial(_do_click, button="right"), _summarize_click),
-    "middle_click": _input(partial(_do_click, button="middle"), _summarize_click),
-    "drag": _input(_do_drag, lambda a, args, fg: (f"drag {args.get('from_element') or args.get('from_coordinate')} → "
-                                                  f"{args.get('to_element') or args.get('to_coordinate')}{fg}")),
-    "scroll": _input(_do_scroll, lambda a, args, fg: f"scroll {args.get('direction', '?')} x{args.get('amount', 3)}{fg}"),
+    "click": _input(_do_click, summarize=_summarize_click),
+    "double_click": _input(partial(_do_click, count=2), summarize=_summarize_click),
+    "right_click": _input(partial(_do_click, button="right"), summarize=_summarize_click),
+    "middle_click": _input(partial(_do_click, button="middle"), summarize=_summarize_click),
+    "drag": _input(_do_drag, summarize=lambda a, args, fg: (
+        f"drag {args.get('from_element') or args.get('from_coordinate')} → "
+        f"{args.get('to_element') or args.get('to_coordinate')}{fg}")),
+    "scroll": _input(_do_scroll, summarize=lambda a, args, fg: f"scroll {args.get('direction', '?')} x{args.get('amount', 3)}{fg}"),
     "type": _input(lambda backend, action, args, **delivery: backend.type_text(args.get("text", ""), **delivery),
-                   lambda a, args, fg: (f"type {args.get('text', '')[:60]!r}"
-                                        + ("..." if len(args.get("text", "")) > 60 else "") + fg)),
+                   summarize=lambda a, args, fg: (f"type {args.get('text', '')[:60]!r}"
+                                                  + ("..." if len(args.get("text", "")) > 60 else "") + fg)),
     "key": _input(lambda backend, action, args, **delivery: backend.key(args.get("keys", ""), **delivery),
-                  lambda a, args, fg: f"key {args.get('keys', '')!r}{fg}"),
+                  summarize=lambda a, args, fg: f"key {args.get('keys', '')!r}{fg}"),
     "set_value": _input(_do_set_value),
     "focus_app": _ActionSpec(_do_focus_app, destructive=True, summarize=lambda a, args, fg: (
         f"focus {args.get('app', '')!r}" + (" (raise)" if args.get("raise_window") else ""))),
     "capture": _ActionSpec(_do_capture),
     "wait": _ActionSpec(lambda backend, action, args, **_: _text_response(backend.wait(float(args.get("seconds", 1.0))))),
-    "list_apps": _ActionSpec(_listing("apps", "list_apps")),
-    "list_windows": _ActionSpec(_listing("windows", "list_windows")),
+    "list_apps": _ActionSpec(partial(_do_listing, key="apps")),
+    "list_windows": _ActionSpec(partial(_do_listing, key="windows")),
 }
 # Native input actions deliver to the backend's sticky target; `app=` on these calls is NOT a
 # targeting parameter — see the mismatch guard in _dispatch.
@@ -477,12 +473,11 @@ _MAX_ELEMENT_LABEL_CHARS = 120
 # Bounded cache trails: every dense capture can spill, and CLI-only sessions never run the gateway's media cleanup.
 _MAX_SPILL_FILES = _MAX_CAPTURE_FILES = 20
 
-def _capture_mime(cap: CaptureResult) -> str:
-    # cua-driver's explicit MIME type, else sniff the base64 prefix (JPEG starts with /9j/, PNG with iVBOR).
-    return cap.image_mime_type or ("image/jpeg" if (cap.png_b64 or "").startswith("/9j/") else "image/png")
-
-def _capture_image_ext(cap: CaptureResult) -> str:
-    return ".jpg" if _capture_mime(cap).lower() == "image/jpeg" else ".png"  # matches on-disk bytes for MIME sniffing
+def _capture_image_format(cap: CaptureResult) -> Tuple[str, str]:
+    # (MIME, file extension): cua-driver's explicit MIME type, else sniff the base64 prefix (JPEG starts with /9j/,
+    # PNG with iVBOR). The extension matches the on-disk bytes for MIME sniffing.
+    mime = cap.image_mime_type or ("image/jpeg" if (cap.png_b64 or "").startswith("/9j/") else "image/png")
+    return mime, (".jpg" if mime.lower() == "image/jpeg" else ".png")
 
 def _bounds_unknown(bounds) -> bool:
     # No real geometry: KDE/Qt apps report [0, 0, 0, 0] for elements clickable by index; serializing that as a
@@ -582,7 +577,7 @@ def _multimodal_capture(v: SimpleNamespace, summary: str) -> Dict[str, Any]:
     return {
         "_multimodal": True,
         "content": [{"type": "text", "text": summary},
-                    {"type": "image_url", "image_url": {"url": f"data:{_capture_mime(cap)};base64,{cap.png_b64}"}}],
+                    {"type": "image_url", "image_url": {"url": f"data:{_capture_image_format(cap)[0]};base64,{cap.png_b64}"}}],
         "text_summary": summary,
         "meta": {"mode": cap.mode, "width": v.width, "height": v.height, "elements": v.total,
                  "png_bytes": cap.png_bytes_len, **_present(screenshot_path=v.screenshot_path,
@@ -682,7 +677,7 @@ def _write_cache_file(what: str, subdir: str, legacy: str, name: str, pattern: s
 def _persist_capture_image(cap: CaptureResult) -> Optional[str]:
     """Copy of the capture in Hermes' media cache so attachment surfaces can deliver it (None without an image)."""
     return _write_cache_file(
-        "screenshot persistence", "cache/images", "image_cache", f"computer_use_{uuid.uuid4().hex}{_capture_image_ext(cap)}",
+        "screenshot persistence", "cache/images", "image_cache", f"computer_use_{uuid.uuid4().hex}{_capture_image_format(cap)[1]}",
         "computer_use_*.*", _MAX_CAPTURE_FILES, lambda p: p.write_bytes(base64.b64decode(cap.png_b64, validate=False)),
     ) if cap.png_b64 else None
 
@@ -776,7 +771,7 @@ def _route_capture_through_aux_vision(
         problem = "failed to decode capture base64"
         raw = base64.b64decode(cap.png_b64, validate=False)
         problem = None  # from here on failures are loud (warning)
-        ext = _capture_image_ext(cap)
+        ext = _capture_image_format(cap)[1]
         temp_image_path = _cache_file("cache/vision", "temp_vision_images", f"computer_use_{uuid.uuid4().hex}{ext}")
         raw, scale_note = _shrink_capture_for_vision(raw, ext)
         temp_image_path.write_bytes(raw)

From 230524b5d1346100135ef91634584182ab42dacc Mon Sep 17 00:00:00 2001
From: Teknium <127238744+teknium1@users.noreply.github.com>
Date: Wed, 2 Sep 2026 23:10:01 -0700
Subject: [PATCH 02/10] refactor(computer_use): _NoopBackend capture/focus_app
 via _noop_stub result factory

---
 tools/computer_use/tool.py | 20 +++++++-------------
 1 file changed, 7 insertions(+), 13 deletions(-)

diff --git a/tools/computer_use/tool.py b/tools/computer_use/tool.py
index 781a853386..7bfc8e660a 100644
--- a/tools/computer_use/tool.py
+++ b/tools/computer_use/tool.py
@@ -219,9 +219,11 @@ def reset_backend_for_tests() -> None:  # pragma: no cover — tear down the cac
     _AUX_VISION_ROUTE_CACHE.clear()
 
 def _noop_stub(name: str, *params: str, result: Any = None):
-    # Recording stub: positional args are folded in under *params* (declared params default to None).
+    # Recording stub: positional args are folded in under *params* (declared params default to None). ``result`` may
+    # be a factory of the recorded kwargs; None -> a trivial ok ActionResult.
     def method(self, *pos, **kw):
-        return self._record(name, {**dict.fromkeys(params), **dict(zip(params, pos)), **kw}, result)
+        self.calls.append((name, call := {**dict.fromkeys(params), **dict(zip(params, pos)), **kw}))
+        return result(call) if callable(result) else ActionResult(ok=True, action=name) if result is None else result
     return method
 
 class _NoopBackend(ComputerUseBackend):  # pragma: no cover
@@ -234,20 +236,12 @@ class _NoopBackend(ComputerUseBackend):  # pragma: no cover
     def stop(self) -> None: pass
     def is_available(self) -> bool: return True
 
-    def _record(self, name: str, kw: Dict[str, Any], result: Any = None) -> Any:
-        self.calls.append((name, kw))
-        return ActionResult(ok=True, action=name) if result is None else result
-
-    def capture(self, mode="som", app=None, pid=None, window_id=None) -> CaptureResult:
-        return self._record("capture", {"mode": mode, "app": app, "pid": pid, "window_id": window_id},
-                            CaptureResult(mode=mode, width=1024, height=768, png_b64=None, elements=[],
-                                          app=app or "", window_title=""))
-
+    capture = _noop_stub("capture", "mode", "app", "pid", "window_id", result=lambda kw: CaptureResult(
+        mode=kw["mode"] or "som", width=1024, height=768, png_b64=None, elements=[], app=kw["app"] or "", window_title=""))
     click, drag, scroll = _noop_stub("click"), _noop_stub("drag"), _noop_stub("scroll")
     type_text, key, set_value = _noop_stub("type", "text"), _noop_stub("key", "keys"), _noop_stub("set_value", "value", "element")
     list_apps, list_windows = _noop_stub("list_apps", result=[]), _noop_stub("list_windows", result=[])
-    def focus_app(self, app: str, raise_window: bool = False) -> ActionResult:
-        return self._record("focus_app", {"app": app, "raise": raise_window})
+    focus_app = _noop_stub("focus_app", "app", "raise_window")
 
 
 # ── Dispatch ────────────────────────────────────────────────────────────────

From b651b77af5d44856dab2d1710fac60038448d31f Mon Sep 17 00:00:00 2001
From: Teknium <127238744+teknium1@users.noreply.github.com>
Date: Wed, 2 Sep 2026 23:12:47 -0700
Subject: [PATCH 03/10] refactor(computer_use): inline _view into its two
 builders; fold constant comments and _VERDICT_FIELDS

---
 tools/computer_use/tool.py | 49 +++++++++++++++-----------------------
 1 file changed, 19 insertions(+), 30 deletions(-)

diff --git a/tools/computer_use/tool.py b/tools/computer_use/tool.py
index 7bfc8e660a..8dc363e789 100644
--- a/tools/computer_use/tool.py
+++ b/tools/computer_use/tool.py
@@ -45,8 +45,7 @@ _BLOCKED_KEY_COMBOS = {
 }
 _KEY_ALIASES = {"command": "cmd", "control": "ctrl", "alt": "option", "⌘": "cmd", "⌥": "option",
                 "windows": "win", "super": "win", "meta": "win"}
-# Dangerous shell patterns for the `type` action (last one: fork bomb).
-_BLOCKED_TYPE_PATTERNS = [re.compile(p, re.IGNORECASE) for p in (
+_BLOCKED_TYPE_PATTERNS = [re.compile(p, re.IGNORECASE) for p in (  # dangerous shell patterns for `type` (last one: fork bomb)
     r"curl\s+[^|]*\|\s*bash", r"curl\s+[^|]*\|\s*sh", r"wget\s+[^|]*\|\s*bash",
     r"\bsudo\s+rm\s+-[rf]", r"\brm\s+-rf\s+/\s*$", r":\s*\(\)\s*\{\s*:\|:\s*&\s*\}")]
 
@@ -80,9 +79,8 @@ def _input_target_mismatch(backend, requested_app: str) -> Optional[str]:
 # ── Backend selection — env-swappable for tests ─────────────────────────────
 
 # Per-Hermes-session cached backends (own cua-driver session, native target, refs, grant namespace).
-# `_backend` is the backward-compatible empty-session injection hook (older tests).
 _backend_lock = threading.Lock()
-_backend: Optional[ComputerUseBackend] = None
+_backend: Optional[ComputerUseBackend] = None  # backward-compatible empty-session injection hook (older tests)
 _backends: Dict[str, ComputerUseBackend] = {}
 _backend_call_locks: Dict[str, threading.RLock] = {}
 _backend_permission_modes: Dict[str, str] = {}
@@ -442,8 +440,6 @@ def _classify_action_result(res: ActionResult) -> Dict[str, Any]:
     return {"decision": "verify_fresh_state",
             "hint": "Transport succeeded but the effect is unproven. Re-capture and confirm before continuing."}
 
-_VERDICT_FIELDS = ("verified", "effect", "escalation", "path", "degraded", "delivery_mode", "code")
-
 def _present(**fields: Any) -> Dict[str, Any]:
     return {k: v for k, v in fields.items() if v}  # only the truthy optional fields, in the given order
 
@@ -451,21 +447,19 @@ def _action_payload(res: ActionResult) -> Dict[str, Any]:
     # cua-driver's structured verdict fields only when returned (None = old driver). ok is transport success;
     # effect/escalation are the semantic verdict.
     return {"ok": res.ok, "action": res.action, **_present(message=res.message),
-            **{k: v for k in _VERDICT_FIELDS if (v := getattr(res, k)) is not None}, **_present(meta=res.meta),
+            **{k: v for k in ("verified", "effect", "escalation", "path", "degraded", "delivery_mode", "code")
+               if (v := getattr(res, k)) is not None}, **_present(meta=res.meta),
             "verdict": _classify_action_result(res)}
 
 def _text_response(res: ActionResult) -> str:
     return json.dumps(_action_payload(res))
 
-# AX `elements` cap: dense UIs publish 500+ nodes (one capture would exhaust context); the full tree spills to a file.
-_DEFAULT_MAX_ELEMENTS = 100
-# Some providers reject images below 8x8 before the model sees the result; such captures fall back to text.
-_MIN_PROVIDER_IMAGE_DIMENSION = 8
+_DEFAULT_MAX_ELEMENTS = 100  # AX `elements` cap: dense UIs publish 500+ nodes (one capture would exhaust context); the full tree spills to a file
+_MIN_PROVIDER_IMAGE_DIMENSION = 8  # some providers reject images below 8x8 before the model sees the result; such captures fall back to text
 # Some AX trees (Discord/Slack via UIA, Electron chat clients) expose ENTIRE message bodies as labels; uncapped
 # they blew the tool-result budget and leaked private chat text. Labels identify a control, not text extraction.
 _MAX_ELEMENT_LABEL_CHARS = 120
-# Bounded cache trails: every dense capture can spill, and CLI-only sessions never run the gateway's media cleanup.
-_MAX_SPILL_FILES = _MAX_CAPTURE_FILES = 20
+_MAX_SPILL_FILES = _MAX_CAPTURE_FILES = 20  # bounded cache trails: every dense capture can spill, and CLI-only sessions never run the gateway's media cleanup
 
 def _capture_image_format(cap: CaptureResult) -> Tuple[str, str]:
     # (MIME, file extension): cua-driver's explicit MIME type, else sniff the base64 prefix (JPEG starts with /9j/,
@@ -518,18 +512,12 @@ def _bounds_hints(elements: List[UIElement], image_width: int, image_height: int
 
 def _bounds_scale(elements: List[UIElement], image_width: int, image_height: int) -> Optional[float]:
     return _bounds_hints(elements, image_width, image_height)[0]
-
 def _bounds_space_note(elements: List[UIElement], image_width: int, image_height: int) -> Optional[str]:
     return _bounds_hints(elements, image_width, image_height)[1]
 
-def _view(cap: CaptureResult, visible: List[UIElement], width: int, height: int, **facts: Any) -> SimpleNamespace:
-    """One capture's derived facts, computed once for every response branch: ``visible`` is the capped element list,
-    ``dims_omitted`` an image below the provider minimum; ``facts`` override the None/False defaults."""
-    return SimpleNamespace(**{**dict(cap=cap, visible=visible, total=len(cap.elements), width=width, height=height,
-                                     truncated=len(cap.elements) - len(visible), bounds_scale=None, bounds_note=None,
-                                     elements_file=None, screenshot_path=None, dims_omitted=None, has_image=False), **facts})
-
 def _capture_view(cap: CaptureResult, max_elements: int) -> SimpleNamespace:
+    """One capture's derived facts, computed once for every response branch: ``visible`` is the capped element list,
+    ``dims_omitted`` an image below the provider minimum."""
     visible, dims = cap.elements[:max_elements], None
     with contextlib.suppress(Exception):  # (width, height) of the inline PNG/JPEG screenshot, else the backend's
         dims = image_dimensions_from_bytes(base64.b64decode(cap.png_b64, validate=False)) if cap.png_b64 else None
@@ -539,10 +527,11 @@ def _capture_view(cap: CaptureResult, max_elements: int) -> SimpleNamespace:
     lost_detail = len(cap.elements) > len(visible) or any(len(e.label) > _MAX_ELEMENT_LABEL_CHARS for e in visible)
     too_small = bool(dims) and min(dims) < _MIN_PROVIDER_IMAGE_DIMENSION
     has_image = bool(cap.png_b64) and cap.mode != "ax" and not too_small
-    return _view(cap, visible, width, height, bounds_scale=scale, bounds_note=note,
-                 elements_file=_spill_elements_to_file(cap) if lost_detail else None,
-                 screenshot_path=_persist_capture_image(cap) if has_image else None,
-                 dims_omitted=dims if too_small else None, has_image=has_image)
+    return SimpleNamespace(cap=cap, visible=visible, total=len(cap.elements), width=width, height=height,
+                           truncated=len(cap.elements) - len(visible), bounds_scale=scale, bounds_note=note,
+                           elements_file=_spill_elements_to_file(cap) if lost_detail else None,
+                           screenshot_path=_persist_capture_image(cap) if has_image else None,
+                           dims_omitted=dims if too_small else None, has_image=has_image)
 
 def _capture_summary_lines(v: SimpleNamespace) -> List[str]:
     """Human-readable capture summary; line ORDER is contract. Lists only what `elements` surfaces, otherwise the
@@ -688,9 +677,8 @@ def _spill_elements_to_file(cap: CaptureResult) -> Optional[str]:
 
 # ── auxiliary.vision routing for captured screenshots ───────────────────────
 
-# 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 vision latency.
-_MAX_VISION_DIM = 1456
+_MAX_VISION_DIM = 1456  # 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 vision latency.
 
 def _shrink_capture_for_vision(raw: bytes, ext: str, max_dim: int = _MAX_VISION_DIM) -> tuple[bytes, Optional[str]]:
     """Downscale encoded image bytes so the longest side is <= max_dim -> ``(bytes, scale_note)``. note is None when
@@ -791,8 +779,9 @@ def _route_capture_through_aux_vision(
         return None
     # Same element cap as every other capture branch; dumping cap.elements in full would bypass max_elements
     # exactly for non-vision main models. Dimensions are the backend's on this branch.
-    view = _view(cap, cap.elements if visible_elements is None else visible_elements, cap.width, cap.height,
-                 truncated=truncated_elements, elements_file=elements_file, screenshot_path=screenshot_path)
+    view = SimpleNamespace(cap=cap, visible=cap.elements if visible_elements is None else visible_elements,
+                           total=len(cap.elements), width=cap.width, height=cap.height, truncated=truncated_elements,
+                           elements_file=elements_file, screenshot_path=screenshot_path, bounds_scale=None)
     return _text_capture_payload(view, summary, {"vision_analysis": analysis_text,
                                                  "vision_analysis_routed_via": "auxiliary.vision"})
 

From 775f1d77dcb5f1df3a0f9a636f1a93e7ccb136d7 Mon Sep 17 00:00:00 2001
From: Teknium <127238744+teknium1@users.noreply.github.com>
Date: Wed, 2 Sep 2026 23:17:04 -0700
Subject: [PATCH 04/10] refactor(computer_use): inline _multimodal_capture,
 drop dead non-JSON fallback in follow capture, _xy as kwargs

---
 tools/computer_use/tool.py | 42 +++++++++++++++-----------------------
 1 file changed, 16 insertions(+), 26 deletions(-)

diff --git a/tools/computer_use/tool.py b/tools/computer_use/tool.py
index 8dc363e789..702a70fdfd 100644
--- a/tools/computer_use/tool.py
+++ b/tools/computer_use/tool.py
@@ -308,13 +308,12 @@ def _summarize_action(action: str, args: Dict[str, Any]) -> str:
 # --- handlers: (backend, action, args, **delivery) -> ActionResult (_dispatch applies the follow-up capture) or a
 #     final str/dict result. `delivery` = delivery_mode + bring_to_front; only input actions use it.
 
-def _xy(args: Dict[str, Any]) -> Tuple[Any, Any]:
+def _xy(args: Dict[str, Any]) -> Dict[str, Any]:
     coord = args.get("coordinate")
-    return (coord[0], coord[1]) if coord and coord[0] is not None else (None, None)
+    return dict(x=coord[0], y=coord[1]) if coord and coord[0] is not None else dict(x=None, y=None)
 
 def _do_click(backend, action, args, button=None, count=1, **delivery):
-    x, y = _xy(args)
-    return backend.click(element=args.get("element"), x=x, y=y, button=button or args.get("button") or "left",
+    return backend.click(element=args.get("element"), **_xy(args), button=button or args.get("button") or "left",
                          click_count=count, modifiers=args.get("modifiers"), **delivery)
 
 def _do_drag(backend, action, args, **delivery):
@@ -326,9 +325,8 @@ def _do_drag(backend, action, args, **delivery):
                         button=args.get("button", "left"), modifiers=args.get("modifiers"), **delivery)
 
 def _do_scroll(backend, action, args, **delivery):
-    x, y = _xy(args)
     return backend.scroll(direction=args.get("direction", "down"), amount=int(args.get("amount", 3)),
-                          element=args.get("element"), x=x, y=y, modifiers=args.get("modifiers"), **delivery)
+                          element=args.get("element"), **_xy(args), modifiers=args.get("modifiers"), **delivery)
 
 def _do_set_value(backend, action, args, **delivery):
     if args.get("value") is None:
@@ -386,8 +384,7 @@ _ACTIONS: Dict[str, _ActionSpec] = {
     "list_apps": _ActionSpec(partial(_do_listing, key="apps")),
     "list_windows": _ActionSpec(partial(_do_listing, key="windows")),
 }
-# Native input actions deliver to the backend's sticky target; `app=` on these calls is NOT a
-# targeting parameter — see the mismatch guard in _dispatch.
+# Native input actions deliver to the backend's sticky target; `app=` is NOT a targeting parameter (guard in _dispatch).
 _INPUT_ACTIONS = frozenset(a for a, s in _ACTIONS.items() if s.input)
 
 # Unknown actions are never aliased (no repairing bad model output), but the nearest real action is
@@ -555,18 +552,6 @@ def _capture_summary_lines(v: SimpleNamespace) -> List[str]:
            f"{_MIN_PROVIDER_IMAGE_DIMENSION}x{_MIN_PROVIDER_IMAGE_DIMENSION} provider minimum)"] if v.dims_omitted else []),
     ]
 
-def _multimodal_capture(v: SimpleNamespace, summary: str) -> Dict[str, Any]:
-    cap = v.cap  # envelope carrying the screenshot (not the elements array, so no truncation note)
-    return {
-        "_multimodal": True,
-        "content": [{"type": "text", "text": summary},
-                    {"type": "image_url", "image_url": {"url": f"data:{_capture_image_format(cap)[0]};base64,{cap.png_b64}"}}],
-        "text_summary": summary,
-        "meta": {"mode": cap.mode, "width": v.width, "height": v.height, "elements": v.total,
-                 "png_bytes": cap.png_bytes_len, **_present(screenshot_path=v.screenshot_path,
-                                                          elements_file=v.elements_file, bounds_scale=v.bounds_scale)},
-    }
-
 def _text_capture_payload(v: SimpleNamespace, summary: str, extra: Optional[Dict[str, Any]] = None) -> str:
     """JSON text payload shared by the AX, vision-unavailable and aux-vision branches. Key order is contract:
     fixed fields, ``extra`` branch markers, then set optionals."""
@@ -587,8 +572,16 @@ def _capture_response(cap: CaptureResult, max_elements: int = _DEFAULT_MAX_ELEME
     if v.has_image:
         # Hand the screenshot to auxiliary.vision (text-only result) when the main model may not consume images
         # natively; returning the multimodal envelope unconditionally tripped HTTP 404/400 at the provider.
-        if not _should_route_through_aux_vision():
-            return _multimodal_capture(v, summary)
+        if not _should_route_through_aux_vision():  # envelope carrying the screenshot (not the elements array, so no truncation note)
+            return {
+                "_multimodal": True,
+                "content": [{"type": "text", "text": summary},
+                            {"type": "image_url", "image_url": {"url": f"data:{_capture_image_format(cap)[0]};base64,{cap.png_b64}"}}],
+                "text_summary": summary,
+                "meta": {"mode": cap.mode, "width": v.width, "height": v.height, "elements": v.total,
+                         "png_bytes": cap.png_bytes_len, **_present(screenshot_path=v.screenshot_path,
+                                                                  elements_file=v.elements_file, bounds_scale=v.bounds_scale)},
+            }
         routed = _route_capture_through_aux_vision(
             cap, summary, visible_elements=v.visible, truncated_elements=v.truncated,
             elements_file=v.elements_file, screenshot_path=v.screenshot_path)
@@ -626,10 +619,7 @@ def _maybe_follow_capture(backend: ComputerUseBackend, res: ActionResult, do_cap
         resp["text_summary"] = prefix + resp["text_summary"]
         resp["action_result"] = payload
         return resp
-    data = {"capture": resp}  # text capture: merge the action payload in
-    with contextlib.suppress(TypeError, json.JSONDecodeError):
-        data = json.loads(resp)
-    return json.dumps({**data, **payload})
+    return json.dumps({**json.loads(resp), **payload})  # text capture: merge the action payload in
 
 
 # ── Cache files (screenshots, element spills, vision temps) ─────────────────

From eebddc22e81b6c4cb07f68d4595c3338c4a739f1 Mon Sep 17 00:00:00 2001
From: Teknium <127238744+teknium1@users.noreply.github.com>
Date: Wed, 2 Sep 2026 23:19:21 -0700
Subject: [PATCH 05/10] refactor(computer_use): drop dead re-raise branch in
 _stop_backend; compact _do_capture target forwarding

---
 tools/computer_use/tool.py | 9 +++------
 1 file changed, 3 insertions(+), 6 deletions(-)

diff --git a/tools/computer_use/tool.py b/tools/computer_use/tool.py
index 702a70fdfd..b5e8b36803 100644
--- a/tools/computer_use/tool.py
+++ b/tools/computer_use/tool.py
@@ -150,15 +150,13 @@ def _detach_locked(sid: str) -> Tuple[Optional[ComputerUseBackend], Optional[thr
     return backend, call_lock
 
 def _stop_backend(backend: ComputerUseBackend, call_lock: Optional[threading.RLock],
-                  on_error: Optional[Callable[[Exception], None]] = None) -> None:
+                  on_error: Callable[[Exception], None]) -> None:
     """Stop under the session call lock (if any) so an in-flight action finishes first. Never called under
-    ``_backend_lock`` (unrelated sessions stay free). Raises unless ``on_error`` absorbs the failure."""
+    ``_backend_lock`` (unrelated sessions stay free). ``on_error`` absorbs the failure (never raises)."""
     try:
         with call_lock if call_lock is not None else contextlib.nullcontext():
             backend.stop()
     except Exception as e:
-        if on_error is None:
-            raise
         on_error(e)
 
 def _get_backend(session_id: str = "") -> ComputerUseBackend:
@@ -338,9 +336,8 @@ def _do_capture(backend, action, args, **_):
     if mode not in {"som", "vision", "ax"}:
         return json.dumps({"error": f"bad mode {mode!r}; use som|vision|ax"})
     # pid/window_id forwarded only when given so older backends keep their defaults.
-    target = {k: args.get(k) for k in ("pid", "window_id")}
     return _capture_response(backend.capture(mode=mode, app=args.get("app"),
-                                             **(target if any(v is not None for v in target.values()) else {})))
+                                             **{k: args[k] for k in ("pid", "window_id") if args.get(k) is not None}))
 
 def _do_focus_app(backend, action, args, **_):
     if not args.get("app"):

From 0bd806bd22fa3643658e510ca70b5ff936a5b24b Mon Sep 17 00:00:00 2001
From: Teknium <127238744+teknium1@users.noreply.github.com>
Date: Wed, 2 Sep 2026 23:21:06 -0700
Subject: [PATCH 06/10] refactor(computer_use): drop cap aliases in capture
 builders, single-line noop __init__, compact section spacing

---
 tools/computer_use/tool.py | 34 ++++++++++------------------------
 1 file changed, 10 insertions(+), 24 deletions(-)

diff --git a/tools/computer_use/tool.py b/tools/computer_use/tool.py
index b5e8b36803..ebaca15d86 100644
--- a/tools/computer_use/tool.py
+++ b/tools/computer_use/tool.py
@@ -24,7 +24,6 @@ from tools.computer_use.backend import ActionResult, CaptureResult, ComputerUseB
 
 logger = logging.getLogger(__name__)
 
-
 # ── Approval & safety ───────────────────────────────────────────────────────
 
 _approval_callback = None
@@ -75,7 +74,6 @@ def _input_target_mismatch(backend, requested_app: str) -> Optional[str]:
     current, wanted = (last_app or "").strip().lower(), requested_app.strip().lower()
     return None if not current or not wanted or wanted in current or current in wanted else last_app
 
-
 # ── Backend selection — env-swappable for tests ─────────────────────────────
 
 # Per-Hermes-session cached backends (own cua-driver session, native target, refs, grant namespace).
@@ -225,9 +223,7 @@ def _noop_stub(name: str, *params: str, result: Any = None):
 class _NoopBackend(ComputerUseBackend):  # pragma: no cover
     """Test/CI stub (HERMES_COMPUTER_USE_BACKEND=noop). Records ``(name, kwargs)`` calls; returns trivial results."""
 
-    def __init__(self) -> None:
-        self.calls: List[Tuple[str, Dict[str, Any]]] = []
-
+    def __init__(self) -> None: self.calls: List[Tuple[str, Dict[str, Any]]] = []
     def start(self) -> None: pass
     def stop(self) -> None: pass
     def is_available(self) -> bool: return True
@@ -239,7 +235,6 @@ class _NoopBackend(ComputerUseBackend):  # pragma: no cover
     list_apps, list_windows = _noop_stub("list_apps", result=[]), _noop_stub("list_windows", result=[])
     focus_app = _noop_stub("focus_app", "app", "raise_window")
 
-
 # ── Dispatch ────────────────────────────────────────────────────────────────
 
 def handle_computer_use(args: Dict[str, Any], **kwargs) -> Any:
@@ -364,9 +359,8 @@ _ACTIONS: Dict[str, _ActionSpec] = {
     "double_click": _input(partial(_do_click, count=2), summarize=_summarize_click),
     "right_click": _input(partial(_do_click, button="right"), summarize=_summarize_click),
     "middle_click": _input(partial(_do_click, button="middle"), summarize=_summarize_click),
-    "drag": _input(_do_drag, summarize=lambda a, args, fg: (
-        f"drag {args.get('from_element') or args.get('from_coordinate')} → "
-        f"{args.get('to_element') or args.get('to_coordinate')}{fg}")),
+    "drag": _input(_do_drag, summarize=lambda a, args, fg: (f"drag {args.get('from_element') or args.get('from_coordinate')} → "
+                                                             f"{args.get('to_element') or args.get('to_coordinate')}{fg}")),
     "scroll": _input(_do_scroll, summarize=lambda a, args, fg: f"scroll {args.get('direction', '?')} x{args.get('amount', 3)}{fg}"),
     "type": _input(lambda backend, action, args, **delivery: backend.type_text(args.get("text", ""), **delivery),
                    summarize=lambda a, args, fg: (f"type {args.get('text', '')[:60]!r}"
@@ -387,9 +381,8 @@ _INPUT_ACTIONS = frozenset(a for a, s in _ACTIONS.items() if s.input)
 # Unknown actions are never aliased (no repairing bad model output), but the nearest real action is
 # named so a bare error isn't the only guidance.
 _ACTION_SUGGESTIONS = {
-    "hotkey": "key", "press_key": "key", "keypress": "key", "key_combo": "key", "shortcut": "key",
-    "type_text": "type", "input_text": "type", "screenshot": "capture", "get_window_state": "capture",
-    "left_click": "click", "mouse_click": "click",
+    "hotkey": "key", "press_key": "key", "keypress": "key", "key_combo": "key", "shortcut": "key", "type_text": "type",
+    "input_text": "type", "screenshot": "capture", "get_window_state": "capture", "left_click": "click", "mouse_click": "click",
 }
 
 def _dispatch(backend: ComputerUseBackend, action: str, args: Dict[str, Any]) -> Any:
@@ -411,7 +404,6 @@ def _dispatch(backend: ComputerUseBackend, action: str, args: Dict[str, Any]) ->
                        bring_to_front=bool(args.get("bring_to_front")))
     return res if isinstance(res, (str, dict)) else _maybe_follow_capture(backend, res, bool(args.get("capture_after")))
 
-
 # ── Response shaping ────────────────────────────────────────────────────────
 
 def _classify_action_result(res: ActionResult) -> Dict[str, Any]:
@@ -430,8 +422,7 @@ def _classify_action_result(res: ActionResult) -> Dict[str, Any]:
             "re-issue by coordinate; 'foreground' (or a failed pixel click) → re-issue with "
             "delivery_mode='foreground' (separate approval). Do not predict the rung from the "
             "app being Electron/Chromium — react to this signal.")}
-    # Transport success without semantic proof is not proof of effect.
-    return {"decision": "verify_fresh_state",
+    return {"decision": "verify_fresh_state",  # transport success without semantic proof is not proof of effect
             "hint": "Transport succeeded but the effect is unproven. Re-capture and confirm before continuing."}
 
 def _present(**fields: Any) -> Dict[str, Any]:
@@ -530,18 +521,17 @@ def _capture_view(cap: CaptureResult, max_elements: int) -> SimpleNamespace:
 def _capture_summary_lines(v: SimpleNamespace) -> List[str]:
     """Human-readable capture summary; line ORDER is contract. Lists only what `elements` surfaces, otherwise the
     summary names indices the model can't find."""
-    cap = v.cap
     notes = (
         v.bounds_note and v.bounds_note + (f"; estimated scale ~{v.bounds_scale}x (screenshot position x "
                                            f"{v.bounds_scale} ≈ native coordinate)" if v.bounds_scale else ""),
         v.screenshot_path and f"shareable screenshot saved to {v.screenshot_path}",
-        cap.note,
+        v.cap.note,
         v.elements_file and (f"full element tree with untruncated labels saved to {v.elements_file} — "
                              "read_file/search_files it if you need dropped label text or elements beyond the cap"),
     )
     return [
-        f"capture mode={cap.mode} {v.width}x{v.height}"
-        + (f" app={cap.app}" if cap.app else "") + (f" window={cap.window_title!r}" if cap.window_title else ""),
+        f"capture mode={v.cap.mode} {v.width}x{v.height}"
+        + (f" app={v.cap.app}" if v.cap.app else "") + (f" window={v.cap.window_title!r}" if v.cap.window_title else ""),
         f"{v.total} interactable element(s):",
         *(f"  ({note})" for note in notes if note),
         *_format_elements(v.visible),
@@ -552,9 +542,8 @@ def _capture_summary_lines(v: SimpleNamespace) -> List[str]:
 def _text_capture_payload(v: SimpleNamespace, summary: str, extra: Optional[Dict[str, Any]] = None) -> str:
     """JSON text payload shared by the AX, vision-unavailable and aux-vision branches. Key order is contract:
     fixed fields, ``extra`` branch markers, then set optionals."""
-    cap = v.cap
     return json.dumps({
-        "mode": cap.mode, "width": v.width, "height": v.height, "app": cap.app, "window_title": cap.window_title,
+        "mode": v.cap.mode, "width": v.width, "height": v.height, "app": v.cap.app, "window_title": v.cap.window_title,
         "elements": [_element_to_dict(e) for e in v.visible], "total_elements": v.total, "summary": summary,
         **(extra or {}),
         **_present(truncated_elements=v.truncated, elements_file=v.elements_file,
@@ -618,7 +607,6 @@ def _maybe_follow_capture(backend: ComputerUseBackend, res: ActionResult, do_cap
         return resp
     return json.dumps({**json.loads(resp), **payload})  # text capture: merge the action payload in
 
-
 # ── Cache files (screenshots, element spills, vision temps) ─────────────────
 
 def _cache_file(subdir: str, legacy: str, name: str, pattern: str = "", cap: int = 0):
@@ -661,7 +649,6 @@ def _spill_elements_to_file(cap: CaptureResult) -> Optional[str]:
     return _write_cache_file("element spill", "cache/computer_use", "computer_use_cache",
                              f"elements_{uuid.uuid4().hex}.json", "elements_*.json", _MAX_SPILL_FILES, write)
 
-
 # ── auxiliary.vision routing for captured screenshots ───────────────────────
 
 _MAX_VISION_DIM = 1456  # longest image side handed to the aux vision model: full-resolution desktop captures tokenize heavily
@@ -772,7 +759,6 @@ def _route_capture_through_aux_vision(
     return _text_capture_payload(view, summary, {"vision_analysis": analysis_text,
                                                  "vision_analysis_routed_via": "auxiliary.vision"})
 
-
 # ── Availability check (used by the tool registry check_fn) ─────────────────
 
 def check_computer_use_requirements() -> bool:

From 06c02431807799dab920c117da659a30ff91a473 Mon Sep 17 00:00:00 2001
From: Teknium <127238744+teknium1@users.noreply.github.com>
Date: Wed, 2 Sep 2026 23:22:16 -0700
Subject: [PATCH 07/10] refactor(computer_use): lambda writer for element
 spill; single prefix assignment in follow capture

---
 tools/computer_use/tool.py | 17 +++++++----------
 1 file changed, 7 insertions(+), 10 deletions(-)

diff --git a/tools/computer_use/tool.py b/tools/computer_use/tool.py
index ebaca15d86..b0fe3122bc 100644
--- a/tools/computer_use/tool.py
+++ b/tools/computer_use/tool.py
@@ -600,9 +600,7 @@ def _maybe_follow_capture(backend: ComputerUseBackend, res: ActionResult, do_cap
     resp, payload = _capture_response(cap), _action_payload(res)
     if isinstance(resp, dict) and resp.get("_multimodal"):
         # Keep the evidence/verdict contract visible alongside the image — it governs whether input may repeat.
-        prefix = json.dumps(payload) + "\n\n"
-        resp["content"][0]["text"] = prefix + resp["content"][0]["text"]
-        resp["text_summary"] = prefix + resp["text_summary"]
+        resp["content"][0]["text"] = resp["text_summary"] = json.dumps(payload) + "\n\n" + resp["text_summary"]
         resp["action_result"] = payload
         return resp
     return json.dumps({**json.loads(resp), **payload})  # text capture: merge the action payload in
@@ -641,13 +639,12 @@ def _persist_capture_image(cap: CaptureResult) -> Optional[str]:
 
 def _spill_elements_to_file(cap: CaptureResult) -> Optional[str]:
     """FULL element tree (untruncated labels) in a cache file — the read_file/search_files escape hatch for capped text."""
-    def write(path) -> None:
-        payload = {"app": cap.app, "window_title": cap.window_title, "total_elements": len(cap.elements),
-                   "elements": [{"index": e.index, "role": e.role, "label": e.label,
-                                 "bounds": list(e.bounds), "app": e.app} for e in cap.elements]}
-        path.write_text(json.dumps(payload, ensure_ascii=False, indent=1), encoding="utf-8")
-    return _write_cache_file("element spill", "cache/computer_use", "computer_use_cache",
-                             f"elements_{uuid.uuid4().hex}.json", "elements_*.json", _MAX_SPILL_FILES, write)
+    payload = {"app": cap.app, "window_title": cap.window_title, "total_elements": len(cap.elements),
+               "elements": [{"index": e.index, "role": e.role, "label": e.label, "bounds": list(e.bounds), "app": e.app}
+                            for e in cap.elements]}
+    return _write_cache_file("element spill", "cache/computer_use", "computer_use_cache", f"elements_{uuid.uuid4().hex}.json",
+                             "elements_*.json", _MAX_SPILL_FILES,
+                             lambda p: p.write_text(json.dumps(payload, ensure_ascii=False, indent=1), encoding="utf-8"))
 
 # ── auxiliary.vision routing for captured screenshots ───────────────────────
 

From b8f6a8693abbb7aa4768c8f421262cd0ec93695e Mon Sep 17 00:00:00 2001
From: Teknium <127238744+teknium1@users.noreply.github.com>
Date: Wed, 2 Sep 2026 23:23:59 -0700
Subject: [PATCH 08/10] refactor(computer_use): set_value/focus_app handlers as
 table lambdas

---
 tools/computer_use/tool.py | 23 ++++++++---------------
 1 file changed, 8 insertions(+), 15 deletions(-)

diff --git a/tools/computer_use/tool.py b/tools/computer_use/tool.py
index b0fe3122bc..67efc387dd 100644
--- a/tools/computer_use/tool.py
+++ b/tools/computer_use/tool.py
@@ -321,11 +321,6 @@ def _do_scroll(backend, action, args, **delivery):
     return backend.scroll(direction=args.get("direction", "down"), amount=int(args.get("amount", 3)),
                           element=args.get("element"), **_xy(args), modifiers=args.get("modifiers"), **delivery)
 
-def _do_set_value(backend, action, args, **delivery):
-    if args.get("value") is None:
-        return json.dumps({"error": "set_value requires `value`"})
-    return backend.set_value(value=str(args["value"]), element=args.get("element"))
-
 def _do_capture(backend, action, args, **_):
     mode = str(args.get("mode", "som"))
     if mode not in {"som", "vision", "ax"}:
@@ -334,11 +329,6 @@ def _do_capture(backend, action, args, **_):
     return _capture_response(backend.capture(mode=mode, app=args.get("app"),
                                              **{k: args[k] for k in ("pid", "window_id") if args.get(k) is not None}))
 
-def _do_focus_app(backend, action, args, **_):
-    if not args.get("app"):
-        return json.dumps({"error": "focus_app requires `app`"})
-    return backend.focus_app(args["app"], raise_window=bool(args.get("raise_window")))
-
 def _do_listing(backend, action, args, key, **_):
     return json.dumps({key: (items := getattr(backend, action)()), "count": len(items)})
 
@@ -367,9 +357,13 @@ _ACTIONS: Dict[str, _ActionSpec] = {
                                                   + ("..." if len(args.get("text", "")) > 60 else "") + fg)),
     "key": _input(lambda backend, action, args, **delivery: backend.key(args.get("keys", ""), **delivery),
                   summarize=lambda a, args, fg: f"key {args.get('keys', '')!r}{fg}"),
-    "set_value": _input(_do_set_value),
-    "focus_app": _ActionSpec(_do_focus_app, destructive=True, summarize=lambda a, args, fg: (
-        f"focus {args.get('app', '')!r}" + (" (raise)" if args.get("raise_window") else ""))),
+    "set_value": _input(lambda backend, action, args, **_: (
+        json.dumps({"error": "set_value requires `value`"}) if args.get("value") is None
+        else backend.set_value(value=str(args["value"]), element=args.get("element")))),
+    "focus_app": _ActionSpec(lambda backend, action, args, **_: (
+        json.dumps({"error": "focus_app requires `app`"}) if not args.get("app")
+        else backend.focus_app(args["app"], raise_window=bool(args.get("raise_window")))), destructive=True,
+        summarize=lambda a, args, fg: f"focus {args.get('app', '')!r}" + (" (raise)" if args.get("raise_window") else "")),
     "capture": _ActionSpec(_do_capture),
     "wait": _ActionSpec(lambda backend, action, args, **_: _text_response(backend.wait(float(args.get("seconds", 1.0))))),
     "list_apps": _ActionSpec(partial(_do_listing, key="apps")),
@@ -460,8 +454,7 @@ def _bounds_unknown(bounds) -> bool:
     return False
 
 def _element_to_dict(e: UIElement) -> Dict[str, Any]:
-    # A zero rect is "geometry unknown", not a position — null it so no coordinate= is ever derived from it.
-    # The element index still works.
+    # A zero rect is "geometry unknown", not a position — null it so no coordinate= is ever derived from it (the index still works).
     return {"index": e.index, "role": e.role, "label": e.label[:_MAX_ELEMENT_LABEL_CHARS],
             "bounds": None if _bounds_unknown(e.bounds) else list(e.bounds), "app": e.app,
             **({"label_truncated": True} if len(e.label) > _MAX_ELEMENT_LABEL_CHARS else {})}

From 1fcb6f521d069e7d7d6154e9614809ac1c7a54a5 Mon Sep 17 00:00:00 2001
From: Teknium <127238744+teknium1@users.noreply.github.com>
Date: Wed, 2 Sep 2026 23:30:17 -0700
Subject: [PATCH 09/10] refactor(computer_use): atexit decorator,
 list-comprehension _format_elements, tighter signatures

---
 tools/computer_use/tool.py | 27 ++++++++++-----------------
 1 file changed, 10 insertions(+), 17 deletions(-)

diff --git a/tools/computer_use/tool.py b/tools/computer_use/tool.py
index 67efc387dd..803ae5c6c7 100644
--- a/tools/computer_use/tool.py
+++ b/tools/computer_use/tool.py
@@ -190,6 +190,7 @@ def release_computer_use_session(session_id: str) -> bool:
                   lambda e: logger.debug("computer_use backend release failed for session %s", sid, exc_info=True))
     return True
 
+@atexit.register
 def _shutdown_backend_atexit() -> None:
     """Stop all cached backends so cua-driver subprocesses don't outlive us. atexit only, no signal handlers: a
     ``SystemExit`` from a prompt_toolkit key binding corrupts its coroutine state and makes the process unkillable.
@@ -206,8 +207,6 @@ def _shutdown_backend_atexit() -> None:
     for backend, call_lock in unique.values():
         _stop_backend(backend, call_lock, lambda e: logger.debug("cua-driver atexit teardown failed: %s", e))
 
-atexit.register(_shutdown_backend_atexit)
-
 def reset_backend_for_tests() -> None:  # pragma: no cover — tear down the cached backend and per-session state
     _shutdown_backend_atexit()
     _AUX_VISION_ROUTE_CACHE.clear()
@@ -353,8 +352,7 @@ _ACTIONS: Dict[str, _ActionSpec] = {
                                                              f"{args.get('to_element') or args.get('to_coordinate')}{fg}")),
     "scroll": _input(_do_scroll, summarize=lambda a, args, fg: f"scroll {args.get('direction', '?')} x{args.get('amount', 3)}{fg}"),
     "type": _input(lambda backend, action, args, **delivery: backend.type_text(args.get("text", ""), **delivery),
-                   summarize=lambda a, args, fg: (f"type {args.get('text', '')[:60]!r}"
-                                                  + ("..." if len(args.get("text", "")) > 60 else "") + fg)),
+                   summarize=lambda a, args, fg: f"type {args.get('text', '')[:60]!r}" + ("..." if len(args.get("text", "")) > 60 else "") + fg),
     "key": _input(lambda backend, action, args, **delivery: backend.key(args.get("keys", ""), **delivery),
                   summarize=lambda a, args, fg: f"key {args.get('keys', '')!r}{fg}"),
     "set_value": _input(lambda backend, action, args, **_: (
@@ -460,13 +458,10 @@ def _element_to_dict(e: UIElement) -> Dict[str, Any]:
             **({"label_truncated": True} if len(e.label) > _MAX_ELEMENT_LABEL_CHARS else {})}
 
 def _format_elements(elements: List[UIElement], max_lines: int = 40) -> List[str]:
-    out: List[str] = []
-    for e in elements[:max_lines]:
-        where = "@ bounds-unknown (click by element index)" if _bounds_unknown(e.bounds) else f"@ {e.bounds}"
-        out.append(f"  #{e.index} {e.role} {e.label.replace(chr(10), ' ')[:60]!r} {where}" + (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
+    out = [f"  #{e.index} {e.role} {e.label.replace(chr(10), ' ')[:60]!r} "
+           + ("@ bounds-unknown (click by element index)" if _bounds_unknown(e.bounds) else f"@ {e.bounds}")
+           + (f" [{e.app}]" if e.app else "") for e in elements[:max_lines]]
+    return out + ([f"  ... +{len(elements) - max_lines} more (call capture with app= to narrow)"] if len(elements) > max_lines else [])
 
 def _bounds_hints(elements: List[UIElement], image_width: int, image_height: int
                   ) -> Tuple[Optional[float], Optional[str]]:
@@ -546,8 +541,7 @@ def _text_capture_payload(v: SimpleNamespace, summary: str, extra: Optional[Dict
 def _capture_response(cap: CaptureResult, max_elements: int = _DEFAULT_MAX_ELEMENTS) -> Any:
     v = _capture_view(cap, max_elements)
     lines = _capture_summary_lines(v)
-    summary = "\n".join(lines)  # multimodal/aux paths use this; text paths append notes and rebuild
-    extra = None
+    summary, extra = "\n".join(lines), None  # multimodal/aux paths use this; text paths append notes and rebuild
     if v.has_image:
         # Hand the screenshot to auxiliary.vision (text-only result) when the main model may not consume images
         # natively; returning the multimodal envelope unconditionally tripped HTTP 404/400 at the provider.
@@ -702,10 +696,9 @@ _VISION_PROMPT = ("Describe what is visible in this desktop application screensh
                   "about. Do not invent details that are not actually visible.\n\nAX/SOM index for "
                   "cross-reference:\n")
 
-def _route_capture_through_aux_vision(
-    cap: CaptureResult, summary: str, *, visible_elements: Optional[List[UIElement]] = None,
-    truncated_elements: int = 0, elements_file: Optional[str] = None, screenshot_path: Optional[str] = None,
-) -> Optional[str]:
+def _route_capture_through_aux_vision(cap: CaptureResult, summary: str, *, visible_elements: Optional[List[UIElement]] = None,
+                                      truncated_elements: int = 0, elements_file: Optional[str] = None,
+                                      screenshot_path: Optional[str] = None) -> Optional[str]:
     """Pre-analyse the capture via ``vision_analyze_tool`` (temp file under ``$HERMES_HOME/cache/vision/``) and merge
     the description with the AX/SOM summary into one text payload. JSON, or None on any failure."""
     if not cap.png_b64:

From de2c49b0d6360c98439dd45da031b22567cb2773 Mon Sep 17 00:00:00 2001
From: Teknium <127238744+teknium1@users.noreply.github.com>
Date: Wed, 2 Sep 2026 23:35:38 -0700
Subject: [PATCH 10/10] refactor(computer_use): compact section headers,
 wrapped literals, and one-shot walrus guards

---
 tools/computer_use/tool.py | 97 +++++++++++++-------------------------
 1 file changed, 34 insertions(+), 63 deletions(-)

diff --git a/tools/computer_use/tool.py b/tools/computer_use/tool.py
index 803ae5c6c7..c4410dced1 100644
--- a/tools/computer_use/tool.py
+++ b/tools/computer_use/tool.py
@@ -25,7 +25,6 @@ from tools.computer_use.backend import ActionResult, CaptureResult, ComputerUseB
 logger = logging.getLogger(__name__)
 
 # ── Approval & safety ───────────────────────────────────────────────────────
-
 _approval_callback = None
 
 def set_approval_callback(cb) -> None:
@@ -57,11 +56,9 @@ def _reject_unsafe(action: str, args: Dict[str, Any]) -> Optional[str]:
     if action == "type" and (pat := next((p.pattern for p in _BLOCKED_TYPE_PATTERNS if p.search(args.get("text", ""))), None)):
         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":
-        combo = _canon_key_combo(args.get("keys", ""))
-        if (blocked := next((b for b in _BLOCKED_KEY_COMBOS if b.issubset(combo)), None)) is not None:
-            return json.dumps({"error": f"blocked key combo: {sorted(blocked)}",
-                               "hint": "Destructive system shortcuts are hard-blocked."})
+    if action == "key" and (blocked := next((b for b in _BLOCKED_KEY_COMBOS
+                                             if b.issubset(_canon_key_combo(args.get("keys", "")))), None)) is not None:
+        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"})
@@ -75,7 +72,6 @@ def _input_target_mismatch(backend, requested_app: str) -> Optional[str]:
     return None if not current or not wanted or wanted in current or current in wanted else last_app
 
 # ── Backend selection — env-swappable for tests ─────────────────────────────
-
 # Per-Hermes-session cached backends (own cua-driver session, native target, refs, grant namespace).
 _backend_lock = threading.Lock()
 _backend: Optional[ComputerUseBackend] = None  # backward-compatible empty-session injection hook (older tests)
@@ -132,8 +128,7 @@ def _install_backend(sid: str, backend: ComputerUseBackend, permission_mode: str
     global _backend
     _backends[sid], _backend_permission_modes[sid] = backend, permission_mode
     _backend_call_locks[sid] = threading.RLock()
-    if sid == "":
-        _backend = backend
+    _backend = backend if sid == "" else _backend
     return backend
 
 def _detach_locked(sid: str) -> Tuple[Optional[ComputerUseBackend], Optional[threading.RLock]]:
@@ -147,8 +142,7 @@ def _detach_locked(sid: str) -> Tuple[Optional[ComputerUseBackend], Optional[thr
         _backend = None if _backend is backend else _backend
     return backend, call_lock
 
-def _stop_backend(backend: ComputerUseBackend, call_lock: Optional[threading.RLock],
-                  on_error: Callable[[Exception], None]) -> None:
+def _stop_backend(backend: ComputerUseBackend, call_lock: Optional[threading.RLock], on_error: Callable[[Exception], None]) -> None:
     """Stop under the session call lock (if any) so an in-flight action finishes first. Never called under
     ``_backend_lock`` (unrelated sessions stay free). ``on_error`` absorbs the failure (never raises)."""
     try:
@@ -223,8 +217,7 @@ class _NoopBackend(ComputerUseBackend):  # pragma: no cover
     """Test/CI stub (HERMES_COMPUTER_USE_BACKEND=noop). Records ``(name, kwargs)`` calls; returns trivial results."""
 
     def __init__(self) -> None: self.calls: List[Tuple[str, Dict[str, Any]]] = []
-    def start(self) -> None: pass
-    def stop(self) -> None: pass
+    start = stop = lambda self: None
     def is_available(self) -> bool: return True
 
     capture = _noop_stub("capture", "mode", "app", "pid", "window_id", result=lambda kw: CaptureResult(
@@ -235,7 +228,6 @@ class _NoopBackend(ComputerUseBackend):  # pragma: no cover
     focus_app = _noop_stub("focus_app", "app", "raise_window")
 
 # ── Dispatch ────────────────────────────────────────────────────────────────
-
 def handle_computer_use(args: Dict[str, Any], **kwargs) -> Any:
     """Main entry point (tools.registry): a JSON string (text-only) or a dict marked `_multimodal`. Order: hard
     blocks (_reject_unsafe) -> approval scopes (destructive action, then 'bring_to_front' — persistent focus is a
@@ -246,9 +238,8 @@ def handle_computer_use(args: Dict[str, Any], **kwargs) -> Any:
     session_id = str(kwargs.get("session_id") or "")  # approval-state / daemon-mode isolation key
     if (err := _reject_unsafe(action, args)) is not None:
         return err
-    scopes = [action] if action in _ACTIONS and _ACTIONS[action].destructive else []
-    if args.get("bring_to_front") or (action == "focus_app" and args.get("raise_window")):
-        scopes.append("bring_to_front")
+    scopes = ([action] if action in _ACTIONS and _ACTIONS[action].destructive else []) + (
+        ["bring_to_front"] if args.get("bring_to_front") or (action == "focus_app" and args.get("raise_window")) else [])
     for scope in scopes:
         if (err := _request_approval(scope, args, session_id)) is not None:
             return err
@@ -301,8 +292,7 @@ def _summarize_action(action: str, args: Dict[str, Any]) -> str:
 #     final str/dict result. `delivery` = delivery_mode + bring_to_front; only input actions use it.
 
 def _xy(args: Dict[str, Any]) -> Dict[str, Any]:
-    coord = args.get("coordinate")
-    return dict(x=coord[0], y=coord[1]) if coord and coord[0] is not None else dict(x=None, y=None)
+    return dict(x=coord[0], y=coord[1]) if (coord := args.get("coordinate")) and coord[0] is not None else dict(x=None, y=None)
 
 def _do_click(backend, action, args, button=None, count=1, **delivery):
     return backend.click(element=args.get("element"), **_xy(args), button=button or args.get("button") or "left",
@@ -321,8 +311,7 @@ def _do_scroll(backend, action, args, **delivery):
                           element=args.get("element"), **_xy(args), modifiers=args.get("modifiers"), **delivery)
 
 def _do_capture(backend, action, args, **_):
-    mode = str(args.get("mode", "som"))
-    if mode not in {"som", "vision", "ax"}:
+    if (mode := str(args.get("mode", "som"))) not in {"som", "vision", "ax"}:
         return json.dumps({"error": f"bad mode {mode!r}; use som|vision|ax"})
     # pid/window_id forwarded only when given so older backends keep their defaults.
     return _capture_response(backend.capture(mode=mode, app=args.get("app"),
@@ -370,8 +359,7 @@ _ACTIONS: Dict[str, _ActionSpec] = {
 # Native input actions deliver to the backend's sticky target; `app=` is NOT a targeting parameter (guard in _dispatch).
 _INPUT_ACTIONS = frozenset(a for a, s in _ACTIONS.items() if s.input)
 
-# Unknown actions are never aliased (no repairing bad model output), but the nearest real action is
-# named so a bare error isn't the only guidance.
+# Unknown actions are never aliased (no repairing bad model output), but the nearest real action is named as guidance.
 _ACTION_SUGGESTIONS = {
     "hotkey": "key", "press_key": "key", "keypress": "key", "key_combo": "key", "shortcut": "key", "type_text": "type",
     "input_text": "type", "screenshot": "capture", "get_window_state": "capture", "left_click": "click", "mouse_click": "click",
@@ -380,9 +368,8 @@ _ACTION_SUGGESTIONS = {
 def _dispatch(backend: ComputerUseBackend, action: str, args: Dict[str, Any]) -> Any:
     spec = _ACTIONS.get(action)
     if spec is None:
-        hint = _ACTION_SUGGESTIONS.get(str(action))
-        return json.dumps({"error": f"unknown action {action!r}"
-                           + (f" — did you mean {hint!r}? See the action enum in the tool schema." if hint else "")})
+        return json.dumps({"error": f"unknown action {action!r}" + (f" — did you mean {hint!r}? See the action enum in the tool schema."
+                                                                 if (hint := _ACTION_SUGGESTIONS.get(str(action))) else "")})
     # app= guard: input goes to the sticky target from the last capture/focus_app and the backend drops app=
     # silently — refuse a clear mismatch rather than type into the wrong window while reporting ok:true.
     if (spec.input and isinstance(requested_app := args.get("app"), str) and requested_app.strip()
@@ -397,23 +384,20 @@ def _dispatch(backend: ComputerUseBackend, action: str, args: Dict[str, Any]) ->
     return res if isinstance(res, (str, dict)) else _maybe_follow_capture(backend, res, bool(args.get("capture_after")))
 
 # ── Response shaping ────────────────────────────────────────────────────────
-
 def _classify_action_result(res: ActionResult) -> Dict[str, Any]:
     """Next ladder step from semantic evidence, in precedence order. Escalation is advisory: it never overrides
     a confirmed effect nor licenses repeating input."""
     if res.effect == "confirmed" or res.verified is True:
         return {"decision": "done"}
     if res.effect == "unverifiable":
-        return {"decision": "verify_fresh_state", "hint": (
-            "Input was delivered but not confirmed. Re-capture and check the result BEFORE any "
-            "retry — do not repeat the input on an escalation recommendation alone.")}
+        return {"decision": "verify_fresh_state", "hint": ("Input was delivered but not confirmed. Re-capture and check the "
+                "result BEFORE any retry — do not repeat the input on an escalation recommendation alone.")}
     if res.effect == "suspected_noop" or not res.ok or res.code is not None:
         return {"decision": "escalate", **({"recommended": res.escalation.get("recommended")}
                                            if isinstance(res.escalation, dict) else {}), "hint": (
-            "The input likely did not land. Climb one rung following `recommended`: 'px' → "
-            "re-issue by coordinate; 'foreground' (or a failed pixel click) → re-issue with "
-            "delivery_mode='foreground' (separate approval). Do not predict the rung from the "
-            "app being Electron/Chromium — react to this signal.")}
+            "The input likely did not land. Climb one rung following `recommended`: 'px' → re-issue by coordinate; "
+            "'foreground' (or a failed pixel click) → re-issue with delivery_mode='foreground' (separate approval). "
+            "Do not predict the rung from the app being Electron/Chromium — react to this signal.")}
     return {"decision": "verify_fresh_state",  # transport success without semantic proof is not proof of effect
             "hint": "Transport succeeded but the effect is unproven. Re-capture and confirm before continuing."}
 
@@ -463,8 +447,7 @@ def _format_elements(elements: List[UIElement], max_lines: int = 40) -> List[str
            + (f" [{e.app}]" if e.app else "") for e in elements[:max_lines]]
     return out + ([f"  ... +{len(elements) - max_lines} more (call capture with app= to narrow)"] if len(elements) > max_lines else [])
 
-def _bounds_hints(elements: List[UIElement], image_width: int, image_height: int
-                  ) -> Tuple[Optional[float], Optional[str]]:
+def _bounds_hints(elements: List[UIElement], image_width: int, image_height: int) -> Tuple[Optional[float], Optional[str]]:
     """(scale, note) when element bounds live in a different coordinate space than the screenshot, else (None, None).
     On HiDPI displays AX bounds are native while the screenshot is downscaled, so coordinate= clicks read off the
     screenshot miss by the scale factor. 5% slack: window chrome can hang a few px past the captured frame without
@@ -483,10 +466,8 @@ def _bounds_hints(elements: List[UIElement], image_width: int, image_height: int
             "space — derive click points from element bounds, or scale screenshot positions up accordingly")
     return round(max(max_x / image_width, max_y / image_height), 2), note
 
-def _bounds_scale(elements: List[UIElement], image_width: int, image_height: int) -> Optional[float]:
-    return _bounds_hints(elements, image_width, image_height)[0]
-def _bounds_space_note(elements: List[UIElement], image_width: int, image_height: int) -> Optional[str]:
-    return _bounds_hints(elements, image_width, image_height)[1]
+_bounds_scale = lambda elements, image_width, image_height: _bounds_hints(elements, image_width, image_height)[0]  # noqa: E731
+_bounds_space_note = lambda elements, image_width, image_height: _bounds_hints(elements, image_width, image_height)[1]  # noqa: E731
 
 def _capture_view(cap: CaptureResult, max_elements: int) -> SimpleNamespace:
     """One capture's derived facts, computed once for every response branch: ``visible`` is the capped element list,
@@ -534,8 +515,8 @@ def _text_capture_payload(v: SimpleNamespace, summary: str, extra: Optional[Dict
         "mode": v.cap.mode, "width": v.width, "height": v.height, "app": v.cap.app, "window_title": v.cap.window_title,
         "elements": [_element_to_dict(e) for e in v.visible], "total_elements": v.total, "summary": summary,
         **(extra or {}),
-        **_present(truncated_elements=v.truncated, elements_file=v.elements_file,
-                   screenshot_path=v.screenshot_path, bounds_scale=v.bounds_scale),
+        **_present(truncated_elements=v.truncated, elements_file=v.elements_file, screenshot_path=v.screenshot_path,
+                   bounds_scale=v.bounds_scale),
     })
 
 def _capture_response(cap: CaptureResult, max_elements: int = _DEFAULT_MAX_ELEMENTS) -> Any:
@@ -551,9 +532,8 @@ def _capture_response(cap: CaptureResult, max_elements: int = _DEFAULT_MAX_ELEME
                 "content": [{"type": "text", "text": summary},
                             {"type": "image_url", "image_url": {"url": f"data:{_capture_image_format(cap)[0]};base64,{cap.png_b64}"}}],
                 "text_summary": summary,
-                "meta": {"mode": cap.mode, "width": v.width, "height": v.height, "elements": v.total,
-                         "png_bytes": cap.png_bytes_len, **_present(screenshot_path=v.screenshot_path,
-                                                                  elements_file=v.elements_file, bounds_scale=v.bounds_scale)},
+                "meta": {"mode": cap.mode, "width": v.width, "height": v.height, "elements": v.total, "png_bytes": cap.png_bytes_len,
+                         **_present(screenshot_path=v.screenshot_path, elements_file=v.elements_file, bounds_scale=v.bounds_scale)},
             }
         routed = _route_capture_through_aux_vision(
             cap, summary, visible_elements=v.visible, truncated_elements=v.truncated,
@@ -577,8 +557,7 @@ def _maybe_follow_capture(backend: ComputerUseBackend, res: ActionResult, do_cap
     try:
         # Recapture the exact window when known: on Linux several unrelated windows may share an app name, so
         # app-only recapture can switch targets.
-        target = getattr(backend, "_last_target", None) or {}
-        exact = {k: target.get(k) for k in ("pid", "window_id")}
+        exact = {k: (getattr(backend, "_last_target", None) or {}).get(k) for k in ("pid", "window_id")}
         cap = backend.capture(mode=_capture_after_mode(), **(exact if None not in exact.values()
                                                             else {"app": getattr(backend, "_last_app", None)}))
     except Exception as e:
@@ -593,7 +572,6 @@ def _maybe_follow_capture(backend: ComputerUseBackend, res: ActionResult, do_cap
     return json.dumps({**json.loads(resp), **payload})  # text capture: merge the action payload in
 
 # ── Cache files (screenshots, element spills, vision temps) ─────────────────
-
 def _cache_file(subdir: str, legacy: str, name: str, pattern: str = "", cap: int = 0):
     """Path for a new file under ``$HERMES_HOME/<subdir>`` (dir created). With ``pattern``/``cap``, first unlinks the
     oldest matching files so at most ``cap - 1`` remain (best-effort)."""
@@ -606,8 +584,7 @@ def _cache_file(subdir: str, legacy: str, name: str, pattern: str = "", cap: int
             stale.unlink(missing_ok=True)
     return cache_dir / name
 
-def _write_cache_file(what: str, subdir: str, legacy: str, name: str, pattern: str, cap: int,
-                      write: Callable[[Any], None]) -> Optional[str]:
+def _write_cache_file(what: str, subdir: str, legacy: str, name: str, pattern: str, cap: int, write: Callable[[Any], None]) -> Optional[str]:
     """Bounded cache write via ``write(path)``; the path, or None on any failure — an unwritable cache must never
     break control (a capture keeps working without its spill/screenshot copy)."""
     try:
@@ -634,7 +611,6 @@ def _spill_elements_to_file(cap: CaptureResult) -> Optional[str]:
                              lambda p: p.write_text(json.dumps(payload, ensure_ascii=False, indent=1), encoding="utf-8"))
 
 # ── auxiliary.vision routing for captured screenshots ───────────────────────
-
 _MAX_VISION_DIM = 1456  # 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 vision latency.
 
@@ -653,12 +629,9 @@ def _shrink_capture_for_vision(raw: bytes, ext: str, max_dim: int = _MAX_VISION_
         new_w, new_h = img.size
         img.save(out, format="JPEG" if ext == ".jpg" else "PNG")
         fx, fy = (orig_w / new_w if new_w else 1.0), (orig_h / new_h if new_h else 1.0)
-        factor_clause = (f"multiply any coordinates you report by {fx:.2f} to map back to the real screen."
-                         if f"{fx:.2f}" == f"{fy:.2f}" else
-                         f"multiply any x coordinates you report by {fx:.2f} and "
-                         f"any y coordinates by {fy:.2f} to map back to the real screen.")
-        return out.getvalue(), (f"Screenshot downscaled from {orig_w}x{orig_h} to "
-                                f"{new_w}x{new_h} for vision; {factor_clause}")
+        factor_clause = (f"multiply any coordinates you report by {fx:.2f} to map back to the real screen." if f"{fx:.2f}" == f"{fy:.2f}"
+                         else f"multiply any x coordinates you report by {fx:.2f} and any y coordinates by {fy:.2f} to map back to the real screen.")
+        return out.getvalue(), f"Screenshot downscaled from {orig_w}x{orig_h} to {new_w}x{new_h} for vision; {factor_clause}"
     except Exception as exc:
         logger.debug("computer_use: vision downscale skipped: %s", exc)
         return raw, None
@@ -690,11 +663,10 @@ def _capture_after_mode() -> str:
         return mode if (mode := mode.strip().lower()) in {"som", "vision", "ax"} else "som"
     return "som"
 
-_VISION_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\nAX/SOM index for "
-                  "cross-reference:\n")
+_VISION_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\nAX/SOM index for cross-reference:\n")
 
 def _route_capture_through_aux_vision(cap: CaptureResult, summary: str, *, visible_elements: Optional[List[UIElement]] = None,
                                       truncated_elements: int = 0, elements_file: Optional[str] = None,
@@ -743,7 +715,6 @@ def _route_capture_through_aux_vision(cap: CaptureResult, summary: str, *, visib
                                                  "vision_analysis_routed_via": "auxiliary.vision"})
 
 # ── Availability check (used by the tool registry check_fn) ─────────────────
-
 def check_computer_use_requirements() -> bool:
     """macOS/Windows/Linux + cua-driver binary (or env override). `hermes computer-use doctor` names blocked checks."""
     if sys.platform not in ("darwin", "win32", "linux"):