Blame for app/api/server.py

"""FastAPI backend for the HCC-MEAT Coder Space.

Endpoints:
    POST /api/code          — submit a note, receive ranked HCC candidates + MEAT evidence
    GET  /api/gallery       — return the preloaded demo gallery (note + precomputed result)
    GET  /api/healthz       — simple health probe
    GET  /api/reference/hcc — static HCC label table for tooltip content in the UI

The app keeps a single `Pipeline` instance in state and reuses it across requests.
"""

from __future__ import annotations

import json
import logging
import traceback
from contextlib import asynccontextmanager
from pathlib import Path

from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel, Field

from app.api.metrics import snapshot as metrics_snapshot
from app.pipeline import Pipeline
from app.pipeline.types import CodingResult, Note

_LOG = logging.getLogger("hcc_meat.api")
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(name)s %(levelname)s %(message)s")

GALLERY_PATH = Path(__file__).resolve().parents[2] / "tests" / "fixtures" / "regression_suite.json"
UI_DIST = Path(__file__).resolve().parents[2] / "ui" / "dist"


class CodeRequest(BaseModel):
    text: str = Field(..., min_length=1, max_length=50_000)
    note_id: str = Field(default="adhoc")


class GalleryItem(BaseModel):
    case_id: str
    name: str
    tags: list[str]
    text: str
    result: CodingResult


@asynccontextmanager
async def lifespan(app: FastAPI):
    """Build the pipeline and prime caches at startup.

    Each step is wrapped defensively: if a step fails, we log the traceback
    and leave the corresponding state empty so the app still boots. /api/status
    reports this as `degraded` and the UI banner surfaces which facet failed.
    A hard crash here would leave HF users with a 502 and no diagnostic path.
    """
    app.state.startup_errors = {}
    try:
        app.state.pipeline = Pipeline()
        _LOG.info("Pipeline built")
    except Exception as e:  # noqa: BLE001
        app.state.pipeline = None
        app.state.startup_errors["pipeline"] = repr(e)
        _LOG.error("Pipeline build failed: %s\n%s", e, traceback.format_exc())

    try:
        app.state.gallery = (
            _build_gallery(app.state.pipeline) if app.state.pipeline is not None else []
        )
        _LOG.info("Gallery primed: %d items", len(app.state.gallery))
    except Exception as e:  # noqa: BLE001
        app.state.gallery = []
        app.state.startup_errors["gallery"] = repr(e)
        _LOG.error("Gallery build failed: %s\n%s", e, traceback.format_exc())

    try:
        app.state.metrics = metrics_snapshot() if app.state.pipeline is not None else {}
        _LOG.info("Metrics snapshot computed")
    except Exception as e:  # noqa: BLE001
        app.state.metrics = {}
        app.state.startup_errors["metrics"] = repr(e)
        _LOG.error("Metrics snapshot failed: %s\n%s", e, traceback.format_exc())

    # Kick off LLM MEAT warm-up in the background when selected.
    # Rule-based path serves until the LLM is ready; /api/status exposes the
    # ready flag so the UI (and smoke tests) can see warm state.
    import threading as _threading

    app.state.meat_llm = None
    extractor = getattr(app.state.pipeline, "meat_extractor", None) if app.state.pipeline else None
    primary = getattr(extractor, "primary", None)
    if primary is not None and hasattr(primary, "warm"):
        app.state.meat_llm = primary

        def _warm_meat_llm() -> None:
            try:
                primary.warm()
                _LOG.info("MEAT LLM ready: %s", getattr(primary, "model_name", "?"))
            except Exception as e:  # noqa: BLE001
                app.state.startup_errors["meat_llm"] = repr(e)
                _LOG.error(
                    "MEAT LLM warm-up failed: %s\n%s", e, traceback.format_exc()
                )

        _threading.Thread(
            target=_warm_meat_llm, name="meat-llm-warmup", daemon=True
        ).start()

    yield


app = FastAPI(
    title="HCC-MEAT Coder",
    version="0.1.0",
    description=(
        "Suggests CMS-HCC v28 codes from a clinical note with MEAT-justified "
        "evidence, calibrated confidence, and an over-coding risk flag."
    ),
    lifespan=lifespan,
)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:5173", "http://127.0.0.1:5173"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


@app.get("/api/healthz")
def healthz() -> dict:
    return {"status": "ok", "version": app.version}


@app.get("/api/status")
def status() -> dict:
    """Runtime diagnostic: are reference CSVs loaded, pipeline built, gallery primed?

    The UI polls this on startup so an HF-Space misdeploy (missing data dir,
    startup crash) surfaces as a visible banner instead of a silent blank state.
    """
    pipeline: Pipeline | None = getattr(app.state, "pipeline", None)
    gallery: list[GalleryItem] = getattr(app.state, "gallery", []) or []
    metrics: dict = getattr(app.state, "metrics", {}) or {}

    crosswalk = pipeline.crosswalk if pipeline else None
    ref_counts: dict = {}
    if crosswalk is not None:
        ref_counts = {
            "hcc_labels": len(crosswalk.all_hccs()),
            "hcc_crosswalk_icd10": len(crosswalk.all_icd10_in_crosswalk()),
        }

    ok = (
        pipeline is not None
        and ref_counts.get("hcc_labels", 0) > 0
        and ref_counts.get("hcc_crosswalk_icd10", 0) > 0
        and len(gallery) > 0
    )
    meat_llm = getattr(app.state, "meat_llm", None)
    meat_llm_model = getattr(meat_llm, "model_name", None) if meat_llm else None
    meat_llm_ready = bool(meat_llm and meat_llm.is_ready()) if meat_llm else None
    return {
        "status": "ok" if ok else "degraded",
        "version": app.version,
        "pipeline_ready": pipeline is not None,
        "reference_counts": ref_counts,
        "gallery_size": len(gallery),
        "metrics_available": bool(metrics),
        "meat_llm_configured": meat_llm is not None,
        "meat_llm_model": meat_llm_model,
        "meat_llm_ready": meat_llm_ready,
        "startup_errors": getattr(app.state, "startup_errors", {}),
    }


@app.post("/api/code", response_model=CodingResult)
def code(req: CodeRequest) -> CodingResult:
    pipeline: Pipeline | None = getattr(app.state, "pipeline", None)
    if pipeline is None:
        raise HTTPException(503, "pipeline not ready — see /api/status")
    note = Note(note_id=req.note_id, text=req.text)
    return pipeline.run(note)


@app.get("/api/gallery", response_model=list[GalleryItem])
def gallery() -> list[GalleryItem]:
    return app.state.gallery


@app.get("/api/metrics")
def metrics() -> dict:
    """Static snapshot built at startup. v0.1 derives from the regression fixture."""
    return app.state.metrics


@app.post("/api/debug/pipeline")
def debug_pipeline(req: CodeRequest) -> dict:
    """Pipeline introspection: sections, spans, attributable spans, and
    per-candidate MEAT extraction trace. Used to diagnose MEAT-coverage gaps
    when the live Space disagrees with local expectations.
    """
    pipeline: Pipeline | None = getattr(app.state, "pipeline", None)
    if pipeline is None:
        raise HTTPException(503, "pipeline not ready")

    from app.pipeline.components import a_segmenter
    from app.pipeline.components.c_negation import is_attributable
    from app.pipeline.components.f_meat import _sentences_with_offsets

    note = Note(note_id=req.note_id, text=req.text)
    note = a_segmenter.segment(note)
    spans = pipeline.ner(note)
    spans = pipeline.negation(note, spans)
    eligible = [s for s in spans if is_attributable(s)]
    links = pipeline.linker(eligible)
    candidates = pipeline.crosswalk.build_candidates(
        links, raf_segment=pipeline.config.raf_segment
    )
    candidates = pipeline.crosswalk.apply_trumping(candidates)

    sentences = _sentences_with_offsets(note.text)
    extraction_trace: list[dict] = []

    primary = getattr(pipeline.meat_extractor, "primary", None)
    primary_has_try = primary is not None and hasattr(primary, "try_extract")
    primary_ready = (
        primary.is_ready() if primary is not None and hasattr(primary, "is_ready") else None
    )

    # Single LLM pass per candidate to keep latency sane: capture raw LLM
    # output, then filter via the same substring rule Component I uses. No
    # double-calling pipeline.meat_extractor.
    for cand in candidates:
        llm_raw: list[dict] = []
        kept: list[dict] = []
        if primary_has_try and primary_ready:
            try:
                res = primary.try_extract(note, cand)
                if res is not None:
                    for ev in res:
                        in_note = ev.quote in note.text
                        llm_raw.append(
                            {
                                "cat": ev.category.value,
                                "quote": ev.quote,
                                "is_substring": in_note,
                            }
                        )
                        if in_note:
                            kept.append(
                                {"cat": ev.category.value, "quote": ev.quote[:120]}
                            )
            except Exception as e:  # noqa: BLE001
                llm_raw = [{"error": f"{type(e).__name__}: {e}"}]

        extraction_trace.append(
            {
                "hcc": cand.hcc,
                "surface": [lnk.condition_span.text for lnk in cand.icd10_links],
                "llm_raw_output": llm_raw,
                "verified_evidence_preview": kept,
            }
        )

    return {
        "sections": [
            {"kind": s.kind, "start": s.start, "end": s.end} for s in note.sections
        ],
        "sentences": [{"start": s, "end": e, "text": t[:80]} for (s, e, t) in sentences],
        "all_spans": [
            {"text": s.text, "type": s.span_type.value, "start": s.start, "end": s.end}
            for s in spans
        ],
        "eligible_spans": len(eligible),
        "candidates": [c.hcc for c in candidates],
        "extractor_chain": {
            "outer_type": type(pipeline.meat_extractor).__name__,
            "primary_type": type(primary).__name__ if primary else None,
            "primary_has_try_extract": primary_has_try,
            "primary_ready": primary_ready,
        },
        "extraction_trace": extraction_trace,
    }


@app.post("/api/benchmark")
def benchmark() -> dict:
    """Re-run every gallery note live (no precomputation) and summarize latency.

    Exercises the CPU-free-tier latency target (S2 in AGENT_BRIEF §3): p95 < 8s.
    """
    import statistics
    import time

    pipeline: Pipeline | None = getattr(app.state, "pipeline", None)
    if pipeline is None:
        raise HTTPException(503, "pipeline not ready — see /api/status")
    items: list[GalleryItem] = app.state.gallery
    latencies: list[float] = []
    timings: list[dict] = []
    for item in items:
        t0 = time.perf_counter()
        pipeline.run(Note(note_id=item.case_id, text=item.text))
        ms = (time.perf_counter() - t0) * 1000.0
        latencies.append(ms)
        timings.append({"case_id": item.case_id, "ms": round(ms, 2), "chars": len(item.text)})

    latencies.sort()
    summary = {
        "count": len(latencies),
        "mean_ms": round(statistics.fmean(latencies), 2) if latencies else 0.0,
        "median_ms": round(statistics.median(latencies), 2) if latencies else 0.0,
        "p95_ms": round(latencies[int(0.95 * (len(latencies) - 1))], 2) if latencies else 0.0,
        "max_ms": round(max(latencies), 2) if latencies else 0.0,
        "min_ms": round(min(latencies), 2) if latencies else 0.0,
        "target_p95_ms": 8000.0,
        "timings": timings,
    }
    return summary


@app.get("/api/reference/hcc")
def reference_hcc() -> list[dict]:
    """Return HCC labels for UI tooltips + model card linking."""
    import csv

    path = Path("data/reference/hcc_v28_labels.csv")
    if not path.exists():
        raise HTTPException(500, "reference data missing")
    with path.open() as f:
        return list(csv.DictReader(f))


def _build_gallery(pipeline: Pipeline) -> list[GalleryItem]:
    if not GALLERY_PATH.exists():
        return []
    cases = json.loads(GALLERY_PATH.read_text())["cases"]
    out: list[GalleryItem] = []
    for c in cases:
        note = Note(note_id=c["case_id"], text=c["note_text"])
        result = pipeline.run(note)
        out.append(
            GalleryItem(
                case_id=c["case_id"],
                name=c["name"],
                tags=c.get("tags", []),
                text=c["note_text"],
                result=result,
            )
        )
    return out


# Mount the built React UI *last* so /api/* routes win.
if UI_DIST.exists():
    app.mount("/", StaticFiles(directory=UI_DIST, html=True), name="ui")