feat(digests): calibrate case radar to the analyst's distilled issues + per-issue attribution

The radar's query was built from dozens of raw claims (procedural-heavy noise), so
matches were thematic but imprecise and gave no reason WHY a lead is relevant. Now:

- Prefer the analyst's distilled legal_arguments (argument_title + legal_topic — one
  crisp CREAC issue per row) over raw claims.
- Search EACH issue separately and MERGE, so every lead is attributed to the case
  issue(s) it answers (`matched_issues`) — the chair sees "this ruling is for your
  'זכות עמידה' issue", not just a blended score.
- Fall back to the raw-claims blended query pre-aggregation; `source` reports the path.
- Shared `_radar_enrich` helper (gap status + action + matched_issues), bounded to 25
  issues to cap the per-issue fan-out.

Validated: 8124-09-24 (32 args → per-issue) surfaces betterment rulings each tagged to
its issue (היעדר השבחה / זהות הנישום / סעיף 7(ב)); 1044-03-26 (0 args) falls back to
claims unchanged. No tool/endpoint signature change (new fields pass through the dict).

Invariants: G2 (reuses the one digest search + arg accessor), INV-DIG1 (radar only).
No schema change.

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

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

@@ -381,6 +381,35 @@ async def link_digest(digest_id: UUID | str, case_law_id: UUID | str) -> dict:
     }
 
 
+async def _radar_enrich(h: dict, score: float, matched_issues: list[str]) -> dict:
+    """Shape one radar hit into a chair lead: gap status + suggested action +
+    which case ISSUE(s) it answers. The action points at the underlying RULING,
+    never the digest (INV-DIG1)."""
+    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
+    return {
+        "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),
+        "matched_issues": matched_issues,
+        "missing_precedent_id": str(gap["id"]) if gap else None,
+        "missing_precedent_status": gap.get("status") if gap else None,
+        "action": action,
+    }
+
+
 async def case_digest_radar(
     case_number: str,
     limit: int = 5,
@@ -389,11 +418,16 @@ async def case_digest_radar(
     """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.
+    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 decided.
+
+    Query calibration: prefer the analyst's DISTILLED legal arguments (one crisp issue
+    per row — ``argument_title`` + ``legal_topic``) over the dozens of raw claims. Each
+    issue is searched separately and the leads are MERGED, so every lead is attributed
+    to the case issue(s) it answers (``matched_issues``) — far higher precision than
+    one blended query of noisy claims. Falls back to raw claims pre-aggregation
+    (``source`` reports which path ran). 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.
@@ -404,52 +438,60 @@ async def case_digest_radar(
     case_id = case["id"]
     if isinstance(case_id, str):
         case_id = UUID(case_id)
-    parts = [case.get("title") or "", case.get("appeal_subtype") or ""]
+    ctx = " ".join(x for x in [case.get("title") or "", case.get("appeal_subtype") or ""] if x).strip()
+
+    # Preferred source: the analyst's distilled CREAC issues (one per legal_argument).
+    issues: list[tuple[str, str]] = []  # (label, embed_text)
     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}
+        from legal_mcp.services import argument_aggregator
+        for a in await argument_aggregator.get_legal_arguments(case_id):
+            label = (a.get("argument_title") or a.get("legal_topic") or "").strip()
+            topic = (a.get("legal_topic") or "").strip()
+            body = f"{label}. {topic}".strip(". ").strip()
+            if body:
+                issues.append((label, f"{ctx} {body}".strip()))
+    except Exception as e:  # noqa: BLE001 — arguments are optional; fall back to claims
+        logger.warning("case_digest_radar: get_legal_arguments failed for %s: %s", case_number, e)
 
-    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)
+    merged: dict[str, dict] = {}  # digest_id -> {"hit", "best", "issues": set}
+    if issues:
+        source = "legal_arguments"
+        issues = issues[:25]  # already distilled — bound the per-issue search fan-out
+        vecs = await embeddings.embed_texts([t for _, t in issues], input_type="query")
+        for (label, _), vec in zip(issues, vecs):
+            for h in await db.search_digests_semantic(vec, limit=6, linked_only=False):
+                s = float(h.get("score", 0) or 0)
+                if s < min_score:
+                    continue
+                m = merged.setdefault(str(h["id"]), {"hit": h, "best": s, "issues": set()})
+                m["best"] = max(m["best"], s)
+                if label:
+                    m["issues"].add(label)
+    else:
+        # Fallback: blended query from raw claims (pre-aggregation), one search.
+        source = "claims"
+        parts = [ctx]
+        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
+            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,
+                    "source": source, "leads": [], "count": 0}
+        vec = (await embeddings.embed_texts([query], input_type="query"))[0]
+        for h in await db.search_digests_semantic(vec, limit=max(limit * 4, 20), linked_only=False):
+            s = float(h.get("score", 0) or 0)
+            if s < min_score:
+                continue
+            merged.setdefault(str(h["id"]), {"hit": h, "best": s, "issues": set()})
 
-    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)}
+    ordered = sorted(merged.values(), key=lambda m: -m["best"])[:max(1, limit)]
+    leads = [await _radar_enrich(m["hit"], m["best"], sorted(m["issues"])) for m in ordered]
+    return {"status": "ok", "case_number": case_number, "source": source,
+            "issues_used": [lbl for lbl, _ in issues] if source == "legal_arguments" else None,
+            "leads": leads, "count": len(leads)}
 
 
 async def relink_digest(digest_id: UUID | str) -> dict: