From b8cfaff69082415826f5fa3261240fe476fd46fc Mon Sep 17 00:00:00 2001 From: ciregenz Date: Thu, 30 Jul 2026 14:55:56 -0700 Subject: [PATCH] [eric] web: stop the fetch cascade on a body with no text layer instead of buying a paid summary of nothing --- backend/apps/agents/tools/web.py | 24 +-- backend/apps/web/web.py | 23 +-- backend/tests/test_web_fetch_cascade.py | 124 +++++++++++++++ backend/tests/test_web_search_cascade.py | 182 ++++------------------- backend/tests/web_cascade_fixtures.py | 66 ++++++++ 5 files changed, 236 insertions(+), 183 deletions(-) create mode 100644 backend/tests/test_web_fetch_cascade.py create mode 100644 backend/tests/web_cascade_fixtures.py diff --git a/backend/apps/agents/tools/web.py b/backend/apps/agents/tools/web.py index c3c63f71..3d14e67d 100644 --- a/backend/apps/agents/tools/web.py +++ b/backend/apps/agents/tools/web.py @@ -14,7 +14,7 @@ from backend.apps.agents.tools.search.search_ddg import ( HTTP_TIMEOUT, USER_AGENT, ) -from backend.apps.agents.tools.fetch.page_text import body_to_text, html_to_text, looks_like_pdf +from backend.apps.agents.tools.fetch.page_text import PageText, body_to_text, html_to_text, looks_like_pdf from backend.apps.agents.tools.search.search_ddg import search_ddg as run_ddg_search from backend.apps.agents.tools.ssrf_guard import SSRFBlocked, safe_fetch @@ -165,9 +165,13 @@ class WebFetchTool(BaseTool): } async def execute(self, input_data: dict, context: ToolContext) -> list[dict]: - url: str = input_data["url"] - prompt: str | None = input_data.get("prompt") + page = await self.fetch_page(input_data["url"], input_data.get("prompt")) + return [{"type": "text", "text": page.text}] + @staticmethod + async def fetch_page(url: str, prompt: str | None = None) -> PageText: + """The page as readable text, plus WHAT it was, so callers can tell a + JS wall (worth another tier) from a PNG (nothing left to try).""" try: resp = await safe_fetch( url, @@ -177,11 +181,11 @@ class WebFetchTool(BaseTool): ) resp.raise_for_status() except SSRFBlocked as exc: - return [{"type": "text", "text": f"Refused to fetch {url}: {exc}"}] + return PageText(text=f"Refused to fetch {url}: {exc}", kind="error") except httpx.HTTPStatusError as exc: - return [{"type": "text", "text": f"HTTP error {exc.response.status_code} fetching {url}"}] + return PageText(text=f"HTTP error {exc.response.status_code} fetching {url}", kind="error") except Exception as exc: - return [{"type": "text", "text": f"Error fetching {url}: {exc}"}] + return PageText(text=f"Error fetching {url}: {exc}", kind="error") content_type = resp.headers.get("content-type", "") # A PDF's content-type often says html, so check the magic bytes before trusting the header. @@ -189,14 +193,12 @@ class WebFetchTool(BaseTool): is_html = not is_pdf and ("html" in content_type or resp.text.strip().startswith(" str: - out: List[str] = [] - for p in parts: - if isinstance(p, dict) and p.get("type") == "text": - out.append(str(p.get("text", ""))) - return "\n".join(out) - - # Drive the packaged app's offscreen Chromium (main-process hidden window) for a fetch/search. Returns the bridge result dict, or None when no Electron main bridge is connected (dev/headless/backend-only) so the cascade just skips this tier. This is the real "browser reachable" gate, OPENSWARM_BROWSER_OK is effectively always "1" and not trustworthy for this. @typechecked async def p_browser_bridge(action: str, params: Dict) -> Optional[Dict]: @@ -256,15 +247,17 @@ async def fetch(body: FetchBody) -> Dict: # Fast path: direct httpx + trafilatura, and it returns the page's ACTUAL text (the grounded fetchers summarize, which is slower and loses detail). Thin/errored reads (JS walls, paywalls, HTTP errors) fall through. nonlocal local_text from backend.apps.agents.tools.web import WebFetchTool - parts = await WebFetchTool().execute({"url": body.url, "prompt": body.prompt or ""}, None) - text = p_join_text(parts) - local_text = text - if text.startswith(("HTTP error", "Error fetching", "Refused to fetch")): + page = await WebFetchTool.fetch_page(body.url, body.prompt) + local_text = page.text + if page.kind == "error": return None - body_text = text.split("\n\n", 1)[-1] if "\n\n" in text else text + # A PNG or a scanned PDF has no text for ANY tier to find, so spending a paid fetcher on it buys nothing. + if page.kind in ("binary", "pdf_unreadable"): + return {"url": body.url, "content": page.text, "backend": "local"} + body_text = page.text.split("\n\n", 1)[-1] if len(body_text.strip()) < 200: return None - return {"url": body.url, "content": text, "backend": "local"} + return {"url": body.url, "content": page.text, "backend": "local"} async def try_browser_fetch() -> Optional[Dict]: # Packaged-app tier: renders the page in a real offscreen Chromium and returns its visible text, so JS-only / SPA / soft-paywall pages that give httpx nothing actually resolve. Shares the user's browser cookies, so pages they're logged into fetch authed. diff --git a/backend/tests/test_web_fetch_cascade.py b/backend/tests/test_web_fetch_cascade.py new file mode 100644 index 00000000..a7ff75d0 --- /dev/null +++ b/backend/tests/test_web_fetch_cascade.py @@ -0,0 +1,124 @@ +"""Deadline-bounded /api/web/fetch cascade. + +Mirrors /search: the local httpx + trafilatura read is the fast path, the +packaged browser and the Wayback archive cover JS walls and dead links, and the +grounded fetchers are the last resort. A body with no text layer at all stops +the chain rather than buying a paid summary of nothing. +""" + +import time + +import pytest + +import backend.apps.agents.tools.fetch.wayback as WB +import backend.apps.web.web as W +from backend.apps.agents.tools.fetch.page_text import PageText +from backend.apps.agents.tools.web import WebFetchTool +from backend.apps.web.web import fetch, FetchBody +from backend.tests.web_cascade_fixtures import ( # noqa: F401 + allow_urls, + patch_browser_bridge, + no_network, +) + + +def p_local_returns(monkeypatch, text, kind="html"): + async def p_fetch(url, prompt=None): + return PageText(text=text, kind=kind) + monkeypatch.setattr(WebFetchTool, "fetch_page", staticmethod(p_fetch)) + + + +@pytest.mark.asyncio +async def test_fetch_local_first_wins_and_is_fast(monkeypatch): + big = "Contents of https://x.example:\n\n" + ("real article body " * 50) + p_local_returns(monkeypatch, big) + async def p_boom(*a, **k): + raise AssertionError("grounded fetch should not run when local has content") + monkeypatch.setattr(W, "gemini_grounded_call", p_boom) + + t = time.monotonic() + res = await fetch(FetchBody(url="https://x.example")) + assert res["backend"] == "local" + assert "real article body" in res["content"] + assert time.monotonic() - t < 1.0 + + +@pytest.mark.asyncio +async def test_fetch_thin_local_falls_to_grounded(monkeypatch): + p_local_returns(monkeypatch, "Contents of https://spa.example:\n\n") # JS wall, empty body + monkeypatch.setattr(W, "resolve_gemini_api_key", lambda: "gkey") + async def p_gem(api_key, prompt, *, use_url_context): + return {"text": "rendered page text from grounding", "chunks": []} + monkeypatch.setattr(W, "gemini_grounded_call", p_gem) + + res = await fetch(FetchBody(url="https://spa.example")) + assert res["backend"] == "gemini_native" + assert "rendered page text" in res["content"] + + +@pytest.mark.asyncio +async def test_fetch_local_error_returned_as_last_resort(monkeypatch): + p_local_returns(monkeypatch, "HTTP error 403 fetching https://blocked.example", kind="error") + # no grounded keys/subs (autouse fixtures) -> all grounded skip/fail + res = await fetch(FetchBody(url="https://blocked.example")) + assert res["backend"] == "local" + assert "HTTP error 403" in res["content"] + + +@pytest.mark.asyncio +async def test_fetch_falls_to_the_archive_when_the_live_page_is_gone(monkeypatch): + """A dead link is exactly what the archive is for; the real text beats a grounded summary of nothing.""" + p_local_returns(monkeypatch, "HTTP error 404 fetching https://gone.example/post") + + async def p_snapshot(url): + return "Archived copy of https://gone.example/post (Wayback Machine snapshot from 2026-05-08)\n\nThe original text." + monkeypatch.setattr(WB, "fetch_wayback", p_snapshot) + + async def p_boom(*a, **k): + raise AssertionError("a paid fetcher must not run once the archive answered") + monkeypatch.setattr(W, "gemini_grounded_call", p_boom) + + res = await fetch(FetchBody(url="https://gone.example/post")) + assert res["backend"] == "wayback" + assert "The original text." in res["content"] + + +@pytest.mark.asyncio +async def test_browser_fetch_tier_fires_when_local_thin(monkeypatch): + p_local_returns(monkeypatch, "Contents of x:\n\ntiny") # <200 chars -> try_local returns None + patch_browser_bridge(monkeypatch, {"title": "T", "text": "the full rendered article body " * 20, "url": "https://x.example"}) + + res = await fetch(FetchBody(url="https://x.example")) + assert res["backend"] == "browser" + assert "rendered article" in res["content"] + + +@pytest.mark.asyncio +async def test_an_unreadable_binary_stops_the_cascade_instead_of_buying_a_summary(monkeypatch): + """A PNG has no text for ANY tier to find; spending a paid fetcher on it buys nothing.""" + p_local_returns(monkeypatch, "This URL is not a readable document: image/png, 219 KB of binary data.", + kind="binary") + monkeypatch.setattr(W, "resolve_gemini_api_key", lambda: "gkey") + + async def p_boom(*a, **k): + raise AssertionError("a paid fetcher must never be spent on a body with no text layer") + monkeypatch.setattr(W, "gemini_grounded_call", p_boom) + + res = await fetch(FetchBody(url="https://x.example/photo.png")) + assert res["backend"] == "local" + assert "image/png" in res["content"] + + +@pytest.mark.asyncio +async def test_a_scanned_pdf_also_stops_the_cascade(monkeypatch): + p_local_returns(monkeypatch, "This URL is a PDF (4 MB) with no extractable text layer.", + kind="pdf_unreadable") + monkeypatch.setattr(W, "resolve_gemini_api_key", lambda: "gkey") + + async def p_boom(*a, **k): + raise AssertionError("a scanned PDF is not worth a paid fetcher either") + monkeypatch.setattr(W, "gemini_grounded_call", p_boom) + + res = await fetch(FetchBody(url="https://x.example/scan.pdf")) + assert res["backend"] == "local" diff --git a/backend/tests/test_web_search_cascade.py b/backend/tests/test_web_search_cascade.py index c29bad62..17c8f011 100644 --- a/backend/tests/test_web_search_cascade.py +++ b/backend/tests/test_web_search_cascade.py @@ -1,9 +1,8 @@ -"""Fast-first, bounded web-search cascade (/api/web/search). +"""Fast-first, deadline-bounded /api/web/search cascade. -Pins the behaviour that fixes the ~75s stall and the human-speed goal: - - DuckDuckGo is tried FIRST and short-circuits the chain when it has results. - - When DDG is throttled, the chain falls over to the grounded backends. - - Every attempt is wait_for-bounded, so a hung provider can't stall the request. + - The free keyless engines are tried FIRST and short-circuit the chain. + - One engine's bot challenge falls over to the other, then to the grounded backends. + - Every attempt is bounded, so a hung provider can't stall the request. - The `primary` hint reorders only the grounded tier. - When everything fails we return an honest message, not a bogus empty result. @@ -15,53 +14,25 @@ import time import pytest -import backend.apps.agents.tools.fetch.wayback as WB -import backend.apps.agents.tools.search.search_startpage as SP import backend.apps.web.web as W from backend.apps.web.web import search, SearchBody -from backend.apps.agents.tools.web import WebSearchTool, DDGRateLimited +from backend.tests.web_cascade_fixtures import ( # noqa: F401 + allow_urls, + patch_browser_bridge, + ddg_returns, + ddg_throttled, + no_network, + startpage_returns, +) -@pytest.fixture(autouse=True) -def p_no_network(monkeypatch): - # Default everything to "unavailable / no network"; each test opts paths in. - monkeypatch.setattr(W, "resolve_gemini_api_key", lambda: None) - monkeypatch.setattr(W, "resolve_openai_api_key", lambda: None) - - async def p_no_subs(): - return set() - monkeypatch.setattr(W, "refresh_9r_connected", p_no_subs) - - async def p_empty(*a, **k): - return {} - # subscription helpers hit localhost:20128 otherwise - monkeypatch.setattr(W, "gemini_grounded_via_9router", p_empty) - monkeypatch.setattr(W, "openai_websearch_via_9router", p_empty) - - async def p_startpage_closed(query, num): - return None - monkeypatch.setattr(SP, "search_startpage", p_startpage_closed) - - async def p_no_snapshot(url): - return None - monkeypatch.setattr(WB, "fetch_wayback", p_no_snapshot) -def p_ddg_returns(monkeypatch, text): - async def p_f(query, num): - return text - monkeypatch.setattr(WebSearchTool, "search_ddg", staticmethod(p_f)) - - -def p_ddg_throttled(monkeypatch): - async def p_f(query, num): - raise DDGRateLimited(query) - monkeypatch.setattr(WebSearchTool, "search_ddg", staticmethod(p_f)) @pytest.mark.asyncio async def test_ddg_is_tried_first_and_wins(monkeypatch): - p_ddg_returns(monkeypatch, "[1] Foo\n https://foo.example") + ddg_returns(monkeypatch, "[1] Foo\n https://foo.example") # grounded would raise if reached; prove it isn't async def p_boom(*a, **k): raise AssertionError("grounded should not be called when DDG has results") @@ -77,7 +48,7 @@ async def test_ddg_is_tried_first_and_wins(monkeypatch): @pytest.mark.asyncio async def test_ddg_throttled_falls_over_to_openai(monkeypatch): - p_ddg_throttled(monkeypatch) + ddg_throttled(monkeypatch) monkeypatch.setattr(W, "resolve_openai_api_key", lambda: "okey") async def p_openai(api_key, query): @@ -91,17 +62,11 @@ async def test_ddg_throttled_falls_over_to_openai(monkeypatch): assert any("ddg" in e for e in res.get("cascade_errors", [])) -def p_startpage_returns(monkeypatch, text): - async def p_f(query, num): - return text - monkeypatch.setattr(SP, "search_startpage", p_f) - - @pytest.mark.asyncio async def test_startpage_rescues_a_ddg_challenge(monkeypatch): """Two independent engines: one operator's bot challenge must not close free search.""" - p_ddg_throttled(monkeypatch) - p_startpage_returns(monkeypatch, "[1] Rescued\n https://sp.example") + ddg_throttled(monkeypatch) + startpage_returns(monkeypatch, "[1] Rescued\n https://sp.example") async def p_boom(*a, **k): raise AssertionError("a paid backend must not run while a free engine still answers") @@ -115,7 +80,7 @@ async def test_startpage_rescues_a_ddg_challenge(monkeypatch): @pytest.mark.asyncio async def test_a_hung_grounded_attempt_is_bounded(monkeypatch): - p_ddg_throttled(monkeypatch) + ddg_throttled(monkeypatch) monkeypatch.setattr(W, "GROUNDED_TIER_SECONDS", 0.3) monkeypatch.setattr(W, "resolve_gemini_api_key", lambda: "gkey") @@ -133,7 +98,7 @@ async def test_a_hung_grounded_attempt_is_bounded(monkeypatch): @pytest.mark.asyncio async def test_primary_openai_reorders_grounded_tier(monkeypatch): - p_ddg_throttled(monkeypatch) + ddg_throttled(monkeypatch) monkeypatch.setattr(W, "resolve_gemini_api_key", lambda: "gkey") monkeypatch.setattr(W, "resolve_openai_api_key", lambda: "okey") @@ -152,7 +117,7 @@ async def test_primary_openai_reorders_grounded_tier(monkeypatch): @pytest.mark.asyncio async def test_everything_fails_is_honest_not_empty(monkeypatch): - p_ddg_throttled(monkeypatch) # no keys, no subs (from fixture) + ddg_throttled(monkeypatch) # no keys, no subs (from fixture) res = await search(SearchBody(query="obscure thing")) assert res["backend"] == "none" assert "obscure thing" in res["results"] @@ -163,7 +128,7 @@ async def test_everything_fails_is_honest_not_empty(monkeypatch): @pytest.mark.asyncio async def test_everything_fails_nudges_browser_not_retry(monkeypatch): # All-fail must hand the model the browser as an escape hatch, not a dead-end "wait and retry". - p_ddg_throttled(monkeypatch) + ddg_throttled(monkeypatch) monkeypatch.setattr(W, "resolve_openai_api_key", lambda: "okey") # configured but errors async def p_openai_boom(*a, **k): @@ -179,7 +144,7 @@ async def test_everything_fails_nudges_browser_not_retry(monkeypatch): @pytest.mark.asyncio async def test_nudge_suppressed_when_browser_denied(monkeypatch): # A session without browser-delegation tools must never be told to call CreateBrowserAgent. - p_ddg_throttled(monkeypatch) + ddg_throttled(monkeypatch) monkeypatch.setattr(W, "resolve_openai_api_key", lambda: "okey") async def p_openai_boom(*a, **k): @@ -192,94 +157,12 @@ async def test_nudge_suppressed_when_browser_denied(monkeypatch): assert "retry" not in res["results"].lower() -# -------------------------------------------------------------------------- /fetch mirrors /search: local httpx + trafilatura is the fast path, grounded fetchers are the fallback for JS/paywalled pages, every attempt is bounded. -------------------------------------------------------------------------- - -from backend.apps.web.web import fetch, FetchBody -from backend.apps.agents.tools.web import WebFetchTool -import backend.apps.agents.tools.ssrf_guard as p_ssrf - - -@pytest.fixture(autouse=True) -def p_allow_urls(monkeypatch): - async def p_ok(url): - return None - monkeypatch.setattr(p_ssrf, "assert_safe_url", p_ok) - - -def p_local_returns(monkeypatch, text): - async def p_exec(self, input_data, context): - return [{"type": "text", "text": text}] - monkeypatch.setattr(WebFetchTool, "execute", p_exec) - - -@pytest.mark.asyncio -async def test_fetch_local_first_wins_and_is_fast(monkeypatch): - big = "Contents of https://x.example:\n\n" + ("real article body " * 50) - p_local_returns(monkeypatch, big) - async def p_boom(*a, **k): - raise AssertionError("grounded fetch should not run when local has content") - monkeypatch.setattr(W, "gemini_grounded_call", p_boom) - - t = time.monotonic() - res = await fetch(FetchBody(url="https://x.example")) - assert res["backend"] == "local" - assert "real article body" in res["content"] - assert time.monotonic() - t < 1.0 - - -@pytest.mark.asyncio -async def test_fetch_thin_local_falls_to_grounded(monkeypatch): - p_local_returns(monkeypatch, "Contents of https://spa.example:\n\n") # JS wall, empty body - monkeypatch.setattr(W, "resolve_gemini_api_key", lambda: "gkey") - async def p_gem(api_key, prompt, *, use_url_context): - return {"text": "rendered page text from grounding", "chunks": []} - monkeypatch.setattr(W, "gemini_grounded_call", p_gem) - - res = await fetch(FetchBody(url="https://spa.example")) - assert res["backend"] == "gemini_native" - assert "rendered page text" in res["content"] - - -@pytest.mark.asyncio -async def test_fetch_local_error_returned_as_last_resort(monkeypatch): - p_local_returns(monkeypatch, "HTTP error 403 fetching https://blocked.example") - # no grounded keys/subs (autouse fixtures) -> all grounded skip/fail - res = await fetch(FetchBody(url="https://blocked.example")) - assert res["backend"] == "local" - assert "HTTP error 403" in res["content"] - - -@pytest.mark.asyncio -async def test_fetch_falls_to_the_archive_when_the_live_page_is_gone(monkeypatch): - """A dead link is exactly what the archive is for; the real text beats a grounded summary of nothing.""" - p_local_returns(monkeypatch, "HTTP error 404 fetching https://gone.example/post") - - async def p_snapshot(url): - return "Archived copy of https://gone.example/post (Wayback Machine snapshot from 2026-05-08)\n\nThe original text." - monkeypatch.setattr(WB, "fetch_wayback", p_snapshot) - - async def p_boom(*a, **k): - raise AssertionError("a paid fetcher must not run once the archive answered") - monkeypatch.setattr(W, "gemini_grounded_call", p_boom) - - res = await fetch(FetchBody(url="https://gone.example/post")) - assert res["backend"] == "wayback" - assert "The original text." in res["content"] - - # --- packaged-browser tier: fires when DDG throttles, skipped when no bridge --- -def p_browser_bridge(monkeypatch, result): - """Patch the offscreen-browser bridge helper; result=None simulates 'no Electron main bridge connected'.""" - async def p_f(action, params): - return result - monkeypatch.setattr(W, "p_browser_bridge", p_f) - - @pytest.mark.asyncio async def test_browser_search_tier_fires_when_ddg_throttled(monkeypatch): - p_ddg_throttled(monkeypatch) - p_browser_bridge(monkeypatch, {"engine": "ddg", "results": "[1] Real\n https://real.example", "count": 1}) + ddg_throttled(monkeypatch) + patch_browser_bridge(monkeypatch, {"engine": "ddg", "results": "[1] Real\n https://real.example", "count": 1}) async def p_boom(*a, **k): raise AssertionError("grounded should not be reached once the browser tier answers") @@ -293,8 +176,8 @@ async def test_browser_search_tier_fires_when_ddg_throttled(monkeypatch): @pytest.mark.asyncio async def test_browser_search_skipped_when_no_bridge(monkeypatch): # DDG throttled, no browser bridge -> must fall THROUGH to grounded, not crash. - p_ddg_throttled(monkeypatch) - p_browser_bridge(monkeypatch, None) + ddg_throttled(monkeypatch) + patch_browser_bridge(monkeypatch, None) monkeypatch.setattr(W, "resolve_openai_api_key", lambda: "okey") async def p_openai(api_key, query): @@ -303,18 +186,3 @@ async def test_browser_search_skipped_when_no_bridge(monkeypatch): res = await search(SearchBody(query="q")) assert res["backend"] == "openai_native" - - -@pytest.mark.asyncio -async def test_browser_fetch_tier_fires_when_local_thin(monkeypatch): - from backend.apps.web.web import fetch, FetchBody - from backend.apps.agents.tools.web import WebFetchTool - - async def p_thin(self, input_data, context): - return [{"type": "text", "text": "Contents of x:\n\ntiny"}] # <200 chars -> try_local returns None - monkeypatch.setattr(WebFetchTool, "execute", p_thin) - p_browser_bridge(monkeypatch, {"title": "T", "text": "the full rendered article body " * 20, "url": "https://x.example"}) - - res = await fetch(FetchBody(url="https://x.example")) - assert res["backend"] == "browser" - assert "rendered article" in res["content"] diff --git a/backend/tests/web_cascade_fixtures.py b/backend/tests/web_cascade_fixtures.py new file mode 100644 index 00000000..eb7e90df --- /dev/null +++ b/backend/tests/web_cascade_fixtures.py @@ -0,0 +1,66 @@ +"""Shared fixtures for the /api/web cascade tests: everything offline by default.""" + +import pytest + +import backend.apps.agents.tools.fetch.wayback as WB +import backend.apps.agents.tools.search.search_startpage as SP +import backend.apps.web.web as W +from backend.apps.agents.tools.web import DDGRateLimited, WebSearchTool +import backend.apps.agents.tools.ssrf_guard as p_ssrf + + +@pytest.fixture(autouse=True) +def no_network(monkeypatch): + # Default everything to "unavailable / no network"; each test opts paths in. + monkeypatch.setattr(W, "resolve_gemini_api_key", lambda: None) + monkeypatch.setattr(W, "resolve_openai_api_key", lambda: None) + + async def p_no_subs(): + return set() + monkeypatch.setattr(W, "refresh_9r_connected", p_no_subs) + + async def p_empty(*a, **k): + return {} + # subscription helpers hit localhost:20128 otherwise + monkeypatch.setattr(W, "gemini_grounded_via_9router", p_empty) + monkeypatch.setattr(W, "openai_websearch_via_9router", p_empty) + + async def p_startpage_closed(query, num): + return None + monkeypatch.setattr(SP, "search_startpage", p_startpage_closed) + + async def p_no_snapshot(url): + return None + monkeypatch.setattr(WB, "fetch_wayback", p_no_snapshot) + + +@pytest.fixture(autouse=True) +def allow_urls(monkeypatch): + async def p_ok(url): + return None + monkeypatch.setattr(p_ssrf, "assert_safe_url", p_ok) + + +def ddg_returns(monkeypatch, text): + async def p_f(query, num): + return text + monkeypatch.setattr(WebSearchTool, "search_ddg", staticmethod(p_f)) + + +def ddg_throttled(monkeypatch): + async def p_f(query, num): + raise DDGRateLimited(query) + monkeypatch.setattr(WebSearchTool, "search_ddg", staticmethod(p_f)) + + +def startpage_returns(monkeypatch, text): + async def p_f(query, num): + return text + monkeypatch.setattr(SP, "search_startpage", p_f) + + +def patch_browser_bridge(monkeypatch, result): + """Patch the offscreen-browser bridge; result=None simulates 'no Electron main bridge connected'.""" + async def p_f(action, params): + return result + monkeypatch.setattr(W, "p_browser_bridge", p_f)