[eric] web: bound the search/fetch cascade with one wall-clock deadline so the later tiers are reachable

backend/apps/agents/web_mcp_server.py

@@ -87,7 +87,11 @@ def send_response(id_, result=None, error=None):
     sys.stdout.flush()
 
 
-def p_post(url: str, body: dict, timeout: float = 60.0) -> dict:
+# The backend bounds its own cascade at 60s and always answers within it (with an honest "here is why every backend failed" body). Waiting slightly longer than that means the useful answer wins; the old 45s cap aborted the call BEFORE the later tiers could even be reached, so search reliability came down to whether an early tier won the race.
+TOOL_TIMEOUT = 75.0
+
+
+def p_post(url: str, body: dict, timeout: float = TOOL_TIMEOUT) -> dict:
     payload = json.dumps(body).encode()
     headers = {"Content-Type": "application/json"}
     if BACKEND_AUTH:
@@ -118,7 +122,7 @@ def handle_tool_call(tool_name: str, arguments: dict) -> dict:
         body = {"query": query, "num_results": num, "browser_ok": BROWSER_OK}
         if PRIMARY_HINT:
             body["primary"] = PRIMARY_HINT
-        r = p_post(SEARCH_URL, body, timeout=45.0)
+        r = p_post(SEARCH_URL, body, timeout=TOOL_TIMEOUT)
         if "error" in r:
             return {"content": [{"type": "text", "text": f"Search failed: {r['error']}"}], "isError": True}
         results = r.get("results", "")
@@ -140,7 +144,7 @@ def handle_tool_call(tool_name: str, arguments: dict) -> dict:
             body["prompt"] = str(prompt)
         if PRIMARY_HINT:
             body["primary"] = PRIMARY_HINT
-        r = p_post(FETCH_URL, body, timeout=45.0)
+        r = p_post(FETCH_URL, body, timeout=TOOL_TIMEOUT)
         if "error" in r:
             return {"content": [{"type": "text", "text": f"Fetch failed: {r['error']}"}], "isError": True}
         content = r.get("content", "")

backend/apps/web/cascade.py

@@ -0,0 +1,62 @@
+"""Deadline-bounded tier runner shared by /api/web/search and /api/web/fetch.
+
+The cascades used to sum their per-tier leashes to 244s (search) and 270s
+(fetch) while the MCP shim calling them gave up at 45s, so the later tiers
+could never run at all: whether a search worked came down to whether an early
+tier happened to win the race before the client-side guillotine. One wall-clock
+deadline for the whole cascade makes that unrepresentable. A tier can only ever
+spend what is LEFT of the budget, so a slow tier cannot starve the ones behind
+it, and the endpoint always answers within the deadline."""
+
+import asyncio
+from typing import Awaitable, Callable, Dict, List, Optional
+
+from pydantic import BaseModel, ConfigDict, Field, InstanceOf
+from typeguard import typechecked
+
+# A tier handed less than this has no realistic chance, and reporting it as a timeout would be a lie; we say the budget ran out instead.
+MIN_TIER_SECONDS = 3.0
+
+
+class CascadeTier(BaseModel):
+    model_config = ConfigDict(validate_assignment=True)
+
+    name: str
+    run: InstanceOf[Callable[[], Awaitable[Optional[Dict]]]]
+    budget: float
+
+
+class CascadeOutcome(BaseModel):
+    model_config = ConfigDict(validate_assignment=True)
+
+    result: Optional[Dict] = None
+    errors: List[str] = Field(default_factory=list)
+
+
+@typechecked
+async def run_cascade(tiers: List[CascadeTier], total_budget: float) -> CascadeOutcome:
+    """Run tiers in order until one returns a result or the budget is gone."""
+    loop = asyncio.get_running_loop()
+    deadline = loop.time() + total_budget
+    errors: List[str] = []
+
+    for tier in tiers:
+        remaining = deadline - loop.time()
+        if remaining < MIN_TIER_SECONDS:
+            skipped = [t.name for t in tiers[tiers.index(tier):]]
+            errors.append(
+                f"{total_budget:.0f}s cascade budget spent; not attempted: {', '.join(skipped)}"
+            )
+            break
+        slice_seconds = min(tier.budget, remaining)
+        try:
+            result = await asyncio.wait_for(tier.run(), timeout=slice_seconds)
+        except asyncio.TimeoutError:
+            errors.append(f"{tier.name}: timed out after {slice_seconds:.0f}s")
+        except Exception as exc:
+            errors.append(f"{tier.name}: {str(exc)[:150]}")
+        else:
+            if result is not None:
+                return CascadeOutcome(result=result, errors=errors)
+
+    return CascadeOutcome(result=None, errors=errors)

backend/apps/web/grounded.py

@@ -0,0 +1,265 @@
+"""Provider-grounded search/fetch backends for the /api/web cascade.
+
+These are the PAID tiers: the user's own Gemini / OpenAI key, or the same
+providers reached through a 9Router subscription. They are slow (tens of
+seconds) but render and reason over pages, so they sit behind the free
+keyless tiers and only run when those come up empty."""
+
+import time
+from typing import Dict, List, Optional, Set, Tuple
+
+import httpx
+from typeguard import typechecked
+
+GEMINI_API_BASE = "https://generativelanguage.googleapis.com/v1beta"
+GEMINI_GROUNDING_MODEL = "gemini-2.5-flash"  # cheapest + fastest for grounded calls
+
+OPENAI_API_BASE = "https://api.openai.com/v1"
+OPENAI_SEARCH_MODEL = "gpt-5-mini"  # cheapest model that supports web_search_preview
+
+NINE_ROUTER_MESSAGES_URL = "http://localhost:20128/v1/messages"
+
+
+@typechecked
+async def gemini_grounded_call(api_key: str, prompt: str, *, use_url_context: bool) -> Dict:
+    """Call Gemini with googleSearch (+ optionally urlContext) grounding.
+
+    Returns {"text": grounded_answer, "chunks": [(title, uri), ...],
+             "queries": [...]} or raises httpx.HTTPError on failure.
+    """
+    tools: List[Dict] = [{"googleSearch": {}}]
+    if use_url_context:
+        tools.append({"urlContext": {}})
+    body = {
+        "contents": [{"role": "user", "parts": [{"text": prompt}]}],
+        "tools": tools,
+        "generationConfig": {"thinkingConfig": {"thinkingBudget": 0}},
+    }
+    url = f"{GEMINI_API_BASE}/models/{GEMINI_GROUNDING_MODEL}:generateContent"
+    async with httpx.AsyncClient(timeout=45.0) as client:
+        r = await client.post(
+            url,
+            headers={"x-goog-api-key": api_key, "Content-Type": "application/json"},
+            json=body,
+        )
+        r.raise_for_status()
+        data = r.json()
+
+    cand = (data.get("candidates") or [{}])[0]
+    text = "".join(
+        p.get("text", "") for p in (cand.get("content", {}).get("parts") or [])
+        if isinstance(p, dict)
+    )
+    gm = cand.get("groundingMetadata") or {}
+    chunks: List[Tuple[str, str]] = []
+    for gc in (gm.get("groundingChunks") or []):
+        web = (gc or {}).get("web") or {}
+        uri = web.get("uri") or web.get("url") or ""
+        title = web.get("title") or uri
+        if uri:
+            chunks.append((title, uri))
+    queries = gm.get("webSearchQueries") or []
+    return {"text": text, "chunks": chunks, "queries": queries}
+
+
+@typechecked
+def format_grounded_as_search_results(grounded: Dict, query: str) -> str:
+    """Format Gemini grounding output to match WebSearchTool's text shape."""
+    lines: List[str] = []
+    chunks = grounded.get("chunks") or []
+    for i, (title, uri) in enumerate(chunks[:10], start=1):
+        lines.append(f"[{i}] {title}\n    {uri}")
+    text = grounded.get("text") or ""
+    if text:
+        lines.append("\n" + text)
+    if not lines:
+        return f"No search results found for: {query}"
+    return "\n\n".join(lines)
+
+
+@typechecked
+def format_grounded_as_fetch(grounded: Dict, url: str) -> str:
+    """Format Gemini urlContext output to match WebFetchTool's text shape."""
+    parts = [f"Contents of {url}:", ""]
+    text = grounded.get("text") or ""
+    if text:
+        parts.append(text)
+    chunks = grounded.get("chunks") or []
+    if chunks:
+        parts.append("\nCited sources:")
+        for i, (title, uri) in enumerate(chunks[:5], start=1):
+            parts.append(f"  [{i}] {title}; {uri}")
+    return "\n".join(parts)
+
+
+@typechecked
+def resolve_gemini_api_key() -> Optional[str]:
+    """Pull the AI Studio API key from settings, or None."""
+    try:
+        from backend.apps.settings.settings import load_settings
+        s = load_settings()
+        return getattr(s, "google_api_key", None) or None
+    except Exception:
+        return None
+
+
+@typechecked
+def resolve_openai_api_key() -> Optional[str]:
+    try:
+        from backend.apps.settings.settings import load_settings
+        s = load_settings()
+        return getattr(s, "openai_api_key", None) or None
+    except Exception:
+        return None
+
+
+# Cache of which 9Router subscriptions are connected. Refreshed rather than hit on every search call; 9Router's /api/providers is fast but not free and we already query it from many places.
+p_nine_router_connected: Set[str] = set()
+p_nine_router_cache_at: float = 0.0
+
+
+@typechecked
+async def refresh_9r_connected() -> Set[str]:
+    """The currently-active 9Router subscription providers, cached 20s."""
+    global p_nine_router_connected, p_nine_router_cache_at
+    now = time.time()
+    if now - p_nine_router_cache_at < 20.0:
+        return p_nine_router_connected
+    try:
+        from backend.apps.nine_router import is_running as p_9r_running, get_providers as p_9r_providers
+        if not p_9r_running():
+            p_nine_router_connected = set()
+        else:
+            conns = await p_9r_providers()
+            p_nine_router_connected = {
+                c.get("provider")
+                for c in conns
+                if isinstance(c, dict) and c.get("isActive") and c.get("provider")
+            }
+        p_nine_router_cache_at = now
+    except Exception:
+        # Cache stays; best-effort.
+        pass
+    return p_nine_router_connected
+
+
+@typechecked
+async def p_nine_router_text(body: Dict) -> str:
+    """POST an Anthropic-shape body to 9Router and concatenate its text blocks."""
+    async with httpx.AsyncClient(timeout=20.0) as client:
+        r = await client.post(
+            NINE_ROUTER_MESSAGES_URL,
+            json=body,
+            headers={"x-api-key": "9router", "anthropic-version": "2023-06-01"},
+        )
+        if r.status_code != 200:
+            return ""
+        data = r.json()
+    return "".join(
+        block.get("text", "")
+        for block in (data.get("content") or [])
+        if isinstance(block, dict) and block.get("type") == "text"
+    )
+
+
+@typechecked
+async def gemini_grounded_via_9router(prompt: str, use_url_context: bool) -> Dict:
+    """Grounded Gemini through the user's OAuth subscription instead of an AI Studio key.
+
+    Shapes its return like `gemini_grounded_call` so the formatters work
+    unchanged. 9Router doesn't surface citations as a structured field
+    uniformly across providers, so we hand back text-only."""
+    connected = await refresh_9r_connected()
+    if "gemini-cli" in connected:
+        model = "gc/gemini-2.5-flash"
+    elif "antigravity" in connected:
+        model = "ag/gemini-3-flash"
+    else:
+        return {}
+    sys_prompt = (
+        "You fetch URLs and return concise summaries with citations."
+        if use_url_context
+        else "You search the web and return concise grounded answers with "
+             "source citations. Always cite the URLs you used."
+    )
+    text = await p_nine_router_text({
+        "model": model,
+        "max_tokens": 1024,
+        "system": sys_prompt,
+        "messages": [{"role": "user", "content": prompt}],
+    })
+    return {"text": text, "chunks": []}
+
+
+@typechecked
+async def openai_websearch_via_9router(query: str) -> Dict:
+    """Same idea for OpenAI, through the user's Codex 9Router connection."""
+    connected = await refresh_9r_connected()
+    if "codex" not in connected:
+        return {}
+    text = await p_nine_router_text({
+        "model": "cx/gpt-5.4-mini",
+        "max_tokens": 1024,
+        "system": (
+            "You search the web and return concise grounded answers "
+            "with source citations. Always cite the URLs you used."
+        ),
+        "messages": [{"role": "user", "content": f"Search the web for: {query}"}],
+    })
+    return {"text": text, "chunks": []}
+
+
+@typechecked
+def p_parse_openai_response(data: Dict) -> Dict:
+    """Pull output_text + url_citation annotations out of a Responses API body."""
+    text_parts: List[str] = []
+    chunks: List[Tuple[str, str]] = []
+    for item in (data.get("output") or []):
+        if not isinstance(item, dict):
+            continue
+        for content in (item.get("content") or []):
+            if not isinstance(content, dict):
+                continue
+            if content.get("type") == "output_text":
+                text_parts.append(content.get("text", ""))
+            for ann in (content.get("annotations") or []):
+                if isinstance(ann, dict) and ann.get("type") == "url_citation":
+                    uri = ann.get("url", "")
+                    if uri:
+                        chunks.append((ann.get("title", uri), uri))
+    return {"text": "".join(text_parts), "chunks": chunks}
+
+
+@typechecked
+async def p_openai_responses(api_key: str, prompt: str) -> Dict:
+    async with httpx.AsyncClient(timeout=45.0) as client:
+        r = await client.post(
+            f"{OPENAI_API_BASE}/responses",
+            headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},
+            json={
+                "model": OPENAI_SEARCH_MODEL,
+                "input": prompt,
+                "tools": [{"type": "web_search_preview"}],
+            },
+        )
+        r.raise_for_status()
+        return p_parse_openai_response(r.json())
+
+
+@typechecked
+async def openai_websearch(api_key: str, query: str) -> Dict:
+    """Call OpenAI Responses API with the web_search_preview tool."""
+    grounded = await p_openai_responses(
+        api_key, f"Search the web for: {query}\n\nReturn a concise summary. Cite sources.",
+    )
+    grounded["queries"] = [query]
+    return grounded
+
+
+@typechecked
+async def openai_urlfetch(api_key: str, url: str, prompt: Optional[str]) -> Dict:
+    """Use OpenAI's web_search_preview to fetch/summarize a specific URL."""
+    prompt_text = f"Fetch and summarize the content at: {url}"
+    if prompt:
+        prompt_text += f"\n\nFocus on: {prompt}"
+    return await p_openai_responses(api_key, prompt_text)

backend/apps/web/web.py

@@ -1,22 +1,35 @@
 """Web search + fetch sub-app.
 
-Thin HTTP wrappers around `WebSearchTool` and `WebFetchTool` from
-`backend.apps.agents.tools.web`. Exists so the in-process MCP server
-(`backend.apps.agents.web_mcp_server`) can proxy tool calls to the
-backend instead of re-implementing DuckDuckGo scraping + trafilatura
-extraction in the MCP process.
+Thin HTTP wrappers around the keyless search rungs and `WebFetchTool` from
+`backend.apps.agents.tools`. Exists so the in-process MCP server
+(`backend.apps.agents.web_mcp_server`) can proxy tool calls to the backend
+instead of re-implementing scraping + trafilatura extraction in the MCP
+process.
 
 Mounted at `/api/web`.
 """
 
-import asyncio
 from contextlib import asynccontextmanager
-from typing import Any
+from typing import Any, Dict, List, Optional
+from uuid import uuid4
 
 from fastapi import HTTPException
 from pydantic import BaseModel, Field
 from typeguard import typechecked
 
+from backend.apps.web.cascade import CascadeTier, run_cascade
+from backend.apps.web.grounded import (
+    format_grounded_as_fetch,
+    format_grounded_as_search_results,
+    gemini_grounded_call,
+    gemini_grounded_via_9router,
+    openai_urlfetch,
+    openai_websearch,
+    openai_websearch_via_9router,
+    refresh_9r_connected,
+    resolve_gemini_api_key,
+    resolve_openai_api_key,
+)
 from backend.config.Apps import SubApp
 
 
@@ -35,48 +48,45 @@ class SearchBody(BaseModel):
     query: str = Field(..., description="The search query.")
     num_results: int = Field(5, ge=1, le=10, description="Max results to return.")
     # Hint from the MCP server about which primary provider the session is using. Lets us route to that provider's native search tool (Gemini googleSearch, OpenAI web_search_preview) when available, costs come out of the user's existing primary budget.
-    primary: str | None = Field(None, description="Primary provider hint: 'gemini' | 'openai' | 'anthropic' | None")
+    primary: Optional[str] = Field(None, description="Primary provider hint: 'gemini' | 'openai' | 'anthropic' | None")
     # Set by the openswarm-web shim from OPENSWARM_BROWSER_OK; the browser-fallback nudge must never fire in a session without browser-delegation tools.
     browser_ok: bool = Field(False, description="Whether this session has browser-delegation tools available.")
 
 
 class FetchBody(BaseModel):
     url: str = Field(..., description="The URL to fetch.")
-    prompt: str | None = Field(None, description="Optional context hint.")
-    primary: str | None = Field(None, description="Primary provider hint.")
+    prompt: Optional[str] = Field(None, description="Optional context hint.")
+    primary: Optional[str] = Field(None, description="Primary provider hint.")
 
 
-# --------------------------------------------------------------------------- Helper; extract plain text from a tool's structured output list ---------------------------------------------------------------------------
+# --------------------------------------------------------------------------- Budgets ---------------------------------------------------------------------------
+
+# Whole-cascade wall clock. Nothing below can push an endpoint past this, and the MCP shim waits LONGER (see web_mcp_server) so our honest "here is why every backend failed" answer beats a client-side abort.
+SEARCH_BUDGET_SECONDS = 60.0
+FETCH_BUDGET_SECONDS = 60.0
+
+KEYLESS_TIER_SECONDS = 8.0        # a search frontend answers in ~1s; >8s is a hang
+BROWSER_TIER_SECONDS = 12.0       # the main-bridge send has its own per-action timeout
+GROUNDED_TIER_SECONDS = 45.0      # grounded native search legitimately takes 30-42s
+LOCAL_FETCH_TIER_SECONDS = 15.0   # normal pages return in <2s
 
 
-def p_join_text(parts: list[dict[str, Any]]) -> str:
-    out = []
+# --------------------------------------------------------------------------- Helpers ---------------------------------------------------------------------------
+
+
+@typechecked
+def p_join_text(parts: List[Dict[str, Any]]) -> 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)
 
 
-# --------------------------------------------------------------------------- Endpoints ---------------------------------------------------------------------------
-
-
-GEMINI_API_BASE = "https://generativelanguage.googleapis.com/v1beta"
-GEMINI_GROUNDING_MODEL = "gemini-2.5-flash"  # cheapest + fastest for grounded calls
-
-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.
-P_DDG_ATTEMPT_TIMEOUT = 6.0       # DDG answers <1s; >6s is a network hang, fall through
-P_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.
-P_LOCAL_FETCH_TIMEOUT = 32.0
-P_BROWSER_TIER_TIMEOUT = 46.0  # main-bridge send has its own per-action timeout; this outer wait_for is just a backstop above it
-
 # 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.
-async def p_browser_bridge(action: str, params: dict) -> dict | None:
+@typechecked
+async def p_browser_bridge(action: str, params: Dict) -> Optional[Dict]:
     from backend.apps.agents.core.ws_manager import ws_manager
-    from uuid import uuid4
     if ws_manager.main_connection is None:
         return None
     res = await ws_manager.send_main_command(uuid4().hex, action, params)
@@ -85,512 +95,158 @@ async def p_browser_bridge(action: str, params: dict) -> dict | None:
     return res
 
 
-# When every search backend fails, point the model at the in-product browser (always-on CreateBrowserAgent tool) instead of telling it to "wait and retry", which it can't do and just relays as a dead end. The real Chromium renders pages and isn't subject to the DDG scrape throttle.
+# When every search backend fails, point the model at the in-product browser (always-on CreateBrowserAgent tool) instead of telling it to "wait and retry", which it can't do and just relays as a dead end. The real Chromium renders pages and isn't subject to the scrape throttle.
+@typechecked
 def p_browser_fallback_nudge(query: str) -> str:
     return (
         "Don't stop here: fall back to the in-product browser, which renders real pages and "
-        "isn't subject to this rate limit. Call CreateBrowserAgent with a task like: "
+        "isn't subject to this block. Call CreateBrowserAgent with a task like: "
         f'"Search the web for: {query}. Report the top results with their titles and URLs, '
         'plus a direct answer if you find one."'
     )
 
 
-async def p_gemini_grounded_call(api_key: str, prompt: str, *, use_url_context: bool) -> dict:
-    """Call Gemini with googleSearch (+ optionally urlContext) grounding.
-
-    Returns {"text": grounded_answer, "chunks": [(title, uri), ...],
-             "queries": [...]} or raises httpx.HTTPError on failure.
-    """
-    import httpx
-    tools = [{"googleSearch": {}}]
-    if use_url_context:
-        tools.append({"urlContext": {}})
-    body = {
-        "contents": [{"role": "user", "parts": [{"text": prompt}]}],
-        "tools": tools,
-        "generationConfig": {"thinkingConfig": {"thinkingBudget": 0}},
-    }
-    url = f"{GEMINI_API_BASE}/models/{GEMINI_GROUNDING_MODEL}:generateContent"
-    async with httpx.AsyncClient(timeout=45.0) as client:
-        r = await client.post(
-            url,
-            headers={"x-goog-api-key": api_key, "Content-Type": "application/json"},
-            json=body,
-        )
-        r.raise_for_status()
-        data = r.json()
-
-    cand = (data.get("candidates") or [{}])[0]
-    text = "".join(
-        p.get("text", "") for p in (cand.get("content", {}).get("parts") or [])
-        if isinstance(p, dict)
-    )
-    gm = cand.get("groundingMetadata") or {}
-    chunks = []
-    for gc in (gm.get("groundingChunks") or []):
-        web = (gc or {}).get("web") or {}
-        uri = web.get("uri") or web.get("url") or ""
-        title = web.get("title") or uri
-        if uri:
-            chunks.append((title, uri))
-    queries = gm.get("webSearchQueries") or []
-    return {"text": text, "chunks": chunks, "queries": queries}
+@typechecked
+def p_grounded_tiers(kind: str, primary: Optional[str], runners: Dict[str, Any]) -> List[CascadeTier]:
+    """Order the four grounded backends, promoting the session's own primary."""
+    names = ["gemini_native", "gemini_subscription", "openai_native", "openai_subscription"]
+    if (primary or "").lower() == "openai":
+        names = names[2:] + names[:2]
+    return [
+        CascadeTier(name=f"{kind}:{n}", run=runners[n], budget=GROUNDED_TIER_SECONDS)
+        for n in names
+    ]
 
 
-def p_format_grounded_as_search_results(grounded: dict, query: str) -> str:
-    """Format Gemini grounding output to match WebSearchTool's text shape."""
-    lines = []
-    chunks = grounded.get("chunks") or []
-    for i, (title, uri) in enumerate(chunks[:10], start=1):
-        lines.append(f"[{i}] {title}\n    {uri}")
-    text = grounded.get("text") or ""
-    if text:
-        lines.append("\n" + text)
-    if not lines:
-        return f"No search results found for: {query}"
-    return "\n\n".join(lines)
-
-
-def p_format_grounded_as_fetch(grounded: dict, url: str) -> str:
-    """Format Gemini urlContext output to match WebFetchTool's text shape."""
-    parts = [f"Contents of {url}:", ""]
-    text = grounded.get("text") or ""
-    if text:
-        parts.append(text)
-    chunks = grounded.get("chunks") or []
-    if chunks:
-        parts.append("\nCited sources:")
-        for i, (title, uri) in enumerate(chunks[:5], start=1):
-            parts.append(f"  [{i}] {title}; {uri}")
-    return "\n".join(parts)
-
-
-def p_resolve_gemini_api_key() -> str | None:
-    """Pull the AI Studio API key from settings, or None."""
-    try:
-        from backend.apps.settings.settings import load_settings
-        s = load_settings()
-        return getattr(s, "google_api_key", None) or None
-    except Exception:
-        return None
-
-
-def p_resolve_openai_api_key() -> str | None:
-    try:
-        from backend.apps.settings.settings import load_settings
-        s = load_settings()
-        return getattr(s, "openai_api_key", None) or None
-    except Exception:
-        return None
-
-
-# Cache of which 9Router subscriptions are connected. Refreshed via `_refresh_9r_connected()` rather than hit on every search call, 9Router's /api/providers is fast but not free, and we already query it from many places.
-P_NINE_ROUTER_CONNECTED: set[str] = set()
-P_NINE_ROUTER_CACHE_AT: float = 0.0
-
-
-async def p_refresh_9r_connected() -> set[str]:
-    """Return the set of currently-active 9Router subscription providers
-    (e.g. {"claude", "codex", "antigravity", "gemini-cli"}). Cached for
-    20s to keep search/fetch endpoints snappy."""
-    global P_NINE_ROUTER_CONNECTED, P_NINE_ROUTER_CACHE_AT
-    import time as p_t
-    now = p_t.time()
-    if now - P_NINE_ROUTER_CACHE_AT < 20.0:
-        return P_NINE_ROUTER_CONNECTED
-    try:
-        from backend.apps.nine_router import is_running as p_9r_running, get_providers as p_9r_providers
-        if not p_9r_running():
-            P_NINE_ROUTER_CONNECTED = set()
-        else:
-            conns = await p_9r_providers()
-            P_NINE_ROUTER_CONNECTED = {
-                c.get("provider")
-                for c in conns
-                if isinstance(c, dict) and c.get("isActive") and c.get("provider")
-            }
-        P_NINE_ROUTER_CACHE_AT = now
-    except Exception:
-        # Cache stays; best-effort.
-        pass
-    return P_NINE_ROUTER_CONNECTED
-
-
-async def p_gemini_grounded_via_9router(prompt: str, use_url_context: bool) -> dict:
-    """Call 9Router's /v1/messages endpoint with a Gemini model so the
-    user's OAuth subscription (Gemini CLI or Antigravity) covers the
-    search call instead of needing a separate AI Studio API key.
-
-    Routes through Anthropic-shape against 9Router's translator. We
-    request a tool result naturally; the translator surfaces grounded
-    URIs as text + cited sources in the response body. Format-shape
-    matches the existing `_gemini_grounded_call` so downstream
-    `_format_grounded_as_search_results` works unchanged."""
-    import httpx
-    # Prefer Gemini CLI (broader model coverage). Fall back to Antigravity if CLI isn't connected.
-    connected = await p_refresh_9r_connected()
-    if "gemini-cli" in connected:
-        model = "gc/gemini-2.5-flash"
-    elif "antigravity" in connected:
-        model = "ag/gemini-3-flash"
-    else:
-        return {}
-
-    sys_prompt = (
-        "You search the web and return concise grounded answers with "
-        "source citations. Always cite the URLs you used."
-        if not use_url_context
-        else "You fetch URLs and return concise summaries with citations."
-    )
-    body = {
-        "model": model,
-        "max_tokens": 1024,
-        "system": sys_prompt,
-        "messages": [{"role": "user", "content": prompt}],
-    }
-    async with httpx.AsyncClient(timeout=20.0) as client:
-        r = await client.post(
-            "http://localhost:20128/v1/messages",
-            json=body,
-            headers={"x-api-key": "9router", "anthropic-version": "2023-06-01"},
-        )
-        if r.status_code != 200:
-            return {}
-        data = r.json()
-    # Synthesize a grounded shape so the existing formatter works: _format_grounded_as_search_results expects {"text": str, "chunks": [(title, uri), ...]}. 9Router doesn't surface citations as a structured field uniformly across providers, so we hand back text-only and let the formatter do its thing.
-    text = ""
-    for block in (data.get("content") or []):
-        if isinstance(block, dict) and block.get("type") == "text":
-            text += block.get("text", "")
-    return {"text": text, "chunks": []}
-
-
-async def p_openai_websearch_via_9router(query: str) -> dict:
-    """Same idea, but for OpenAI's web_search_preview through Codex's
-    9Router connection. Goes through 9Router's openai-compat endpoint
-    (the responses API) so the user's Codex subscription covers it."""
-    import httpx
-    connected = await p_refresh_9r_connected()
-    if "codex" not in connected:
-        return {}
-    body = {
-        "model": "cx/gpt-5.4-mini",
-        "max_tokens": 1024,
-        "system": (
-            "You search the web and return concise grounded answers "
-            "with source citations. Always cite the URLs you used."
-        ),
-        "messages": [{"role": "user", "content": f"Search the web for: {query}"}],
-    }
-    async with httpx.AsyncClient(timeout=20.0) as client:
-        r = await client.post(
-            "http://localhost:20128/v1/messages",
-            json=body,
-            headers={"x-api-key": "9router", "anthropic-version": "2023-06-01"},
-        )
-        if r.status_code != 200:
-            return {}
-        data = r.json()
-    text = ""
-    for block in (data.get("content") or []):
-        if isinstance(block, dict) and block.get("type") == "text":
-            text += block.get("text", "")
-    return {"text": text, "chunks": []}
-
-
-async def p_openai_websearch(api_key: str, query: str) -> dict:
-    """Call OpenAI Responses API with the web_search_preview tool.
-
-    Returns {"text": grounded_answer, "chunks": [(title, uri), ...]}.
-    """
-    import httpx
-    body = {
-        "model": OPENAI_SEARCH_MODEL,
-        "input": f"Search the web for: {query}\n\nReturn a concise summary. Cite sources.",
-        "tools": [{"type": "web_search_preview"}],
-    }
-    async with httpx.AsyncClient(timeout=45.0) as client:
-        r = await client.post(
-            f"{OPENAI_API_BASE}/responses",
-            headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},
-            json=body,
-        )
-        r.raise_for_status()
-        data = r.json()
-
-    text_parts = []
-    chunks: list[tuple[str, str]] = []
-    for item in (data.get("output") or []):
-        if not isinstance(item, dict):
-            continue
-        for content in (item.get("content") or []):
-            if not isinstance(content, dict):
-                continue
-            if content.get("type") == "output_text":
-                text_parts.append(content.get("text", ""))
-            for ann in (content.get("annotations") or []):
-                if isinstance(ann, dict) and ann.get("type") == "url_citation":
-                    uri = ann.get("url", "")
-                    title = ann.get("title", uri)
-                    if uri:
-                        chunks.append((title, uri))
-    return {"text": "".join(text_parts), "chunks": chunks, "queries": [query]}
-
-
-async def p_openai_urlfetch(api_key: str, url: str, prompt: str | None) -> dict:
-    """Use OpenAI's web_search_preview to fetch/summarize a specific URL."""
-    prompt_text = f"Fetch and summarize the content at: {url}"
-    if prompt:
-        prompt_text += f"\n\nFocus on: {prompt}"
-    import httpx
-    body = {
-        "model": OPENAI_SEARCH_MODEL,
-        "input": prompt_text,
-        "tools": [{"type": "web_search_preview"}],
-    }
-    async with httpx.AsyncClient(timeout=45.0) as client:
-        r = await client.post(
-            f"{OPENAI_API_BASE}/responses",
-            headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},
-            json=body,
-        )
-        r.raise_for_status()
-        data = r.json()
-    text_parts = []
-    chunks = []
-    for item in (data.get("output") or []):
-        for content in (item.get("content") or []):
-            if isinstance(content, dict) and content.get("type") == "output_text":
-                text_parts.append(content.get("text", ""))
-            for ann in (content.get("annotations") or []) if isinstance(content, dict) else []:
-                if isinstance(ann, dict) and ann.get("type") == "url_citation":
-                    uri = ann.get("url", "")
-                    title = ann.get("title", uri)
-                    if uri:
-                        chunks.append((title, uri))
-    return {"text": "".join(text_parts), "chunks": chunks}
+# --------------------------------------------------------------------------- Endpoints ---------------------------------------------------------------------------
 
 
 @web.router.post("/search")
 @typechecked
-async def search(body: SearchBody) -> dict:
-    """Web search, primary-aware. Prefers the native search tool of the
-    provider the user is already paying for:
+async def search(body: SearchBody) -> Dict:
+    """Web search, primary-aware.
 
-        Gemini primary + Gemini key → googleSearch grounding
-        OpenAI primary + OpenAI key → web_search_preview
-        Anthropic path available   → handled at agent_manager layer
-                                     (MCP isn't registered; built-in
-                                     WebSearch routes through 9Router
-                                     → Anthropic's server-side tool)
-        otherwise                  → DDG fallback (CAPTCHA-prone)
+    Free keyless rungs lead (they answer at human speed and cost nothing),
+    then the packaged app's real browser, then the provider-grounded backends
+    of whichever provider the user already pays for."""
+    gemini_key = resolve_gemini_api_key()
+    openai_key = resolve_openai_api_key()
 
-    If the primary's own native path fails, we cascade to whichever
-    other provider's credentials are available, then DDG last."""
-    gemini_key = p_resolve_gemini_api_key()
-    openai_key = p_resolve_openai_api_key()
-    primary = (body.primary or "").lower()
-    errors: list[str] = []
-
-    async def try_gemini():
-        if not gemini_key:
-            return None
-        prompt = (
-            f"Search the web for: {body.query}\n\n"
-            f"Return a concise summary of what you found. Cite sources."
-        )
-        grounded = await p_gemini_grounded_call(gemini_key, prompt, use_url_context=False)
-        return {
-            "query": body.query,
-            "results": p_format_grounded_as_search_results(grounded, body.query),
-            "backend": "gemini_native",
-        }
-
-    async def try_openai():
-        if not openai_key:
-            return None
-        grounded = await p_openai_websearch(openai_key, body.query)
-        return {
-            "query": body.query,
-            "results": p_format_grounded_as_search_results(grounded, body.query),
-            "backend": "openai_native",
-        }
-
-    async def try_gemini_subscription():
-        prompt = (
-            f"Search the web for: {body.query}\n\n"
-            f"Return a concise summary of what you found. Cite sources."
-        )
-        grounded = await p_gemini_grounded_via_9router(prompt, use_url_context=False)
-        if not grounded.get("text"):
-            return None
-        return {
-            "query": body.query,
-            "results": p_format_grounded_as_search_results(grounded, body.query),
-            "backend": "gemini_subscription",
-        }
-
-    async def try_openai_subscription():
-        grounded = await p_openai_websearch_via_9router(body.query)
-        if not grounded.get("text"):
-            return None
-        return {
-            "query": body.query,
-            "results": p_format_grounded_as_search_results(grounded, body.query),
-            "backend": "openai_subscription",
-        }
-
-    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
+    async def try_keyless() -> Optional[Dict]:
+        # DuckDuckGo (html then lite); sub-second when it isn't challenged. None on a real no-hits so the chain falls through.
+        from backend.apps.agents.tools.web import DDGRateLimited, WebSearchTool
         try:
             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
+            # Surface the challenge as a recorded error (not a silent None) so the caller can see WHY we fell through to a slower backend.
+            raise RuntimeError("DuckDuckGo served its bot challenge (HTTP 202)") from None
         if not text:
             return None
         return {"query": body.query, "results": text, "backend": "ddg"}
 
-    async def try_browser_search():
-        # Packaged-app tier: a real Chromium's fingerprint isn't subject to the httpx DDG 202 throttle (proven: browser-DDG returns hits where our headless client 202s), and it can scrape Google/Bing directly. Skipped (None) when no Electron main bridge is connected.
+    async def try_browser_search() -> Optional[Dict]:
+        # Packaged-app tier: a real Chromium's fingerprint isn't subject to the headless-client challenge, and it can scrape Google/Bing directly. Skipped (None) when no Electron main bridge is connected.
         res = await p_browser_bridge("browser_search", {"query": body.query, "num_results": body.num_results})
         if not res or not res.get("results"):
             return None
         return {"query": body.query, "results": res["results"], "backend": f"browser_{res.get('engine', 'search')}"}
 
-    # Fast-first cascade: DDG leads (~1s = human speed); then the packaged browser (real fingerprint, throttle-immune, no LLM cost); then the 30-42s LLM-grounded backends. The primary hint only reorders the grounded tier. Every attempt is wait_for-bounded so a slow/hung provider fails over fast.
-    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]
+    async def try_gemini() -> Optional[Dict]:
+        if not gemini_key:
+            return None
+        grounded = await gemini_grounded_call(
+            gemini_key,
+            f"Search the web for: {body.query}\n\nReturn a concise summary of what you found. Cite sources.",
+            use_url_context=False,
+        )
+        return {"query": body.query, "results": format_grounded_as_search_results(grounded, body.query),
+                "backend": "gemini_native"}
 
-    cascade = [
-        ("ddg", try_ddg, P_DDG_ATTEMPT_TIMEOUT),
-        ("browser_search", try_browser_search, P_BROWSER_TIER_TIMEOUT),
-    ] + [
-        (name, fn, P_GROUNDED_ATTEMPT_TIMEOUT) for name, fn in grounded
-    ]
+    async def try_openai() -> Optional[Dict]:
+        if not openai_key:
+            return None
+        grounded = await openai_websearch(openai_key, body.query)
+        return {"query": body.query, "results": format_grounded_as_search_results(grounded, body.query),
+                "backend": "openai_native"}
 
-    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]}")
+    async def try_gemini_subscription() -> Optional[Dict]:
+        grounded = await gemini_grounded_via_9router(
+            f"Search the web for: {body.query}\n\nReturn a concise summary of what you found. Cite sources.",
+            False,
+        )
+        if not grounded.get("text"):
+            return None
+        return {"query": body.query, "results": format_grounded_as_search_results(grounded, body.query),
+                "backend": "gemini_subscription"}
+
+    async def try_openai_subscription() -> Optional[Dict]:
+        grounded = await openai_websearch_via_9router(body.query)
+        if not grounded.get("text"):
+            return None
+        return {"query": body.query, "results": format_grounded_as_search_results(grounded, body.query),
+                "backend": "openai_subscription"}
+
+    tiers = [
+        CascadeTier(name="ddg", run=try_keyless, budget=KEYLESS_TIER_SECONDS),
+        CascadeTier(name="browser_search", run=try_browser_search, budget=BROWSER_TIER_SECONDS),
+    ] + p_grounded_tiers("search", body.primary, {
+        "gemini_native": try_gemini,
+        "gemini_subscription": try_gemini_subscription,
+        "openai_native": try_openai,
+        "openai_subscription": try_openai_subscription,
+    })
+
+    outcome = await run_cascade(tiers, SEARCH_BUDGET_SECONDS)
+    if outcome.result is not None:
+        if outcome.errors:
+            outcome.result["cascade_errors"] = outcome.errors
+        return outcome.result
 
     # Everything failed. Be honest about why instead of an empty "no results".
-    connected = await p_refresh_9r_connected()
+    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 "
+            "Every free search frontend refused this request and no paid search backend "
+            "is configured. 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)."
+            "Every free search frontend refused this request and every configured "
+            "provider errored (see details below)."
         )
     nudge = p_browser_fallback_nudge(body.query) if body.browser_ok else ""
-    p_results_text = f"No results for: {body.query}\n\n{tail}" + (f"\n\n{nudge}" if nudge else "")
+    results_text = f"No results for: {body.query}\n\n{tail}" + (f"\n\n{nudge}" if nudge else "")
     return {
         "query": body.query,
-        "results": p_results_text,
+        "results": results_text,
         "backend": "none",
-        "cascade_errors": errors,
+        "cascade_errors": outcome.errors,
     }
 
 
 @web.router.post("/fetch")
 @typechecked
-async def fetch(body: FetchBody) -> dict:
-    """Fetch a URL, primary-aware. Mirrors /search cascade logic."""
+async def fetch(body: FetchBody) -> Dict:
+    """Fetch a URL, primary-aware. Mirrors the /search cascade."""
     # Belt-and-suspenders: even though we delegate to remote Gemini/OpenAI fetchers (which can't reach private IPs), validating the URL here means a private/metadata URL gets a 4xx instead of being silently forwarded.
     from backend.apps.agents.tools.ssrf_guard import SSRFBlocked, assert_safe_url
     try:
         await assert_safe_url(body.url)
     except SSRFBlocked as exc:
-        from fastapi import HTTPException
         raise HTTPException(status_code=400, detail=f"Refused: {exc}")
-    gemini_key = p_resolve_gemini_api_key()
-    openai_key = p_resolve_openai_api_key()
-    primary = (body.primary or "").lower()
+    gemini_key = resolve_gemini_api_key()
+    openai_key = resolve_openai_api_key()
 
-    async def try_gemini():
-        if not gemini_key:
-            return None
-        prompt_bits = [f"Fetch and summarize this URL: {body.url}"]
-        if body.prompt:
-            prompt_bits.append(f"Focus on: {body.prompt}")
-        grounded = await p_gemini_grounded_call(
-            gemini_key, "\n".join(prompt_bits), use_url_context=True,
-        )
-        return {
-            "url": body.url,
-            "content": p_format_grounded_as_fetch(grounded, body.url),
-            "backend": "gemini_native",
-        }
+    # Remembered so a thin/errored local read is still returned as the last resort if every other tier also fails (never worse than before).
+    local_text: Optional[str] = None
 
-    async def try_openai():
-        if not openai_key:
-            return None
-        grounded = await p_openai_urlfetch(openai_key, body.url, body.prompt)
-        return {
-            "url": body.url,
-            "content": p_format_grounded_as_fetch(grounded, body.url),
-            "backend": "openai_native",
-        }
-
-    async def try_gemini_subscription():
-        prompt_bits = [f"Fetch and summarize this URL: {body.url}"]
-        if body.prompt:
-            prompt_bits.append(f"Focus on: {body.prompt}")
-        grounded = await p_gemini_grounded_via_9router(
-            "\n".join(prompt_bits), use_url_context=True,
-        )
-        if not grounded.get("text"):
-            return None
-        return {
-            "url": body.url,
-            "content": p_format_grounded_as_fetch(grounded, body.url),
-            "backend": "gemini_subscription",
-        }
-
-    async def try_openai_subscription():
-        # Codex's web_search is general; URL fetch via search query works adequately for our use.
-        prompt = f"Fetch this URL and summarize: {body.url}"
-        if body.prompt:
-            prompt += f"\nFocus on: {body.prompt}"
-        grounded = await p_openai_websearch_via_9router(prompt)
-        if not grounded.get("text"):
-            return None
-        return {
-            "url": body.url,
-            "content": p_format_grounded_as_fetch(grounded, body.url),
-            "backend": "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
-
-    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.
+    async def try_local() -> Optional[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,
-        )
+        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")):
@@ -600,44 +256,69 @@ async def fetch(body: FetchBody) -> dict:
             return None
         return {"url": body.url, "content": text, "backend": "local"}
 
-    async def try_browser_fetch():
-        # 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 (the try_local thin-read case) actually resolve. Shares the user's browser cookies, so pages they're logged into fetch authed. Skipped (None) with no Electron main bridge.
+    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.
         res = await p_browser_bridge("browser_fetch", {"url": body.url})
         if not res or not res.get("text"):
             return None
         return {"url": body.url, "content": f"Contents of {body.url}:\n\n{res['text']}", "backend": "browser"}
 
-    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]
+    async def try_gemini() -> Optional[Dict]:
+        if not gemini_key:
+            return None
+        prompt_bits = [f"Fetch and summarize this URL: {body.url}"]
+        if body.prompt:
+            prompt_bits.append(f"Focus on: {body.prompt}")
+        grounded = await gemini_grounded_call(gemini_key, "\n".join(prompt_bits), use_url_context=True)
+        return {"url": body.url, "content": format_grounded_as_fetch(grounded, body.url),
+                "backend": "gemini_native"}
 
-    cascade = [
-        ("local", try_local, P_LOCAL_FETCH_TIMEOUT),
-        ("browser", try_browser_fetch, P_BROWSER_TIER_TIMEOUT),
-    ] + [
-        (name, fn, P_GROUNDED_ATTEMPT_TIMEOUT) for name, fn in grounded
-    ]
+    async def try_openai() -> Optional[Dict]:
+        if not openai_key:
+            return None
+        grounded = await openai_urlfetch(openai_key, body.url, body.prompt)
+        return {"url": body.url, "content": format_grounded_as_fetch(grounded, body.url),
+                "backend": "openai_native"}
 
-    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]}")
+    async def try_gemini_subscription() -> Optional[Dict]:
+        prompt_bits = [f"Fetch and summarize this URL: {body.url}"]
+        if body.prompt:
+            prompt_bits.append(f"Focus on: {body.prompt}")
+        grounded = await gemini_grounded_via_9router("\n".join(prompt_bits), True)
+        if not grounded.get("text"):
+            return None
+        return {"url": body.url, "content": format_grounded_as_fetch(grounded, body.url),
+                "backend": "gemini_subscription"}
 
-    # Grounded all failed; hand back whatever the local read got (even an error string is useful signal) rather than nothing.
+    async def try_openai_subscription() -> Optional[Dict]:
+        # Codex's web_search is general; URL fetch via search query works adequately for our use.
+        prompt = f"Fetch this URL and summarize: {body.url}"
+        if body.prompt:
+            prompt += f"\nFocus on: {body.prompt}"
+        grounded = await openai_websearch_via_9router(prompt)
+        if not grounded.get("text"):
+            return None
+        return {"url": body.url, "content": format_grounded_as_fetch(grounded, body.url),
+                "backend": "openai_subscription"}
+
+    tiers = [
+        CascadeTier(name="local", run=try_local, budget=LOCAL_FETCH_TIER_SECONDS),
+        CascadeTier(name="browser", run=try_browser_fetch, budget=BROWSER_TIER_SECONDS),
+    ] + p_grounded_tiers("fetch", body.primary, {
+        "gemini_native": try_gemini,
+        "gemini_subscription": try_gemini_subscription,
+        "openai_native": try_openai,
+        "openai_subscription": try_openai_subscription,
+    })
+
+    outcome = await run_cascade(tiers, FETCH_BUDGET_SECONDS)
+    if outcome.result is not None:
+        if outcome.errors:
+            outcome.result["cascade_errors"] = outcome.errors
+        return outcome.result
+
+    # Every tier 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])
+                **({"cascade_errors": outcome.errors} if outcome.errors else {})}
+    raise HTTPException(status_code=502, detail=f"Fetch failed for {body.url}: " + "; ".join(outcome.errors)[:400])

backend/tests/test_web_cascade_budget.py

@@ -0,0 +1,131 @@
+"""The /api/web cascade is bounded by ONE wall-clock budget.
+
+Seals the class of bug where the cascade's own leashes summed to 244s (search)
+/ 270s (fetch) while the MCP shim gave up at 45s, so the later tiers were
+unreachable and search "worked" only when an early tier won the race.
+"""
+
+import asyncio
+import time
+
+import pytest
+
+import backend.apps.web.cascade as C
+from backend.apps.web.cascade import CascadeTier, run_cascade
+
+
+@pytest.fixture
+def p_tiny_floor(monkeypatch):
+    """Shrink the honest-slice floor so budget tests run in milliseconds, not seconds."""
+    monkeypatch.setattr(C, "MIN_TIER_SECONDS", 0.05)
+
+
+def p_tier(name, budget, fn):
+    return CascadeTier(name=name, run=fn, budget=budget)
+
+
+async def p_hangs():
+    await asyncio.sleep(60)
+
+
+async def p_none():
+    return None
+
+
+@pytest.mark.asyncio
+async def test_first_result_short_circuits():
+    async def p_hit():
+        return {"backend": "one"}
+
+    async def p_boom():
+        raise AssertionError("later tiers must not run once a tier answers")
+
+    out = await run_cascade([p_tier("a", 5, p_hit), p_tier("b", 5, p_boom)], 10.0)
+    assert out.result == {"backend": "one"}
+    assert out.errors == []
+
+
+@pytest.mark.asyncio
+async def test_total_budget_bounds_the_whole_cascade(p_tiny_floor):
+    # Four tiers each willing to burn 60s: without a shared deadline this runs for minutes.
+    tiers = [p_tier(f"t{i}", 60.0, p_hangs) for i in range(4)]
+    t0 = time.monotonic()
+    out = await run_cascade(tiers, 1.0)
+    elapsed = time.monotonic() - t0
+    assert elapsed < 2.0, f"cascade overran its budget: {elapsed:.2f}s"
+    assert out.result is None
+    assert any("timed out" in e for e in out.errors)
+
+
+@pytest.mark.asyncio
+async def test_a_slow_tier_cannot_starve_the_ones_behind_it(p_tiny_floor):
+    # A hanging tier is cut at ITS OWN budget, so the tier behind it still gets its turn and wins.
+    async def p_hit():
+        return {"backend": "rescue"}
+
+    out = await run_cascade([p_tier("slow", 0.4, p_hangs), p_tier("fast", 5.0, p_hit)], 3.0)
+    assert out.result == {"backend": "rescue"}
+    assert any("slow" in e and "timed out" in e for e in out.errors)
+
+
+@pytest.mark.asyncio
+async def test_unreachable_tiers_are_reported_not_silently_dropped():
+    async def p_slow():
+        await asyncio.sleep(0.6)
+        return None
+
+    out = await run_cascade(
+        [p_tier("burn", 5.0, p_slow), p_tier("never_a", 5.0, p_none), p_tier("never_b", 5.0, p_none)],
+        0.6 + C.MIN_TIER_SECONDS - 0.1,
+    )
+    assert out.result is None
+    tail = out.errors[-1]
+    assert "budget spent" in tail
+    assert "never_a" in tail and "never_b" in tail
+
+
+@pytest.mark.asyncio
+async def test_a_tier_slice_never_drops_below_the_honest_floor():
+    # A tier handed 0.2s would "time out" on a hair trigger; we must say the budget ran out instead.
+    async def p_slow():
+        await asyncio.sleep(0.5)
+        return None
+
+    out = await run_cascade([p_tier("burn", 5.0, p_slow), p_tier("squeezed", 5.0, p_hangs)], 0.5 + 1.0)
+    assert out.result is None
+    assert not any("squeezed" in e and "timed out" in e for e in out.errors)
+    assert any("budget spent" in e for e in out.errors)
+
+
+@pytest.mark.asyncio
+async def test_tier_exception_is_recorded_and_the_chain_continues():
+    async def p_boom():
+        raise RuntimeError("provider down")
+
+    async def p_hit():
+        return {"backend": "next"}
+
+    out = await run_cascade([p_tier("bad", 5.0, p_boom), p_tier("good", 5.0, p_hit)], 10.0)
+    assert out.result == {"backend": "next"}
+    assert out.errors == ["bad: provider down"]
+
+
+def test_mcp_shim_waits_longer_than_the_server_budget():
+    """The shim must outlast the cascade, or the later tiers are unreachable again."""
+    from backend.apps.agents.web_mcp_server import TOOL_TIMEOUT
+    from backend.apps.web.web import FETCH_BUDGET_SECONDS, SEARCH_BUDGET_SECONDS
+
+    assert TOOL_TIMEOUT > SEARCH_BUDGET_SECONDS
+    assert TOOL_TIMEOUT > FETCH_BUDGET_SECONDS
+
+
+def test_search_tier_budgets_leave_room_for_the_grounded_tier():
+    """The free rungs must not eat the whole budget before a paid backend is tried."""
+    import backend.apps.web.web as W
+
+    # A grounded native call needs ~30-42s, so the free rungs must leave it a usable slice.
+    grounded_floor = 20.0
+    search_cheap = W.KEYLESS_TIER_SECONDS + W.BROWSER_TIER_SECONDS
+    assert W.SEARCH_BUDGET_SECONDS - search_cheap >= grounded_floor
+    fetch_cheap = W.LOCAL_FETCH_TIER_SECONDS + W.BROWSER_TIER_SECONDS
+    assert W.FETCH_BUDGET_SECONDS - fetch_cheap >= grounded_floor

backend/tests/test_web_search_cascade.py

@@ -23,18 +23,18 @@ from backend.apps.agents.tools.web import WebSearchTool, DDGRateLimited
 @pytest.fixture(autouse=True)
 def p_no_network(monkeypatch):
     # Default everything to "unavailable / no network"; each test opts paths in.
-    monkeypatch.setattr(W, "p_resolve_gemini_api_key", lambda: None)
-    monkeypatch.setattr(W, "p_resolve_openai_api_key", lambda: None)
+    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, "p_refresh_9r_connected", p_no_subs)
+    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, "p_gemini_grounded_via_9router", p_empty)
-    monkeypatch.setattr(W, "p_openai_websearch_via_9router", p_empty)
+    monkeypatch.setattr(W, "gemini_grounded_via_9router", p_empty)
+    monkeypatch.setattr(W, "openai_websearch_via_9router", p_empty)
 
 
 def p_ddg_returns(monkeypatch, text):
@@ -55,7 +55,7 @@ async def test_ddg_is_tried_first_and_wins(monkeypatch):
     # 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")
-    monkeypatch.setattr(W, "p_gemini_grounded_call", p_boom)
+    monkeypatch.setattr(W, "gemini_grounded_call", p_boom)
 
     t = time.monotonic()
     res = await search(SearchBody(query="foo"))
@@ -68,11 +68,11 @@ 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)
-    monkeypatch.setattr(W, "p_resolve_openai_api_key", lambda: "okey")
+    monkeypatch.setattr(W, "resolve_openai_api_key", lambda: "okey")
 
     async def p_openai(api_key, query):
         return {"text": "grounded answer", "chunks": [("Title", "https://u.example")]}
-    monkeypatch.setattr(W, "p_openai_websearch", p_openai)
+    monkeypatch.setattr(W, "openai_websearch", p_openai)
 
     res = await search(SearchBody(query="x"))
     assert res["backend"] == "openai_native"
@@ -84,12 +84,12 @@ async def test_ddg_throttled_falls_over_to_openai(monkeypatch):
 @pytest.mark.asyncio
 async def test_a_hung_grounded_attempt_is_bounded(monkeypatch):
     p_ddg_throttled(monkeypatch)
-    monkeypatch.setattr(W, "P_GROUNDED_ATTEMPT_TIMEOUT", 0.3)
-    monkeypatch.setattr(W, "p_resolve_gemini_api_key", lambda: "gkey")
+    monkeypatch.setattr(W, "GROUNDED_TIER_SECONDS", 0.3)
+    monkeypatch.setattr(W, "resolve_gemini_api_key", lambda: "gkey")
 
     async def p_hangs(*a, **k):
         await asyncio.sleep(30)
-    monkeypatch.setattr(W, "p_gemini_grounded_call", p_hangs)
+    monkeypatch.setattr(W, "gemini_grounded_call", p_hangs)
 
     t = time.monotonic()
     res = await search(SearchBody(query="x"))
@@ -102,15 +102,15 @@ 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)
-    monkeypatch.setattr(W, "p_resolve_gemini_api_key", lambda: "gkey")
-    monkeypatch.setattr(W, "p_resolve_openai_api_key", lambda: "okey")
+    monkeypatch.setattr(W, "resolve_gemini_api_key", lambda: "gkey")
+    monkeypatch.setattr(W, "resolve_openai_api_key", lambda: "okey")
 
     async def p_gem(*a, **k):
         return {"text": "GEM", "chunks": [("g", "https://gem.example")]}
     async def p_oai(api_key, query):
         return {"text": "OAI", "chunks": [("o", "https://oai.example")]}
-    monkeypatch.setattr(W, "p_gemini_grounded_call", p_gem)
-    monkeypatch.setattr(W, "p_openai_websearch", p_oai)
+    monkeypatch.setattr(W, "gemini_grounded_call", p_gem)
+    monkeypatch.setattr(W, "openai_websearch", p_oai)
 
     res = await search(SearchBody(query="x", primary="openai"))
     # openai must be tried before gemini when it's the primary
@@ -132,11 +132,11 @@ async def test_everything_fails_is_honest_not_empty(monkeypatch):
 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)
-    monkeypatch.setattr(W, "p_resolve_openai_api_key", lambda: "okey")  # configured but errors
+    monkeypatch.setattr(W, "resolve_openai_api_key", lambda: "okey")  # configured but errors
 
     async def p_openai_boom(*a, **k):
         raise RuntimeError("openai down")
-    monkeypatch.setattr(W, "p_openai_websearch", p_openai_boom)
+    monkeypatch.setattr(W, "openai_websearch", p_openai_boom)
 
     res = await search(SearchBody(query="sony zv-e10 price", browser_ok=True))
     assert res["backend"] == "none"
@@ -148,11 +148,11 @@ async def test_everything_fails_nudges_browser_not_retry(monkeypatch):
 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)
-    monkeypatch.setattr(W, "p_resolve_openai_api_key", lambda: "okey")
+    monkeypatch.setattr(W, "resolve_openai_api_key", lambda: "okey")
 
     async def p_openai_boom(*a, **k):
         raise RuntimeError("openai down")
-    monkeypatch.setattr(W, "p_openai_websearch", p_openai_boom)
+    monkeypatch.setattr(W, "openai_websearch", p_openai_boom)
 
     res = await search(SearchBody(query="sony zv-e10 price"))
     assert res["backend"] == "none"
@@ -186,7 +186,7 @@ async def test_fetch_local_first_wins_and_is_fast(monkeypatch):
     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, "p_gemini_grounded_call", p_boom)
+    monkeypatch.setattr(W, "gemini_grounded_call", p_boom)
 
     t = time.monotonic()
     res = await fetch(FetchBody(url="https://x.example"))
@@ -198,10 +198,10 @@ async def test_fetch_local_first_wins_and_is_fast(monkeypatch):
 @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, "p_resolve_gemini_api_key", lambda: "gkey")
+    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, "p_gemini_grounded_call", p_gem)
+    monkeypatch.setattr(W, "gemini_grounded_call", p_gem)
 
     res = await fetch(FetchBody(url="https://spa.example"))
     assert res["backend"] == "gemini_native"
@@ -233,7 +233,7 @@ async def test_browser_search_tier_fires_when_ddg_throttled(monkeypatch):
 
     async def p_boom(*a, **k):
         raise AssertionError("grounded should not be reached once the browser tier answers")
-    monkeypatch.setattr(W, "p_gemini_grounded_call", p_boom)
+    monkeypatch.setattr(W, "gemini_grounded_call", p_boom)
 
     res = await search(SearchBody(query="q"))
     assert res["backend"] == "browser_ddg"
@@ -245,11 +245,11 @@ 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)
-    monkeypatch.setattr(W, "p_resolve_openai_api_key", lambda: "okey")
+    monkeypatch.setattr(W, "resolve_openai_api_key", lambda: "okey")
 
     async def p_openai(api_key, query):
         return {"text": "grounded", "chunks": [("T", "https://u.example")]}
-    monkeypatch.setattr(W, "p_openai_websearch", p_openai)
+    monkeypatch.setattr(W, "openai_websearch", p_openai)
 
     res = await search(SearchBody(query="q"))
     assert res["backend"] == "openai_native"