diff --git a/tests/tools/test_computer_use_capture_routing.py b/tests/tools/test_computer_use_capture_routing.py new file mode 100644 index 0000000000000..44084fabbea67 --- /dev/null +++ b/tests/tools/test_computer_use_capture_routing.py @@ -0,0 +1,431 @@ +"""End-to-end regression for #24015 — capture routing via auxiliary.vision. + +When ``computer_use(action='capture', mode='som'|'vision')`` returns a +screenshot, ``_capture_response`` previously always returned a +``_multimodal`` envelope. For non-vision main models, or when the user +explicitly configured ``auxiliary.vision`` in ``config.yaml``, that +envelope tripped HTTP 404 / 400 at the provider boundary even though a +perfectly good vision backend was sitting in config waiting to be used. + +This file exercises the integrated ``_capture_response`` flow with +deterministic stubs for: + +* ``should_route_capture_to_aux_vision`` (the policy decision) +* ``_run_async`` (sync->async bridge) +* ``vision_analyze_tool`` (the aux LLM call) +* ``hermes_constants.get_hermes_dir`` (cache path) + +…so the full code path is covered without a live cua-driver, a real +auxiliary client, or network access. +""" + +from __future__ import annotations + +import base64 +import json +import os +from pathlib import Path +from typing import Any +from unittest.mock import MagicMock, patch + +import pytest + + +# --------------------------------------------------------------------------- +# Fixtures / helpers +# --------------------------------------------------------------------------- + +# 1×1 PNG (transparent) — minimal bytes that decode cleanly. +_PNG_B64 = ( + "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42m" + "NkYAAAAAYAAjCB0C8AAAAASUVORK5CYII=" +) + +# 1×1 JPEG — used to verify mime detection works for either stream type. +_JPEG_B64 = ( + "/9j/4AAQSkZJRgABAQEAYABgAAD/2wBDAAEBAQEBAQEBAQEBAQEBAQEBAQEB" + "AQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQH/" +) + + +@pytest.fixture +def tmp_cache_dir(tmp_path): + """Override get_hermes_dir so cache writes land under tmp_path.""" + cache_dir = tmp_path / "cache_vision" + cache_dir.mkdir() + + def _fake_get(*_args, **_kw): + return cache_dir + + with patch("hermes_constants.get_hermes_dir", _fake_get): + yield cache_dir + + +def _make_capture( + *, + png_b64: str = _PNG_B64, + mode: str = "som", + elements=None, + app: str = "Safari", + window_title: str = "GitHub – Issue #24015", + width: int = 1280, + height: int = 800, +): + from tools.computer_use.backend import CaptureResult, UIElement + + elements = list(elements or [ + UIElement(index=0, role="AXButton", label="Sign in", + bounds=(10, 20, 80, 30)), + UIElement(index=1, role="AXTextField", label="username", + bounds=(10, 60, 200, 24)), + ]) + raw = base64.b64decode(png_b64, validate=False) + return CaptureResult( + mode=mode, + width=width, + height=height, + png_b64=png_b64, + elements=elements, + app=app, + window_title=window_title, + png_bytes_len=len(raw), + ) + + +def _stub_aux_analysis(text: str): + """Return a fake vision_analyze_tool coroutine result (JSON envelope).""" + return json.dumps({"success": True, "analysis": text}) + + +# --------------------------------------------------------------------------- +# _capture_response: routing OFF (current/native behaviour) +# --------------------------------------------------------------------------- + +class TestCaptureResponseDefaultPath: + """When routing helper says 'native', the existing multimodal envelope wins.""" + + def test_som_capture_returns_multimodal_envelope_when_native(self): + from tools.computer_use import tool as cu_tool + + cap = _make_capture(png_b64=_PNG_B64, mode="som") + with patch.object(cu_tool, "_should_route_through_aux_vision", + return_value=False): + resp = cu_tool._capture_response(cap) + + assert isinstance(resp, dict) + assert resp.get("_multimodal") is True + # Image part must use image/png MIME for a PNG payload. + image_part = next( + p for p in resp["content"] if p.get("type") == "image_url" + ) + url = image_part["image_url"]["url"] + assert url.startswith("data:image/png;base64,") + assert "vision_analysis" not in resp + + def test_jpeg_capture_returns_image_jpeg_mime_when_native(self): + from tools.computer_use import tool as cu_tool + + cap = _make_capture(png_b64=_JPEG_B64, mode="som") + with patch.object(cu_tool, "_should_route_through_aux_vision", + return_value=False): + resp = cu_tool._capture_response(cap) + + url = next(p for p in resp["content"] if p.get("type") == "image_url") + assert url["image_url"]["url"].startswith("data:image/jpeg;base64,") + + def test_ax_only_capture_returns_text_regardless_of_routing(self): + from tools.computer_use import tool as cu_tool + + cap = _make_capture(mode="ax", png_b64="") + # ax mode never has a PNG so neither path matters; assert pure text. + with patch.object(cu_tool, "_should_route_through_aux_vision", + return_value=True) as routing: + resp = cu_tool._capture_response(cap) + + # ax never even consults the routing helper — short-circuited above + # the image branch. + routing.assert_not_called() + assert isinstance(resp, str) + body = json.loads(resp) + assert body["mode"] == "ax" + + +# --------------------------------------------------------------------------- +# _capture_response: routing ON (the #24015 fix) +# --------------------------------------------------------------------------- + +class TestCaptureResponseRoutedToAuxVision: + """When routing helper says 'aux', the PNG is pre-analysed and a text + response is returned with no image_url parts at all.""" + + def test_som_capture_returns_text_with_vision_analysis( + self, tmp_cache_dir, + ): + from tools.computer_use import tool as cu_tool + + cap = _make_capture(mode="som") + + captured_calls = {} + + def _fake_run_async(coro): + captured_calls["called"] = True + return _stub_aux_analysis( + "A Safari window showing a GitHub issue page with a 'Sign " + "in' button and a 'username' text field." + ) + + # vision_analyze_tool is async; force a sync MagicMock so we can + # assert positional args without dealing with awaitables. + fake_vat = MagicMock(return_value="") + + with patch.object(cu_tool, "_should_route_through_aux_vision", + return_value=True), \ + patch("model_tools._run_async", side_effect=_fake_run_async), \ + patch("tools.vision_tools.vision_analyze_tool", + new_callable=lambda: fake_vat): + resp = cu_tool._capture_response(cap) + + # Must be a JSON string, NOT a multimodal envelope. This is exactly + # the contract that prevents #24015's HTTP 404 from firing on the + # next agent turn. + assert isinstance(resp, str) + body = json.loads(resp) + assert body["mode"] == "som" + assert body["app"] == "Safari" + assert "Sign in" in body["vision_analysis"] + assert body["vision_analysis_routed_via"] == "auxiliary.vision" + # The original AX-only metadata (window title, element index, app) + # is preserved alongside the new vision analysis so the agent loses + # no context vs the multimodal path. + assert body["window_title"] == "GitHub – Issue #24015" + assert len(body["elements"]) == 2 + + assert captured_calls.get("called") is True + # vision_analyze_tool was invoked with a path under the patched cache + # and a non-empty prompt. + args, _kwargs = fake_vat.call_args + path_arg, prompt_arg = args[0], args[1] + assert str(tmp_cache_dir) in path_arg + assert "macOS application screenshot" in prompt_arg + # AX summary is included so the aux model can ground its description + # against the same set-of-mark index the agent will see. + assert "Sign in" in prompt_arg + + def test_temp_screenshot_file_is_cleaned_up_after_routing( + self, tmp_cache_dir, + ): + from tools.computer_use import tool as cu_tool + + cap = _make_capture(mode="som") + # We capture the path the aux call sees so we can assert it's gone + # after _capture_response returns. + observed_path = {} + + def _fake_run_async(_coro): + return _stub_aux_analysis("description goes here") + + def _fake_vat(image_path, _prompt): + observed_path["path"] = image_path + # File must exist while aux is being arranged. + assert os.path.exists(image_path) + return "" + + fake_vat = MagicMock(side_effect=_fake_vat) + + with patch.object(cu_tool, "_should_route_through_aux_vision", + return_value=True), \ + patch("model_tools._run_async", side_effect=_fake_run_async), \ + patch("tools.vision_tools.vision_analyze_tool", + new_callable=lambda: fake_vat): + cu_tool._capture_response(cap) + + # File must be unlinked after _capture_response returns. + assert observed_path["path"] + assert not os.path.exists(observed_path["path"]) + + def test_temp_file_cleaned_up_even_when_aux_call_raises( + self, tmp_cache_dir, + ): + from tools.computer_use import tool as cu_tool + + cap = _make_capture(mode="som") + observed_path = {} + + def _fake_vat(image_path, _prompt): + observed_path["path"] = image_path + return "" + + def _fake_run_async(_coro): + raise RuntimeError("aux LLM down") + + fake_vat = MagicMock(side_effect=_fake_vat) + + with patch.object(cu_tool, "_should_route_through_aux_vision", + return_value=True), \ + patch("model_tools._run_async", side_effect=_fake_run_async), \ + patch("tools.vision_tools.vision_analyze_tool", + new_callable=lambda: fake_vat): + resp = cu_tool._capture_response(cap) + + # Aux failure → fall back to multimodal envelope (so the user still + # gets *something* useful even if vision is broken). + assert isinstance(resp, dict) + assert resp.get("_multimodal") is True + # Temp file must still be cleaned up. + assert observed_path["path"] + assert not os.path.exists(observed_path["path"]) + + def test_empty_aux_analysis_falls_back_to_multimodal(self, tmp_cache_dir): + from tools.computer_use import tool as cu_tool + + cap = _make_capture(mode="som") + + def _fake_run_async(_coro): + return _stub_aux_analysis("") + + fake_vat = MagicMock(return_value="") + + with patch.object(cu_tool, "_should_route_through_aux_vision", + return_value=True), \ + patch("model_tools._run_async", side_effect=_fake_run_async), \ + patch("tools.vision_tools.vision_analyze_tool", + new_callable=lambda: fake_vat): + resp = cu_tool._capture_response(cap) + + # Empty analysis is treated as failure — we'd rather show pixels + # than embed an empty 'vision_analysis' string into the result. + assert isinstance(resp, dict) + assert resp.get("_multimodal") is True + + def test_invalid_aux_response_falls_back_to_multimodal(self, tmp_cache_dir): + from tools.computer_use import tool as cu_tool + + cap = _make_capture(mode="som") + + def _fake_run_async(_coro): + return 1234 # not a string at all + + fake_vat = MagicMock(return_value="") + + with patch.object(cu_tool, "_should_route_through_aux_vision", + return_value=True), \ + patch("model_tools._run_async", side_effect=_fake_run_async), \ + patch("tools.vision_tools.vision_analyze_tool", + new_callable=lambda: fake_vat): + resp = cu_tool._capture_response(cap) + + assert isinstance(resp, dict) + assert resp.get("_multimodal") is True + + +# --------------------------------------------------------------------------- +# _should_route_through_aux_vision: end-to-end with real config plumbing +# --------------------------------------------------------------------------- + +class TestRoutingDecisionWiring: + """Verify _should_route_through_aux_vision wires the right config + helper.""" + + def test_explicit_aux_vision_in_config_routes_to_aux(self): + from tools.computer_use import tool as cu_tool + + cfg = { + "model": {"default": "tencent/hy3-preview", "provider": "openrouter"}, + "auxiliary": { + "vision": { + "provider": "openrouter", + "model": "google/gemini-2.5-flash", + } + }, + } + with patch("agent.auxiliary_client._read_main_provider", + return_value="openrouter"), \ + patch("agent.auxiliary_client._read_main_model", + return_value="tencent/hy3-preview"), \ + patch("hermes_cli.config.load_config", return_value=cfg): + assert cu_tool._should_route_through_aux_vision() is True + + def test_no_explicit_aux_and_vision_capable_main_keeps_multimodal(self): + from tools.computer_use import tool as cu_tool + + cfg = { + "model": {"default": "claude-opus-4-5", "provider": "anthropic"}, + } + with patch("agent.auxiliary_client._read_main_provider", + return_value="anthropic"), \ + patch("agent.auxiliary_client._read_main_model", + return_value="claude-opus-4-5"), \ + patch("hermes_cli.config.load_config", return_value=cfg), \ + patch("tools.computer_use.vision_routing._lookup_supports_vision", + return_value=True), \ + patch("tools.computer_use.vision_routing." + "_provider_accepts_multimodal_tool_result", + return_value=True): + assert cu_tool._should_route_through_aux_vision() is False + + def test_config_load_failure_disables_routing_safely(self): + from tools.computer_use import tool as cu_tool + + with patch("hermes_cli.config.load_config", + side_effect=RuntimeError("config.yaml unreadable")): + # No exception should bubble up — fail open by returning False + # so the legacy multimodal envelope continues to work. + assert cu_tool._should_route_through_aux_vision() is False + + def test_helper_decision_exception_is_swallowed(self): + from tools.computer_use import tool as cu_tool + from tools.computer_use import vision_routing as vr_mod + + with patch("agent.auxiliary_client._read_main_provider", + return_value="openrouter"), \ + patch("agent.auxiliary_client._read_main_model", + return_value="x"), \ + patch("hermes_cli.config.load_config", return_value={}), \ + patch.object(vr_mod, "should_route_capture_to_aux_vision", + side_effect=ValueError("policy bug")): + assert cu_tool._should_route_through_aux_vision() is False + + +# --------------------------------------------------------------------------- +# Bug reproduction marker — proves the fix is needed. +# --------------------------------------------------------------------------- + +class TestBugReproductionAnchor: + """Without the fix, this test would assert the wrong thing. + + On upstream/main HEAD prior to this branch, _capture_response returns a + multimodal envelope unconditionally — so when a non-vision main model + is configured, the captured PNG is delivered to the main provider as + image_url content and the request is rejected with HTTP 404. We don't + have a live provider here, but we can pin the contract: with routing + enabled the response MUST be a JSON string with no image_url parts. + """ + + def test_non_vision_main_model_never_returns_image_url_when_routed( + self, tmp_cache_dir, + ): + from tools.computer_use import tool as cu_tool + + cap = _make_capture(mode="som") + + def _fake_run_async(_coro): + return _stub_aux_analysis( + "Screenshot showing a GitHub.com window with a sign-in " + "form." + ) + + fake_vat = MagicMock(return_value="") + + with patch.object(cu_tool, "_should_route_through_aux_vision", + return_value=True), \ + patch("model_tools._run_async", side_effect=_fake_run_async), \ + patch("tools.vision_tools.vision_analyze_tool", + new_callable=lambda: fake_vat): + resp = cu_tool._capture_response(cap) + + # Must be a string (text-only result). + assert isinstance(resp, str) + # Must NOT contain a base64 image URL anywhere — that's what tripped + # 'No endpoints found that support image input' on the reporter's + # main provider in #24015. + assert "data:image" not in resp + assert "image_url" not in resp diff --git a/tests/tools/test_computer_use_vision_routing.py b/tests/tools/test_computer_use_vision_routing.py new file mode 100644 index 0000000000000..b0ae4566994e5 --- /dev/null +++ b/tests/tools/test_computer_use_vision_routing.py @@ -0,0 +1,260 @@ +"""Unit tests for tools.computer_use.vision_routing. + +Cover the small ``should_route_capture_to_aux_vision`` policy helper that +decides whether a captured screenshot from ``computer_use(action='capture')`` +should be returned as a multimodal envelope (main model handles vision +natively) or pre-analysed via the ``auxiliary.vision`` pipeline so the +main model only sees text. + +The companion end-to-end regression for #24015 lives in +``tests/tools/test_computer_use_capture_routing.py``; this file pins the +unit contract of the helper in isolation so behaviour does not regress +silently if the surrounding ``computer_use`` plumbing is refactored. +""" + +from __future__ import annotations + +from unittest.mock import patch + +import pytest + + +# --------------------------------------------------------------------------- +# _explicit_aux_vision_override +# --------------------------------------------------------------------------- + +class TestExplicitAuxVisionOverride: + """Mirror agent.image_routing — config detection must agree across paths.""" + + def test_returns_false_for_none_cfg(self): + from tools.computer_use.vision_routing import _explicit_aux_vision_override + assert _explicit_aux_vision_override(None) is False + + def test_returns_false_for_non_dict_cfg(self): + from tools.computer_use.vision_routing import _explicit_aux_vision_override + assert _explicit_aux_vision_override("not-a-dict") is False + assert _explicit_aux_vision_override([]) is False + + def test_returns_false_when_auxiliary_block_missing(self): + from tools.computer_use.vision_routing import _explicit_aux_vision_override + assert _explicit_aux_vision_override({}) is False + assert _explicit_aux_vision_override({"model": {"default": "x"}}) is False + + def test_returns_false_when_vision_block_missing(self): + from tools.computer_use.vision_routing import _explicit_aux_vision_override + cfg = {"auxiliary": {"compression": {"provider": "openai"}}} + assert _explicit_aux_vision_override(cfg) is False + + def test_returns_false_for_blank_provider_no_model_no_base_url(self): + from tools.computer_use.vision_routing import _explicit_aux_vision_override + cfg = {"auxiliary": {"vision": {"provider": "", "model": "", "base_url": ""}}} + assert _explicit_aux_vision_override(cfg) is False + + def test_returns_false_for_provider_auto(self): + from tools.computer_use.vision_routing import _explicit_aux_vision_override + cfg = {"auxiliary": {"vision": {"provider": "auto"}}} + assert _explicit_aux_vision_override(cfg) is False + + def test_returns_false_for_provider_AUTO_uppercase(self): + from tools.computer_use.vision_routing import _explicit_aux_vision_override + cfg = {"auxiliary": {"vision": {"provider": " AUTO "}}} + assert _explicit_aux_vision_override(cfg) is False + + def test_returns_true_for_explicit_provider(self): + from tools.computer_use.vision_routing import _explicit_aux_vision_override + cfg = {"auxiliary": {"vision": {"provider": "openrouter"}}} + assert _explicit_aux_vision_override(cfg) is True + + def test_returns_true_for_explicit_model_only(self): + from tools.computer_use.vision_routing import _explicit_aux_vision_override + cfg = {"auxiliary": {"vision": {"model": "google/gemini-2.5-flash"}}} + assert _explicit_aux_vision_override(cfg) is True + + def test_returns_true_for_explicit_base_url_only(self): + from tools.computer_use.vision_routing import _explicit_aux_vision_override + cfg = {"auxiliary": {"vision": {"base_url": "http://localhost:1234/v1"}}} + assert _explicit_aux_vision_override(cfg) is True + + def test_returns_true_for_provider_auto_plus_explicit_model(self): + """``provider: auto`` + an explicit model still counts as override.""" + from tools.computer_use.vision_routing import _explicit_aux_vision_override + cfg = { + "auxiliary": { + "vision": {"provider": "auto", "model": "claude-3-haiku"}, + } + } + assert _explicit_aux_vision_override(cfg) is True + + def test_handles_non_dict_vision_block(self): + from tools.computer_use.vision_routing import _explicit_aux_vision_override + cfg = {"auxiliary": {"vision": "not-a-dict"}} + assert _explicit_aux_vision_override(cfg) is False + + +# --------------------------------------------------------------------------- +# should_route_capture_to_aux_vision +# --------------------------------------------------------------------------- + +class TestRouteDecision: + """End-to-end policy: explicit override > tool-result support > vision caps.""" + + def test_explicit_override_routes_to_aux_even_for_vision_main(self): + """Issue #24015 core repro: explicit aux config must win. + + Even if the main model fully supports vision (Anthropic / Claude), + an explicit ``auxiliary.vision`` block means the user wants their + configured backend used. Don't silently bypass it. + """ + from tools.computer_use import vision_routing + + cfg = { + "auxiliary": { + "vision": { + "provider": "openrouter", + "model": "google/gemini-2.5-flash", + } + } + } + with patch.object(vision_routing, "_lookup_supports_vision", return_value=True), \ + patch.object(vision_routing, + "_provider_accepts_multimodal_tool_result", + return_value=True): + assert vision_routing.should_route_capture_to_aux_vision( + "anthropic", "claude-opus-4-5", cfg + ) is True + + def test_non_vision_main_model_routes_to_aux(self): + """The reported #24015 scenario: tencent/hy3-preview has no vision.""" + from tools.computer_use import vision_routing + + cfg = {"model": {"default": "tencent/hy3-preview", "provider": "openrouter"}} + with patch.object(vision_routing, "_lookup_supports_vision", return_value=False), \ + patch.object(vision_routing, + "_provider_accepts_multimodal_tool_result", + return_value=True): + assert vision_routing.should_route_capture_to_aux_vision( + "openrouter", "tencent/hy3-preview", cfg + ) is True + + def test_vision_main_model_no_override_keeps_multimodal(self): + """Default path: vision-capable main model + no aux override → native.""" + from tools.computer_use import vision_routing + + with patch.object(vision_routing, "_lookup_supports_vision", return_value=True), \ + patch.object(vision_routing, + "_provider_accepts_multimodal_tool_result", + return_value=True): + assert vision_routing.should_route_capture_to_aux_vision( + "anthropic", "claude-opus-4-5", None + ) is False + + def test_provider_rejects_multimodal_tool_results_routes_to_aux(self): + """Some providers' tool-result messages won't carry images at all.""" + from tools.computer_use import vision_routing + + with patch.object(vision_routing, "_lookup_supports_vision", return_value=True), \ + patch.object(vision_routing, + "_provider_accepts_multimodal_tool_result", + return_value=False): + assert vision_routing.should_route_capture_to_aux_vision( + "some-aggregator", "some-vision-model", {} + ) is True + + def test_unknown_provider_capabilities_fail_closed(self): + """When tool-result lookup returns None, route to aux (safe default).""" + from tools.computer_use import vision_routing + + with patch.object(vision_routing, "_lookup_supports_vision", return_value=True), \ + patch.object(vision_routing, + "_provider_accepts_multimodal_tool_result", + return_value=None): + assert vision_routing.should_route_capture_to_aux_vision( + "exotic-provider", "exotic-model", {} + ) is True + + def test_unknown_vision_capability_fails_closed(self): + """When models.dev has no entry, prefer aux over a likely 404.""" + from tools.computer_use import vision_routing + + with patch.object(vision_routing, "_lookup_supports_vision", return_value=None), \ + patch.object(vision_routing, + "_provider_accepts_multimodal_tool_result", + return_value=True): + assert vision_routing.should_route_capture_to_aux_vision( + "openrouter", "novel/never-seen-model", {} + ) is True + + def test_explicit_override_wins_over_unknown_caps(self): + """Explicit aux config wins regardless of unknown caps elsewhere.""" + from tools.computer_use import vision_routing + + cfg = {"auxiliary": {"vision": {"provider": "openrouter"}}} + with patch.object(vision_routing, "_lookup_supports_vision", return_value=None), \ + patch.object(vision_routing, + "_provider_accepts_multimodal_tool_result", + return_value=None): + assert vision_routing.should_route_capture_to_aux_vision( + "openrouter", "tencent/hy3-preview", cfg + ) is True + + +# --------------------------------------------------------------------------- +# Internal lookups — defensive paths +# --------------------------------------------------------------------------- + +class TestLookupHelpers: + def test_lookup_supports_vision_returns_none_for_blank_provider(self): + from tools.computer_use.vision_routing import _lookup_supports_vision + assert _lookup_supports_vision("", "claude") is None + + def test_lookup_supports_vision_returns_none_for_blank_model(self): + from tools.computer_use.vision_routing import _lookup_supports_vision + assert _lookup_supports_vision("anthropic", "") is None + + def test_lookup_supports_vision_handles_lookup_exception(self): + """Underlying caps lookup may raise; helper must swallow + return None.""" + from tools.computer_use import vision_routing + + def _boom(_provider, _model): + raise RuntimeError("models.dev unreachable") + + with patch("agent.models_dev.get_model_capabilities", side_effect=_boom): + assert vision_routing._lookup_supports_vision("anthropic", "claude") is None + + def test_lookup_supports_vision_returns_none_when_caps_missing(self): + from tools.computer_use import vision_routing + + with patch("agent.models_dev.get_model_capabilities", return_value=None): + assert vision_routing._lookup_supports_vision("anthropic", "claude") is None + + def test_provider_accepts_multimodal_tool_result_returns_none_for_blank_provider(self): + from tools.computer_use.vision_routing import ( + _provider_accepts_multimodal_tool_result, + ) + assert _provider_accepts_multimodal_tool_result("", "claude") is None + + +# --------------------------------------------------------------------------- +# Module surface +# --------------------------------------------------------------------------- + +class TestModuleSurface: + """Pin the public surface so dependents stay in lockstep.""" + + def test_should_route_capture_to_aux_vision_is_exported(self): + from tools.computer_use import vision_routing + + assert "should_route_capture_to_aux_vision" in vision_routing.__all__ + assert callable(vision_routing.should_route_capture_to_aux_vision) + + @pytest.mark.parametrize("name", [ + "_explicit_aux_vision_override", + "_lookup_supports_vision", + "_provider_accepts_multimodal_tool_result", + ]) + def test_internal_helpers_are_addressable(self, name): + """Internal helpers stay importable so tests can monkeypatch them.""" + from tools.computer_use import vision_routing + + assert hasattr(vision_routing, name) + assert callable(getattr(vision_routing, name)) diff --git a/tools/computer_use/tool.py b/tools/computer_use/tool.py index 63a5076c17181..2c0b3382d3eda 100644 --- a/tools/computer_use/tool.py +++ b/tools/computer_use/tool.py @@ -423,6 +423,21 @@ def _capture_response(cap: CaptureResult) -> Any: summary = "\n".join(summary_lines) if cap.png_b64 and cap.mode != "ax": + # 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 (no vision client, aux + # call raised, etc.). Fall through to the multimodal envelope — + # better to surface a tool-result error from the main model + # than to silently drop the screenshot entirely. + # Detect actual image format from base64 magic bytes so the MIME type # matches what the data contains (cua-driver may return JPEG or PNG). # JPEG: base64 starts with /9j/ PNG: starts with iVBOR @@ -451,6 +466,140 @@ def _capture_response(cap: CaptureResult) -> Any: }) +# --------------------------------------------------------------------------- +# auxiliary.vision routing for captured screenshots (#24015) +# --------------------------------------------------------------------------- + +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() + model = _read_main_model() + cfg = load_config() + except Exception as exc: # pragma: no cover - defensive + logger.debug("computer_use: aux-vision routing config read failed: %s", exc) + return False + try: + return 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 + + +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. + ext = ".jpg" if cap.png_b64[:8].startswith("/9j/") else ".png" + cache_dir = get_hermes_dir("cache/vision", "temp_vision_images") + temp_image_path = cache_dir / f"computer_use_{_uuid.uuid4().hex}{ext}" + temp_image_path.write_bytes(raw) + + prompt = ( + "Describe what is visible in this macOS 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); " + "falling back to native multimodal envelope", + 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: diff --git a/tools/computer_use/vision_routing.py b/tools/computer_use/vision_routing.py new file mode 100644 index 0000000000000..3b4be1e15a657 --- /dev/null +++ b/tools/computer_use/vision_routing.py @@ -0,0 +1,152 @@ +"""Vision-routing decisions for ``computer_use`` capture results. + +Background +---------- +``computer_use(action='capture', mode='som'|'vision')`` returns a +``_multimodal`` envelope containing the captured screenshot. That envelope +is delivered back to the **active session model** as the tool result. When +the active main model has no vision capability (e.g. text-only or +text+code-only models), or when the active provider rejects multimodal +content inside tool-result messages, the screenshot trips a 404 / 400 at +the provider boundary and the agent loop reports a hard tool failure. + +Issue #24015 reports this regression for the ``cua-driver`` backend: +configuring ``auxiliary.vision`` (a dedicated vision-capable model) in +``config.yaml`` was silently ignored — the screenshot was still routed at +the *main* model and failed with HTTP 404 ``No endpoints found that +support image input`` even though a perfectly good vision backend was +sitting in config waiting to be used. + +This module centralises the small policy decision: should a captured +screenshot be returned as multimodal content (main model handles vision +natively) or pre-analysed via the auxiliary vision pipeline so the main +model only ever sees text? + +Behaviour (mirrors ``vision_analyze`` for consistency) +------------------------------------------------------ +* If the user explicitly configured ``auxiliary.vision`` (any of + ``provider``, ``model``, or ``base_url`` non-empty / not ``"auto"``), + the screenshot is routed through the aux vision pipeline. Users who + pay for a dedicated vision model usually want it used. +* Otherwise, if the active main model+provider can carry an image inside + a tool-result message AND the model reports ``supports_vision=True`` + in models.dev metadata, return ``False`` (use the multimodal path). +* In every other case (non-vision main model, provider that does not + accept multimodal tool results, lookup failure), route through aux + vision so the main model receives a text description it can act on. + +The decision intentionally fails *closed* (i.e. towards aux routing) when +metadata is missing or ambiguous: returning a screenshot to a model that +cannot read it is a hard tool failure, while routing it through aux costs +one extra LLM call and yields a usable description. +""" + +from __future__ import annotations + +import logging +from typing import Any, Dict, Optional + +logger = logging.getLogger(__name__) + + +def _explicit_aux_vision_override(cfg: Optional[Dict[str, Any]]) -> bool: + """True when ``auxiliary.vision`` carries a non-default user override. + + Mirrors ``agent.image_routing._explicit_aux_vision_override`` so the + capture path and the user-attached-image path agree on what counts as + an explicit user request for the aux vision pipeline. ``provider: + "auto"``, blank values, or a missing block all count as *not* + explicit. + """ + if not isinstance(cfg, dict): + return False + aux = cfg.get("auxiliary") or {} + if not isinstance(aux, dict): + return False + vision = aux.get("vision") or {} + if not isinstance(vision, dict): + return False + + provider = str(vision.get("provider") or "").strip().lower() + model = str(vision.get("model") or "").strip() + base_url = str(vision.get("base_url") or "").strip() + + if provider in ("", "auto") and not model and not base_url: + return False + return True + + +def _lookup_supports_vision(provider: str, model: str) -> Optional[bool]: + """Return models.dev ``supports_vision`` for *(provider, model)* or None.""" + if not provider or not model: + return None + try: + from agent.models_dev import get_model_capabilities + caps = get_model_capabilities(provider, model) + except Exception as exc: # pragma: no cover - defensive + logger.debug( + "computer_use vision_routing: caps lookup failed for %s:%s — %s", + provider, model, exc, + ) + return None + if caps is None: + return None + return bool(getattr(caps, "supports_vision", False)) + + +def _provider_accepts_multimodal_tool_result(provider: str, model: str) -> Optional[bool]: + """Return whether *provider*+*model* carries images inside tool-result messages. + + Reuses ``tools.vision_tools._supports_media_in_tool_results`` so the + capture-routing decision stays in lockstep with the + ``vision_analyze`` native fast path. Returns None on import failure + so callers fall back to aux routing rather than guessing. + """ + if not provider: + return None + try: + from tools.vision_tools import _supports_media_in_tool_results + except Exception as exc: # pragma: no cover - defensive + logger.debug( + "computer_use vision_routing: tool-result support lookup failed: %s", + exc, + ) + return None + return bool(_supports_media_in_tool_results(provider, model)) + + +def should_route_capture_to_aux_vision( + provider: str, + model: str, + cfg: Optional[Dict[str, Any]], +) -> bool: + """Return True iff the captured screenshot should be pre-analysed via aux vision. + + Args: + provider: active inference provider id (e.g. ``"openrouter"``, + ``"anthropic"``, ``"openai-codex"``). Lower-case canonical id. + model: active main model slug as it would be sent to the provider. + cfg: loaded ``config.yaml`` dict (or None). + + Returns: + ``True`` when the caller should hand the screenshot to the aux vision + pipeline (and surface a text-only tool result). ``False`` when the + caller should keep the existing multimodal envelope (main model + handles vision natively). + """ + if _explicit_aux_vision_override(cfg): + return True + + accepts_tool_image = _provider_accepts_multimodal_tool_result(provider, model) + if accepts_tool_image is None or accepts_tool_image is False: + return True + + supports_vision = _lookup_supports_vision(provider, model) + if supports_vision is True: + return False + return True + + +__all__ = [ + "should_route_capture_to_aux_vision", +]