feat(search): add Serper as a keyed fallback behind SearXNG (ADR-002)

Stage 2. SearXNG stays primary and the fallback engages only when a query
returns nothing -- which is what a fully blocked engine pool looks like -- so
the private path remains the default and the paid path is insurance.

Serper was chosen because it returns organic URLs and snippets and nothing
else. This stack already owns crawling, so providers bundling page content
(Tavily, Exa, Firecrawl) would be paid for output it discards. Brave remains
rejected pending a licensing question: its terms restrict storing API
results, and this system exists to build a persistent corpus.

Inert without SERPER_API_KEY, so acquisition behaves exactly as before until
a key is present. Never raises: a quota error or transport failure returns no
results, exactly as SearXNG does, so a dead fallback degrades acquisition
rather than failing a job -- and unlike a silent CAPTCHA, a quota error is
logged legibly. The key is excluded from the config repr, matching the judge
key, and a test asserts it cannot reach a log that way.

Fallback URLs go through the unchanged source policy and SSRF vetting, so a
new provider cannot bypass either. Job progress records which provider served
each query.

339 tests green in-container, covering result mapping, inert-without-key,
quota errors, transport failure, entries without a link, and repr exclusion.
ruff clean, compose valid.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: funsaized

.env.example

@@ -9,6 +9,9 @@
 SEARXNG_ENGINES=duckduckgo,bing,brave,startpage,mojeek,qwant
 # Delay between a plan's searches. Bursts are what trigger CAPTCHAs.
 SEARCH_PACING_SECONDS=2.0
+# Optional keyed search fallback, used only when every SearXNG engine returns
+# nothing. Leave blank to run on SearXNG alone. Never commit a real key.
+SERPER_API_KEY=
 SEARXNG_SECRET=
 # Shared bearer token between research-hub/research-worker and the Crawl4AI API.
 CRAWL4AI_API_TOKEN=

docker-compose.yml

@@ -225,6 +225,8 @@ services:
       - SEARXNG_URL=http://hub-searxng:8080
       - SEARXNG_ENGINES=${SEARXNG_ENGINES:-duckduckgo,bing,brave,startpage,mojeek,qwant}
       - SEARCH_PACING_SECONDS=${SEARCH_PACING_SECONDS:-2.0}
+      # Optional keyed fallback; unset means SearXNG only (ADR-002).
+      - SERPER_API_KEY=${SERPER_API_KEY:-}
       - CRAWL4AI_URL=http://hub-crawl4ai:11235
       - CRAWL4AI_TOKEN=${CRAWL4AI_API_TOKEN:?Set CRAWL4AI_API_TOKEN in .env — generate with openssl rand -hex 32}
       - LOG_LEVEL=info
@@ -301,6 +303,8 @@ services:
       - SEARXNG_URL=http://hub-searxng:8080
       - SEARXNG_ENGINES=${SEARXNG_ENGINES:-duckduckgo,bing,brave,startpage,mojeek,qwant}
       - SEARCH_PACING_SECONDS=${SEARCH_PACING_SECONDS:-2.0}
+      # Optional keyed fallback; unset means SearXNG only (ADR-002).
+      - SERPER_API_KEY=${SERPER_API_KEY:-}
       - CRAWL4AI_URL=http://hub-crawl4ai:11235
       - CRAWL4AI_TOKEN=${CRAWL4AI_API_TOKEN:?Set CRAWL4AI_API_TOKEN in .env — generate with openssl rand -hex 32}
       - WORKER_LEASE_SECONDS=${WORKER_LEASE_SECONDS:-60}

research-hub/app/clients.py

@@ -424,6 +424,73 @@ async def search(self, query: str, max_results: int = 20, language: str = "en")
             return []
 
 
+class SerperClient:
+    """Keyed Google-results fallback for when every SearXNG engine is blocked.
+
+    Deliberately a fallback, not a replacement: SearXNG stays the default so
+    the private path is the normal one and the paid path is insurance
+    (ADR-002). Returns URLs and snippets only -- this stack does its own
+    crawling, so a provider bundling page content would be paid for output it
+    discards.
+
+    Never raises: a failure returns no results, exactly as SearXNG does, so a
+    dead fallback degrades acquisition rather than failing the job.
+    """
+
+    def __init__(self, api_key: str, base_url: str = "https://google.serper.dev",
+                 timeout: float = 30.0):
+        self._api_key = api_key
+        self._configured = bool(api_key)
+        self._client = httpx.AsyncClient(base_url=base_url.rstrip("/"),
+                                         timeout=timeout)
+
+    @property
+    def configured(self) -> bool:
+        return self._configured
+
+    async def close(self):
+        await self._client.aclose()
+
+    async def search(self, query: str, max_results: int = 20,
+                     language: str = "en") -> list[dict]:
+        if not self._configured:
+            return []
+        try:
+            response = await self._client.post(
+                "/search",
+                headers={"X-API-KEY": self._api_key,
+                         "Content-Type": "application/json"},
+                json={"q": query, "hl": language,
+                      "num": max(1, min(int(max_results), 100))},
+            )
+            if response.status_code != 200:
+                # A quota error is a documented, legible failure -- unlike the
+                # silent CAPTCHA this exists to survive.
+                logger.warning("serper_search_failed", extra={"diagnostic": {
+                    "status_code": response.status_code,
+                }})
+                return []
+            payload = response.json()
+        except Exception as exc:
+            logger.warning("serper_search_failed", extra={"diagnostic": {
+                "failure_type": type(exc).__name__,
+            }})
+            return []
+
+        results = []
+        for item in payload.get("organic") or []:
+            url = item.get("link")
+            if not url:
+                continue
+            results.append({
+                "url": url,
+                "title": item.get("title", ""),
+                "snippet": item.get("snippet", ""),
+                "published_at": item.get("date"),
+            })
+        return results
+
+
 def crawl_markdown_text(value: Any) -> str:
     """Normalize Crawl4AI markdown across string and structured responses."""
     if isinstance(value, str):

research-hub/app/config.py

@@ -25,6 +25,13 @@ class Config:
     # succession is what triggers a CAPTCHA; SearXNG's own guidance is to back
     # off. Seconds per job against a ~180s job is a rounding error (ADR-002).
     search_pacing_seconds: float = 2.0
+    # Keyed fallback for when every SearXNG engine is blocked (ADR-002 stage
+    # 2). SearXNG stays primary: this engages only when a query returns
+    # nothing, so the private path remains the default and the paid path is
+    # insurance. Unset means disabled, and acquisition behaves exactly as
+    # before. Excluded from repr so the key can never reach a log.
+    serper_base_url: str = "https://google.serper.dev"
+    serper_api_key: str = field(default="", repr=False)
     qdrant_collection: str = "research_corpus"
     embedding_dimension: int = 768
     chunk_size: int = 800
@@ -153,6 +160,9 @@ def load_config() -> Config:
         llm_model=os.environ.get("LLM_MODEL", "qwen3.5:9b"),
         embedding_model=os.environ.get("EMBEDDING_MODEL", "nomic-embed-text"),
         searxng_url=os.environ.get("SEARXNG_URL", "http://localhost:8080"),
+        serper_base_url=os.environ.get("SERPER_BASE_URL",
+                                       "https://google.serper.dev"),
+        serper_api_key=os.environ.get("SERPER_API_KEY", ""),
         search_pacing_seconds=float(
             os.environ.get("SEARCH_PACING_SECONDS", "2.0")
         ),

research-hub/app/research.py

@@ -17,7 +17,9 @@
 import redis.asyncio as redis_async
 
 from .config import Config
-from .clients import OllamaClient, QdrantClient, SearXNGClient, Crawl4AIClient
+from .clients import (
+    Crawl4AIClient, OllamaClient, QdrantClient, SearXNGClient, SerperClient,
+)
 from .context import classify_and_sanitize
 from .models import JobStatus, ResearchRequest
 from .document_store import DocumentStore
@@ -316,6 +318,10 @@ def __init__(self, cfg: Config):
         self.searxng = SearXNGClient(
             cfg.searxng_url, getattr(cfg, "searxng_engines", None)
         )
+        self.serper = SerperClient(
+            getattr(cfg, "serper_api_key", ""),
+            getattr(cfg, "serper_base_url", "https://google.serper.dev"),
+        )
         self.crawl4ai = Crawl4AIClient(cfg.crawl4ai_url, cfg.crawl4ai_token or None)
         self.documents = DocumentStore(cfg.document_store_path)
         self.retrieval = ScopedRetrievalService(
@@ -337,6 +343,7 @@ async def init(self):
     async def close(self):
         await self.ollama.close()
         await self.searxng.close()
+        await self.serper.close()
         await self.crawl4ai.close()
         await self.claim_verifier.close()
         if self._redis:
@@ -665,6 +672,7 @@ async def run_job(self, job_id: str):
             issued_vectors = list(plan.vectors)
             search_pacing = getattr(self.cfg, "search_pacing_seconds", 0.0)
             snippet_ranking: dict | None = None
+            search_providers: Counter = Counter()
             query_results: dict[str, set[str]] = {}
             covered_facets = 0
             issued_facets = 0
@@ -754,9 +762,30 @@ async def crawl_one(url: str, idx: int, round_selected: list[dict]):
                     for index, query in enumerate(round_queries):
                         if index and search_pacing > 0:
                             await asyncio.sleep(search_pacing)
-                        results_by_facet.append(await self.searxng.search(
+                        facet_results = await self.searxng.search(
                             query, max_results=max_sources, language=language
-                        ))
+                        )
+                        # SearXNG stays primary. The keyed fallback engages
+                        # only when a query returns nothing -- which is what a
+                        # fully blocked engine pool looks like -- so the
+                        # private path remains the default and the paid path
+                        # is insurance (ADR-002 stage 2).
+                        if not facet_results and self.serper.configured:
+                            facet_results = await self.serper.search(
+                                query, max_results=max_sources, language=language
+                            )
+                            search_providers["serper" if facet_results
+                                             else "none"] += 1
+                            if facet_results:
+                                logger.info("search_fallback_used", extra={
+                                    "job_id": job_id, "phase": "search",
+                                    "diagnostic": {"provider": "serper",
+                                                   "results": len(facet_results)},
+                                })
+                        else:
+                            search_providers[
+                                "searxng" if facet_results else "none"] += 1
+                        results_by_facet.append(facet_results)
                 # Spend the crawl cap on the most relevant candidates rather
                 # than on whatever the engine happened to rank first.
                 if planning_enabled:
@@ -1083,6 +1112,7 @@ async def crawl_one(url: str, idx: int, round_selected: list[dict]):
                     "robots_respected": respect_robots,
                     "source_screening": source_screening,
                     "snippet_ranking": snippet_ranking,
+                    "search_providers": dict(search_providers),
                     "query_plan": acquisition_provenance(
                         plan, issued_queries=issued_queries,
                         decisions=plan_decisions, rounds=rounds,

research-hub/tests/test_query_plan.py

@@ -10,6 +10,7 @@
 """
 
 import asyncio
+import json
 
 import pytest
 
@@ -431,6 +432,15 @@ def save(self, *_args, **_kwargs):
                                      reply_queue=reply_queue,
                                      vector_queue=vector_queue)
     orchestrator.searxng = Searx()
+
+    class NoFallback:
+        """The keyed fallback is unset in tests: SearXNG-only behaviour."""
+        configured = False
+
+        async def search(self, *_args, **_kwargs):
+            return []
+
+    orchestrator.serper = NoFallback()
     orchestrator.crawl4ai = Crawler()
     orchestrator.documents = Documents()
 
@@ -1213,3 +1223,74 @@ def test_search_pacing_default_is_set_and_bounded():
         _config(search_pacing_seconds=-1.0)
     with pytest.raises(ValueError):
         _config(search_pacing_seconds=99.0)
+
+
+# --- ADR-002 stage 2: the keyed search fallback -----------------------------
+
+def _serper(handler, key="test-serper-key"):
+    import httpx
+    from app.clients import SerperClient
+    subject = SerperClient(key)
+    subject._client = httpx.AsyncClient(
+        base_url="https://google.serper.dev",
+        transport=httpx.MockTransport(handler),
+    )
+    return subject
+
+
+def test_serper_maps_organic_results_to_the_shared_shape():
+    import httpx
+
+    def handler(request):
+        assert request.headers["X-API-KEY"] == "test-serper-key"
+        body = json.loads(request.content)
+        assert body["q"] == "a query"
+        return httpx.Response(200, json={"organic": [
+            {"title": "T", "link": "https://a.example", "snippet": "S",
+             "date": "2026-01-02", "position": 1},
+            {"title": "U", "link": "https://b.example", "snippet": "V"},
+        ]})
+
+    results = run(_serper(handler).search("a query", max_results=10))
+    assert results == [
+        {"url": "https://a.example", "title": "T", "snippet": "S",
+         "published_at": "2026-01-02"},
+        {"url": "https://b.example", "title": "U", "snippet": "V",
+         "published_at": None},
+    ]
+
+
+def test_serper_is_inert_without_a_key():
+    """Unset means acquisition behaves exactly as it did before."""
+    from app.clients import SerperClient
+    subject = SerperClient("")
+    assert subject.configured is False
+    assert run(subject.search("a query")) == []
+
+
+def test_serper_quota_error_returns_no_results_rather_than_raising():
+    import httpx
+    subject = _serper(lambda _r: httpx.Response(429, json={"message": "quota"}))
+    assert run(subject.search("a query")) == []
+
+
+def test_serper_transport_failure_returns_no_results():
+    import httpx
+
+    def handler(request):
+        raise httpx.ConnectError("down", request=request)
+
+    assert run(_serper(handler).search("a query")) == []
+
+
+def test_serper_drops_entries_without_a_link():
+    import httpx
+    subject = _serper(lambda _r: httpx.Response(200, json={"organic": [
+        {"title": "no link"}, {"title": "ok", "link": "https://c.example"},
+    ]}))
+    assert [r["url"] for r in run(subject.search("q"))] == ["https://c.example"]
+
+
+def test_serper_key_is_excluded_from_config_repr():
+    """The key must never reach a log through a Config repr."""
+    assert "shhh" not in repr(_config(serper_api_key="shhh"))