"""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"}, )