Computer-use AX tree context size and routing fixes

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #30126 fix(computer_use): route SOM/vision captures via auxiliary.vision (#24015)

  merged to main

  files: tests/tools/test_computer_use_capture_routing.py, tests/tools/test_computer_use_vision_routing.py, tools/computer_use/tool.py, tools/computer_use/vision_routing.py

  Show the diff

  • 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/"
    +)
    +
    +
    [redacted:email]
    +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="<coro>")
    +
    +        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 "<coro>"
    +
    +        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 "<coro>"
    +
    +        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="<coro>")
    +
    +        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="<coro>")
    +
    +        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="<coro>")
    +
    +        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

  • 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))

  • tools/computer_use/tool.py

    @@ -429,6 +429,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
    @@ -457,6 +472,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:

  • 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",
    +]

• #30145 fix(computer-use): cap AX elements array to prevent context blowup (#22865)

  merged to main

  files: tests/tools/test_computer_use.py, tools/computer_use/schema.py, tools/computer_use/tool.py

  Show the diff

  • tests/tools/test_computer_use.py

    @@ -76,6 +76,27 @@ def test_capture_mode_enum_has_som_vision_ax(self):
             modes = set(COMPUTER_USE_SCHEMA["parameters"]["properties"]["mode"]["enum"])
             assert modes == {"som", "vision", "ax"}
     
    +    def test_schema_exposes_max_elements_cap_for_capture(self):
    +        from tools.computer_use.schema import COMPUTER_USE_SCHEMA
    +        props = COMPUTER_USE_SCHEMA["parameters"]["properties"]
    +        assert "max_elements" in props
    +        assert props["max_elements"]["type"] == "integer"
    +        assert props["max_elements"].get("minimum", 1) >= 1
    +
    +    def test_schema_max_elements_documents_default_and_upper_bound(self):
    +        """Schema description must agree with the runtime. The original PR
    +        text said "Default 100" without a corresponding `default` field, and
    +        had no upper bound — both Copilot findings.
    +        """
    +        from tools.computer_use.schema import COMPUTER_USE_SCHEMA
    +        from tools.computer_use.tool import (
    +            _DEFAULT_MAX_ELEMENTS,
    +            _MAX_ALLOWED_MAX_ELEMENTS,
    +        )
    +        prop = COMPUTER_USE_SCHEMA["parameters"]["properties"]["max_elements"]
    +        assert prop.get("default") == _DEFAULT_MAX_ELEMENTS
    +        assert prop.get("maximum") == _MAX_ALLOWED_MAX_ELEMENTS
    +
     
     class TestRegistration:
         def test_tool_registers_with_registry(self):
    @@ -337,6 +358,193 @@ def focus_app(self, app, raise_window=False): ...
             assert "AXButton" in text_part["text"]
             assert "AXTextField" in text_part["text"]
     
    +    def _ax_backend_with(self, count: int):
    +        """Construct a fake backend that yields ``count`` AX elements."""
    +        from tools.computer_use.backend import CaptureResult, UIElement
    +
    +        elements = [
    +            UIElement(index=i + 1, role="AXButton", label=f"el-{i}", bounds=(0, 0, 1, 1))
    +            for i in range(count)
    +        ]
    +
    +        class FakeBackend:
    +            def start(self): pass
    +            def stop(self): pass
    +            def is_available(self): return True
    +            def capture(self, mode="som", app=None):
    +                return CaptureResult(
    +                    mode=mode, width=800, height=600,
    +                    png_b64="",
    +                    elements=list(elements),
    +                    app="Obsidian",
    +                )
    +            def click(self, **kw): ...
    +            def drag(self, **kw): ...
    +            def scroll(self, **kw): ...
    +            def type_text(self, text): ...
    +            def key(self, keys): ...
    +            def list_apps(self): return []
    +            def focus_app(self, app, raise_window=False): ...
    +
    +        return FakeBackend()
    +
    +    def test_capture_ax_caps_elements_at_default_for_dense_trees(self):
    +        """Regression for #22865: an Electron-style 600-element AX tree must
    +        not emit the entire array verbatim into the tool result.
    +        """
    +        from tools.computer_use import tool as cu_tool
    +
    +        fake_backend = self._ax_backend_with(600)
    +        cu_tool.reset_backend_for_tests()
    +        with patch.object(cu_tool, "_get_backend", return_value=fake_backend):
    +            out = cu_tool.handle_computer_use({"action": "capture", "mode": "ax"})
    +
    +        parsed = json.loads(out)
    +        assert parsed["mode"] == "ax"
    +        assert parsed["total_elements"] == 600
    +        assert len(parsed["elements"]) == cu_tool._DEFAULT_MAX_ELEMENTS
    +        assert parsed["truncated_elements"] == 600 - cu_tool._DEFAULT_MAX_ELEMENTS
    +        # Truncation must be visible in the human summary so the model knows
    +        # the JSON view is partial and can re-issue with a tighter scope.
    +        assert "truncated to" in parsed["summary"]
    +
    +    def test_capture_ax_honors_explicit_max_elements_override(self):
    +        from tools.computer_use import tool as cu_tool
    +
    +        fake_backend = self._ax_backend_with(600)
    +        cu_tool.reset_backend_for_tests()
    +        with patch.object(cu_tool, "_get_backend", return_value=fake_backend):
    +            out = cu_tool.handle_computer_use(
    +                {"action": "capture", "mode": "ax", "max_elements": 250}
    +            )
    +
    +        parsed = json.loads(out)
    +        assert len(parsed["elements"]) == 250
    +        assert parsed["truncated_elements"] == 350
    +
    +    def test_capture_ax_below_cap_is_unchanged(self):
    +        """Backwards-compat: small captures keep the full elements array and
    +        do not surface a `truncated_elements` field.
    +        """
    +        from tools.computer_use import tool as cu_tool
    +
    +        fake_backend = self._ax_backend_with(5)
    +        cu_tool.reset_backend_for_tests()
    +        with patch.object(cu_tool, "_get_backend", return_value=fake_backend):
    +            out = cu_tool.handle_computer_use({"action": "capture", "mode": "ax"})
    +
    +        parsed = json.loads(out)
    +        assert len(parsed["elements"]) == 5
    +        assert parsed["total_elements"] == 5
    +        assert "truncated_elements" not in parsed
    +        assert "truncated to" not in parsed["summary"]
    +
    +    def test_capture_ax_invalid_max_elements_falls_back_to_default(self):
    +        """Malformed `max_elements` (string, negative, zero) must not silently
    +        disable the cap and re-introduce the original unbounded behavior.
    +        """
    +        from tools.computer_use import tool as cu_tool
    +
    +        fake_backend = self._ax_backend_with(600)
    +        cu_tool.reset_backend_for_tests()
    +        for bad in ("not-a-number", 0, -10):
    +            with patch.object(cu_tool, "_get_backend", return_value=fake_backend):
    +                out = cu_tool.handle_computer_use(
    +                    {"action": "capture", "mode": "ax", "max_elements": bad}
    +                )
    +            parsed = json.loads(out)
    +            assert len(parsed["elements"]) == cu_tool._DEFAULT_MAX_ELEMENTS, (
    +                f"bad max_elements={bad!r} disabled the cap"
    +            )
    +
    +    def test_capture_ax_clamps_oversized_max_elements_to_hard_cap(self):
    +        """A caller passing a very large `max_elements` must not be able to
    +        disable the safeguard. The cap is clamped to a hard upper bound so
    +        the context-blow-up protection cannot be bypassed by argument.
    +        """
    +        from tools.computer_use import tool as cu_tool
    +
    +        fake_backend = self._ax_backend_with(5000)
    +        cu_tool.reset_backend_for_tests()
    +        with patch.object(cu_tool, "_get_backend", return_value=fake_backend):
    +            out = cu_tool.handle_computer_use(
    +                {"action": "capture", "mode": "ax", "max_elements": 10_000}
    +            )
    +        parsed = json.loads(out)
    +        assert len(parsed["elements"]) == cu_tool._MAX_ALLOWED_MAX_ELEMENTS
    +        assert parsed["total_elements"] == 5000
    +        assert parsed["truncated_elements"] == 5000 - cu_tool._MAX_ALLOWED_MAX_ELEMENTS
    +
    +    def test_capture_ax_summary_indices_match_returned_elements(self):
    +        """When `max_elements` is below the human-summary's own line cap, the
    +        summary must not index elements that aren't in the returned array.
    +        Otherwise the model sees `#15` in the summary and finds no matching
    +        entry in `elements`.
    +        """
    +        from tools.computer_use import tool as cu_tool
    +
    +        fake_backend = self._ax_backend_with(600)
    +        cu_tool.reset_backend_for_tests()
    +        with patch.object(cu_tool, "_get_backend", return_value=fake_backend):
    +            out = cu_tool.handle_computer_use(
    +                {"action": "capture", "mode": "ax", "max_elements": 5}
    +            )
    +        parsed = json.loads(out)
    +        returned_indices = {e["index"] for e in parsed["elements"]}
    +        summary_lines = parsed["summary"].splitlines()
    +        indexed_lines = [ln for ln in summary_lines if ln.lstrip().startswith("#")]
    +        for ln in indexed_lines:
    +            idx_token = [redacted:secret-assignment]"#")
    +            idx = int(idx_token)
    +            assert idx in returned_indices, (
    +                f"summary references #{idx} but it is absent from elements payload "
    +                f"(returned: {sorted(returned_indices)})"
    +            )
    +
    +    def test_capture_multimodal_summary_omits_truncation_note(self):
    +        """The som/vision multimodal envelope returns a screenshot, not an
    +        `elements` array — so a "response truncated to N of M elements"
    +        claim in the summary would be inaccurate.
    +        """
    +        from tools.computer_use.backend import CaptureResult, UIElement
    +        from tools.computer_use import tool as cu_tool
    +
    +        fake_png = "iVBORw0KGgo="
    +        elements = [
    +            UIElement(index=i + 1, role="AXButton", label=f"el-{i}", bounds=(0, 0, 1, 1))
    +            for i in range(600)
    +        ]
    +
    +        class FakeBackend:
    +            def start(self): pass
    +            def stop(self): pass
    +            def is_available(self): return True
    +            def capture(self, mode="som", app=None):
    +                return CaptureResult(
    +                    mode=mode, width=800, height=600,
    +                    png_b64=fake_png, elements=list(elements),
    +                    app="Obsidian",
    +                )
    +            def click(self, **kw): ...
    +            def drag(self, **kw): ...
    +            def scroll(self, **kw): ...
    +            def type_text(self, text): ...
    +            def key(self, keys): ...
    +            def list_apps(self): return []
    +            def focus_app(self, app, raise_window=False): ...
    +
    +        cu_tool.reset_backend_for_tests()
    +        with patch.object(cu_tool, "_get_backend", return_value=FakeBackend()):
    +            out = cu_tool.handle_computer_use({"action": "capture", "mode": "som"})
    +
    +        assert isinstance(out, dict) and out["_multimodal"] is True
    +        text_part = next(p for p in out["content"] if p.get("type") == "text")
    +        assert "truncated to" not in text_part["text"], (
    +            "multimodal response carries an image, not an elements array; "
    +            "the truncation note describes a payload field that isn't present"
    +        )
    +        assert "truncated to" not in out["text_summary"]
    +
     
     # ---------------------------------------------------------------------------
     # Anthropic adapter: multimodal tool-result conversion

  • tools/computer_use/schema.py

    @@ -75,6 +75,28 @@
                         "frontmost app's window or the whole screen."
                     ),
                 },
    +            "max_elements": {
    +                "type": "integer",
    +                "description": (
    +                    "Optional cap on the AX `elements` array returned by "
    +                    "`action='capture'`. Default 100, hard maximum 1000. "
    +                    "Dense UIs (Electron apps such as Obsidian or VS Code, "
    +                    "JetBrains IDEs) can publish 500+ AX nodes — capping "
    +                    "prevents a single capture from blowing session "
    +                    "context. When the cap trims the response, "
    +                    "`total_elements` and `truncated_elements` are "
    +                    "surfaced in the result so you can re-call with "
    +                    "`app=` to narrow scope or raise `max_elements` when "
    +                    "the full tree is required. Has no effect on "
    +                    "`mode='som'` / `mode='vision'` when a screenshot is "
    +                    "included in the response; only the rare image-"
    +                    "missing fallback returns an `elements` array and is "
    +                    "subject to the cap."
    +                ),
    +                "default": 100,
    +                "minimum": 1,
    +                "maximum": 1000,
    +            },
                 # ── click / drag / scroll targeting ────────────────────
                 "element": {
                     "type": "integer",

  • tools/computer_use/tool.py

    @@ -317,7 +317,7 @@ def _dispatch(backend: ComputerUseBackend, action: str, args: Dict[str, Any]) ->
             if mode not in {"som", "vision", "ax"}:
                 return json.dumps({"error": f"bad mode {mode!r}; use som|vision|ax"})
             cap = backend.capture(mode=mode, app=args.get("app"))
    -        return _capture_response(cap)
    +        return _capture_response(cap, max_elements=_coerce_max_elements(args.get("max_elements")))
     
         if action == "wait":
             seconds = float(args.get("seconds", 1.0))
    @@ -416,16 +416,62 @@ def _text_response(res: ActionResult) -> str:
         return json.dumps(payload)
     
     
    -def _capture_response(cap: CaptureResult) -> Any:
    -    element_index = _format_elements(cap.elements)
    +# Default cap for the AX `elements` array returned by capture. Dense UIs
    +# (Electron apps, Obsidian, JetBrains IDEs) can publish 500+ AX nodes, which
    +# can exhaust session context after a single capture. The model-facing
    +# `max_elements` argument lets callers raise this when they need the full tree.
    +_DEFAULT_MAX_ELEMENTS = 100
    +# Hard upper bound on caller-supplied `max_elements`. Without this, a tool
    +# call passing a very large integer would silently disable the safeguard and
    +# reintroduce the original unbounded behavior.
    +_MAX_ALLOWED_MAX_ELEMENTS = 1000
    +
    +
    +def _coerce_max_elements(value: Any) -> int:
    +    """Validate the caller-supplied ``max_elements``.
    +
    +    Falls back to :data:`_DEFAULT_MAX_ELEMENTS` for missing / non-integer /
    +    sub-1 inputs so the cap can never be silently disabled by a malformed
    +    tool-call argument. Clamps oversized values to
    +    :data:`_MAX_ALLOWED_MAX_ELEMENTS` so a caller cannot bypass the
    +    safeguard by passing a very large integer.
    +    """
    +    if value is None:
    +        return _DEFAULT_MAX_ELEMENTS
    +    try:
    +        n = int(value)
    +    except (TypeError, ValueError):
    +        return _DEFAULT_MAX_ELEMENTS
    +    if n < 1:
    +        return _DEFAULT_MAX_ELEMENTS
    +    if n > _MAX_ALLOWED_MAX_ELEMENTS:
    +        return _MAX_ALLOWED_MAX_ELEMENTS
    +    return n
    +
    +
    +def _capture_response(cap: CaptureResult, max_elements: int = _DEFAULT_MAX_ELEMENTS) -> Any:
    +    total_elements = len(cap.elements)
    +    visible_elements = cap.elements[:max_elements]
    +    truncated_elements = max(0, total_elements - len(visible_elements))
    +
    +    # Index only what's actually surfaced in the response — otherwise the
    +    # human-readable summary references element indices the model cannot
    +    # find in the JSON `elements` array (e.g. max_elements=10 vs the default
    +    # 40-line index window).
    +    element_index = _format_elements(visible_elements)
         summary_lines = [
             f"capture mode={cap.mode} {cap.width}x{cap.height}"
             + (f" app={cap.app}" if cap.app else "")
             + (f" window={cap.window_title!r}" if cap.window_title else ""),
    -        f"{len(cap.elements)} interactable element(s):",
    +        f"{total_elements} interactable element(s):",
         ]
         if element_index:
             summary_lines.extend(element_index)
    +    # Multimodal and AX paths both reference `summary`; build it once up-front
    +    # so the aux-vision routing branch (which fires before either path is
    +    # selected) has a valid value to hand to _route_capture_through_aux_vision.
    +    # The AX path appends the "truncated to N of M" note to summary_lines
    +    # below and rebuilds; the multimodal path keeps this version untouched.
         summary = "\n".join(summary_lines)
     
         if cap.png_b64 and cap.mode != "ax":
    @@ -449,6 +495,9 @@ def _capture_response(cap: CaptureResult) -> Any:
             # JPEG: base64 starts with /9j/   PNG: starts with iVBOR
             _b64_prefix = cap.png_b64[:8]
             _mime = "image/jpeg" if _b64_prefix.startswith("/9j/") else "image/png"
    +        # The multimodal response carries the screenshot, not the AX
    +        # elements array, so a "response truncated to N of M elements"
    +        # note would be inaccurate — skip it on this branch.
             return {
                 "_multimodal": True,
                 "content": [
    @@ -458,18 +507,29 @@ def _capture_response(cap: CaptureResult) -> Any:
                 ],
                 "text_summary": summary,
                 "meta": {"mode": cap.mode, "width": cap.width, "height": cap.height,
    -                     "elements": len(cap.elements), "png_bytes": cap.png_bytes_len},
    +                     "elements": total_elements, "png_bytes": cap.png_bytes_len},
             }
    -    # AX-only (or image missing): text path.
    -    return json.dumps({
    +    # AX-only (or image-missing fallback): text path actually carries the
    +    # `elements` array, so the truncation note applies here.
    +    if truncated_elements:
    +        summary_lines.append(
    +            f"  (response truncated to {len(visible_elements)} of {total_elements} elements; "
    +            f"raise max_elements or pass app= to narrow)"
    +        )
    +    summary = "\n".join(summary_lines)
    +    payload: Dict[str, Any] = {
             "mode": cap.mode,
             "width": cap.width,
             "height": cap.height,
             "app": cap.app,
             "window_title": cap.window_title,
    -        "elements": [_element_to_dict(e) for e in cap.elements],
    +        "elements": [_element_to_dict(e) for e in visible_elements],
    +        "total_elements": total_elements,
             "summary": summary,
    -    })
    +    }
    +    if truncated_elements:
    +        payload["truncated_elements"] = truncated_elements
    +    return json.dumps(payload)
     
     
     # ---------------------------------------------------------------------------