perf(computer_use): cap capture size and cache vision routing

Cut steady-state Computer Use latency without changing default behavior
or waiting on cua-driver:

- Cap screenshots via set_config(max_image_dimension) on session start
  (config: computer_use.max_image_dimension, default 1456)
- Cache aux-vision routing per (provider, model) so captures skip
  repeated load_config()
- Add computer_use.capture_after_mode (default som) so users can opt
  follow-ups down to ax (elements only) for speed
This commit is contained in:
Brooklyn Nicholson 2026-07-23 00:48:20 -05:00
parent de5ece9944
commit 12ad13ddca
4 changed files with 141 additions and 19 deletions

View file

@ -3474,6 +3474,12 @@ DEFAULT_CONFIG = {
# every invocation (MCP backend, status, doctor, install). Set true
# to let cua-driver use its own default (telemetry on).
"cua_telemetry": False,
# Cap driver screenshot longest edge (pixels) via set_config on
# session start. Shrinks SOM multimodal payloads; 0 disables.
"max_image_dimension": 1456,
# Mode for capture_after follow-ups: som (screenshot + overlays —
# default), ax (elements only, no PNG — faster), vision (pixels only).
"capture_after_mode": "som",
},
# Hermes Desktop (Electron app) launch options. These only affect

View file

@ -0,0 +1,66 @@
"""Behavior contracts for computer_use latency knobs."""
from unittest.mock import patch
from tools.computer_use import cua_backend
from tools.computer_use import tool as cu_tool
def test_max_image_dimension_default():
with patch("hermes_cli.config.load_config", return_value={}):
assert cua_backend._computer_use_max_image_dimension() == 1456
def test_max_image_dimension_zero_disables():
with patch(
"hermes_cli.config.load_config",
return_value={"computer_use": {"max_image_dimension": 0}},
):
assert cua_backend._computer_use_max_image_dimension() is None
def test_capture_after_mode_default_som():
with patch("hermes_cli.config.load_config", return_value={}):
assert cu_tool._capture_after_mode() == "som"
def test_capture_after_mode_ax_override():
with patch(
"hermes_cli.config.load_config",
return_value={"computer_use": {"capture_after_mode": "ax"}},
):
assert cu_tool._capture_after_mode() == "ax"
def test_capture_after_mode_invalid_falls_back_to_som():
with patch(
"hermes_cli.config.load_config",
return_value={"computer_use": {"capture_after_mode": "bogus"}},
):
assert cu_tool._capture_after_mode() == "som"
def test_aux_vision_route_caches_per_provider_model(monkeypatch):
cu_tool._AUX_VISION_ROUTE_CACHE.clear()
calls = {"n": 0}
monkeypatch.setattr(
"agent.auxiliary_client._read_main_provider", lambda: "openai"
)
monkeypatch.setattr(
"agent.auxiliary_client._read_main_model", lambda: "gpt-test"
)
def fake_load():
calls["n"] += 1
return {"auxiliary": {"vision": {}}}
monkeypatch.setattr("hermes_cli.config.load_config", fake_load)
monkeypatch.setattr(
"tools.computer_use.vision_routing.should_route_capture_to_aux_vision",
lambda *a, **k: True,
)
assert cu_tool._should_route_through_aux_vision() is True
assert cu_tool._should_route_through_aux_vision() is True
assert calls["n"] == 1

View file

@ -171,23 +171,37 @@ _DESKTOP_WINDOW_NAMES = (
_CUA_TELEMETRY_ENV_VAR = "CUA_DRIVER_RS_TELEMETRY_ENABLED"
def _cua_telemetry_disabled() -> bool:
"""True when Hermes should disable cua-driver telemetry for this user.
Reads ``computer_use.cua_telemetry`` from config.yaml. Default is False
(telemetry off). Any failure to read config fails SAFE toward the
privacy-preserving default of telemetry disabled.
"""
def _computer_use_cfg() -> Dict[str, Any]:
"""The ``computer_use`` config block, or ``{}`` when config is unreadable."""
try:
from hermes_cli.config import load_config
cfg = load_config() or {}
cu = cfg.get("computer_use") or {}
# opt-in flag: True => user wants telemetry => do NOT disable.
return not bool(cu.get("cua_telemetry", False))
return (load_config() or {}).get("computer_use") or {}
except Exception:
# Config unreadable — default to disabling telemetry (fail safe).
return True
return {}
def _cua_telemetry_disabled() -> bool:
"""True when Hermes should disable cua-driver telemetry for this user.
Reads ``computer_use.cua_telemetry`` (default False telemetry off).
Unreadable config falls SAFE toward disabling telemetry.
"""
# opt-in flag: True => user wants telemetry => do NOT disable.
return not bool(_computer_use_cfg().get("cua_telemetry", False))
def _computer_use_max_image_dimension() -> Optional[int]:
"""Longest-edge cap for cua-driver screenshots, or None to leave unset.
Reads ``computer_use.max_image_dimension`` (default 1456, matching the
aux-vision downscale). ``0`` / negative / non-numeric None.
"""
try:
dim = int(_computer_use_cfg().get("max_image_dimension", 1456))
except (TypeError, ValueError):
return 1456
return dim if dim > 0 else None
def cua_driver_child_env(base_env: Optional[Dict[str, str]] = None) -> Dict[str, str]:
@ -1418,6 +1432,18 @@ class CuaDriverBackend(ComputerUseBackend):
except Exception as e:
logger.debug("cua-driver start_session failed (continuing anonymous): %s", e)
# Cap screenshot size early so every later get_window_state / SOM
# capture pays less over the daemon socket and in the model turn.
# Only when the session handshake actually flipped `_started` —
# otherwise call_tool would re-enter session.start() (see
# _LIFECYCLE_CALLS) and tests that stub start() would recurse.
max_dim = _computer_use_max_image_dimension()
if max_dim and self._session._started:
try:
self.set_config(max_image_dimension=max_dim)
except Exception as e:
logger.debug("cua-driver set_config(max_image_dimension) failed: %s", e)
def stop(self) -> None:
# Tear the cua-driver session down before disconnecting so the
# driver can clean up per-session state (cursor overlay, recording

View file

@ -138,6 +138,8 @@ def _is_blocked_type(text: str) -> Optional[str]:
# Per-process cached backend; lazily instantiated on first call.
_backend_lock = threading.Lock()
# Process-scoped aux-vision routing cache: (provider, model) → bool.
_AUX_VISION_ROUTE_CACHE: Dict[Tuple[str, str], bool] = {}
_backend: Optional[ComputerUseBackend] = None
# Approval state, scoped per conversation/run (keyed by session_id) so a
# gateway serving concurrent sessions can't leak one run's "always approve"
@ -185,6 +187,7 @@ def reset_backend_for_tests() -> None: # pragma: no cover
except Exception:
pass
_backend = None
_AUX_VISION_ROUTE_CACHE.clear()
with _approval_lock:
_session_auto_approve.clear()
_always_allow.clear()
@ -792,17 +795,37 @@ def _should_route_through_aux_vision() -> bool:
logger.debug("computer_use: aux-vision routing import failed: %s", exc)
return False
try:
provider = _read_main_provider()
model = _read_main_model()
cfg = load_config()
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:
return bool(should_route_capture_to_aux_vision(provider, model, cfg))
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(
@ -927,10 +950,11 @@ def _maybe_follow_capture(
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="som", pid=pid, window_id=window_id)
cap = backend.capture(mode=mode, pid=pid, window_id=window_id)
else:
cap = backend.capture(mode="som", app=getattr(backend, "_last_app", None))
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)