openswarm/backend/apps/web/web.py

"""Web search + fetch sub-app.

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

from contextlib import asynccontextmanager
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


@asynccontextmanager
async def web_lifespan():
    yield


web = SubApp("web", web_lifespan)


# --------------------------------------------------------------------------- Request models ---------------------------------------------------------------------------


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: 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: Optional[str] = Field(None, description="Optional context hint.")
    primary: Optional[str] = Field(None, description="Primary provider hint.")


# --------------------------------------------------------------------------- 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
ARCHIVE_TIER_SECONDS = 10.0       # the Wayback redirect path answered in 0.5-3s


# --------------------------------------------------------------------------- Helpers ---------------------------------------------------------------------------


# 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]:
    from backend.apps.agents.core.ws_manager import ws_manager
    if ws_manager.main_connection is None:
        return None
    res = await ws_manager.send_main_command(uuid4().hex, action, params)
    if not res or res.get("error"):
        return 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. A real Chromium carries a real browser fingerprint, which is what the challenge is actually keyed on.
@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 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."'
    )


@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
    ]


# --------------------------------------------------------------------------- Endpoints ---------------------------------------------------------------------------


@web.router.post("/search")
@typechecked
async def search(body: SearchBody) -> Dict:
    """Web search, primary-aware.

    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()

    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 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_startpage() -> Optional[Dict]:
        # Second independent engine (Google's index), so DuckDuckGo's bot challenge is no longer a single point of failure for keyless users.
        from backend.apps.agents.tools.search.search_startpage import search_startpage
        text = await search_startpage(body.query, body.num_results)
        if not text:
            return None
        return {"query": body.query, "results": text, "backend": "startpage"}

    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')}"}

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

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

    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, breaker=True),
        CascadeTier(name="startpage", run=try_startpage, budget=KEYLESS_TIER_SECONDS, breaker=True),
        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 refresh_9r_connected()
    has_subscription = bool(connected & {"codex", "antigravity", "gemini-cli"})
    if not (gemini_key or openai_key or has_subscription):
        tail = (
            "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 = (
            "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 ""
    results_text = f"No results for: {body.query}\n\n{tail}" + (f"\n\n{nudge}" if nudge else "")
    return {
        "query": body.query,
        "results": results_text,
        "backend": "none",
        "cascade_errors": outcome.errors,
    }


@web.router.post("/fetch")
@typechecked
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 DomainUnreachable, SSRFBlocked, assert_safe_url
    try:
        await assert_safe_url(body.url)
    except DomainUnreachable:
        # A domain that no longer resolves is the archive's whole reason for existing, so let the cascade run instead of 400ing here.
        pass
    except SSRFBlocked as exc:
        raise HTTPException(status_code=400, detail=f"Refused: {exc}")
    gemini_key = resolve_gemini_api_key()
    openai_key = resolve_openai_api_key()

    # 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_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
        page = await WebFetchTool.fetch_page(body.url, body.prompt)
        local_text = page.text
        if page.kind == "error":
            return None
        # 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": 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.
        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"}

    async def try_wayback() -> Optional[Dict]:
        # A dead link or a hard bot wall is exactly what the archive is for, and it returns the page's REAL text where a grounded fetcher can only summarise a page it also can't read.
        from backend.apps.agents.tools.fetch.wayback import fetch_wayback
        text = await fetch_wayback(body.url)
        if not text:
            return None
        return {"url": body.url, "content": text, "backend": "wayback"}

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

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

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

    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),
        CascadeTier(name="wayback", run=try_wayback, budget=ARCHIVE_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": outcome.errors} if outcome.errors else {})}
    raise HTTPException(status_code=502, detail=f"Fetch failed for {body.url}: " + "; ".join(outcome.errors)[:400])