"""Orchestrator for the Digests radar (X12). A digest ("כל יום" daily one-pager) is a SECONDARY source that POINTS at a ruling — it is never cited in a decision (INV-DIG1) and never enters the precedent/halacha pipeline (INV-DIG2). Ingest reuses only ATOMIC services (extract_text, embeddings), NOT the canonical ``ingest.ingest_document``. Two intake paths share one enrichment core: - ``ingest_digest`` (local/MCP, e.g. batch script) — does everything synchronously: stage → extract_text → create → LLM enrich → embed → autolink → completed. - ``create_pending_digest`` (CONTAINER-SAFE — the web upload) — stage → extract_text → create row with status='pending'. No LLM, no embedding. ``process_pending_digests`` (local/MCP) drains the queue and enriches. claude_session rule: ``digest_metadata_extractor`` (local CLI) is imported LAZILY inside the enrichment core only, so this module stays import-safe from the FastAPI container for create_pending / search / list / link / delete (DB + voyage only — voyage embedding only runs in the local enrich path). """ from __future__ import annotations import logging from datetime import date from pathlib import Path from typing import Awaitable, Callable from uuid import UUID from legal_mcp import config from legal_mcp.services import db, embeddings, extractor, ingest, storage logger = logging.getLogger(__name__) ProgressCb = Callable[[str, int, str], Awaitable[None]] DIGEST_LIBRARY_DIR = Path(config.DATA_DIR) / "digests" _VALID_PRACTICE_AREAS = frozenset( {"", "rishuy_uvniya", "betterment_levy", "compensation_197"} ) async def _noop_progress(_status: str, _percent: int, _msg: str) -> None: return None def _coerce_date(v) -> date | None: if v is None or v == "": return None if isinstance(v, date): return v if isinstance(v, str): try: return date.fromisoformat(v[:10]) except ValueError: return None return None def _embedding_text(row: dict) -> str: """The single vector indexes the digest as an atomic discovery unit.""" parts = [ row.get("concept_tag", ""), row.get("headline_holding", ""), row.get("summary", ""), row.get("analysis_text", ""), ] return "\n".join(p for p in parts if p).strip() async def try_autolink(digest_id: UUID | str, underlying_citation: str) -> str | None: """Best-effort link of a digest to the underlying ruling in case_law (INV-DIG3). Returns the case_law_id (str) if linked, else None. Never raises.""" citation = (underlying_citation or "").strip() if not citation: return None try: match = await db.find_case_law_by_citation_fuzzy(citation) except Exception as e: logger.warning("digest try_autolink lookup failed for %r: %s", citation, e) return None if not match: # Gap (INV-DIG3): the underlying ruling isn't in the corpus. Surface it — # never drop silently (INV-CF2). Court verdicts (supreme/admin) get an X13 # auto-fetch job; ועדת-ערר / unknown — which נט-המשפט can't serve — get a # missing_precedent the chair sees on /missing-precedents (#136). Never # raises. await _handle_unlinked_citation(digest_id, citation) return None await db.link_digest_to_case_law(digest_id, match["id"]) return str(match["id"]) async def _handle_unlinked_citation(digest_id: UUID | str, citation: str) -> None: """Surface an unlinked digest citation — auto-fetch if possible, else record a missing_precedent. Closes the silent-drop gap (#136, INV-DIG3/CF2). Routing via the ONE canonical classifier (``court_citation.classify``): * supreme/admin → ``court_fetch_jobs`` (drained by X13; on fetch failure the orchestrator opens its own missing_precedent, so no double-record here). * skip (ערר/בל"מ) / unknown → ``missing_precedents`` (needs Nevo / manual; נט-המשפט can't serve it). Deduped designator-aware via citation_norm (#143) so re-runs and overlaps don't pile up. """ try: from legal_mcp.services import court_citation cit = court_citation.classify(citation) if cit.tier in ("supreme", "admin"): await db.court_fetch_job_upsert( case_number_norm=cit.case_number_norm, citation_raw=citation, tier=cit.tier, court=cit.court_prefix, digest_id=UUID(str(digest_id)), ) logger.info("digest %s: enqueued court-fetch for %r (tier=%s)", digest_id, citation, cit.tier) return # Non-fetchable (ערר/בל"מ/unknown) — open a missing_precedent gap so it's # visible and actionable instead of vanishing. Dedup first (#143). if await db.find_missing_precedent_by_citation(citation): return digest = await db.get_digest(digest_id) yomon = (digest or {}).get("yomon_number") or "" note = (f"זוהה דרך יומון מס' {yomon} (digest_id={digest_id})" if yomon else f"זוהה דרך יומון (digest_id={digest_id})") await db.create_missing_precedent( citation=citation, discovery_source="digest", notes=note, ) logger.info("digest %s: opened missing_precedent for %r (tier=%s)", digest_id, citation, cit.tier) except Exception as e: # never break digest ingest logger.warning("digest unlinked-citation handling failed for %r: %s", citation, e) # ── Container-safe creation (web upload) — no LLM, no embedding ────── async def create_pending_digest( *, file_path: str | Path, yomon_number: str = "", digest_date: date | str | None = None, practice_area: str = "", appeal_subtype: str = "", subject_tags: list[str] | None = None, progress: ProgressCb | None = None, ) -> dict: """Stage the file, extract text (PyMuPDF — container-safe), and create a digest row with extraction_status='pending'. The LLM metadata extraction, embedding, and autolink are deferred to ``process_pending_digests`` (local). Returns {status, digest_id, extraction_status} or {status:'exists', ...}. Idempotent on content_hash (INV-G3). """ progress = progress or _noop_progress if practice_area and practice_area not in _VALID_PRACTICE_AREAS: raise ValueError(f"invalid practice_area: {practice_area!r}") src = Path(file_path) if not src.exists(): raise ValueError(f"file not found: {file_path}") await progress("staging", 10, "מעתיק קובץ") # ``_stage_file`` returns the storage KEY (DATA_DIR-relative path). Resolve a # real local path to read from — on s3-only this downloads to a temp file we # own and remove after extraction (INV-STG1; the key is not guaranteed to be # on local disk). rel_path = await ingest._stage_file(src, DIGEST_LIBRARY_DIR, "incoming") local = await storage.ensure_local(rel_path, bucket=storage.Bucket.DOCUMENTS) local_is_tmp = storage.local_path(rel_path, bucket=storage.Bucket.DOCUMENTS) is None await progress("extracting_text", 50, "מחלץ טקסט") try: raw_text, _pc, _off = await extractor.extract_text(str(local)) finally: if local_is_tmp: try: local.unlink(missing_ok=True) except OSError as e: # noqa: BLE001 — temp cleanup, never fatal logger.debug("could not remove temp digest file %s: %s", local, e) raw_text = (raw_text or "").strip() if not raw_text: raise ValueError("no text extracted from digest") content_hash = db._content_hash(raw_text) existing = await db.get_digest_by_content_hash(content_hash) if existing: await progress("completed", 100, "יומון זהה כבר קיים") return {"status": "exists", "digest_id": existing["id"], "extraction_status": existing.get("extraction_status")} record = await db.create_digest( analysis_text=raw_text, yomon_number=yomon_number.strip(), digest_date=_coerce_date(digest_date), practice_area=practice_area, appeal_subtype=appeal_subtype.strip(), subject_tags=list(subject_tags) if subject_tags else [], source_document_path=rel_path, extraction_status="pending", ) await progress("queued", 100, "ממתין לעיבוד מקומי (LLM)") return {"status": "pending", "digest_id": record["id"], "extraction_status": "pending"} # ── Local enrichment core (LLM + embed + autolink) ────────────────── async def enrich_digest(digest_id: UUID | str, progress: ProgressCb | None = None) -> dict: """Run LLM metadata extraction over a digest's analysis_text, fill ONLY empty fields (preserve user-supplied values), embed, autolink, complete. **MCP-tool-only path** (uses the local LLM extractor). Idempotent. """ progress = progress or _noop_progress row = await db.get_digest(digest_id) if not row: raise ValueError("digest not found") analysis = (row.get("analysis_text") or "").strip() if not analysis: await db.update_digest(digest_id, extraction_status="failed") return {"status": "no_text", "digest_id": str(digest_id)} await db.update_digest(digest_id, extraction_status="processing") await progress("extracting_metadata", 40, "מחלץ מטא-דאטה (LLM)") from legal_mcp.services import digest_metadata_extractor extracted = await digest_metadata_extractor.extract(analysis) # Fill only empty fields (preserve user-supplied values from the form). fields: dict = {} for key in ("yomon_number", "concept_tag", "headline_holding", "summary", "underlying_citation", "underlying_court", "underlying_judge", "practice_area", "appeal_subtype"): if not (row.get(key) or "").strip() and extracted.get(key): fields[key] = extracted[key] if row.get("digest_date") is None and extracted.get("digest_date"): fields["digest_date"] = extracted["digest_date"] if row.get("underlying_date") is None and extracted.get("underlying_date"): fields["underlying_date"] = extracted["underlying_date"] if not (row.get("subject_tags") or []) and extracted.get("subject_tags"): fields["subject_tags"] = extracted["subject_tags"] # digest_kind classifies the issue (decision vs announcement). A successful # extraction (any field returned) must end with a non-empty kind — that is the # signal the drain self-heal uses to tell "enriched" from "failed". If the # model omitted it, infer: a ruling citation → decision, else announcement. if extracted and not (row.get("digest_kind") or "").strip(): kind = extracted.get("digest_kind") if kind not in ("decision", "announcement", "other"): cite = fields.get("underlying_citation") or row.get("underlying_citation") or "" kind = "decision" if cite.strip() else "announcement" fields["digest_kind"] = kind if fields: try: await db.update_digest(digest_id, **fields) except Exception as e: # The same yomon issue can arrive as two different PDFs (re-sent / # forwarded twice → different bytes → content_hash dedup misses it), # but the yomon_number is unique. The extracted number then collides # on uq_digests_yomon_number. This row is a duplicate of an already- # ingested yomon → drop it so it isn't retried forever by the cron. if "uq_digests_yomon_number" in str(e): await db.delete_digest(digest_id) logger.info( "digest %s is a duplicate yomon (%s) — deleted", digest_id, fields.get("yomon_number"), ) return {"status": "duplicate", "digest_id": str(digest_id), "yomon_number": fields.get("yomon_number")} raise merged = await db.get_digest(digest_id) await progress("embedding", 75, "מחשב embedding") emb_text = _embedding_text(merged) if emb_text: try: vecs = await embeddings.embed_texts([emb_text], input_type="document") if vecs: await db.store_digest_embedding(digest_id, vecs[0]) except Exception as e: # surfaced, not swallowed (§6) logger.warning("digest embedding failed for %s: %s", digest_id, e) await progress("linking", 90, "מנסה לקשר לפסק המקורי") linked_id = None if not merged.get("linked_case_law_id"): linked_id = await try_autolink(digest_id, merged.get("underlying_citation", "")) await db.update_digest(digest_id, extraction_status="completed") await progress("completed", 100, "הושלם") return { "status": "completed", "digest_id": str(digest_id), "yomon_number": merged.get("yomon_number", ""), "underlying_citation": merged.get("underlying_citation", ""), "linked_case_law_id": merged.get("linked_case_law_id") or linked_id, "fields_filled": sorted(fields.keys()), } async def process_pending_digests(limit: int = 20) -> dict: """Drain the digest extraction queue (rows stamped extraction_status='pending' by the web upload). Local/MCP only — runs the LLM enrichment per row. Sequential (avoids LLM rate-limit storms), mirrors process_pending_extractions.""" pending = await db.list_pending_digests(limit=limit) if not pending: return {"status": "no_pending", "processed": 0, "results": []} results = [] processed = 0 for row in pending: did = row["id"] try: res = await enrich_digest(did) processed += 1 results.append({"digest_id": str(did), "status": res.get("status"), "linked": bool(res.get("linked_case_law_id"))}) except Exception as e: logger.exception("process_pending_digests failed for %s: %s", did, e) try: await db.update_digest(did, extraction_status="failed") except Exception: logger.exception("could not mark digest %s failed", did) results.append({"digest_id": str(did), "status": "failed", "error": str(e)}) return {"status": "completed", "processed": processed, "total_pending": len(pending), "results": results} # ── Full synchronous ingest (local/MCP, e.g. batch script) ────────── async def ingest_digest( *, file_path: str | Path, yomon_number: str = "", digest_date: date | str | None = None, practice_area: str = "", appeal_subtype: str = "", subject_tags: list[str] | None = None, progress: ProgressCb | None = None, ) -> dict: """Ingest one digest synchronously. **MCP-tool-only** (uses the LLM). Creates the row (with any user-supplied values) then enriches in place. Idempotent on content_hash (INV-G3). """ progress = progress or _noop_progress created = await create_pending_digest( file_path=file_path, yomon_number=yomon_number, digest_date=digest_date, practice_area=practice_area, appeal_subtype=appeal_subtype, subject_tags=subject_tags, progress=progress, ) if created.get("status") == "exists": return created digest_id = created["digest_id"] enriched = await enrich_digest(digest_id, progress=progress) return enriched # ── Linking (INV-DIG3) ────────────────────────────────────────────── async def link_digest(digest_id: UUID | str, case_law_id: UUID | str) -> dict: """Manually link a digest to an underlying ruling (INV-DIG3). Idempotent.""" digest = await db.get_digest(digest_id) if not digest: raise ValueError("digest not found") ruling = await db.get_case_law( case_law_id if isinstance(case_law_id, UUID) else UUID(str(case_law_id)) ) if not ruling: raise ValueError("case_law not found") updated = await db.link_digest_to_case_law(digest_id, case_law_id) return { "linked": True, "digest_id": str(digest_id), "case_law_id": str(case_law_id), "case_number": ruling.get("case_number"), "digest": updated, } 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, 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 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. """ 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) 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: 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) 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()}) 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: """Re-run autolink for an unlinked digest. No-op if already linked / no match.""" digest = await db.get_digest(digest_id) if not digest: raise ValueError("digest not found") if digest.get("linked_case_law_id"): return {"linked": True, "digest_id": str(digest_id), "case_law_id": digest["linked_case_law_id"], "changed": False} linked_id = await try_autolink(digest_id, digest.get("underlying_citation", "")) return { "linked": linked_id is not None, "digest_id": str(digest_id), "case_law_id": linked_id, "changed": linked_id is not None, } async def unlink_digest(digest_id: UUID | str) -> dict: """Clear a digest's link to the underlying ruling.""" updated = await db.link_digest_to_case_law(digest_id, None) if updated is None: raise ValueError("digest not found") return {"unlinked": True, "digest_id": str(digest_id)} # ── Read / search (container-safe: DB + voyage) ───────────────────── async def search_digests( query: str, practice_area: str = "", subject_tag: str = "", concept_tag: str = "", limit: int = 10, ) -> list[dict]: """Semantic search over the digests radar. Container-safe (voyage + DB).""" if not query.strip(): return [] query_vec = await embeddings.embed_query(query) return await db.search_digests_semantic( query_embedding=query_vec, practice_area=practice_area, subject_tag=subject_tag, concept_tag=concept_tag, limit=limit, ) async def get_digest(digest_id: UUID | str) -> dict | None: return await db.get_digest(digest_id) async def list_digests( practice_area: str = "", concept_tag: str = "", linked: bool | None = None, search: str = "", publication: str = "", limit: int = 100, offset: int = 0, ) -> list[dict]: return await db.list_digests( practice_area=practice_area, concept_tag=concept_tag, linked=linked, search=search, publication=publication, limit=limit, offset=offset, ) async def update_digest(digest_id: UUID | str, **fields) -> dict | None: return await db.update_digest(digest_id, **fields) async def delete_digest(digest_id: UUID | str) -> bool: return await db.delete_digest(digest_id)