phase1/scripts/serve.py

"""Moonley Phase-1 fallback serving backend (Thor, bound to the tailnet).

  GET /                       -> v2 frontend (results + judgment views)
  GET /api/search_stream      -> reviewer-IMPROVED retrieval, STREAMED stepwise (SSE):
                                 emits live step events (search -> rerank -> paralegal
                                 review -> drop -> bounded re-query -> good-law) then
                                 streams the grounded answer token-by-token.
  GET /api/search             -> same pipeline, non-streamed (fallback). {answer, results[], steps}
  GET /api/judgment?id=       -> full judgment: metadata + verbatim issue/held + citator +
                                 dark good-law (provenance) + reassembled text
  GET /api/ask_judgment?id=&q= -> grounded Q&A over a single judgment

Retrieval: dense (BGE bf16) + cross-encoder rerank. Answer / paralegal-review / ask:
The answer model uses the key in the local service environment. The local Qwen model is reserved
for the citator Tier-2 batch (whole-doc treatment), NOT serving. Good-law: DARK
(overruled/doubted/per_incuriam/unknown).
Run:  uvicorn serve:app --host 0.0.0.0 --port 8000
"""
import json, os, re
from collections import defaultdict, Counter
import numpy as np, torch, requests
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse, StreamingResponse, FileResponse
from sentence_transformers import SentenceTransformer, CrossEncoder

HERE = os.path.dirname(os.path.abspath(__file__))
DEV = "cuda" if torch.cuda.is_available() else "cpu"
CAND, BGE_Q = 40, "Represent this sentence for searching relevant passages: "

# ============================ PILOT LOGGING ============================
# Pilot stage (George, 2026-06): capture as much as we can to improve the product for lawyers.
# Privacy gating is intentionally OFF — the ONLY hard rule is that secrets (the DeepSeek API key /
# Clerk bearer token) must NEVER land in a log. Two append-only JSONL streams on Thor, joined by req_id:
#   usage-YYYY-MM-DD.jsonl     one line per user-facing request (who ran what, RAW query, judgments opened)
#   internal-YYYY-MM-DD.jsonl  per-DeepSeek-call FULL prompt+response + latency/tokens, plus surfaced errors
# Logging is fire-and-forget on a daemon thread: it never blocks the SSE stream and can never crash a request.
import threading, queue, uuid, time, contextvars
from contextlib import contextmanager
from datetime import datetime, timezone

LOG_DIR   = os.environ.get("MOONLEY_LOG_DIR") or os.environ.get("THEMIS_LOG_DIR") or os.path.join(os.path.expanduser("~"), "moonley", "logs")
LOG_FULL  = os.environ.get("THEMIS_LOG_PROMPTS", "1") != "0"   # full DeepSeek prompt+response — ON by default for the pilot
REQ_ID    = contextvars.ContextVar("req_id", default="")
FN_LABEL  = contextvars.ContextVar("fn_label", default="")
USAGE_CTX = contextvars.ContextVar("usage_ctx", default=None)  # {claimed_user, client_ip, user_agent} from the gate

_LOG_Q = queue.Queue(maxsize=20000); _LOG_DROPPED = [0]
_SECRET_KEY = re.compile(r"^(authorization|deepseek_api_key|api[_-]?key|clerk_secret_key|x-api-key)$", re.I)
_SECRET_VAL = re.compile(r"Bearer\s+\S+|sk-[A-Za-z0-9]{8,}")
_LOGCTRL    = re.compile(r"[\x00-\x08\x0b\x0c\x0e-\x1f]")

def _scrub(d):                                  # defense-in-depth: secrets never reach disk even via an echoed error
    key = globals().get("DS_KEY") or ""
    out = {}
    for k, v in d.items():
        if _SECRET_KEY.match(k): continue        # never log a secret-named field
        if isinstance(v, str):
            v = _SECRET_VAL.sub("[REDACTED]", v)
            if key: v = v.replace(key, "[REDACTED]")
        out[k] = v
    return out

def _log_writer():
    while True:
        try:
            stream, obj = _LOG_Q.get()
            os.makedirs(LOG_DIR, mode=0o700, exist_ok=True)
            day = datetime.now(timezone.utc).strftime("%Y-%m-%d")
            fd = os.open(os.path.join(LOG_DIR, f"{stream}-{day}.jsonl"), os.O_CREAT | os.O_WRONLY | os.O_APPEND, 0o600)
            with os.fdopen(fd, "a", encoding="utf-8") as f:
                f.write(json.dumps(obj, ensure_ascii=False) + "\n")
        except Exception:
            pass
threading.Thread(target=_log_writer, daemon=True).start()

def log_event(stream_name, **fields):           # fire-and-forget; never raises into request code
    try:
        obj = {"ts": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), "req_id": REQ_ID.get()}
        obj.update(fields)
        _LOG_Q.put_nowait((stream_name, _scrub(obj)))
    except queue.Full:
        _LOG_DROPPED[0] += 1
    except Exception:
        pass

def _clip(s, n=20000):
    s = "" if s is None else str(s)
    s = _LOGCTRL.sub(" ", s)
    return s if len(s) <= n else s[:n] + "…"

@contextmanager
def fn(label):                                  # tags every DeepSeek call made within the block (no signature changes)
    tok = FN_LABEL.set(label)
    try: yield
    finally: FN_LABEL.reset(tok)

def _bind_ctx(it, ctx):
    """Iterate a streaming generator inside a FIXED context so REQ_ID/FN_LABEL set in the endpoint
    persist across yields — and into llm() called mid-stream. (Starlette would otherwise run each
    next() in a fresh context, losing the request id on the DeepSeek-call logs.)"""
    while True:
        try: yield ctx.run(next, it)
        except StopIteration: return
# ======================================================================

# --- DeepSeek (serving LLM) — key from .env, never logged ---
def _load_env(path):
    if os.path.exists(path):
        for ln in open(path):
            ln = ln.strip()
            if ln and not ln.startswith("#") and "=" in ln:
                k, v = ln.split("=", 1)
                os.environ.setdefault(k.strip(), v.strip().strip('"').strip("'"))
_load_env(os.path.join(HERE, ".env"))
from clerk_auth import (  # noqa: E402 - the local .env must be loaded first
    PUBLIC_PATHS,
    authenticate_clerk_request,
    cors_origins,
    frontend_auth_config,
)
DS_KEY = os.environ.get("DEEPSEEK_API_KEY", "")
DS_URL = "https://api.deepseek.com/chat/completions"
DS_HDR = {"Authorization": f"Bearer {DS_KEY}", "Content-Type": "application/json"}
DS_MODEL = "deepseek-chat"

def _ds_meta(label, t0, status, msgs, out, streamed):
    rec = {"lvl": "INFO", "stage": label, "fn": label, "ds_model": DS_MODEL, "ds_status": status,
           "ds_latency_ms": int((time.time() - t0) * 1000), "streamed": streamed,
           "prompt_chars": sum(len(m.get("content", "")) for m in msgs), "completion_chars": len(out)}
    if LOG_FULL:                                       # full prompt + response (pilot: max data)
        rec["prompt"] = [{"role": m.get("role"), "content": _clip(m.get("content"))} for m in msgs]
        rec["completion"] = _clip(out)
    return rec

def llm(msgs, max_new=256):
    t0 = time.time(); label = FN_LABEL.get() or "llm"
    try:
        r = requests.post(DS_URL, headers=DS_HDR, timeout=120,
                          json={"model": DS_MODEL, "messages": msgs, "max_tokens": max_new, "temperature": 0})
        r.raise_for_status()
        j = r.json(); out = j["choices"][0]["message"]["content"].strip(); u = j.get("usage") or {}
        rec = _ds_meta(label, t0, r.status_code, msgs, out, False)
        rec["prompt_tokens"] = u.get("prompt_tokens"); rec["completion_tokens"] = u.get("completion_tokens")
        rec["finish_reason"] = (j["choices"][0] or {}).get("finish_reason")
        log_event("internal", **rec)
        return out
    except Exception as e:
        log_event("internal", lvl="ERROR", stage=label, fn=label, ds_latency_ms=int((time.time() - t0) * 1000),
                  exc_type=type(e).__name__, exc_msg=_clip(str(e), 300), timed_out=isinstance(e, requests.exceptions.Timeout))
        raise

def llm_stream(msgs, max_new=256):
    t0 = time.time(); label = FN_LABEL.get() or "llm"; acc = []; status = None
    try:
        with requests.post(DS_URL, headers=DS_HDR, timeout=120, stream=True,
                           json={"model": DS_MODEL, "messages": msgs, "max_tokens": max_new,
                                 "temperature": 0, "stream": True}) as r:
            status = r.status_code; r.raise_for_status()
            for raw in r.iter_lines():
                if not raw:
                    continue
                ln = raw.decode("utf-8", "ignore")
                if not ln.startswith("data: "):
                    continue
                payload = ln[6:]
                if payload == "[DONE]":
                    break
                try:
                    delta = json.loads(payload)["choices"][0]["delta"].get("content")
                except Exception:
                    delta = None
                if delta:
                    acc.append(delta); yield delta
        log_event("internal", **_ds_meta(label, t0, status, msgs, "".join(acc), True))
    except Exception as e:
        log_event("internal", lvl="ERROR", stage=label, fn=label, ds_latency_ms=int((time.time() - t0) * 1000),
                  exc_type=type(e).__name__, exc_msg=_clip(str(e), 300), timed_out=isinstance(e, requests.exceptions.Timeout))
        raise

print("loading index...", flush=True)
chunks = [json.loads(l) for l in open("escr_chunks.jsonl")]
texts = [c["text"] for c in chunks]; chunk_doc = [c["doc_id"] for c in chunks]
M = np.load("escr_vectors.npy")
meta = {}; goodlaw = {}
for l in open("escr_meta.jsonl"): m = json.loads(l); meta[m["doc_id"]] = m
for l in open("good_law.jsonl"): g = json.loads(l); goodlaw[g["doc_id"]] = g
doc_chunks = defaultdict(list)
for i, d in enumerate(chunk_doc): doc_chunks[d].append(i)
nc2doc = {m.get("neutral_citation"): d for d, m in meta.items() if m.get("neutral_citation")}
NDOCS = len(meta)
# doc_id -> (year, path) for constructing the open-registry PDF URL (path lives in corpus_full, not meta)
pdfmap = {}
if os.path.exists("escr_pdfmap.jsonl"):
    for l in open("escr_pdfmap.jsonl"):
        try: r = json.loads(l); pdfmap[r["doc_id"]] = (str(r.get("year") or ""), r["path"])
        except Exception: pass
print(f"pdfmap: {len(pdfmap)} judgments have a source PDF", flush=True)

# --- source PDF cache: pull the authoritative SCR PDF from the open registry, keep a small LRU on
#     disk, serve it from OUR origin (so it embeds inline — no cross-origin iframe blocking) ---
PDF_BASE  = "https://indian-supreme-court-judgments.s3.ap-south-1.amazonaws.com"
PDF_CACHE = os.environ.get("THEMIS_PDF_CACHE") or os.path.join(HERE, "pdf_cache")
PDF_CACHE_MAX = int(os.environ.get("THEMIS_PDF_CACHE_MAX", "20"))
_PDF_LOCK = threading.Lock()
os.makedirs(PDF_CACHE, exist_ok=True)

def _pdf_evict():                                            # keep at most PDF_CACHE_MAX files (oldest-used go first)
    files = [os.path.join(PDF_CACHE, f) for f in os.listdir(PDF_CACHE) if f.endswith(".pdf")]
    if len(files) <= PDF_CACHE_MAX: return
    files.sort(key=lambda p: os.path.getmtime(p))
    for p in files[:len(files) - PDF_CACHE_MAX]:
        try: os.remove(p)
        except Exception: pass

def fetch_pdf(d):
    """Local cached path to doc d's source PDF; pull from the open registry on a miss.
    Returns (path, 'hit'|'miss') on success, or (None, reason)."""
    yp = pdfmap.get(d)
    if not yp: return None, "no_pdf"
    year, path = yp
    local = os.path.join(PDF_CACHE, path + "_EN.pdf")
    if os.path.exists(local):
        try: os.utime(local, None)                          # touch = mark recently used (LRU)
        except Exception: pass
        return local, "hit"
    url = f"{PDF_BASE}/data/pdf/year={year}/english/{path}_EN.pdf"
    try:
        r = requests.get(url, timeout=30)
        if r.status_code != 200 or r.content[:4] != b"%PDF":
            return None, f"upstream_{r.status_code}"
    except Exception:
        return None, "fetch_error"
    with _PDF_LOCK:
        tmp = local + ".tmp"
        with open(tmp, "wb") as f: f.write(r.content)
        os.replace(tmp, local)
        _pdf_evict()
    return local, "miss"

# --- Stage-3: citation-graph tools (edges.jsonl) + BM25 keyword index ---
from rank_bm25 import BM25Okapi
out_edges = defaultdict(list); in_edges = defaultdict(list); edge_meta = {}
cite_indeg = defaultdict(int)                       # CITE-edge in-degree only — the RELIABLE salience/display count
for _l in open("edges.jsonl"):
    _e = json.loads(_l); _f, _t = _e["from"], _e["target"]
    out_edges[_f].append(_t); in_edges[_t].append(_f)
    edge_meta[(_f, _t)] = {"treatment": _e.get("treatment"), "method": _e.get("method")}
    if _e.get("method") == "cite": cite_indeg[_t] += 1
print(f"graph: {len(edge_meta)} edges", flush=True)
_tok = lambda s: re.findall(r"[a-z0-9]+", s.lower())
bm25 = BM25Okapi([_tok(t) for t in texts])
print("BM25 index built", flush=True)
import difflib
name_vocab = set()                                 # case-name token vocabulary for fuzzy lookup (typo tolerance)
for _m in meta.values():
    for _w in re.findall(r"[a-z]+", (_m.get("case_name") or "").lower()):
        if len(_w) >= 4: name_vocab.add(_w)

# --- citation resolver (neutral + equivalent) — only RESOLVABLE cites become inline links ---
def norm_cite(c): return re.sub(r"\s+", " ", (c or "").replace(".", "")).strip().upper()  # dots stripped: S.C.R.==SCR
cite_resolver = {}
for _d, _m in meta.items():
    for _k in [_m.get("neutral_citation")] + (_m.get("equivalent_citations") or []):
        if _k: cite_resolver.setdefault(norm_cite(_k), _d)
CITE_RE = re.compile(r"\[\d{4}\]\s*\d+\s*S\.?C\.?R\.?\s*\d+|\(\d{4}\)\s*\d+\s*SCC\s*\d+|\d{4}\s+INSC\s+\d+|AIR\s+\d{4}\s+SC\s+\d+")

# --- deterministic reporter-chrome cleaner (DELETE-ONLY; raw is preserved for audit) ---
_CTRL     = re.compile(r"[\x00-\x08\x0b\x0c\x0e-\x1f]")                  # PDF/OCR control chars (e.g. \x08 wedged into running headers)
_MARGIN   = re.compile(r"(?m)^[ \t]*[A-H][ \t]*$")                       # reporter gutter letters A–H on their own line
_RUNHDR   = re.compile(r"\s*\d{1,4}\s+(?:\[\d{4}\]\s*\d+\s*S\.?C\.?R\.?[^A-Za-z]*)?Digital Supreme Court Reports\s*")
_CITELINE = re.compile(r"(?im)^[ \t]*(?:\[\d{4}\]\s*\d+\s*S\.?C\.?R\.?\s*\d+|\(\d{4}\)\s*\d+\s*SCC\s*\d+|\d{4}\s+INSC\s+\d+|AIR\s+\d{4}\s+SC\s+\d+)[ \t.:]*$")
_CSTART   = re.compile(r"(?im)^[ \t]*(?:\d{1,3}\.\s|Issue\s+for\s+Consideration|Head\s*notes?\b|IN THE SUPREME COURT|The appellants?\b|This appeal\b|These appeals\b|Leave granted\b|Heard\b)")
_DEHYPH   = re.compile(r"([A-Za-z])-\n[ \t]*([a-z])")                    # join words split by a line-break hyphen (judg-\nment)
_GUTTER   = re.compile(r" ([A-H]) ?\n")                                  # SCR right-margin gutter letter wedged at a line end ("…the D\n")

def clean_headnote(s):
    if not s: return s
    s = _CTRL.sub("", s)                # strip control chars so the running-header regex can match
    s = _RUNHDR.sub(" ", s)
    s = re.sub(r"\bHead\s*notes?\s*†?", "", s, flags=re.I)   # drop the "Headnotes†" marker
    s = s.replace("†", "")
    s = re.split(r"\*\s*Author\b", s)[0]                      # cut the reporter author/citation footer ("*Author [2024] 12 S.C.R. 1437 …")
    s = re.sub(r"^[\s:–—-]+", "", s)                          # leading ":" / dashes
    return re.sub(r"[ \t]{2,}", " ", s).strip()

def clean_judgment(raw):
    if not raw: return ""
    t = _CTRL.sub("", raw)              # strip PDF/OCR control chars first
    t = _MARGIN.sub("", t)              # drop margin letters
    t = _RUNHDR.sub(" ", t)             # kill the "130 [2024] 6 S.C.R. Digital Supreme Court Reports" triad anywhere
    t = _CITELINE.sub("", t)            # drop page-top citation-only echo lines
    # Prefer starting the opinion at its first numbered paragraph — eSCR formats these as a bare "1.\n"/"2.\n"
    # line — which drops the duplicated reporter headnote (already shown verbatim in the ISSUE/HELD cards; the
    # full unedited text remains one click away via "View raw text"). Fall back to the caption-preamble slice.
    mo = re.search(r"(?m)^[ \t]*1\.[ \t]*$", t) or re.search(r"(?m)^[ \t]*2\.[ \t]*$", t)
    if mo and mo.start() > 200:
        t = t[mo.start():]
    else:
        m = _CSTART.search(t[:2000])    # slice the duplicated reporter caption preamble, only if an anchor is found early
        if m: t = t[m.start():]
    t = _DEHYPH.sub(r"\1\2", t)         # repair hyphen-split words across line breaks
    t = _GUTTER.sub("\n", t)            # drop SCR right-margin gutter letters (A–H) wedged at line ends
    t = re.sub(r"[ \t]{2,}", " ", t)
    t = re.sub(r"\n{3,}", "\n\n", t)
    return t.strip()

def doc_links(d, text):
    out = {}
    for c in CITE_RE.findall(text):
        rid = cite_resolver.get(norm_cite(c))
        if rid and rid != d and c not in out: out[c] = rid
    return [{"cite": k, "id": v} for k, v in out.items()]

def passage_snippet(raw, n=300):
    """Search results show retrieval CHUNKS, which start mid-sentence. Clean reporter chrome
    and snap the start to a sentence/word boundary so the snippet reads cleanly."""
    t = re.sub(r"\s+", " ", clean_headnote(raw or "")).strip()
    if not t: return ""
    m = re.search(r"[.?!]\s+([A-Z])", t[:90])          # prefer a sentence start near the front
    if m: t = t[m.start(1):]
    elif t[0].islower():                                # else drop a leading partial word
        sp = t.find(" ")
        if 0 <= sp <= 30: t = "…" + t[sp + 1:]
    if len(t) > n:                                      # truncate on a word boundary
        cut = t.rfind(" ", 0, n)
        t = (t[:cut] if cut > 0 else t[:n]).rstrip(" ,;:–-") + "…"
    return t

def resolve_cited(cases_cited, self_id):
    """Cases THIS judgment relies on (from metadata), each resolved to a corpus doc where the
    parallel citation matches (cross-reporter via the ' : '-joined citation string)."""
    out = []
    for c in (cases_cited or []):
        cites = c.get("citations") or []
        rid = None
        for cstr in cites:
            for part in re.split(r"\s*[:;]\s*", cstr):
                rid = cite_resolver.get(norm_cite(part))
                if rid and rid != self_id: break
                rid = None
            if rid: break
        if not rid and c.get("name"):                      # citation didn't resolve (e.g. cited only by SCC / SCC OnLine,
            hits = name_search(c["name"], 1)               # which our resolver doesn't index) — fall back to a fuzzy NAME
            if hits and hits[0] != self_id: rid = hits[0]  # match (needs a distinctive token, so generics won't mislink)
        out.append({"name": c.get("name"), "citation": (cites[0] if cites else ""),
                    "treatment": c.get("treatment"), "id": rid})
    return out
st = SentenceTransformer("BAAI/bge-small-en-v1.5", device=DEV,
                         model_kwargs={"torch_dtype": torch.bfloat16 if DEV == "cuda" else torch.float32})  # bf16 only helps on GPU; CPU box loads fp32
ce = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2", device=DEV)
print(f"READY — {NDOCS} judgments, DeepSeek serving={'yes' if DS_KEY else 'NO KEY'}", flush=True)

def card(d, s, ci):
    m = meta.get(d, {}); gl = goodlaw.get(d, {})
    return {"doc_id": d, "case_name": m.get("case_name"), "neutral_citation": m.get("neutral_citation"),
            "equivalent_citations": m.get("equivalent_citations"), "court": m.get("court"), "date": m.get("date"),
            "bench_strength": m.get("bench_strength"), "disposition": m.get("disposition"),
            "good_law_status": gl.get("good_law_status", "unknown"), "good_law_prov": gl.get("provenance"),
            "cited_by": cite_indeg.get(d, 0), "rr": round(s, 2), "passage": passage_snippet(texts[ci]),
            "chunk": re.sub(r"\s+", " ", clean_headnote(texts[ci]))[:1600]}

def dense(q, n=CAND):
    qv = st.encode(BGE_Q + q, normalize_embeddings=True, convert_to_numpy=True).astype(np.float32)
    sim = M @ qv
    cand = np.argpartition(-sim, n)[:n]
    return [int(ci) for ci in cand[np.argsort(-sim[cand])]]

def rerank(q, cand, topk=12):
    rr = ce.predict([(q, texts[ci]) for ci in cand]); best = {}
    for ci, s in zip(cand, rr):
        d = chunk_doc[ci]
        if d not in best or s > best[d][0]: best[d] = (float(s), ci)
    return [card(d, s, ci) for d, (s, ci) in sorted(best.items(), key=lambda x: x[1][0], reverse=True)[:topk]]

def bm25_top(q, n=CAND):
    s = bm25.get_scores(_tok(q))
    return [int(i) for i in np.argsort(-s)[:n] if s[i] > 0]

def candidates(q, n=CAND):
    """Hybrid candidate pool: dense (semantic) + BM25 (exact terms / section nums / names), RRF-fused."""
    dc, bc = dense(q, n), bm25_top(q, n)
    sc = defaultdict(float)
    for r, ci in enumerate(dc): sc[ci] += 1.0 / (60 + r + 1)
    for r, ci in enumerate(bc): sc[ci] += 1.0 / (60 + r + 1)
    return [ci for ci, _ in sorted(sc.items(), key=lambda x: -x[1])][:max(n, 48)]

def retrieve(q, topk=12):
    return rerank(q, candidates(q), topk)

# --- citation-graph tools (Stage 3) ---
def cited_by_docs(d):   # who cites d (inbound), de-duped
    return list(dict.fromkeys(in_edges.get(d, [])))
def cites_docs(d):      # what d cites (outbound)
    return list(dict.fromkeys(out_edges.get(d, [])))

def card_for_doc(q, d):
    """Build a card for a doc by reranking its own chunks against q (real relevance score for added cases)."""
    cis = doc_chunks.get(d, [])
    if not cis: return id_card(d)
    rr = ce.predict([(q, texts[ci]) for ci in cis[:6]])
    bi = int(np.argmax(rr))
    c = card(d, float(rr[bi]), cis[bi]); c["relevance"] = "partial"
    return c

def verify(q, cases):
    """Fresh paralegal reviewer (DeepSeek) — only the passages, nothing else."""
    listing = "\n".join(f"[{i}] {c['case_name']}: {c['passage'][:280]}" for i, c in enumerate(cases))
    msg = [{"role": "system", "content": 'You are a paralegal screening search results. Judge whether each case is relevant to the legal query. Output ONLY a JSON array like [{"i":0,"v":"relevant"}] where v is relevant, partial, or not.'},
           {"role": "user", "content": f"Query: {q}\n\nCases:\n{listing}\n\nJSON:"}]
    try:
        with fn("verify"): t = llm(msg, 400)
        j = json.loads(t[t.find("["):t.rfind("]") + 1]); vm = {d["i"]: d["v"] for d in j}
        for i, c in enumerate(cases): c["relevance"] = vm.get(i, "partial")
    except Exception as e:
        log_event("internal", lvl="WARN", stage="verify", exc_type=type(e).__name__, exc_msg=_clip(str(e), 300))
        for c in cases: c["relevance"] = "partial"
    return cases

# --- exact identity lookup (case name / citation) — a name or cite is not a legal QUESTION,
#     so dense-passage search + the relevance reviewer miss it; route it to metadata instead. ---
_NAME_STOP = {"v", "vs", "of", "and", "the", "ors", "anr", "etc", "state", "union", "govt", "government", "in", "re"}
_CITE_ANY = re.compile(r"\[\d{4}\]\s*\d+\s*S\.?C\.?R\.?\s*\d+|\(\d{4}\)\s*\d+\s*SCC\s*\d+|\d{4}\s+INSC\s+\d+|AIR\s+\d{4}\s+SC\s+\d+", re.I)

def name_search(q, k=6):
    raw = [t for t in re.findall(r"[a-z]+", q.lower()) if t not in _NAME_STOP and len(t) > 1]
    if not raw: return []
    qtok = set()                                            # fuzzy-expand each token vs the name vocab (visaka -> vishaka)
    for t in raw:
        if t in name_vocab or len(t) <= 3: qtok.add(t)
        else: qtok.update(difflib.get_close_matches(t, name_vocab, n=3, cutoff=0.82) or [t])
    scored = []
    for d, m in meta.items():
        ntok = set(re.findall(r"[a-z]+", (m.get("case_name") or "").lower()))
        ov = qtok & ntok
        if len(ov) >= 2 or (len(ov) == 1 and any(len(t) >= 7 for t in ov)):
            scored.append((len(ov), cite_indeg.get(d, 0), d))   # SALIENCE tiebreak: the landmark (more-cited) wins over a namesake
    scored.sort(reverse=True)
    return [d for _, _, d in scored[:k]]

def identity_hits(q):
    ql = q.strip()
    m = _CITE_ANY.search(ql)
    if m:
        rid = cite_resolver.get(norm_cite(m.group(0))) or nc2doc.get(m.group(0))
        if rid: return [rid], "citation"
    if re.search(r"\bv[s.]?\b|\bversus\b", ql, re.I) and len(ql) <= 90:
        hits = name_search(ql)
        if hits: return hits, "case name"
    return [], None

def id_card(d):
    cis = doc_chunks.get(d)
    c = card(d, 9.9, cis[0]) if cis else {"doc_id": d, "rr": 9.9}
    m = meta.get(d, {}); gl = goodlaw.get(d, {})
    c.update({"case_name": m.get("case_name"), "neutral_citation": m.get("neutral_citation"),
              "equivalent_citations": m.get("equivalent_citations"), "court": m.get("court"), "date": m.get("date"),
              "bench_strength": m.get("bench_strength"), "disposition": m.get("disposition"),
              "good_law_status": gl.get("good_law_status", "unknown"), "good_law_prov": gl.get("provenance"),
              "cited_by": cite_indeg.get(d, 0), "relevance": "relevant",
              "passage": passage_snippet(m.get("held") or m.get("issue") or c.get("passage") or "")})
    return c

def improve(q):
    ids, kind = identity_hits(q)
    if ids:
        cases = [id_card(d) for d in ids][:8]
        if kind == "case name":
            seen = {c["doc_id"] for c in cases}
            for c in rerank(q, candidates(q), 6):
                if c["doc_id"] not in seen and len(cases) < 8:
                    c["relevance"] = "partial"; cases.append(c); seen.add(c["doc_id"])
        return cases, {"identity": kind, "retrieved": len(cases), "dropped": 0, "requeried": False}
    steps = {"retrieved": 0, "dropped": 0, "requeried": False}
    cases = verify(q, retrieve(q, 12)); steps["retrieved"] = len(cases)
    kept = [c for c in cases if c["relevance"] in ("relevant", "partial")]
    steps["dropped"] = len(cases) - len(kept)
    if len(kept) < 4:                                            # one bounded re-query on weak recall
        steps["requeried"] = True
        try:
            with fn("requery"): rw = llm([{"role": "user", "content": f'Rewrite this as a precise legal-register search query (one line, no preamble): "{q}"'}], 60).strip().strip('"')
        except Exception as e:
            log_event("internal", lvl="WARN", stage="requery", exc_type=type(e).__name__, exc_msg=_clip(str(e), 300)); rw = q
        seen = {c["doc_id"] for c in kept}
        kept += [c for c in verify(q, retrieve(rw, 8)) if c["relevance"] in ("relevant", "partial") and c["doc_id"] not in seen]
    kept.sort(key=lambda c: (c["relevance"] != "relevant", -c["rr"]))
    return kept[:8], steps

# --- RENDER-FROM-LEDGER grounding gate (anti-hallucination, all-paths) ---
# The synthesiser must back each claim with a VERBATIM quote copied from the case's loaded text;
# we drop any claim whose quote is not a true substring of that case's chunk, or whose [n] is out
# of range. The user-facing answer is rendered ONLY from the surviving (verified) claims — so a
# hallucinated holding or a fabricated/un-loaded case has no surface to appear on.
def _ground_msgs(q, cases):
    ctx = "\n\n".join(f"[{i+1}] {c['case_name']} ({c.get('neutral_citation') or ''}):\n{c.get('chunk') or c.get('passage')}" for i, c in enumerate(cases[:5]))
    sysmsg = ('You are summarising SEARCH RESULTS for a lawyer. Using ONLY the supplied case texts, output a JSON array of 2-4 items '
              'that SUMMARISE what the retrieved cases hold on the issue — a neutral digest of the line of authority to help a lawyer scan the results. '
              'This is a SUMMARY OF THE CASES, NOT legal advice, NOT a recommendation, NOT guidance to a client — never say what the lawyer or client "should" do. '
              'Each item: {"claim": one plain sentence stating what that case holds/establishes, "n": the [n] of the case, "quote": a SHORT span (6-20 words) copied EXACTLY, character-for-character, from case [n]\'s supplied text}. '
              'The quote MUST be a verbatim substring of case [n]. Never paraphrase the quote, never invent. If the cases do not address the issue, output [].')
    return [{"role": "system", "content": sysmsg},
            {"role": "user", "content": f"Query: {q}\n\nCases:\n{ctx}\n\nJSON array:"}]

def _norm(s): return re.sub(r"\s+", " ", (s or "")).strip().lower()

def verify_claims(arr, cases):
    """THE gate (pure, unit-testable): keep a claim only if its [n] is in range AND its quote is a
    verbatim substring of case [n]'s loaded text. A fabricated/un-loaded case or invented quote drops."""
    texts_norm = [_norm(c.get("chunk") or c.get("passage")) for c in cases[:5]]
    verified, dropped = [], []                                        # dropped: [{claim, reason}] for transparency
    for it in (arr if isinstance(arr, list) else []):
        n = (it or {}).get("n"); claim = ((it or {}).get("claim") or "").strip(); quote = ((it or {}).get("quote") or "").strip()
        if not claim or not isinstance(n, int) or isinstance(n, bool) or not (1 <= n <= len(texts_norm)):
            if claim: dropped.append({"claim": claim[:240], "reason": "no valid case reference"})
            continue                                                  # bool-n guard: isinstance(True,int) is True
        nq = _norm(quote)
        if nq and len(nq.split()) >= 4 and nq in texts_norm[n - 1]:   # the gate: a substantive (>=4-word) verbatim substring
            verified.append({"claim": claim, "n": n, "quote": quote})
        else:
            dropped.append({"claim": claim[:240], "reason": "could not be traced to a verbatim passage in the cited case"})
    return verified, dropped

def grounded_answer(q, cases):
    """Returns {text, claims:[{claim,n,quote}], dropped:int}. text is rendered only from verified claims."""
    if not cases:
        return {"text": "No relevant judgments found for this query.", "claims": [], "dropped": 0}
    try:
        with fn("ground"): raw = llm(_ground_msgs(q, cases), 700)
        arr = json.loads(raw[raw.find("["):raw.rfind("]") + 1])
    except Exception as e:
        log_event("internal", lvl="WARN", stage="ground", exc_type=type(e).__name__, exc_msg=_clip(str(e), 300))
        return {"text": "No grounded synthesis could be verified — see the cases below.", "claims": [], "dropped": 0}
    verified, dropped = verify_claims(arr, cases)
    if not verified:
        return {"text": "No grounded synthesis could be verified against the retrieved cases — review the cases below directly.", "claims": [], "dropped": len(dropped), "dropped_items": dropped}
    # Render the VERIFIED words, not just the unchecked paraphrase: show each claim with the verbatim
    # quote it rests on, so the user reads what is actually grounded (deep claim↔quote entailment is Stage-2).
    text = "  ".join(f'{v["claim"]} — "…{v["quote"]}…" [{v["n"]}]' for v in verified)
    return {"text": text, "claims": verified, "dropped": len(dropped), "dropped_items": dropped}

def answer(q, cases):                       # non-stream callers (/api/search)
    return grounded_answer(q, cases)["text"]

def answer_events(q, cases, stats=None):    # SSE: the verified grounded answer + provenance
    yield sse({"t": "step", "k": "answer", "s": "run", "label": "Summarising the cases"})
    ga = grounded_answer(q, cases)
    buf = ""
    for w in ga["text"].split(" "):
        buf += w + " "
        if len(buf) >= 14:
            yield sse({"t": "answer_delta", "text": buf}); buf = ""
    if buf: yield sse({"t": "answer_delta", "text": buf})
    if ga["claims"]:
        yield sse({"t": "claims", "claims": ga["claims"]})
    if ga.get("dropped_items"):                     # transparency: what we REFUSED to assert (couldn't ground)
        yield sse({"t": "dropped_claims", "items": ga["dropped_items"]})
    n = len(ga["claims"])
    if stats is not None: stats["n_verified"] = n; stats["n_dropped"] = ga["dropped"]
    lab = (f"Summary grounded in {n} verbatim holding{'s' if n != 1 else ''}" + (f" · set aside {ga['dropped']} the cases didn't support" if ga["dropped"] else "")) if n else "Couldn't ground a summary — review the cases below"
    yield sse({"t": "step", "k": "answer", "s": "done", "label": lab})

def doc_text(d):
    cs = [texts[i] for i in doc_chunks.get(d, [])]
    return ("".join(c[:1200] for c in cs[:-1]) + cs[-1]) if cs else ""

app = FastAPI(title="Moonley API", description="Grounded Indian legal research API")

# --- Clerk access gate for public hosting ---
@app.middleware("http")
async def _clerk_gate(request, call_next):
    rid = uuid.uuid4().hex[:16]; REQ_ID.set(rid)
    rejection = None
    if request.method != "OPTIONS" and request.url.path not in PUBLIC_PATHS:
        rejection = authenticate_clerk_request(request)
    cu = getattr(request.state, "clerk_user_id", "")
    ip = request.client.host if request.client else ""
    ua = request.headers.get("user-agent", "")
    request.state.req_id = rid; request.state.claimed_user = cu
    request.state.client_ip = ip; request.state.user_agent = ua
    USAGE_CTX.set({"claimed_user": cu, "client_ip": ip, "user_agent": ua})
    if rejection is not None:
        log_event("usage", endpoint=request.url.path, claimed_user=cu, client_ip=ip, user_agent=ua,
                  http_status=rejection.status_code, outcome="denied")
        return rejection
    return await call_next(request)

from fastapi.middleware.cors import CORSMiddleware
app.add_middleware(CORSMiddleware, allow_origins=cors_origins(), allow_methods=["*"],
                   allow_headers=["*"], expose_headers=["*"])

@app.get("/api/v2/auth/config")
def auth_config():
    return frontend_auth_config()

def sse(o): return "data: " + json.dumps(o, ensure_ascii=False) + "\n\n"

@app.get("/api/search_stream")
def search_stream(q: str, request: Request):
    rid = getattr(request.state, "req_id", ""); REQ_ID.set(rid)
    uc = USAGE_CTX.get() or {"claimed_user": getattr(request.state, "claimed_user", ""),
                             "client_ip": getattr(request.state, "client_ip", ""),
                             "user_agent": getattr(request.state, "user_agent", "")}
    t0 = time.time()
    def gen():
        route = "doctrinal"; final = []; thin = False; outcome = "ok"; stats = {}
        try:
            yield sse({"t": "meta", "req_id": rid})
            ids, kind = identity_hits(q)
            if ids:                                            # exact case-name / citation lookup
                route = "identity"
                yield sse({"t": "step", "k": "identity", "s": "done", "label": f"Matched {len(ids)} judgment{'s' if len(ids) != 1 else ''} by {kind}"})
                cases = [id_card(d) for d in ids][:8]
                if kind == "case name":
                    seen = {c["doc_id"] for c in cases}
                    for c in rerank(q, candidates(q), 6):
                        if c["doc_id"] not in seen and len(cases) < 8:
                            c["relevance"] = "partial"; cases.append(c); seen.add(c["doc_id"])
                final = cases
                yield sse({"t": "step", "k": "goodlaw", "s": "done", "label": "Checked which results are still good law"})
                yield sse({"t": "results", "results": cases})
                for ev in answer_events(q, cases, stats): yield ev
                yield sse({"t": "done"})
                return
            yield sse({"t": "step", "k": "search", "s": "run", "label": f"Searching all {NDOCS:,} reportable Supreme Court judgments"})
            cand = candidates(q)
            yield sse({"t": "step", "k": "search", "s": "done", "label": f"Searched all {NDOCS:,} judgments by meaning and keywords"})

            yield sse({"t": "step", "k": "rerank", "s": "run", "label": "Ranking the closest matches to your issue"})
            cases = rerank(q, cand, 12)
            yield sse({"t": "step", "k": "rerank", "s": "done", "label": f"Shortlisted the {len(cases)} closest judgments"})

            yield sse({"t": "step", "k": "review", "s": "run", "label": "Reviewing each result for relevance to your issue"})
            cases = verify(q, cases)
            kept = [c for c in cases if c["relevance"] in ("relevant", "partial")]
            dropped = len(cases) - len(kept)
            yield sse({"t": "step", "k": "review", "s": "done", "label": f"Reviewed {len(cases)} — kept {len(kept)} on-point, set aside {dropped}"})

            if len(kept) < 4:
                thin = True
                yield sse({"t": "step", "k": "requery", "s": "run", "label": "Few on-point results — rephrasing the search once"})
                try:
                    with fn("requery"): rw = llm([{"role": "user", "content": f'Rewrite this as a precise legal-register search query (one line, no preamble): "{q}"'}], 60).strip().strip('"')
                except Exception as e:
                    log_event("internal", lvl="WARN", stage="requery", exc_type=type(e).__name__, exc_msg=_clip(str(e), 300)); rw = q
                seen = {c["doc_id"] for c in kept}
                extra = [c for c in verify(q, rerank(rw, candidates(rw), 8)) if c["relevance"] in ("relevant", "partial") and c["doc_id"] not in seen]
                kept += extra
                yield sse({"t": "step", "k": "requery", "s": "done", "label": f'Rephrased the search — found {len(extra)} more'})

            kept.sort(key=lambda c: (c["relevance"] != "relevant", -c["rr"]))
            kept = kept[:8]; final = kept
            yield sse({"t": "step", "k": "goodlaw", "s": "done", "label": "Checked which results are still good law"})
            yield sse({"t": "results", "results": kept})
            for ev in answer_events(q, kept, stats): yield ev
            yield sse({"t": "done"})
        except Exception as e:
            outcome = "error"
            log_event("internal", lvl="ERROR", stage="search_stream", exc_type=type(e).__name__, exc_msg=_clip(str(e), 300))
            yield sse({"t": "error", "message": str(e)[:200]})
            yield sse({"t": "done"})
        finally:
            log_event("usage", **uc, endpoint="search_stream", mode="fast", route=route, http_status=200,
                      q=_clip(q, 2000), n_results=len(final), top_doc_ids=[c.get("doc_id") for c in final[:5]],
                      n_claims_verified=stats.get("n_verified", 0), n_claims_dropped=stats.get("n_dropped", 0),
                      thin=thin, latency_ms=int((time.time() - t0) * 1000), outcome=(outcome if final or outcome == "error" else "empty"))
    ctx = contextvars.copy_context()
    return StreamingResponse(_bind_ctx(gen(), ctx), media_type="text/event-stream",
                             headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no", "Connection": "keep-alive"})

def _plan(q):
    """DeepSeek controller: decompose the issue + name the LEADING authorities a lawyer expects.
    Names are only candidates — each is grounded via name_search; a hallucinated name simply fails to resolve."""
    try:
        with fn("plan"):
            t = llm([{"role": "system", "content": 'Indian Supreme Court legal-research planner. For the query output JSON {"sub_issues":[1-3 short issue phrases],"authorities":[up to 5 LEADING / LANDMARK SC case names a lawyer would expect on this exact issue — case names only, no citations]}. Name only genuinely well-known authorities; every name is verified against our corpus, so do not pad. [] if unsure.'},
                     {"role": "user", "content": q}], 320)
        j = json.loads(t[t.find("{"):t.rfind("}") + 1])
        return (j.get("sub_issues") or [])[:3], (j.get("authorities") or [])[:5]
    except Exception as e:
        log_event("internal", lvl="WARN", stage="plan", exc_type=type(e).__name__, exc_msg=_clip(str(e), 300))
        return [], []

@app.get("/api/deep_search_stream")
def deep_search_stream(q: str, request: Request):
    rid = getattr(request.state, "req_id", ""); REQ_ID.set(rid)
    uc = USAGE_CTX.get() or {"claimed_user": getattr(request.state, "claimed_user", ""),
                             "client_ip": getattr(request.state, "client_ip", ""),
                             "user_agent": getattr(request.state, "user_agent", "")}
    t0 = time.time()
    def gen():
        route = "deep"; final = []; outcome = "ok"; stats = {}
        try:
            yield sse({"t": "meta", "req_id": rid})
            ids, kind = identity_hits(q)                          # a bare name/cite lookup never needs the deep loop
            if ids:
                route = "identity"
                cases = [id_card(d) for d in ids][:8]; final = cases
                yield sse({"t": "step", "k": "identity", "s": "done", "label": f"Matched {len(ids)} by {kind}"})
                yield sse({"t": "results", "results": cases})
                for ev in answer_events(q, cases, stats): yield ev
                yield sse({"t": "done"}); return

            yield sse({"t": "step", "k": "plan", "s": "run", "label": "Identifying the leading authorities a lawyer would expect"})
            subs, auths = _plan(q)
            yield sse({"t": "step", "k": "plan", "s": "done", "label": ("Checking for the leading authorities on this issue: " + ", ".join(auths[:5])) if auths else f"Broke the issue into {len(subs)} sub-issue(s)"})

            yield sse({"t": "step", "k": "seed", "s": "run", "label": f"Searching all {NDOCS:,} reportable Supreme Court judgments"})
            cases = verify(q, retrieve(q, 12))
            kept = [c for c in cases if c["relevance"] in ("relevant", "partial")]
            seen = {c["doc_id"] for c in kept}
            yield sse({"t": "step", "k": "seed", "s": "done", "label": f"{len(kept)} on-point results from the search"})

            yield sse({"t": "step", "k": "expand", "s": "run", "label": "Bringing in the leading authorities and the cases they rely on"})
            add, auth_docs = [], set()
            auth_hits = {}                                        # authority NAME -> resolved corpus doc_ids (to close the loop)
            for nm in auths:                                      # ground each proposed authority (name -> corpus doc)
                docs = name_search(nm, 2); auth_hits[nm] = docs
                for d in docs:
                    if d not in seen and d not in add: add.append(d); auth_docs.add(d)
            nbr = Counter()                                       # cases the seed results commonly rely on (shared citations)
            for c in kept[:6]:
                for t in cites_docs(c["doc_id"]):
                    if t not in seen: nbr[t] += 1
            for t, _ in nbr.most_common(6):
                if t not in add: add.append(t)
            new_cards = verify(q, [card_for_doc(q, d) for d in add[:14]])
            for c in new_cards: c["authority"] = c["doc_id"] in auth_docs   # a PLAN-named leading authority
            added = [c for c in new_cards if c["relevance"] in ("relevant", "partial")
                     and c["good_law_status"] not in ("overruled", "partly_overruled", "per_incuriam")]
            kept += added
            yield sse({"t": "step", "k": "expand", "s": "done", "label": f"Added {len(added)} more after review (leading authorities + frequently-cited cases)"})

            # rank: a reviewer-confirmed leading authority earns a top slot (authority buys a seat AFTER vetting);
            # then relevant seed results by rerank; then partials. (Pure rerank would bury old foundational cases.)
            def _sk(c): return (not (c.get("authority") and c["relevance"] == "relevant"), c["relevance"] != "relevant", -c.get("rr", 0))
            uniq, s2 = [], set()
            for c in sorted(kept, key=_sk):
                if c["doc_id"] not in s2: s2.add(c["doc_id"]); uniq.append(c)
            kept = uniq[:10]; final = kept
            # close the authority loop: tell the lawyer which expected landmarks actually made it into the results
            if auths:
                kept_ids = {c["doc_id"] for c in kept}
                got = [nm for nm in auths if any(dd in kept_ids for dd in auth_hits.get(nm, []))]
                miss = [nm for nm in auths if nm not in got]
                lab = (("Leading authorities now in your results: " + ", ".join(got)) if got else "None of the expected landmark authorities were on point here") \
                      + (" · not on point here: " + ", ".join(miss) if miss else "")
                yield sse({"t": "step", "k": "authcheck", "s": "done", "label": lab})
            yield sse({"t": "step", "k": "goodlaw", "s": "done", "label": "Checked which results are still good law"})
            yield sse({"t": "results", "results": kept})
            for ev in answer_events(q, kept, stats): yield ev
            yield sse({"t": "done"})
        except Exception as e:
            outcome = "error"
            log_event("internal", lvl="ERROR", stage="deep_search_stream", exc_type=type(e).__name__, exc_msg=_clip(str(e), 300))
            yield sse({"t": "error", "message": str(e)[:200]}); yield sse({"t": "done"})
        finally:
            log_event("usage", **uc, endpoint="deep_search_stream", mode="deep", route=route, http_status=200,
                      q=_clip(q, 2000), n_results=len(final), top_doc_ids=[c.get("doc_id") for c in final[:5]],
                      n_claims_verified=stats.get("n_verified", 0), n_claims_dropped=stats.get("n_dropped", 0),
                      latency_ms=int((time.time() - t0) * 1000), outcome=(outcome if final or outcome == "error" else "empty"))
    ctx = contextvars.copy_context()
    return StreamingResponse(_bind_ctx(gen(), ctx), media_type="text/event-stream",
                             headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no", "Connection": "keep-alive"})

@app.get("/api/search")
def search(q: str, request: Request):
    REQ_ID.set(getattr(request.state, "req_id", ""))
    uc = USAGE_CTX.get() or {"claimed_user": getattr(request.state, "claimed_user", ""),
                             "client_ip": getattr(request.state, "client_ip", ""),
                             "user_agent": getattr(request.state, "user_agent", "")}
    t0 = time.time()
    try:
        cases, steps = improve(q)
        log_event("usage", **uc, endpoint="search", mode="fallback", q=_clip(q, 2000), http_status=200,
                  n_results=len(cases), top_doc_ids=[c.get("doc_id") for c in cases[:5]],
                  latency_ms=int((time.time() - t0) * 1000), outcome=("ok" if cases else "empty"))
        return JSONResponse({"query": q, "answer": answer(q, cases), "results": cases, "steps": steps})
    except Exception as e:
        log_event("usage", **uc, endpoint="search", mode="fallback", q=_clip(q, 2000), http_status=500,
                  latency_ms=int((time.time() - t0) * 1000), outcome="error")
        return JSONResponse({"query": q, "answer": "", "results": [], "error": str(e)[:200]}, status_code=500)

@app.get("/api/judgment")
def judgment(id: str, request: Request):
    REQ_ID.set(getattr(request.state, "req_id", ""))
    uc = USAGE_CTX.get() or {"claimed_user": getattr(request.state, "claimed_user", ""),
                             "client_ip": getattr(request.state, "client_ip", ""),
                             "user_agent": getattr(request.state, "user_agent", "")}
    d = id if id in meta else nc2doc.get(id)
    if not d:
        log_event("usage", **uc, endpoint="judgment", doc_id=_clip(id, 200), http_status=404, outcome="not_found")
        return JSONResponse({"error": "not found"}, status_code=404)
    m = meta.get(d, {}); gl = goodlaw.get(d, {})
    log_event("usage", **uc, endpoint="judgment", doc_id=d, neutral_citation=m.get("neutral_citation"),
              case_name=_clip(m.get("case_name"), 300), http_status=200, outcome="ok")
    raw = doc_text(d); clean = clean_judgment(raw)[:80000]
    return JSONResponse({"doc_id": d, "case_name": m.get("case_name"), "neutral_citation": m.get("neutral_citation"),
        "equivalent_citations": m.get("equivalent_citations"), "court": m.get("court"), "date": m.get("date"),
        "bench": m.get("bench"), "author_judge": m.get("author_judge"), "bench_strength": m.get("bench_strength"),
        "case_number": m.get("case_number"), "disposition": m.get("disposition"), "acts": m.get("acts"),
        "issue": clean_headnote(m.get("issue")), "held": clean_headnote(m.get("held")),
        "good_law_status": gl.get("good_law_status", "unknown"),
        "good_law_prov": gl.get("provenance"), "as_of": gl.get("as_of"), "cited_by": cite_indeg.get(d, 0),
        "treatment_breakdown": gl.get("treatment_breakdown", {}), "corpus_n": NDOCS,
        "cnr": m.get("cnr"), "year": m.get("year"), "cited_cases": resolve_cited(m.get("cases_cited"), d),
        "has_pdf": d in pdfmap,
        "text": clean, "text_raw": raw[:80000], "links": doc_links(d, clean)})

@app.get("/api/pdf")
def pdf(id: str, request: Request, dl: int = 0):
    REQ_ID.set(getattr(request.state, "req_id", ""))
    uc = USAGE_CTX.get() or {"claimed_user": getattr(request.state, "claimed_user", ""),
                             "client_ip": getattr(request.state, "client_ip", ""),
                             "user_agent": getattr(request.state, "user_agent", "")}
    t0 = time.time()
    d = id if id in meta else nc2doc.get(id)
    if not d:
        log_event("usage", **uc, endpoint="pdf", doc_id=_clip(id, 200), http_status=404, outcome="not_found")
        return JSONResponse({"error": "not found"}, status_code=404)
    local, status = fetch_pdf(d)
    if not local:
        log_event("usage", **uc, endpoint="pdf", doc_id=d, http_status=502, outcome=status,
                  latency_ms=int((time.time() - t0) * 1000))
        return JSONResponse({"error": "pdf unavailable", "reason": status}, status_code=502)
    log_event("usage", **uc, endpoint="pdf", doc_id=d, http_status=200, outcome="ok", cache=status,
              mode=("download" if dl else "inline"), latency_ms=int((time.time() - t0) * 1000))
    fname = (d.replace(" ", "_") + ".pdf") if dl else None
    disp = f'attachment; filename="{fname}"' if dl else "inline"
    return FileResponse(local, media_type="application/pdf",
                        headers={"Content-Disposition": disp, "Cache-Control": "private, max-age=3600"})

@app.get("/")
def home():
    return JSONResponse(
        {"service": "Moonley API", "status": "ok", "ui": "https://moonley-pilot.vercel.app"},
        headers={"Cache-Control": "no-store"},
    )