src/ragforge/api.py

from __future__ import annotations

import tempfile
import time
from pathlib import Path
from typing import Annotated

from fastapi import Depends, FastAPI, File, Form, Header, HTTPException, Request, UploadFile
from fastapi.responses import Response
from prometheus_client import CONTENT_TYPE_LATEST, Counter, Histogram, generate_latest

from .config import get_settings
from .evaluation import demo_benchmark_metadata, run_demo_eval
from .pipeline import RAGEngine
from .rate_limit import RateLimitExceeded, limiter
from .schemas import CorpusSummary, EvaluationRequest, QueryRequest, QueryResponse, SessionResponse
from .workspace import registry

REQUESTS = Counter("ragforge_requests_total", "Total API requests", ["endpoint", "status"])
LATENCY = Histogram("ragforge_request_latency_seconds", "API request latency", ["endpoint"])


def _auth(authorization: Annotated[str | None, Header()] = None) -> None:
    token = get_settings().app_api_token
    if not token:
        return
    if authorization != f"Bearer {token}":
        raise HTTPException(status_code=401, detail="Invalid or missing Bearer token")


def create_api() -> FastAPI:
    app = FastAPI(title="RAGForge API", version="2.0.3")

    @app.get("/api/health")
    def health():
        return {"status": "ok", "service": "RAGForge", "model": get_settings().default_model}

    @app.get("/api/v1/info")
    def info():
        s = get_settings()
        return {
            "app": s.app_name,
            "default_model": s.default_model,
            "embedding_model": s.embedding_model,
            "reranker_model": s.reranker_model,
            "native_search_model": s.native_search_model,
            "features": [
                "semantic-query-planning", "source-profile-index", "hierarchical-retrieval",
                "source-balanced-global-retrieval", "hybrid-search", "reranking", "hyde",
                "corrective-rag", "conditional-web", "self-rag", "text2sql", "ask-the-web",
                "citations", "guardrails", "layered-evaluation", "retrieval-ablation", "quality-gated-evaluation",
                "cache-bypassed-benchmarking", "quota-aware-evaluation", "retry-after-backoff",
                "workspace-preflight", "browser-session-continuity", "lazy-demo-recovery", "explicit-abstention",
                "saved-evaluation-history", "incremental-deep-evaluation", "typed-text2sql-evaluation",
                "adaptive-reranking", "semantic-citation-attribution", "insight-synthesis",
                "table-citations", "hard-mode-evaluation", "chunk-level-reranker-ablation",
                "optional-profile-benchmark", "node-latency-observability", "evaluation-run-history",
                "grounded-absence-handling", "markdown-aware-citation-coverage", "evaluation-run-provenance",
                "profile-policy-diagnostics", "context-efficiency-diagnostics",
                "focused-context-pruning", "adaptive-context-budgeting", "adaptive-retrieval-depth",
                "focused-evidence-compression", "context-budget-ablation", "compression-ablation",
                "scale-stress-evaluation", "release-readiness-checklist", "prompt-budget-observability",
                "retrieval-confidence-telemetry", "workspace-health-diagnostics"
            ],
        }

    @app.post("/api/v1/session", response_model=SessionResponse, dependencies=[Depends(_auth)])
    def new_session():
        ws = registry.create()
        return SessionResponse(session_id=ws.session_id)

    @app.get("/api/v1/session/{session_id}", dependencies=[Depends(_auth)])
    def session_status(session_id: str):
        try:
            return registry.require(session_id).stats()
        except Exception as exc:
            raise HTTPException(status_code=404, detail=str(exc)) from exc

    @app.get("/api/v1/session/{session_id}/diagnostics", dependencies=[Depends(_auth)])
    def session_diagnostics(session_id: str):
        try:
            return registry.require(session_id).health_snapshot()
        except Exception as exc:
            raise HTTPException(status_code=404, detail=str(exc)) from exc

    @app.get("/api/v1/evaluation/benchmark")
    def evaluation_benchmark():
        return demo_benchmark_metadata()

    @app.get("/api/v1/evaluation/saved/{session_id}", dependencies=[Depends(_auth)])
    def saved_evaluations(session_id: str):
        try:
            return {"session_id": session_id, "runs": registry.require(session_id).evaluation_inventory()}
        except Exception as exc:
            raise HTTPException(status_code=404, detail=str(exc)) from exc

    @app.get("/api/v1/evaluation/history/{session_id}", dependencies=[Depends(_auth)])
    def evaluation_history(session_id: str):
        try:
            return {"session_id": session_id, "runs": registry.require(session_id).evaluation_history_inventory()}
        except Exception as exc:
            raise HTTPException(status_code=404, detail=str(exc)) from exc

    @app.get("/api/v1/evaluation/saved/{session_id}/{level}", dependencies=[Depends(_auth)])
    def saved_evaluation(session_id: str, level: str):
        normalized = level.strip().title()
        if normalized not in {"Quick", "Standard", "Deep"}:
            raise HTTPException(status_code=400, detail="level must be Quick, Standard or Deep")
        try:
            ws = registry.require(session_id)
            report = ws.get_evaluation(normalized, require_current_corpus=False)
            if not report:
                raise HTTPException(status_code=404, detail=f"No saved {normalized} evaluation")
            return report
        except HTTPException:
            raise
        except Exception as exc:
            raise HTTPException(status_code=404, detail=str(exc)) from exc

    @app.post("/api/v1/ingest", response_model=CorpusSummary, dependencies=[Depends(_auth)])
    async def ingest(
        request: Request,
        session_id: Annotated[str, Form()],
        files: Annotated[list[UploadFile], File()],
        use_ocr: Annotated[bool, Form()] = False,
        semantic_chunking: Annotated[bool, Form()] = False,
    ):
        started = time.perf_counter()
        try:
            client = request.client.host if request.client else "unknown"
            limiter.check(f"api-ingest:{client}")
            ws = registry.require(session_id)
            paths = []
            max_bytes = get_settings().max_upload_mb * 1024 * 1024
            with tempfile.TemporaryDirectory() as td:
                for upload in files:
                    target = Path(td) / Path(upload.filename or "upload").name
                    written = 0
                    with target.open("wb") as fh:
                        while True:
                            block = await upload.read(1024 * 1024)
                            if not block:
                                break
                            written += len(block)
                            if written > max_bytes:
                                raise ValueError(f"{target.name} exceeds the configured upload limit")
                            fh.write(block)
                    paths.append(target)
                result = ws.ingest(paths, ocr=use_ocr, semantic_chunking=semantic_chunking)
            REQUESTS.labels("ingest", "ok").inc()
            return result
        except RateLimitExceeded as exc:
            REQUESTS.labels("ingest", "rate_limited").inc()
            raise HTTPException(status_code=429, detail=str(exc)) from exc
        except Exception as exc:
            REQUESTS.labels("ingest", "error").inc()
            raise HTTPException(status_code=400, detail=str(exc)) from exc
        finally:
            LATENCY.labels("ingest").observe(time.perf_counter() - started)

    @app.post("/api/v1/query", response_model=QueryResponse, dependencies=[Depends(_auth)])
    def query(payload: QueryRequest, request: Request):
        started = time.perf_counter()
        try:
            client = request.client.host if request.client else "unknown"
            limiter.check(f"api:{client}")
            ws = registry.require(payload.session_id)
            result = RAGEngine(ws).ask(payload.query, payload.config)
            REQUESTS.labels("query", "ok").inc()
            return result
        except RateLimitExceeded as exc:
            REQUESTS.labels("query", "rate_limited").inc()
            raise HTTPException(status_code=429, detail=str(exc)) from exc
        except Exception as exc:
            REQUESTS.labels("query", "error").inc()
            raise HTTPException(status_code=400, detail=str(exc)) from exc
        finally:
            LATENCY.labels("query").observe(time.perf_counter() - started)

    @app.post("/api/v1/evaluate/demo", dependencies=[Depends(_auth)])
    def evaluate_demo(payload: EvaluationRequest, request: Request):
        started = time.perf_counter()
        try:
            client = request.client.host if request.client else "unknown"
            limiter.check(f"api-eval:{client}")
            ws = registry.require(payload.session_id)
            required_demo_sources = {
                "acme_cloud_runbook.md",
                "orbitpay_policy.txt",
                "release_notes.html",
                "support_matrix.csv",
                "NIST_AI_RMF_1.0.pdf",
            }
            if not required_demo_sources.issubset(set(ws.sources)):
                raise ValueError(
                    "The bundled demo benchmark requires the five bundled demo sources to be indexed in this session."
                )
            benchmark_version = str(demo_benchmark_metadata().get("version", ""))
            if payload.reuse_saved:
                cached = ws.get_evaluation(
                    payload.level,
                    model=payload.model,
                    benchmark_version=benchmark_version,
                    require_current_corpus=True,
                )
                if cached:
                    REQUESTS.labels("evaluate_demo", "ok").inc()
                    return cached

            standard_base = None
            if payload.level == "Deep" and payload.reuse_saved and not payload.include_profile_benchmark:
                standard_base = ws.get_evaluation(
                    "Standard",
                    model=payload.model,
                    benchmark_version=benchmark_version,
                    require_current_corpus=True,
                )

            report = run_demo_eval(
                ws,
                api_key=None,
                model=payload.model,
                level=payload.level,
                target_rpm=payload.target_rpm,
                base_standard_report=standard_base,
                include_profile_benchmark=payload.include_profile_benchmark,
            )
            report = ws.save_evaluation(
                payload.level,
                report,
                model=payload.model,
                benchmark_version=benchmark_version,
            )
            REQUESTS.labels("evaluate_demo", "ok").inc()
            return report
        except RateLimitExceeded as exc:
            REQUESTS.labels("evaluate_demo", "rate_limited").inc()
            raise HTTPException(status_code=429, detail=str(exc)) from exc
        except Exception as exc:
            REQUESTS.labels("evaluate_demo", "error").inc()
            raise HTTPException(status_code=400, detail=str(exc)) from exc
        finally:
            LATENCY.labels("evaluate_demo").observe(time.perf_counter() - started)

    @app.get("/metrics", include_in_schema=False)
    def metrics():
        return Response(generate_latest(), media_type=CONTENT_TYPE_LATEST)

    return app