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

"""Citation-verification view (X11 Phase 2 / #154) — the chair's "אימות פסיקה" tab.

Assembles, per legal ARGUMENT of a case, the supporting precedents the chair should
verify before the writer cites them:

  • in-corpus suggestions — per-issue semantic retrieval over the authoritative
    precedent library (``search_library``), each carrying the cumulative authority
    signal (``cited_by``: followed/distinguished — db.citation_authority, X11).
  • attached/verified state — any ``case_precedents`` row already attached to the
    argument (verified flag + chair_note), merged onto the matching suggestion.
  • radar — UNLINKED digests relevant to the same issue (rulings we don't hold yet),
    from ``case_digest_radar`` grouped by matched issue.

Pure read/assembly — never writes, never cites (INV-DIG1/INV-AH). The chair verifies
through ``db.set_case_precedent_verified`` / attach; the writer consumes only verified
rows. Reuses the one corpus search + the one authority query + the one radar (G2).
"""

from __future__ import annotations

import asyncio
import logging
from uuid import UUID

from legal_mcp.services import (
    argument_aggregator,
    db,
    digest_library,
    precedent_library,
)

logger = logging.getLogger(__name__)

_SUGGEST_PER_ISSUE = 4
_SUGGEST_FLOOR = 0.45


async def build_view(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": []}
    case_id = case["id"]
    if isinstance(case_id, str):
        case_id = UUID(case_id)
    ctx = " ".join(x for x in [case.get("title") or "", case.get("appeal_subtype") or ""] if x).strip()

    args = await argument_aggregator.get_legal_arguments(case_id)

    # Attached precedents already on the case → grouped by argument_id, keyed by
    # the resolved corpus ruling so we can merge verify-state onto a suggestion.
    attached = await db.list_case_precedents(case_id)
    attached_by_arg: dict[str, dict[str, dict]] = {}
    for p in attached:
        aid = str(p.get("argument_id") or "")
        clid = str(p.get("case_law_id") or "")
        if aid and clid:
            attached_by_arg.setdefault(aid, {})[clid] = p

    # Radar (unlinked digests) once, grouped by the issue label it matched.
    radar_by_issue: dict[str, list[dict]] = {}
    try:
        radar = await digest_library.case_digest_radar(case_number, limit=12, min_score=0.42)
        for lead in radar.get("leads", []):
            for label in (lead.get("matched_issues") or ["—"]):
                radar_by_issue.setdefault(label, []).append(lead)
    except Exception as e:  # noqa: BLE001 — radar is best-effort
        logger.warning("citation_verification radar failed for %s: %s", case_number, e)

    async def _fetch(a: dict) -> tuple[list[dict], dict]:
        """Per-argument corpus search + batched authority — run concurrently.

        Each call is one Voyage embed + one vector search (+ one batched authority
        query); independent across arguments, so they fan out rather than waterfall.
        """
        title = (a.get("argument_title") or "").strip()
        topic = (a.get("legal_topic") or "").strip()
        query = f"{ctx} {title}. {topic}".strip()
        hits: list[dict] = []
        try:
            hits = await precedent_library.search_library(
                query=query, limit=_SUGGEST_PER_ISSUE, include_halachot=True)
        except Exception as e:  # noqa: BLE001
            logger.warning("citation_verification search failed (%s): %s", title[:30], e)
        # Resolve the authority breakdown for the hit set in one batched query.
        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]
        authority = await db.citation_authority(clids) if clids else {}
        return hits, authority

    # Fan out the expensive per-argument retrieval concurrently — was a sequential
    # waterfall (N args × Voyage embed + vector search each). gather preserves order.
    fetched = await asyncio.gather(*(_fetch(a) for a in args)) if args else []

    out_args: list[dict] = []
    n_verified = 0
    for a, (hits, authority) in zip(args, fetched):
        aid = str(a["id"])
        title = (a.get("argument_title") or "").strip()
        topic = (a.get("legal_topic") or "").strip()

        seen: set[str] = set()
        supporting: list[dict] = []
        for h in hits:
            clid = str(h.get("case_law_id") or "")
            if not clid or clid in seen:
                continue
            if float(h.get("score", 0) or 0) < _SUGGEST_FLOOR:
                continue
            seen.add(clid)
            att = attached_by_arg.get(aid, {}).get(clid)
            if att and att.get("verified"):
                n_verified += 1
            supporting.append({
                "case_law_id": clid,
                "case_number": h.get("case_number") or "",
                "case_name": h.get("case_name") or "",
                "quote": h.get("supporting_quote") or h.get("rule_statement") or "",
                "score": round(float(h.get("score", 0) or 0), 3),
                "cited_by": authority.get(clid, {"total": 0, "positive": 0,
                                                 "negative": 0, "unclassified": 0,
                                                 "by_treatment": {}}),
                "attached_id": str(att["id"]) if att else None,
                "verified": bool(att.get("verified")) if att else False,
                "chair_note": (att.get("chair_note") or "") if att else "",
            })

        out_args.append({
            "argument_id": aid,
            "title": title,
            "legal_topic": topic,
            "priority": a.get("priority") or "",
            "party": a.get("party") or "",
            "supporting": supporting,
            "radar": radar_by_issue.get(title, []),
        })

    return {
        "status": "ok",
        "case_number": case_number,
        "arguments": out_args,
        "summary": {
            "arguments_total": len(out_args),
            "arguments_with_support": sum(1 for x in out_args if x["supporting"]),
            "verified": n_verified,
            "radar_leads": sum(len(x["radar"]) for x in out_args),
        },
    }