Fix the defects a mutation probe found, and the architecture that hid them (#6)

A mutation probe of the existing suite killed 5 of 6 mutants aimed at pure
functions and 1 of 7 aimed at the I/O boundary - deleting the API-key
middleware entirely left it green, as did replacing the Bing search URL
with a constant. Line coverage read 93% and could not see the difference.

Fixes every defect that hid behind that gap: the blocking DuckDuckGo call
that froze the event loop on the primary search path, an empty query list
500ing all six search endpoints, a non-ASCII API key crashing the auth
check, upstream failures reported as "patent not found" (which
MCP_INSTRUCTIONS tells agents to believe and not retry), unbounded bulk
fan-out, and three MCP tools advertising no output schema.

Then the cause: orchestration policy moved out of the route handlers into
services.py, with its stateful collaborators injected via Depends. Also
adds ruff to CI, closes the httpx client in the lifespan, and stops the
circuit breaker treating "zero results" as evidence a backend is down.

97 -> 238 tests, app.py 65% -> 97% coverage, mutants killed 6/13 -> 18/18.

.github/workflows/deploy-to-hf.yml

@@ -24,6 +24,9 @@ jobs:
       - name: Install Playwright's Chromium
         run: playwright install --with-deps chromium
 
+      - name: Lint
+        run: ruff check .
+
       - name: Run tests
         run: pytest
 

.github/workflows/test.yml

@@ -29,5 +29,8 @@ jobs:
       - name: Install Playwright's Chromium
         run: playwright install --with-deps chromium
 
+      - name: Lint
+        run: ruff check .
+
       - name: Run tests
         run: pytest

.gitignore

@@ -1,4 +1,7 @@
 __pycache__
 .vscode
 .venv
-.env
+.env
+.coverage
+htmlcov
+.ruff_cache

CONTRIBUTING.md

@@ -26,10 +26,11 @@ dependency update broke this."
 ## Running tests
 
 ```bash
+ruff check .   # same lint CI runs
 pytest
 ```
 
-97 tests, a few seconds, fully offline. All outbound HTTP is mocked with
+238 tests, a few seconds, fully offline. All outbound HTTP is mocked with
 `respx`; the Playwright-driven scrapers (Bing, Brave, Google Scholar,
 Google Patents search) run against a real headless Chromium but navigate to
 local fixture HTML instead of the live sites — see `tests/helpers.py` for
@@ -54,6 +55,34 @@ bug here — fix it the same way as anything else (see below): update the
 fixture HTML to match the new real markup, confirm the test fails against
 the current selectors, fix the selectors, confirm green.
 
+## How the code is laid out
+
+- **`app.py`** is HTTP wiring only: request validation, dependency
+  providers, the lifespan, and mapping domain errors onto status codes.
+- **`services.py`** holds the orchestration policy - which backend to try,
+  in what order, when to fall back, what counts as evidence a backend is
+  unhealthy, and when a patent is genuinely absent rather than merely
+  unreachable. None of that is HTTP-specific, and keeping it out of route
+  handlers is why most of its tests need no ASGI client.
+- **`serp.py` / `scrap.py` / `ops.py`** are the backends. Each scraper is
+  split into a pure URL builder, a navigation step, and an `_extract_*`
+  function taking an already-loaded page, so all three can be tested
+  separately.
+- **`utils.py` / `circuit_breaker.py`** are dependency-free helpers.
+
+The services take their stateful collaborators (HTTP client, browser,
+circuit breaker, OPS credentials) as constructor arguments. That is
+deliberate and worth preserving: it is what lets a test build a service
+with its own circuit breaker instead of resetting a process-wide singleton
+between tests, and hand in a stand-in for the OPS credentials instead of
+assigning to private attributes on the real token manager. `app.py`'s
+`get_search_service` / `get_patent_service` providers exist so
+`app.dependency_overrides` can replace them wholesale in a test.
+
+The stateless backend functions stay module-level imports in
+`services.py`; monkeypatching one of those in a test is fine, because they
+hold no state to leak between tests.
+
 ## The development loop (TDD, not test-after)
 
 Every change in this codebase's history followed the same loop, and new
@@ -77,6 +106,22 @@ changes should too:
    confirm the test actually catches it, then revert. This caught a real
    gap once already: a circuit breaker test that looked correct was
    actually vacuous until this step exposed it.
+
+   Applying this systematically is worth a pass of its own, and one such
+   pass is where most of the tests added since came from. Line coverage
+   was 93% and hid the problem completely: the suite killed five of six
+   mutants aimed at pure functions, and one of seven aimed at the I/O
+   boundary. Deleting the entire API-key middleware left it green, as did
+   replacing the Bing search URL with a constant. The lesson worth keeping
+   is *where* to point this technique - at the code that touches the
+   outside world, not the code that's easy to test.
+
+   A useful heuristic fell out of it: if a property has no behavioural
+   signature at all, no black-box test can pin it. `==` and
+   `secrets.compare_digest` return the same answer for every input and
+   differ only in timing, so that one test asserts structurally that the
+   comparison goes through `compare_digest`. Reach for that only when
+   there is genuinely nothing observable to assert on.
 6. **Watch for shared mutable state across tests.** A few characterization
    tests in this codebase turned out to be recording real state into a
    module-level singleton (a circuit breaker) with nothing resetting it
@@ -93,8 +138,13 @@ changes should too:
 
 ## CI / deployment
 
-- **`.github/workflows/test.yml`** runs the full suite on every pull
-  request against `main`, so a change gets feedback before merge.
+- **`.github/workflows/test.yml`** runs `ruff check .` and then the full
+  suite on every pull request against `main`, so a change gets feedback
+  before merge. The lint rule set (see `pyproject.toml`) is deliberately
+  small - pyflakes-equivalent checks plus bugbear - because the first pass
+  over this repo found a duplicate import shadowing another, four unused
+  imports and an f-string with no placeholders. Widen it when the team
+  wants more rather than starting broad and accumulating ignores.
 - **`.github/workflows/deploy-to-hf.yml`** runs the same suite again
   (`test` job) on every push to `main`, and only then (`needs: test`)
   pushes the repo to the Hugging Face Space (`deploy` job) with

README.md

@@ -19,13 +19,15 @@ pinned: true
 ```
 pip install -r requirements.txt -r requirements-dev.txt
 playwright install chromium   # if not already present on the machine
+ruff check .
 pytest
 ```
 
 The suite mocks all outbound HTTP (via `respx`) and drives the Playwright-based
-scrapers against local fixture HTML instead of live search engines - see
-`tests/helpers.py` for how - so it runs fully offline and doesn't depend on,
-or risk tripping, any site's anti-bot measures.
+scrapers against local fixture HTML loaded straight into a real headless
+Chromium - see the `page_factory` fixture in `tests/conftest.py` - so it runs
+fully offline and doesn't depend on, or risk tripping, any site's anti-bot
+measures.
 
 See [`CONTRIBUTING.md`](CONTRIBUTING.md) for the development methodology
 (TDD workflow, dependency pinning policy, CI/deployment) behind this project.

app.py

@@ -1,24 +1,24 @@
-import asyncio
 import logging
 import os
 import secrets
 from contextlib import asynccontextmanager
 from typing import Annotated, Optional
-from fastapi import APIRouter, FastAPI, HTTPException, Request
+
+import httpx
+import uvicorn
+from fastapi import Depends, FastAPI, Request
 from fastapi.responses import JSONResponse
 from fastapi.routing import APIRouter
-import httpx
-from httpx import HTTPStatusError
+from playwright.async_api import Browser, async_playwright
 from pydantic import BaseModel, Field, StringConstraints
-from playwright.async_api import async_playwright, Browser, BrowserContext, Page
-import uvicorn
 
-from circuit_breaker import CircuitBreaker, CircuitOpenError
-from scrap import PatentScrapBulkResponse, PatentScrapResult, scrap_patent_async, scrap_patent_bulk_async
-from serp import PATENT_ID_CORE, SerpQuery, SerpResults, query_arxiv, query_bing_search, query_brave_search, query_ddg_search, query_google_patents, query_google_scholar
-from ops import OPSBulkResponse, OPSNotConfigured, ops_scrap_patent, ops_scrap_patent_bulk, ops_search, token_manager as ops_token_manager
-from utils import log_gathered_exceptions
+from circuit_breaker import CircuitBreaker
 from mcp_server import mount_mcp_server
+from ops import OPSBulkResponse, token_manager as ops_token_manager
+from scrap import PatentScrapBulkResponse, PatentScrapResult
+from serp import PATENT_ID_CORE, SerpQuery, SerpResults
+from services import (OPSUnconfigured, PatentNotFound, PatentService,
+                      SearchService, UpstreamUnavailable)
 
 # Anchored version of serp.py's PATENT_ID_CORE: validates a whole
 # user-supplied patent id rather than finding one inside free text. Rejects
@@ -26,6 +26,11 @@ from mcp_server import mount_mcp_server
 # instead of it reaching Google Patents/OPS as a confusing request.
 PatentId = Annotated[str, StringConstraints(pattern=rf"^{PATENT_ID_CORE}$")]
 
+# Same reasoning as MAX_QUERIES_PER_REQUEST in serp.py: one accepted
+# request must not be able to queue unbounded outbound work. The OPS
+# variant is the expensive one - three requests per id.
+MAX_BULK_PATENT_IDS = 50
+
 logging.basicConfig(
     level=logging.INFO,
     format='[%(asctime)s][%(levelname)s][%(filename)s:%(lineno)d]: %(message)s',
@@ -36,10 +41,19 @@ logging.basicConfig(
 playwright = None
 pw_browser: Optional[Browser] = None
 
-# httpx client
+# httpx client. Constructed at import so the dependency providers below can
+# close over it, but its lifetime is owned by `api_lifespan`, which closes
+# it on shutdown alongside the browser.
 httpx_client = httpx.AsyncClient(timeout=30, limits=httpx.Limits(
     max_connections=30, max_keepalive_connections=20))
 
+# Shared across all queries and requests: once a backend has failed
+# `failure_threshold` times in a row, skip it for `cooldown_seconds` instead
+# of attempting (and paying the timeout cost of) another call that's very
+# likely to fail - and, more importantly, stop hammering a backend that may
+# already be rate-limiting or blocking this deployment's IP.
+_backend_circuit_breaker = CircuitBreaker(failure_threshold=3, cooldown_seconds=60.0)
+
 # ===================== Optional API key protection =====================
 # Unset by default, which keeps the API exactly as open as before. Set
 # SERPENT_API_KEY on the deployment to require every request (REST and MCP
@@ -89,23 +103,78 @@ async def api_lifespan(app: FastAPI):
         await pw_browser.close()
     if playwright:
         await playwright.stop()
+    await httpx_client.aclose()
 
 app = FastAPI(lifespan=api_lifespan, docs_url="/",
               title="SERPent", description=_load_docs())
 
 
+# ============================ service dependencies ============================
+# The orchestration itself lives in services.py; these just wire the app's
+# collaborators into it. Overriding one of these via
+# `app.dependency_overrides` is how a test supplies its own circuit breaker
+# or OPS credentials, rather than reassigning module globals.
+
+
+def get_search_service() -> SearchService:
+    return SearchService(
+        http_client=httpx_client,
+        # A provider, not the browser itself: the lifespan starts it after
+        # import, and leaves it None when Playwright fails to start.
+        browser_provider=lambda: pw_browser,
+        circuit_breaker=_backend_circuit_breaker,
+        ops_tokens=ops_token_manager,
+    )
+
+
+def get_patent_service() -> PatentService:
+    return PatentService(http_client=httpx_client, ops_tokens=ops_token_manager)
+
+
+SearchServiceDep = Annotated[SearchService, Depends(get_search_service)]
+PatentServiceDep = Annotated[PatentService, Depends(get_patent_service)]
+
+
 @app.middleware("http")
 async def api_key_guard(request: Request, call_next):
     """No-op unless SERPENT_API_KEY is set; then gates every path but the docs."""
     if SERPENT_API_KEY and request.url.path not in _PUBLIC_PATHS:
         supplied = _extract_supplied_key(request)
-        if not supplied or not secrets.compare_digest(supplied, SERPENT_API_KEY):
+        # Compare as bytes: compare_digest raises TypeError on str arguments
+        # containing non-ASCII characters, and Starlette decodes headers as
+        # latin-1, so an accented byte in the header would otherwise turn
+        # this check into an unhandled 500 for any anonymous caller.
+        if not supplied or not secrets.compare_digest(
+                supplied.encode("utf-8"), SERPENT_API_KEY.encode("utf-8")):
             return JSONResponse(
                 {"detail": "Missing or invalid API key. Supply it via the X-API-Key or Authorization: Bearer header."},
                 status_code=401,
             )
     return await call_next(request)
 
+
+# ========================= domain errors -> status codes =========================
+# 404 is a claim that the patent does not exist, and MCP_INSTRUCTIONS tells
+# agents to believe it and move on without retrying - so the services raise
+# PatentNotFound only when a backend positively said so, and everything else
+# arrives here as UpstreamUnavailable.
+
+
+@app.exception_handler(PatentNotFound)
+async def _patent_not_found_handler(request: Request, exc: PatentNotFound):
+    return JSONResponse({"detail": str(exc)}, status_code=404)
+
+
+@app.exception_handler(UpstreamUnavailable)
+async def _upstream_unavailable_handler(request: Request, exc: UpstreamUnavailable):
+    return JSONResponse({"detail": str(exc)}, status_code=504 if exc.timeout else 502)
+
+
+@app.exception_handler(OPSUnconfigured)
+async def _ops_unconfigured_handler(request: Request, exc: OPSUnconfigured):
+    return JSONResponse({"detail": str(exc)}, status_code=503)
+
+
 # Router for scrapping related endpoints
 scrap_router = APIRouter(prefix="/scrap", tags=["scrapping"])
 # Router for SERP-scrapping related endpoints
@@ -116,243 +185,105 @@ ops_router = APIRouter(prefix="/ops", tags=["EPO OPS"])
 # ===================== Search endpoints =====================
 
 
-def _shape_serp_results(results: list) -> SerpResults:
-    """Flatten a list of per-query result-lists (or Exceptions, from a
-    `return_exceptions=True` gather) into one SerpResults, surfacing the
-    last error only when every query failed.
-    """
-    filtered_results = [r for r in results if not isinstance(r, Exception)]
-    flattened_results = [
-        item for sublist in filtered_results for item in sublist]
-
-    if len(filtered_results) == 0:
-        return SerpResults(results=[], error=str(results[-1]))
-
-    return SerpResults(results=flattened_results, error=None)
-
-
-async def _run_serp_queries(fn, params: SerpQuery, context: str) -> SerpResults:
-    """Run `fn(query, n_results)` concurrently for every query in `params`,
-    log any failures, and shape the result the way every simple /serp and
-    /ops search endpoint below expects.
-    """
-    results = await asyncio.gather(
-        *[fn(q, params.n_results) for q in params.queries], return_exceptions=True)
-    log_gathered_exceptions(results, context, params)
-    return _shape_serp_results(results)
-
-
 @serp_router.post("/search_scholar")
-async def search_google_scholar(params: SerpQuery):
+async def search_google_scholar(params: SerpQuery, service: SearchServiceDep) -> SerpResults:
     """Queries google scholar for the specified query"""
     logging.info(f"Searching Google Scholar for queries: {params.queries}")
-    return await _run_serp_queries(
-        lambda q, n: query_google_scholar(pw_browser, q, n), params, "google scholar search")
+    return await service.google_scholar(params)
 
 
 @serp_router.post("/search_arxiv")
-async def search_arxiv(params: SerpQuery):
+async def search_arxiv(params: SerpQuery, service: SearchServiceDep) -> SerpResults:
     """Searches arxiv for the specified queries and returns the found documents."""
     logging.info(f"Searching Arxiv for queries: {params.queries}")
-    return await _run_serp_queries(
-        lambda q, n: query_arxiv(httpx_client, q, n), params, "arxiv search")
+    return await service.arxiv(params)
 
 
 @serp_router.post("/search_patents")
-async def search_patents(params: SerpQuery) -> SerpResults:
+async def search_patents(params: SerpQuery, service: SearchServiceDep) -> SerpResults:
     """Searches google patents for the specified queries and returns the found documents.
 
     Falls back to the EPO OPS API for any query Google Patents returns nothing
     for, when OPS credentials are configured.
     """
     logging.info(f"Searching Google Patents for queries: {params.queries}")
-    results = await asyncio.gather(*[query_google_patents(pw_browser, q, params.n_results) for q in params.queries], return_exceptions=True)
-    log_gathered_exceptions(results, "google patent search", params)
-
-    # Fall back to OPS for queries that errored or returned no results.
-    if ops_token_manager.configured:
-        for i, res in enumerate(results):
-            if isinstance(res, Exception) or not res:
-                try:
-                    logging.info(
-                        f"Google Patents empty for `{params.queries[i]}`, trying OPS.")
-                    results[i] = await ops_search(httpx_client, params.queries[i], params.n_results)
-                except Exception as e:
-                    logging.warning(f"OPS fallback failed: {e}")
-
-    return _shape_serp_results(results)
+    return await service.patents(params)
 
 
 @serp_router.post("/search_brave")
-async def search_brave(params: SerpQuery) -> SerpResults:
+async def search_brave(params: SerpQuery, service: SearchServiceDep) -> SerpResults:
     """Searches brave search for the specified queries and returns the found documents."""
     logging.info(f"Searching Brave Search for queries: {params.queries}")
-    return await _run_serp_queries(
-        lambda q, n: query_brave_search(pw_browser, q, n), params, "brave search")
+    return await service.brave(params)
 
 
 @serp_router.post("/search_bing")
-async def search_bing(params: SerpQuery) -> SerpResults:
+async def search_bing(params: SerpQuery, service: SearchServiceDep) -> SerpResults:
     """Searches Bing search for the specified queries and returns the found documents."""
     logging.info(f"Searching Bing Search for queries: {params.queries}")
-    return await _run_serp_queries(
-        lambda q, n: query_bing_search(pw_browser, q, n), params, "bing search")
+    return await service.bing(params)
 
 
 @serp_router.post("/search_duck")
-async def search_duck(params: SerpQuery) -> SerpResults:
+async def search_duck(params: SerpQuery, service: SearchServiceDep) -> SerpResults:
     """Searches duckduckgo for the specified queries and returns the found documents"""
     logging.info(f"Searching DuckDuckGo for queries: {params.queries}")
-    return await _run_serp_queries(
-        lambda q, n: query_ddg_search(q, n), params, "duckduckgo search")
-
-
-# Shared across all queries and requests: once a backend has failed
-# `failure_threshold` times in a row, skip it for `cooldown_seconds` instead
-# of attempting (and paying the timeout cost of) another call that's very
-# likely to fail - and, more importantly, stop hammering a backend that may
-# already be rate-limiting or blocking this deployment's IP.
-_backend_circuit_breaker = CircuitBreaker(failure_threshold=3, cooldown_seconds=60.0)
-
-
-async def _search_one(q: str, n_results: int) -> tuple[str, list[dict], Optional[str]]:
-    """Try DDG, then Brave, then Bing for a single query; stop at the first success."""
-    backends = [
-        ("DuckDuckGo", lambda: query_ddg_search(q, n_results)),
-        ("Brave Search", lambda: query_brave_search(pw_browser, q, n_results)),
-        ("Bing", lambda: query_bing_search(pw_browser, q, n_results)),
-    ]
-    last_error: Optional[Exception] = None
-    for name, call in backends:
-        try:
-            _backend_circuit_breaker.before_call(name)
-        except CircuitOpenError as e:
-            logging.info(f"Skipping {name} for query `{q}`: {e}")
-            last_error = e
-            continue
-
-        try:
-            logging.info(f"Querying {name} with query: `{q}`")
-            result = await call()
-            _backend_circuit_breaker.record_success(name)
-            return q, result, None
-        except Exception as e:
-            logging.error(f"Failed to query {name} with query `{q}`: {e}")
-            _backend_circuit_breaker.record_failure(name)
-            last_error = e
-
-    return q, [], f"All backends failed for query '{q}': {last_error}"
+    return await service.duckduckgo(params)
 
 
 @serp_router.post("/search")
-async def search(params: SerpQuery):
+async def search(params: SerpQuery, service: SearchServiceDep) -> SerpResults:
     """Attempts to search the specified queries using ALL backends"""
-    outcomes = await asyncio.gather(*[_search_one(q, params.n_results) for q in params.queries])
-
-    results: list[dict] = []
-    errors: list[str] = []
-    for _q, res, err in outcomes:
-        results.extend(res)
-        if err:
-            errors.append(err)
-
-    if len(results) == 0:
-        return SerpResults(results=[], error="; ".join(errors) if errors else "All backends are rate-limited.")
-
-    return SerpResults(results=results, error="; ".join(errors) if errors else None)
+    return await service.search(params)
 
 # =========================== Scrapping endpoints ===========================
 
 
 @scrap_router.get("/scrap_patent/{patent_id}")
-async def scrap_patent(patent_id: PatentId) -> PatentScrapResult:
+async def scrap_patent(patent_id: PatentId, service: PatentServiceDep) -> PatentScrapResult:
     """Scraps the specified patent from Google Patents.
 
     Falls back to the EPO OPS API (which covers patents missing from Google
     Patents) when the scrape fails and OPS credentials are configured.
     """
-    try:
-        return await scrap_patent_async(httpx_client, f"https://patents.google.com/patent/{patent_id}/en")
-    except HTTPStatusError as e:
-        logging.warning(
-            f"Google Patents returned {e.response.status_code} for {patent_id}.")
-    except Exception as e:
-        logging.warning(f"Failed to scrap patent {patent_id}: {e}")
-
-    if not ops_token_manager.configured:
-        raise HTTPException(
-            status_code=404,
-            detail=f"Patent '{patent_id}' not found on Google Patents (EPO OPS fallback not configured).")
-
-    try:
-        logging.info(f"Trying OPS for patent {patent_id}.")
-        return await ops_scrap_patent(httpx_client, patent_id)
-    except HTTPStatusError as e:
-        if e.response.status_code == 404:
-            raise HTTPException(
-                status_code=404,
-                detail=f"Patent '{patent_id}' not found on Google Patents or EPO OPS.")
-        raise HTTPException(
-            status_code=502,
-            detail=f"EPO OPS returned {e.response.status_code} for '{patent_id}'.")
-    except Exception as e:
-        logging.warning(f"OPS fallback failed for {patent_id}: {e}")
-        raise HTTPException(
-            status_code=502,
-            detail=f"EPO OPS request failed for '{patent_id}': {e}")
+    return await service.scrap(patent_id)
 
 
 class ScrapPatentsRequest(BaseModel):
     """Request model for scrapping multiple patents."""
     patent_ids: list[PatentId] = Field(...,
+                                       min_length=1,
+                                       max_length=MAX_BULK_PATENT_IDS,
                                        description="List of patent IDs to scrap")
 
 
 @scrap_router.post("/scrap_patents_bulk", response_model=PatentScrapBulkResponse)
-async def scrap_patents(params: ScrapPatentsRequest) -> PatentScrapBulkResponse:
+async def scrap_patents(params: ScrapPatentsRequest,
+                        service: PatentServiceDep) -> PatentScrapBulkResponse:
     """Scraps multiple patents from Google Patents."""
-    patents = await scrap_patent_bulk_async(httpx_client, params.patent_ids)
-    return patents
+    return await service.scrap_bulk(params.patent_ids)
 
 # =========================== EPO OPS endpoints ===========================
 
 
 @ops_router.post("/search")
-async def ops_keyword_search(params: SerpQuery) -> SerpResults:
+async def ops_keyword_search(params: SerpQuery, service: SearchServiceDep) -> SerpResults:
     """Keyword-searches patents via the official EPO OPS API."""
     logging.info(f"Searching EPO OPS for queries: {params.queries}")
-    return await _run_serp_queries(
-        lambda q, n: ops_search(httpx_client, q, n), params, "OPS search")
+    return await service.ops_keyword_search(params)
 
 
 @ops_router.get("/scrap_patent/{patent_id}")
-async def ops_get_patent(patent_id: PatentId) -> PatentScrapResult:
+async def ops_get_patent(patent_id: PatentId, service: PatentServiceDep) -> PatentScrapResult:
     """Retrieves a patent (biblio + abstract + claims + description) via EPO OPS."""
-    if not ops_token_manager.configured:
-        raise HTTPException(
-            status_code=503,
-            detail="EPO OPS is not configured (OPS_CONSUMER_KEY / OPS_CONSUMER_SECRET missing).")
-    try:
-        return await ops_scrap_patent(httpx_client, patent_id)
-    except OPSNotConfigured:
-        raise HTTPException(status_code=503, detail="EPO OPS is not configured.")
-    except HTTPStatusError as e:
-        if e.response.status_code == 404:
-            raise HTTPException(
-                status_code=404, detail=f"Patent '{patent_id}' not found in EPO OPS.")
-        raise HTTPException(
-            status_code=502,
-            detail=f"EPO OPS returned {e.response.status_code} for '{patent_id}'.")
-    except Exception as e:
-        logging.warning(f"Failed to retrieve patent {patent_id} from OPS: {e}")
-        raise HTTPException(
-            status_code=502, detail=f"EPO OPS request failed for '{patent_id}': {e}")
+    return await service.ops_scrap(patent_id)
 
 
 @ops_router.post("/scrap_patents_bulk", response_model=OPSBulkResponse)
-async def ops_get_patents_bulk(params: ScrapPatentsRequest) -> OPSBulkResponse:
+async def ops_get_patents_bulk(params: ScrapPatentsRequest,
+                               service: PatentServiceDep) -> OPSBulkResponse:
     """Retrieves multiple patents via EPO OPS."""
-    return await ops_scrap_patent_bulk(httpx_client, params.patent_ids)
+    return await service.ops_scrap_bulk(params.patent_ids)
 
 # ===============================================================================
 

ops.py

@@ -312,9 +312,20 @@ class OPSBulkResponse(BaseModel):
     failed_ids: list[str]
 
 
+# Lower than the Google Patents equivalent because each id here costs three
+# OPS requests (biblio + claims + description) rather than one page fetch.
+BULK_OPS_CONCURRENCY_LIMIT = 5
+
+
 async def ops_scrap_patent_bulk(client: AsyncClient, numbers: list[str]) -> OPSBulkResponse:
+    semaphore = asyncio.Semaphore(BULK_OPS_CONCURRENCY_LIMIT)
+
+    async def scrap_one(number: str):
+        async with semaphore:
+            return await ops_scrap_patent(client, number)
+
     results = await asyncio.gather(
-        *[ops_scrap_patent(client, n) for n in numbers], return_exceptions=True)
+        *[scrap_one(n) for n in numbers], return_exceptions=True)
     patents = [r for r in results if not isinstance(r, Exception)]
     failed = [numbers[i] for i, r in enumerate(
         results) if isinstance(r, Exception)]

pyproject.toml

@@ -0,0 +1,17 @@
+# Lint configuration. The rule set is deliberately small: this repo had no
+# linter, and the first pass found a duplicate import shadowing another, four
+# unused imports and an f-string with no placeholders - all of which a
+# default pyflakes-equivalent catches. Widen the selection when the team
+# wants more, rather than starting from a large set and adding ignores.
+[tool.ruff]
+line-length = 120
+
+[tool.ruff.lint]
+select = [
+    "F",    # pyflakes: unused/duplicate imports, f-strings without placeholders
+    "E9",   # syntax and IO errors
+    "B",    # bugbear: likely-wrong constructs
+]
+ignore = [
+    "B008",  # FastAPI's Depends()/Field() in defaults is the documented idiom
+]

requirements-dev.txt

@@ -10,3 +10,5 @@
 pytest>=8,<10
 pytest-asyncio>=1.4.0,<1.5
 respx>=0.23.1,<0.24
+# Lint. Pinned like everything else; `ruff check .` is what CI runs.
+ruff>=0.14.6,<0.15

scrap.py

@@ -1,5 +1,4 @@
 import asyncio
-import logging
 import re
 from typing import Optional
 from httpx import AsyncClient
@@ -131,11 +130,24 @@ class PatentScrapBulkResponse(BaseModel):
     failed_ids: list[str]
 
 
+# Cap how many patent pages are fetched at once. Without this a single
+# accepted request opened one outbound scrape per id, all at the same time;
+# the Playwright scrapers have had an equivalent limit from the start and
+# this brings the HTTP path in line with them.
+BULK_SCRAP_CONCURRENCY_LIMIT = 10
+
+
 async def scrap_patent_bulk_async(client: AsyncClient, patent_ids: list[str]) -> PatentScrapBulkResponse:
-    """Scrape multiple patents asynchronously."""
+    """Scrape multiple patents asynchronously, a bounded number at a time."""
     urls = [
         f"https://patents.google.com/patent/{pid}/en" for pid in patent_ids]
-    results = await asyncio.gather(*[scrap_patent_async(client, url) for url in urls], return_exceptions=True)
+    semaphore = asyncio.Semaphore(BULK_SCRAP_CONCURRENCY_LIMIT)
+
+    async def scrap_one(url: str):
+        async with semaphore:
+            return await scrap_patent_async(client, url)
+
+    results = await asyncio.gather(*[scrap_one(url) for url in urls], return_exceptions=True)
 
     filtered_results = [
         res for res in results if not isinstance(res, Exception)]

serp.py

@@ -3,11 +3,11 @@ from typing import Optional
 from duckduckgo_search import DDGS
 import httpx
 from pydantic import BaseModel, Field, field_validator
-from playwright.async_api import Browser, BrowserContext, Page, TimeoutError
+from playwright.async_api import Browser, BrowserContext, Page
 from urllib.parse import quote_plus
-import logging
 import re
 from lxml import etree
+import asyncio
 from asyncio import Semaphore
 
 # Concurrency limit for Playwright browser contexts.
@@ -15,8 +15,18 @@ from asyncio import Semaphore
 PLAYWRIGHT_CONCURRENCY_LIMIT = 10
 
 
+# One request must not be able to fan out without limit. Every search
+# endpoint issues one outbound call per query (and `/serp/search` may try
+# three backends for each), so the length of this list is a direct
+# multiplier on outbound load - `n_results` was clamped from the start
+# while the list holding the queries was not bounded at all.
+MAX_QUERIES_PER_REQUEST = 50
+
+
 class SerpQuery(BaseModel):
     queries: list[str] = Field(...,
+                               min_length=1,
+                               max_length=MAX_QUERIES_PER_REQUEST,
                                description="The list of queries to search for")
     n_results: int = Field(
         10, description="Number of results to return for each query. Valid values are 10, 25, 50 and 100")
@@ -50,14 +60,26 @@ class BrowserUnavailableError(Exception):
         super().__init__("Playwright browser is not available (it failed to start at startup).")
 
 
-class DuckDuckGoBlockedException(Exception):
+class EmptyResultsError(Exception):
+    """A backend returned zero results.
+
+    Deliberately distinct from a backend erroring, because zero results is
+    ambiguous: it can mean the query genuinely has no hits, or that the
+    backend served an anti-bot page that parsed to nothing. Callers (the
+    `/serp/search` fallback chain in particular) rely on an exception to
+    move on to the next backend, but should not treat this one as evidence
+    that the backend itself is unhealthy.
+    """
+
+
+class DuckDuckGoBlockedException(EmptyResultsError):
     """Raised when DuckDuckGo returns zero results.
 
     The duckduckgo_search library doesn't always raise when its request is
     blocked/rate-limited — it can also just parse zero results from the
-    anti-bot response page and return an empty list. Since callers (the
-    `/serp/search` fallback chain in particular) rely on an exception to know
-    a backend failed, treat an empty result as a failure too.
+    anti-bot response page and return an empty list. A hard block usually
+    does raise (and that still counts against the backend); this covers the
+    soft case, which is indistinguishable from a genuinely empty search.
     """
 
     def __init__(self, *args):
@@ -78,8 +100,12 @@ async def playwright_open_page(browser: Optional[Browser]):
         try:
             yield page
         finally:
-            await page.close()
-            await context.close()
+            # Close the context even if closing the page fails (a crashed
+            # renderer), otherwise the context leaks for the process's life.
+            try:
+                await page.close()
+            finally:
+                await context.close()
 
 
 async def _block_stylesheet_and_image_resources(route, request):
@@ -113,13 +139,21 @@ async def _extract_google_scholar_results(page: Page, n_results: int) -> list[di
     return results
 
 
+def google_scholar_url(q: str, n_results: int) -> str:
+    """The Google Scholar results URL for a query.
+
+    Split out from the scraper so it can be asserted on directly: the
+    Playwright tests replace `page.goto` and never see the URL.
+    """
+    return f"https://scholar.google.com/scholar?q={quote_plus(q)}&num={n_results}"
+
+
 async def query_google_scholar(browser: Browser, q: str, n_results: int = 10):
     """Queries google scholar for the specified query and number of results. Returns relevant papers"""
     async with playwright_open_page(browser) as page:
         await page.route("**/*", _block_stylesheet_and_image_resources)
 
-        url = f"https://scholar.google.com/scholar?q={quote_plus(q)}&num={n_results}"
-        await page.goto(url)
+        await page.goto(google_scholar_url(q, n_results))
 
         return await _extract_google_scholar_results(page, n_results)
 
@@ -139,7 +173,7 @@ async def _extract_google_patents_results(page: Page, n_results: int) -> list[di
     # Wait for at least one search result item to appear
     # This ensures the page has loaded enough to start scraping
     await page.wait_for_function(
-        f"""() => document.querySelectorAll('search-result-item').length >= 1""",
+        "() => document.querySelectorAll('search-result-item').length >= 1",
         timeout=30_000
     )
 
@@ -169,13 +203,17 @@ async def _extract_google_patents_results(page: Page, n_results: int) -> list[di
     return results[:n_results]
 
 
+def google_patents_search_url(q: str, n_results: int) -> str:
+    """The Google Patents search URL for a query."""
+    return f"https://patents.google.com/?q={quote_plus(q)}&num={n_results}"
+
+
 async def query_google_patents(browser: Browser, q: str, n_results: int = 10):
     """Queries google patents for the specified query and number of results. Returns relevant patents"""
     async with playwright_open_page(browser) as page:
         await page.route("**/*", _block_stylesheet_and_image_resources)
 
-        url = f"https://patents.google.com/?q={quote_plus(q)}&num={n_results}"
-        await page.goto(url)
+        await page.goto(google_patents_search_url(q, n_results))
 
         return await _extract_google_patents_results(page, n_results)
 
@@ -217,13 +255,22 @@ async def _extract_brave_results(page: Page, n_results: int) -> list[dict]:
     return results
 
 
+def brave_search_url(q: str, n_results: int = 10) -> str:
+    """The Brave Search results URL for a query.
+
+    Brave paginates rather than taking a result count, so `n_results` is
+    applied while extracting rather than in the URL; it is accepted here to
+    keep the builders interchangeable.
+    """
+    return f"https://search.brave.com/search?q={quote_plus(q)}"
+
+
 async def query_brave_search(browser: Browser, q: str, n_results: int = 10):
     """Queries Brave Search for the specified query."""
     async with playwright_open_page(browser) as page:
         await page.route("**/*", _block_stylesheet_and_image_resources)
 
-        url = f"https://search.brave.com/search?q={quote_plus(q)}"
-        await page.goto(url)
+        await page.goto(brave_search_url(q, n_results))
 
         return await _extract_brave_results(page, n_results)
 
@@ -266,25 +313,50 @@ async def _extract_bing_results(page: Page, n_results: int) -> list[dict]:
     return results
 
 
+def bing_search_url(q: str, n_results: int = 10) -> str:
+    """The Bing results URL for a query.
+
+    Bing paginates rather than taking a result count, so `n_results` is
+    applied while extracting rather than in the URL; it is accepted here to
+    keep the builders interchangeable.
+    """
+    return f"https://www.bing.com/search?q={quote_plus(q)}"
+
+
 async def query_bing_search(browser: Browser, q: str, n_results: int = 10):
     """Queries bing search for the specified query"""
     async with playwright_open_page(browser) as page:
         await page.route("**/*", _block_stylesheet_and_image_resources)
 
-        url = f"https://www.bing.com/search?q={quote_plus(q)}"
-        await page.goto(url)
+        await page.goto(bing_search_url(q, n_results))
 
         return await _extract_bing_results(page, n_results)
 
 
+def _ddg_text_blocking(q: str, n_results: int) -> list[dict]:
+    """The synchronous half of a DuckDuckGo query.
+
+    `DDGS.text()` performs blocking HTTP I/O. It is deliberately isolated
+    here so `query_ddg_search` can hand it to a worker thread rather than
+    calling it on the event loop.
+    """
+    return list(DDGS().text(q, max_results=n_results))
+
+
 async def query_ddg_search(q: str, n_results: int = 10):
-    """Queries duckduckgo search for the specified query"""
-    ddgs = DDGS()
-    results = []
+    """Queries duckduckgo search for the specified query.
+
+    The underlying library call is synchronous, so it runs in a worker
+    thread: calling it directly would freeze the whole event loop for the
+    duration of the request, and DuckDuckGo is the first backend tried for
+    every query in `/serp/search`.
+    """
+    raw_results = await asyncio.to_thread(_ddg_text_blocking, q, n_results)
 
-    for result in ddgs.text(q, max_results=n_results):
-        results.append(
-            {"title": result["title"], "body": result["body"], "href": result["href"]})
+    results = [
+        {"title": r["title"], "body": r["body"], "href": r["href"]}
+        for r in raw_results
+    ]
 
     if not results:
         raise DuckDuckGoBlockedException()

services.py

@@ -0,0 +1,337 @@
+"""Backend orchestration, separated from HTTP wiring.
+
+The policy in here - which backend to try, in what order, when to fall back
+to another, what counts as evidence a backend is unhealthy, and when a
+patent is genuinely absent rather than merely unreachable - is domain
+policy. It would be equally true behind a CLI or a queue consumer, and it
+was previously written inline in FastAPI route handlers, which meant
+exercising any of it cost an ASGI round-trip.
+
+The stateful collaborators (HTTP client, browser, circuit breaker, OPS
+credentials) are constructor arguments rather than module globals reached
+into at call time. That is what lets a test build a service with its own
+circuit breaker instead of resetting a process-wide singleton between
+tests, and hand in a stand-in for the OPS credentials instead of assigning
+to private attributes on the real token manager.
+
+The stateless backend functions (`query_arxiv` and friends) stay
+module-level imports: they hold no state, so a test that wants to replace
+one can monkeypatch it here without any of the problems shared mutable
+state causes.
+
+Services raise domain errors - `PatentNotFound`, `UpstreamUnavailable` -
+rather than `HTTPException`. Mapping those onto status codes is the HTTP
+layer's job, and lives in app.py.
+"""
+
+import asyncio
+import logging
+from typing import Callable, Optional
+
+import httpx
+from httpx import AsyncClient, HTTPStatusError
+from playwright.async_api import Browser
+
+from circuit_breaker import CircuitBreaker, CircuitOpenError
+from ops import (OPSBulkResponse, ops_scrap_patent, ops_scrap_patent_bulk,
+                 ops_search)
+from scrap import (PatentScrapBulkResponse, PatentScrapResult,
+                   scrap_patent_async, scrap_patent_bulk_async)
+from serp import (EmptyResultsError, SerpQuery, SerpResults, query_arxiv,
+                  query_bing_search, query_brave_search, query_ddg_search,
+                  query_google_patents, query_google_scholar)
+from utils import log_gathered_exceptions
+
+logger = logging.getLogger(__name__)
+
+
+# ================================ domain errors ================================
+
+
+class PatentNotFound(Exception):
+    """Every backend consulted positively said the document is absent.
+
+    Distinct from UpstreamUnavailable on purpose: MCP_INSTRUCTIONS tells an
+    agent that a not-found patent is genuinely absent everywhere and to
+    move on without retrying, so this must never stand in for "we couldn't
+    reach anyone".
+    """
+
+
+class UpstreamUnavailable(Exception):
+    """A backend failed to answer. Says nothing about whether the document
+    exists."""
+
+    def __init__(self, message: str, *, timeout: bool = False):
+        super().__init__(message)
+        self.timeout = timeout
+
+
+class OPSUnconfigured(Exception):
+    """An OPS-only operation was requested without credentials configured."""
+
+
+def _is_not_found(exc: BaseException) -> bool:
+    """True only when a backend positively said the document is absent."""
+    return isinstance(exc, HTTPStatusError) and exc.response.status_code == 404
+
+
+def _as_upstream(exc: BaseException, message: str) -> UpstreamUnavailable:
+    return UpstreamUnavailable(message, timeout=isinstance(exc, httpx.TimeoutException))
+
+
+# ================================ result shaping ================================
+
+
+def shape_serp_results(results: list) -> SerpResults:
+    """Flatten a list of per-query result-lists (or Exceptions, from a
+    `return_exceptions=True` gather) into one SerpResults, surfacing the
+    last error only when every query failed.
+    """
+    if not results:
+        # The request models bound the query list, so this is unreachable
+        # from the endpoints; keep the helper total anyway rather than
+        # indexing into an empty list, since every search path shares it.
+        return SerpResults(results=[], error="No queries were provided.")
+
+    filtered_results = [r for r in results if not isinstance(r, Exception)]
+    flattened_results = [
+        item for sublist in filtered_results for item in sublist]
+
+    if len(filtered_results) == 0:
+        return SerpResults(results=[], error=str(results[-1]))
+
+    return SerpResults(results=flattened_results, error=None)
+
+
+# ================================= search service =================================
+
+
+class SearchService:
+    """Runs SERP queries across the available backends."""
+
+    def __init__(self, *, http_client: AsyncClient,
+                 browser_provider: Callable[[], Optional[Browser]],
+                 circuit_breaker: CircuitBreaker, ops_tokens):
+        self._http = http_client
+        # A provider rather than the browser itself: the browser is started
+        # by the app's lifespan, after this service may already exist, and
+        # is None when Playwright failed to start.
+        self._browser_provider = browser_provider
+        self._breaker = circuit_breaker
+        self._ops_tokens = ops_tokens
+
+    @property
+    def browser(self) -> Optional[Browser]:
+        return self._browser_provider()
+
+    async def _run_queries(self, fn, params: SerpQuery, context: str) -> SerpResults:
+        """Run `fn(query, n_results)` concurrently for every query, log any
+        failures, and shape the result the way every simple search path
+        expects.
+        """
+        results = await asyncio.gather(
+            *[fn(q, params.n_results) for q in params.queries], return_exceptions=True)
+        log_gathered_exceptions(results, context, params.queries)
+        return shape_serp_results(results)
+
+    # ------------------------------ single backend ------------------------------
+
+    async def google_scholar(self, params: SerpQuery) -> SerpResults:
+        return await self._run_queries(
+            lambda q, n: query_google_scholar(self.browser, q, n), params,
+            "google scholar search")
+
+    async def arxiv(self, params: SerpQuery) -> SerpResults:
+        return await self._run_queries(
+            lambda q, n: query_arxiv(self._http, q, n), params, "arxiv search")
+
+    async def brave(self, params: SerpQuery) -> SerpResults:
+        return await self._run_queries(
+            lambda q, n: query_brave_search(self.browser, q, n), params, "brave search")
+
+    async def bing(self, params: SerpQuery) -> SerpResults:
+        return await self._run_queries(
+            lambda q, n: query_bing_search(self.browser, q, n), params, "bing search")
+
+    async def duckduckgo(self, params: SerpQuery) -> SerpResults:
+        return await self._run_queries(
+            lambda q, n: query_ddg_search(q, n), params, "duckduckgo search")
+
+    async def ops_keyword_search(self, params: SerpQuery) -> SerpResults:
+        return await self._run_queries(
+            lambda q, n: ops_search(self._http, q, n), params, "OPS search")
+
+    # -------------------------------- patents --------------------------------
+
+    async def patents(self, params: SerpQuery) -> SerpResults:
+        """Google Patents, falling back to EPO OPS for any query it returns
+        nothing for (when OPS credentials are configured)."""
+        results = await asyncio.gather(
+            *[query_google_patents(self.browser, q, params.n_results)
+              for q in params.queries],
+            return_exceptions=True)
+        log_gathered_exceptions(results, "google patent search", params.queries)
+
+        # Gathered rather than awaited one at a time: this path exists for
+        # when Google Patents is unavailable, so it is exactly when a
+        # request can least afford N sequential round-trips to the slower
+        # backend.
+        if self._ops_tokens.configured:
+            needs_fallback = [
+                i for i, res in enumerate(results)
+                if isinstance(res, Exception) or not res]
+            if needs_fallback:
+                logger.info(
+                    f"Google Patents empty for {len(needs_fallback)} quer(y/ies), trying OPS.")
+                fallback = await asyncio.gather(
+                    *[ops_search(self._http, params.queries[i], params.n_results)
+                      for i in needs_fallback],
+                    return_exceptions=True)
+                # strict: gather returns exactly one result per index, so a
+                # length mismatch here would be a bug worth surfacing.
+                for i, res in zip(needs_fallback, fallback, strict=True):
+                    if isinstance(res, Exception):
+                        logger.warning(f"OPS fallback failed for `{params.queries[i]}`: {res}")
+                    else:
+                        results[i] = res
+
+        return shape_serp_results(results)
+
+    # ----------------------------- all backends -----------------------------
+
+    async def _search_one(self, q: str, n_results: int) -> tuple[str, list[dict], Optional[str]]:
+        """Try DDG, then Brave, then Bing for a single query; stop at the
+        first success."""
+        backends = [
+            ("DuckDuckGo", lambda: query_ddg_search(q, n_results)),
+            ("Brave Search", lambda: query_brave_search(self.browser, q, n_results)),
+            ("Bing", lambda: query_bing_search(self.browser, q, n_results)),
+        ]
+        last_error: Optional[Exception] = None
+        for name, call in backends:
+            try:
+                self._breaker.before_call(name)
+            except CircuitOpenError as e:
+                logger.info(f"Skipping {name} for query `{q}`: {e}")
+                last_error = e
+                continue
+
+            try:
+                logger.info(f"Querying {name} with query: `{q}`")
+                result = await call()
+                self._breaker.record_success(name)
+                return q, result, None
+            except EmptyResultsError as e:
+                # Zero results is ambiguous - no hits, or a soft block that
+                # parsed to nothing. Move on to the next backend, but don't
+                # hold it against this one: the breaker is shared across
+                # every request, so counting it would let a few
+                # unusual-but-legitimate queries disable the primary backend
+                # for everyone. A hard block normally raises a real error,
+                # which is handled below.
+                logger.info(f"{name} returned no results for query `{q}`")
+                last_error = e
+            except Exception as e:
+                logger.error(f"Failed to query {name} with query `{q}`: {e}")
+                self._breaker.record_failure(name)
+                last_error = e
+
+        return q, [], f"All backends failed for query '{q}': {last_error}"
+
+    async def search(self, params: SerpQuery) -> SerpResults:
+        """Search every query across the full backend fallback chain."""
+        outcomes = await asyncio.gather(
+            *[self._search_one(q, params.n_results) for q in params.queries])
+
+        results: list[dict] = []
+        errors: list[str] = []
+        for _q, res, err in outcomes:
+            results.extend(res)
+            if err:
+                errors.append(err)
+
+        if len(results) == 0:
+            return SerpResults(
+                results=[],
+                error="; ".join(errors) if errors else "All backends are rate-limited.")
+
+        return SerpResults(results=results, error="; ".join(errors) if errors else None)
+
+
+# ================================= patent service =================================
+
+
+class PatentService:
+    """Retrieves patent full text, from Google Patents or EPO OPS."""
+
+    def __init__(self, *, http_client: AsyncClient, ops_tokens):
+        self._http = http_client
+        self._ops_tokens = ops_tokens
+
+    async def scrap(self, patent_id: str) -> PatentScrapResult:
+        """Scrape from Google Patents, falling back to EPO OPS.
+
+        Raises PatentNotFound only when every backend consulted said the
+        document is absent; anything else raises UpstreamUnavailable.
+        """
+        google_error: BaseException
+        try:
+            return await scrap_patent_async(
+                self._http, f"https://patents.google.com/patent/{patent_id}/en")
+        except HTTPStatusError as e:
+            google_error = e
+            logger.warning(
+                f"Google Patents returned {e.response.status_code} for {patent_id}.")
+        except Exception as e:
+            google_error = e
+            logger.warning(f"Failed to scrap patent {patent_id}: {e}")
+
+        if not self._ops_tokens.configured:
+            if _is_not_found(google_error):
+                raise PatentNotFound(
+                    f"Patent '{patent_id}' not found on Google Patents "
+                    "(EPO OPS fallback not configured).")
+            raise _as_upstream(
+                google_error,
+                f"Google Patents request failed for '{patent_id}' and the EPO OPS "
+                f"fallback is not configured: {google_error}")
+
+        try:
+            logger.info(f"Trying OPS for patent {patent_id}.")
+            return await ops_scrap_patent(self._http, patent_id)
+        except Exception as e:
+            logger.warning(f"OPS fallback failed for {patent_id}: {e}")
+            # Only claim the patent doesn't exist when *both* backends said so.
+            if _is_not_found(google_error) and _is_not_found(e):
+                raise PatentNotFound(
+                    f"Patent '{patent_id}' not found on Google Patents or EPO OPS.") from e
+            if isinstance(e, HTTPStatusError):
+                raise _as_upstream(
+                    e, f"EPO OPS returned {e.response.status_code} for '{patent_id}'.") from e
+            raise _as_upstream(e, f"EPO OPS request failed for '{patent_id}': {e}") from e
+
+    async def scrap_bulk(self, patent_ids: list[str]) -> PatentScrapBulkResponse:
+        return await scrap_patent_bulk_async(self._http, patent_ids)
+
+    async def ops_scrap(self, patent_id: str) -> PatentScrapResult:
+        """Retrieve via EPO OPS only, with no Google Patents fallback."""
+        if not self._ops_tokens.configured:
+            raise OPSUnconfigured(
+                "EPO OPS is not configured (OPS_CONSUMER_KEY / OPS_CONSUMER_SECRET missing).")
+        try:
+            return await ops_scrap_patent(self._http, patent_id)
+        except HTTPStatusError as e:
+            if e.response.status_code == 404:
+                # OPS is the only backend asked here, so its 404 is the
+                # whole answer.
+                raise PatentNotFound(
+                    f"Patent '{patent_id}' not found in EPO OPS.") from e
+            raise _as_upstream(
+                e, f"EPO OPS returned {e.response.status_code} for '{patent_id}'.") from e
+        except Exception as e:
+            logger.warning(f"Failed to retrieve patent {patent_id} from OPS: {e}")
+            raise _as_upstream(e, f"EPO OPS request failed for '{patent_id}': {e}") from e
+
+    async def ops_scrap_bulk(self, patent_ids: list[str]) -> OPSBulkResponse:
+        return await ops_scrap_patent_bulk(self._http, patent_ids)

tests/conftest.py

@@ -59,3 +59,29 @@ async def browser():
     yield b
     await b.close()
     await pw.stop()
+
+
+@pytest_asyncio.fixture
+async def page_factory(browser):
+    """Returns a callable producing a real Page with fixture HTML loaded.
+
+    This is the whole seam the Playwright-driven tests need now that every
+    scraper is split into a URL builder, a navigation step and an
+    `_extract_*` function: the extractors take an already-loaded page, so
+    `set_content` reaches them directly. It replaces the goto-patching
+    Browser wrapper that used to be required, which had to accept the
+    navigation URL and throw it away - and so hid every URL bug.
+    """
+    contexts = []
+
+    async def make(html: str):
+        context = await browser.new_context()
+        contexts.append(context)
+        page = await context.new_page()
+        await page.set_content(html)
+        return page
+
+    yield make
+
+    for context in contexts:
+        await context.close()

tests/helpers.py

@@ -1,7 +1,4 @@
-"""Shared test helpers: fixture loading and a fake-navigation Playwright
-Browser wrapper used to characterize the scraper functions in `serp.py`
-without hitting live search engines.
-"""
+"""Shared test helpers: fixture loading."""
 
 from pathlib import Path
 
@@ -11,63 +8,3 @@ FIXTURES_DIR = Path(__file__).resolve().parent / "fixtures"
 def load_fixture(*parts: str) -> str:
     """Read a text fixture from tests/fixtures/<parts...>."""
     return FIXTURES_DIR.joinpath(*parts).read_text(encoding="utf-8")
-
-
-class _GotoPatchingContext:
-    """Wraps a real Playwright BrowserContext so every Page it creates loads
-    fixture HTML via `set_content()` instead of navigating over the network.
-    """
-
-    def __init__(self, real_context, html: str):
-        self._real_context = real_context
-        self._html = html
-
-    async def new_page(self):
-        page = await self._real_context.new_page()
-        html = self._html
-
-        async def fake_goto(url, **kwargs):
-            # The scraper functions don't use the response object `goto`
-            # normally returns, so a plain None is a safe stand-in.
-            await page.set_content(html)
-            return None
-
-        page.goto = fake_goto
-        return page
-
-    async def close(self):
-        await self._real_context.close()
-
-
-class _GotoPatchingBrowser:
-    """Wraps a real Playwright Browser the same way, so it can be passed
-    directly to `serp.py` functions in place of the real `pw_browser`.
-    """
-
-    def __init__(self, real_browser, html: str):
-        self._real_browser = real_browser
-        self._html = html
-
-    async def new_context(self, **kwargs):
-        real_context = await self._real_browser.new_context(**kwargs)
-        return _GotoPatchingContext(real_context, self._html)
-
-
-def make_fixture_browser(real_browser, html: str) -> _GotoPatchingBrowser:
-    """Build a Browser-like object for characterizing `serp.py`'s scraper
-    functions against fixture HTML.
-
-    Why this exists: each scraper (e.g. `query_bing_search`) registers its
-    own catch-all `page.route("**/*", ...)` resource blocker *inside* the
-    function, and that handler calls `route.continue_()` (not
-    `route.fallback()`) for the main document request. Playwright invokes
-    the most-recently-registered matching route first, so any route we
-    register on the page *before* calling the function never runs -
-    verified empirically, it's not a documented guarantee either way.
-    Patching `page.goto` itself, on a page created through this wrapper, is
-    the only reliable seam into these functions without touching their
-    source (see the code review this test suite followed from: the
-    fetch/navigate and parse logic aren't separated, which is exactly what
-    makes this workaround necessary).
-    """
-    return _GotoPatchingBrowser(real_browser, html)

tests/test_app.py

@@ -1,12 +1,10 @@
-"""Characterization tests for app.py's search endpoints.
-
-These monkeypatch the underlying query_*/ops_search functions rather than
-hitting real search engines - app.py's own job here is just wiring
-(gather queries concurrently, log failures, flatten results, decide when to
-report an error), which is exactly what's being tested. The app's lifespan
-(which starts a real Playwright browser) is never triggered: ASGITransport
-doesn't run it, and these tests don't need it since pw_browser is never
-actually used once the query functions are patched out.
+"""app.py: HTTP wiring only.
+
+The orchestration these endpoints used to contain now lives in
+services.py and is tested in test_search_service.py and
+test_patent_fallback.py without an HTTP layer. What's left here is what
+app.py is actually responsible for: request validation, dependency wiring,
+the lifespan, and turning domain results into responses.
 """
 
 import httpx
@@ -14,9 +12,8 @@ import pytest
 from httpx import ASGITransport
 
 import app as app_module
-from circuit_breaker import CircuitBreaker
-from scrap import PatentScrapResult
-from serp import SerpQuery
+from scrap import PatentScrapBulkResponse, PatentScrapResult
+from serp import SerpResults
 
 
 @pytest.fixture
@@ -26,151 +23,134 @@ async def client():
         yield c
 
 
-@pytest.fixture(autouse=True)
-def _fresh_backend_circuit_breaker(monkeypatch):
-    """`_backend_circuit_breaker` is a module-level singleton shared across
-    every request in production, which is the point - but it means a
-    backend failure recorded by one test would otherwise leak into the
-    next. Give every test its own instance.
-    """
-    monkeypatch.setattr(app_module, "_backend_circuit_breaker", CircuitBreaker())
-
-
-# ------------------------- _shape_serp_results / _run_serp_queries -------------------------
+@pytest.fixture
+def override_services():
+    """Install stand-in services for the duration of one test."""
+    def _install(*, search=None, patent=None):
+        if search is not None:
+            app_module.app.dependency_overrides[app_module.get_search_service] = lambda: search
+        if patent is not None:
+            app_module.app.dependency_overrides[app_module.get_patent_service] = lambda: patent
 
+    yield _install
+    app_module.app.dependency_overrides.clear()
 
-def test_shape_serp_results_flattens_successful_lists():
-    result = app_module._shape_serp_results([[{"a": 1}], [{"b": 2}, {"c": 3}]])
-    assert result.results == [{"a": 1}, {"b": 2}, {"c": 3}]
-    assert result.error is None
 
+class _RecordingSearchService:
+    """Records which service method each endpoint dispatched to."""
 
-def test_shape_serp_results_surfaces_last_error_when_all_failed():
-    result = app_module._shape_serp_results([ValueError("first"), RuntimeError("second")])
-    assert result.results == []
-    assert "second" in result.error
+    def __init__(self):
+        self.calls = []
 
+    def _record(self, name):
+        async def _fn(params):
+            self.calls.append((name, list(params.queries), params.n_results))
+            return SerpResults(results=[{"title": name}], error=None)
+        return _fn
 
-def test_shape_serp_results_ignores_failures_when_some_succeed():
-    result = app_module._shape_serp_results([[{"a": 1}], RuntimeError("boom")])
-    assert result.results == [{"a": 1}]
-    assert result.error is None
+    def __getattr__(self, name):
+        return self._record(name)
 
 
-async def test_run_serp_queries_calls_fn_per_query_and_shapes_result():
-    params = SerpQuery(queries=["x", "y"], n_results=5)
-    calls = []
+class _StubPatentService:
+    def __init__(self):
+        self.calls = []
 
-    async def fn(q, n):
-        calls.append((q, n))
-        return [{"q": q}]
+    async def scrap(self, patent_id):
+        self.calls.append(("scrap", patent_id))
+        return PatentScrapResult(title="Widget apparatus")
 
-    result = await app_module._run_serp_queries(fn, params, "test context")
+    async def scrap_bulk(self, patent_ids):
+        self.calls.append(("scrap_bulk", patent_ids))
+        return PatentScrapBulkResponse(patents=[], failed_ids=list(patent_ids))
 
-    assert sorted(calls) == [("x", 5), ("y", 5)]
-    assert sorted(result.results, key=str) == sorted([{"q": "x"}, {"q": "y"}], key=str)
+    async def ops_scrap(self, patent_id):
+        self.calls.append(("ops_scrap", patent_id))
+        return PatentScrapResult(title="From OPS")
 
+    async def ops_scrap_bulk(self, patent_ids):
+        self.calls.append(("ops_scrap_bulk", patent_ids))
+        return PatentScrapBulkResponse(patents=[], failed_ids=list(patent_ids))
 
-# ------------------------------------- endpoints -------------------------------------
 
+# ------------------------------- endpoint dispatch -------------------------------
 
-async def test_search_arxiv_flattens_results_from_all_queries(client, monkeypatch):
-    async def fake_query_arxiv(client_arg, q, n):
-        return [{"title": f"paper about {q}", "href": "x", "body": "y", "id": "1"}]
 
-    monkeypatch.setattr(app_module, "query_arxiv", fake_query_arxiv)
+@pytest.mark.parametrize("path, expected_method", [
+    ("/serp/search_scholar", "google_scholar"),
+    ("/serp/search_arxiv", "arxiv"),
+    ("/serp/search_patents", "patents"),
+    ("/serp/search_brave", "brave"),
+    ("/serp/search_bing", "bing"),
+    ("/serp/search_duck", "duckduckgo"),
+    ("/serp/search", "search"),
+    ("/ops/search", "ops_keyword_search"),
+])
+async def test_each_search_endpoint_dispatches_to_its_service_method(
+        client, override_services, path, expected_method):
+    """Guards the wiring itself: every endpoint reaching the right method
+    with the request's queries and result count intact."""
+    service = _RecordingSearchService()
+    override_services(search=service)
 
-    resp = await client.post("/serp/search_arxiv", json={"queries": ["a", "b"], "n_results": 5})
+    resp = await client.post(path, json={"queries": ["a", "b"], "n_results": 25})
 
     assert resp.status_code == 200
-    data = resp.json()
-    assert data["error"] is None
-    assert {r["title"] for r in data["results"]} == {"paper about a", "paper about b"}
+    assert service.calls == [(expected_method, ["a", "b"], 25)]
+    assert resp.json()["results"] == [{"title": expected_method}]
 
 
-async def test_search_arxiv_all_queries_failing_returns_error(client, monkeypatch):
-    async def failing_query_arxiv(client_arg, q, n):
-        raise RuntimeError(f"boom for {q}")
+@pytest.mark.parametrize("method, path, expected_call", [
+    ("get", "/scrap/scrap_patent/US11930446B2", ("scrap", "US11930446B2")),
+    ("get", "/ops/scrap_patent/US11930446B2", ("ops_scrap", "US11930446B2")),
+])
+async def test_patent_endpoints_dispatch_to_their_service_method(
+        client, override_services, method, path, expected_call):
+    service = _StubPatentService()
+    override_services(patent=service)
 
-    monkeypatch.setattr(app_module, "query_arxiv", failing_query_arxiv)
-
-    resp = await client.post("/serp/search_arxiv", json={"queries": ["a"], "n_results": 5})
+    resp = await getattr(client, method)(path)
 
     assert resp.status_code == 200
-    data = resp.json()
-    assert data["results"] == []
-    assert "boom for a" in data["error"]
-
-
-async def test_search_patents_does_not_fall_back_when_google_patents_succeeds(client, monkeypatch):
-    async def fake_google_patents(browser, q, n):
-        return [{"id": "US1", "href": "h", "title": "t", "body": "b"}]
-
-    ops_search_called = False
-
-    async def fake_ops_search(client_arg, q, n):
-        nonlocal ops_search_called
-        ops_search_called = True
-        return []
-
-    monkeypatch.setattr(app_module, "query_google_patents", fake_google_patents)
-    monkeypatch.setattr(app_module, "ops_search", fake_ops_search)
-    monkeypatch.setattr(app_module.ops_token_manager, "_key", "k")
-    monkeypatch.setattr(app_module.ops_token_manager, "_secret", "s")
-
-    resp = await client.post("/serp/search_patents", json={"queries": ["widget"], "n_results": 5})
-
-    data = resp.json()
-    assert data["results"] == [{"id": "US1", "href": "h", "title": "t", "body": "b"}]
-    assert ops_search_called is False
-
-
-async def test_search_patents_falls_back_to_ops_when_google_patents_empty(client, monkeypatch):
-    async def fake_google_patents(browser, q, n):
-        return []
+    assert service.calls == [expected_call]
 
-    async def fake_ops_search(client_arg, q, n):
-        return [{"title": "OPS result", "body": "", "href": "", "id": "US123"}]
 
-    monkeypatch.setattr(app_module, "query_google_patents", fake_google_patents)
-    monkeypatch.setattr(app_module, "ops_search", fake_ops_search)
-    monkeypatch.setattr(app_module.ops_token_manager, "_key", "k")
-    monkeypatch.setattr(app_module.ops_token_manager, "_secret", "s")
+@pytest.mark.parametrize("path, expected_method", [
+    ("/scrap/scrap_patents_bulk", "scrap_bulk"),
+    ("/ops/scrap_patents_bulk", "ops_scrap_bulk"),
+])
+async def test_bulk_endpoints_dispatch_to_their_service_method(
+        client, override_services, path, expected_method):
+    service = _StubPatentService()
+    override_services(patent=service)
 
-    resp = await client.post("/serp/search_patents", json={"queries": ["widget"], "n_results": 5})
+    resp = await client.post(path, json={"patent_ids": ["US11930446B2"]})
 
-    data = resp.json()
-    assert data["results"] == [{"title": "OPS result", "body": "", "href": "", "id": "US123"}]
-
-
-async def test_search_patents_no_fallback_when_ops_not_configured(client, monkeypatch):
-    async def fake_google_patents(browser, q, n):
-        return []
-
-    monkeypatch.setattr(app_module, "query_google_patents", fake_google_patents)
-    monkeypatch.setattr(app_module.ops_token_manager, "_key", None)
-    monkeypatch.setattr(app_module.ops_token_manager, "_secret", None)
-
-    resp = await client.post("/serp/search_patents", json={"queries": ["widget"], "n_results": 5})
-
-    data = resp.json()
-    assert data["results"] == []
-    assert data["error"] is None
+    assert resp.status_code == 200
+    assert service.calls == [(expected_method, ["US11930446B2"])]
 
 
 # ------------------------------- patent_id validation -------------------------------
 
 
-async def test_scrap_patent_rejects_malformed_patent_id(client):
-    resp = await client.get("/scrap/scrap_patent/not-a-patent-id")
+@pytest.mark.parametrize("path", [
+    "/scrap/scrap_patent/not-a-patent-id",
+    "/ops/scrap_patent/not-a-patent-id",
+])
+async def test_malformed_patent_id_is_rejected(client, path):
+    resp = await client.get(path)
     assert resp.status_code == 422
 
 
-async def test_scrap_patent_accepts_well_formed_patent_id(client, monkeypatch):
-    async def fake_scrap(client_arg, url):
-        return PatentScrapResult(title="Widget apparatus")
+@pytest.mark.parametrize("path", [
+    "/scrap/scrap_patents_bulk", "/ops/scrap_patents_bulk"])
+async def test_malformed_patent_id_in_a_bulk_list_is_rejected(client, path):
+    resp = await client.post(path, json={"patent_ids": ["US11930446B2", "garbage!!"]})
+    assert resp.status_code == 422
 
-    monkeypatch.setattr(app_module, "scrap_patent_async", fake_scrap)
+
+async def test_well_formed_patent_id_is_accepted(client, override_services):
+    override_services(patent=_StubPatentService())
 
     resp = await client.get("/scrap/scrap_patent/US11930446B2")
 
@@ -178,38 +158,47 @@ async def test_scrap_patent_accepts_well_formed_patent_id(client, monkeypatch):
     assert resp.json()["title"] == "Widget apparatus"
 
 
-async def test_ops_get_patent_rejects_malformed_patent_id(client):
-    resp = await client.get("/ops/scrap_patent/not-a-patent-id")
-    assert resp.status_code == 422
+# ---------------------------- default dependency wiring ----------------------------
 
 
-async def test_scrap_patents_bulk_rejects_malformed_id_in_list(client):
-    resp = await client.post(
-        "/scrap/scrap_patents_bulk", json={"patent_ids": ["US11930446B2", "garbage!!"]})
-    assert resp.status_code == 422
+def test_the_default_search_service_gets_the_apps_collaborators():
+    """The providers exist so tests can override them; this pins that the
+    un-overridden ones still hand the service the real collaborators."""
+    service = app_module.get_search_service()
 
+    assert service._http is app_module.httpx_client
+    assert service._breaker is app_module._backend_circuit_breaker
+    assert service._ops_tokens is app_module.ops_token_manager
 
-async def test_ops_scrap_patents_bulk_rejects_malformed_id_in_list(client):
-    resp = await client.post("/ops/scrap_patents_bulk", json={"patent_ids": ["garbage!!"]})
-    assert resp.status_code == 422
 
+def test_the_search_service_reads_the_browser_lazily(monkeypatch):
+    """The browser is started by the lifespan, after the service may already
+    exist, and stays None when Playwright fails to start - so the service
+    has to read it per call rather than capture it at construction."""
+    service = app_module.get_search_service()
+    monkeypatch.setattr(app_module, "pw_browser", "a-browser")
 
-async def test_ops_keyword_search_flattens_results(client, monkeypatch):
-    async def fake_ops_search(client_arg, q, n):
-        return [{"title": f"OPS {q}", "body": "", "href": "", "id": "1"}]
+    assert service.browser == "a-browser"
 
-    monkeypatch.setattr(app_module, "ops_search", fake_ops_search)
 
-    resp = await client.post("/ops/search", json={"queries": ["a", "b"], "n_results": 5})
+def test_the_default_patent_service_gets_the_apps_collaborators():
+    service = app_module.get_patent_service()
 
-    data = resp.json()
-    assert data["error"] is None
-    assert {r["title"] for r in data["results"]} == {"OPS a", "OPS b"}
+    assert service._http is app_module.httpx_client
+    assert service._ops_tokens is app_module.ops_token_manager
 
 
 # ------------------------------------- api_lifespan -------------------------------------
 
 
+class _RecordingClient:
+    def __init__(self):
+        self.closed = False
+
+    async def aclose(self):
+        self.closed = True
+
+
 async def test_api_lifespan_launches_chromium_without_a_sandbox(monkeypatch):
     """The container runs as a non-root user (see the Dockerfile), and
     Chromium's own internal sandbox needs privileges a non-root container
@@ -243,6 +232,10 @@ async def test_api_lifespan_launches_chromium_without_a_sandbox(monkeypatch):
     monkeypatch.setattr(app_module, "pw_browser", None)
     monkeypatch.setattr(app_module, "playwright", None)
     monkeypatch.setattr(app_module, "async_playwright", lambda: FakePlaywrightContextManager())
+    # The lifespan also closes the shared httpx client on shutdown; swap in
+    # a stand-in so this test doesn't close the real one for the rest of
+    # the session.
+    monkeypatch.setattr(app_module, "httpx_client", _RecordingClient())
 
     async with app_module.api_lifespan(app_module.app):
         pass
@@ -251,94 +244,35 @@ async def test_api_lifespan_launches_chromium_without_a_sandbox(monkeypatch):
     assert "--no-sandbox" in launch_calls[0].get("args", [])
 
 
-# --------------------------------- _search_one fallback ---------------------------------
-
-
-async def test_search_one_returns_first_backend_that_succeeds(monkeypatch):
-    calls = []
-
-    async def fake_ddg(q, n):
-        calls.append("ddg")
-        return [{"title": "from ddg"}]
-
-    async def fake_brave(browser, q, n):
-        calls.append("brave")
-        return [{"title": "from brave"}]
-
-    monkeypatch.setattr(app_module, "query_ddg_search", fake_ddg)
-    monkeypatch.setattr(app_module, "query_brave_search", fake_brave)
-
-    q, results, err = await app_module._search_one("widget", 5)
-
-    assert calls == ["ddg"]
-    assert results == [{"title": "from ddg"}]
-    assert err is None
-
-
-async def test_search_one_falls_through_to_the_next_backend_on_failure(monkeypatch):
-    calls = []
-
-    async def failing_ddg(q, n):
-        calls.append("ddg")
-        raise RuntimeError("ddg blocked")
-
-    async def fake_brave(browser, q, n):
-        calls.append("brave")
-        return [{"title": "from brave"}]
-
-    monkeypatch.setattr(app_module, "query_ddg_search", failing_ddg)
-    monkeypatch.setattr(app_module, "query_brave_search", fake_brave)
-
-    q, results, err = await app_module._search_one("widget", 5)
-
-    assert calls == ["ddg", "brave"]
-    assert results == [{"title": "from brave"}]
-    assert err is None
-
-
-async def test_search_one_reports_an_error_when_every_backend_fails(monkeypatch):
-    async def failing(*args):
-        raise RuntimeError("blocked")
-
-    monkeypatch.setattr(app_module, "query_ddg_search", failing)
-    monkeypatch.setattr(app_module, "query_brave_search", failing)
-    monkeypatch.setattr(app_module, "query_bing_search", failing)
-
-    q, results, err = await app_module._search_one("widget", 5)
-
-    assert results == []
-    assert "All backends failed for query 'widget'" in err
-    assert "blocked" in err
-
-
-async def test_search_one_skips_a_backend_whose_circuit_is_open(monkeypatch):
-    """A backend that's failed `failure_threshold` times in a row shouldn't
-    be attempted again until its cooldown elapses - repeatedly hammering a
-    backend that's already rate-limiting us only makes that worse.
+async def test_api_lifespan_closes_the_http_client_on_shutdown(monkeypatch):
+    """The client was created at import and never closed - the browser was
+    torn down on shutdown but its connection pool was not.
     """
-    from circuit_breaker import CircuitBreaker
-
-    fresh_breaker = CircuitBreaker(failure_threshold=1, cooldown_seconds=60)
-    monkeypatch.setattr(app_module, "_backend_circuit_breaker", fresh_breaker)
-
-    ddg_calls = []
+    recording_client = _RecordingClient()
+    monkeypatch.setattr(app_module, "pw_browser", None)
+    monkeypatch.setattr(app_module, "playwright", None)
+    monkeypatch.setattr(app_module, "httpx_client", recording_client)
+    # Playwright failing to start is the documented degraded mode; the
+    # client must still be closed cleanly on the way out.
+    monkeypatch.setattr(
+        app_module, "async_playwright", lambda: (_ for _ in ()).throw(RuntimeError("no browser")))
 
-    async def failing_ddg(q, n):
-        ddg_calls.append(q)
-        raise RuntimeError("ddg blocked")
+    async with app_module.api_lifespan(app_module.app):
+        pass
 
-    async def fake_brave(browser, q, n):
-        return [{"title": "from brave"}]
+    assert recording_client.closed is True
 
-    monkeypatch.setattr(app_module, "query_ddg_search", failing_ddg)
-    monkeypatch.setattr(app_module, "query_brave_search", fake_brave)
 
-    # First call: DDG fails, opening its circuit (threshold=1); Brave picks it up.
-    await app_module._search_one("first query", 5)
-    assert ddg_calls == ["first query"]
+async def test_api_lifespan_survives_playwright_failing_to_start(monkeypatch):
+    """Documented behaviour: the app still comes up, pw_browser stays None,
+    and the browser-backed endpoints fail individually rather than the whole
+    deployment refusing to boot.
+    """
+    monkeypatch.setattr(app_module, "pw_browser", None)
+    monkeypatch.setattr(app_module, "playwright", None)
+    monkeypatch.setattr(app_module, "httpx_client", _RecordingClient())
+    monkeypatch.setattr(
+        app_module, "async_playwright", lambda: (_ for _ in ()).throw(RuntimeError("no browser")))
 
-    # Second call: DDG's circuit is now open, so it should be skipped
-    # entirely (not called again) and go straight to Brave.
-    q, results, err = await app_module._search_one("second query", 5)
-    assert ddg_calls == ["first query"]  # unchanged - DDG was skipped
-    assert results == [{"title": "from brave"}]
+    async with app_module.api_lifespan(app_module.app):
+        assert app_module.pw_browser is None

tests/test_auth.py

@@ -0,0 +1,180 @@
+"""Tests for the optional API-key middleware.
+
+This is the only security control in the codebase and had no tests at all:
+mutation testing showed that replacing `secrets.compare_digest` with `==`
+(a timing-oracle regression) or disabling the guard's condition outright
+both left the full suite green. These pin it.
+
+`SERPENT_API_KEY` is read into a module global that the middleware resolves
+at call time, so monkeypatching the module attribute is enough - no import
+juggling or module reloading needed.
+"""
+
+import httpx
+import pytest
+from httpx import ASGITransport
+
+import app as app_module
+import services as services_module
+
+
+@pytest.fixture
+async def client():
+    transport = ASGITransport(app=app_module.app)
+    async with httpx.AsyncClient(transport=transport, base_url="http://test") as c:
+        yield c
+
+
+@pytest.fixture
+def api_key(monkeypatch):
+    """Configure a known key for the duration of one test."""
+    monkeypatch.setattr(app_module, "SERPENT_API_KEY", "s3cret-key")
+    return "s3cret-key"
+
+
+@pytest.fixture(autouse=True)
+def _unset_by_default(monkeypatch):
+    """Every test starts from the shipped default (no key configured), so a
+    stray SERPENT_API_KEY in the developer's environment can't change the
+    outcome of the tests below.
+    """
+    monkeypatch.setattr(app_module, "SERPENT_API_KEY", None)
+
+
+# ------------------------------- disabled by default -------------------------------
+
+
+async def test_no_key_configured_leaves_the_api_open(client, monkeypatch):
+    """The middleware is a no-op unless SERPENT_API_KEY is set - this keeps
+    the default deployment exactly as open as it was before it existed.
+    """
+    monkeypatch.setattr(services_module, "query_arxiv", _stub_arxiv)
+
+    resp = await client.post("/serp/search_arxiv", json={"queries": ["a"]})
+
+    assert resp.status_code == 200
+
+
+async def _stub_arxiv(client_arg, q, n):
+    return [{"title": q, "href": "h", "body": "b", "id": "1"}]
+
+
+# ------------------------------- gating when enabled -------------------------------
+
+
+async def test_missing_key_is_rejected(client, api_key):
+    resp = await client.post("/serp/search_arxiv", json={"queries": ["a"]})
+
+    assert resp.status_code == 401
+    assert "API key" in resp.json()["detail"]
+
+
+async def test_wrong_key_is_rejected(client, api_key):
+    resp = await client.post(
+        "/serp/search_arxiv", json={"queries": ["a"]}, headers={"X-API-Key": "wrong"})
+
+    assert resp.status_code == 401
+
+
+async def test_correct_key_via_x_api_key_header_is_accepted(client, api_key, monkeypatch):
+    monkeypatch.setattr(services_module, "query_arxiv", _stub_arxiv)
+
+    resp = await client.post(
+        "/serp/search_arxiv", json={"queries": ["a"]}, headers={"X-API-Key": api_key})
+
+    assert resp.status_code == 200
+
+
+async def test_correct_key_via_bearer_header_is_accepted(client, api_key, monkeypatch):
+    monkeypatch.setattr(services_module, "query_arxiv", _stub_arxiv)
+
+    resp = await client.post(
+        "/serp/search_arxiv", json={"queries": ["a"]},
+        headers={"Authorization": f"Bearer {api_key}"})
+
+    assert resp.status_code == 200
+
+
+async def test_bearer_scheme_match_is_case_insensitive(client, api_key, monkeypatch):
+    monkeypatch.setattr(services_module, "query_arxiv", _stub_arxiv)
+
+    resp = await client.post(
+        "/serp/search_arxiv", json={"queries": ["a"]},
+        headers={"Authorization": f"bearer {api_key}"})
+
+    assert resp.status_code == 200
+
+
+# --------------------------------- public paths ---------------------------------
+
+
+@pytest.mark.parametrize("path", ["/", "/openapi.json"])
+async def test_docs_paths_stay_public(client, api_key, path):
+    """The docs UI and the schema it fetches must stay reachable, otherwise
+    a key-protected deployment serves an unusable landing page.
+    """
+    resp = await client.get(path)
+
+    assert resp.status_code == 200
+
+
+async def test_a_key_that_is_a_prefix_of_the_real_one_is_rejected(client, api_key):
+    """Guards the length check inside the comparison."""
+    resp = await client.post(
+        "/serp/search_arxiv", json={"queries": ["a"]}, headers={"X-API-Key": "s3cret"})
+
+    assert resp.status_code == 401
+
+
+# ------------------------------ constant-time comparison ------------------------------
+
+
+async def test_key_comparison_is_constant_time(client, api_key, monkeypatch):
+    """`==` and `secrets.compare_digest` are functionally identical - they
+    differ only in timing, so no black-box assertion can tell them apart.
+    (Confirmed by mutation: swapping in `==` leaves every other test in this
+    file green.) A timing oracle on the API key is worth pinning anyway, so
+    assert structurally that the comparison goes through compare_digest.
+    """
+    import secrets as secrets_module
+
+    calls = []
+    real = secrets_module.compare_digest
+
+    def spy(a, b):
+        calls.append((a, b))
+        return real(a, b)
+
+    monkeypatch.setattr(secrets_module, "compare_digest", spy)
+
+    await client.post(
+        "/serp/search_arxiv", json={"queries": ["a"]}, headers={"X-API-Key": "wrong"})
+
+    assert calls, "the API key must be compared with secrets.compare_digest"
+
+
+async def test_non_ascii_key_is_rejected_not_crashed(client, api_key):
+    """`secrets.compare_digest` raises TypeError ("comparing strings with
+    non-ASCII characters is not supported") on str arguments outside ASCII,
+    and Starlette decodes headers as latin-1 - so an accented byte in the
+    header reached it intact and turned the security control into an
+    unhandled 500 for any anonymous caller. It must fail closed with a 401
+    like any other wrong key.
+
+    The header is passed as raw latin-1 bytes because that is what arrives
+    off the wire; handing httpx a str with non-ASCII in it fails in the
+    client before the request is ever sent, which would test nothing.
+    """
+    resp = await client.post(
+        "/serp/search_arxiv", json={"queries": ["a"]},
+        headers={"X-API-Key": "café".encode("latin-1")})
+
+    assert resp.status_code == 401
+
+
+async def test_non_ascii_key_is_rejected_via_bearer_header_too(client, api_key):
+    resp = await client.post(
+        "/serp/search_arxiv", json={"queries": ["a"]},
+        headers={"Authorization": "Bearer café".encode("latin-1")})
+
+    assert resp.status_code == 401

tests/test_input_bounds.py

@@ -0,0 +1,149 @@
+"""Bounds on request-shaped input.
+
+`n_results` was defended by a validator and a parametrized test while the
+list fields on the same models were undefended entirely: an empty query
+list crashed every search endpoint with an IndexError, and the bulk
+endpoints would fan out one concurrent outbound request per id with no
+upper limit at all.
+"""
+
+import httpx
+import pytest
+from httpx import ASGITransport
+from pydantic import ValidationError
+
+import app as app_module
+import scrap as scrap_module
+from app import MAX_BULK_PATENT_IDS, ScrapPatentsRequest
+from serp import MAX_QUERIES_PER_REQUEST, SerpQuery
+from services import shape_serp_results
+
+
+@pytest.fixture
+async def client():
+    transport = ASGITransport(app=app_module.app)
+    async with httpx.AsyncClient(transport=transport, base_url="http://test") as c:
+        yield c
+
+
+# --------------------------------- query list bounds ---------------------------------
+
+
+def test_empty_query_list_is_rejected_by_the_model():
+    with pytest.raises(ValidationError):
+        SerpQuery(queries=[])
+
+
+def test_too_many_queries_are_rejected_by_the_model():
+    with pytest.raises(ValidationError):
+        SerpQuery(queries=["q"] * (MAX_QUERIES_PER_REQUEST + 1))
+
+
+async def test_empty_query_list_returns_422_not_500(client):
+    """`_shape_serp_results` read `results[-1]` to build its error message,
+    so an empty list raised IndexError - an unhandled 500 on all six search
+    endpoints, reachable with a two-character request body.
+    """
+    resp = await client.post("/serp/search_arxiv", json={"queries": []})
+
+    assert resp.status_code == 422
+
+
+async def test_oversized_query_list_returns_422(client):
+    resp = await client.post(
+        "/serp/search_arxiv", json={"queries": ["q"] * (MAX_QUERIES_PER_REQUEST + 1)})
+
+    assert resp.status_code == 422
+
+
+def test_shape_serp_results_is_total_for_an_empty_list():
+    """Defence in depth: the helper must not depend on its caller having
+    validated the query list, since it is shared by every search path.
+    """
+    result = shape_serp_results([])
+
+    assert result.results == []
+    assert result.error is not None
+
+
+# ---------------------------------- bulk id bounds ----------------------------------
+
+
+def test_empty_patent_id_list_is_rejected():
+    with pytest.raises(ValidationError):
+        ScrapPatentsRequest(patent_ids=[])
+
+
+def test_too_many_patent_ids_are_rejected():
+    with pytest.raises(ValidationError):
+        ScrapPatentsRequest(patent_ids=["US1234567"] * (MAX_BULK_PATENT_IDS + 1))
+
+
+async def test_oversized_bulk_request_returns_422(client):
+    resp = await client.post(
+        "/scrap/scrap_patents_bulk",
+        json={"patent_ids": ["US1234567"] * (MAX_BULK_PATENT_IDS + 1)})
+
+    assert resp.status_code == 422
+
+
+# ------------------------------- bulk fan-out is bounded -------------------------------
+
+
+async def test_bulk_scrape_limits_concurrent_outbound_requests(monkeypatch):
+    """Playwright work is bounded by a semaphore; HTTP work was not bounded
+    at all, so a single accepted request could open one outbound scrape per
+    id. Track how many are in flight simultaneously.
+    """
+    import asyncio
+
+    in_flight = 0
+    peak = 0
+
+    async def slow_scrape(client_arg, url):
+        nonlocal in_flight, peak
+        in_flight += 1
+        peak = max(peak, in_flight)
+        await asyncio.sleep(0.01)
+        in_flight -= 1
+        raise httpx.ConnectTimeout("nope")
+
+    monkeypatch.setattr(scrap_module, "scrap_patent_async", slow_scrape)
+
+    ids = [f"US{1000000 + i}" for i in range(40)]
+    result = await scrap_module.scrap_patent_bulk_async(None, ids)
+
+    assert peak <= scrap_module.BULK_SCRAP_CONCURRENCY_LIMIT, (
+        f"{peak} concurrent scrapes for {len(ids)} ids; "
+        f"limit is {scrap_module.BULK_SCRAP_CONCURRENCY_LIMIT}")
+    assert result.failed_ids == ids
+
+
+async def test_bulk_ops_retrieval_limits_concurrent_requests(monkeypatch):
+    """The OPS bulk path is the expensive one - three requests per id
+    (biblio, claims, description) - so it carries a tighter limit.
+    """
+    import asyncio
+
+    import ops as ops_module
+
+    in_flight = 0
+    peak = 0
+
+    async def slow_ops_scrap(client_arg, number, *args, **kwargs):
+        nonlocal in_flight, peak
+        in_flight += 1
+        peak = max(peak, in_flight)
+        await asyncio.sleep(0.01)
+        in_flight -= 1
+        raise RuntimeError("nope")
+
+    monkeypatch.setattr(ops_module, "ops_scrap_patent", slow_ops_scrap)
+
+    numbers = [f"US{1000000 + i}" for i in range(30)]
+    result = await ops_module.ops_scrap_patent_bulk(None, numbers)
+
+    assert peak <= ops_module.BULK_OPS_CONCURRENCY_LIMIT, (
+        f"{peak} concurrent OPS retrievals; limit is "
+        f"{ops_module.BULK_OPS_CONCURRENCY_LIMIT}")
+    assert result.failed_ids == numbers

tests/test_mcp_server.py

@@ -0,0 +1,102 @@
+"""The MCP tool surface.
+
+mcp_server.py had no tests, and it degrades silently. `_tool_names` walks
+the route tree through three version-dependent shapes - DefaultPlaceholder
+unwrapping, nested `.routes`, and FastAPI >= 0.141's `original_router` -
+each a reasonable guess about a library's internals. If any stops
+resolving, tools quietly revert to names like
+`search_arxiv_serp_search_arxiv_post`, tool selection degrades, and
+nothing goes red.
+
+These pin the surface an agent actually sees, so a dependency bump reports
+the breakage instead of shipping it.
+"""
+
+import pytest
+
+import app as app_module
+from mcp_server import EXCLUDED_ROUTES, _tool_names
+
+# The curated set. /ops/* and the single-backend search endpoints are
+# deliberately excluded (see EXCLUDED_ROUTES): they stay available over
+# REST, but exposing them as tools invites a model to pick a path that
+# skips the fallback chains.
+EXPECTED_TOOLS = {
+    "search",
+    "search_arxiv",
+    "search_google_scholar",
+    "search_patents",
+    "scrap_patent",
+    "scrap_patents",
+}
+
+
+@pytest.fixture
+async def tools():
+    return {t.name: t for t in await app_module.mcp.list_tools()}
+
+
+async def test_the_exposed_tool_set_is_the_curated_one(tools):
+    assert set(tools) == EXPECTED_TOOLS
+
+
+async def test_tool_names_are_handler_names_not_generated_operation_ids(tools):
+    """FastAPI's generated operationIds (`search_arxiv_serp_search_arxiv_post`)
+    are what an LLM would otherwise see. _tool_names maps them back to the
+    Python handler name.
+    """
+    for name in tools:
+        assert "_post" not in name and "_get" not in name
+        assert "_serp_" not in name and "_scrap_" not in name
+
+
+async def test_operation_ids_all_resolve_to_a_handler_name():
+    """If the route walk stops finding routes, this mapping silently
+    empties and every tool falls back to its generated operationId."""
+    names = _tool_names(app_module.app)
+
+    assert len(names) >= len(EXPECTED_TOOLS)
+    assert "search_arxiv_serp_search_arxiv_post" in names
+    assert names["search_arxiv_serp_search_arxiv_post"] == "search_arxiv"
+
+
+@pytest.mark.parametrize("excluded", [
+    "ops_keyword_search", "ops_get_patent", "ops_get_patents_bulk",
+    "search_brave", "search_bing", "search_duck",
+])
+async def test_excluded_endpoints_are_not_exposed_as_tools(tools, excluded):
+    assert excluded not in tools
+
+
+def test_excluded_routes_patterns_are_anchored():
+    """An unanchored pattern would silently exclude more than intended."""
+    for pattern in EXCLUDED_ROUTES:
+        assert pattern.startswith("^")
+
+
+# ---------------------------------- output schemas ----------------------------------
+
+
+@pytest.mark.parametrize("tool_name", sorted(EXPECTED_TOOLS))
+async def test_every_tool_declares_its_output_schema(tools, tool_name):
+    """FastMCP builds tool schemas from the app's OpenAPI document, so a
+    handler with no return annotation advertises an untyped result to the
+    model. `search` - the tool the instructions push agents toward first -
+    was one of three that did.
+    """
+    assert tools[tool_name].output_schema, (
+        f"tool '{tool_name}' advertises no output schema; "
+        "the handler is probably missing a return annotation")
+
+
+@pytest.mark.parametrize("path", [
+    "/serp/search", "/serp/search_arxiv", "/serp/search_scholar",
+    "/serp/search_patents", "/scrap/scrap_patent/{patent_id}",
+    "/scrap/scrap_patents_bulk",
+])
+def test_every_exposed_route_declares_a_200_response_model(path):
+    schema = app_module.app.openapi()
+    for operation in schema["paths"][path].values():
+        content = (operation.get("responses", {}).get("200", {})
+                   .get("content", {}).get("application/json", {}).get("schema", {}))
+        assert content, f"{path} has no 200 response schema"

tests/test_patent_fallback.py

@@ -0,0 +1,232 @@
+"""The Google Patents -> EPO OPS fallback chain.
+
+The policy lives in services.PatentService, so most of this is plain unit
+testing with no ASGI client involved. Only the last section goes through
+the app, and only to check that the domain errors the service raises map
+onto the right status codes.
+
+The distinction these tests exist to protect: a 404 means "this patent does
+not exist", and an agent is explicitly told by MCP_INSTRUCTIONS to believe
+it and move on without retrying. Anything else - an upstream outage, a
+timeout, a blocked scrape - must not be reported that way.
+"""
+
+import httpx
+import pytest
+from httpx import ASGITransport, HTTPStatusError, Request, Response
+
+import app as app_module
+import services as services_module
+from scrap import PatentScrapResult
+from services import (OPSUnconfigured, PatentNotFound, PatentService,
+                      UpstreamUnavailable)
+
+PATENT_ID = "US11930446B2"
+
+
+class FakeOPSTokens:
+    """Stand-in for the OPS token manager. Tests only care whether
+    credentials are configured, and this avoids assigning to private
+    attributes on the real module-level singleton.
+    """
+
+    def __init__(self, configured: bool):
+        self.configured = configured
+
+
+def make_service(*, ops_configured: bool) -> PatentService:
+    return PatentService(http_client=None, ops_tokens=FakeOPSTokens(ops_configured))
+
+
+def _http_error(status: int) -> HTTPStatusError:
+    request = Request("GET", "https://patents.google.com/")
+    return HTTPStatusError(
+        str(status), request=request, response=Response(status, request=request))
+
+
+def _raises(exc):
+    async def _fn(*args, **kwargs):
+        raise exc
+    return _fn
+
+
+def _returns(value):
+    async def _fn(*args, **kwargs):
+        return value
+    return _fn
+
+
+# ------------------------------- the happy path -------------------------------
+
+
+async def test_successful_scrape_never_touches_ops(monkeypatch):
+    ops_called = False
+
+    async def fake_ops(*args, **kwargs):
+        nonlocal ops_called
+        ops_called = True
+        return PatentScrapResult(title="from OPS")
+
+    monkeypatch.setattr(
+        services_module, "scrap_patent_async",
+        _returns(PatentScrapResult(title="from Google Patents")))
+    monkeypatch.setattr(services_module, "ops_scrap_patent", fake_ops)
+
+    result = await make_service(ops_configured=True).scrap(PATENT_ID)
+
+    assert result.title == "from Google Patents"
+    assert ops_called is False
+
+
+# ------------------------- genuine miss vs upstream failure -------------------------
+
+
+async def test_a_real_404_from_google_patents_is_not_found(monkeypatch):
+    """The one case where "not found" is the truth."""
+    monkeypatch.setattr(services_module, "scrap_patent_async", _raises(_http_error(404)))
+
+    with pytest.raises(PatentNotFound):
+        await make_service(ops_configured=False).scrap(PATENT_ID)
+
+
+@pytest.mark.parametrize("status", [429, 500, 502, 503])
+async def test_an_upstream_failure_is_not_reported_as_not_found(monkeypatch, status):
+    """MCP_INSTRUCTIONS tells the model a "not found" patent is genuinely
+    absent everywhere and to move on rather than retrying. Reporting a
+    transient upstream failure that way teaches an agent - with the
+    server's explicit encouragement - that a real patent does not exist.
+    """
+    monkeypatch.setattr(services_module, "scrap_patent_async", _raises(_http_error(status)))
+
+    with pytest.raises(UpstreamUnavailable) as exc_info:
+        await make_service(ops_configured=False).scrap(PATENT_ID)
+
+    assert exc_info.value.timeout is False
+
+
+async def test_a_timeout_is_flagged_as_a_timeout(monkeypatch):
+    monkeypatch.setattr(
+        services_module, "scrap_patent_async", _raises(httpx.ConnectTimeout("timed out")))
+
+    with pytest.raises(UpstreamUnavailable) as exc_info:
+        await make_service(ops_configured=False).scrap(PATENT_ID)
+
+    assert exc_info.value.timeout is True
+
+
+async def test_an_unparseable_page_is_an_upstream_failure(monkeypatch):
+    """parse_patent_html raises ValueError when the page isn't a patent page
+    (an interstitial, or a markup change). That is our problem or theirs,
+    but it is not evidence the patent doesn't exist.
+    """
+    monkeypatch.setattr(
+        services_module, "scrap_patent_async", _raises(ValueError("no DC.title")))
+
+    with pytest.raises(UpstreamUnavailable):
+        await make_service(ops_configured=False).scrap(PATENT_ID)
+
+
+# --------------------------------- the OPS fallback ---------------------------------
+
+
+async def test_falls_back_to_ops_when_google_patents_fails(monkeypatch):
+    monkeypatch.setattr(services_module, "scrap_patent_async", _raises(_http_error(503)))
+    monkeypatch.setattr(
+        services_module, "ops_scrap_patent", _returns(PatentScrapResult(title="from OPS")))
+
+    result = await make_service(ops_configured=True).scrap(PATENT_ID)
+
+    assert result.title == "from OPS"
+
+
+async def test_missing_from_both_backends_is_not_found(monkeypatch):
+    monkeypatch.setattr(services_module, "scrap_patent_async", _raises(_http_error(404)))
+    monkeypatch.setattr(services_module, "ops_scrap_patent", _raises(_http_error(404)))
+
+    with pytest.raises(PatentNotFound):
+        await make_service(ops_configured=True).scrap(PATENT_ID)
+
+
+async def test_ops_failing_after_a_google_patents_404_is_not_a_miss(monkeypatch):
+    """Google Patents says the patent is missing, but OPS - the backend that
+    covers what Google Patents doesn't - never answered. That is unknown,
+    not absent.
+    """
+    monkeypatch.setattr(services_module, "scrap_patent_async", _raises(_http_error(404)))
+    monkeypatch.setattr(services_module, "ops_scrap_patent", _raises(_http_error(500)))
+
+    with pytest.raises(UpstreamUnavailable):
+        await make_service(ops_configured=True).scrap(PATENT_ID)
+
+
+# ------------------------------ the OPS-only path ------------------------------
+
+
+async def test_ops_only_retrieval_requires_credentials():
+    with pytest.raises(OPSUnconfigured):
+        await make_service(ops_configured=False).ops_scrap(PATENT_ID)
+
+
+async def test_ops_only_retrieval_reports_a_real_miss(monkeypatch):
+    """Here OPS is the only backend asked, so its 404 is the whole answer."""
+    monkeypatch.setattr(services_module, "ops_scrap_patent", _raises(_http_error(404)))
+
+    with pytest.raises(PatentNotFound):
+        await make_service(ops_configured=True).ops_scrap(PATENT_ID)
+
+
+async def test_ops_only_retrieval_flags_a_timeout(monkeypatch):
+    monkeypatch.setattr(
+        services_module, "ops_scrap_patent", _raises(httpx.ConnectTimeout("timed out")))
+
+    with pytest.raises(UpstreamUnavailable) as exc_info:
+        await make_service(ops_configured=True).ops_scrap(PATENT_ID)
+
+    assert exc_info.value.timeout is True
+
+
+# ---------------------------- domain error -> status code ----------------------------
+
+
+@pytest.fixture
+async def client():
+    transport = ASGITransport(app=app_module.app)
+    async with httpx.AsyncClient(transport=transport, base_url="http://test") as c:
+        yield c
+
+
+@pytest.fixture
+def override_patent_service():
+    """Swap the app's patent service for one the test controls, then put the
+    real provider back."""
+    def _install(service):
+        app_module.app.dependency_overrides[app_module.get_patent_service] = lambda: service
+
+    yield _install
+    app_module.app.dependency_overrides.clear()
+
+
+class _StubService:
+    def __init__(self, error):
+        self._error = error
+
+    async def scrap(self, patent_id):
+        raise self._error
+
+    ops_scrap = scrap
+
+
+@pytest.mark.parametrize("error, expected_status", [
+    (PatentNotFound("gone"), 404),
+    (UpstreamUnavailable("upstream broke"), 502),
+    (UpstreamUnavailable("slow", timeout=True), 504),
+    (OPSUnconfigured("no credentials"), 503),
+])
+async def test_domain_errors_map_onto_status_codes(
+        client, override_patent_service, error, expected_status):
+    override_patent_service(_StubService(error))
+
+    resp = await client.get(f"/scrap/scrap_patent/{PATENT_ID}")
+
+    assert resp.status_code == expected_status
+    assert resp.json()["detail"] == str(error)

tests/test_scrap.py

@@ -89,3 +89,15 @@ async def test_scrap_patent_bulk_async_separates_successes_from_failures():
     assert len(result.patents) == 1
     assert result.patents[0].title == "Widget with improved gadget mechanism"
     assert result.failed_ids == ["US_FAIL"]
+
+
+def test_parse_patent_html_strips_whitespace_around_the_title():
+    """Google Patents' DC.title meta content is frequently padded with
+    newlines and indentation from the surrounding markup, which would
+    otherwise end up in the API response and in every downstream citation.
+    """
+    html = '<html><head><meta name="DC.title" content="  Widget apparatus\n  "></head><body></body></html>'
+
+    result = parse_patent_html(html, "https://patents.google.com/patent/US1/en")
+
+    assert result.title == "Widget apparatus"

tests/test_search_service.py

@@ -0,0 +1,390 @@
+"""SearchService: the backend orchestration policy.
+
+Which backend to try, in what order, when to fall back, and what counts as
+evidence a backend is unhealthy. None of this needs an HTTP layer, so none
+of these tests use one.
+
+Each test builds a service with its own circuit breaker. That replaces the
+autouse fixture that used to reset a process-wide singleton between tests -
+the breaker being shared across every request is correct in production and
+was only a problem because tests had no way to opt out of it.
+"""
+
+
+import services as services_module
+from circuit_breaker import CircuitBreaker
+from serp import DuckDuckGoBlockedException, SerpQuery
+from services import SearchService, shape_serp_results
+
+
+class FakeOPSTokens:
+    def __init__(self, configured: bool):
+        self.configured = configured
+
+
+def make_service(*, ops_configured: bool = False, breaker: CircuitBreaker = None,
+                 browser=None) -> SearchService:
+    return SearchService(
+        http_client=None,
+        browser_provider=lambda: browser,
+        circuit_breaker=breaker or CircuitBreaker(),
+        ops_tokens=FakeOPSTokens(ops_configured),
+    )
+
+
+# --------------------------------- result shaping ---------------------------------
+
+
+class TestShapeSerpResults:
+    def test_flattens_successful_lists(self):
+        result = shape_serp_results([[{"a": 1}], [{"b": 2}, {"c": 3}]])
+        assert result.results == [{"a": 1}, {"b": 2}, {"c": 3}]
+        assert result.error is None
+
+    def test_surfaces_last_error_when_all_failed(self):
+        result = shape_serp_results([ValueError("first"), RuntimeError("second")])
+        assert result.results == []
+        assert "second" in result.error
+
+    def test_ignores_failures_when_some_succeed(self):
+        result = shape_serp_results([[{"a": 1}], RuntimeError("boom")])
+        assert result.results == [{"a": 1}]
+        assert result.error is None
+
+    def test_is_total_for_an_empty_list(self):
+        """Defence in depth: the request models bound the query list, but
+        this helper is shared by every search path and must not depend on
+        its caller having validated anything."""
+        result = shape_serp_results([])
+        assert result.results == []
+        assert result.error is not None
+
+
+# ------------------------------- per-query dispatch -------------------------------
+
+
+async def test_every_query_is_dispatched_with_the_result_count():
+    calls = []
+
+    async def fn(q, n):
+        calls.append((q, n))
+        return [{"q": q}]
+
+    result = await make_service()._run_queries(
+        fn, SerpQuery(queries=["x", "y"], n_results=5), "test context")
+
+    assert sorted(calls) == [("x", 5), ("y", 5)]
+    assert sorted(result.results, key=str) == sorted([{"q": "x"}, {"q": "y"}], key=str)
+
+
+async def test_arxiv_flattens_results_from_all_queries(monkeypatch):
+    async def fake_query_arxiv(client, q, n):
+        return [{"title": f"paper about {q}"}]
+
+    monkeypatch.setattr(services_module, "query_arxiv", fake_query_arxiv)
+
+    result = await make_service().arxiv(SerpQuery(queries=["a", "b"], n_results=5))
+
+    assert {r["title"] for r in result.results} == {"paper about a", "paper about b"}
+    assert result.error is None
+
+
+async def test_arxiv_reports_an_error_when_every_query_fails(monkeypatch):
+    async def failing(client, q, n):
+        raise RuntimeError(f"boom for {q}")
+
+    monkeypatch.setattr(services_module, "query_arxiv", failing)
+
+    result = await make_service().arxiv(SerpQuery(queries=["a"], n_results=5))
+
+    assert result.results == []
+    assert "boom for a" in result.error
+
+
+# ------------------------------ patents + OPS fallback ------------------------------
+
+
+async def test_patents_does_not_fall_back_when_google_patents_succeeds(monkeypatch):
+    ops_called = False
+
+    async def fake_ops(client, q, n):
+        nonlocal ops_called
+        ops_called = True
+        return []
+
+    monkeypatch.setattr(
+        services_module, "query_google_patents",
+        lambda b, q, n: _ok([{"id": "US1", "title": "t"}]))
+    monkeypatch.setattr(services_module, "ops_search", fake_ops)
+
+    result = await make_service(ops_configured=True).patents(
+        SerpQuery(queries=["widget"], n_results=5))
+
+    assert result.results == [{"id": "US1", "title": "t"}]
+    assert ops_called is False
+
+
+async def _ok(value):
+    return value
+
+
+async def test_patents_falls_back_to_ops_when_google_patents_is_empty(monkeypatch):
+    monkeypatch.setattr(services_module, "query_google_patents", lambda b, q, n: _ok([]))
+    monkeypatch.setattr(
+        services_module, "ops_search", lambda c, q, n: _ok([{"id": "US123", "title": "OPS"}]))
+
+    result = await make_service(ops_configured=True).patents(
+        SerpQuery(queries=["widget"], n_results=5))
+
+    assert result.results == [{"id": "US123", "title": "OPS"}]
+
+
+async def test_patents_does_not_fall_back_when_ops_is_unconfigured(monkeypatch):
+    monkeypatch.setattr(services_module, "query_google_patents", lambda b, q, n: _ok([]))
+
+    result = await make_service(ops_configured=False).patents(
+        SerpQuery(queries=["widget"], n_results=5))
+
+    assert result.results == []
+    assert result.error is None
+
+
+async def test_ops_fallback_runs_concurrently_not_one_query_at_a_time(monkeypatch):
+    """This path exists for when Google Patents is unavailable, which is
+    exactly when a request can least afford N sequential round-trips to the
+    slower backend."""
+    import asyncio
+
+    in_flight = 0
+    peak = 0
+
+    async def slow_ops(client, q, n):
+        nonlocal in_flight, peak
+        in_flight += 1
+        peak = max(peak, in_flight)
+        await asyncio.sleep(0.02)
+        in_flight -= 1
+        return [{"id": q}]
+
+    monkeypatch.setattr(services_module, "query_google_patents", lambda b, q, n: _ok([]))
+    monkeypatch.setattr(services_module, "ops_search", slow_ops)
+
+    await make_service(ops_configured=True).patents(
+        SerpQuery(queries=["a", "b", "c", "d"], n_results=5))
+
+    assert peak > 1, "OPS fallback queries were issued one at a time"
+
+
+async def test_a_failing_ops_fallback_does_not_lose_other_queries_results(monkeypatch):
+    async def selective_ops(client, q, n):
+        if q == "bad":
+            raise RuntimeError("OPS down")
+        return [{"id": q}]
+
+    monkeypatch.setattr(services_module, "query_google_patents", lambda b, q, n: _ok([]))
+    monkeypatch.setattr(services_module, "ops_search", selective_ops)
+
+    result = await make_service(ops_configured=True).patents(
+        SerpQuery(queries=["good", "bad"], n_results=5))
+
+    assert result.results == [{"id": "good"}]
+
+
+# -------------------------------- the fallback chain --------------------------------
+
+
+async def test_returns_the_first_backend_that_succeeds(monkeypatch):
+    calls = []
+
+    async def fake_ddg(q, n):
+        calls.append("ddg")
+        return [{"title": "from ddg"}]
+
+    async def fake_brave(browser, q, n):
+        calls.append("brave")
+        return [{"title": "from brave"}]
+
+    monkeypatch.setattr(services_module, "query_ddg_search", fake_ddg)
+    monkeypatch.setattr(services_module, "query_brave_search", fake_brave)
+
+    q, results, err = await make_service()._search_one("widget", 5)
+
+    assert calls == ["ddg"]
+    assert results == [{"title": "from ddg"}]
+    assert err is None
+
+
+async def test_falls_through_to_the_next_backend_on_failure(monkeypatch):
+    calls = []
+
+    async def failing_ddg(q, n):
+        calls.append("ddg")
+        raise RuntimeError("ddg blocked")
+
+    async def fake_brave(browser, q, n):
+        calls.append("brave")
+        return [{"title": "from brave"}]
+
+    monkeypatch.setattr(services_module, "query_ddg_search", failing_ddg)
+    monkeypatch.setattr(services_module, "query_brave_search", fake_brave)
+
+    q, results, err = await make_service()._search_one("widget", 5)
+
+    assert calls == ["ddg", "brave"]
+    assert results == [{"title": "from brave"}]
+    assert err is None
+
+
+async def test_reports_an_error_when_every_backend_fails(monkeypatch):
+    async def failing(*args):
+        raise RuntimeError("blocked")
+
+    for name in ("query_ddg_search", "query_brave_search", "query_bing_search"):
+        monkeypatch.setattr(services_module, name, failing)
+
+    q, results, err = await make_service()._search_one("widget", 5)
+
+    assert results == []
+    assert "All backends failed for query 'widget'" in err
+    assert "blocked" in err
+
+
+# ------------------------------ the circuit breaker ------------------------------
+
+
+async def test_skips_a_backend_whose_circuit_is_open(monkeypatch):
+    """A backend that's failed `failure_threshold` times in a row shouldn't
+    be attempted again until its cooldown elapses - repeatedly hammering a
+    backend that's already rate-limiting us only makes that worse.
+    """
+    ddg_calls = []
+
+    async def failing_ddg(q, n):
+        ddg_calls.append(q)
+        raise RuntimeError("ddg blocked")
+
+    async def fake_brave(browser, q, n):
+        return [{"title": "from brave"}]
+
+    monkeypatch.setattr(services_module, "query_ddg_search", failing_ddg)
+    monkeypatch.setattr(services_module, "query_brave_search", fake_brave)
+
+    service = make_service(breaker=CircuitBreaker(failure_threshold=1, cooldown_seconds=60))
+
+    await service._search_one("first query", 5)
+    assert ddg_calls == ["first query"]
+
+    q, results, err = await service._search_one("second query", 5)
+
+    assert ddg_calls == ["first query"]  # unchanged - DDG was skipped
+    assert results == [{"title": "from brave"}]
+
+
+async def test_a_zero_result_query_does_not_open_the_circuit(monkeypatch):
+    """An empty result is ambiguous - it can mean the query genuinely has no
+    hits, or that the backend served an anti-bot page that parsed to
+    nothing. The breaker is shared across every request, so counting the
+    ambiguous case meant a few unusual-but-legitimate queries could disable
+    the primary backend for everyone for a full cooldown.
+    """
+    ddg_calls = []
+
+    async def empty_ddg(q, n):
+        ddg_calls.append(q)
+        raise DuckDuckGoBlockedException()
+
+    monkeypatch.setattr(services_module, "query_ddg_search", empty_ddg)
+    monkeypatch.setattr(
+        services_module, "query_brave_search", lambda b, q, n: _ok([{"title": "brave"}]))
+
+    service = make_service(breaker=CircuitBreaker(failure_threshold=1, cooldown_seconds=60))
+
+    await service._search_one("obscure", 5)
+    await service._search_one("another obscure", 5)
+
+    assert ddg_calls == ["obscure", "another obscure"]
+
+
+async def test_an_empty_result_still_falls_through_to_the_next_backend(monkeypatch):
+    """Not counting it against the backend must not stop the fallback: the
+    caller still needs results from somewhere."""
+    async def empty_ddg(q, n):
+        raise DuckDuckGoBlockedException()
+
+    monkeypatch.setattr(services_module, "query_ddg_search", empty_ddg)
+    monkeypatch.setattr(
+        services_module, "query_brave_search", lambda b, q, n: _ok([{"title": "from brave"}]))
+
+    _q, results, err = await make_service()._search_one("widget", 5)
+
+    assert results == [{"title": "from brave"}]
+    assert err is None
+
+
+async def test_a_real_backend_error_still_opens_the_circuit(monkeypatch):
+    """The distinction has to cut both ways: a backend raising a genuine
+    error (a rate limit, a transport failure) is still evidence."""
+    ddg_calls = []
+
+    async def failing_ddg(q, n):
+        ddg_calls.append(q)
+        raise RuntimeError("rate limited")
+
+    monkeypatch.setattr(services_module, "query_ddg_search", failing_ddg)
+    monkeypatch.setattr(
+        services_module, "query_brave_search", lambda b, q, n: _ok([{"title": "brave"}]))
+
+    service = make_service(breaker=CircuitBreaker(failure_threshold=1, cooldown_seconds=60))
+
+    await service._search_one("first", 5)
+    await service._search_one("second", 5)
+
+    assert ddg_calls == ["first"]
+
+
+# ------------------------------- search aggregation -------------------------------
+
+
+async def test_search_merges_results_from_every_query(monkeypatch):
+    monkeypatch.setattr(
+        services_module, "query_ddg_search", lambda q, n: _ok([{"title": f"result for {q}"}]))
+
+    result = await make_service().search(SerpQuery(queries=["a", "b"], n_results=5))
+
+    assert {r["title"] for r in result.results} == {"result for a", "result for b"}
+    assert result.error is None
+
+
+async def test_search_reports_partial_failure_alongside_results(monkeypatch):
+    """A query no backend could answer must be reported even when another
+    query succeeded - otherwise the caller silently receives fewer results
+    than they asked for with no indication why."""
+    async def selective_ddg(q, n):
+        if q == "bad":
+            raise RuntimeError("blocked")
+        return [{"title": f"result for {q}"}]
+
+    async def failing(*args):
+        raise RuntimeError("blocked")
+
+    monkeypatch.setattr(services_module, "query_ddg_search", selective_ddg)
+    monkeypatch.setattr(services_module, "query_brave_search", failing)
+    monkeypatch.setattr(services_module, "query_bing_search", failing)
+
+    result = await make_service().search(SerpQuery(queries=["good", "bad"], n_results=5))
+
+    assert [r["title"] for r in result.results] == ["result for good"]
+    assert "bad" in result.error
+
+
+async def test_search_names_every_failed_query_not_just_the_last(monkeypatch):
+    async def failing(*args):
+        raise RuntimeError("blocked")
+
+    for name in ("query_ddg_search", "query_brave_search", "query_bing_search"):
+        monkeypatch.setattr(services_module, name, failing)
+
+    result = await make_service().search(SerpQuery(queries=["a", "b"], n_results=5))
+
+    assert result.results == []
+    assert "'a'" in result.error and "'b'" in result.error

tests/test_serp_ddg_concurrency.py

@@ -0,0 +1,72 @@
+"""The DuckDuckGo backend must not block the event loop.
+
+`DDGS.text()` is a synchronous, blocking method. Calling it directly from
+an `async def` freezes the whole process for the duration of the HTTP
+request: no other request progresses, no in-flight Playwright scrape
+progresses, and nothing else on the loop gets a turn.
+
+That matters more here than it looks. DuckDuckGo is the *first* backend
+tried for every query in `/serp/search`, which is the flagship MCP tool,
+and both the endpoint docstrings and MCP_INSTRUCTIONS actively encourage
+clients to batch many queries into one call - which under a blocking
+implementation serializes into N sequential stalls instead of N concurrent
+requests.
+
+These tests use a blocking stand-in with a known duration rather than the
+real library, so they assert the concurrency property itself and stay fast
+and offline.
+"""
+
+import asyncio
+import time
+
+import serp
+from serp import query_ddg_search
+
+BLOCK_SECONDS = 0.25
+
+
+class _BlockingDDGS:
+    """Stands in for the real DDGS: a synchronous call that takes real time."""
+
+    def text(self, q, max_results=10):
+        time.sleep(BLOCK_SECONDS)
+        return [{"title": q, "body": "b", "href": "h"}]
+
+
+async def test_concurrent_queries_do_not_serialize(monkeypatch):
+    """Four gathered queries should take about as long as one, not four."""
+    monkeypatch.setattr(serp, "DDGS", lambda: _BlockingDDGS())
+
+    started = time.perf_counter()
+    await asyncio.gather(*[query_ddg_search(f"q{i}", 5) for i in range(4)])
+    elapsed = time.perf_counter() - started
+
+    assert elapsed < BLOCK_SECONDS * 2, (
+        f"4 concurrent queries took {elapsed:.2f}s; "
+        f"serialized would be ~{BLOCK_SECONDS * 4:.2f}s, concurrent ~{BLOCK_SECONDS:.2f}s")
+
+
+async def test_the_event_loop_stays_responsive_during_a_query(monkeypatch):
+    """Anything else scheduled on the loop must keep getting turns while a
+    DuckDuckGo query is in flight - this is what a blocked loop costs every
+    other request being served at the same time.
+    """
+    monkeypatch.setattr(serp, "DDGS", lambda: _BlockingDDGS())
+
+    ticks = 0
+    stop = asyncio.Event()
+
+    async def heartbeat():
+        nonlocal ticks
+        while not stop.is_set():
+            await asyncio.sleep(0.01)
+            ticks += 1
+
+    beat = asyncio.create_task(heartbeat())
+    await query_ddg_search("widget", 5)
+    stop.set()
+    await beat
+
+    # A responsive loop ticks ~25 times in 0.25s; a blocked one manages ~1.
+    assert ticks > 10, f"event loop only got {ticks} turns during a {BLOCK_SECONDS}s query"

tests/test_serp_navigation.py

@@ -0,0 +1,156 @@
+"""The glue holding each scraper together.
+
+Every Playwright scraper is now: build a URL, open a page, block
+decorative resources, navigate, extract. The builders are tested in
+test_serp_urls.py and the extractors in test_serp_playwright.py; this
+covers the composition - specifically that each scraper navigates to the
+URL its own builder produced.
+
+That link is the one the old goto-patching harness could not test, because
+the stand-in it installed accepted the navigation URL and discarded it.
+These use a recording stand-in browser instead: no real Chromium, and the
+URL is the thing being asserted on.
+"""
+
+import pytest
+
+import serp
+from serp import (bing_search_url, brave_search_url, google_patents_search_url,
+                  google_scholar_url, playwright_open_page,
+                  query_bing_search, query_brave_search, query_google_patents,
+                  query_google_scholar)
+
+SENTINEL = [{"title": "extracted"}]
+
+
+class _RecordingPage:
+    def __init__(self):
+        self.goto_urls = []
+        self.route_patterns = []
+        self.closed = False
+
+    async def route(self, pattern, handler):
+        self.route_patterns.append(pattern)
+
+    async def goto(self, url, **kwargs):
+        self.goto_urls.append(url)
+        return None
+
+    async def close(self):
+        self.closed = True
+
+
+class _RecordingContext:
+    def __init__(self, page):
+        self._page = page
+        self.closed = False
+
+    async def new_page(self):
+        return self._page
+
+    async def close(self):
+        self.closed = True
+
+
+class _RecordingBrowser:
+    def __init__(self):
+        self.page = _RecordingPage()
+        self.context = _RecordingContext(self.page)
+
+    async def new_context(self, **kwargs):
+        return self.context
+
+
+async def _sentinel_extractor(page, n_results):
+    return SENTINEL
+
+
+SCRAPERS = [
+    ("query_google_scholar", query_google_scholar,
+     "_extract_google_scholar_results", google_scholar_url),
+    ("query_google_patents", query_google_patents,
+     "_extract_google_patents_results", google_patents_search_url),
+    ("query_brave_search", query_brave_search,
+     "_extract_brave_results", brave_search_url),
+    ("query_bing_search", query_bing_search,
+     "_extract_bing_results", bing_search_url),
+]
+
+
+@pytest.mark.parametrize("name, scraper, extractor_attr, builder",
+                         SCRAPERS, ids=[s[0] for s in SCRAPERS])
+async def test_scraper_navigates_to_the_url_its_builder_produced(
+        monkeypatch, name, scraper, extractor_attr, builder):
+    monkeypatch.setattr(serp, extractor_attr, _sentinel_extractor)
+    browser = _RecordingBrowser()
+
+    await scraper(browser, "agentic ai", 25)
+
+    assert browser.page.goto_urls == [builder("agentic ai", 25)]
+
+
+@pytest.mark.parametrize("name, scraper, extractor_attr, builder",
+                         SCRAPERS, ids=[s[0] for s in SCRAPERS])
+async def test_scraper_returns_what_its_extractor_produced(
+        monkeypatch, name, scraper, extractor_attr, builder):
+    monkeypatch.setattr(serp, extractor_attr, _sentinel_extractor)
+
+    results = await scraper(_RecordingBrowser(), "widgets", 10)
+
+    assert results == SENTINEL
+
+
+@pytest.mark.parametrize("name, scraper, extractor_attr, builder",
+                         SCRAPERS, ids=[s[0] for s in SCRAPERS])
+async def test_scraper_blocks_decorative_resources(
+        monkeypatch, name, scraper, extractor_attr, builder):
+    """Skipping stylesheets and images meaningfully speeds up navigation,
+    and every scraper is supposed to do it."""
+    monkeypatch.setattr(serp, extractor_attr, _sentinel_extractor)
+    browser = _RecordingBrowser()
+
+    await scraper(browser, "widgets", 10)
+
+    assert browser.page.route_patterns == ["**/*"]
+
+
+@pytest.mark.parametrize("name, scraper, extractor_attr, builder",
+                         SCRAPERS, ids=[s[0] for s in SCRAPERS])
+async def test_scraper_closes_its_page_and_context(
+        monkeypatch, name, scraper, extractor_attr, builder):
+    monkeypatch.setattr(serp, extractor_attr, _sentinel_extractor)
+    browser = _RecordingBrowser()
+
+    await scraper(browser, "widgets", 10)
+
+    assert browser.page.closed and browser.context.closed
+
+
+async def test_page_and_context_are_closed_even_when_extraction_raises(monkeypatch):
+    async def exploding_extractor(page, n_results):
+        raise RuntimeError("selectors changed")
+
+    monkeypatch.setattr(serp, "_extract_bing_results", exploding_extractor)
+    browser = _RecordingBrowser()
+
+    with pytest.raises(RuntimeError):
+        await query_bing_search(browser, "widgets", 10)
+
+    assert browser.page.closed and browser.context.closed
+
+
+async def test_context_is_closed_even_when_closing_the_page_raises():
+    """A crashed renderer can make page.close() throw; the context still has
+    to be released or it leaks for the process's lifetime."""
+    browser = _RecordingBrowser()
+
+    async def exploding_close():
+        raise RuntimeError("renderer gone")
+
+    browser.page.close = exploding_close
+
+    with pytest.raises(RuntimeError):
+        async with playwright_open_page(browser):
+            pass
+
+    assert browser.context.closed

tests/test_serp_playwright.py

@@ -1,36 +1,36 @@
-"""Characterization tests for the Playwright-driven scrapers in serp.py.
-
-These run against a real headless Chromium (same engine production uses) so
-that DOM/CSS-selector behavior is genuinely exercised, but the target page is
-served locally from a fixture instead of navigating to the live site - see
-`helpers.make_fixture_browser` for how, and why a simpler
-`page.route()`-based approach doesn't work here.
+"""Tests for the DOM-extraction half of the Playwright-driven scrapers.
+
+Each scraper in serp.py is three separable pieces: a URL builder (pure,
+tested in test_serp_urls.py), a navigation step, and an `_extract_*`
+function that reads an already-loaded page. These test the third piece
+against a real headless Chromium - the same engine production uses, so
+selector behaviour is genuinely exercised - with the fixture HTML loaded
+via `set_content`.
+
+This used to require a wrapper that replaced `page.goto` on a
+purpose-built fake Browser, because fetch and parse weren't separated.
+That wrapper accepted the navigation URL and discarded it, which is
+precisely why URL bugs were invisible to the suite; splitting the scrapers
+made it unnecessary and it has been deleted.
 """
 
 import pytest
 
-from helpers import load_fixture, make_fixture_browser
+from helpers import load_fixture
 from serp import (
     BraveSearchBlockedException,
     BrowserUnavailableError,
+    _block_stylesheet_and_image_resources,
     _extract_bing_results,
+    _extract_brave_results,
+    _extract_google_patents_results,
+    _extract_google_scholar_results,
     playwright_open_page,
-    query_bing_search,
-    query_brave_search,
-    query_google_patents,
-    query_google_scholar,
 )
 
 
-async def test_extract_bing_results_works_on_an_already_loaded_page(browser):
-    """Demonstrates the payoff of splitting fetch (navigate) from parse
-    (extract): `_extract_bing_results` takes a Page that's already loaded,
-    so this needs nothing beyond `set_content` - no goto-patching wrapper,
-    no browser-as-a-parameter indirection.
-    """
-    context = await browser.new_context()
-    page = await context.new_page()
-    await page.set_content(load_fixture("bing_results.html"))
+async def test_bing_results_are_extracted(page_factory):
+    page = await page_factory(load_fixture("bing_results.html"))
 
     results = await _extract_bing_results(page, 10)
 
@@ -39,24 +39,19 @@ async def test_extract_bing_results_works_on_an_already_loaded_page(browser):
         {"title": "Result B", "href": "https://example.org/b", "body": "Snippet B"},
     ]
 
-    await context.close()
 
+async def test_bing_extraction_honours_the_result_limit(page_factory):
+    page = await page_factory(load_fixture("bing_results.html"))
 
-async def test_bing_search_parses_results(browser):
-    fixture_browser = make_fixture_browser(browser, load_fixture("bing_results.html"))
+    results = await _extract_bing_results(page, 1)
 
-    results = await query_bing_search(fixture_browser, "test query", 10)
-
-    assert results == [
-        {"title": "Result A", "href": "https://example.org/a", "body": "Snippet A"},
-        {"title": "Result B", "href": "https://example.org/b", "body": "Snippet B"},
-    ]
+    assert [r["title"] for r in results] == ["Result A"]
 
 
-async def test_brave_search_parses_results(browser):
-    fixture_browser = make_fixture_browser(browser, load_fixture("brave_results.html"))
+async def test_brave_results_are_extracted(page_factory):
+    page = await page_factory(load_fixture("brave_results.html"))
 
-    results = await query_brave_search(fixture_browser, "test query", 10)
+    results = await _extract_brave_results(page, 10)
 
     assert results == [
         {"title": "Brave Result A", "body": "Brave snippet A", "href": "https://example.org/brave-a"},
@@ -64,17 +59,30 @@ async def test_brave_search_parses_results(browser):
     ]
 
 
-async def test_brave_search_detects_the_suspicious_block_page(browser):
-    fixture_browser = make_fixture_browser(browser, load_fixture("brave_blocked.html"))
+async def test_brave_extraction_honours_the_result_limit(page_factory):
+    page = await page_factory(load_fixture("brave_results.html"))
+
+    results = await _extract_brave_results(page, 1)
+
+    assert [r["title"] for r in results] == ["Brave Result A"]
+
+
+async def test_brave_block_page_raises(page_factory):
+    """Brave's anti-bot interstitial has no result cards, so it would
+    otherwise look like a legitimate zero-result search - and the
+    `/serp/search` fallback chain needs an exception to move to the next
+    backend.
+    """
+    page = await page_factory(load_fixture("brave_blocked.html"))
 
     with pytest.raises(BraveSearchBlockedException):
-        await query_brave_search(fixture_browser, "test query", 10)
+        await _extract_brave_results(page, 10)
 
 
-async def test_google_scholar_parses_results(browser):
-    fixture_browser = make_fixture_browser(browser, load_fixture("scholar_results.html"))
+async def test_google_scholar_results_are_extracted(page_factory):
+    page = await page_factory(load_fixture("scholar_results.html"))
 
-    results = await query_google_scholar(fixture_browser, "test query", 10)
+    results = await _extract_google_scholar_results(page, 10)
 
     assert results == [
         {"title": "Scholar Paper A", "body": "Scholar snippet A", "href": "https://example.org/scholar-a"},
@@ -82,10 +90,18 @@ async def test_google_scholar_parses_results(browser):
     ]
 
 
-async def test_google_patents_parses_results(browser):
-    fixture_browser = make_fixture_browser(browser, load_fixture("patents", "search_results.html"))
+async def test_google_scholar_extraction_honours_the_result_limit(page_factory):
+    page = await page_factory(load_fixture("scholar_results.html"))
+
+    results = await _extract_google_scholar_results(page, 1)
+
+    assert [r["title"] for r in results] == ["Scholar Paper A"]
 
-    results = await query_google_patents(fixture_browser, "test query", 10)
+
+async def test_google_patents_results_are_extracted(page_factory):
+    page = await page_factory(load_fixture("patents", "search_results.html"))
+
+    results = await _extract_google_patents_results(page, 10)
 
     assert results == [
         {
@@ -103,7 +119,57 @@ async def test_google_patents_parses_results(browser):
     ]
 
 
+async def test_google_patents_extraction_honours_the_result_limit(page_factory):
+    page = await page_factory(load_fixture("patents", "search_results.html"))
+
+    results = await _extract_google_patents_results(page, 1)
+
+    assert [r["id"] for r in results] == ["US11930446B2"]
+
+
+# --------------------------------- shared plumbing ---------------------------------
+
+
 async def test_playwright_open_page_raises_when_browser_unavailable():
     with pytest.raises(BrowserUnavailableError):
         async with playwright_open_page(None):
             pass
+
+
+class _FakeRoute:
+    def __init__(self):
+        self.aborted = False
+        self.continued = False
+
+    async def abort(self):
+        self.aborted = True
+
+    async def continue_(self):
+        self.continued = True
+
+
+class _FakeRequest:
+    def __init__(self, resource_type):
+        self.resource_type = resource_type
+
+
+@pytest.mark.parametrize("resource_type", ["stylesheet", "image"])
+async def test_decorative_resources_are_blocked(resource_type):
+    """Only page text matters to these scrapers, and skipping stylesheets
+    and images meaningfully speeds up navigation."""
+    route = _FakeRoute()
+
+    await _block_stylesheet_and_image_resources(route, _FakeRequest(resource_type))
+
+    assert route.aborted and not route.continued
+
+
+@pytest.mark.parametrize("resource_type", ["document", "script", "xhr", "fetch"])
+async def test_everything_else_is_allowed_through(resource_type):
+    """Blocking the main document (or the XHR that renders results) would
+    break the scrape entirely."""
+    route = _FakeRoute()
+
+    await _block_stylesheet_and_image_resources(route, _FakeRequest(resource_type))
+
+    assert route.continued and not route.aborted

tests/test_serp_urls.py

@@ -0,0 +1,78 @@
+"""The search URLs each scraper navigates to.
+
+These had no coverage at all, and the gap was invisible: the Playwright
+tests drive a real browser but patch `page.goto` with a stand-in that
+accepts the URL and discards it, so everything downstream of navigation
+was verified and everything upstream was not. Mutation testing confirmed
+it - replacing the Bing URL with a constant, dropping `&num=` from the
+Scholar URL, and making the Google Patents URL ignore the query all left
+the full suite green. A scraper that searched for the wrong thing entirely
+passed.
+
+These are pure string assertions: no browser, no fixtures, microseconds.
+"""
+
+import pytest
+
+from serp import (
+    bing_search_url,
+    brave_search_url,
+    google_patents_search_url,
+    google_scholar_url,
+)
+
+
+class TestQueryIsIncluded:
+    """The single most important property: the URL must ask for what the
+    caller asked for."""
+
+    @pytest.mark.parametrize("builder", [
+        bing_search_url, brave_search_url, google_patents_search_url, google_scholar_url])
+    def test_the_query_reaches_the_url(self, builder):
+        assert "widgets" in builder("widgets", 10)
+
+    @pytest.mark.parametrize("builder", [
+        bing_search_url, brave_search_url, google_patents_search_url, google_scholar_url])
+    def test_a_different_query_produces_a_different_url(self, builder):
+        assert builder("widgets", 10) != builder("gadgets", 10)
+
+
+class TestQueryIsEncoded:
+    """A raw query in a URL is both a correctness bug (spaces, ampersands
+    and hashes truncate or corrupt the search) and the kind of thing that
+    turns user input into URL structure."""
+
+    @pytest.mark.parametrize("builder", [
+        bing_search_url, brave_search_url, google_patents_search_url, google_scholar_url])
+    def test_spaces_are_encoded(self, builder):
+        url = builder("agentic ai", 10)
+        assert "agentic+ai" in url or "agentic%20ai" in url
+        assert "agentic ai" not in url
+
+    @pytest.mark.parametrize("builder", [
+        bing_search_url, brave_search_url, google_patents_search_url, google_scholar_url])
+    def test_url_significant_characters_are_escaped(self, builder):
+        url = builder("a&b=c#d", 10)
+        query_part = url.split("?", 1)[1]
+        # Exactly one separator per parameter the builder itself added.
+        assert "a&b=c#d" not in query_part
+        assert "%26" in query_part
+
+
+class TestResultCount:
+    def test_scholar_asks_for_the_requested_number_of_results(self):
+        assert "num=25" in google_scholar_url("widgets", 25)
+
+    def test_google_patents_asks_for_the_requested_number_of_results(self):
+        assert "num=25" in google_patents_search_url("widgets", 25)
+
+
+class TestHosts:
+    @pytest.mark.parametrize("builder, host", [
+        (bing_search_url, "https://www.bing.com/search"),
+        (brave_search_url, "https://search.brave.com/search"),
+        (google_patents_search_url, "https://patents.google.com/"),
+        (google_scholar_url, "https://scholar.google.com/scholar"),
+    ])
+    def test_points_at_the_expected_endpoint(self, builder, host):
+        assert builder("widgets", 10).startswith(host)

tests/test_utils.py

@@ -1,15 +1,13 @@
 import logging
 
-from serp import SerpQuery
 from utils import log_gathered_exceptions
 
 
 def test_logs_only_the_exceptions_with_their_query(caplog):
-    params = SerpQuery(queries=["q1", "q2", "q3"])
     results = ["ok result", ValueError("boom"), "another ok"]
 
     with caplog.at_level(logging.WARNING):
-        log_gathered_exceptions(results, "test context", params)
+        log_gathered_exceptions(results, "test context", ["q1", "q2", "q3"])
 
     warnings = [r for r in caplog.records if r.levelno == logging.WARNING]
     assert len(warnings) == 1
@@ -19,11 +17,8 @@ def test_logs_only_the_exceptions_with_their_query(caplog):
 
 
 def test_logs_nothing_when_all_results_succeeded(caplog):
-    params = SerpQuery(queries=["q1", "q2"])
-    results = ["ok", "also ok"]
-
     with caplog.at_level(logging.WARNING):
-        log_gathered_exceptions(results, "test context", params)
+        log_gathered_exceptions(["ok", "also ok"], "test context", ["q1", "q2"])
 
     assert [r for r in caplog.records if r.levelno == logging.WARNING] == []
 
@@ -33,8 +28,29 @@ def test_uses_its_own_logger_not_the_asyncio_one(caplog):
     misattributed these warnings to the 'asyncio' logging namespace instead
     of the app's own. Fixed to `logging.getLogger(__name__)`.
     """
-    params = SerpQuery(queries=["q1"])
     with caplog.at_level(logging.WARNING):
-        log_gathered_exceptions([ValueError("boom")], "ctx", params)
+        log_gathered_exceptions([ValueError("boom")], "ctx", ["q1"])
 
     assert caplog.records[0].name == "utils"
+
+
+def test_takes_a_plain_sequence_of_queries(caplog):
+    """This helper is the lowest layer in the codebase; it used to import
+    SerpQuery from serp.py purely to read one attribute, inverting the
+    dependency. It now accepts any iterable of query strings, so the patent
+    and OPS paths can use it too.
+    """
+    with caplog.at_level(logging.WARNING):
+        log_gathered_exceptions([ValueError("boom")], "ctx", iter(["only-query"]))
+
+    assert "only-query" in caplog.records[0].message
+
+
+def test_extra_results_without_a_matching_query_are_ignored(caplog):
+    """zip stops at the shorter sequence - pinning that so a mismatch can't
+    raise inside an error-logging path.
+    """
+    with caplog.at_level(logging.WARNING):
+        log_gathered_exceptions([ValueError("a"), ValueError("b")], "ctx", ["q1"])
+
+    assert len([r for r in caplog.records if r.levelno == logging.WARNING]) == 1

utils.py

@@ -1,12 +1,20 @@
 import logging
-
-from serp import SerpQuery
+from typing import Iterable
 
 logger = logging.getLogger(__name__)
 
 
-def log_gathered_exceptions(results: list, context: str, params: SerpQuery):
-    """Logs gathered exceptions with context and parameters."""
-    for exc, q in zip(results, params.queries):
+def log_gathered_exceptions(results: list, context: str, queries: Iterable[str]):
+    """Log the exceptions in a `return_exceptions=True` gather, named by the
+    query that produced each.
+
+    Takes the queries themselves rather than a SerpQuery: this is the
+    lowest layer in the codebase and importing the domain model inverted
+    the dependency for the sake of one attribute access. It also means the
+    patent and OPS paths can use it.
+    """
+    # strict=False deliberately: this is an error-logging path and must
+    # never raise on a length mismatch of its own.
+    for exc, q in zip(results, queries, strict=False):
         if isinstance(exc, Exception):
             logger.warning(f"Error during {context} for query '{q}': {exc}")