feat(digests): case-contextual digest radar — surface unlinked digests as chair leads (X12)

A digest pointing at a ruling we don't hold yet ("unlinked") was captured globally
(missing_precedents inbox) but never surfaced IN THE CONTEXT of the case being
decided — so a relevant ruling known only via a digest could fall through the cracks
at the moment it matters. Adds the case-contextual radar:

- db.search_digests_semantic: new `linked_only` filter (False = unlinked-only target set).
- digest_library.case_digest_radar(case_number): builds the case topic from title +
  appeal_subtype + the analyst's claims, embeds once, matches against UNLINKED digests,
  and returns leads enriched with the underlying ruling's gap status + suggested action
  (new_lead / gap_open / fetched / available_link).
- MCP tool `digest_radar` + endpoint GET /api/cases/{n}/digest-radar (for the agent and
  the future case-page lead).

INV-DIG1 preserved: radar only — every lead points at the underlying RULING (fetch /
upload / link), never cites the digest. Read-only.

Validated on 8124-09-24 (היטל השבחה): 5 on-topic unlinked digests, scores 0.64–0.72.
The visible case-page panel is a separate UI change → goes through the design gate.

Invariants: G2 (reuses the one digest semantic-search + gap/citation resolvers),
INV-DIG1 (no digest citation), INV-DIG3 (gap surfacing). No schema change.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

mcp-server/src/legal_mcp/server.py

@@ -418,6 +418,12 @@ async def digest_process_pending(limit: int = 20) -> str:
     return await digest_tools.digest_process_pending(_clamp_limit(limit))
 
 
+@mcp.tool()
+async def digest_radar(case_number: str, limit: int = 5, min_score: float = 0.45) -> str:
+    """רדאר-יומונים הקשרי-לתיק (X12) — מחזיר יומונים לא-מקושרים (פס"ד שעוד אין לנו) שהנושא שלהם קרוב סמנטית לתיק, כדי שלא ייפול פס"ד רלוונטי בזמן ההכרעה. כל ליד מצביע על מראה-המקום של הפס"ד המקורי + סטטוס-הפער + פעולה מוצעת. ⚠️ radar בלבד — היומון אינו מצוטט (INV-DIG1)."""
+    return await digest_tools.digest_radar(case_number, _clamp_limit(limit), min_score)
+
+
 @mcp.tool()
 async def halacha_review(
     halacha_id: str,

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

@@ -5018,14 +5018,24 @@ async def search_digests_semantic(
     subject_tag: str = "",
     concept_tag: str = "",
     limit: int = 10,
+    linked_only: bool | None = None,
 ) -> list[dict]:
     """Pure-semantic search over the digests radar (X12). Single vector per row
     (no chunks/halachot), so no RRF here — see X12 §6. Joins the linked ruling's
-    citation when present so the researcher sees the pointer target directly."""
+    citation when present so the researcher sees the pointer target directly.
+
+    ``linked_only``: None = all digests (default); False = only UNLINKED digests
+    (``linked_case_law_id IS NULL`` — rulings we don't hold yet, the case radar's
+    target set); True = only linked digests.
+    """
     pool = await get_pool()
     conditions = ["d.embedding IS NOT NULL"]
     params: list = [query_embedding, limit]
     idx = 3
+    if linked_only is True:
+        conditions.append("d.linked_case_law_id IS NOT NULL")
+    elif linked_only is False:
+        conditions.append("d.linked_case_law_id IS NULL")
     if practice_area:
         conditions.append(f"d.practice_area = ${idx}")
         params.append(practice_area)

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

@@ -381,6 +381,77 @@ async def link_digest(digest_id: UUID | str, case_law_id: UUID | str) -> dict:
     }
 
 
+async def case_digest_radar(
+    case_number: str,
+    limit: int = 5,
+    min_score: float = 0.45,
+) -> dict:
+    """Case-contextual digest radar (X12) — the chair-facing "שים לב" lead.
+
+    Surfaces UNLINKED digests (``linked_case_law_id IS NULL`` — rulings we don't hold
+    yet) whose TOPIC is semantically close to THIS case, so a relevant ruling we
+    only know about via a digest doesn't fall through the cracks while the case is
+    being decided. The case topic is built from title + appeal_subtype + the analyst's
+    extracted claims, embedded once, and matched against unlinked digests. Each lead
+    carries the underlying ruling's gap status + a suggested action.
+
+    INV-DIG1: this is RADAR — the digest is never cited; the lead points at the
+    underlying *ruling* (fetch / upload / link), never the digest itself. Read-only.
+    """
+    case = await db.get_case_by_number(case_number)
+    if not case:
+        return {"status": "case_not_found", "case_number": case_number, "leads": [], "count": 0}
+    case_id = case["id"]
+    if isinstance(case_id, str):
+        case_id = UUID(case_id)
+    parts = [case.get("title") or "", case.get("appeal_subtype") or ""]
+    try:
+        claims = await db.get_claims(case_id)
+        parts += [(c.get("claim_text") or "") for c in claims[:20]]
+    except Exception as e:  # noqa: BLE001 — claims are optional enrichment
+        logger.warning("case_digest_radar: get_claims failed for %s: %s", case_number, e)
+    query = " ".join(p for p in parts if p).strip()
+    if not query:
+        return {"status": "no_topic", "case_number": case_number, "leads": [], "count": 0}
+
+    vec = (await embeddings.embed_texts([query], input_type="query"))[0]
+    # Over-fetch, then filter by score + cap — semantic relevance, domain-agnostic
+    # (a cross-domain ruling on the same issue is still a valid lead).
+    hits = await db.search_digests_semantic(vec, limit=max(limit * 4, 20), linked_only=False)
+
+    leads: list[dict] = []
+    for h in hits:
+        score = float(h.get("score", 0) or 0)
+        if score < min_score:
+            continue
+        cit = (h.get("underlying_citation") or "").strip()
+        gap = await db.find_missing_precedent_by_citation(cit) if cit else None
+        in_corpus = await db.find_case_law_by_citation_fuzzy(cit) if cit else None
+        if in_corpus:
+            action = "available_link"   # ruling actually IS in the corpus → just link the digest
+        elif gap and (gap.get("status") in ("uploaded", "closed")):
+            action = "fetched"          # already obtained
+        elif gap:
+            action = "gap_open"         # flagged as missing — can request a fetch
+        else:
+            action = "new_lead"         # not even flagged yet — the highest-value alert
+        leads.append({
+            "digest_id": str(h["id"]),
+            "yomon_number": h.get("yomon_number"),
+            "headline": h.get("headline_holding") or h.get("summary") or "",
+            "underlying_citation": cit,
+            "underlying_court": h.get("underlying_court") or "",
+            "score": round(score, 3),
+            "missing_precedent_id": str(gap["id"]) if gap else None,
+            "missing_precedent_status": gap.get("status") if gap else None,
+            "action": action,
+        })
+        if len(leads) >= limit:
+            break
+    return {"status": "ok", "case_number": case_number,
+            "topic_query": query[:200], "leads": leads, "count": len(leads)}
+
+
 async def relink_digest(digest_id: UUID | str) -> dict:
     """Re-run autolink for an unlinked digest. No-op if already linked / no match."""
     digest = await db.get_digest(digest_id)

mcp-server/src/legal_mcp/tools/digests.py

@@ -170,3 +170,30 @@ async def digest_process_pending(limit: int = 20) -> str:
     except Exception as e:
         return _err(str(e))
     return _ok(result)
+
+
+async def digest_radar(case_number: str, limit: int = 5, min_score: float = 0.45) -> str:
+    """רדאר-יומונים הקשרי-לתיק (X12) — "שים לב" ליו"ר.
+
+    מחזיר יומונים **לא-מקושרים** (פס"ד שעוד אין לנו בקורפוס) שהנושא שלהם קרוב
+    סמנטית לתיק הזה — כדי שפס"ד רלוונטי שמוכר רק דרך יומון לא ייפול בין הכיסאות
+    בזמן הכרעת התיק. נושא-התיק נבנה מ-title + appeal_subtype + טענות-הצדדים, ומותאם
+    מול יומונים לא-מקושרים. כל ליד נושא את מראה-המקום של הפס"ד המקורי, סטטוס-הפער
+    (new_lead / gap_open / fetched / available_link) ופעולה מוצעת.
+
+    INV-DIG1: זהו radar — היומון לעולם אינו מצוטט; הליד מצביע על **הפס"ד** (להזמין
+    משיכה / להעלות / לקשר), לא על היומון. read-only.
+
+    Args:
+        case_number: מספר התיק (למשל "8124-09-24").
+        limit: מספר לידים מקסימלי.
+        min_score: סף-דמיון סמנטי (0-1) לסינון רעש.
+    """
+    if not case_number.strip():
+        return _err("case_number חובה")
+    try:
+        result = await digest_library.case_digest_radar(
+            case_number.strip(), limit=max(1, int(limit)), min_score=float(min_score))
+    except Exception as e:
+        return _err(str(e))
+    return _ok(result)

web/app.py

@@ -6444,6 +6444,17 @@ class DigestLinkRequest(BaseModel):
     case_law_id: str
 
 
+@app.get("/api/cases/{case_number}/digest-radar")
+async def api_case_digest_radar(case_number: str, limit: int = 5, min_score: float = 0.45):
+    """Case-contextual digest radar (X12) — UNLINKED digests whose topic is close to
+    this case (rulings we don't hold yet). Powers the case-page "📡 רדאר יומונים" lead
+    so a relevant ruling known only via a digest doesn't fall through the cracks while
+    the case is decided. INV-DIG1: points at the underlying ruling, never cites the
+    digest."""
+    return await digest_service.case_digest_radar(
+        case_number, limit=max(1, min(int(limit), 20)), min_score=float(min_score))
+
+
 @app.post("/api/digests/upload")
 async def digest_upload(
     file: UploadFile = File(...),