"""FastAPI serving layer for the legal agent (Phase 9A). Endpoints: POST /chat — synchronous single-turn invoke POST /stream — SSE streaming of graph events GET /history/{id} — conversation history from the checkpointer GET /document/{id} — the full reassembled parent judgment POST /review/{id} — resume an interrupted draft (approve/reject) Design notes: - The graph is built once at startup (module scope) — LangGraph graphs are re-entrant across threads; per-conversation state is isolated by thread_id via the checkpointer. - Config comes from the process env / .env (same PrecedentConfig as everywhere else); LEGAL_AI_ENABLE_LLM=true activates the LLM stage. Retrieval defaults to the fused backend, which needs a built ChromaDB index (scripts/build_index.py); /health reports `index`. Set LEGAL_AI_RETRIEVAL_BACKEND=keyword for the offline fixture floor. - The interrupt contract: when the graph pauses for human review, /chat returns {"interrupted": true, ...} and /review resumes it. """ from __future__ import annotations import asyncio import contextlib import logging import tarfile from collections.abc import AsyncIterator from datetime import date from pathlib import Path from typing import Annotated, Any, Literal from uuid import UUID from fastapi import FastAPI, HTTPException, Query, Request from fastapi.exceptions import ResponseValidationError from fastapi.openapi.utils import get_openapi from fastapi.responses import FileResponse, JSONResponse, Response, StreamingResponse from langgraph.types import Command from pydantic import BaseModel, ConfigDict, Field from agent.core import build_graph from agent.supervisor import build_supervisor_graph from api.access import PublicAccess, account_id, conversation_id, session_id from api.accounts import create_router from api.contracts import ( ApprovalReceipt, CancelResponse, CaseResolution, ChatResult, ContractError, CoverageReport, DecisionPage, ErrorResponse, HealthReport, HistoryResponse, Judgment, ReadinessReport, RecoveryConversation, ServiceLimitsReport, SupervisorResponse, UsageReport, public_chat, stream_adapter, stream_frame, ) from api.readiness import Readiness from shared.accounts import HistoryError from shared.approvals import ApprovalError, Approvals, approval_id, review_deadline from shared.config import get_config from shared.database import StorageUnavailable from shared.execution import ExecutionStopped, execution, stop_request from shared.feedback import FeedbackError from shared.recovery import Recovery, RecoveryCheckpointer, RecoveryError from shared.usage_budget import request_budget from tools.corpus_catalog import ( CatalogError, catalog_coverage, catalog_document, catalog_list, catalog_resolve, ) from tools.source_pdf import DOCUMENT_ID, read_source_pdf, source_pdf_path app = FastAPI( title="PrecedentAI", version="0.6.0", responses={ status: {"model": ErrorResponse} for status in (400, 401, 403, 404, 409, 413, 422, 429, 500, 503, 504) }, ) # A dedicated metadata-only event sink; do not enable verbose provider logging. _retrieval_log = logging.getLogger("precedent.retrieval") if not _retrieval_log.handlers: _retrieval_handler = logging.StreamHandler() _retrieval_handler.setFormatter(logging.Formatter("%(message)s")) _retrieval_log.addHandler(_retrieval_handler) _retrieval_log.setLevel(logging.INFO) _retrieval_log.propagate = False def openapi_contract() -> dict[str, Any]: """Include the actual SSE frames alongside ordinary route response schemas.""" if app.openapi_schema is None: schema = get_openapi(title=app.title, version=app.version, routes=app.routes) events = stream_adapter.json_schema(ref_template="#/components/schemas/{model}") schema.setdefault("components", {}).setdefault("schemas", {}).update(events.pop("$defs")) schema["components"]["schemas"]["StreamEvent"] = events schema["paths"]["/stream"]["post"]["responses"]["200"]["content"] = { "text/event-stream": {"schema": {"$ref": "#/components/schemas/StreamEvent"}} } app.openapi_schema = schema return app.openapi_schema app.openapi = openapi_contract _STATIC = Path(__file__).parent / "static" # One graph per process; checkpointer gives /history and /review their memory. _config = get_config() _checkpointer = RecoveryCheckpointer(lambda: _config.quota_database_url.get_secret_value() if _config.quota_database_url else None) app.include_router(create_router(lambda: _config)) app.add_middleware(PublicAccess, config_provider=lambda: _config) _graph = build_graph(config=_config, checkpointer=_checkpointer) _supervisor = build_supervisor_graph(_config) _readiness = Readiness() @app.exception_handler(ContractError) @app.exception_handler(ResponseValidationError) def contract_error(request: Request, exc: Exception) -> JSONResponse: return JSONResponse( { "detail": "The research response was incomplete. Please try again.", "code": "response_contract_invalid", }, status_code=500, ) @app.exception_handler(HistoryError) def history_error(request: Request, exc: HistoryError) -> JSONResponse: return JSONResponse({"detail": str(exc), "code": exc.code}, status_code=exc.status) @app.exception_handler(FeedbackError) def feedback_error(request: Request, exc: FeedbackError) -> JSONResponse: return JSONResponse({"detail": str(exc), "code": "feedback_rejected"}, status_code=exc.status) @app.exception_handler(StorageUnavailable) def storage_error(request: Request, exc: StorageUnavailable) -> JSONResponse: return JSONResponse( {"detail": "Application storage is temporarily unavailable.", "code": "storage_unavailable"}, status_code=503, ) @app.exception_handler(ApprovalError) def approval_error(request: Request, exc: ApprovalError) -> JSONResponse: return JSONResponse({"detail": str(exc), "code": exc.code}, status_code=exc.status) @app.exception_handler(RecoveryError) def recovery_error(request: Request, exc: RecoveryError) -> JSONResponse: return JSONResponse({"detail": str(exc), "code": exc.code}, status_code=exc.status) @app.exception_handler(CatalogError) def inventory_error(request: Request, exc: CatalogError) -> JSONResponse: return JSONResponse( status_code=503, content={ "detail": "The configured document inventories need reconciliation before research can continue.", "code": "source_inventory_invalid", }, ) class ConversationTurn(BaseModel): model_config = ConfigDict(extra="forbid") role: Literal["user", "assistant"] content: str = Field(min_length=1, max_length=2000) class ChatRequest(BaseModel): message: str = Field(min_length=1, max_length=8000) thread_id: str = Field(default="default", min_length=1, max_length=128) history: list[ConversationTurn] = Field(default_factory=list, max_length=12) class ReviewRequest(BaseModel): model_config = ConfigDict(extra="forbid") approved: bool = Field(strict=True) note: str = Field(default="", max_length=2000, strict=True) # Omission supports embedded local callers only. Protected sessions must # bind their decision to the pending interrupt, including before DB rollout. approval_id: str | None = Field(default=None, pattern=r"^[a-f0-9]{64}$") @app.exception_handler(ExecutionStopped) def execution_error(request: Request, exc: ExecutionStopped) -> JSONResponse: return JSONResponse( {"detail": str(exc), "code": exc.code}, status_code=504 if exc.reason == "deadline" else 499, ) @app.post("/cancel/{request_id}", response_model=CancelResponse) def cancel_research(request_id: UUID) -> dict[str, bool]: owner = session_id.get() if not owner: raise HTTPException(status_code=401, detail="Use the authenticated website session.") if not stop_request(owner, request_id): raise HTTPException(status_code=404, detail="No active request belongs to this session.") return {"stopped": True} def _cfg_for(thread_id: str) -> dict[str, Any]: cfg: dict[str, Any] = {"thread_id": conversation_id(thread_id)} account = account_id.get() if account and _config.quota_database_url: settings = Recovery(_config.quota_database_url.get_secret_value()).settings(UUID(hex=account)) if settings["enabled"]: cfg.update(recovery_owner=account, recovery_generation=settings["generation"], recovery_thread_id=thread_id) return {"configurable": cfg} def _review_prompt(task: Any, thread_id: str) -> dict[str, Any]: value = dict(getattr(task, "value", None) or {}) value["approval_id"] = approval_id(conversation_id(thread_id), task.id) return value def _invoke(payload: dict[str, Any], thread_id: str) -> dict[str, Any]: """One graph turn, with interrupt translated into a review prompt.""" state = _graph.invoke(payload, _cfg_for(thread_id)) tasks = state.get("__interrupt__") or [] if tasks: task = tasks[0] return public_chat( { "interrupted": True, "review": _review_prompt(task, thread_id), "thread_id": thread_id, } ) return _public_state(state, thread_id) def _chat_input(req: ChatRequest) -> dict[str, Any]: payload: dict[str, Any] = {"request": req.message} if req.history: state = _graph.get_state(_cfg_for(req.thread_id)) if state.values or state.next: raise HTTPException( status_code=409, detail="Earlier context requires a new conversation." ) # Browser copies are untrusted conversational context, never evidence # or persisted approval decisions. Seed a fresh owner-scoped graph only. payload["history"] = [turn.model_dump() for turn in req.history] return payload def _public_state(state: dict[str, Any], thread_id: str) -> dict[str, Any]: """Project graph state into an API-safe response (drop internals).""" result = state.get("result") or {} ctx = state.get("context") or {} return public_chat( { "interrupted": False, "thread_id": thread_id, "routed_to": result.get("tool") or ctx.get("routed_to"), "routing": ctx.get("routing"), "evaluation": state.get("evaluation"), "result": _with_pdf_links(result), "sources": _with_pdf_links(ctx.get("sources")), } ) def _with_pdf_links(value: Any) -> Any: """Attach local source access to a copy, without mutating graph history.""" if isinstance(value, list): return [_with_pdf_links(item) for item in value] if not isinstance(value, dict): return value result = {key: _with_pdf_links(item) for key, item in value.items()} path = source_pdf_path(value, _config.source_archive_root) if path: result["source_pdf_path"] = path result.pop("source_archive", None) return result @app.get("/") def index() -> FileResponse: """Serve the minimal chat UI (Phase 9C).""" return FileResponse(_STATIC / "index.html") @app.get("/health", response_model=HealthReport) def health() -> dict[str, str]: """Liveness and index-file presence; not a retrieval readiness check.""" from tools.lexical_retrieve import lexical_index_path return { "status": "ok", "llm": str(_config.enable_llm), "web_fallback": str(_config.enable_web_fallback), "backend": _config.retrieval_backend, "index": str((_config.chroma_dir / "chroma.sqlite3").is_file()), "lexical_index": str( lexical_index_path(_config.chroma_dir, _config.chroma_collection).is_file() ), } @app.get("/coverage", response_model=CoverageReport) def coverage() -> dict[str, Any]: """Describe the loaded inventory, including source permissions and gaps.""" try: return catalog_coverage(_config) except FileNotFoundError: raise HTTPException(status_code=503, detail="Document inventory is not available yet") @app.get("/ready", response_model=ReadinessReport, responses={503: {"model": ReadinessReport}}) def readiness(response: Response) -> dict: report = _readiness.get(_config) response.headers["Cache-Control"] = "no-store" if report["status"] in {"warming", "unavailable", "degraded"}: response.status_code = 503 response.headers["Retry-After"] = "5" if report["status"] == "warming" else "30" return report @app.get("/decisions", response_model=DecisionPage) def decisions( start: Annotated[date | None, Query()] = None, end: Annotated[date | None, Query()] = None, court: Annotated[str | None, Query(max_length=100)] = None, limit: Annotated[int, Query(ge=1, le=100)] = 20, offset: Annotated[int, Query(ge=0, le=1_000_000)] = 0, ) -> dict[str, Any]: """List retained decisions in an inclusive decision-date range.""" try: return catalog_list( _config, start=start.isoformat() if start else None, end=end.isoformat() if end else None, court=court, limit=limit, offset=offset, ) except ValueError as exc: raise HTTPException(status_code=422, detail=str(exc)) except FileNotFoundError: raise HTTPException(status_code=503, detail="Document inventory is not available yet") @app.get("/cases/resolve", response_model=CaseResolution) def resolve_named_case(q: Annotated[str, Query(min_length=1, max_length=1000)]) -> dict[str, Any]: try: return catalog_resolve(q, _config) except FileNotFoundError: raise HTTPException(status_code=503, detail="Document inventory is not available yet") @app.post("/chat", response_model=ChatResult) def chat(req: ChatRequest) -> dict[str, Any]: if not req.message.strip(): raise HTTPException(status_code=422, detail="empty message") return _invoke(_chat_input(req), req.thread_id) @app.post("/stream") async def stream(req: ChatRequest) -> StreamingResponse: """SSE stream of graph progress (Phase 8C, event-level not token-level). Events: {"event": "node", "name": ...} per finished node, without raw internal state. A final {"event": "done", ...} carries the same shape as /chat — the finished state, or the interrupt/review payload when the graph paused. The done event carries the state deliberately: the client renders from it instead of calling /chat again, which would re-run the whole graph (and its LLM calls) a second time for every query. Cancellation or deadline emits a terminal error instead of a done event. """ graph_input = await asyncio.to_thread(_chat_input, req) cfg = await asyncio.to_thread(_cfg_for, req.thread_id) async def updates() -> AsyncIterator[str]: # The interrupt arrives as a `__interrupt__` update whose payload is # a LIST of Interrupt objects, not a dict — so it cannot be read # back out of the accumulated node payloads, and it has to be caught # by name as it goes past. interrupt_value: Any = None async for chunk in _graph.astream( graph_input, cfg, stream_mode="updates" ): # stream_mode="updates" yields {node_name: update_dict} per step node_name = next(iter(chunk), "unknown") payload = chunk.get(node_name) if isinstance(chunk, dict) else None if node_name == "__interrupt__" and isinstance(payload, (list, tuple)) and payload: interrupt_value = payload[0] event = { "event": "node", "name": node_name, } yield stream_frame(event) if interrupt_value is not None: done: dict[str, Any] = { "event": "done", **public_chat( { "interrupted": True, "review": _review_prompt(interrupt_value, req.thread_id), "thread_id": req.thread_id, } ), } else: # A checkpoint read, not a re-run. snapshot = await asyncio.to_thread(_graph.get_state, cfg) done = {"event": "done", **_public_state(snapshot.values or {}, req.thread_id)} yield stream_frame(done) async def event_source() -> AsyncIterator[str]: control = execution.get() source = updates() pending = None try: while True: if control: control.check() if pending is None: pending = asyncio.create_task(anext(source)) finished, _ = await asyncio.wait({pending}, timeout=0.2) if not finished: continue try: value = pending.result() except StopAsyncIteration: break pending = None if control: control.check() yield value except ExecutionStopped as exc: yield stream_frame({"event": "error", "code": exc.code, "detail": str(exc)}) except ContractError as exc: yield stream_frame( {"event": "error", "code": "response_contract_invalid", "detail": str(exc)} ) finally: if pending is not None and not pending.done(): if control: control.stop() pending.cancel() if pending is not None: with contextlib.suppress( asyncio.CancelledError, ExecutionStopped, ContractError, StopAsyncIteration ): await pending await source.aclose() return StreamingResponse( event_source(), media_type="text/event-stream", headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}, ) def _safe_json(value: Any) -> Any: """Best-effort JSON-ification for SSE payloads.""" if value is None or isinstance(value, (str, int, float, bool)): return value if isinstance(value, dict): return {k: _safe_json(v) for k, v in value.items()} if isinstance(value, (list, tuple)): return [_safe_json(v) for v in value] return str(value) @app.get("/history/{thread_id}", response_model=HistoryResponse) def history(thread_id: str) -> dict[str, Any]: snapshot = _graph.get_state(_cfg_for(thread_id)) values = snapshot.values or {} return { "thread_id": thread_id, "history": values.get("history", []), "status": values.get("status"), } # Judgment ids as the corpus writes them ("2017-INSC-801", "const-art-21"). # Checked before any lookup, so a path segment can only ever reach the corpus # layer as a bounded id. _PARENT_ID = DOCUMENT_ID @app.get("/document/{parent_id}", response_model=Judgment) def document(parent_id: str) -> dict[str, Any]: """The full judgment a retrieved chunk came from, reassembled. Retrieval ranks CHUNKS, so a source card can only ever show one slice of a judgment. The console's source inspector reads this endpoint to put the whole case next to the slice that matched — the chunk is what ranked, the judgment is what it means. 404 when the corpus is not on disk or the id is not in it; the caller still holds the chunk, which is the evidence that matched. """ if not _PARENT_ID.fullmatch(parent_id): raise HTTPException(status_code=422, detail="malformed judgment id") record = catalog_document(_config, parent_id) if record is None: raise HTTPException(status_code=404, detail="unknown judgment") text = str(record.get("text") or "") meta = { key: record[key] for key in ( "parent_id", "case_id", "case_id_aliases", "citation_aliases", "neutral_citation_aliases", "title_aliases", "title", "citation", "court", "year", "decision_date", "judge", "document_id", "document_type", "edition_as_of", "article_number", "constitution_group", "provision_status", "source_section", "case_number", "case_group_id", "proceeding_id", "proceeding_basis", "neutral_citation", "source_url", "original_pdf_url", "pdf_sha256", "text_sha256", "normalization_version", "metadata_date_status", "decision_date_original", "pdf_header_date", "source_revision", "source_archive_sha256", "case_number_original", "case_number_status", "corpus_snapshot", ) if record.get(key) is not None } path = source_pdf_path(record, _config.source_archive_root) if path: meta["source_pdf_path"] = path return {**meta, "text": text, "characters": len(text)} @app.get("/document/{parent_id}/pdf") def document_pdf(parent_id: str) -> Response: """Serve the retained PDF edition used during ingestion, verified by hash.""" if not _PARENT_ID.fullmatch(parent_id): raise HTTPException(status_code=422, detail="malformed judgment id") record = catalog_document(_config, parent_id) if record is None: raise HTTPException(status_code=404, detail="unknown judgment") try: data = read_source_pdf(record, _config.source_archive_root) except (FileNotFoundError, OSError) as exc: raise HTTPException(status_code=404, detail="retained source PDF unavailable") from exc except (ValueError, tarfile.TarError) as exc: raise HTTPException( status_code=409, detail="retained source PDF failed verification" ) from exc return Response( content=data, media_type="application/pdf", headers={ "Content-Disposition": 'inline; filename="judgment.pdf"', "ETag": f'"{record["pdf_sha256"]}"', "Cache-Control": "private, max-age=86400, immutable", "X-Content-Type-Options": "nosniff", }, ) @app.post("/supervise", response_model=SupervisorResponse) def supervise(req: ChatRequest) -> dict[str, Any]: """Multi-agent supervised run (Phase 8B): researcher/analyst/drafter orchestrated by an LLM planner (deterministic fallback). One-shot: no thread state, no interrupts — the drafter's review notice rides in the payload and drafting stays human-reviewed downstream.""" if not req.message.strip(): raise HTTPException(status_code=422, detail="empty message") out = _supervisor.invoke({"request": req.message}) final = out.get("final") or {} final["evaluation"] = out.get("evaluation") final["thread_id"] = req.thread_id return _with_pdf_links(final) @app.post("/review/{thread_id}", response_model=ChatResult) def review(thread_id: str, req: ReviewRequest) -> dict[str, Any]: """Consume the exact pending decision before resuming any drafting work.""" store = None budget = request_budget.get() if _config.quota_database_url is not None: if budget is None or budget.owner != session_id.get(): raise ApprovalError("A server-owned request is required.", code="approval_owner_invalid", status=401) store = Approvals(_config.quota_database_url.get_secret_value()) if req.approval_id: store.check_unused(budget.owner, conversation_id(thread_id), req.approval_id, approved=req.approved, note=req.note) snapshot = _graph.get_state(_cfg_for(thread_id)) pending = [item for task in snapshot.tasks for item in task.interrupts] if snapshot.next != ("act",) or len(pending) != 1: raise HTTPException(status_code=409, detail="no interrupted task on this thread") prompt = _review_prompt(pending[0], thread_id) if req.approval_id != prompt["approval_id"] and (req.approval_id or session_id.get()): raise ApprovalError("This decision belongs to a different review. Request a new draft.", code="approval_id_mismatch") deadline = review_deadline(prompt) if store: store.claim(budget.owner, conversation_id(thread_id), prompt["approval_id"], budget.request_id, approved=req.approved, note=req.note, expires_at=deadline) try: state = _graph.invoke( Command(resume={"approved": req.approved, "note": req.note}), _cfg_for(thread_id), ) result = _public_state(state, thread_id) except Exception: if store: # Accepted is already a non-reusable receipt even if settlement # cannot reach storage. Never refund authorization on an exception. store.settle(budget.owner, prompt["approval_id"], completed=False) raise if store: store.settle(budget.owner, prompt["approval_id"], completed=True) return result @app.get("/review/{thread_id}", response_model=ApprovalReceipt) def review_receipt(thread_id: str, approval_id: str = Query(pattern=r"^[a-f0-9]{64}$")) -> dict: """Read metadata after an uncertain response; this never resumes the graph.""" if _config.quota_database_url is None: raise HTTPException(status_code=503, detail="Durable review receipts are not configured.") return Approvals(_config.quota_database_url.get_secret_value()).lookup( session_id.get(), conversation_id(thread_id), approval_id, ) @app.get("/recovery/{thread_id}", response_model=RecoveryConversation) def recover_conversation(thread_id: str) -> dict: """Read retained state; never rerun interrupted research or consume approval.""" if not account_id.get() or not _config.quota_database_url: raise RecoveryError("Sign in to recover a conversation.", code="recovery_signin_required", status=401) cfg = _cfg_for(thread_id) if not cfg["configurable"].get("recovery_owner"): raise RecoveryError("Enable temporary recovery first.", code="recovery_disabled") snapshot = _graph.get_state(cfg) if not snapshot.values: raise RecoveryError("No recent state belongs to this conversation.", code="recovery_not_found", status=404) pending = [item for task in snapshot.tasks for item in task.interrupts] result = None status = "interrupted" if snapshot.next == ("act",) and len(pending) == 1: prompt = _review_prompt(pending[0], thread_id) try: review_deadline(prompt) Approvals(_config.quota_database_url.get_secret_value()).lookup( session_id.get(), conversation_id(thread_id), prompt["approval_id"], ) except ApprovalError as exc: if exc.code == "approval_not_found": result = public_chat({"interrupted": True, "review": prompt, "thread_id": thread_id}) status = "review_required" elif exc.code == "approval_expired": status = "review_expired" else: raise else: status = "decision_consumed" elif not snapshot.next and snapshot.values.get("status") == "done": result = _public_state(snapshot.values, thread_id) status = "completed" return {"thread_id": thread_id, "request": snapshot.values.get("request") or "", "status": status, "history": snapshot.values.get("history") or [], "result": result} @app.get("/limits", response_model=ServiceLimitsReport) def local_limits() -> dict[str, Any]: """PublicAccess supplies owner-scoped limits for authenticated proxy calls.""" if _config.enable_web_fallback: from tools.web_research import memory_limits return { "persistence": "memory", "limits": memory_limits(_config), "note": "The shared web allowance resets when this development backend restarts. Public turn limits are disabled.", } return { "persistence": "disabled", "limits": {}, "note": "Public access limits are not enabled in this local configuration.", } @app.get("/usage", response_model=UsageReport) def usage() -> dict[str, Any]: """LLM usage ledger for this process: per-model token totals and ESTIMATED dollar cost (MODEL_PRICES are point-in-time figures — treat the dollars as indicative, the token counts as exact). Reset only on process restart.""" from tools.llm_client import get_usage report = get_usage() return { "calls": report.calls, "failed_calls": report.failed_calls, "prompt_tokens": report.prompt_tokens, "completion_tokens": report.completion_tokens, "total_tokens": report.total_tokens, "cost_usd_estimate": report.cost_usd, "by_model": report.by_model, "note": "dollar figures are point-in-time estimates; token counts are exact", }