perf(citation-view): cache מבוסס-טביעה — 24.6 שנ' לטעינה חוזרת → 0.01 שנ'

הרכבת תצוגת אימות-הפסיקה עולה 20+ שניות של embeds וחיפושים וקטוריים,
והתוצאה משתנה רק כשמשתנים הקלטים שלה. עד עכשיו כל פתיחת-דף שילמה את
המחיר המלא מחדש, וכל רענון תוך-כדי-טעינה הוסיף הרצה מקבילה — כך העומס
המקורי התגלגל ל-deadlocks ב-Postgres.

למה טביעה ולא TTL: יו"ר שמאמת תקדים חייב לראות זאת בטעינה **הבאה**, לא
כשיפוג טיימר. `_inputs_fingerprint` הוא שאילתה אחת זולה שמסכמת בדיוק את
מה שהתצוגה נגזרת ממנו — טיעונים · שיוכי-פסיקה · גודל-הקורפוס. השתנה
משהו → הטביעה זזה → נבנה מחדש. **הנכונות אינה תלויה בכך שמישהו יזכור
לבטל** (G1): הבדיקה יושבת במקור-האמת ולא בזיכרון של כל כותב. `invalidate()`
קיים לנוחות, לא לתקינות.

מה שונה
- cache פר-תיק, מפתח = טביעת-הקלטים. תצוגה שלמה אינה פגה בטיימר.
- **מנעול פר-תיק נגד היצף** — שש טעינות מקבילות מריצות הרכבה אחת. זה
  בדיוק התרחיש שהפיל את המערכת.
- תצוגה **חלקית** (תקציב-האחזור נגמר) נשמרת ל-90 שנ' בלבד: מספיק כדי
  לעצור סופת-רענונים, לא מספיק כדי שההצעות החסרות ייתקעו לנצח.
- `cached: true/false` בתשובה, ו-`use_cache=False` למי שצריך רענון כפוי.

מדידה מול הקורפוס החי (8124-09-24):
    טעינה קרה ........ 24.6 שנ'   cached=false
    טעינה חוזרת ...... 0.01 שנ'   cached=true      ← פי 2,074
    5 מקבילות ........ 0.02 שנ'   כולן מה-cache
    אחרי invalidate .. 24.8 שנ'   cached=false

invariants: G1 (השער במקור-האמת, לא בכל כותב) · §6 (חלקי מסומן ופג,
לא נשמר כשלם)

טסטים: 6 חדשים (tests/test_citation_view_cache.py) — כולל היצף-מקבילי,
פקיעת תצוגה חלקית, וביטול-אוטומטי בלי קריאה מפורשת ל-invalidate.
537 עוברים.

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

@@ -20,6 +20,7 @@ from __future__ import annotations
 
 import asyncio
 import logging
+import time
 from uuid import UUID
 
 from legal_mcp.services import (
@@ -74,7 +75,92 @@ def _passes_floor(hit: dict) -> bool:
 _RETRIEVAL_BUDGET_S = 22.0
 
 
-async def build_view(case_number: str) -> dict:
+#: 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)
     if not case:
         return {"status": "case_not_found", "case_number": case_number, "arguments": []}