diff --git a/mcp-server/src/legal_mcp/services/digest_library.py b/mcp-server/src/legal_mcp/services/digest_library.py index dc310e7..13ecaf8 100644 --- a/mcp-server/src/legal_mcp/services/digest_library.py +++ b/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: