src/endopath/api.py

"""FastAPI backend: serves extraction results + page images, accepts
confirm/edit actions (docs/prd.md section 8.4), and -- when the frontend has been
built -- serves the SPA itself so the whole app lives on one origin.

Every JSON endpoint lives under `/api`. That prefix is load-bearing rather
than cosmetic: the SPA's client-side routes `/dashboard`, `/export`, and
`/cases/:barcode` are character-for-character the same paths as three of the
API's own routes, so serving both from one origin at the root would make a
deep link to `/dashboard` return JSON instead of the app.

Endpoints that spend money or change reviewer state require a per-user login
(see `authorize`, #128). Reads stay open: the corpus is public, de-identified
TCGA data, and the SPA has to render before anyone can authenticate.
"""

from __future__ import annotations

import json
import logging
import os
import re
import sqlite3
import uuid
from contextlib import asynccontextmanager
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Optional

from fastapi import APIRouter, Depends, FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse, RedirectResponse
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from pydantic import BaseModel, Field, ValidationError
from sqlalchemy import text

from endopath import auth, colpali_retrieval, fields, precomputed_retrieval, taxonomy, textsource
from endopath.auth import Identity
from endopath.dictionary import DataDictionary
from endopath.schema import CaseStatus, EndometrialChecklist, EvidenceSpan
from endopath.storage import (
    DEFAULT_DB_PATH,
    DEFAULT_PROJECT_ID,
    connect,
    count_users,
    create_project,
    get_case,
    get_user,
    list_cases,
    list_projects,
    upsert_case,
    upsert_user,
)

logger = logging.getLogger(__name__)

IMAGE_DIR = Path("data") / "images"

FRONTEND_DIST_ENV = "ENDOPATH_FRONTEND_DIST"

# Hugging Face sets SPACE_ID in every Space container. Used only to tell "this
# process is reachable from the public internet" apart from "this is someone's
# laptop", so authentication can be optional in one case and mandatory in the other.
SPACE_ENV = "SPACE_ID"

_bearer_scheme = HTTPBearer(auto_error=False)


def authorize(
    credentials: Optional[HTTPAuthorizationCredentials] = Depends(_bearer_scheme),
) -> Optional[Identity]:
    """Resolve the authenticated account for a write endpoint, or None when auth is disabled (#128).

    Expects `Authorization: Bearer <session-token>`, the token endopath.auth issues at login. With no
    session secret configured the gate is open and this returns None, which keeps local dev and the test
    suite free of ceremony; that default is only safe because a Space without a secret refuses to start
    (see the lifespan below). With a secret set, a missing, malformed, or expired token is a 401, and a
    valid one resolves to the Identity the endpoints read the confirmer and role from.
    """
    secret = auth.session_secret()
    if not secret:
        return None
    token = credentials.credentials if credentials is not None else ""
    identity = auth.decode_token(token, secret)
    if identity is None:
        raise HTTPException(
            status_code=401,
            detail="missing or invalid session token",
            headers={"WWW-Authenticate": "Bearer"},
        )
    return identity


def _attribution(
    identity: Optional[Identity], body_confirmer: Optional[str], body_role: Optional[str]
) -> tuple[Optional[str], Optional[str]]:
    """The confirmer name and role recorded on a decision. When authenticated, the account's, ignoring any
    body-supplied values, so attribution is non-repudiable (PRD section 4.4). When auth is disabled (local
    dev, the test suite), the body-supplied session identity, preserving the open-path behavior."""
    if identity is not None:
        return identity.confirmer, identity.role
    return body_confirmer, body_role


def _require_licence(identity: Optional[Identity], licence_required: bool) -> None:
    """A licence_required decision may only be confirmed by an account holding a pathology licence (PRD
    section 4.4). Enforced only when authenticated; the open dev path does not gate on licence."""
    if identity is not None and licence_required and not identity.holds_licence:
        raise HTTPException(
            status_code=403,
            detail="confirming this requires a pathology licence; the signed-in account holds none",
        )


def _seed_users() -> None:
    """Provision the accounts named in ENDOPATH_SEED_USERS into the store at startup, so a fresh
    deployment has logins. Upserts, so a redeploy rotates a seeded password in place. A malformed seed
    raises, failing the boot loudly rather than starting an auth-enabled Space with no way to log in."""
    seeds = auth.parse_seed_users(os.environ.get(auth.SEED_USERS_ENV, ""))
    if not seeds:
        return
    conn = connect(_db_path)
    try:
        for seed in seeds:
            upsert_user(
                conn,
                username=seed.username,
                password_hash=auth.hash_password(seed.password),
                role=seed.role,
                holds_licence=seed.holds_licence,
                display_name=seed.display_name,
            )
        total = count_users(conn)
    finally:
        conn.close()
    logger.info("seeded %d reviewer account(s); %d total", len(seeds), total)


@asynccontextmanager
async def lifespan(_app: FastAPI):
    if auth.session_secret():
        _seed_users()
        logger.info(
            "%s is set; write endpoints require a reviewer login", auth.SESSION_SECRET_ENV
        )
    elif os.environ.get(SPACE_ENV):
        # Fail closed. A public Space with no session secret would let any visitor spend the Anthropic
        # key via /process, so refuse to boot rather than serve.
        raise RuntimeError(
            f"{auth.SESSION_SECRET_ENV} is not set but {SPACE_ENV} is: refusing to expose "
            "unauthenticated extraction and review endpoints on a public URL. Set "
            f"{auth.SESSION_SECRET_ENV} and seed accounts via {auth.SEED_USERS_ENV} as Space Secrets, "
            "or make the Space private."
        )
    else:
        logger.warning(
            "%s is not set: extraction and review endpoints are unauthenticated. "
            "Fine locally, not for a deployment reachable from the internet.",
            auth.SESSION_SECRET_ENV,
        )

    # Record every model call's provenance (#116) to the same store, so the export bundle can summarize
    # what the cohort cost. Configured before the worker starts, so the worker thread's calls record too.
    from endopath import llm, provenance

    llm.configure_recorder(provenance.make_recorder(_db_path))

    # Drain the Intake follow-up queue in the background (report induction #92,
    # dictionary compilation #91). Disabled under pytest, so the suite drives the
    # executors synchronously and never starts a thread.
    from endopath import worker

    if worker.start(_db_path):
        logger.info("Intake follow-up worker started against %s", _db_path)
    try:
        yield
    finally:
        worker.stop()


app = FastAPI(title="Provinans API", lifespan=lifespan)
api = APIRouter(prefix="/api")

# Applied to endpoints that spend money (the Anthropic key), spend a lot of CPU,
# or mutate the reviewer's dataset. New state-changing routes (the #25 upload
# endpoint) belong on this list too.
authenticated = [Depends(authorize)]

# Two-port local dev (vite :5173 + api :8000) is cross-origin and still needs
# CORS. The deployed Space serves the built SPA from this same app, so there
# the browser never makes a cross-origin request and this middleware is inert.
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:5173", "http://127.0.0.1:5173"],
    allow_methods=["*"],
    allow_headers=["*"],
)

_db_path = DEFAULT_DB_PATH
_frontend_dist = Path(os.environ.get(FRONTEND_DIST_ENV, "frontend/dist"))


def configure_db_path(path: Path | str) -> None:
    """Test/CLI hook to point the API at a different SQLite file."""
    global _db_path
    _db_path = Path(path)


def configure_frontend_dist(path: Path | str) -> None:
    """Test/CLI hook to point the SPA fallback at a different build directory."""
    global _frontend_dist
    _frontend_dist = Path(path)


def get_db():
    conn = connect(_db_path)
    try:
        yield conn
    finally:
        conn.close()


@api.get("/trace")
def model_trace(limit: int = 50, conn: sqlite3.Connection = Depends(get_db)) -> dict:
    """The live model-call trace (#221): the most recent LLM calls with the requested and the resolved
    model, token usage, latency, and the Anthropic request id, plus the running aggregate. Read only,
    metadata only. A viewer watches this to confirm the app calls the model it claims to."""
    from endopath import provenance

    return {"calls": provenance.recent(conn, limit), "summary": provenance.summary(conn)}


@api.get("/coverage")
def coverage(project: Optional[str] = None, conn: sqlite3.Connection = Depends(get_db)) -> dict:
    """The Field coverage view (#237): a project's target dictionary, DB-backed, with the record type that
    fulfils each field and its fill config. The live fill is computed in the browser from /api/dashboard and
    /api/longitudinal/cohort. Seeds the default project's coverage on first read."""
    from endopath import coverage as coverage_mod
    from endopath import coverage_data

    coverage_mod.seed_coverage(conn)
    return coverage_mod.build_coverage(conn, project or coverage_data.PROJECT_ID)


class ReportSource(BaseModel):
    """Where the reports come from. `prepared_corpus` walks the pre-fetched TCGA
    UCEC corpus; `folder` ingests a local directory of report PDFs (#25)."""

    kind: str = "prepared_corpus"
    folder_path: Optional[str] = None


class IngestionRunRequest(BaseModel):
    """One Intake submission: a report source, a target dictionary, and the text
    source that reads each report. All optional, so a bare POST still runs the
    prepared corpus against the built-in CAP checklist."""

    source: ReportSource = Field(default_factory=ReportSource)
    dictionary: Optional[DataDictionary] = None
    text_source: Optional[str] = None
    limit: Optional[int] = None


def _default_text_source_for(source_kind: str) -> str:
    """The prepared corpus already carries recognition text, so it reads through
    corpus OCR by default; a folder source has none, so it reads the rendered
    pages with the vision model, the MVP path. Either is overridable per
    submission."""
    return textsource.CORPUS_OCR if source_kind == "prepared_corpus" else textsource.DEFAULT_SOURCE_ID


@api.post("/ingestion/run", dependencies=authenticated)
def trigger_ingestion(
    limit: Optional[int] = None,
    body: Optional[IngestionRunRequest] = None,
    conn: sqlite3.Connection = Depends(get_db),
) -> dict:
    """Intake screen action (docs/prd.md section 8.1): point the tool at a report
    source and a target dictionary, ingest, and enqueue the follow-up jobs for
    that pair (report induction #92, dictionary compilation #91).

    The report's text is read through a pluggable text source (docs/prd.md
    section 7, "Two points plug out"): corpus OCR, the PDF text layer (#45), or a
    vision model read of the rendered pages. This call site names no provider;
    it selects a source by id, so a local on-premise model swaps in unchanged.

    The corpus pipeline is idempotent, so re-running it (re-clicking Intake) is
    safe and cheap once the corpus is already local.
    """
    from pathlib import Path as _Path

    import pandas as pd

    from endopath import dictionary as dictionary_mod
    from endopath import ingestion, jobs

    req = body or IngestionRunRequest()
    effective_limit = req.limit if req.limit is not None else limit
    source = req.source
    data_dictionary = dictionary_mod.register(
        conn, req.dictionary or dictionary_mod.builtin_cap_dictionary()
    )

    source_id = req.text_source or _default_text_source_for(source.kind)
    try:
        text_source = textsource.select(source_id)
    except ValueError as exc:
        raise HTTPException(status_code=422, detail=str(exc)) from exc

    if source.kind == "folder":
        if not source.folder_path:
            raise HTTPException(status_code=422, detail="folder source requires folder_path")
        folder = _Path(source.folder_path)
        try:
            results_df = ingestion.run_folder_pipeline(folder, limit=effective_limit)
        except (NotADirectoryError, FileNotFoundError) as exc:
            raise HTTPException(status_code=422, detail=str(exc)) from exc
        # A folder case has no corpus recognition row, so seed with an empty
        # manifest (no network) and let the text source read the PDF or pages.
        empty_manifest = pd.DataFrame({"case_barcode": [], "text": []})
        seeded = ingestion.seed_cases_into_store(
            conn, results_df, manifest=empty_manifest, text_source=text_source, folder=folder
        )
    elif source.kind == "prepared_corpus":
        results_df = ingestion.run_pipeline(limit=effective_limit)
        seeded = ingestion.seed_cases_into_store(conn, results_df, text_source=text_source)
    else:
        raise HTTPException(status_code=422, detail=f"unknown report source kind: {source.kind!r}")

    case_count = len(results_df)
    failures = int(results_df.error.notna().sum()) if case_count else 0
    enqueued = jobs.enqueue_intake_followups(
        conn,
        dictionary_id=data_dictionary.id,
        source_kind=source.kind,
        case_count=case_count,
        variable_count=len(data_dictionary.variables),
    )
    return {
        "ingested": case_count,
        "failures": failures,
        "seeded": seeded,
        "source": source.model_dump(),
        "text_source": {"id": text_source.id, "label": text_source.label},
        "dictionary": {
            "id": data_dictionary.id,
            "title": data_dictionary.title,
            "variable_count": len(data_dictionary.variables),
            "licence_required_variables": data_dictionary.licence_required_variables,
        },
        "enqueued": [job.model_dump() for job in enqueued],
    }


@api.get("/ingestion/sources")
def ingestion_text_sources() -> list[dict]:
    """The selectable text sources for the Intake screen (id + label). An open
    read: the SPA renders the picker before anyone authenticates."""
    return textsource.available_sources()


@api.get("/ingestion/dictionary")
def builtin_dictionary() -> dict:
    """The built-in CAP endometrium checklist as a dictionary, the default
    target an Intake submission points at. An open read, so the SPA can render
    the dictionary preview before a reviewer has a token."""
    from endopath import dictionary as dictionary_mod

    return dictionary_mod.builtin_cap_dictionary().model_dump()


@api.get("/ingestion/jobs")
def ingestion_jobs(conn: sqlite3.Connection = Depends(get_db)) -> list[dict]:
    """The follow-up jobs Intake submissions have enqueued (report induction
    #92, dictionary compilation #91), read back from the durable queue. An open read."""
    from endopath import jobs

    return [job.model_dump() for job in jobs.pending(conn)]


@api.get("/fields")
def checklist_fields() -> list[dict]:
    """The checklist definition, so the SPA renders a schema it is handed
    rather than a hand-mirrored copy (#37). Order is meaningful: it is the
    order Case Review lays out evidence cards.

    `enum_values` is served alongside the type so the edit path can offer a
    dropdown of the permitted values instead of a free-text box.
    """
    return [
        {
            "name": spec.name,
            "label": spec.label,
            "value_type": spec.value_type,
            "enum_values": spec.enum_values,
        }
        for spec in fields.FIELDS
    ]


@api.get("/concepts")
def concepts() -> dict:
    """The canonical concept list (#91) with its registry evidence, dependency
    relationships, and decisions joined on (#35), for the registry browser
    (#94, docs/prd.md section 8.1). The concept spine is
    `mapping.concept_list()`, served rather than hand-mirrored, so the browser
    never diverges from the exported column spec. An open read: the SPA renders
    the browser before anyone authenticates.
    """
    from endopath import concept_registry

    return concept_registry.concept_registry()


@api.get("/mapping")
def mapping(dictionary: Optional[str] = None, conn: sqlite3.Connection = Depends(get_db)) -> dict:
    """The crosswalk mapping-review surface (#95, #109, docs/prd.md sections 8.1,
    3; design-guideline section 9.1): the drafted edges from a source dictionary's
    fields to the canonical concepts, colored by category, with the unresolved and
    low-confidence edges marked as the reviewer queue and each queued edge carrying
    its maieutic question and two candidate resolutions.

    `dictionary` selects a submitted dictionary's drafted crosswalk (#108); omitted,
    it serves the built-in CAP `crosswalk.reference_crosswalk()`, the committed
    reference instance. Joined to the confirmations recorded so far. An open read:
    the SPA renders the diagram before anyone authenticates.
    """
    from endopath import mapping_review

    try:
        return mapping_review.mapping_review(conn, dictionary)
    except mapping_review.MappingReviewError as exc:
        # The dictionary was never submitted, or its crosswalk has not been drafted yet.
        raise HTTPException(status_code=404, detail=str(exc)) from exc


@api.get("/dictionaries")
def dictionaries(conn: sqlite3.Connection = Depends(get_db)) -> list[dict]:
    """The dictionaries a reviewer can open on the mapping screen: the built-in CAP
    checklist always, then any submitted dictionary, each flagged with whether its
    crosswalk has been drafted yet (#109). An open read."""
    from endopath import dictionary as dictionary_mod
    from endopath import storage as storage_mod

    out = [
        {
            "id": dictionary_mod.BUILTIN_DICTIONARY_ID,
            "title": dictionary_mod.BUILTIN_DICTIONARY_TITLE,
            "builtin": True,
            "ready": True,
        }
    ]
    for row in storage_mod.list_dictionaries(conn):
        if row["id"] == dictionary_mod.BUILTIN_DICTIONARY_ID:
            continue
        ready = storage_mod.get_submission_artifact(conn, row["id"], "crosswalk_draft") is not None
        out.append({"id": row["id"], "title": row["title"], "builtin": False, "ready": ready})
    return out


class LoginRequest(BaseModel):
    """A reviewer's credentials at login (#128)."""

    username: str
    password: str


@api.post("/login")
def login(body: LoginRequest, conn: sqlite3.Connection = Depends(get_db)) -> dict:
    """Exchange a username and password for a signed session token (#128). The token carries the account's
    role and licence, which the write endpoints read the confirmer and role from, so a confirmation's
    attribution is the authenticated account's rather than a self-declared field. Only meaningful when auth
    is enabled; with no session secret the write endpoints are open and no login is needed."""
    secret = auth.session_secret()
    if not secret:
        raise HTTPException(status_code=409, detail="authentication is not enabled on this deployment")
    user = get_user(conn, body.username)
    if user is None or not auth.verify_password(body.password, user["password_hash"]):
        # One message for an unknown user and a wrong password, so the response does not disclose which
        # usernames exist.
        raise HTTPException(status_code=401, detail="invalid username or password")
    identity = Identity(
        username=user["username"],
        role=user["role"],
        holds_licence=user["holds_licence"],
        display_name=user["display_name"],
    )
    return {
        "token": auth.issue_token(identity, secret),
        "username": identity.username,
        "display_name": identity.display_name,
        "role": identity.role,
        "holds_licence": identity.holds_licence,
    }


# ============================================================================
# Longitudinal Records & Joins (multi-record clinical data harmonization)
# ============================================================================

class LongitudinalRecord(BaseModel):
    """A single longitudinal event (follow-up, treatment, recurrence, vital status)."""

    record_id: str
    participant_id: str
    case_barcode: Optional[str] = None
    record_type: str
    record_index: int
    days_to_followup: Optional[int] = None
    days_to_treatment: Optional[int] = None
    days_to_recurrence: Optional[int] = None
    biopsy_result: Optional[str] = None
    followup_treatment_success: Optional[str] = None
    recurrence_type: Optional[str] = None
    therapy_type: Optional[str] = None
    structured_data: dict[str, Any] = {}


class LongitudinalJoin(BaseModel):
    """A confirmed association between a case and a longitudinal record."""

    join_id: str
    case_barcode: str
    record_id: str
    disposition: str  # confirmed, rejected, pending, uncertain
    confirmed_by: Optional[str] = None
    role: Optional[str] = None
    confirmed_at: Optional[str] = None
    note: Optional[str] = None
    question: Optional[str] = None


class JoinRequest(BaseModel):
    """Request to confirm/reject a longitudinal record with a case."""

    record_id: str
    disposition: str
    note: Optional[str] = None
    question: Optional[str] = None


class LinkageRequest(BaseModel):
    """Request to affirm a case's whole GDC longitudinal record in one action (#180)."""

    note: Optional[str] = None


# ============================================================================

class MappingComponentInput(BaseModel):
    """One component concept a split resolves a source field to (#103 follow-on):
    the canonical concept and how the field carries it. Never 'unresolved'."""

    concept_id: str
    relation: str


class MappingConfirmRequest(BaseModel):
    """One reviewer's decision on a crosswalk edge (#95, #103): the edge, the
    chosen concept and relation (concept null for an out-of-scope resolution),
    the per-variable licence-required flag, and the decision-ledger context.
    `disposition` is `confirmed`, `flagged`, or `deferred`; `decided_via` names
    how the answer was reached; `question` and `options` record what the decision
    weighed. `relation` is optional for a flag or a deferral, which resolves to none.

    `components`, when present, splits the field across several canonical concepts
    (#103 follow-on): `concept_id` and `relation` become the derived headline, and
    each component is a primitive the field also resolves to. A split is always a
    `confirmed` resolution.

    `confirmer` and `role` are optional (#128): when auth is enabled the server
    reads them from the authenticated account and ignores the body, so a
    confirmation's attribution is non-repudiable; the open dev path still reads
    them from the body-supplied session identity."""

    mapping_id: str
    concept_id: Optional[str] = None
    relation: str = "unresolved"
    confirmer: Optional[str] = None
    role: Optional[str] = None
    note: Optional[str] = None
    licence_required: bool = False
    disposition: str = "confirmed"
    decided_via: Optional[str] = None
    question: Optional[str] = None
    options: Optional[list] = None
    components: Optional[list[MappingComponentInput]] = None


@api.post("/mapping/confirm", dependencies=authenticated)
def confirm_mapping(
    body: MappingConfirmRequest,
    conn: sqlite3.Connection = Depends(get_db),
    identity: Optional[Identity] = Depends(authorize),
) -> dict:
    """Record one reviewer's decision on a crosswalk edge (#95, #103): the
    disposition, the resolved concept and relation, the confirmer and their
    role, the licence-required flag, the time, and the ledger context. Any edge
    is decidable, not only the queued ones. Resolving the last queued edge, with
    nothing flagged or deferred, confirms the whole mapping, at which point
    `GET /api/mapping` exposes the confirmed field set (#96) reads.

    When auth is enabled (#128) the confirmer and role are the authenticated
    account's, not the body's, and a licence_required edge is refused unless the
    account holds a pathology licence. On the open dev path the body-supplied
    session identity is used, as before.
    """
    from endopath import mapping_review

    _require_licence(identity, body.licence_required)
    confirmer, role = _attribution(identity, body.confirmer, body.role)

    try:
        if body.components:
            mapping_review.confirm_split(
                conn,
                mapping_id=body.mapping_id,
                headline_concept_id=body.concept_id,
                headline_relation=body.relation,
                components=[c.model_dump() for c in body.components],
                confirmer=confirmer,
                role=role,
                note=body.note,
                licence_required=body.licence_required,
                decided_via=body.decided_via,
                question=body.question,
                options=body.options,
            )
        else:
            mapping_review.confirm_edge(
                conn,
                mapping_id=body.mapping_id,
                concept_id=body.concept_id,
                relation=body.relation,
                confirmer=confirmer,
                role=role,
                note=body.note,
                licence_required=body.licence_required,
                disposition=body.disposition,
                decided_via=body.decided_via,
                question=body.question,
                options=body.options,
            )
    except mapping_review.MappingReviewError as exc:
        raise HTTPException(status_code=422, detail=str(exc)) from exc

    return mapping_review.mapping_review(conn)


class MappingChatTurn(BaseModel):
    """One prior turn of an edge chat: the speaker (`user` or `assistant`) and
    the text, so a follow-up question carries its context."""

    role: str
    text: str


class MappingChatRequest(BaseModel):
    """A reviewer's question about one crosswalk edge (#103), with the prior
    turns of that edge's conversation."""

    mapping_id: str
    question: str
    history: list[MappingChatTurn] = Field(default_factory=list)


@api.post("/mapping/edge/chat", dependencies=authenticated)
def mapping_edge_chat(
    body: MappingChatRequest, conn: sqlite3.Connection = Depends(get_db)
) -> dict:
    """Reply to a reviewer's message about one crosswalk edge (#103): a concise
    answer and, when the message points at a better resolution, one concrete remap
    (a re-target or a split) the reviewer can apply. The exchange resolves into a
    filed decision when the reviewer applies the proposal through
    `POST /api/mapping/confirm` with `decided_via='chat'`.

    Gated by the shared token: the call spends the Anthropic key. The chat model
    is named explicitly in `mapping_review` (claude-sonnet-5), behind a
    provider-agnostic seam a fake substitutes for in the fast test suite.
    """
    from endopath import mapping_review

    try:
        return mapping_review.explain_edge(
            conn,
            mapping_id=body.mapping_id,
            question=body.question,
            history=[turn.model_dump() for turn in body.history],
        )
    except mapping_review.MappingReviewError as exc:
        raise HTTPException(status_code=422, detail=str(exc)) from exc
    except RuntimeError as exc:  # e.g. ANTHROPIC_API_KEY not set (llm_extraction.get_client)
        raise HTTPException(status_code=503, detail=str(exc)) from exc


@api.get("/dashboard")
def dashboard(
    project: Optional[str] = None,
    conn: sqlite3.Connection = Depends(get_db),
) -> dict:
    """The cohort dashboard (#100, docs/prd.md sections 8.1, 4.3, 4.5, 4.7): the completion rate and
    worst-field-first summary, then per field the confirmed-value and coded-absence-reason distributions
    and the absence-versus-grade association, and per case the absent variables that would require ordering
    slides. `project` scopes the whole payload to one project's cases (#149); omitted, the whole cohort is
    summarized. An open read: the SPA renders the dashboard before anyone authenticates.
    """
    from endopath.dashboard import cohort_dashboard

    return cohort_dashboard(conn, project)


@api.get("/export")
def export(
    staging_edition: Optional[str] = None,
    project: Optional[str] = None,
    conn: sqlite3.Connection = Depends(get_db),
) -> dict:
    """The cohort export (#101, docs/prd.md sections 3, 8.1): the confirmed case records projected onto
    the canonical concept list as the column spec, one row per case, each value column carrying a parallel
    coded absent-reason column, alongside the reproduction bundle and the three rendered download files.

    One projection feeds the on-screen preview and the CSV, XLSX, and JSON downloads, so the files and the
    preview never disagree. `staging_edition` names the edition the researcher chose at export (#40); the
    FIGO stage is re-projected under it through `taxonomy.derive_stage`, rendering a determined code, a
    constrained candidate set, or an indeterminate reason distinctly. Omitted, each case projects under its
    own recorded edition. `project` scopes the export to one project's cases (#149); omitted, every case in
    the store is projected. An open read: the SPA renders the preview before anyone authenticates.
    """
    from endopath import export as export_mod
    from endopath import projection

    if staging_edition is not None and not projection.is_offered_edition(staging_edition):
        raise HTTPException(status_code=422, detail=f"unknown staging edition: {staging_edition}")
    try:
        result = export_mod.build_export(
            conn,
            staging_edition=staging_edition,
            project=project,
            statuses=export_mod.EXPORTABLE_STATUSES,
        )
    except taxonomy.RegistryError as exc:
        raise HTTPException(status_code=422, detail=str(exc)) from exc
    return {
        "columns": result["columns"],
        "rows": result["rows"],
        "provenance": result["provenance"],
        "bundle": result["bundle"],
        "files": export_mod.render_files(result),
    }


@api.get("/staging_editions")
def staging_editions() -> list[dict]:
    """The FIGO editions the Export target-schema selector offers (#40). An open read."""
    from endopath import projection

    return projection.STAGING_EDITIONS


@api.get("/cases/{case_barcode}/projection")
def case_projection(
    case_barcode: str,
    edition: Optional[str] = None,
    conn: sqlite3.Connection = Depends(get_db),
) -> dict:
    """Project one case's FIGO stage under a target edition (#40): a determined code, a constrained
    candidate set the answer narrows to, or an indeterminate reason, each with its confidence and the
    derivation that produced it. Omitted, the case's own recorded edition. An open read."""
    from endopath import projection

    result = get_case(conn, case_barcode)
    if result is None:
        raise HTTPException(status_code=404, detail=f"case {case_barcode} not found")
    case, _ = result
    if edition is not None and not projection.is_offered_edition(edition):
        raise HTTPException(status_code=422, detail=f"unknown staging edition: {edition}")
    target = edition or case.staging_edition.value
    try:
        verdict = projection.project_stage(case.checklist, target)
    except taxonomy.RegistryError as exc:
        raise HTTPException(status_code=422, detail=str(exc)) from exc
    return {"case_barcode": case_barcode, "edition": target, "figo_stage": verdict}


@api.get("/cases")
def worklist(
    status: Optional[str] = None,
    field_name: Optional[str] = None,
    field_status: Optional[str] = None,
    max_confidence: Optional[float] = None,
    project: Optional[str] = None,
    conn: sqlite3.Connection = Depends(get_db),
) -> list[dict]:
    return list_cases(
        conn,
        status=status,
        field_name=field_name,
        field_status=field_status,
        max_confidence=max_confidence,
        project=project,
    )


def _slugify_project_id(name: str, existing: set[str]) -> str:
    """Derive a URL-safe project_id from a project name: lowercased, each run of non-alphanumeric
    characters collapsed to a single hyphen, leading and trailing hyphens stripped. A numeric suffix is
    appended when the base slug is already taken, so two projects sharing a name get distinct ids. Falls
    back to 'project' when the name carries no alphanumeric character."""
    base = re.sub(r"[^a-z0-9]+", "-", name.lower()).strip("-") or "project"
    if base not in existing:
        return base
    suffix = 2
    while f"{base}-{suffix}" in existing:
        suffix += 1
    return f"{base}-{suffix}"


class CreateProjectRequest(BaseModel):
    """A new project's human-readable name. The project_id slug is derived server-side from it."""

    name: str


@api.get("/projects")
def projects_list(conn: sqlite3.Connection = Depends(get_db)) -> dict:
    """Every project with its case count, the default project first. An open read: the SPA renders
    the project picker before anyone authenticates."""
    return {"projects": list_projects(conn), "default": DEFAULT_PROJECT_ID}


@api.post("/projects")
def projects_create(
    body: CreateProjectRequest,
    conn: sqlite3.Connection = Depends(get_db),
    identity: Optional[Identity] = Depends(authorize),
) -> dict:
    """Create a project: derive a slug project_id from the name and insert it. Admin-gated when
    auth is enabled: an authenticated non-admin account is refused with a 403; the open dev path (no
    identity, local dev and the test suite) allows it. Returns the created project with a zero case count."""
    if identity is not None and identity.role != "admin":
        raise HTTPException(status_code=403, detail="admin role required")
    name = body.name.strip()
    if not name:
        raise HTTPException(status_code=422, detail="project name is required")
    existing = {project["project_id"] for project in list_projects(conn)}
    project_id = _slugify_project_id(name, existing)
    created_by = identity.confirmer if identity is not None else None
    try:
        return create_project(conn, project_id=project_id, name=name, created_by=created_by)
    except ValueError as exc:
        raise HTTPException(status_code=409, detail=str(exc)) from exc


@api.post("/cases/extract_all", dependencies=authenticated)
def extract_all(conn: sqlite3.Connection = Depends(get_db)) -> dict:
    """Enqueue a cohort-extraction job (#115): the worker drains every queued case through the shared
    pipeline in the background, so a whole cohort is extracted without one /process call per case. Gated
    by the shared token, since the job spends the Anthropic key. Returns the enqueued job."""
    from endopath import jobs

    queued = list_cases(conn, status=CaseStatus.QUEUED.value)
    job = jobs.enqueue_cohort_extraction(conn, case_count=len(queued))
    return job.model_dump()


def _page_count(patient_filename: Optional[str]) -> int:
    if patient_filename is None:
        return 0
    page_dir = IMAGE_DIR / patient_filename
    if not page_dir.is_dir():
        return 0
    return len(list(page_dir.glob("page_*.png")))


@api.get("/cases/{case_barcode}")
def case_detail(case_barcode: str, conn: sqlite3.Connection = Depends(get_db)) -> dict:
    result = get_case(conn, case_barcode)
    if result is None:
        raise HTTPException(status_code=404, detail="case not found")
    case, patient_filename = result
    return {
        "case_barcode": case.case_barcode,
        "patient_filename": patient_filename,
        "report_date": case.report_date.isoformat() if case.report_date else None,
        "staging_edition": case.staging_edition.value,
        "status": case.status.value,
        "page_count": _page_count(patient_filename),
        "fields": {
            name: field.model_dump() for name, field in case.checklist.all_fields().items()
        },
    }


@api.get("/cases/{case_barcode}/review")
def case_review_surface(case_barcode: str, conn: sqlite3.Connection = Depends(get_db)) -> dict:
    """The case-review confirmation surface (#98, docs/prd.md sections 8.1, 8.3, 4.6; design-guideline
    section 9.3): the case's fields grouped into sections as record cards, each carrying its B1 validation
    block (#97) that promotes a field from needs-review to confirmed-eligible, and the decision ledger.

    An open read: the SPA renders the surface before anyone authenticates. The value chat, the confirm,
    and the edit actions on it are gated.
    """
    from endopath import case_review as case_review_mod

    result = get_case(conn, case_barcode)
    if result is None:
        raise HTTPException(status_code=404, detail="case not found")
    case, patient_filename = result
    return case_review_mod.case_review(
        conn, case, patient_filename=patient_filename, page_count=_page_count(patient_filename)
    )


@api.post("/cases/{case_barcode}/process", dependencies=authenticated)
def process_case_endpoint(
    case_barcode: str,
    use_visual: bool = False,
    conn: sqlite3.Connection = Depends(get_db),
) -> dict:
    """Run first-pass inference on a queued case (docs/prd.md section 8.2:
    Queued -> Processing -> Ready for Review). The LLM text-extraction pass is
    the fast core; use_visual additionally runs the ColPali consensus/resolution
    layers, which are minutes-per-page on CPU (see pipeline.py) and default off.
    """
    from endopath import pipeline

    result = get_case(conn, case_barcode)
    if result is None:
        raise HTTPException(status_code=404, detail="case not found")
    case, _ = result
    if case.status != CaseStatus.QUEUED:
        raise HTTPException(
            status_code=409,
            detail=f"case is {case.status.value}, not queued; only queued cases can be processed",
        )

    try:
        pipeline.process_case_in_store(conn, case_barcode, use_visual=use_visual)
    except pipeline.NoReportTextError:
        raise HTTPException(status_code=422, detail="no OCR report text available for this case")
    except RuntimeError as exc:  # e.g. ANTHROPIC_API_KEY not set (llm_extraction.get_client)
        raise HTTPException(status_code=503, detail=str(exc))

    return case_detail(case_barcode, conn)


def _maybe_advance_case(conn: sqlite3.Connection, case_barcode: str) -> None:
    """Section 7.2: confirming the last open field should move a case
    toward Confirmed, not require a separate manual action.
    """
    result = get_case(conn, case_barcode)
    if result is None:
        return
    case, patient_filename = result
    changed = False
    # A confirm or edit is the act of review beginning. Nothing else performs the READY_FOR_REVIEW ->
    # IN_REVIEW hop, so without it the IN_REVIEW -> CONFIRMED guard below is never satisfied and a case
    # confirmed field by field in the UI stays at READY_FOR_REVIEW forever (#154). Move it on the first
    # review action so it can reach CONFIRMED once every field is settled.
    if case.status == CaseStatus.READY_FOR_REVIEW:
        case.transition_to(CaseStatus.IN_REVIEW)
        changed = True
    if case.status == CaseStatus.IN_REVIEW and case.is_ready_for_export():
        case.transition_to(CaseStatus.CONFIRMED)
        changed = True
    if changed:
        upsert_case(conn, case, patient_filename)


class FieldDecisionRequest(BaseModel):
    """The reviewer-session context recorded with a confirm on the case-review surface (#98, #103): the
    confirmer and their role drawn from the session, an optional note, how the answer was reached, and the
    question and options the decision weighed. All optional, so a bare confirm still works; a row is
    written to the decision ledger only when the confirmer and role are present."""

    confirmer: Optional[str] = None
    role: Optional[str] = None
    note: Optional[str] = None
    decided_via: Optional[str] = None
    question: Optional[str] = None
    options: Optional[list] = None


def _record_field_decision(
    conn: sqlite3.Connection,
    *,
    case_barcode: str,
    field_name: str,
    field: Any,
    default_decided_via: str,
    ctx: Optional[FieldDecisionRequest],
) -> None:
    """Write a decision-ledger row for a confirm or edit, but only when the session identity is present.
    A bare confirm with no identity (the backward-compatible path) performs the state change without a
    ledger row; the confirmation surface always sends the session identity, so it always files one."""
    if ctx is None or not (ctx.confirmer and ctx.confirmer.strip() and ctx.role and ctx.role.strip()):
        return
    from endopath import case_review as case_review_mod

    case_review_mod.record_decision(
        conn,
        case_barcode=case_barcode,
        field=field,
        field_name=field_name,
        confirmer=ctx.confirmer,
        role=ctx.role,
        note=ctx.note,
        disposition="confirmed",
        decided_via=ctx.decided_via or default_decided_via,
        question=ctx.question,
        options=ctx.options,
    )


def _with_identity(
    ctx: Optional[FieldDecisionRequest], identity: Optional[Identity]
) -> Optional[FieldDecisionRequest]:
    """Force a field decision's confirmer and role to the authenticated account's, so an authenticated
    confirm files a ledger row under the real account even when the client sent no identity (#128). On the
    open dev path (no identity) the client's context passes through unchanged."""
    if identity is None:
        return ctx
    base = ctx if ctx is not None else FieldDecisionRequest()
    return base.model_copy(update={"confirmer": identity.confirmer, "role": identity.role})


@api.post("/cases/{case_barcode}/fields/{field_name}/confirm", dependencies=authenticated)
def confirm_field(
    case_barcode: str,
    field_name: str,
    body: Optional[FieldDecisionRequest] = None,
    conn: sqlite3.Connection = Depends(get_db),
    identity: Optional[Identity] = Depends(authorize),
) -> dict:
    result = get_case(conn, case_barcode)
    if result is None:
        raise HTTPException(status_code=404, detail="case not found")
    case, patient_filename = result
    if field_name not in case.checklist.all_fields():
        raise HTTPException(status_code=404, detail="field not found")

    field = getattr(case.checklist, field_name)
    field.confirm()
    setattr(case.checklist, field_name, field)

    upsert_case(conn, case, patient_filename)
    _record_field_decision(
        conn,
        case_barcode=case_barcode,
        field_name=field_name,
        field=field,
        default_decided_via="accept",
        ctx=_with_identity(body, identity),
    )
    _maybe_advance_case(conn, case_barcode)

    return case_detail(case_barcode, conn)


class EditFieldRequest(BaseModel):
    value: Any = None
    evidence: Optional[EvidenceSpan] = None
    # The reviewer-session context (#98), recorded to the decision ledger the same way a confirm is.
    decision: Optional[FieldDecisionRequest] = None


@api.post("/cases/{case_barcode}/fields/{field_name}/edit", dependencies=authenticated)
def edit_field(
    case_barcode: str,
    field_name: str,
    body: EditFieldRequest,
    conn: sqlite3.Connection = Depends(get_db),
    identity: Optional[Identity] = Depends(authorize),
) -> dict:
    result = get_case(conn, case_barcode)
    if result is None:
        raise HTTPException(status_code=404, detail="case not found")
    case, patient_filename = result
    if field_name not in case.checklist.all_fields():
        raise HTTPException(status_code=404, detail="field not found")

    field = getattr(case.checklist, field_name)
    field.edit(body.value, body.evidence)
    setattr(case.checklist, field_name, field)

    # ChecklistField.edit() sets .value via plain attribute assignment,
    # which pydantic does NOT type-check by default -- an arbitrary string
    # would otherwise sail through here, get persisted, and only blow up
    # the *next* time the case is read back (get_case reconstructs
    # EndometrialChecklist from scratch, which does validate). Force that
    # same validation now, before persisting, so a bad edit fails cleanly
    # with a 422 instead of corrupting the stored case.
    try:
        EndometrialChecklist.model_validate(case.checklist.model_dump())
    except ValidationError as exc:
        raise HTTPException(status_code=422, detail=f"invalid value for {field_name}: {exc}") from exc

    # A gate field is itself a reviewed value: editing histologic_type can open
    # or close the FIGO-grade gate. Re-run applicability (only now that the edit
    # is known valid) so a corrected histotype returns a previously suppressed
    # grade with its preserved draft, or suppresses a newly-inapplicable one
    # (issue #43). Total and idempotent, so a non-gate edit leaves gating alone.
    case.checklist.apply_applicability_rules(case.staging_edition)

    upsert_case(conn, case, patient_filename)
    _record_field_decision(
        conn,
        case_barcode=case_barcode,
        field_name=field_name,
        field=field,
        default_decided_via="edit",
        ctx=_with_identity(body.decision, identity),
    )
    _maybe_advance_case(conn, case_barcode)

    return case_detail(case_barcode, conn)


@api.post("/cases/{case_barcode}/fields/{field_name}/escalate", dependencies=authenticated)
def escalate_field(
    case_barcode: str,
    field_name: str,
    body: Optional[FieldDecisionRequest] = None,
    conn: sqlite3.Connection = Depends(get_db),
    identity: Optional[Identity] = Depends(authorize),
) -> dict:
    """Route a drafted field to the licensed reviewer's queue (#119, #161). Records an 'escalated'
    disposition in the decision ledger without confirming the value: the field stays in review, and the
    escalation queue reads these rows. Resolving it is confirming or editing the field, which upserts the
    ledger row to a settled disposition and drops it from the queue."""
    result = get_case(conn, case_barcode)
    if result is None:
        raise HTTPException(status_code=404, detail="case not found")
    case, _ = result
    if field_name not in case.checklist.all_fields():
        raise HTTPException(status_code=404, detail="field not found")

    field = getattr(case.checklist, field_name)
    ctx = _with_identity(body, identity)
    if ctx is not None and ctx.confirmer and ctx.role:
        from endopath import case_review as case_review_mod

        case_review_mod.record_decision(
            conn,
            case_barcode=case_barcode,
            field=field,
            field_name=field_name,
            confirmer=ctx.confirmer,
            role=ctx.role,
            note=ctx.note,
            disposition="escalated",
            decided_via="escalate",
            question=ctx.question,
            options=ctx.options,
        )
    return case_detail(case_barcode, conn)


@api.get("/escalations")
def list_escalations(
    conn: sqlite3.Connection = Depends(get_db),
    project: Optional[str] = None,
) -> dict:
    """The escalation queue (#161): fields a reviewer routed to the licensed account, across cases. Reads
    the decision ledger for the escalated disposition; a later confirm or edit upserts the row to a settled
    disposition, so it leaves the queue. Scoped to a project when one is given."""
    if project:
        rows = conn.execute(
            text(
                "SELECT fc.case_barcode, fc.field_name, fc.value_json, fc.confirmed_by, fc.role, "
                "fc.note, fc.confirmed_at FROM field_confirmations fc "
                "JOIN cases c ON c.case_barcode = fc.case_barcode "
                "WHERE fc.disposition = 'escalated' AND c.project_id = :project "
                "ORDER BY fc.confirmed_at DESC"
            ),
            {"project": project},
        )
    else:
        rows = conn.execute(
            text(
                "SELECT case_barcode, field_name, value_json, confirmed_by, role, note, confirmed_at "
                "FROM field_confirmations WHERE disposition = 'escalated' ORDER BY confirmed_at DESC"
            )
        )
    escalations = []
    for r in rows.mappings():
        value: Any = None
        if r["value_json"]:
            try:
                value = json.loads(r["value_json"])
            except (ValueError, TypeError):
                value = r["value_json"]
        escalations.append(
            {
                "case_barcode": r["case_barcode"],
                "field_name": r["field_name"],
                "value": value,
                "escalated_by": r["confirmed_by"],
                "role": r["role"],
                "note": r["note"],
                "escalated_at": r["confirmed_at"],
            }
        )
    return {"escalations": escalations}


@api.post("/demo/reset")
def demo_reset(
    conn: sqlite3.Connection = Depends(get_db),
    case_barcode: Optional[str] = None,
    field_name: str = "molecular_classification",
    seed_escalation: bool = False,
    clear_longitudinal: bool = False,
) -> dict:
    """Clear one beat of the integrated demo's slice so a run starts fresh. Called three ways, and none
    must step on each other in the full walkthrough:

    - `clear_longitudinal` with a case: drop that case's GDC linkage joins (#180), so the annotator's
      patient-record beat starts from an unconfirmed panel and its Confirm-linkage click has something to
      do on every run. It touches only `longitudinal_joins`, leaving the case's field decisions and the
      mapping resolution untouched, and returns without running the field or mapping reset below.
    - No case: clear the data manager's molecular mapping resolution (and its split components). The
      mapping and the case-field beats live in different tables, so this is the mapping-only reset.
    - A case: clear that case's decision on the one escalated field (the annotator's escalation and the
      pathologist's confirmation), and nothing else, so the case's other confirmed values and the data
      manager's earlier mapping resolution both survive.

    With `seed_escalation`, it then re-routes that field to the escalation queue as the annotator, so the
    pathologist story finds an escalation to resolve on its own, without the annotator story running
    first. Demo-only and idempotent; it touches only the rows the demo writes, on a public demo corpus,
    so it stays open like the reads."""
    if clear_longitudinal:
        if case_barcode:
            conn.execute(
                text("DELETE FROM longitudinal_joins WHERE case_barcode = :cb"),
                {"cb": case_barcode},
            )
            conn.commit()
        return {"reset": True, "case_barcode": case_barcode, "cleared": "longitudinal_joins"}

    if case_barcode:
        conn.execute(
            text(
                "DELETE FROM field_confirmations WHERE case_barcode = :cb AND field_name = :fn"
            ),
            {"cb": case_barcode, "fn": field_name},
        )
    else:
        conn.execute(
            text(
                "DELETE FROM mapping_confirmation_components WHERE mapping_id IN "
                "(SELECT mapping_id FROM mapping_confirmations WHERE LOWER(source_field) LIKE '%molecular%')"
            )
        )
        conn.execute(
            text("DELETE FROM mapping_confirmations WHERE LOWER(source_field) LIKE '%molecular%'")
        )
    conn.commit()

    if case_barcode and seed_escalation:
        from endopath import case_review as case_review_mod

        seed_notes = {
            "lymphovascular_space_invasion": "Report reads 'Identified' but not focal vs substantial; the FIGO stage turns on it.",
            "molecular_classification": "Molecular class needs POLE, MMR, and p53 the report never assays.",
        }
        result = get_case(conn, case_barcode)
        if result is not None:
            case, _ = result
            if field_name in case.checklist.all_fields():
                case_review_mod.record_decision(
                    conn,
                    case_barcode=case_barcode,
                    field=getattr(case.checklist, field_name),
                    field_name=field_name,
                    confirmer="Annotator",
                    role="reviewer",
                    note=seed_notes.get(field_name, "Needs a pathologist's read."),
                    disposition="escalated",
                    decided_via="escalate",
                    question=None,
                    options=None,
                )

    return {"reset": True, "case_barcode": case_barcode, "field_name": field_name}


class FieldChatTurn(BaseModel):
    """One prior turn of a value chat: the speaker (`user` or `assistant`) and the text, so a follow-up
    question carries its context."""

    role: str
    text: str


class FieldChatRequest(BaseModel):
    """A reviewer's question about one checklist value (#98), with the prior turns of that value's
    conversation."""

    question: str
    history: list[FieldChatTurn] = Field(default_factory=list)


@api.post("/cases/{case_barcode}/fields/{field_name}/chat", dependencies=authenticated)
def field_chat(
    case_barcode: str,
    field_name: str,
    body: FieldChatRequest,
    conn: sqlite3.Connection = Depends(get_db),
) -> dict:
    """Answer a reviewer's question about one extracted value (#98): the engine's reasoning, the evidence
    it read, and, when the question points at a correction, the value to file. The exchange resolves into
    a filed decision when the reviewer accepts the suggestion through the edit endpoint with
    `decided_via='chat'`.

    Gated by the shared token: the call spends the Anthropic key. The chat model is named explicitly in
    `case_review` (claude-sonnet-5), behind a provider-agnostic seam a fake substitutes for in the fast
    test suite.
    """
    from endopath import case_review as case_review_mod

    result = get_case(conn, case_barcode)
    if result is None:
        raise HTTPException(status_code=404, detail="case not found")
    case, _ = result
    if field_name not in case.checklist.all_fields():
        raise HTTPException(status_code=404, detail="field not found")

    try:
        return case_review_mod.explain_field(
            conn,
            case=case,
            field_name=field_name,
            question=body.question,
            history=[turn.model_dump() for turn in body.history],
        )
    except case_review_mod.CaseReviewError as exc:
        raise HTTPException(status_code=422, detail=str(exc)) from exc
    except RuntimeError as exc:  # e.g. ANTHROPIC_API_KEY not set (llm_extraction.get_client)
        raise HTTPException(status_code=503, detail=str(exc)) from exc


# Page embeddings for a case, keyed by patient_filename. issue #18 measured
# ColPali embedding at ~1-2s/page on a GPU and roughly an order of
# magnitude slower per page on this app's local CPU deployment -- re-embedding
# a case's pages from scratch for every checklist field would multiply that
# cost by up to nine. Caching in-process means the first visual-evidence
# request for a case pays the embedding cost once; every other field on the
# same case reuses it. Single-process, no eviction, no persistence across
# restarts -- proportionate to this app's local-first/no-auth/MVP-scale
# posture (see module docstring), not a production cache.
_page_embedding_cache: dict[str, object] = {}


def _get_case_page_embeddings(patient_filename: str):
    if patient_filename not in _page_embedding_cache:
        image_paths = colpali_retrieval.case_page_images(patient_filename, image_dir=IMAGE_DIR)
        if not image_paths:
            raise HTTPException(status_code=404, detail="no page images ingested for this case")
        _page_embedding_cache[patient_filename] = colpali_retrieval.embed_pages(image_paths)
    return _page_embedding_cache[patient_filename]


# Precomputed field-query vectors (issue #19), scored on CPU (issue #21) so the
# deployed serving path loads no model. ENDOPATH_EMBEDDINGS_DIR lets the Space
# point this at wherever the HF Dataset was downloaded.
_EMBEDDINGS_DIR = Path(os.environ.get("ENDOPATH_EMBEDDINGS_DIR", str(precomputed_retrieval.EMBEDDINGS_DIR)))
_query_embeddings_cache: Optional[dict] = None
_query_embeddings_loaded = False


def _get_query_embeddings():
    """The nine precomputed query vectors, loaded once. None when they aren't
    present (then visual retrieval falls back to embedding live)."""
    global _query_embeddings_cache, _query_embeddings_loaded
    if not _query_embeddings_loaded:
        path = _EMBEDDINGS_DIR / precomputed_retrieval.QUERY_EMBEDDINGS_NAME
        _query_embeddings_cache = (
            precomputed_retrieval.load_query_embeddings(_EMBEDDINGS_DIR) if path.exists() else None
        )
        _query_embeddings_loaded = True
    return _query_embeddings_cache


def _visual_rank(patient_filename: str, field_name: str) -> tuple[int, float]:
    """(1-indexed page, score) for a field's visual retrieval. Prefers
    precomputed vectors scored on CPU (deployed path, no torch/ColPali); falls
    back to embedding the case's pages live when they aren't precomputed."""
    queries = _get_query_embeddings()
    if (
        queries is not None
        and field_name in queries
        and precomputed_retrieval.has_case_embeddings(patient_filename, _EMBEDDINGS_DIR)
    ):
        pages = precomputed_retrieval.load_case_page_embeddings(patient_filename, _EMBEDDINGS_DIR)
        page_number, score, _ = precomputed_retrieval.rank_pages_with_margin(queries[field_name], pages)
        return page_number, float(score)

    if not colpali_retrieval.is_available():
        raise HTTPException(
            status_code=503,
            detail="no precomputed embeddings for this case and colpali_engine/torch not installed",
        )
    page_embeddings = _get_case_page_embeddings(patient_filename)
    page_number, score = colpali_retrieval.rank_pages(
        colpali_retrieval.FIELD_QUERIES[field_name], page_embeddings
    )
    return page_number, float(score)


@api.get("/cases/{case_barcode}/fields/{field_name}/visual_evidence", dependencies=authenticated)
def visual_evidence(case_barcode: str, field_name: str, conn: sqlite3.Connection = Depends(get_db)) -> dict:
    """Visual retrieval for one checklist field, returning an EvidenceSpan the
    same shape llm_extraction.py produces. Scores precomputed page vectors
    against precomputed query vectors on CPU (issue #21, no model loaded), and
    falls back to embedding live with ColPali when a case isn't precomputed.
    Stateless. Issue #18: raw scores aren't comparable across different fields.
    """
    if field_name not in colpali_retrieval.FIELD_QUERIES:
        raise HTTPException(status_code=404, detail="field not found")

    result = get_case(conn, case_barcode)
    if result is None:
        raise HTTPException(status_code=404, detail="case not found")
    _, patient_filename = result
    if patient_filename is None:
        raise HTTPException(status_code=404, detail="no page images ingested for this case")

    page_number, score = _visual_rank(patient_filename, field_name)
    evidence = EvidenceSpan(quote="", page_number=page_number, source="visual")
    return {"evidence": evidence.model_dump(), "score": score}


@api.get("/cases/{case_barcode}/pages/{page_number}")
def get_page_image(case_barcode: str, page_number: int, conn: sqlite3.Connection = Depends(get_db)):
    result = get_case(conn, case_barcode)
    if result is None:
        raise HTTPException(status_code=404, detail="case not found")
    _, patient_filename = result
    if patient_filename is None:
        raise HTTPException(status_code=404, detail="no page images ingested for this case")

    image_path = IMAGE_DIR / patient_filename / f"page_{page_number:03d}.png"
    if not image_path.exists():
        raise HTTPException(status_code=404, detail="page image not found")
    return FileResponse(image_path)


# ============================================================================
# Longitudinal Records Endpoints
# ============================================================================

@api.get("/longitudinal/records/{case_barcode}")
def longitudinal_records(case_barcode: str, conn = Depends(get_db)) -> dict:
    """List all longitudinal records for a participant (via case barcode)."""
    # Verify case exists
    result = conn.execute(
        text("SELECT case_barcode FROM cases WHERE case_barcode = :barcode"),
        {"barcode": case_barcode}
    ).fetchone()

    if not result:
        raise HTTPException(status_code=404, detail="case not found")

    # Extract participant ID from case barcode (first 12 chars = TCGA-XX-XXXX)
    participant_id = case_barcode[:12] if len(case_barcode) >= 12 else case_barcode

    # Get all records for this participant
    rows = conn.execute(
        text("""
        SELECT record_id, participant_id, case_barcode, record_type, record_index,
               days_to_followup, days_to_treatment, days_to_recurrence,
               biopsy_result, followup_treatment_success, recurrence_type, therapy_type,
               structured_data
        FROM longitudinal_records
        WHERE participant_id = :pid
        ORDER BY record_type, record_index
        """),
        {"pid": participant_id},
    ).mappings().fetchall()

    records = []
    for row in rows:
        record_dict = dict(row)
        if record_dict.get("structured_data"):
            try:
                record_dict["structured_data"] = json.loads(record_dict["structured_data"])
            except (json.JSONDecodeError, TypeError):
                record_dict["structured_data"] = {}
        records.append(record_dict)

    return {"case_barcode": case_barcode, "participant_id": participant_id, "records": records}


@api.get("/longitudinal/joins/{case_barcode}")
def longitudinal_joins(case_barcode: str, conn = Depends(get_db)) -> dict:
    """List all confirmed joins for a case."""
    rows = conn.execute(
        text("""
        SELECT lj.join_id, lj.case_barcode, lj.record_id, lj.disposition,
               lj.confirmed_by, lj.role, lj.confirmed_at, lj.note, lj.question,
               lr.record_type, lr.days_to_followup, lr.therapy_type, lr.recurrence_type
        FROM longitudinal_joins lj
        JOIN longitudinal_records lr ON lj.record_id = lr.record_id
        WHERE lj.case_barcode = :barcode
        ORDER BY lj.confirmed_at DESC
        """),
        {"barcode": case_barcode},
    ).mappings().fetchall()

    joins = [dict(row) for row in rows]
    return {"case_barcode": case_barcode, "joins": joins}


@api.post("/longitudinal/join/{case_barcode}", dependencies=authenticated)
def confirm_join(
    case_barcode: str,
    request: JoinRequest,
    identity: Optional[Identity] = Depends(authorize),
    conn = Depends(get_db),
) -> dict:
    """Confirm or reject a join between a case and a longitudinal record."""
    join_id = str(uuid.uuid4())
    confirmed_at = datetime.now(timezone.utc).isoformat()

    # Verify the record exists
    result = conn.execute(
        text("SELECT record_id FROM longitudinal_records WHERE record_id = :rid"),
        {"rid": request.record_id}
    ).fetchone()

    if not result:
        raise HTTPException(status_code=404, detail="record not found")

    # Check if a join already exists
    existing = conn.execute(
        text("SELECT join_id FROM longitudinal_joins WHERE case_barcode = :cb AND record_id = :rid"),
        {"cb": case_barcode, "rid": request.record_id}
    ).fetchone()

    if existing:
        # Update existing join
        conn.execute(
            text("""
            UPDATE longitudinal_joins
            SET disposition = :disp, confirmed_by = :cb, role = :role, confirmed_at = :cat, note = :note, question = :q
            WHERE join_id = :jid
            """),
            {
                "disp": request.disposition,
                "cb": identity.confirmer if identity else None,
                "role": identity.role if identity else None,
                "cat": confirmed_at,
                "note": request.note,
                "q": request.question,
                "jid": existing[0],
            }
        )
        join_id = existing[0]
    else:
        # Create new join
        conn.execute(
            text("""
            INSERT INTO longitudinal_joins (
                join_id, case_barcode, record_id, disposition, confirmed_by, role, confirmed_at, note, question
            ) VALUES (:jid, :cb, :rid, :disp, :cb2, :role, :cat, :note, :q)
            """),
            {
                "jid": join_id,
                "cb": case_barcode,
                "rid": request.record_id,
                "disp": request.disposition,
                "cb2": identity.confirmer if identity else None,
                "role": identity.role if identity else None,
                "cat": confirmed_at,
                "note": request.note,
                "q": request.question,
            }
        )

    conn.commit()

    return {
        "join_id": join_id,
        "case_barcode": case_barcode,
        "record_id": request.record_id,
        "disposition": request.disposition,
        "confirmed_by": identity.confirmer if identity else None,
        "confirmed_at": confirmed_at,
    }


@api.post("/longitudinal/confirm-linkage/{case_barcode}", dependencies=authenticated)
def confirm_linkage(
    case_barcode: str,
    request: LinkageRequest = LinkageRequest(),
    identity: Optional[Identity] = Depends(authorize),
    conn = Depends(get_db),
) -> dict:
    """Affirm in one action that every GDC longitudinal record joined to this case belongs to it (#180).

    Upserts a `confirmed` join for each of the participant's records, recording the reviewer's identity and
    role, and leaves a record the reviewer has already flagged `rejected` untouched so a confirm-all does not
    silently un-flag it. Idempotent: re-running re-stamps the confirmations without duplicating rows."""
    result = conn.execute(
        text("SELECT case_barcode FROM cases WHERE case_barcode = :b"), {"b": case_barcode}
    ).fetchone()
    if not result:
        raise HTTPException(status_code=404, detail="case not found")

    participant_id = case_barcode[:12] if len(case_barcode) >= 12 else case_barcode
    record_ids = [
        row[0]
        for row in conn.execute(
            text("SELECT record_id FROM longitudinal_records WHERE participant_id = :p"),
            {"p": participant_id},
        ).fetchall()
    ]
    existing = {
        row[0]: row[1]
        for row in conn.execute(
            text("SELECT record_id, disposition FROM longitudinal_joins WHERE case_barcode = :b"),
            {"b": case_barcode},
        ).fetchall()
    }
    confirmed_at = datetime.now(timezone.utc).isoformat()
    confirmer = identity.confirmer if identity else None
    role = identity.role if identity else None

    confirmed = 0
    for record_id in record_ids:
        if existing.get(record_id) == "rejected":
            continue  # a flagged record stays rejected; a confirm-all does not un-flag it
        if record_id in existing:
            conn.execute(
                text(
                    "UPDATE longitudinal_joins SET disposition='confirmed', confirmed_by=:c, "
                    "role=:r, confirmed_at=:t, note=:n WHERE case_barcode=:b AND record_id=:rid"
                ),
                {"c": confirmer, "r": role, "t": confirmed_at, "n": request.note, "b": case_barcode, "rid": record_id},
            )
        else:
            conn.execute(
                text(
                    "INSERT INTO longitudinal_joins (join_id, case_barcode, record_id, disposition, "
                    "confirmed_by, role, confirmed_at, note, question) "
                    "VALUES (:j, :b, :rid, 'confirmed', :c, :r, :t, :n, NULL)"
                ),
                {"j": str(uuid.uuid4()), "b": case_barcode, "rid": record_id, "c": confirmer, "r": role, "t": confirmed_at, "n": request.note},
            )
        confirmed += 1
    conn.commit()

    rejected = sum(1 for disposition in existing.values() if disposition == "rejected")
    return {
        "case_barcode": case_barcode,
        "confirmed": confirmed,
        "rejected": rejected,
        "total_records": len(record_ids),
        "confirmed_by": confirmer,
        "role": role,
        "confirmed_at": confirmed_at,
    }


@api.get("/longitudinal/summary/{case_barcode}")
def longitudinal_summary(case_barcode: str, conn = Depends(get_db)) -> dict:
    """Summarize longitudinal data availability for a case."""
    # Verify case exists
    result = conn.execute(
        text("SELECT case_barcode FROM cases WHERE case_barcode = :barcode"),
        {"barcode": case_barcode}
    ).fetchone()

    if not result:
        raise HTTPException(status_code=404, detail="case not found")

    # Extract participant ID from case barcode
    participant_id = case_barcode[:12] if len(case_barcode) >= 12 else case_barcode

    # Count records by type
    rows = conn.execute(
        text("""
        SELECT record_type, COUNT(*) as count
        FROM longitudinal_records
        WHERE participant_id = :pid
        GROUP BY record_type
        """),
        {"pid": participant_id}
    ).fetchall()

    records_by_type = {row[0]: row[1] for row in rows}

    # Count joins by disposition
    rows = conn.execute(
        text("""
        SELECT disposition, COUNT(*) as count
        FROM longitudinal_joins
        WHERE case_barcode = :barcode
        GROUP BY disposition
        """),
        {"barcode": case_barcode}
    ).fetchall()

    joins_by_disposition = {row[0]: row[1] for row in rows}

    return {
        "case_barcode": case_barcode,
        "participant_id": participant_id,
        "records_available": records_by_type,
        "joins_confirmed": joins_by_disposition,
    }


@api.get("/longitudinal/cohort")
def longitudinal_cohort(project: Optional[str] = None, conn = Depends(get_db)) -> dict:
    """Cohort-level longitudinal outcomes for the dashboard module (#181): recurrence, survival, and
    adjuvant-treatment counts, plus median follow-up and survival times, over the cases the GDC join
    reached. Projects each case through the same `project_outcomes` the export uses, so the dashboard and
    the export never disagree. `project` scopes to one project's cases; omitted, every case is aggregated.
    An open read."""
    from endopath import export as export_mod
    from endopath import longitudinal

    by_participant = export_mod._load_longitudinal(conn)
    joins_by_case = export_mod._load_longitudinal_joins(conn)
    if project is not None:
        case_rows = conn.execute(
            text("SELECT case_barcode FROM cases WHERE project_id = :p ORDER BY case_barcode"),
            {"p": project},
        ).fetchall()
    else:
        case_rows = conn.execute(
            text("SELECT case_barcode FROM cases ORDER BY case_barcode")
        ).fetchall()

    outcomes_per_case = []
    for (barcode,) in case_rows:
        participant_id = barcode[:12] if len(barcode) >= 12 else barcode
        records = by_participant.get(participant_id, [])
        if not records:
            continue  # only cases the join reached carry longitudinal outcomes
        # Exclude records a reviewer rejected, the same rule the export applies, so the dashboard and the
        # export agree over the join-reached case set (#217).
        rejected = export_mod._rejected_record_ids(joins_by_case.get(barcode, []))
        if rejected:
            records = [r for r in records if r.record_id not in rejected]
        outcomes_per_case.append(longitudinal.project_outcomes(records))

    return longitudinal.aggregate_cohort(outcomes_per_case)


app.include_router(api)


def _static_file_within_dist(dist: Path, relative_path: str) -> Optional[Path]:
    """Resolve `relative_path` against the build directory, or None if it isn't
    a real file inside it. A path that names a directory holding an index.html
    resolves to that index.html, so a bundled subfolder page (the case study at
    /case-study/) serves without falling through to the SPA and rendering blank.

    The containment check is not decorative: `spa_path` is whatever the client
    put in the URL, and a percent-encoded `..` survives the routing layer, so
    resolving without checking would hand out arbitrary files from the host.
    """
    if not relative_path:
        return None
    dist_root = dist.resolve()
    candidate = (dist_root / relative_path).resolve()
    if dist_root not in candidate.parents:
        return None
    if candidate.is_file():
        return candidate
    index = candidate / "index.html"
    if candidate.is_dir() and index.is_file():
        return index
    return None


@app.get("/{spa_path:path}", include_in_schema=False)
def serve_frontend(spa_path: str):
    """Serve the built SPA: real files (hashed JS/CSS, favicon) as themselves,
    every other path as index.html so client-side routes survive a hard refresh
    or a deep link.

    Registered last, so it only sees paths no API route claimed. Unmatched
    `/api/...` paths are excluded explicitly -- an API client that typos a URL
    should get a JSON 404, not a 200 and a page of HTML.
    """
    if spa_path == "api" or spa_path.startswith("api/"):
        raise HTTPException(status_code=404, detail="not found")

    index = _frontend_dist / "index.html"
    if not index.is_file():
        raise HTTPException(
            status_code=404,
            detail=f"frontend build not found at {_frontend_dist}; run `npm run build` in frontend/",
        )

    static_file = _static_file_within_dist(_frontend_dist, spa_path)
    if static_file is None:
        return FileResponse(index)
    # A directory URL served via its index.html needs a trailing slash, or the
    # page's relative asset paths (./assets/...) resolve against the parent and
    # 404. Redirect /case-study to /case-study/ so the assets load. An explicit
    # /case-study/index.html (or any trailing-slash path) is served as-is.
    if (
        static_file.name == "index.html"
        and static_file != index
        and not spa_path.endswith("/")
        and not spa_path.endswith("index.html")
    ):
        return RedirectResponse(url=f"/{spa_path}/", status_code=308)
    return FileResponse(static_file)