diff --git a/backend/apps/web/web.py b/backend/apps/web/web.py index 5b1583ea..a15ef232 100644 --- a/backend/apps/web/web.py +++ b/backend/apps/web/web.py @@ -9,6 +9,7 @@ extraction in the MCP process. Mounted at `/api/web`. """ +import asyncio from contextlib import asynccontextmanager from typing import Any @@ -76,6 +77,21 @@ GEMINI_GROUNDING_MODEL = "gemini-2.5-flash" # cheapest + fastest for grounded c OPENAI_API_BASE = "https://api.openai.com/v1" OPENAI_SEARCH_MODEL = "gpt-5-mini" # cheapest model that supports web_search_preview +# Per-attempt timeouts for the search/fetch cascade. The fast-first ORDERING is +# what fixes the ~75s stall (DDG answers in ~1s so the slow grounded backends are +# rarely reached); these bounds are hang safety-nets, set just ABOVE each path's +# own httpx timeout so a normally-slow call still completes and only a truly hung +# provider (no response at all) gets cut. Grounded native search legitimately +# takes 32-42s (httpx ceiling 45s), so its leash sits at 48s, NOT below 45, or +# we'd clip the slow tail of a valid paid call. +_DDG_ATTEMPT_TIMEOUT = 6.0 # DDG answers <1s; >6s is a network hang, fall through +_GROUNDED_ATTEMPT_TIMEOUT = 48.0 # just above the providers' own 45s httpx timeout +# Local httpx + trafilatura fetch of a real page; the fast path for /fetch +# (normal pages return in <2s). Set just above WebFetchTool's own 30s httpx +# ceiling so a valid-but-slow page still completes locally instead of being +# clipped down to a grounded summary; only a truly hung server gets cut. +_LOCAL_FETCH_TIMEOUT = 32.0 + async def _gemini_grounded_call(api_key: str, prompt: str, *, use_url_context: bool) -> dict: """Call Gemini with googleSearch (+ optionally urlContext) grounding. @@ -430,72 +446,70 @@ async def search(body: SearchBody) -> dict: "backend": "openai_subscription", } - # Ordered cascade: primary's native API key first (most direct), then - # the user's connected subscriptions (free via OAuth), then the - # opposite-provider native key, then DuckDuckGo last as a guaranteed - # fallback (which is rate-limit-prone but free). - if primary == "openai": - cascade = [ - ("openai_native", try_openai), - ("openai_subscription", try_openai_subscription), - ("gemini_native", try_gemini), - ("gemini_subscription", try_gemini_subscription), - ] - elif primary in ("gemini", "google"): - cascade = [ - ("gemini_native", try_gemini), - ("gemini_subscription", try_gemini_subscription), - ("openai_native", try_openai), - ("openai_subscription", try_openai_subscription), - ] - else: - cascade = [ - ("gemini_native", try_gemini), - ("gemini_subscription", try_gemini_subscription), - ("openai_native", try_openai), - ("openai_subscription", try_openai_subscription), - ] - - for name, fn in cascade: + async def try_ddg(): + # Fast path: direct HTML search, sub-second when DDG isn't throttling us. + # Returns None on a real no-hits OR a 202 throttle so the chain falls + # through to the slower-but-grounded backends. + from backend.apps.agents.tools.web import WebSearchTool, DDGRateLimited try: - res = await fn() + text = await WebSearchTool._search_ddg(body.query, body.num_results) + except DDGRateLimited: + # Surface the throttle as a recorded error (not a silent None) so the + # caller can see WHY we fell through to a slower backend. + raise RuntimeError("DuckDuckGo rate-limited (HTTP 202)") from None + if not text: + return None + return {"query": body.query, "results": text, "backend": "ddg"} + + # Fast-first cascade: DDG leads (~1s = human speed); the 30-42s LLM-grounded + # backends are the reliable fallback when DDG is throttled or empty. The + # primary hint only reorders the grounded tier (native key before the + # same-provider subscription). Every attempt is wait_for-bounded so a slow + # or hung provider fails over fast instead of stalling the whole request. + grounded = [ + ("gemini_native", try_gemini), + ("gemini_subscription", try_gemini_subscription), + ("openai_native", try_openai), + ("openai_subscription", try_openai_subscription), + ] + if primary == "openai": + grounded = grounded[2:] + grounded[:2] + + cascade = [("ddg", try_ddg, _DDG_ATTEMPT_TIMEOUT)] + [ + (name, fn, _GROUNDED_ATTEMPT_TIMEOUT) for name, fn in grounded + ] + + for name, fn, timeout in cascade: + try: + res = await asyncio.wait_for(fn(), timeout=timeout) if res is not None: + if errors: + res["cascade_errors"] = errors return res + except asyncio.TimeoutError: + errors.append(f"{name}: timed out after {timeout:.0f}s") except Exception as e: errors.append(f"{name}: {str(e)[:150]}") - # DDG fallback. - from backend.apps.agents.tools.web import WebSearchTool - try: - tool = WebSearchTool() - parts = await tool.execute( - {"query": body.query, "num_results": body.num_results}, - None, + # Everything failed. Be honest about why instead of an empty "no results". + connected = await _refresh_9r_connected() + has_subscription = bool(connected & {"codex", "antigravity", "gemini-cli"}) + if not (gemini_key or openai_key or has_subscription): + tail = ( + "No search backend is configured and DuckDuckGo is rate-limiting this " + "network. Connect Codex / Antigravity / Gemini CLI in Settings, or add " + "an OpenAI / Gemini API key, for reliable search." + ) + else: + tail = ( + "DuckDuckGo is rate-limiting this network and every configured provider " + "errored (see details below). Wait a moment and retry." ) - except Exception as e: - raise HTTPException(status_code=500, detail=f"Search failed: {e}") - - text = _join_text(parts) - hint = "" - if text.startswith("No search results found"): - connected = await _refresh_9r_connected() - has_subscription = bool(connected & {"codex", "antigravity", "gemini-cli"}) - if not (gemini_key or openai_key or has_subscription): - hint = ( - "\n\n(DuckDuckGo returned no results; likely rate-limiting this IP. " - "Connect Codex / Antigravity / Gemini CLI in Settings, or add an " - "OpenAI / Gemini API key, for reliable native search.)" - ) - else: - hint = ( - "\n\n(DuckDuckGo returned no results and the connected providers " - "didn't return useful results either; try rephrasing the query.)" - ) return { "query": body.query, - "results": text + hint, - "backend": "ddg", - **({"cascade_errors": errors} if errors else {}), + "results": f"No results for: {body.query}\n\n{tail}", + "backend": "none", + "cascade_errors": errors, } @@ -571,28 +585,58 @@ async def fetch(body: FetchBody) -> dict: "backend": "openai_subscription", } - if primary == "openai": - cascade = [try_openai, try_openai_subscription, try_gemini, try_gemini_subscription] - else: - cascade = [try_gemini, try_gemini_subscription, try_openai, try_openai_subscription] + # Remembered so a thin/errored local read is still returned as the last + # resort if every grounded fetcher also fails (never worse than before). + local_text: str | None = None - for fn in cascade: - try: - res = await fn() - if res is not None: - return res - except Exception: - continue - - # Local httpx + trafilatura fallback. - from backend.apps.agents.tools.web import WebFetchTool - try: - tool = WebFetchTool() - parts = await tool.execute( - {"url": body.url, "prompt": body.prompt or ""}, - None, + async def try_local(): + # Fast path: direct httpx + trafilatura, sub-second to a few seconds and + # 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 to the grounded fetchers that can render them. + nonlocal local_text + from backend.apps.agents.tools.web import WebFetchTool + parts = await WebFetchTool().execute( + {"url": body.url, "prompt": body.prompt or ""}, None, ) - except Exception as e: - raise HTTPException(status_code=500, detail=f"Fetch failed: {e}") + text = _join_text(parts) + local_text = text + if text.startswith(("HTTP error", "Error fetching", "Refused to fetch")): + return None + body_text = text.split("\n\n", 1)[-1] if "\n\n" in text else text + if len(body_text.strip()) < 200: + return None + return {"url": body.url, "content": text, "backend": "local"} - return {"url": body.url, "content": _join_text(parts), "backend": "local"} + grounded = [ + ("gemini_native", try_gemini), + ("gemini_subscription", try_gemini_subscription), + ("openai_native", try_openai), + ("openai_subscription", try_openai_subscription), + ] + if primary == "openai": + grounded = grounded[2:] + grounded[:2] + + cascade = [("local", try_local, _LOCAL_FETCH_TIMEOUT)] + [ + (name, fn, _GROUNDED_ATTEMPT_TIMEOUT) for name, fn in grounded + ] + + errors: list[str] = [] + for name, fn, timeout in cascade: + try: + res = await asyncio.wait_for(fn(), timeout=timeout) + if res is not None: + if errors: + res["cascade_errors"] = errors + return res + except asyncio.TimeoutError: + errors.append(f"{name}: timed out after {timeout:.0f}s") + except Exception as e: + errors.append(f"{name}: {str(e)[:150]}") + + # Grounded all failed; hand back whatever the local read got (even an error + # string is useful signal) rather than nothing. + if local_text is not None: + return {"url": body.url, "content": local_text, "backend": "local", + **({"cascade_errors": errors} if errors else {})} + raise HTTPException(status_code=502, detail=f"Fetch failed for {body.url}: " + "; ".join(errors)[:400]) diff --git a/backend/tests/test_web_search_cascade.py b/backend/tests/test_web_search_cascade.py new file mode 100644 index 00000000..8eb46537 --- /dev/null +++ b/backend/tests/test_web_search_cascade.py @@ -0,0 +1,188 @@ +"""Fast-first, bounded web-search cascade (/api/web/search). + +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 `primary` hint reorders only the grounded tier. + - When everything fails we return an honest message, not a bogus empty result. + +All providers are mocked, so the test is deterministic and offline. +""" + +import asyncio +import time + +import pytest + +import backend.apps.web.web as W +from backend.apps.web.web import search, SearchBody +from backend.apps.agents.tools.web import WebSearchTool, DDGRateLimited + + +@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 _no_subs(): + return set() + monkeypatch.setattr(W, "_refresh_9r_connected", _no_subs) + + async def _empty(*a, **k): + return {} + # subscription helpers hit localhost:20128 otherwise + monkeypatch.setattr(W, "_gemini_grounded_via_9router", _empty) + monkeypatch.setattr(W, "_openai_websearch_via_9router", _empty) + + +def _ddg_returns(monkeypatch, text): + async def _f(query, num): + return text + monkeypatch.setattr(WebSearchTool, "_search_ddg", staticmethod(_f)) + + +def _ddg_throttled(monkeypatch): + async def _f(query, num): + raise DDGRateLimited(query) + monkeypatch.setattr(WebSearchTool, "_search_ddg", staticmethod(_f)) + + +@pytest.mark.asyncio +async def test_ddg_is_tried_first_and_wins(monkeypatch): + _ddg_returns(monkeypatch, "[1] Foo\n https://foo.example") + # grounded would raise if reached; prove it isn't + async def _boom(*a, **k): + raise AssertionError("grounded should not be called when DDG has results") + monkeypatch.setattr(W, "_gemini_grounded_call", _boom) + + t = time.monotonic() + res = await search(SearchBody(query="foo")) + assert res["backend"] == "ddg" + assert "foo.example" in res["results"] + assert "cascade_errors" not in res + assert time.monotonic() - t < 1.0 + + +@pytest.mark.asyncio +async def test_ddg_throttled_falls_over_to_openai(monkeypatch): + _ddg_throttled(monkeypatch) + monkeypatch.setattr(W, "_resolve_openai_api_key", lambda: "okey") + + async def _openai(api_key, query): + return {"text": "grounded answer", "chunks": [("Title", "https://u.example")]} + monkeypatch.setattr(W, "_openai_websearch", _openai) + + res = await search(SearchBody(query="x")) + assert res["backend"] == "openai_native" + assert "u.example" in res["results"] + # DDG's throttle is recorded so the caller knows why we fell through + assert any("ddg" in e for e in res.get("cascade_errors", [])) + + +@pytest.mark.asyncio +async def test_a_hung_grounded_attempt_is_bounded(monkeypatch): + _ddg_throttled(monkeypatch) + monkeypatch.setattr(W, "_GROUNDED_ATTEMPT_TIMEOUT", 0.3) + monkeypatch.setattr(W, "_resolve_gemini_api_key", lambda: "gkey") + + async def _hangs(*a, **k): + await asyncio.sleep(30) + monkeypatch.setattr(W, "_gemini_grounded_call", _hangs) + + t = time.monotonic() + res = await search(SearchBody(query="x")) + elapsed = time.monotonic() - t + assert elapsed < 2.0, f"hung provider should be bounded, took {elapsed:.2f}s" + assert res["backend"] == "none" + assert any("timed out" in e for e in res["cascade_errors"]) + + +@pytest.mark.asyncio +async def test_primary_openai_reorders_grounded_tier(monkeypatch): + _ddg_throttled(monkeypatch) + monkeypatch.setattr(W, "_resolve_gemini_api_key", lambda: "gkey") + monkeypatch.setattr(W, "_resolve_openai_api_key", lambda: "okey") + + async def _gem(*a, **k): + return {"text": "GEM", "chunks": [("g", "https://gem.example")]} + async def _oai(api_key, query): + return {"text": "OAI", "chunks": [("o", "https://oai.example")]} + monkeypatch.setattr(W, "_gemini_grounded_call", _gem) + monkeypatch.setattr(W, "_openai_websearch", _oai) + + res = await search(SearchBody(query="x", primary="openai")) + # openai must be tried before gemini when it's the primary + assert res["backend"] == "openai_native" + assert "oai.example" in res["results"] + + +@pytest.mark.asyncio +async def test_everything_fails_is_honest_not_empty(monkeypatch): + _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"] + # points the user at how to get reliable search + assert "Settings" in res["results"] or "API key" in res["results"] + + +# -------------------------------------------------------------------------- +# /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 _ssrf + + +@pytest.fixture(autouse=True) +def _allow_urls(monkeypatch): + async def _ok(url): + return None + monkeypatch.setattr(_ssrf, "assert_safe_url", _ok) + + +def _local_returns(monkeypatch, text): + async def _exec(self, input_data, context): + return [{"type": "text", "text": text}] + monkeypatch.setattr(WebFetchTool, "execute", _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) + _local_returns(monkeypatch, big) + async def _boom(*a, **k): + raise AssertionError("grounded fetch should not run when local has content") + monkeypatch.setattr(W, "_gemini_grounded_call", _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): + _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 _gem(api_key, prompt, *, use_url_context): + return {"text": "rendered page text from grounding", "chunks": []} + monkeypatch.setattr(W, "_gemini_grounded_call", _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): + _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"]