From 82e3db979dddb222b9cc3db2b3f28b8715794ae8 Mon Sep 17 00:00:00 2001 From: ciregenz Date: Tue, 23 Jun 2026 21:01:55 -0700 Subject: [PATCH] [eric] web: leading-_ -> p_/P_ across web.py (module fns/consts + locals/args via AST renamer), fix test refs --- backend/apps/web/web.py | 112 +++++++++++------------ backend/tests/test_web_search_cascade.py | 36 ++++---- 2 files changed, 74 insertions(+), 74 deletions(-) diff --git a/backend/apps/web/web.py b/backend/apps/web/web.py index a15ef232..e98567d7 100644 --- a/backend/apps/web/web.py +++ b/backend/apps/web/web.py @@ -58,7 +58,7 @@ class FetchBody(BaseModel): # --------------------------------------------------------------------------- -def _join_text(parts: list[dict[str, Any]]) -> str: +def p_join_text(parts: list[dict[str, Any]]) -> str: out = [] for p in parts: if isinstance(p, dict) and p.get("type") == "text": @@ -84,16 +84,16 @@ OPENAI_SEARCH_MODEL = "gpt-5-mini" # cheapest model that supports web_search_pr # 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 +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. -_LOCAL_FETCH_TIMEOUT = 32.0 +P_LOCAL_FETCH_TIMEOUT = 32.0 -async def _gemini_grounded_call(api_key: str, prompt: str, *, use_url_context: bool) -> dict: +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), ...], @@ -135,7 +135,7 @@ async def _gemini_grounded_call(api_key: str, prompt: str, *, use_url_context: b return {"text": text, "chunks": chunks, "queries": queries} -def _format_grounded_as_search_results(grounded: dict, query: str) -> str: +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 [] @@ -149,7 +149,7 @@ def _format_grounded_as_search_results(grounded: dict, query: str) -> str: return "\n\n".join(lines) -def _format_grounded_as_fetch(grounded: dict, url: str) -> str: +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 "" @@ -163,7 +163,7 @@ def _format_grounded_as_fetch(grounded: dict, url: str) -> str: return "\n".join(parts) -def _resolve_gemini_api_key() -> str | None: +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 @@ -173,7 +173,7 @@ def _resolve_gemini_api_key() -> str | None: return None -def _resolve_openai_api_key() -> str | None: +def p_resolve_openai_api_key() -> str | None: try: from backend.apps.settings.settings import load_settings s = load_settings() @@ -186,38 +186,38 @@ def _resolve_openai_api_key() -> str | None: # `_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. -_NINE_ROUTER_CONNECTED: set[str] = set() -_NINE_ROUTER_CACHE_AT: float = 0.0 +P_NINE_ROUTER_CONNECTED: set[str] = set() +P_NINE_ROUTER_CACHE_AT: float = 0.0 -async def _refresh_9r_connected() -> set[str]: +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 _NINE_ROUTER_CONNECTED, _NINE_ROUTER_CACHE_AT - import time as _t - now = _t.time() - if now - _NINE_ROUTER_CACHE_AT < 20.0: - return _NINE_ROUTER_CONNECTED + 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 _9r_running, get_providers as _9r_providers - if not _9r_running(): - _NINE_ROUTER_CONNECTED = set() + 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 _9r_providers() - _NINE_ROUTER_CONNECTED = { + 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") } - _NINE_ROUTER_CACHE_AT = now + P_NINE_ROUTER_CACHE_AT = now except Exception: # Cache stays; best-effort. pass - return _NINE_ROUTER_CONNECTED + return P_NINE_ROUTER_CONNECTED -async def _gemini_grounded_via_9router(prompt: str, use_url_context: bool) -> dict: +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. @@ -230,7 +230,7 @@ async def _gemini_grounded_via_9router(prompt: str, use_url_context: bool) -> di import httpx # Prefer Gemini CLI (broader model coverage). Fall back to # Antigravity if CLI isn't connected. - connected = await _refresh_9r_connected() + connected = await p_refresh_9r_connected() if "gemini-cli" in connected: model = "gc/gemini-2.5-flash" elif "antigravity" in connected: @@ -271,12 +271,12 @@ async def _gemini_grounded_via_9router(prompt: str, use_url_context: bool) -> di return {"text": text, "chunks": []} -async def _openai_websearch_via_9router(query: str) -> dict: +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 _refresh_9r_connected() + connected = await p_refresh_9r_connected() if "codex" not in connected: return {} body = { @@ -304,7 +304,7 @@ async def _openai_websearch_via_9router(query: str) -> dict: return {"text": text, "chunks": []} -async def _openai_websearch(api_key: str, query: str) -> dict: +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), ...]}. @@ -343,7 +343,7 @@ async def _openai_websearch(api_key: str, query: str) -> dict: return {"text": "".join(text_parts), "chunks": chunks, "queries": [query]} -async def _openai_urlfetch(api_key: str, url: str, prompt: str | None) -> dict: +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: @@ -393,8 +393,8 @@ async def search(body: SearchBody) -> dict: If the primary's own native path fails, we cascade to whichever other provider's credentials are available, then DDG last.""" - gemini_key = _resolve_gemini_api_key() - openai_key = _resolve_openai_api_key() + gemini_key = p_resolve_gemini_api_key() + openai_key = p_resolve_openai_api_key() primary = (body.primary or "").lower() errors: list[str] = [] @@ -405,20 +405,20 @@ async def search(body: SearchBody) -> dict: f"Search the web for: {body.query}\n\n" f"Return a concise summary of what you found. Cite sources." ) - grounded = await _gemini_grounded_call(gemini_key, prompt, use_url_context=False) + grounded = await p_gemini_grounded_call(gemini_key, prompt, use_url_context=False) return { "query": body.query, - "results": _format_grounded_as_search_results(grounded, 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 _openai_websearch(openai_key, body.query) + grounded = await p_openai_websearch(openai_key, body.query) return { "query": body.query, - "results": _format_grounded_as_search_results(grounded, body.query), + "results": p_format_grounded_as_search_results(grounded, body.query), "backend": "openai_native", } @@ -427,22 +427,22 @@ async def search(body: SearchBody) -> dict: f"Search the web for: {body.query}\n\n" f"Return a concise summary of what you found. Cite sources." ) - grounded = await _gemini_grounded_via_9router(prompt, use_url_context=False) + grounded = await p_gemini_grounded_via_9router(prompt, use_url_context=False) if not grounded.get("text"): return None return { "query": body.query, - "results": _format_grounded_as_search_results(grounded, body.query), + "results": p_format_grounded_as_search_results(grounded, body.query), "backend": "gemini_subscription", } async def try_openai_subscription(): - grounded = await _openai_websearch_via_9router(body.query) + grounded = await p_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), + "results": p_format_grounded_as_search_results(grounded, body.query), "backend": "openai_subscription", } @@ -475,8 +475,8 @@ async def search(body: SearchBody) -> dict: 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 + cascade = [("ddg", try_ddg, P_DDG_ATTEMPT_TIMEOUT)] + [ + (name, fn, P_GROUNDED_ATTEMPT_TIMEOUT) for name, fn in grounded ] for name, fn, timeout in cascade: @@ -492,7 +492,7 @@ async def search(body: SearchBody) -> dict: errors.append(f"{name}: {str(e)[:150]}") # Everything failed. Be honest about why instead of an empty "no results". - connected = await _refresh_9r_connected() + connected = await p_refresh_9r_connected() has_subscription = bool(connected & {"codex", "antigravity", "gemini-cli"}) if not (gemini_key or openai_key or has_subscription): tail = ( @@ -526,8 +526,8 @@ async def fetch(body: FetchBody) -> dict: except SSRFBlocked as exc: from fastapi import HTTPException raise HTTPException(status_code=400, detail=f"Refused: {exc}") - gemini_key = _resolve_gemini_api_key() - openai_key = _resolve_openai_api_key() + gemini_key = p_resolve_gemini_api_key() + openai_key = p_resolve_openai_api_key() primary = (body.primary or "").lower() async def try_gemini(): @@ -536,22 +536,22 @@ async def fetch(body: FetchBody) -> 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_call( + grounded = await p_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), + "content": p_format_grounded_as_fetch(grounded, body.url), "backend": "gemini_native", } async def try_openai(): if not openai_key: return None - grounded = await _openai_urlfetch(openai_key, body.url, body.prompt) + grounded = await p_openai_urlfetch(openai_key, body.url, body.prompt) return { "url": body.url, - "content": _format_grounded_as_fetch(grounded, body.url), + "content": p_format_grounded_as_fetch(grounded, body.url), "backend": "openai_native", } @@ -559,14 +559,14 @@ async def fetch(body: FetchBody) -> 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( + 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": _format_grounded_as_fetch(grounded, body.url), + "content": p_format_grounded_as_fetch(grounded, body.url), "backend": "gemini_subscription", } @@ -576,12 +576,12 @@ async def fetch(body: FetchBody) -> dict: 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) + grounded = await p_openai_websearch_via_9router(prompt) if not grounded.get("text"): return None return { "url": body.url, - "content": _format_grounded_as_fetch(grounded, body.url), + "content": p_format_grounded_as_fetch(grounded, body.url), "backend": "openai_subscription", } @@ -599,7 +599,7 @@ async def fetch(body: FetchBody) -> dict: parts = await WebFetchTool().execute( {"url": body.url, "prompt": body.prompt or ""}, None, ) - text = _join_text(parts) + text = p_join_text(parts) local_text = text if text.startswith(("HTTP error", "Error fetching", "Refused to fetch")): return None @@ -617,8 +617,8 @@ async def fetch(body: FetchBody) -> dict: 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 + cascade = [("local", try_local, P_LOCAL_FETCH_TIMEOUT)] + [ + (name, fn, P_GROUNDED_ATTEMPT_TIMEOUT) for name, fn in grounded ] errors: list[str] = [] diff --git a/backend/tests/test_web_search_cascade.py b/backend/tests/test_web_search_cascade.py index 659764b7..00cc8b41 100644 --- a/backend/tests/test_web_search_cascade.py +++ b/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 _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) + monkeypatch.setattr(W, "p_resolve_gemini_api_key", lambda: None) + monkeypatch.setattr(W, "p_resolve_openai_api_key", lambda: None) async def _no_subs(): return set() - monkeypatch.setattr(W, "_refresh_9r_connected", _no_subs) + monkeypatch.setattr(W, "p_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) + monkeypatch.setattr(W, "p_gemini_grounded_via_9router", _empty) + monkeypatch.setattr(W, "p_openai_websearch_via_9router", _empty) def _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 _boom(*a, **k): raise AssertionError("grounded should not be called when DDG has results") - monkeypatch.setattr(W, "_gemini_grounded_call", _boom) + monkeypatch.setattr(W, "p_gemini_grounded_call", _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): _ddg_throttled(monkeypatch) - monkeypatch.setattr(W, "_resolve_openai_api_key", lambda: "okey") + monkeypatch.setattr(W, "p_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) + monkeypatch.setattr(W, "p_openai_websearch", _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): _ddg_throttled(monkeypatch) - monkeypatch.setattr(W, "_GROUNDED_ATTEMPT_TIMEOUT", 0.3) - monkeypatch.setattr(W, "_resolve_gemini_api_key", lambda: "gkey") + monkeypatch.setattr(W, "P_GROUNDED_ATTEMPT_TIMEOUT", 0.3) + monkeypatch.setattr(W, "p_resolve_gemini_api_key", lambda: "gkey") async def _hangs(*a, **k): await asyncio.sleep(30) - monkeypatch.setattr(W, "_gemini_grounded_call", _hangs) + monkeypatch.setattr(W, "p_gemini_grounded_call", _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): _ddg_throttled(monkeypatch) - monkeypatch.setattr(W, "_resolve_gemini_api_key", lambda: "gkey") - monkeypatch.setattr(W, "_resolve_openai_api_key", lambda: "okey") + monkeypatch.setattr(W, "p_resolve_gemini_api_key", lambda: "gkey") + monkeypatch.setattr(W, "p_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) + monkeypatch.setattr(W, "p_gemini_grounded_call", _gem) + monkeypatch.setattr(W, "p_openai_websearch", _oai) res = await search(SearchBody(query="x", primary="openai")) # openai must be tried before gemini when it's the primary @@ -157,7 +157,7 @@ async def test_fetch_local_first_wins_and_is_fast(monkeypatch): _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) + monkeypatch.setattr(W, "p_gemini_grounded_call", _boom) t = time.monotonic() res = await fetch(FetchBody(url="https://x.example")) @@ -169,10 +169,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): _local_returns(monkeypatch, "Contents of https://spa.example:\n\n") # JS wall, empty body - monkeypatch.setattr(W, "_resolve_gemini_api_key", lambda: "gkey") + monkeypatch.setattr(W, "p_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) + monkeypatch.setattr(W, "p_gemini_grounded_call", _gem) res = await fetch(FetchBody(url="https://spa.example")) assert res["backend"] == "gemini_native"