From d65c335a4a9fc7dcc4d4c12beee8f11ec8628210 Mon Sep 17 00:00:00 2001 From: Chaim Date: Wed, 5 Aug 2026 13:25:41 +0000 Subject: [PATCH] =?UTF-8?q?fix(citation-view):=20=D7=A4=D7=99=D7=96=D7=95?= =?UTF-8?q?=D7=A8=20=D7=91=D7=9C=D7=AA=D7=99-=D7=9E=D7=95=D7=92=D7=91?= =?UTF-8?q?=D7=9C=20=D7=94=D7=A4=D7=99=D7=9C=20=D7=90=D7=AA=20=D7=93=D7=A3?= =?UTF-8?q?=20=D7=90=D7=99=D7=9E=D7=95=D7=AA-=D7=94=D7=A4=D7=A1=D7=99?= =?UTF-8?q?=D7=A7=D7=94=20=D7=91-500?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit פתיחת טאב "אימות פסיקה" ירתה `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 עוברים. --- .../services/case_citation_verification.py | 64 ++++++++++++++- mcp-server/tests/test_citation_view_fanout.py | 81 +++++++++++++++++++ 2 files changed, 142 insertions(+), 3 deletions(-) create mode 100644 mcp-server/tests/test_citation_view_fanout.py diff --git a/mcp-server/src/legal_mcp/services/case_citation_verification.py b/mcp-server/src/legal_mcp/services/case_citation_verification.py index 26c2f97..bbeb0ff 100644 --- a/mcp-server/src/legal_mcp/services/case_citation_verification.py +++ b/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), diff --git a/mcp-server/tests/test_citation_view_fanout.py b/mcp-server/tests/test_citation_view_fanout.py new file mode 100644 index 0000000..2f0226c --- /dev/null +++ b/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