Blame for app.py

"""LibBee FastAPI application for the private Hugging Face origin."""

import asyncio
import hmac
import logging
from contextlib import asynccontextmanager, suppress

from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse, Response

from src.config import LIBBEE_VERSION, get_settings
from src.services.metrics_service import MetricsService
from src.services.rag_service import RAGService

logging.basicConfig(level=logging.INFO)
# HTTPX INFO logs include complete request URLs, including research queries.
logging.getLogger("httpx").setLevel(logging.WARNING)
logging.getLogger("httpcore").setLevel(logging.WARNING)
logger = logging.getLogger(__name__)
settings = get_settings()
rag_service = RAGService()
metrics_service = MetricsService(settings.metrics_path)
_agent_slots = asyncio.Semaphore(settings.max_concurrent_agent_requests)
_startup_state = "initializing"

# Hugging Face displays a Space inside an iframe whose parent is
# huggingface.co. Only the public status pages may be framed there. Admin and
# API routes remain protected against framing/clickjacking.
_HF_PREVIEW_PATHS = {"/", "/health/live", "/health/ready"}
_HF_FRAME_ANCESTORS = "frame-ancestors 'self' https://huggingface.co https://*.huggingface.co"


def get_rag_service() -> RAGService:
    return rag_service


def get_metrics_service() -> MetricsService:
    return metrics_service


def _runtime_config() -> dict:
    from src.services.runtime_store import JsonRuntimeStore

    defaults = settings.runtime_defaults
    try:
        return JsonRuntimeStore(settings.config_path, default=defaults).load()
    except Exception:
        return defaults


async def _metrics_flush_loop() -> None:
    while True:
        await asyncio.sleep(30)
        try:
            await metrics_service.flush()
        except Exception as exc:
            logger.warning("Metrics flush error: %s", exc)


async def _initialize_services() -> None:
    global _startup_state
    from src.services.staff_service import build_staff_index

    _startup_state = "initializing"
    try:
        await asyncio.to_thread(build_staff_index)
    except Exception as exc:
        logger.error("Staff index build failed: %s", exc)
    try:
        # Local lexical retrieval can initialize without model credentials;
        # the RAG service upgrades to dense retrieval when they are available.
        await rag_service.initialize(openai_api_key=settings.openai_api_key)
        _startup_state = "ready" if rag_service.is_ready() else "failed"
    except asyncio.CancelledError:
        raise
    except Exception as exc:
        _startup_state = "failed"
        logger.error("RAG initialization failed: %s", exc, exc_info=True)


@asynccontextmanager
async def lifespan(_app: FastAPI):
    # Do not hold the HTTP listener closed while fetching remote embeddings.
    # /health/ready reports actual readiness independently of /health/live.
    startup_task = asyncio.create_task(_initialize_services())
    flush_task = asyncio.create_task(_metrics_flush_loop())
    try:
        yield
    finally:
        for task in (startup_task, flush_task):
            task.cancel()
        for task in (startup_task, flush_task):
            with suppress(asyncio.CancelledError):
                await task
        try:
            await metrics_service.flush()
        except Exception as exc:
            logger.warning("Final metrics flush failed: %s", exc)


class RequestBodyLimitMiddleware:
    """Bound received bytes, including chunked bodies without Content-Length."""

    def __init__(self, app, max_bytes: int):
        self.app = app
        self.max_bytes = max_bytes

    async def __call__(self, scope, receive, send):
        if scope["type"] != "http" or scope.get("method") not in {"POST", "PUT", "PATCH"}:
            await self.app(scope, receive, send)
            return

        async def reject():
            response = JSONResponse(status_code=413, content={"detail": "Request body too large."})
            await response(scope, receive, send)

        lengths = [value for name, value in scope.get("headers", []) if name.lower() == b"content-length"]
        if any(value.isdigit() and int(value) > self.max_bytes for value in lengths):
            await reject()
            return
        chunks = []
        total = 0
        while True:
            message = await receive()
            if message["type"] == "http.disconnect":
                return
            chunk = message.get("body", b"")
            total += len(chunk)
            if total > self.max_bytes:
                await reject()
                return
            chunks.append(chunk)
            if not message.get("more_body", False):
                break
        body = b"".join(chunks)
        delivered = False

        async def replay():
            nonlocal delivered
            if not delivered:
                delivered = True
                return {"type": "http.request", "body": body, "more_body": False}
            return await receive()

        await self.app(scope, replay, send)


app = FastAPI(
    title="LibBee - KU Library AI",
    description="Private origin API for the Khalifa University Library AI Assistant",
    version=LIBBEE_VERSION,
    lifespan=lifespan,
    docs_url=None if settings.environment == "production" else "/docs",
    openapi_url=None if settings.environment == "production" else "/openapi.json",
    redoc_url=None,
)

app.add_middleware(
    CORSMiddleware,
    allow_origins=settings.cors_origins,
    allow_credentials=False,
    allow_methods=["GET", "POST", "OPTIONS"],
    allow_headers=["Content-Type", "X-LibBee-Origin-Token", "X-Request-ID", "X-LibBee-Canary"],
    max_age=3600,
)
app.add_middleware(RequestBodyLimitMiddleware, max_bytes=settings.max_request_body_bytes)


@app.middleware("http")
async def secure_origin(request: Request, call_next):
    path = request.url.path
    protected = any(path == prefix or path.startswith(prefix + "/") for prefix in ("/agent", "/search", "/feedback"))
    acquired = False

    if path.startswith("/admin") and request.method not in {"GET", "HEAD", "OPTIONS"}:
        fetch_site = request.headers.get("sec-fetch-site", "")
        if fetch_site and fetch_site not in {"same-origin", "none"}:
            return JSONResponse(status_code=403, content={"detail": "Cross-site admin request denied."})

    # CORS preflight contains no credentials or application body. The CORS
    # middleware checks allowed origins, methods and headers before responding.
    if protected and request.method != "OPTIONS":
        supplied = request.headers.get("x-libbee-origin-token", "")
        configured = settings.origin_token
        local_insecure = settings.environment in {"development", "test"} and not configured
        if not local_insecure and (not configured or not hmac.compare_digest(supplied.encode("utf-8"), configured.encode("utf-8"))):
            return JSONResponse(status_code=401, content={"detail": "Origin authentication required."})

        runtime = _runtime_config()
        if runtime.get("maintenance_mode") and request.headers.get("x-libbee-canary") != "1":
            return JSONResponse(
                status_code=503,
                content={"detail": str(runtime.get("maintenance_message") or settings.maintenance_message)[:300]},
            )

        if path == "/agent":
            try:
                await asyncio.wait_for(_agent_slots.acquire(), timeout=0.25)
                acquired = True
            except TimeoutError:
                return JSONResponse(status_code=503, content={"detail": "LibBee is busy. Please retry shortly."})

    try:
        response = await call_next(request)
    except Exception as exc:
        logger.exception("Unhandled application error: %s", exc)
        metrics_service.incr_bucket("errors", "unhandled_exception")
        response = JSONResponse(status_code=500, content={"detail": "Unexpected server error."})
    finally:
        if acquired:
            _agent_slots.release()

    response.headers["X-Content-Type-Options"] = "nosniff"
    response.headers["Referrer-Policy"] = "no-referrer"
    response.headers["Permissions-Policy"] = "geolocation=(), microphone=(), camera=(), payment=()"
    if path in _HF_PREVIEW_PATHS:
        # X-Frame-Options cannot express an allowlist. Omit it only on these
        # harmless GET endpoints and use CSP's exact parent-origin allowlist.
        response.headers["Content-Security-Policy"] = _HF_FRAME_ANCESTORS
    else:
        response.headers["X-Frame-Options"] = "SAMEORIGIN" if path.startswith("/admin") else "DENY"
        response.headers["Content-Security-Policy"] = (
            "frame-ancestors 'self'" if path.startswith("/admin") else "frame-ancestors 'none'"
        )
    if protected or path.startswith("/admin"):
        response.headers["Cache-Control"] = "no-store"
    return response


from src.api import admin, agent, feedback, search  # noqa: E402

app.include_router(agent.router, prefix="/agent", tags=["Agent"])
app.include_router(search.router, prefix="/search", tags=["Search"])
app.include_router(admin.router, prefix="/admin", tags=["Admin"])
app.include_router(feedback.router, prefix="/feedback", tags=["Feedback"])


@app.get("/")
@app.get("/health/live")
def health_live():
    return {"status": "ok", "version": LIBBEE_VERSION, "service": "LibBee private origin"}


@app.get("/health/ready")
def health_ready():
    runtime = _runtime_config()
    rag_ready = rag_service.is_ready()
    origin_ready = bool(settings.origin_token) or settings.environment in {"development", "test"}
    if runtime.get("maintenance_mode"):
        status = "maintenance"
    elif not origin_ready:
        status = "misconfigured"
    elif not rag_ready:
        status = "initializing" if _startup_state == "initializing" else "unavailable"
    elif not settings.openai_api_key:
        status = "degraded"
    else:
        status = "ok"
    return JSONResponse(status_code=200 if status in {"ok", "degraded"} else 503, content={
        "status": status,
        "version": LIBBEE_VERSION,
        "rag_ready": rag_ready,
        "retrieval_mode": rag_service.get_stats().get("retrieval_mode", "unavailable"),
        "model_configured": bool(settings.openai_api_key),
        "maintenance_mode": bool(runtime.get("maintenance_mode")),
    })


@app.get("/config")
def public_config():
    runtime = _runtime_config()
    return {
        "welcome_message": str(runtime.get("welcome_message", ""))[:500],
        "announcement": str(runtime.get("announcement", ""))[:500],
        "maintenance_mode": bool(runtime.get("maintenance_mode")),
        "maintenance_message": str(runtime.get("maintenance_message", ""))[:300],
    }


@app.get("/favicon.ico", include_in_schema=False)
def favicon():
    return Response(status_code=204)