fix(citation-view): פיזור בלתי-מוגבל הפיל את דף אימות-הפסיקה ב-500

פתיחת טאב "אימות פסיקה" ירתה `asyncio.gather` על **כל** טיעוני התיק בבת
אחת — טיעון אחד = קריאת embed ל-Voyage + חיפוש היברידי + שאילתת סמכות.
מעל ~8 קריאות במקביל Voyage מווסת, וכולן נתקעות יחד. הפיזור הבלתי-מוגבל
היה לא רק מסוכן אלא **איטי יותר**:

    חיפוש בודד ............  1.2 שנ'
    32 ללא מגבלה .......... 30.4 שנ'   ← פי 25 מחיפוש בודד
    32 עם מגבלת 8 ......... 22.1 שנ'   ← מבוקר = מהיר יותר

תיק עם 69 טיעונים (1069-04-26) חרג מ-timeout ה-30 שניות של הפרוקסי והחזיר
500; רענוני-דף חופפים הערימו תחרות עד `deadlock detected` ב-Postgres,
והאטו את כל המערכת.

מה שונה
- `_MAX_CONCURRENT_LOOKUPS = 8` — semaphore על הפיזור.
- `_RETRIEVAL_BUDGET_S = 22` — תקציב זמן שנגמר לפני הפרוקסי.
- בפקיעת התקציב **קוצרים את מה שהספיק** במקום לבטל הכל. במפורש *לא*
  `wait_for(gather(...))`: הוא מבטל כל משימה, כך שאיחור אחד היה זורק 30
  תוצאות שכבר הצליחו. משימות ממתינות מבוטלות ונאספות לפני שהמאגר
  מתפרק, אחרת נותרת `connection_lost` יתומה.
- `retrieval_complete: false` בתשובה — הצעות חסרות מסומנות ולא מוצגות
  כ"אין תקדים תומך" (§6). שתי הטענות שונות.

מדידה אחרי: 24 מתוך 32 טיעונים ו-26 מתוך 69 חוזרים עם הצעות, במקום 500.

invariants: §6 (חלקי מסומן, לא נבלע) · INV-AH (לא להציג היעדר-מידע
כהיעדר-תקדים)

⚠️ **נותר פתוח, נפרד:** גם כשהחיפוש מצליח `arguments_with_support=0` —
`hybrid_search.py:305` דורס את ציון הקוסינוס בציון RRF כשה-leg הלקסיקלי
מחזיר תוצאות, כך ש-`_SUGGEST_FLOOR=0.45` (מכויל לקוסינוס) מסנן הכל.
מטופל ב-PR נפרד.

טסטים: 4 חדשים (tests/test_citation_view_fanout.py) — הראשונים נועלים את
המגבלה והתקציב, האחרון משחזר בדיוק את הרגרסיה: קציר מול ביטול-הכל.
525 עוברים.

mcp-server/src/legal_mcp/services/case_citation_verification.py

@@ -34,6 +34,26 @@ logger = logging.getLogger(__name__)
 _SUGGEST_PER_ISSUE = 4
 _SUGGEST_FLOOR = 0.45
 
+#: Concurrent per-argument retrievals. The fan-out used to be unbounded — one
+#: task per legal argument — which is self-defeating, not merely risky: each
+#: task opens a Voyage embed call, and past ~8 in flight Voyage throttles, so
+#: every request stalls together. Measured on this corpus (32 arguments):
+#:
+#:     one search alone ......  1.2s
+#:     32 unbounded .......... 30.4s   ← 25× a single search
+#:     32 at 8 concurrent .... 22.1s   ← bounded is FASTER
+#:
+#: A 69-argument case (1069-04-26) therefore blew past the 30s proxy timeout and
+#: returned 500, and overlapping page reloads piled contention into Postgres
+#: deadlocks. Bounding the fan-out both fixes the failure and speeds it up.
+_MAX_CONCURRENT_LOOKUPS = 8
+
+#: Wall-clock ceiling for the whole retrieval phase. The proxy gives up at 30s;
+#: cutting ourselves off earlier lets us return the suggestions that DID land
+#: instead of a 500 that shows the chair nothing. Partial results are labelled
+#: (``retrieval_complete: false``) rather than passed off as the full picture.
+_RETRIEVAL_BUDGET_S = 22.0
+
 
 async def build_view(case_number: str) -> dict:
     case = await db.get_case_by_number(case_number)
@@ -87,9 +107,42 @@ async def build_view(case_number: str) -> dict:
         authority = await db.citation_authority(clids) if clids else {}
         return hits, authority
 
-    # Fan out the expensive per-argument retrieval concurrently — was a sequential
-    # waterfall (N args × Voyage embed + vector search each). gather preserves order.
-    fetched = await asyncio.gather(*(_fetch(a) for a in args)) if args else []
+    # Fan out the expensive per-argument retrieval, but BOUNDED — see
+    # _MAX_CONCURRENT_LOOKUPS. gather preserves order, so the zip below still
+    # pairs each argument with its own result.
+    _sem = asyncio.Semaphore(_MAX_CONCURRENT_LOOKUPS)
+
+    async def _fetch_bounded(a: dict) -> tuple[list[dict], dict]:
+        async with _sem:
+            return await _fetch(a)
+
+    retrieval_complete = True
+    fetched: list[tuple[list[dict], dict]] = []
+    if args:
+        # Harvest whatever finished inside the budget, per argument. Deliberately
+        # NOT wait_for(gather(...)): that cancels every task on timeout, so one
+        # slow lookup would throw away the 30 that already succeeded and the page
+        # would show nothing at all.
+        tasks = [asyncio.ensure_future(_fetch_bounded(a)) for a in args]
+        done, pending = await asyncio.wait(tasks, timeout=_RETRIEVAL_BUDGET_S)
+        for t in pending:
+            t.cancel()
+        if pending:
+            # Let the cancellations settle before the caller's DB pool unwinds —
+            # a task cancelled mid-query otherwise surfaces as a stray
+            # "connection_lost" future with no owner.
+            await asyncio.gather(*pending, return_exceptions=True)
+            retrieval_complete = False
+            logger.warning(
+                "citation_verification: retrieval budget of %.0fs exhausted for %s "
+                "— %d of %d arguments returned suggestions, the rest are empty",
+                _RETRIEVAL_BUDGET_S, case_number, len(done), len(args),
+            )
+        for t in tasks:
+            if t in done and not t.cancelled() and t.exception() is None:
+                fetched.append(t.result())
+            else:
+                fetched.append(([], {}))
 
     out_args: list[dict] = []
     n_verified = 0
@@ -137,6 +190,11 @@ async def build_view(case_number: str) -> dict:
     return {
         "status": "ok",
         "case_number": case_number,
+        # False when the retrieval budget ran out: the attached/verified rows and
+        # the radar are complete, but the corpus SUGGESTIONS are missing. The UI
+        # must say so — an empty suggestion list otherwise reads as "no precedent
+        # in the corpus supports this argument", which is a different claim.
+        "retrieval_complete": retrieval_complete,
         "arguments": out_args,
         "summary": {
             "arguments_total": len(out_args),

mcp-server/tests/test_citation_view_fanout.py

@@ -0,0 +1,81 @@
+"""The citation-verification view must not fan out without a bound.
+
+`build_view` used to spawn one retrieval task per legal argument with a bare
+`asyncio.gather`. Each task opens a Voyage embed call, and past ~8 in flight
+Voyage throttles — so the unbounded version was *slower* than a bounded one
+(32 arguments: 30.4s unbounded vs 22.1s at 8), and a 69-argument case blew past
+the 30s proxy timeout and returned 500 while overlapping reloads piled
+contention into Postgres deadlocks.
+
+These tests pin the two properties that fix gave us: the fan-out is bounded,
+and a timeout yields the results that DID land instead of nothing.
+"""
+
+import asyncio
+
+import pytest
+
+from legal_mcp.services import case_citation_verification as ccv
+
+
+def test_concurrency_bound_is_set_and_modest():
+    assert 1 <= ccv._MAX_CONCURRENT_LOOKUPS <= 16, (
+        "the bound exists to stay under Voyage's throttle point — a large value "
+        "reintroduces the stall this was added to fix"
+    )
+
+
+def test_retrieval_budget_leaves_room_under_the_proxy_timeout():
+    """The proxy gives up at 30s; we must cut ourselves off before that."""
+    assert 0 < ccv._RETRIEVAL_BUDGET_S < 30
+
+
+@pytest.mark.asyncio
+async def test_semaphore_actually_caps_in_flight_work():
+    """A semaphore of N never lets N+1 coroutines run the body at once."""
+    limit = ccv._MAX_CONCURRENT_LOOKUPS
+    sem = asyncio.Semaphore(limit)
+    in_flight = 0
+    peak = 0
+
+    async def worker():
+        nonlocal in_flight, peak
+        async with sem:
+            in_flight += 1
+            peak = max(peak, in_flight)
+            await asyncio.sleep(0.01)
+            in_flight -= 1
+
+    await asyncio.gather(*(worker() for _ in range(limit * 4)))
+    assert peak <= limit
+
+
+@pytest.mark.asyncio
+async def test_timeout_harvests_finished_work_instead_of_discarding_it():
+    """The regression: wait_for(gather(...)) cancels everything on timeout, so
+    one slow lookup threw away every result that had already succeeded. The
+    harvest pattern must keep them."""
+
+    async def quick(i):
+        await asyncio.sleep(0.01)
+        return i
+
+    async def never():
+        await asyncio.sleep(30)
+        return "unreachable"
+
+    tasks = [asyncio.ensure_future(quick(i)) for i in range(5)]
+    tasks.append(asyncio.ensure_future(never()))
+
+    done, pending = await asyncio.wait(tasks, timeout=0.3)
+    for t in pending:
+        t.cancel()
+    await asyncio.gather(*pending, return_exceptions=True)
+
+    harvested = [
+        t.result() if (t in done and not t.cancelled() and t.exception() is None) else None
+        for t in tasks
+    ]
+    assert harvested[:5] == [0, 1, 2, 3, 4], "finished work must survive the timeout"
+    assert harvested[5] is None, "the unfinished one is empty, not fabricated"
+    assert len(pending) == 1