fix(web): harden extract input and display boundaries

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)
 # =========================================================================

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."""
 

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 = {

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] == ""

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}