"""Moonley API service. Loads the legal tool registry once and streams grounded research over SSE. The React interface is hosted only on Vercel; this process is the private HF backend. Run: MOONLEY_DATA=.../thor_artifacts MOONLEY_STATUTE=".../statute corpus" \ .venv/bin/uvicorn --app-dir phase1/scripts serve_agent:app --host 127.0.0.1 --port 8001 """ import os, sys, re, json, time, hashlib from urllib.parse import unquote def _promote_moonley_environment() -> None: """Let unchanged corpus internals consume canonical Moonley configuration.""" for name, value in list(os.environ.items()): if name.startswith("MOONLEY_"): os.environ.setdefault("THEMIS_" + name[len("MOONLEY_"):], value) def _load_env(path: str) -> None: if os.path.exists(path): for line in open(path): line = line.strip() if line and not line.startswith("#") and "=" in line: key, value = line.split("=", 1) os.environ.setdefault(key.strip(), value.strip().strip('"').strip("'")) HERE = os.path.dirname(os.path.abspath(__file__)) _load_env(os.path.join(HERE, ".env")) _promote_moonley_environment() sys.path.insert(0, HERE) from tools import Corpus as LegacyCorpus from corpus_v5 import CorpusV5 import agent as A import requests from pdf_sources import PdfSourceResolver from bharat_courts_source import BharatCourtsPdfError, resolve_and_fetch_pdf from graph_view import graph_node_card from project_store import ProjectStore, ProjectStoreError, QuotaExceeded from drafting_service import ( DRAFT_PROFILES, DraftingError, TemplateRegistry, apply_drafting_intake, draft_docx, draft_pdf, draft_profile, drafting_intake_messages, drafting_messages, finalization_messages, infer_draft_profile, missing_draft_fields, public_draft_profile, revision_messages, extract_uploaded_template, ) from knowledge_service import KnowledgeService, KnowledgeServiceError from research_release import build_research_release from statute_crosswalk import ACT_NAMES, normalise_act, normalise_section from fastapi import BackgroundTasks, FastAPI, Request from fastapi.responses import FileResponse, StreamingResponse, JSONResponse, RedirectResponse, Response from clerk_auth import ( # noqa: E402 - the local .env must be loaded first PUBLIC_PATHS, authenticate_clerk_request, clerk_settings, cors_origins, frontend_auth_config, ) HDR = {"Authorization": f"Bearer {os.environ.get('DEEPSEEK_API_KEY','')}", "Content-Type": "application/json"} def llm_fn(msgs): for _ in range(2): try: r = requests.post("https://api.deepseek.com/chat/completions", headers=HDR, timeout=60, json={"model": "deepseek-v4-flash", "temperature": 0, "max_tokens": 700, "thinking": {"type": "disabled"}, "messages": msgs}) if r.status_code == 200: return r.json()["choices"][0]["message"]["content"] except Exception: time.sleep(1) return "{}" def fast_llm_fn(msgs): """Fast user-facing turn: fail closed instead of holding the interface through retries.""" try: r = requests.post("https://api.deepseek.com/chat/completions", headers=HDR, timeout=30, json={"model": "deepseek-v4-flash", "temperature": 0, "max_tokens": 600, "thinking": {"type": "disabled"}, "messages": msgs}) if r.status_code == 200: return r.json()["choices"][0]["message"]["content"] except Exception: pass return "{}" def ds_call(messages, tools): """DeepSeek function-calling turn -> the assistant message (with tool_calls or content).""" r = requests.post("https://api.deepseek.com/chat/completions", headers=HDR, timeout=90, json={"model": "deepseek-v4-flash", "temperature": 0, "max_tokens": 800, "thinking": {"type": "disabled"}, "messages": messages, "tools": tools, "tool_choice": "auto"}) return r.json()["choices"][0]["message"] DATA_DIR = os.environ.get("THEMIS_DATA", ".") STATUTE_DIR = os.environ.get("THEMIS_STATUTE", ".") if os.path.exists(os.path.join(DATA_DIR, "release_manifest.json")): C = CorpusV5(DATA_DIR, STATUTE_DIR, device=os.environ.get("THEMIS_DEVICE", "cpu")) RUNTIME_KIND = "schema-v5-qwen" else: C = LegacyCorpus(DATA_DIR, STATUTE_DIR, device=os.environ.get("THEMIS_DEVICE", "cpu")) RUNTIME_KIND = "legacy-bge" RESEARCH_RELEASE = build_research_release(C, A) # citation resolution for the judgment view (neutral + equivalent -> doc) def norm_cite(c): return re.sub(r"\s+", " ", (c or "").replace(".", "")).strip().upper() cite_resolver = {}; nc2doc = {} for _d, _m in C.meta.items(): if _m.get("neutral_citation"): nc2doc[_m["neutral_citation"]] = _d 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+") def doc_links(text, self_id): out = {} for c in CITE_RE.findall(text or ""): rid = cite_resolver.get(norm_cite(c)) if rid and rid != self_id and C.is_retrieval_eligible(rid) and c not in out: out[c] = rid return [{"cite": k, "id": v} for k, v in out.items()] def resolve_cited(cases_cited, self_id): out = [] for c in (cases_cited or []): rid = None for cstr in (c.get("citations") or []): 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"): hits = C.name_lookup(c["name"], 1) if hits and hits[0]["doc_id"] != self_id: rid = hits[0]["doc_id"] if rid: card = graph_node_card( rid, C.meta.get(rid, {}), treatment=c.get("treatment"), cited_by=C.cite_indeg.get(rid, 0), good_law_status=C.goodlaw.get(rid, {}).get( "good_law_status", "unknown" ), ) if not card["hover"]["case_name"] and c.get("name"): card["display_name"] = c["name"] card["name"] = c["name"] card["hover"]["case_name"] = c["name"] out.append(card) else: out.append( { "name": c.get("name"), "display_name": c.get("name") or "Unresolved cited case", "citation": ((c.get("citations") or [""])[0]), "treatment": c.get("treatment"), "id": None, "node_id": None, "judgment_id": None, "label": None, } ) return out # --- verified source PDFs ----------------------------------------------------- # Map membership is not availability: the upstream bucket contains a small number # of application/pdf objects whose payload is actually an HTML error page. Probe a # bounded byte range, expose the PDF only after its payload is verified, and let the # browser load the public source directly so large PDFs retain byte-range support. PDF_SOURCES = PdfSourceResolver(os.path.join(DATA_DIR, "escr_pdfmap.jsonl")) print(f"[serve_agent] pdfmap: {PDF_SOURCES.mapped_count} unique judgments have mapped source candidates", flush=True) app = FastAPI(title="Moonley API", description="Private grounded Indian legal research API", version="2") RUNTIME_WARM = RUNTIME_KIND != "schema-v5-qwen" PROJECTS = ProjectStore.from_env() DRAFTING = TemplateRegistry(os.path.join(HERE, "..", "drafting")) KNOWLEDGE = KnowledgeService(PROJECTS, C) @app.on_event("startup") def _warm_runtime(): global RUNTIME_WARM if RUNTIME_KIND == "schema-v5-qwen" and os.environ.get("THEMIS_WARM_QUERY_MODEL", "1") == "1": C.warmup() RUNTIME_WARM = True print(f"[serve_agent] READY runtime={RUNTIME_KIND} accepted={len(C.eligible_doc_ids)}", flush=True) # --- Clerk access gate (public hosting) --- @app.middleware("http") async def _clerk_gate(request: Request, call_next): # CORS preflights and the boot/config endpoints must be reachable before sign-in. if request.method != "OPTIONS" and request.url.path not in PUBLIC_PATHS: rejection = authenticate_clerk_request(request) if rejection is not None: return rejection return await call_next(request) # CORS and Clerk's authorized-parties check share one explicit origin allow-list. from fastapi.middleware.cors import CORSMiddleware app.add_middleware(CORSMiddleware, allow_origins=cors_origins(), allow_methods=["*"], allow_headers=["*"], expose_headers=["*"]) def sse(o): return "data: " + json.dumps(o, ensure_ascii=False) + "\n\n" # --- SESSION CAPTURE (the pooled-verification machine for daily lawyer sessions) --- # Every search + every 👍/👎 lands in append-only JSONL; each graded result is a future qrel row. LOG_DIR = os.environ.get("THEMIS_LOG_DIR") or os.path.join(HERE, "..", "logs") os.makedirs(LOG_DIR, exist_ok=True) def _log(name, obj): try: obj = {"ts": time.strftime("%Y-%m-%dT%H:%M:%S"), **obj} with open(os.path.join(LOG_DIR, f"{name}.jsonl"), "a", encoding="utf-8") as fh: fh.write(json.dumps(obj, ensure_ascii=False) + "\n") except Exception: pass LOG_RAW_QUERIES = os.environ.get("THEMIS_LOG_RAW_QUERIES", "0") == "1" def _query_log_fields(query: str) -> dict: normalized = re.sub(r"\s+", " ", query or "").strip() fields = { "query_sha256": hashlib.sha256(normalized.encode("utf-8")).hexdigest(), "query_chars": len(normalized), } if LOG_RAW_QUERIES: fields["q"] = normalized[:2000] return fields from pydantic import BaseModel, Field class CaseChatTurn(BaseModel): role: str content: str class QueryBriefRequest(BaseModel): query: str refinements: list[str] = Field(default_factory=list) history: list[CaseChatTurn] = Field(default_factory=list) active_case_id: str | None = None recent_case_ids: list[str] = Field(default_factory=list) class CaseChatRequest(BaseModel): doc_id: str question: str history: list[CaseChatTurn] = Field(default_factory=list) class SearchRequest(BaseModel): q: str original_q: str | None = None approved: bool = True search_frame: dict | None = None brief_revision: int | None = None route: str = "legal_research" retrieval_scope: str = "global" active_case_id: str | None = None recent_case_ids: list[str] = Field(default_factory=list) history: list[CaseChatTurn] = Field(default_factory=list) case_question: str | None = None class ProjectRequest(BaseModel): name: str class DraftChatSource(BaseModel): id: str = Field(default="", max_length=200) title: str = Field(default="Saved chat", max_length=120) content: str = Field(max_length=16_000) class DraftMatterDetails(BaseModel): matter_title: str = Field(default="", max_length=500) parties: str = Field(default="", max_length=4_000) lower_court: str = Field(default="", max_length=500) case_number: str = Field(default="", max_length=300) impugned_order_date: str = Field(default="", max_length=100) synopsis: str = Field(default="", max_length=8_000) list_of_dates: str = Field(default="", max_length=8_000) questions_of_law: str = Field(default="", max_length=8_000) grounds: str = Field(default="", max_length=8_000) relief: str = Field(default="", max_length=8_000) advocate: str = Field(default="", max_length=500) class DraftRequest(BaseModel): template_id: str = Field(default="", max_length=100) template_text: str = Field(default="", max_length=60_000) document_type: str = Field(default="", max_length=100) project_id: str | None = None document_ids: list[str] = Field(default_factory=list) chat_sources: list[DraftChatSource] = Field(default_factory=list) matter_details: DraftMatterDetails = Field(default_factory=DraftMatterDetails) intake_details: dict[str, str] = Field(default_factory=dict) instructions: str = Field(default="", max_length=6_000) class DraftExportRequest(BaseModel): title: str = Field(default="Moonley working draft", max_length=180) draft: str = Field(max_length=80_000) class DraftIntakeTurn(BaseModel): role: str = Field(max_length=20) content: str = Field(max_length=2_000) class DraftIntakeRequest(BaseModel): message: str = Field(max_length=4_000) document_type: str = Field(default="", max_length=100) details: dict[str, str] = Field(default_factory=dict) history: list[DraftIntakeTurn] = Field(default_factory=list) class DraftFinalizeRequest(DraftExportRequest): document_type: str = Field(default="", max_length=100) class DraftRevisionRequest(DraftFinalizeRequest): instruction: str = Field(max_length=4_000) def _project_owner(request: Request) -> str: return str(getattr(request.state, "clerk_user_id", "")) def _project_error(exc: ProjectStoreError) -> JSONResponse: return JSONResponse( {"error": exc.code, "message": exc.message}, status_code=exc.status_code, headers={"Cache-Control": "no-store"}, ) @app.get("/api/v2/auth/config") def auth_config(): return frontend_auth_config() @app.get("/api/v2/projects") def list_projects(request: Request): try: projects = PROJECTS.list_projects(_project_owner(request)) return JSONResponse( {"projects": projects, "storage": PROJECTS.status()}, headers={"Cache-Control": "no-store"}, ) except ProjectStoreError as exc: return _project_error(exc) @app.post("/api/v2/projects") def create_project(request: Request, body: ProjectRequest): try: project = PROJECTS.create_project(_project_owner(request), body.name) return JSONResponse( {"project": project, "storage": PROJECTS.status()}, status_code=201, headers={"Cache-Control": "no-store"}, ) except ProjectStoreError as exc: return _project_error(exc) @app.get("/api/v2/projects/{project_id}") def get_project(project_id: str, request: Request): try: return JSONResponse( {"project": PROJECTS.get_project(_project_owner(request), project_id)}, headers={"Cache-Control": "no-store"}, ) except ProjectStoreError as exc: return _project_error(exc) @app.patch("/api/v2/projects/{project_id}") def rename_project(project_id: str, request: Request, body: ProjectRequest): try: return JSONResponse( {"project": PROJECTS.rename_project(_project_owner(request), project_id, body.name)}, headers={"Cache-Control": "no-store"}, ) except ProjectStoreError as exc: return _project_error(exc) @app.delete("/api/v2/projects/{project_id}") def delete_project(project_id: str, request: Request): try: KNOWLEDGE.delete_project(_project_owner(request), project_id) PROJECTS.delete_project(_project_owner(request), project_id) return Response(status_code=204, headers={"Cache-Control": "no-store"}) except ProjectStoreError as exc: return _project_error(exc) except KnowledgeServiceError as exc: return JSONResponse( {"error": "knowledge_delete_failed", "message": str(exc)}, status_code=502, headers={"Cache-Control": "no-store"}, ) @app.post("/api/v2/projects/{project_id}/documents") async def upload_project_document(project_id: str, request: Request, background_tasks: BackgroundTasks): try: content_length = request.headers.get("content-length", "").strip() if content_length and int(content_length) > PROJECTS.limits.max_file_bytes: raise QuotaExceeded( f"Each document must be {PROJECTS.limits.max_file_bytes // (1024 * 1024)} MiB or smaller." ) filename = unquote(request.headers.get("x-document-name", "")) content = bytearray() async for chunk in request.stream(): content.extend(chunk) if len(content) > PROJECTS.limits.max_file_bytes: raise QuotaExceeded( f"Each document must be {PROJECTS.limits.max_file_bytes // (1024 * 1024)} MiB or smaller." ) document = PROJECTS.add_document(_project_owner(request), project_id, filename, bytes(content)) background_tasks.add_task( KNOWLEDGE.ingest, _project_owner(request), project_id, document["id"] ) return JSONResponse( {"document": document, "project": PROJECTS.get_project(_project_owner(request), project_id)}, status_code=201, headers={"Cache-Control": "no-store"}, ) except (ValueError, ProjectStoreError) as exc: if isinstance(exc, ProjectStoreError): return _project_error(exc) return JSONResponse({"error": "invalid_content_length"}, status_code=400) @app.delete("/api/v2/projects/{project_id}/documents/{document_id}") def delete_project_document(project_id: str, document_id: str, request: Request): try: KNOWLEDGE.delete(_project_owner(request), project_id, document_id) PROJECTS.delete_document(_project_owner(request), project_id, document_id) return Response(status_code=204, headers={"Cache-Control": "no-store"}) except ProjectStoreError as exc: return _project_error(exc) except KnowledgeServiceError as exc: return JSONResponse( {"error": "knowledge_delete_failed", "message": str(exc)}, status_code=502, headers={"Cache-Control": "no-store"}, ) @app.get("/api/v2/projects/{project_id}/documents/{document_id}") def download_project_document(project_id: str, document_id: str, request: Request): try: document = PROJECTS.document_record(_project_owner(request), project_id, document_id) path = PROJECTS.document_path(_project_owner(request), project_id, document_id) return FileResponse( path, media_type=document.get("media_type") or "application/octet-stream", filename=document.get("name") or "document", headers={"Cache-Control": "private, no-store"}, ) except ProjectStoreError as exc: return _project_error(exc) @app.post("/api/v2/projects/{project_id}/documents/{document_id}/ingest", status_code=202) def ingest_project_document( project_id: str, document_id: str, request: Request, background_tasks: BackgroundTasks ): try: PROJECTS.document_record(_project_owner(request), project_id, document_id) background_tasks.add_task(KNOWLEDGE.ingest, _project_owner(request), project_id, document_id) return JSONResponse( {"status": "queued", "document_id": document_id}, status_code=202, headers={"Cache-Control": "no-store"}, ) except ProjectStoreError as exc: return _project_error(exc) @app.get("/api/v2/statute-crosswalk") def statute_crosswalk(act: str, section: str): code, number = normalise_act(act), normalise_section(section) if not code or not number: return JSONResponse( { "error": "invalid_provision", "message": "Choose IPC, BNS, CrPC, BNSS, IEA, or BSA and enter a section number.", }, status_code=400, ) result = C.statute_crosswalk(code, number) result["provision"] = C.statute_provision(code, number) for item in result.get("corresponding") or []: item["provision"] = C.statute_provision(item["act"], item["section"]) result["supported_acts"] = [ {"act": value, "name": ACT_NAMES[value]} for value in ("IPC", "BNS", "CRPC", "BNSS", "IEA", "BSA") ] return JSONResponse(result, headers={"Cache-Control": "private, max-age=3600"}) @app.get("/api/v2/statute-lookup") def statute_lookup(act: str, section: str): """Exact statutory-text lookup; never infer or substitute another section.""" code, number = normalise_act(act), normalise_section(section) if not code or not number: return JSONResponse( {"found": False, "error": "invalid_provision", "message": "Use IPC, BNS, CrPC, BNSS, IEA, or BSA with an exact section number."}, status_code=400, ) provision = C.statute_provision(code, number) if not provision: return JSONResponse( {"found": False, "act": code, "section": number, "message": "The exact provision is not available in the private statute source; do not guess it."}, headers={"Cache-Control": "private, max-age=300"}, ) return JSONResponse( {"found": True, "act": code, "act_name": ACT_NAMES.get(code), "section": number, "provision": provision}, headers={"Cache-Control": "private, max-age=3600"}, ) @app.get("/api/v2/drafting/templates") def drafting_templates(): return JSONResponse( { "version": DRAFTING.version, "templates": DRAFTING.list(), "document_types": [ public_draft_profile(profile) for profile in DRAFT_PROFILES.values() ], "knowledge": KNOWLEDGE.status(), }, headers={"Cache-Control": "private, max-age=300"}, ) @app.get("/api/v2/drafting/templates/{template_id}/pdf") def drafting_template_pdf(template_id: str): try: template = DRAFTING.get(template_id) return FileResponse( template["path"], media_type="application/pdf", filename=template["filename"], headers={"Cache-Control": "private, max-age=3600"}, ) except DraftingError as exc: return JSONResponse({"error": "template_not_found", "message": str(exc)}, status_code=404) @app.get("/api/v2/drafting/templates/{template_id}/text") def drafting_template_text(template_id: str): try: template = DRAFTING.get(template_id) return JSONResponse( { "template": { key: value for key, value in template.items() if key not in {"path", "filename"} }, "text": DRAFTING.text(template_id), "editable": True, }, headers={"Cache-Control": "private, no-store"}, ) except DraftingError as exc: return JSONResponse({"error": "template_not_found", "message": str(exc)}, status_code=404) @app.post("/api/v2/drafting/templates/extract") async def extract_private_drafting_template(request: Request): """Extract one authenticated user's template without retaining the uploaded file.""" max_bytes = 10 * 1024 * 1024 content_length = request.headers.get("content-length", "").strip() if content_length and int(content_length) > max_bytes: return JSONResponse( {"error": "template_too_large", "message": "Template must be 10 MiB or smaller."}, status_code=413, headers={"Cache-Control": "no-store"}, ) filename = unquote(request.headers.get("x-template-name", "")) content = bytearray() async for chunk in request.stream(): content.extend(chunk) if len(content) > max_bytes: return JSONResponse( {"error": "template_too_large", "message": "Template must be 10 MiB or smaller."}, status_code=413, headers={"Cache-Control": "no-store"}, ) try: extracted = extract_uploaded_template(filename, bytes(content), request.headers.get("content-type", "")) return JSONResponse( extracted, headers={"Cache-Control": "no-store"}, ) except DraftingError as exc: return JSONResponse( {"error": "template_extraction_failed", "message": str(exc)}, status_code=400, headers={"Cache-Control": "no-store"}, ) def draft_llm_fn(messages, *, max_tokens: int = 5000, timeout: int = 150): try: response = requests.post( "https://api.deepseek.com/chat/completions", headers=HDR, timeout=timeout, json={ "model": "deepseek-v4-flash", "temperature": 0, "max_tokens": max_tokens, "thinking": {"type": "disabled"}, "messages": messages, }, ) if response.status_code == 200: return str(response.json()["choices"][0]["message"]["content"] or "").strip() except Exception: pass return "" @app.post("/api/v2/drafting/intake") def drafting_intake(request: Request, body: DraftIntakeRequest): message = re.sub(r"\x00", "", body.message or "").strip()[:4_000] if not message: return JSONResponse( {"error": "empty_message", "message": "Tell Moonley what you want drafted."}, status_code=400, ) current_profile = draft_profile(body.document_type) prompt_profile = current_profile or draft_profile(infer_draft_profile(message)) messages = drafting_intake_messages( message, prompt_profile, body.details, [turn.dict() for turn in body.history[-8:]], ) output = draft_llm_fn(messages, max_tokens=800, timeout=60) model_returned = bool(output) if not output and current_profile: missing = missing_draft_fields(current_profile, body.details) if missing: output = json.dumps( { "document_type": current_profile["id"], "updates": {missing[0]["key"]: message}, "acknowledgement": "Noted.", } ) state = apply_drafting_intake( message, body.document_type, body.details, output, ) _log( "drafting_intake", { "owner_sha256": hashlib.sha256(_project_owner(request).encode("utf-8")).hexdigest(), "document_type": state.get("document_type"), "detail_count": len(state.get("details") or {}), "ready": bool(state.get("ready")), "model_returned": model_returned, }, ) return JSONResponse( { **state, "model_call": { "provider": "deepseek", "attempted": True, "succeeded": model_returned, }, }, headers={"Cache-Control": "no-store"}, ) @app.post("/api/v2/drafting/generate") def generate_draft(request: Request, body: DraftRequest): owner = _project_owner(request) try: profile = draft_profile(body.document_type) if profile: missing = missing_draft_fields(profile, body.intake_details) if missing: return JSONResponse( { "error": "draft_intake_incomplete", "message": f"Complete the drafting chat first: {missing[0]['label']} is still required.", "missing_fields": [field["key"] for field in missing], }, status_code=409, ) template_id = body.template_id or str(profile.get("template_id") or "") else: template_id = body.template_id if template_id: template = DRAFTING.get(template_id) edited_template = re.sub(r"\x00", "", body.template_text or "").strip()[:60_000] template_text = edited_template or DRAFTING.text(template_id) elif profile: template = { "id": profile["id"], "title": profile["title"], "description": profile["description"], "category": "Chat-led", } edited_template = re.sub(r"\x00", "", body.template_text or "").strip()[:60_000] template_text = edited_template or str(profile.get("structure") or "") else: raise DraftingError("Tell Moonley what document to draft first.") sources = [] document_ids = list(dict.fromkeys(body.document_ids))[:8] if document_ids and not body.project_id: raise DraftingError("Choose the project that owns the selected documents.") for document_id in document_ids: document = PROJECTS.document_record(owner, body.project_id or "", document_id) text, extraction = KNOWLEDGE.source_text(owner, body.project_id or "", document_id) if text: sources.append( { "label": f"Project document: {document.get('name')}", "text": text, "kind": "document", "document_id": document_id, "extraction": extraction, } ) for chat in body.chat_sources[:5]: text = re.sub(r"\x00", "", chat.content or "").strip()[:16_000] if text: sources.append({"label": f"Selected chat: {chat.title[:120]}", "text": text, "kind": "chat"}) messages = drafting_messages( template, template_text, body.instructions, sources, body.matter_details.dict(), intake_details=body.intake_details, profile=profile, ) draft = draft_llm_fn(messages) if not draft: return JSONResponse( {"error": "draft_generation_unavailable", "message": "The drafting model did not return a draft. Try again."}, status_code=503, ) _log( "drafting", { "owner_sha256": hashlib.sha256(owner.encode("utf-8")).hexdigest(), "template_id": template_id or profile.get("id"), "document_type": body.document_type, "document_count": len(document_ids), "chat_count": len(body.chat_sources[:5]), }, ) return JSONResponse( { "draft": draft[:80_000], "template": {key: value for key, value in template.items() if key not in {"path", "filename"}}, "sources": [ {key: value for key, value in source.items() if key not in {"text"}} for source in sources ], "notice": "Working draft only. Verify every fact, authority, annexure and filing requirement before use.", }, headers={"Cache-Control": "no-store"}, ) except ProjectStoreError as exc: return _project_error(exc) except DraftingError as exc: return JSONResponse({"error": "invalid_draft_request", "message": str(exc)}, status_code=400) @app.post("/api/v2/drafting/finalize") def finalize_draft(request: Request, body: DraftFinalizeRequest): draft = re.sub(r"\x00", "", body.draft or "").strip()[:80_000] if not draft: return JSONResponse( {"error": "empty_draft", "message": "Generate or enter a draft before finalizing."}, status_code=400, ) profile = draft_profile(body.document_type) final = draft_llm_fn( finalization_messages(body.title, draft, profile), max_tokens=6_000, timeout=150, ) if not final: return JSONResponse( {"error": "finalization_unavailable", "message": "The drafting model did not return a final version. Your editable draft is unchanged."}, status_code=503, ) _log( "drafting_finalize", { "owner_sha256": hashlib.sha256(_project_owner(request).encode("utf-8")).hexdigest(), "document_type": body.document_type, "input_chars": len(draft), }, ) return JSONResponse( { "draft": final[:80_000], "notice": "Finalized working draft only. Counsel must verify the record, law and filing requirements.", }, headers={"Cache-Control": "no-store"}, ) @app.post("/api/v2/drafting/revise") def revise_draft(request: Request, body: DraftRevisionRequest): draft = re.sub(r"\x00", "", body.draft or "").strip()[:80_000] instruction = re.sub(r"\x00", "", body.instruction or "").strip()[:4_000] if not draft: return JSONResponse( {"error": "empty_draft", "message": "Generate or enter a draft before asking for changes."}, status_code=400, ) if not instruction: return JSONResponse( {"error": "empty_instruction", "message": "Tell Moonley what to change or what new draft to prepare."}, status_code=400, ) profile = draft_profile(body.document_type) revised = draft_llm_fn( revision_messages(body.title, draft, instruction, profile), max_tokens=6_000, timeout=150, ) if not revised: return JSONResponse( {"error": "revision_unavailable", "message": "The drafting model did not return an update. Your editable draft is unchanged."}, status_code=503, ) _log( "drafting_revision", { "owner_sha256": hashlib.sha256(_project_owner(request).encode("utf-8")).hexdigest(), "document_type": body.document_type, "input_chars": len(draft), "instruction_chars": len(instruction), }, ) return JSONResponse( { "draft": revised[:80_000], "model_call": {"provider": "deepseek", "attempted": True, "succeeded": True}, "notice": "AI-updated working draft only. Review every change before finalizing.", }, headers={"Cache-Control": "no-store"}, ) @app.post("/api/v2/drafting/export/docx") def export_draft_docx(body: DraftExportRequest): draft = re.sub(r"\x00", "", body.draft or "").strip()[:80_000] if not draft: return JSONResponse( {"error": "empty_draft", "message": "Generate or enter a draft before exporting."}, status_code=400, ) title = re.sub(r"\s+", " ", body.title or "").strip()[:180] or "Moonley working draft" filename = re.sub(r"[^A-Za-z0-9._-]+", "-", title).strip("-.")[:80] or "moonley-working-draft" return Response( content=draft_docx(title, draft), media_type="application/vnd.openxmlformats-officedocument.wordprocessingml.document", headers={ "Cache-Control": "private, no-store", "Content-Disposition": f'attachment; filename="{filename}.docx"', }, ) @app.post("/api/v2/drafting/export/pdf") def export_draft_pdf(body: DraftExportRequest): draft = re.sub(r"\x00", "", body.draft or "").strip()[:80_000] if not draft: return JSONResponse( {"error": "empty_draft", "message": "Generate or enter a draft before exporting."}, status_code=400, ) title = re.sub(r"\s+", " ", body.title or "").strip()[:180] or "Moonley working draft" filename = re.sub(r"[^A-Za-z0-9._-]+", "-", title).strip("-.")[:80] or "moonley-working-draft" try: payload = draft_pdf(title, draft) except DraftingError as exc: return JSONResponse({"error": "pdf_export_failed", "message": str(exc)}, status_code=400) return Response( content=payload, media_type="application/pdf", headers={ "Cache-Control": "private, no-store", "Content-Disposition": f'attachment; filename="{filename}.pdf"', }, ) @app.post("/api/v2/query_brief") @app.post("/api/query_brief") def query_brief(req: QueryBriefRequest): query = re.sub(r"\s+", " ", req.query or "").strip() if not query: return JSONResponse({"error": "query is required"}, status_code=400) history = [ turn.model_dump() if hasattr(turn, "model_dump") else turn.dict() for turn in req.history[-8:] ] active_doc = _eligible_doc(req.active_case_id or "") active_case = C._card(active_doc) if active_doc else None brief = A.query_brief( query, req.refinements, fast_llm_fn, history=history, active_case=active_case, ) route = str(brief.get("route") or "legal_research") scope = str(brief.get("retrieval_scope") or "global") if route.startswith("case_") or scope == "case_plus_global": resolution = A.resolve_case_reference( C, brief.get("case_reference") or query, active_case_id=active_doc, recent_case_ids=req.recent_case_ids, ) brief["case_resolution"] = resolution if resolution.get("status") == "resolved": brief["active_case"] = resolution.get("case") elif resolution.get("status") == "ambiguous": brief["case_message"] = ( "I found more than one plausible case-title match in the corpus. " "Choose the intended judgment; Moonley will not silently substitute one case for another." ) else: reference = re.sub(r"\s+", " ", str(brief.get("case_reference") or query)).strip() brief["case_message"] = ( f"I could not find an exact or reliable close match for {reference!r} in the Supreme Court corpus. " "Add a citation, year, another party name, or subject if you want me to search differently." ) crosswalks = [] for mention in A.extract_statute_mentions(" ".join([query, *req.refinements])): result = C.statute_crosswalk(mention["act"], mention["section"]) if result.get("found"): result["provision"] = C.statute_provision( mention["act"], mention["section"] ) for item in result.get("corresponding") or []: item["provision"] = C.statute_provision( item["act"], item["section"] ) crosswalks.append(result) if crosswalks: brief["statute_crosswalks"] = crosswalks if brief.get("mode") == "research": provisions = list(brief.get("provisions") or []) for item in crosswalks: label = f"{item['from']} corresponds directly to {item['to']}" if label not in provisions: provisions.append(label) brief["provisions"] = provisions[:8] _log("query_briefs", { **_query_log_fields(query), "refinement_count": len([x for x in req.refinements if str(x).strip()]), "history_turn_count": len(history), "route": brief.get("route"), "case_resolution": (brief.get("case_resolution") or {}).get("status"), }) return JSONResponse(brief) def _eligible_results(rows): out, seen = [], set() for card in rows or []: if not isinstance(card, dict): continue d = str(card.get("judgment_id") or card.get("doc_id") or "") if not d or d in seen or not C.is_retrieval_eligible(d): continue copy = dict(card) copy["doc_id"] = d copy["judgment_id"] = d out.append(copy) seen.add(d) return out def _search_response( q: str, *, original_q: str | None = None, approved_frame: dict | None = None, route: str = "legal_research", retrieval_scope: str = "global", active_case_id: str | None = None, case_question: str | None = None, history: list[dict] | None = None, primary_limit: int = 6, more_limit: int = 14, ): q = re.sub(r"\s+", " ", q or "").strip() original_q = re.sub(r"\s+", " ", original_q or q).strip() if not q: return JSONResponse({"error": "query is required"}, status_code=400) case_doc = _eligible_doc(active_case_id or "") requested_scope = retrieval_scope if retrieval_scope in {"case", "graph", "case_plus_global", "global"} else "global" if route in {"case_lookup", "case_question"}: scope = "case" elif route == "case_lineage": scope = "graph" else: scope = requested_scope if requested_scope in {"global", "case_plus_global"} else "global" frame = dict(approved_frame or {}) if approved_frame else None if case_doc and scope == "case_plus_global": frame = dict(frame or {}) known = list(frame.get("known_citations") or []) case_name = str(C.meta.get(case_doc, {}).get("case_name") or "").strip() if case_name and case_name not in known: known.insert(0, case_name) frame["known_citations"] = known[:4] t0 = time.time() def gen(): yield sse({"t": "meta", "corpus": C.coverage(), "grounding": "stored-source-only", "research_release": RESEARCH_RELEASE}) if case_doc: card = C._card(case_doc) yield sse({ "t": "case_context", "route": route, "retrieval_scope": scope, "source": "verified_doc_id", "case": card, }) final, pending_more, more_sent = [], [], False try: if case_doc and scope == "case": events = A.case_context_stream( C, re.sub(r"\s+", " ", str(case_question or q)).strip()[:1200], case_doc, history or [], fast_llm_fn, ) elif case_doc and scope == "graph": events = A.case_lineage_stream(C, case_question or q, case_doc) else: events = A.structured_search_stream( C, q, llm_fn, approved_frame=frame, identity_query=original_q ) for ev in events: if ev.get("t") == "_trace": # stage-level instrumentation -> log only _log("trace", {**_query_log_fields(q), "stage": ev.get("stage"), "data": ev.get("data")}) continue if ev.get("t") == "results": final = _eligible_results(ev.get("results")) pending_more = final[primary_limit:] ev = {**ev, "results": final[:primary_limit]} elif ev.get("t") == "more_results": combined = _eligible_results(pending_more + list(ev.get("results") or [])) pending_more = [] more_sent = True ev = {**ev, "results": combined[:more_limit]} if not ev["results"]: continue elif ev.get("t") == "done" and pending_more and not more_sent: yield sse({"t": "more_results", "results": pending_more[:more_limit]}) pending_more = [] yield sse(ev) except Exception as e: yield sse({"t": "error", "message": str(e)[:200]}); yield sse({"t": "done"}) _log("searches", {**_query_log_fields(q), "latency_s": round(time.time() - t0, 1), "result_ids": [c.get("doc_id") for c in final[:20]]}) query_log = _query_log_fields(q) print( f"[agent] query_sha256={query_log['query_sha256']} " f"query_chars={query_log['query_chars']} {time.time()-t0:.1f}s", flush=True, ) return StreamingResponse(gen(), media_type="text/event-stream", headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no", "Connection": "keep-alive"}) @app.get("/api/search_stream") def search_stream(q: str, request: Request): return _search_response(q) @app.post("/api/v2/search_stream") def search_stream_v2(req: SearchRequest, request: Request): direct_case = req.route in {"case_lookup", "case_question", "case_lineage"} if not req.approved and not direct_case: return JSONResponse({"error": "query understanding must be approved"}, status_code=409) if direct_case and not _eligible_doc(req.active_case_id or ""): return JSONResponse({"error": "selected judgment is required"}, status_code=409) return _search_response( req.q, original_q=req.original_q, approved_frame=req.search_frame, route=req.route, retrieval_scope=req.retrieval_scope, active_case_id=req.active_case_id, case_question=req.case_question, history=[ turn.model_dump() if hasattr(turn, "model_dump") else turn.dict() for turn in req.history[-6:] ], ) def _graph_card(t, src_dst): tm = C.meta.get(t, {}) return graph_node_card( t, tm, treatment=C.edge_meta.get(src_dst, {}).get("treatment"), cited_by=C.cite_indeg.get(t, 0), good_law_status=C.goodlaw.get(t, {}).get("good_law_status", "unknown"), ) @app.get("/api/deep_search_stream") def deep_search_stream(q: str, request: Request): # the agent IS the unified deep pipeline — alias so the frontend's 'deep' toggle never 404s return _search_response(q) def _eligible_doc(value: str): d = value if value in C.meta else nc2doc.get(value) return d if d and C.is_retrieval_eligible(d) else None def _pdf_aliases(metadata: dict) -> list[str]: return [ value for value in [ metadata.get("neutral_citation"), *(metadata.get("equivalent_citations") or []), ] if value ] @app.get("/api/v2/judgment") @app.get("/api/judgment") def judgment(id: str, q: str = ""): d = _eligible_doc(id) if not d: return JSONResponse({"error": "not found"}, status_code=404) m = C.meta.get(d, {}); jv = C.judgment_view(d) jv["judgment_id"] = str(d) jv["bench"] = m.get("bench"); jv["author_judge"] = m.get("author_judge"); jv["acts"] = m.get("acts") jv["case_number"] = m.get("case_number"); jv["year"] = m.get("year") jv["text_raw"] = jv.get("text", "") aliases = _pdf_aliases(m) pdf_status = PDF_SOURCES.probe(d, aliases=aliases) pdf_public = pdf_status.public_dict() # A mapped individual object is the fast path. Bharat Courts can still # resolve the same public archive by year and identity if that map misses. pdf_public["fallback_available"] = bool( pdf_public.get("fallback_available") or (m.get("year") and (m.get("neutral_citation") or m.get("case_name"))) ) pdf_public["fallback_provider"] = "bharat_courts" pdf_public["route"] = f"/api/v2/pdf?id={d}" jv["pdf"] = pdf_public jv["has_pdf"] = pdf_status.verified # compatibility with cached/older frontends text_provider = m.get("source_provider") or m.get("provider") or "Supreme Court Reports open registry" jv["grounding"] = { "text_available": bool((jv.get("text") or "").strip()), "retrieval_eligible": True, "text_origin": f"judgment text extracted from {text_provider}", "pdf_status": pdf_status.status, "source_name": text_provider, "source_url": m.get("source_url"), } clean_query = re.sub(r"\s+", " ", q or "").strip()[:4000] if clean_query: if hasattr(C, "relevant_passages"): highlights = C.relevant_passages(clean_query, d, k=6) else: highlights = C.case_chat_passages(clean_query, d, k=6) jv["relevance_highlights"] = highlights jv["highlighting"] = { "query_specific": True, "method": "case-local semantic retrieval resolved to stored paragraphs", "grounding": "stored_paragraph_ids_only", } else: jv["relevance_highlights"] = [] jv["highlighting"] = {"query_specific": False, "grounding": "stored_paragraph_ids_only"} jv["corpus_notice"] = ( f"Searched {C.coverage()['accepted_judgments']:,} accepted Supreme Court judgments. " "Unavailable or unmapped judgments were not evaluated." ) if not pdf_status.verified: _log("pdf_sources", {"doc_id": d, "status": pdf_status.status, "reason": pdf_status.reason}) # CITATOR from the citation GRAPH (meta.cases_cited is only ~3% populated): note-up + note-down jv["cited_cases"] = [ _graph_card(t, (d, t)) for t in list(dict.fromkeys(C.out_edges.get(d, []))) if C.is_retrieval_eligible(t) ][:20] jv["citing_cases"] = [ _graph_card(s, (s, d)) for s in sorted(set(C.in_edges.get(d, [])), key=lambda x: -C.cite_indeg.get(x, 0)) if C.is_retrieval_eligible(s) ][:20] if not jv["cited_cases"]: # fallback to the sparse metadata if the graph has nothing jv["cited_cases"] = resolve_cited(m.get("cases_cited"), d) jv["links"] = doc_links(jv.get("text"), d) return JSONResponse(jv) @app.get("/api/v2/judgment/{judgment_id}/paragraphs") def judgment_paragraphs(judgment_id: str, offset: int = 0, limit: int = 50): d = _eligible_doc(judgment_id) if not d: return JSONResponse({"error": "judgment not found"}, status_code=404) limit = max(1, min(int(limit), 100)) offset = max(0, int(offset)) if hasattr(C, "judgment_paragraphs"): return JSONResponse(C.judgment_paragraphs(d, offset=offset, limit=limit)) cis = C.doc_chunks.get(d, []) rows = [ { "paragraph_id": f"{d}:chunk:{ci}", "label": f"Indexed passage {position + 1}", "sequence": position + 1, "text": C.texts[ci], "html_anchor": f"paragraph-{d}-chunk-{ci}", "source_kind": "legacy_chunk", } for position, ci in enumerate(cis[offset:offset + limit], start=offset) if str(C.texts[ci]).strip() ] return JSONResponse({ "judgment_id": str(d), "paragraphs": rows, "offset": offset, "limit": limit, "total": len(cis), "next_offset": offset + len(rows) if offset + len(rows) < len(cis) else None, }) @app.get("/api/v2/graph") def graph(id: str, direction: str = "both", limit: int = 50): d = _eligible_doc(id) if not d: return JSONResponse({"error": "judgment not found"}, status_code=404) direction = direction if direction in {"incoming", "outgoing", "both"} else "both" limit = max(1, min(int(limit), 100)) root = graph_node_card( d, C.meta.get(d, {}), cited_by=C.cite_indeg.get(d, 0), good_law_status=C.goodlaw.get(d, {}).get("good_law_status", "unknown"), ) nodes = {d: root} edges = [] if direction in {"outgoing", "both"}: for target in list(dict.fromkeys(C.out_edges.get(d, []))): if len(edges) >= limit or not C.is_retrieval_eligible(target): continue card = _graph_card(target, (d, target)); nodes[target] = card edge = C.edge_meta.get((d, target), {}) edges.append({ "source_id": str(d), "target_id": str(target), "relation": edge.get("treatment") or "referred_to", "scope": edge.get("scope") or "unknown", "confidence": edge.get("confidence"), "direction": "outgoing", "evidence": edge.get("evidence") or [], }) if direction in {"incoming", "both"}: for source in sorted(set(C.in_edges.get(d, [])), key=lambda x: -C.cite_indeg.get(x, 0)): if len(edges) >= limit or not C.is_retrieval_eligible(source): continue card = _graph_card(source, (source, d)); nodes[source] = card edge = C.edge_meta.get((source, d), {}) edges.append({ "source_id": str(source), "target_id": str(d), "relation": edge.get("treatment") or "referred_to", "scope": edge.get("scope") or "unknown", "confidence": edge.get("confidence"), "direction": "incoming", "evidence": edge.get("evidence") or [], }) return JSONResponse({ "judgment_id": str(d), "root": root, "nodes": list(nodes.values()), "edges": edges, "unresolved_edges_hidden": True, }) @app.post("/api/v2/judgment_chat") @app.post("/api/judgment_chat") def judgment_chat(req: CaseChatRequest): d = _eligible_doc(req.doc_id) if not d: return JSONResponse({"error": "judgment not found"}, status_code=404) question = re.sub(r"\s+", " ", req.question or "").strip() if not question: return JSONResponse({"error": "question is required"}, status_code=400) jv = C.judgment_view(d) summary = jv.get("summary") or {} if not summary.get("available") or not summary.get("text"): return JSONResponse( {"error": "case summary unavailable", "code": "summary_unavailable"}, status_code=409, ) passages = C.case_chat_passages(question, d, k=5) response = A.case_chat_grounded_response( summary["text"], passages, question, [turn.dict() for turn in req.history], jv.get("case_name"), jv.get("neutral_citation"), fast_llm_fn, ) if not response.get("answer"): return JSONResponse({"error": "case chat unavailable"}, status_code=502) _log("judgment_chats", {"doc_id": d, **_query_log_fields(question), "summary_source": summary.get("source")}) return JSONResponse({ "doc_id": d, "judgment_id": str(d), "answer": response["answer"], "evidence": response.get("evidence") or [], "supported": bool(response.get("supported")), "grounded_in": "case_summary_and_stored_passages", "summary_source": summary.get("source"), }) @app.get("/api/v2/pdf/status") @app.get("/api/pdf/status") def pdf_status(id: str, refresh: int = 0): d = _eligible_doc(id) if not d: return JSONResponse({"error": "not found"}, status_code=404) m = C.meta.get(d, {}) status = PDF_SOURCES.probe(d, aliases=_pdf_aliases(m), force=bool(refresh)) public = status.public_dict() public["fallback_available"] = bool( public.get("fallback_available") or (m.get("year") and (m.get("neutral_citation") or m.get("case_name"))) ) public["fallback_provider"] = "bharat_courts" return JSONResponse({"doc_id": d, "judgment_id": str(d), **public}) @app.get("/api/v2/pdf") @app.get("/api/pdf") async def pdf(id: str, dl: int = 0): d = _eligible_doc(id) if not d: return JSONResponse({"error": "not found"}, status_code=404) m = C.meta.get(d, {}) aliases = _pdf_aliases(m) status = PDF_SOURCES.probe(d, aliases=aliases) if status.verified: # Redirect instead of downloading the complete document into the Space. # The public object supports byte ranges used by native PDF viewers. return RedirectResponse( status.url, status_code=307, headers={"Cache-Control": "private, max-age=3600", "X-PDF-Provider": "aws_open_data"}, ) archive = PDF_SOURCES.archive_candidate(d, aliases) try: data, provenance = await resolve_and_fetch_pdf( year=(archive or {}).get("year") or m.get("year"), path=(archive or {}).get("path"), case_name=m.get("case_name") or "", neutral_citation=m.get("neutral_citation") or "", equivalent_citations=m.get("equivalent_citations") or [], decision_date=m.get("date") or "", ) except BharatCourtsPdfError as exc: _log("pdf_sources", {"doc_id": d, "status": "bharat_courts_unavailable", "reason": str(exc)[:200]}) return JSONResponse( { "error": "pdf unavailable", "reason": str(exc)[:300], "pdf_status": "bharat_courts_unavailable", "official_search_url": status.public_dict()["official_search_url"], }, status_code=503 if status.status == "temporarily_unavailable" else 422, ) citation = re.sub(r"[^A-Za-z0-9._-]+", "-", m.get("neutral_citation") or "judgment").strip("-") disposition = "attachment" if dl else "inline" _log("pdf_sources", {"doc_id": d, "status": "verified", "provider": provenance["provider"]}) return Response( content=data, media_type="application/pdf", headers={ "Cache-Control": "private, max-age=86400", "Content-Disposition": f'{disposition}; filename="{citation or "judgment"}.pdf"', "X-PDF-Provider": "bharat_courts", }, ) @app.get("/api/v2/health") def health(): return JSONResponse({ "service": "Moonley API", "status": "ok", "auth_configured": clerk_settings().configured, "api_version": "v2-grounded-preview", "runtime": RUNTIME_KIND, "warm": RUNTIME_WARM, "corpus": C.coverage(), "research_release": RESEARCH_RELEASE, "device": C.device, "pdf_sources": { "mapped_identities": PDF_SOURCES.mapped_count, "primary": "aws_open_data", "fallback": "bharat_courts", }, "project_storage": PROJECTS.status(), "knowledge": KNOWLEDGE.status(), "statute_crosswalk": { "loaded": True, "indexed_directions": C.crosswalk.mapping_count, }, "statute_library": C.statute_library.status(), "drafting_templates": len(DRAFTING.list()), }, headers={"Cache-Control": "no-store"}) @app.get("/api/v2/ready") def ready(): return JSONResponse({ "ready": bool(C.eligible_doc_ids) and RUNTIME_WARM, "api_version": "v2-grounded-preview", "runtime": RUNTIME_KIND, "accepted_judgments": len(C.eligible_doc_ids), "research_release": RESEARCH_RELEASE, }, headers={"Cache-Control": "no-store"}) @app.get("/") def home(): return JSONResponse( {"service": "Moonley API", "status": "ok", "ui": "https://moonley-pilot.vercel.app"}, headers={"Cache-Control": "no-store"}, ) print(f"[serve_agent] boot configured runtime={RUNTIME_KIND}", flush=True)