From de33c2413b19a8ab5d928468a7c623d469b53f04 Mon Sep 17 00:00:00 2001 From: kshitijk4poor <82637225+kshitijk4poor@users.noreply.github.com> Date: Fri, 10 Jul 2026 18:55:55 +0530 Subject: [PATCH] fix(web): harden extract input and display boundaries --- agent/display.py | 33 +++- tests/agent/test_display.py | 12 ++ tests/agent/test_display_todo_progress.py | 6 +- tests/tools/test_web_tools_dict_urls.py | 190 ++++++++++------------ tools/web_tools.py | 46 ++++-- 5 files changed, 163 insertions(+), 124 deletions(-) diff --git a/agent/display.py b/agent/display.py index 25b1ade44d..39879e2631 100644 --- a/agent/display.py +++ b/agent/display.py @@ -27,6 +27,14 @@ logger = logging.getLogger(__name__) _ANSI_RESET = "\033[0m" + +def _display_url(value: Any) -> str: + """Extract a display-only URL without assuming model argument types.""" + if isinstance(value, dict): + value = value.get("url") or value.get("href") + return value.strip() if isinstance(value, str) else "" + + # Diff colors — resolved lazily from the skin engine so they adapt # to light/dark themes. Falls back to sensible defaults on import # failure. We cache after first resolution for performance. @@ -1259,7 +1267,7 @@ def _detect_tool_failure(tool_name: str, result: str | None) -> tuple[bool, str] return False, "" -def get_cute_tool_message( +def _get_cute_tool_message( tool_name: str, args: dict, duration: float, result: str | None = None, ) -> str: """Generate a formatted tool completion line for CLI quiet mode. @@ -1301,14 +1309,11 @@ def get_cute_tool_message( if tool_name == "web_extract": urls = args.get("urls", []) if urls: - url = urls[0] if isinstance(urls, list) else str(urls) - # Handle dict objects from web_search results - if isinstance(url, dict): - url = url.get("url") or url.get("href") or "" - if not isinstance(url, str): - url = str(url) + url = _display_url(urls[0] if isinstance(urls, list) else urls) + if not url: + return _wrap(f"┊ 📄 fetch pages {dur}") domain = url.replace("https://", "").replace("http://", "").split("/")[0] - extra = f" +{len(urls)-1}" if len(urls) > 1 else "" + extra = f" +{len(urls)-1}" if isinstance(urls, list) and len(urls) > 1 else "" return _wrap(f"┊ 📄 fetch {_trunc(domain, 35)}{extra} {dur}") return _wrap(f"┊ 📄 fetch pages {dur}") if tool_name == "terminal": @@ -1438,6 +1443,18 @@ def get_cute_tool_message( return _wrap(f"┊ ⚡ {tool_name[:9]:9} {_trunc(preview, 35)} {dur}") +def get_cute_tool_message( + tool_name: str, args: dict, duration: float, result: str | None = None, +) -> str: + """Render a completion label without letting cosmetic failures escape.""" + try: + return _get_cute_tool_message(tool_name, args, duration, result=result) + except Exception as exc: # noqa: BLE001 — display must never abort a turn + logger.debug("Tool completion label failed for %s: %s", tool_name, exc) + safe_name = str(tool_name or "tool")[:9] + return f"┊ ⚡ {safe_name:9} completed {duration:.1f}s" + + # ========================================================================= # Honcho session line (one-liner with clickable OSC 8 hyperlink) # ========================================================================= diff --git a/tests/agent/test_display.py b/tests/agent/test_display.py index ab390fac24..b777e3e7f6 100644 --- a/tests/agent/test_display.py +++ b/tests/agent/test_display.py @@ -4,6 +4,7 @@ import json import pytest from unittest.mock import MagicMock +import agent.display as display_module from agent.display import ( build_tool_preview, capture_local_edit_snapshot, @@ -24,6 +25,17 @@ def reset_tool_preview_max_len(): set_tool_preview_max_len(0) +def test_cute_tool_message_falls_back_when_renderer_raises(monkeypatch): + def _boom(*_args, **_kwargs): + raise RuntimeError("cosmetic failure") + + monkeypatch.setattr(display_module, "_get_cute_tool_message", _boom) + + assert get_cute_tool_message("web_extract", {"urls": []}, 0.25) == ( + "┊ ⚡ web_extra completed 0.2s" + ) + + class TestBuildToolPreview: """Tests for build_tool_preview defensive handling and normal operation.""" diff --git a/tests/agent/test_display_todo_progress.py b/tests/agent/test_display_todo_progress.py index 4940e32e3a..653e247c06 100644 --- a/tests/agent/test_display_todo_progress.py +++ b/tests/agent/test_display_todo_progress.py @@ -280,10 +280,14 @@ class TestWebExtractDisplay: ] } msg = get_cute_tool_message("web_extract", args, 0.2) - # Should handle gracefully, default to empty string domain assert "📄" in msg + assert "pages" in msg assert "0.2s" in msg + def test_web_extract_with_non_string_item_uses_generic_label(self): + msg = get_cute_tool_message("web_extract", {"urls": [123]}, 0.2) + assert "pages" in msg + def test_web_extract_with_multiple_dicts(self): """Multiple dict URLs show '+N' suffix.""" args = { diff --git a/tests/tools/test_web_tools_dict_urls.py b/tests/tools/test_web_tools_dict_urls.py index 41cc67ec9f..026376426b 100644 --- a/tests/tools/test_web_tools_dict_urls.py +++ b/tests/tools/test_web_tools_dict_urls.py @@ -1,114 +1,96 @@ -"""Test web_extract_tool handles dict objects from web_search results. +"""Regression tests for model-forwarded web-search result objects.""" -Reproduces and verifies fix for #61693 where web_search result dicts -caused TypeError when web_extract tried to normalize and validate URLs. -""" import json -from typing import List + +import pytest + +from agent import web_search_registry +from agent.web_search_provider import WebSearchProvider +from tools import web_tools -def test_web_extract_handles_dict_urls(): - """web_extract_tool should extract URL strings from dict objects. +class _FakeExtractProvider(WebSearchProvider): + def __init__(self) -> None: + self.received_urls: list[str] = [] - When the model passes web_search result dicts (with url/href fields), - web_extract_tool should extract the actual URL strings before processing. - This test verifies the dict-to-string coercion logic works without errors. + @property + def name(self) -> str: + return "dict-url-test" - This is a minimal smoke test - it doesn't actually call web_extract_tool - (which requires external services), but verifies the core logic that - dict URLs are converted to strings before regex operations. - """ - # Simulate web_search result dicts - dict_urls: List[dict] = [ - {"url": "https://example.com/page1", "title": "Example 1", "snippet": "..."}, - {"url": "https://example.com/page2", "title": "Example 2", "snippet": "..."}, - {"href": "https://alternate.com/page", "title": "Alternate", "snippet": "..."}, - {"title": "No URL field"}, # Missing url/href + @property + def display_name(self) -> str: + return "Dict URL Test" + + def is_available(self) -> bool: + return True + + def supports_extract(self) -> bool: + return True + + async def extract(self, urls, **kwargs): + self.received_urls.extend(urls) + return [ + {"url": url, "title": "", "content": "ok"} + for url in urls + ] + + +@pytest.fixture +def extract_provider(monkeypatch): + with web_search_registry._lock: + previous = dict(web_search_registry._providers) + web_search_registry._providers.clear() + + provider = _FakeExtractProvider() + web_search_registry.register_provider(provider) + monkeypatch.setattr(web_tools, "_ensure_web_plugins_loaded", lambda: None) + monkeypatch.setattr( + web_tools, + "_load_web_config", + lambda: {"extract_backend": provider.name}, + ) + + async def _safe(_url): + return True + + monkeypatch.setattr(web_tools, "async_is_safe_url", _safe) + yield provider + + with web_search_registry._lock: + web_search_registry._providers.clear() + web_search_registry._providers.update(previous) + + +@pytest.mark.asyncio +async def test_web_extract_dispatches_urls_from_search_result_objects(extract_provider): + result = json.loads(await web_tools.web_extract_tool([ + {"url": "https://example.com/a", "title": "A"}, + {"href": "https://example.org/b"}, + ])) + + assert extract_provider.received_urls == [ + "https://example.com/a", + "https://example.org/b", ] + assert [entry["url"] for entry in result["results"]] == extract_provider.received_urls - # This is the coercion logic from the fix in web_tools.py - processed_urls: List[str] = [] - for _url in dict_urls: - # Handle dict objects from web_search results - if isinstance(_url, dict): - _url = _url.get("url") or _url.get("href") or "" - elif not isinstance(_url, str): - _url = str(_url) - processed_urls.append(_url) - # Verify extraction works - assert processed_urls == [ - "https://example.com/page1", - "https://example.com/page2", - "https://alternate.com/page", - "", # Missing url/href → empty string +@pytest.mark.asyncio +async def test_web_extract_reports_invalid_items_without_dispatching_them(extract_provider): + result = json.loads(await web_tools.web_extract_tool([ + {"url": "https://example.com/good"}, + {"title": "missing URL"}, + {"url": 123}, + None, + ])) + + assert extract_provider.received_urls == ["https://example.com/good"] + errors = [entry["error"] for entry in result["results"] if entry["error"]] + assert errors == [ + "Invalid URL item at index 1: expected a URL string or an object " + "with a string 'url' or 'href' field", + "Invalid URL item at index 2: expected a URL string or an object " + "with a string 'url' or 'href' field", + "Invalid URL item at index 3: expected a URL string or an object " + "with a string 'url' or 'href' field", ] - - -def test_web_extract_handles_mixed_string_dict_urls(): - """web_extract_tool should handle mix of string URLs and dict objects.""" - mixed_urls = [ - "https://direct.com/page", - {"url": "https://dict.com/page", "title": "Dict URL"}, - {"href": "https://href.com/page"}, - "https://another.com/page", - {"title": "No URL field"}, - ] - - processed_urls: List[str] = [] - for _url in mixed_urls: - if isinstance(_url, dict): - _url = _url.get("url") or _url.get("href") or "" - elif not isinstance(_url, str): - _url = str(_url) - processed_urls.append(_url) - - assert processed_urls == [ - "https://direct.com/page", - "https://dict.com/page", - "https://href.com/page", - "https://another.com/page", - "", - ] - - -def test_web_extract_dict_coercion_preserves_valid_urls(): - """Dict coercion should not break valid URL strings.""" - valid_url = "https://example.com/path?query=value#fragment" - - # String URL should pass through unchanged - if isinstance(valid_url, dict): - processed = valid_url.get("url") or valid_url.get("href") or "" - elif not isinstance(valid_url, str): - processed = str(valid_url) - else: - processed = valid_url - - assert processed == valid_url - - -def test_web_extract_dict_coercion_handles_edge_cases(): - """Test edge cases: non-dict, non-string objects, empty strings.""" - edge_cases = [ - 123, # int - None, # None - True, # bool - "", # empty string - ] - - processed: List[str] = [] - for item in edge_cases: - if isinstance(item, dict): - processed_item = item.get("url") or item.get("href") or "" - elif not isinstance(item, str): - processed_item = str(item) - else: - processed_item = item - processed.append(processed_item) - - # Non-dict, non-string → str() conversion - assert processed[0] == "123" - assert processed[1] == "None" - assert processed[2] == "True" - # Empty string stays empty - assert processed[3] == "" \ No newline at end of file diff --git a/tools/web_tools.py b/tools/web_tools.py index a5366bccb9..71a53f982e 100644 --- a/tools/web_tools.py +++ b/tools/web_tools.py @@ -103,6 +103,21 @@ import sys logger = logging.getLogger(__name__) +def _web_extract_url(value: Any) -> Optional[str]: + """Return a usable URL from a model-supplied extract item. + + Models sometimes forward a complete web-search result instead of its URL. + Accept the two common URL keys, but reject missing/non-string values rather + than stringifying arbitrary objects into misleading fetch targets. + """ + if isinstance(value, dict): + value = value.get("url") or value.get("href") + if not isinstance(value, str): + return None + value = value.strip() + return value or None + + # ─── Backend Selection ──────────────────────────────────────────────────────── def _env_value(name: str) -> str: @@ -726,7 +741,7 @@ def web_search_tool(query: str, limit: int = 5) -> str: async def web_extract_tool( - urls: List[str], + urls: List[Any], format: str = None, char_limit: Optional[int] = None, ) -> str: @@ -742,7 +757,8 @@ async def web_extract_tool( ``[IMAGE: alt]`` placeholders (real image URLs are preserved as links). Args: - urls (List[str]): List of URLs to extract content from + urls (List[Any]): URL strings or search-result objects containing a + string ``url`` or ``href`` field format (str): Desired output format ("markdown" or "html", optional) char_limit (Optional[int]): Per-page char budget sent to the model (default: web.extract_char_limit or 15000). Larger pages truncate. @@ -762,12 +778,20 @@ async def web_extract_tool( from agent.redact import _PREFIX_RE from urllib.parse import unquote normalized_urls: List[str] = [] - for _url in urls: - # Handle dict objects from web_search results - if isinstance(_url, dict): - _url = _url.get("url") or _url.get("href") or "" - elif not isinstance(_url, str): - _url = str(_url) + invalid_urls: List[Dict[str, Any]] = [] + for index, item in enumerate(urls): + _url = _web_extract_url(item) + if _url is None: + invalid_urls.append({ + "url": "", + "title": "", + "content": "", + "error": ( + f"Invalid URL item at index {index}: expected a URL string " + "or an object with a string 'url' or 'href' field" + ), + }) + continue normalized_url = normalize_url_for_request(_url) if ( _PREFIX_RE.search(_url) @@ -915,9 +939,9 @@ async def web_extract_tool( provider.extract, safe_urls, format=format ) - # Merge any SSRF-blocked results back in - if ssrf_blocked: - results = ssrf_blocked + results + # Merge rejected and SSRF-blocked inputs back into the result envelope. + if invalid_urls or ssrf_blocked: + results = invalid_urls + ssrf_blocked + results response = {"results": results}