Bound the list inputs and the outbound fan-out they cause

`n_results` was defended by a validator and a seven-case parametrized test
while the list fields alongside it were undefended entirely. Two
consequences, both reachable from an unauthenticated request:

An empty query list crashed every search endpoint. `_shape_serp_results`
reads `results[-1]` to build its error message, so `{"queries": []}`
raised IndexError - an unhandled 500 on all six endpoints, from a
two-character body.

The bulk endpoints fanned out without any limit. `patent_ids` had no
maximum and both bulk helpers gathered the whole list at once, so 3000 ids
opened 3000 concurrent scrapes; the OPS variant is three times worse again,
since each id costs a biblio, a claims and a description request. The
Playwright scrapers have had a concurrency semaphore from the start - this
brings the HTTP paths in line with them.

- min_length/max_length on SerpQuery.queries and ScrapPatentsRequest.patent_ids
- semaphores in scrap_patent_bulk_async and ops_scrap_patent_bulk
- _shape_serp_results is now total for an empty list, so it doesn't depend
on its caller having validated anything

The concurrency tests assert the observed peak rather than the limit
constant, so they fail if the semaphore is removed rather than just
re-stating the configuration.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01MSNSYnceqvdVz7Csis4K9e

app.py

@@ -126,6 +126,12 @@ def _shape_serp_results(results: list) -> SerpResults:
     `return_exceptions=True` gather) into one SerpResults, surfacing the
     last error only when every query failed.
     """
+    if not results:
+        # SerpQuery bounds 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 endpoint 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]
@@ -307,9 +313,17 @@ async def scrap_patent(patent_id: PatentId) -> PatentScrapResult:
             detail=f"EPO OPS request failed for '{patent_id}': {e}")
 
 
+# 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
+
+
 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")
 
 

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

scrap.py

@@ -131,11 +131,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

@@ -16,8 +16,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")

tests/test_input_bounds.py

@@ -0,0 +1,148 @@
+"""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
+
+
+@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 endpoint.
+    """
+    result = app_module._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