Compare commits
7 Commits
worktree-g
...
main
| Author | SHA1 | Date | |
|---|---|---|---|
| 1f2ff4f1ec | |||
| 6c870ac691 | |||
| ff3a2f398c | |||
| 5a2a989e9b | |||
| df60636876 | |||
| d65c335a4a | |||
| dc203c77eb |
@@ -20,6 +20,7 @@ from __future__ import annotations
|
|||||||
|
|
||||||
import asyncio
|
import asyncio
|
||||||
import logging
|
import logging
|
||||||
|
import time
|
||||||
from uuid import UUID
|
from uuid import UUID
|
||||||
|
|
||||||
from legal_mcp.services import (
|
from legal_mcp.services import (
|
||||||
@@ -34,8 +35,132 @@ logger = logging.getLogger(__name__)
|
|||||||
_SUGGEST_PER_ISSUE = 4
|
_SUGGEST_PER_ISSUE = 4
|
||||||
_SUGGEST_FLOOR = 0.45
|
_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
|
||||||
|
|
||||||
async def build_view(case_number: str) -> dict:
|
def _passes_floor(hit: dict) -> bool:
|
||||||
|
"""Is this hit similar enough to the argument to suggest to the chair?
|
||||||
|
|
||||||
|
Reads ``relevance`` (always a cosine similarity), NOT ``score`` — ``score``
|
||||||
|
becomes a rank-fusion value (~0.008-0.02) as soon as the lexical leg returns
|
||||||
|
rows, and comparing that against a cosine-calibrated floor rejected every
|
||||||
|
hit. That is why this tab showed no supporting precedent for any argument.
|
||||||
|
|
||||||
|
``relevance is None`` means the row came from the lexical leg only, so no
|
||||||
|
cosine was ever computed. It is KEPT: it earned its place by BM25 rank
|
||||||
|
inside an already-tiny top-k, and dropping it would silently hide exact
|
||||||
|
phrase/docket matches — the very hits a chair searches for by name.
|
||||||
|
"""
|
||||||
|
rel = hit.get("relevance")
|
||||||
|
if rel is None:
|
||||||
|
return True
|
||||||
|
return float(rel) >= _SUGGEST_FLOOR
|
||||||
|
|
||||||
|
|
||||||
|
#: 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
|
||||||
|
|
||||||
|
|
||||||
|
#: Cached views, keyed by case number → (fingerprint, expires_at|None, view).
|
||||||
|
#: Assembling a view costs 20+ seconds of Voyage embeds and vector searches, and
|
||||||
|
#: the result changes only when its inputs do. Keyed by a FINGERPRINT of those
|
||||||
|
#: inputs rather than a guessed TTL: a chair who verifies a precedent must see it
|
||||||
|
#: reflected on the next load, not after a timer expires.
|
||||||
|
_view_cache: dict[str, tuple[str, float | None, dict]] = {}
|
||||||
|
|
||||||
|
#: One lock per case so two concurrent loads of the same page don't both compute.
|
||||||
|
#: Reload-during-a-slow-load is how the original outage compounded into Postgres
|
||||||
|
#: deadlocks; the second reader now waits for the first and gets its result.
|
||||||
|
_view_locks: dict[str, asyncio.Lock] = {}
|
||||||
|
|
||||||
|
#: A partial view (retrieval budget exhausted) is still worth caching — otherwise
|
||||||
|
#: every reload re-pays 22 seconds — but only briefly, so the missing suggestions
|
||||||
|
#: get another chance while the corpus is quieter.
|
||||||
|
_PARTIAL_TTL_S = 90.0
|
||||||
|
|
||||||
|
|
||||||
|
async def _inputs_fingerprint(case_id: UUID) -> str:
|
||||||
|
"""Cheap signature of everything the view is derived from.
|
||||||
|
|
||||||
|
Changes when: arguments are re-aggregated, the chair attaches or verifies a
|
||||||
|
precedent, or the corpus grows (new rulings change the suggestions). One
|
||||||
|
round-trip — the point is to be far cheaper than the 20s it guards.
|
||||||
|
"""
|
||||||
|
pool = await db.get_pool()
|
||||||
|
async with pool.acquire() as conn:
|
||||||
|
row = await conn.fetchrow(
|
||||||
|
"""
|
||||||
|
SELECT
|
||||||
|
(SELECT count(*)::text || ':' || coalesce(max(updated_at)::text, '-')
|
||||||
|
FROM legal_arguments WHERE case_id = $1) AS args,
|
||||||
|
(SELECT count(*)::text || ':' || coalesce(max(updated_at)::text, '-')
|
||||||
|
FROM case_precedents WHERE case_id = $1) AS attached,
|
||||||
|
(SELECT count(*)::text || ':' || coalesce(max(created_at)::text, '-')
|
||||||
|
FROM case_law WHERE searchable = true) AS corpus
|
||||||
|
""",
|
||||||
|
case_id,
|
||||||
|
)
|
||||||
|
return f"{row['args']}|{row['attached']}|{row['corpus']}"
|
||||||
|
|
||||||
|
|
||||||
|
def invalidate(case_number: str = "") -> None:
|
||||||
|
"""Drop cached views — one case, or all when called with no argument.
|
||||||
|
|
||||||
|
Callers that mutate the inputs may use this for immediacy, but correctness
|
||||||
|
does not depend on it: the fingerprint check catches any change on the next
|
||||||
|
read regardless of whether anyone remembered to invalidate (G1 — the guard
|
||||||
|
lives at the source of truth, not in every writer's memory).
|
||||||
|
"""
|
||||||
|
if case_number:
|
||||||
|
_view_cache.pop(case_number, None)
|
||||||
|
else:
|
||||||
|
_view_cache.clear()
|
||||||
|
|
||||||
|
|
||||||
|
async def build_view(case_number: str, *, use_cache: bool = True) -> dict:
|
||||||
|
"""Assemble the citation-verification view, served from cache when unchanged."""
|
||||||
|
if not use_cache:
|
||||||
|
return await _build_view_uncached(case_number)
|
||||||
|
|
||||||
|
lock = _view_locks.setdefault(case_number, asyncio.Lock())
|
||||||
|
async with lock:
|
||||||
|
case = await db.get_case_by_number(case_number)
|
||||||
|
if not case:
|
||||||
|
return {"status": "case_not_found", "case_number": case_number,
|
||||||
|
"arguments": []}
|
||||||
|
cid = case["id"]
|
||||||
|
fp = await _inputs_fingerprint(UUID(cid) if isinstance(cid, str) else cid)
|
||||||
|
|
||||||
|
hit = _view_cache.get(case_number)
|
||||||
|
if hit is not None:
|
||||||
|
cached_fp, expires_at, view = hit
|
||||||
|
fresh = cached_fp == fp and (expires_at is None or time.monotonic() < expires_at)
|
||||||
|
if fresh:
|
||||||
|
return {**view, "cached": True}
|
||||||
|
|
||||||
|
view = await _build_view_uncached(case_number)
|
||||||
|
if view.get("status") == "ok":
|
||||||
|
expires_at = (None if view.get("retrieval_complete", True)
|
||||||
|
else time.monotonic() + _PARTIAL_TTL_S)
|
||||||
|
_view_cache[case_number] = (fp, expires_at, view)
|
||||||
|
return {**view, "cached": False}
|
||||||
|
|
||||||
|
|
||||||
|
async def _build_view_uncached(case_number: str) -> dict:
|
||||||
case = await db.get_case_by_number(case_number)
|
case = await db.get_case_by_number(case_number)
|
||||||
if not case:
|
if not case:
|
||||||
return {"status": "case_not_found", "case_number": case_number, "arguments": []}
|
return {"status": "case_not_found", "case_number": case_number, "arguments": []}
|
||||||
@@ -83,13 +208,46 @@ async def build_view(case_number: str) -> dict:
|
|||||||
logger.warning("citation_verification search failed (%s): %s", title[:30], e)
|
logger.warning("citation_verification search failed (%s): %s", title[:30], e)
|
||||||
# Resolve the authority breakdown for the hit set in one batched query.
|
# Resolve the authority breakdown for the hit set in one batched query.
|
||||||
clids = [UUID(str(h["case_law_id"])) for h in hits
|
clids = [UUID(str(h["case_law_id"])) for h in hits
|
||||||
if h.get("case_law_id") and float(h.get("score", 0) or 0) >= _SUGGEST_FLOOR]
|
if h.get("case_law_id") and _passes_floor(h)]
|
||||||
authority = await db.citation_authority(clids) if clids else {}
|
authority = await db.citation_authority(clids) if clids else {}
|
||||||
return hits, authority
|
return hits, authority
|
||||||
|
|
||||||
# Fan out the expensive per-argument retrieval concurrently — was a sequential
|
# Fan out the expensive per-argument retrieval, but BOUNDED — see
|
||||||
# waterfall (N args × Voyage embed + vector search each). gather preserves order.
|
# _MAX_CONCURRENT_LOOKUPS. gather preserves order, so the zip below still
|
||||||
fetched = await asyncio.gather(*(_fetch(a) for a in args)) if args else []
|
# 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] = []
|
out_args: list[dict] = []
|
||||||
n_verified = 0
|
n_verified = 0
|
||||||
@@ -104,7 +262,7 @@ async def build_view(case_number: str) -> dict:
|
|||||||
clid = str(h.get("case_law_id") or "")
|
clid = str(h.get("case_law_id") or "")
|
||||||
if not clid or clid in seen:
|
if not clid or clid in seen:
|
||||||
continue
|
continue
|
||||||
if float(h.get("score", 0) or 0) < _SUGGEST_FLOOR:
|
if not _passes_floor(h):
|
||||||
continue
|
continue
|
||||||
seen.add(clid)
|
seen.add(clid)
|
||||||
att = attached_by_arg.get(aid, {}).get(clid)
|
att = attached_by_arg.get(aid, {}).get(clid)
|
||||||
@@ -137,6 +295,11 @@ async def build_view(case_number: str) -> dict:
|
|||||||
return {
|
return {
|
||||||
"status": "ok",
|
"status": "ok",
|
||||||
"case_number": case_number,
|
"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,
|
"arguments": out_args,
|
||||||
"summary": {
|
"summary": {
|
||||||
"arguments_total": len(out_args),
|
"arguments_total": len(out_args),
|
||||||
|
|||||||
@@ -7979,6 +7979,14 @@ async def search_precedent_library_semantic(
|
|||||||
# Calibrated so the average (≈0.85) stays at +0.05 (legacy value).
|
# Calibrated so the average (≈0.85) stays at +0.05 (legacy value).
|
||||||
_conf = float(d.get("confidence") or 0.0)
|
_conf = float(d.get("confidence") or 0.0)
|
||||||
d["score"] = float(d["score"]) + max(_conf * 0.06, 0.0)
|
d["score"] = float(d["score"]) + max(_conf * 0.06, 0.0)
|
||||||
|
# Stable cosine-scale relevance, carried alongside ``score``.
|
||||||
|
# ``score`` is the RANKING signal and downstream fusion overwrites it
|
||||||
|
# with an RRF value (~0.008-0.02) whenever the lexical leg returns
|
||||||
|
# rows — a different scale entirely. Anything that THRESHOLDS must
|
||||||
|
# read ``relevance`` instead, which always means "cosine similarity
|
||||||
|
# to the query" no matter which fusion stages ran. See
|
||||||
|
# hybrid_search._merge_sem_lex.
|
||||||
|
d["relevance"] = d["score"]
|
||||||
d["type"] = "halacha"
|
d["type"] = "halacha"
|
||||||
# authority is DERIVED from the source, never stored (INV-DM7)
|
# authority is DERIVED from the source, never stored (INV-DM7)
|
||||||
d["authority"] = halacha_quality.derive_authority(d.get("precedent_level"))
|
d["authority"] = halacha_quality.derive_authority(d.get("precedent_level"))
|
||||||
@@ -7990,6 +7998,7 @@ async def search_precedent_library_semantic(
|
|||||||
if d.get("decision_date") is not None:
|
if d.get("decision_date") is not None:
|
||||||
d["decision_date"] = d["decision_date"].isoformat()
|
d["decision_date"] = d["decision_date"].isoformat()
|
||||||
d["score"] = float(d["score"])
|
d["score"] = float(d["score"])
|
||||||
|
d["relevance"] = d["score"] # cosine anchor — see the halacha branch above
|
||||||
d["type"] = "passage"
|
d["type"] = "passage"
|
||||||
_maybe_swap_parent(d)
|
_maybe_swap_parent(d)
|
||||||
results.append(d)
|
results.append(d)
|
||||||
|
|||||||
@@ -303,6 +303,21 @@ def _merge_sem_lex(
|
|||||||
if key in lex_row_by_key else 0.0
|
if key in lex_row_by_key else 0.0
|
||||||
d["lex_rank"] = lex_rank or 0
|
d["lex_rank"] = lex_rank or 0
|
||||||
d["score"] = sem_term + lex_term
|
d["score"] = sem_term + lex_term
|
||||||
|
# ``score`` is now an RRF value (~0.008-0.02), NOT a cosine. Carry the
|
||||||
|
# cosine forward under ``relevance`` so thresholding callers keep a
|
||||||
|
# stable scale — without it, a caller comparing score >= 0.45 silently
|
||||||
|
# drops every result the moment the lexical leg returns anything, which
|
||||||
|
# is exactly how the citation-verification tab ended up showing no
|
||||||
|
# supporting precedent for any argument.
|
||||||
|
#
|
||||||
|
# A lexical-only row has no cosine at all. It gets ``relevance = None``
|
||||||
|
# rather than 0.0: "we did not measure this" is not "measured as
|
||||||
|
# irrelevant", and the row earned its place by BM25 rank. Callers decide
|
||||||
|
# (see case_citation_verification) — but they must decide knowingly.
|
||||||
|
if key in sem_row_by_key:
|
||||||
|
d["relevance"] = float(sem_row_by_key[key].get("relevance", d["sem_score"]))
|
||||||
|
else:
|
||||||
|
d["relevance"] = None
|
||||||
merged.append(d)
|
merged.append(d)
|
||||||
|
|
||||||
merged.sort(key=lambda x: -float(x["score"]))
|
merged.sort(key=lambda x: -float(x["score"]))
|
||||||
|
|||||||
125
mcp-server/tests/test_citation_view_cache.py
Normal file
125
mcp-server/tests/test_citation_view_cache.py
Normal file
@@ -0,0 +1,125 @@
|
|||||||
|
"""The citation view is cached by input fingerprint, not by a guessed TTL.
|
||||||
|
|
||||||
|
Assembling it costs 20+ seconds of Voyage embeds and vector searches. A TTL
|
||||||
|
would force a choice between staleness and cost: a chair who verifies a
|
||||||
|
precedent must see it on the very next load, and a timer cannot promise that.
|
||||||
|
Keying on a fingerprint of the inputs (arguments · attachments · corpus) makes
|
||||||
|
freshness a property of the data rather than of the clock.
|
||||||
|
"""
|
||||||
|
|
||||||
|
import asyncio
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from legal_mcp.services import case_citation_verification as ccv
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture(autouse=True)
|
||||||
|
def _clean_cache():
|
||||||
|
ccv.invalidate()
|
||||||
|
yield
|
||||||
|
ccv.invalidate()
|
||||||
|
|
||||||
|
|
||||||
|
def _stub(monkeypatch, fingerprint: str, calls: list, *, complete: bool = True):
|
||||||
|
"""Point build_view at a fake case, a controllable fingerprint, and a
|
||||||
|
counted assembler, so these tests never touch the corpus."""
|
||||||
|
async def fake_case(_cn):
|
||||||
|
return {"id": "00000000-0000-0000-0000-000000000001"}
|
||||||
|
|
||||||
|
async def fake_fp(_cid):
|
||||||
|
return fingerprint
|
||||||
|
|
||||||
|
async def fake_build(cn):
|
||||||
|
calls.append(cn)
|
||||||
|
return {"status": "ok", "case_number": cn, "arguments": [],
|
||||||
|
"retrieval_complete": complete, "summary": {}}
|
||||||
|
|
||||||
|
monkeypatch.setattr(ccv.db, "get_case_by_number", fake_case)
|
||||||
|
monkeypatch.setattr(ccv, "_inputs_fingerprint", fake_fp)
|
||||||
|
monkeypatch.setattr(ccv, "_build_view_uncached", fake_build)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_second_load_is_served_from_cache(monkeypatch):
|
||||||
|
calls: list = []
|
||||||
|
_stub(monkeypatch, "fp-1", calls)
|
||||||
|
first = await ccv.build_view("1069-04-26")
|
||||||
|
second = await ccv.build_view("1069-04-26")
|
||||||
|
assert first["cached"] is False
|
||||||
|
assert second["cached"] is True
|
||||||
|
assert len(calls) == 1, "the expensive assembly must run once"
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_changed_inputs_invalidate_without_anyone_calling_invalidate(monkeypatch):
|
||||||
|
"""The chair verifies a precedent → fingerprint moves → next load rebuilds.
|
||||||
|
Correctness must not depend on a writer remembering to clear the cache."""
|
||||||
|
calls: list = []
|
||||||
|
_stub(monkeypatch, "fp-before", calls)
|
||||||
|
await ccv.build_view("1069-04-26")
|
||||||
|
|
||||||
|
_stub(monkeypatch, "fp-after", calls) # e.g. a new case_precedents row
|
||||||
|
again = await ccv.build_view("1069-04-26")
|
||||||
|
assert again["cached"] is False
|
||||||
|
assert len(calls) == 2
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_concurrent_loads_do_not_stampede(monkeypatch):
|
||||||
|
"""Reload-during-a-slow-load is how the original outage compounded into
|
||||||
|
Postgres deadlocks. The later readers must wait, not pile on."""
|
||||||
|
calls: list = []
|
||||||
|
|
||||||
|
async def fake_case(_cn):
|
||||||
|
return {"id": "00000000-0000-0000-0000-000000000001"}
|
||||||
|
|
||||||
|
async def fake_fp(_cid):
|
||||||
|
return "fp-1"
|
||||||
|
|
||||||
|
async def slow_build(cn):
|
||||||
|
calls.append(cn)
|
||||||
|
await asyncio.sleep(0.05)
|
||||||
|
return {"status": "ok", "case_number": cn, "arguments": [],
|
||||||
|
"retrieval_complete": True, "summary": {}}
|
||||||
|
|
||||||
|
monkeypatch.setattr(ccv.db, "get_case_by_number", fake_case)
|
||||||
|
monkeypatch.setattr(ccv, "_inputs_fingerprint", fake_fp)
|
||||||
|
monkeypatch.setattr(ccv, "_build_view_uncached", slow_build)
|
||||||
|
|
||||||
|
await asyncio.gather(*(ccv.build_view("1069-04-26") for _ in range(6)))
|
||||||
|
assert len(calls) == 1, "six concurrent loads, one assembly"
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_partial_view_expires_so_missing_suggestions_get_retried(monkeypatch):
|
||||||
|
"""A budget-exhausted view is cached to stop reload storms, but it must not
|
||||||
|
become permanent — the arguments with no suggestions deserve another try."""
|
||||||
|
calls: list = []
|
||||||
|
_stub(monkeypatch, "fp-1", calls, complete=False)
|
||||||
|
await ccv.build_view("1069-04-26")
|
||||||
|
_fp, expires_at, _view = ccv._view_cache["1069-04-26"]
|
||||||
|
assert expires_at is not None, "a partial view must carry an expiry"
|
||||||
|
|
||||||
|
monkeypatch.setattr(ccv.time, "monotonic", lambda: expires_at + 1)
|
||||||
|
after = await ccv.build_view("1069-04-26")
|
||||||
|
assert after["cached"] is False
|
||||||
|
assert len(calls) == 2
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_complete_view_does_not_expire_on_a_timer(monkeypatch):
|
||||||
|
calls: list = []
|
||||||
|
_stub(monkeypatch, "fp-1", calls, complete=True)
|
||||||
|
await ccv.build_view("1069-04-26")
|
||||||
|
_fp, expires_at, _view = ccv._view_cache["1069-04-26"]
|
||||||
|
assert expires_at is None, "freshness comes from the fingerprint, not a clock"
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_use_cache_false_always_rebuilds(monkeypatch):
|
||||||
|
calls: list = []
|
||||||
|
_stub(monkeypatch, "fp-1", calls)
|
||||||
|
await ccv.build_view("1069-04-26")
|
||||||
|
await ccv.build_view("1069-04-26", use_cache=False)
|
||||||
|
assert len(calls) == 2
|
||||||
81
mcp-server/tests/test_citation_view_fanout.py
Normal file
81
mcp-server/tests/test_citation_view_fanout.py
Normal file
@@ -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
|
||||||
68
mcp-server/tests/test_relevance_scale.py
Normal file
68
mcp-server/tests/test_relevance_scale.py
Normal file
@@ -0,0 +1,68 @@
|
|||||||
|
"""`score` and `relevance` are different things — thresholds must use `relevance`.
|
||||||
|
|
||||||
|
Retrieval returns cosine similarities (~0.4-0.75) until the lexical leg returns
|
||||||
|
rows; then `_merge_sem_lex` replaces `score` with an RRF value (~0.008-0.02).
|
||||||
|
Both are legitimate *ranking* signals, but they are not on the same scale, so a
|
||||||
|
caller comparing `score >= 0.45` rejected every hit the moment BM25 matched
|
||||||
|
anything. That is how the citation-verification tab came to show no supporting
|
||||||
|
precedent for a single argument, on every case.
|
||||||
|
|
||||||
|
`relevance` is the fix: always a cosine, or None when the row came from the
|
||||||
|
lexical leg alone and no cosine was ever computed.
|
||||||
|
"""
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from legal_mcp.services.case_citation_verification import _SUGGEST_FLOOR, _passes_floor
|
||||||
|
from legal_mcp.services.hybrid_search import _merge_sem_lex
|
||||||
|
|
||||||
|
|
||||||
|
def _sem(key: str, score: float) -> dict:
|
||||||
|
return {"chunk_id": key, "case_law_id": "c1", "score": score, "relevance": score}
|
||||||
|
|
||||||
|
|
||||||
|
def _lex(key: str, score: float) -> dict:
|
||||||
|
# The lexical leg emits ts_rank_cd, never a cosine — so no `relevance`.
|
||||||
|
return {"chunk_id": key, "case_law_id": "c1", "score": score}
|
||||||
|
|
||||||
|
|
||||||
|
def test_fusion_replaces_score_but_preserves_the_cosine():
|
||||||
|
"""The regression in one assertion."""
|
||||||
|
merged = _merge_sem_lex([_sem("a", 0.73)], [_lex("a", 0.31)], limit=10)
|
||||||
|
row = merged[0]
|
||||||
|
assert row["score"] < 0.1, "fused score is an RRF value, not a cosine"
|
||||||
|
assert row["relevance"] == pytest.approx(0.73), "the cosine must survive fusion"
|
||||||
|
|
||||||
|
|
||||||
|
def test_lexical_only_row_has_no_fabricated_cosine():
|
||||||
|
"""None means 'not measured' — not 'measured as irrelevant'."""
|
||||||
|
merged = _merge_sem_lex([], [_lex("b", 0.31)], limit=10)
|
||||||
|
assert merged[0]["relevance"] is None
|
||||||
|
|
||||||
|
|
||||||
|
def test_semantic_only_row_keeps_its_cosine():
|
||||||
|
merged = _merge_sem_lex([_sem("c", 0.62)], [], limit=10)
|
||||||
|
assert merged[0]["relevance"] == pytest.approx(0.62)
|
||||||
|
|
||||||
|
|
||||||
|
def test_floor_would_have_rejected_everything_on_the_fused_score():
|
||||||
|
"""Guards the exact production symptom: fused scores are ~0.008, the floor
|
||||||
|
is 0.45, so score-based filtering wipes the result set."""
|
||||||
|
merged = _merge_sem_lex(
|
||||||
|
[_sem(k, 0.70) for k in "abcd"], [_lex(k, 0.30) for k in "abcd"], limit=10)
|
||||||
|
assert all(r["score"] < _SUGGEST_FLOOR for r in merged) # the bug
|
||||||
|
assert all(_passes_floor(r) for r in merged) # the fix
|
||||||
|
|
||||||
|
|
||||||
|
def test_floor_still_rejects_genuinely_weak_hits():
|
||||||
|
"""The fix must not become 'accept everything'."""
|
||||||
|
assert not _passes_floor({"relevance": 0.10})
|
||||||
|
assert not _passes_floor({"relevance": _SUGGEST_FLOOR - 0.01})
|
||||||
|
assert _passes_floor({"relevance": _SUGGEST_FLOOR})
|
||||||
|
|
||||||
|
|
||||||
|
def test_lexical_only_hits_are_kept_deliberately():
|
||||||
|
"""An exact docket/phrase match reaches the top by BM25 rank with no cosine.
|
||||||
|
Dropping it would hide precisely what a chair searches for by name."""
|
||||||
|
assert _passes_floor({"relevance": None})
|
||||||
|
assert _passes_floor({}) # missing key behaves the same as None
|
||||||
Reference in New Issue
Block a user