"""FastAPI backend for Amana — the moderator triage console. A thin HTTP layer over the existing, UI-agnostic `src/` functions. The React SPA (in `frontend/`) talks to `/api/*`; in production this same app also serves the built SPA as static files, so the whole thing runs from one origin in one container. Design notes: - **The AI recommends; a human decides.** The override-requires-a-reason rule is enforced HERE (server-side), not just in the UI — a client cannot bypass it. - Triage is billed, so its result is cached per campaign (last-run-wins) to avoid re-charging on every interaction; pass `force=true` to re-run. - Heavy imports (pydantic-ai, chroma, sentence-transformers) stay lazy — imported inside the endpoints that need them — so startup stays fast (repo convention). Run locally: uvicorn api:app --reload --port 8000 """ from __future__ import annotations from pathlib import Path from fastapi import FastAPI, HTTPException from fastapi.middleware.cors import CORSMiddleware from fastapi.staticfiles import StaticFiles from pydantic import BaseModel from src.audit import append_decision, decided_campaign_ids, load_decisions from src.campaigns import list_campaign_paths, load_campaign from src.config import CONFIG from src.policy import get_rule app = FastAPI(title="Amana Triage API", version="1.0") # Dev convenience: the Vite dev server is a different origin. In production the SPA is served from # this same app (see the static mount at the bottom), so CORS is a no-op there. app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"], ) # Last triage shown per campaign id: {campaign_id: {"provider": str, "gated": GatedDecision}}. # In-memory by design — a demo cache, reset on restart; the audit log is the durable record. _triage_cache: dict[str, dict] = {} _PROVIDERS = ("anthropic", "ollama") # UI buttons map onto the three recommendation outcomes. _VALID_DECISIONS = ("APPROVE", "REJECT", "ESCALATE") # --------------------------------------------------------------------------- helpers def _paths_by_id() -> dict[str, Path]: """{campaign_id (file stem) -> path} for the camp-*.json submissions.""" return {p.stem: p for p in list_campaign_paths()} def _campaign_or_404(cid: str): path = _paths_by_id().get(cid) if path is None: raise HTTPException(status_code=404, detail=f"Unknown campaign {cid!r}") return load_campaign(path) def _enrich_rule(item: dict) -> dict: """Attach the cited rule's full policy text + section so the UI can show it without another call.""" rule = get_rule(item.get("rule_id", "")) item["rule_text"] = rule.text if rule else None item["section"] = rule.section if rule else None return item # --------------------------------------------------------------------------- request models class TriageRequest(BaseModel): campaign_id: str provider: str = CONFIG.llm_provider force: bool = False class DecisionRequest(BaseModel): campaign_id: str human_decision: str # APPROVE | REJECT | ESCALATE moderator: str = "" # who made the call — required (accountability) reason: str = "" # --------------------------------------------------------------------------- endpoints @app.get("/api/stats") def stats() -> dict: from src.store import count # lazy: pulls in chroma return { "policy_rules": count(CONFIG.policy_collection), "precedent_cases": count(CONFIG.cases_collection), "provider": CONFIG.llm_provider, "providers": list(_PROVIDERS), # On the public demo the provider is forced to Anthropic (see run_triage), so the UI hides # the otherwise-inert provider toggle. "public_demo": CONFIG.public_demo, } @app.get("/api/policy") def policy() -> list[dict]: """The policy grouped by section, for the moderator reference drawer (acronym glossary + full rule text). Static and cheap — no auth or cache needed.""" from src.policy import policy_sections # light parse; keep the import local to the route return policy_sections() @app.get("/api/campaigns") def list_campaigns() -> list[dict]: decided = decided_campaign_ids() out: list[dict] = [] for cid, path in _paths_by_id().items(): c = load_campaign(path) rec = decided.get(cid) out.append({ "id": c.id, "title": c.title, "category": c.category, "goal_amount": c.goal_amount, "currency": c.currency, "beneficiary": c.beneficiary.model_dump(), "organizer": c.organizer.model_dump(), "decided": rec is not None, "decision": rec.get("human_decision") if rec else None, "is_override": bool(rec.get("is_override")) if rec else False, }) return out @app.get("/api/campaigns/{cid}") def get_campaign(cid: str) -> dict: return _campaign_or_404(cid).model_dump() @app.post("/api/triage") def run_triage(req: TriageRequest) -> dict: provider = req.provider.lower() force = req.force # Public-demo spend cap (PUBLIC_DEMO=1, set on the deployed Space): bound billing BY CONSTRUCTION. # The endpoint already only triages the 18 fixed campaigns (arbitrary text can't reach the model), # so the only spend faucet is force-rerunning the cache — disabled here — and a non-Anthropic # provider (Ollama can't run on the Space anyway). With both shut, max spend is 18 calls/restart, # then cache hits. Local dev (no PUBLIC_DEMO) keeps force + the provider toggle. if CONFIG.public_demo: provider = "anthropic" force = False if provider not in _PROVIDERS: raise HTTPException(status_code=400, detail=f"provider must be one of {_PROVIDERS}") campaign = _campaign_or_404(req.campaign_id) cached = _triage_cache.get(req.campaign_id) if cached and not force and cached["provider"] == provider: gated = cached["gated"] else: from src.agent import triage, _resolve_model # lazy: pulls in pydantic-ai try: gated = triage(campaign, model=_resolve_model(provider)) except Exception as e: # surface provider/setup errors as a clean 502 raise HTTPException(status_code=502, detail=f"Triage failed: {type(e).__name__}: {e}") _triage_cache[req.campaign_id] = {"provider": provider, "gated": gated} data = gated.model_dump() for rv in data["decision"]["rule_violations"]: _enrich_rule(rv) for o in data["overrides"]: _enrich_rule(o) data["provider"] = provider return data @app.get("/api/decisions") def get_decisions() -> list[dict]: return load_decisions() @app.post("/api/decisions") def record_decision(req: DecisionRequest) -> dict: human = req.human_decision.upper() if human not in _VALID_DECISIONS: raise HTTPException(status_code=400, detail=f"human_decision must be one of {_VALID_DECISIONS}") moderator = req.moderator.strip() # Accountability: a human owns every decision, so we record who made it. if not moderator: raise HTTPException(status_code=400, detail="A moderator name is required — a human owns the decision.") cached = _triage_cache.get(req.campaign_id) if not cached: raise HTTPException(status_code=409, detail="Run AI triage for this campaign before deciding.") gated = cached["gated"] campaign = _campaign_or_404(req.campaign_id) is_override = human != gated.decision.recommendation # Server-enforced governance: contradicting the AI requires a written reason (the human owns it). if is_override and not req.reason.strip(): raise HTTPException( status_code=400, detail=f"Overriding the AI's {gated.decision.recommendation} with {human} requires a written reason.", ) record = append_decision({ "campaign_id": req.campaign_id, "title": campaign.title, "ai_recommendation": gated.decision.recommendation, "ai_llm_recommendation": gated.llm_recommendation, "gate_overrides": [o.rule_id for o in gated.overrides], "ai_confidence": gated.decision.confidence, "provider": cached["provider"], "moderator": moderator, "human_decision": human, "is_override": is_override, "reason": req.reason.strip(), }) return record # --------------------------------------------------------------------------- static SPA (prod) # In production the built React app lives in frontend/dist and is served from this same origin. # During local dev that directory doesn't exist yet — the Vite dev server serves the SPA instead. _DIST = Path(__file__).parent / "frontend" / "dist" if _DIST.is_dir(): app.mount("/", StaticFiles(directory=str(_DIST), html=True), name="spa")