Core¶

def add_exception_handlers(app: FastAPI) -> None:
    """Attach application-level exception handlers to FastAPI.

    Registers custom exception handlers for application-level errors:
    - AppBaseException and its subclasses (ModelNotFoundException,
      InvalidInputException, EngineException, ...)
    - plain Exception as a last-resort fallback, returning the same
      structured 500 JSON shape and logging the traceback with the
      request id.

    Args:
        app: The FastAPI application instance to attach handlers to.

    Returns:
        None. Modifies the app instance in-place.

    Example:
        ::

            from fastapi import FastAPI
            from src.core.api import add_exception_handlers

            app = FastAPI()
            add_exception_handlers(app)
            # Exceptions now return structured JSON responses with proper HTTP codes

    Note:
        See src.core.exceptions.app_base_exception_handler for response format.
    """
    app.add_exception_handler(AppBaseException, app_base_exception_handler)
    app.add_exception_handler(Exception, unhandled_exception_handler)

def add_middleware(app: FastAPI) -> None:
    """Configure the middleware stack: host pinning + CORS + request logging.

    The API binds 127.0.0.1 without auth (single-user desktop), so the
    browser's same-origin machinery is the wall between a malicious website
    and the user's data (#89): CORS grants are restricted to the two real
    renderer origins with credentials off, and TrustedHostMiddleware rejects
    non-local Host headers (DNS rebinding). Request logging is added last so
    it sits outermost and also covers CORS-short-circuited requests.

    Args:
        app: The FastAPI application instance to configure.

    Returns:
        None. Modifies the app instance in-place.
    """
    app.add_middleware(TrustedHostMiddleware, allowed_hosts=LOCAL_HOSTS)
    app.add_middleware(
        CORSMiddleware,
        allow_origins=RENDERER_ORIGINS,
        allow_credentials=False,
        allow_methods=["*"],
        allow_headers=["*"],
        expose_headers=["X-Request-ID"],
    )
    # Added last -> outermost: every request gets an id and one access line.
    app.add_middleware(RequestLoggingMiddleware)

def apply_inference_backend_preference() -> None:
    """Replace the auto-detected CUDA engine with CPU when the user asked for it.

    Runs AFTER the migrations and the catalog population, which is the earliest
    point where ``user_settings.inference_backend`` is guaranteed to exist and
    the latest point that still precedes any inference. ``ERUDI_FORCE_CPU``
    wins over the setting: the developer override already returned
    ``CPU_Engine`` from ``get_engine()`` and there is nothing left to swap.

    A no-op unless the selected engine is ``CUDA_Engine`` and the preference is
    ``"cpu"`` -- MLX has no CPU story on Apple Silicon and an already-CPU
    machine has nothing to change.
    """
    from src.engines.cpu_engine import CPU_Engine
    from src.engines.cuda_engine import CUDA_Engine
    from src.domains.user_settings.repository import read_inference_backend

    if os.environ.get("ERUDI_FORCE_CPU"):
        return
    if config.LLM_Engine is not CUDA_Engine:
        return
    if read_inference_backend() != "cpu":
        return
    logger.info("Inference backend preference is 'cpu': switching CUDA_Engine to CPU_Engine.")
    config.LLM_Engine = CPU_Engine

def emit_engine_notice(app: FastAPI) -> None:
    """Emit an ``engine_notice`` when the selected GPU cannot run our CUDA build.

    Two NVML reads (see ``src/engines/cuda_compatibility.py``); nothing is
    applied, the frontend renders the verdict and the user decides. Silent on
    every engine but CUDA, and silent when the machine checks out.

    The event goes out on the same newline-JSON stdout channel as the launcher's
    own lifecycle events, which ``frontend/src/main.js`` forwards wholesale to
    the renderer. ``app.state.emit_event`` overrides the emitter (tests capture
    it there, exactly as they capture phases through ``app.state.emit_phase``).
    """
    from src.engines.cuda_engine import CUDA_Engine

    if config.LLM_Engine is not CUDA_Engine:
        return
    notice = cuda_preflight_notice()
    if notice is None:
        return
    emit = getattr(app.state, "emit_event", None) or emit_event
    emit(notice)

@asynccontextmanager
async def lifespan(app: FastAPI):
    """Manage application lifespan with startup and shutdown hooks.

    This async context manager orchestrates the full application lifecycle:
    - Startup: Engine selection, database initialization, cleanup task scheduling
    - Shutdown: Task cancellation, engine cleanup, resource deallocation

    Args:
        app: The FastAPI application instance (passed by FastAPI framework).

    Yields:
        None. Control is yielded to the FastAPI application for request handling.

    Raises:
        Exception: Propagates any exception during startup (database errors,
            engine initialization failures). Application will not start if
            startup fails.

    Example:
        ::

            from fastapi import FastAPI
            from src.core.api import lifespan

            app = FastAPI(lifespan=lifespan)
            # Engine initialized automatically on startup
            # Cleanup runs automatically on shutdown

    Note:
        The cleanup task runs every 30 seconds to free a model idle for over 300 seconds.
        See BaseEngine.start_cleanup_task() for details.

    Lifecycle Flow:
        1. Start the embedded PostgreSQL cluster (postgres_runtime)
        2. Bind the SQLAlchemy engine/session factory (init_database)
        3. Select engine via platform detection (BaseEngine.get_engine)
        4. Migrate the schema to head (Alembic, forward-only)
        5. Seed database with default models (startup_populate_database)
        6. Apply the persisted inference-backend preference and run the CUDA
           pre-flight (apply_inference_backend_preference / emit_engine_notice)
        7. Open the LangGraph checkpointer (app.state.checkpointer)
        8. Start the cleanup background task (30s interval) and the memory
           pressure monitor (60s interval)
        9. **[YIELD]** → Application handles requests
        10. Shutdown (reverse order): cleanup task → pressure monitor (awaited)
            → pending prefix-cache reset drained → engine → checkpointer
            → embedded PostgreSQL cluster last
    """
    # Before yield comes the startup code
    logger.info("==== Starting up... ====")
    # Startup-progress phases for the Electron loader. run.py injects the emitter
    # on app.state (same process/stdout); absent (plain uvicorn in dev/tests) this
    # is a no-op. Phases are informational only — readiness is still the `ready`
    # event / a confirming health check, never a phase.
    _emit_phase = getattr(app.state, "emit_phase", None)

    def _phase(name: str) -> None:
        if _emit_phase is not None:
            _emit_phase(name)

    # A startup failure is logged HERE, with its traceback, before it
    # propagates: uvicorn reports "Application startup failed" through its own
    # logger, which writes to stderr and never to backend.log, so without this
    # record the backend log of a backend that would not start says nothing
    # about why. run.py then turns the dead server thread into a
    # CRASH_BEFORE_READY event for Electron.
    try:
        await _start_services(app, _phase)
    except Exception as exc:
        logger.error(f"Startup failed: {exc}", exc_info=exc)
        raise
    yield
    logger.info("==== Shutting down... ====")
    wire_backfill_task = getattr(app.state, "wire_backfill_task", None)
    if wire_backfill_task is not None and not wire_backfill_task.done():
        wire_backfill_task.cancel()
    await stop_watchdog()
    config.LLM_Engine.stop_cleanup_task()
    # The monitor's sample in flight finishes first (it is shielded), and a
    # reset a torn-down turn left behind is drained: the child is never
    # stopped under a request still talking to it.
    await config.LLM_Engine.stop_pressure_task()
    await drain_pending_reset()
    config.LLM_Engine.cleanup()
    await app.state.checkpointer_cm.__aexit__(None, None, None)
    close_kb_store()
    # Cluster stops LAST: every DB consumer above must be closed first.
    stop_postgres(app.state.postgres)

def register_routers(app: FastAPI) -> None:
    """Register all domain routers to the FastAPI application.

    Attaches endpoint routers from each domain module with the /erudi prefix.
    Order of registration does not affect routing behavior due to FastAPI's
    path-matching algorithm.

    Args:
        app: The FastAPI application instance to register routers to.

    Returns:
        None. Modifies the app instance in-place.

    Example:
        ::

            from fastapi import FastAPI
            from src.core.api import register_routers

            app = FastAPI()
            register_routers(app)
            # All domain endpoints now accessible under /erudi/*
    """

    app.include_router(llms_router, prefix="/erudi")
    app.include_router(hardware_router, prefix="/erudi")
    app.include_router(arena_router, prefix="/erudi")
    app.include_router(knowledge_base_router, prefix="/erudi")
    app.include_router(conversations_router, prefix="/erudi")
    app.include_router(health_router, prefix="/erudi")
    app.include_router(startup_router, prefix="/erudi")
    app.include_router(user_settings_router, prefix="/erudi")
    app.include_router(diagnostics_router, prefix="/erudi")

def get_hf_api() -> HfApi:
    """Get or initialize HuggingFace API client (lazy-loaded singleton).

    Provides thread-safe lazy initialization of HfApi client. The client is
    created only when first requested, avoiding unnecessary initialization
    during imports or when working offline.

    Returns:
        HfApi: Configured HuggingFace API client with authentication token.

    Note:
        Thread-safe through Python's GIL. Multiple calls return same instance.
        Token is loaded from HF_TOKEN environment variable.

    Example:
        >>> from src.core import config
        >>> api = config.get_hf_api()
        >>> models = api.list_models(search="gemma")
    """
    global _HF_API
    if _HF_API is None:
        _HF_API = _RetryingHfApi(token=HF_TOKEN)
    return _HF_API

def parse_kb_agentic_flag(raw: Optional[str]) -> Optional[bool]:
    """Parse the tri-state ERUDI_KB_AGENTIC value (#298).

    Unset/empty/unrecognized -> None (per-model routing); "1"/"true" -> True
    (force agentic); "0"/"false" -> False (force systematic). Case-insensitive,
    surrounding whitespace ignored.
    """
    if raw is None:
        return None
    value = raw.strip().lower()
    if value in ("1", "true"):
        return True
    if value in ("0", "false"):
        return False
    return None

async def app_base_exception_handler(request: Request, exc: AppBaseException):
    """Global exception handler for AppBaseException and subclasses.

    Converts application exceptions into structured JSON responses with
    appropriate HTTP status codes, and writes the request's single log
    record -- method, path, status, Erudi code, message, and the trace the
    raiser attached -- at the level :func:`log_level_for_status` gives. A
    5xx record also carries the traceback (``exc_info``), which includes the
    wrapped exception when the raiser chained one. The query string is left
    out: it can carry a search term, and the path is what identifies the
    route.

    Args:
        request: Incoming FastAPI request object.
        exc: The raised AppBaseException or subclass instance.

    Returns:
        JSONResponse with structured error information.

    Examples:
        Register in FastAPI application:
            from fastapi import FastAPI
            from src.core.exceptions import app_base_exception_handler, AppBaseException

            app = FastAPI()
            app.add_exception_handler(AppBaseException, app_base_exception_handler)

    """
    level = log_level_for_status(exc.status_code)
    record = (
        f"{request.method} {request.url.path} -> {exc.status_code} {exc.erudi_code}: {exc.message}"
    )
    if exc.trace:
        record = f"{record}\n- Trace: {exc.trace}"
    logger.log(level, record, exc_info=exc if level >= logging.ERROR else None)
    error: dict = {
        "type": exc.erudi_code,
        "message": exc.message,
    }
    # Structured payload for clients to act on (e.g. the choice dialog on a
    # guarded base-model delete). Omitted entirely when unset.
    if getattr(exc, "detail", None) is not None:
        error["detail"] = exc.detail
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "success": False,
            "error": error,
        },
    )

def log_level_for_status(status_code: int) -> int:
    """The log level of a request that ends with ``status_code``.

    ``ERROR`` at 500 and above: the operation the user asked for did not
    happen and a maintainer needs the record. ``INFO`` below 500: a 404 for a
    model deleted a moment ago, a 409 on a guarded delete or a 422 on a
    rejected input is the expected outcome of asking for something that
    legitimately is not there or not allowed, not a defect. The Diagnostics
    panel shows WARNING and above, so a client error logged any higher would
    fill a page meant for bug reports with normal traffic.
    """
    return logging.ERROR if status_code >= 500 else logging.INFO

async def unhandled_exception_handler(request: Request, exc: Exception) -> JSONResponse:
    """Fallback handler for exceptions that escaped every domain handler.

    Logs the full traceback tagged with the current request id and returns a
    500 JSON body shaped exactly like app_base_exception_handler responses.
    Registered for plain ``Exception``: Starlette wires it into
    ServerErrorMiddleware, which sits OUTSIDE the request-logging middleware,
    so this response must carry the X-Request-ID header itself.

    Args:
        request: Incoming FastAPI request object.
        exc: The unhandled exception instance.

    Returns:
        JSONResponse with structured error information (HTTP 500).
    """
    request_id = get_request_id()
    logger.error(
        f"Unhandled exception on {request.method} {request.url.path}: {exc}",
        exc_info=exc,
    )
    return JSONResponse(
        status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
        content={
            "success": False,
            "error": {
                "type": "INTERNAL_SERVER_ERROR",
                "message": f"Unexpected server error: {exc}",
            },
        },
        headers={"X-Request-ID": request_id},
    )

@router.get("/")
async def health():
    """Health check endpoint returning basic service status.

    The HTTP status is ALWAYS 200 (Electron's boot readiness must not confuse
    a dead database with a dead backend); the ``db`` field carries the truth --
    ``"ok" | "recovering" | "failed"`` as seen by the DB watchdog (#162).

    Returns:
        dict: Status object with "ok" indicator, descriptive message, and the
        current database state.

    Example:
        ::

            import requests

            response = requests.get("http://127.0.0.1:27182/erudi/health/")
            assert response.json()["status"] == "ok"
    """
    return {"status": "ok", "message": "Backend is running", "db": _db_state()}