Spaces:

lothnic
/

precedentai

Running on Zero

Release code against the existing immutable corpus
66bab44 verified 18 days ago
29.6 kB
"""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",
    }