refactor(phase 14): Model Endpoints Registry and Per-Partition Presets

docs/refactoring/PHASE_14_MIGRATION.md

@@ -0,0 +1,140 @@
+# Phase 14 Migration — Model Endpoints & Per-Partition Presets
+
+> **Audience:** operators upgrading an existing OpenRAG deployment to the
+> Phase 14 release.
+> **TL;DR:** the upgrade is backward compatible. Existing partitions keep
+> working untouched; the only new step is a one-time seed of the default model
+> endpoints and presets, which the startup path (or `scripts/seed_presets.py`)
+> performs automatically.
+
+---
+
+## What changed
+
+Phase 14 introduces two DB-backed registries that replace the previous
+"single global config" model:
+
+1. **Model Endpoint Registry** (`model_endpoints` table → `config.models`):
+   named endpoints per model type (`embedder`, `reranker`, `llm`, `vlm`).
+   Components are built on demand from these named entries by the DI
+   component factories.
+2. **Per-Partition Preset System** (`pipeline_presets` table →
+   `config.presets`): named bundles of pipeline configuration, of two types —
+   `indexation` and `retrieval`. A partition references one preset of each type
+   by name; the retrieval preset in turn references a reranker/llm *endpoint* by
+   name.
+
+A partition row now carries `embedder`, `indexation_preset` and
+`retrieval_preset` columns. At startup these names are resolved into a cached
+`PartitionConfig` (`config.partitions`) used by the retrieval/indexing
+pipelines.
+
+> **Note:** there is no "reranker preset". The reranker is a model *endpoint*.
+> The retrieval preset only *references* it via the `reranker` (endpoint name)
+> and `enable_reranker` (on/off) fields.
+
+---
+
+## Backward compatibility
+
+The upgrade is designed to be a no-op for existing deployments:
+
+- **Schema:** the Phase 14B migration adds the new columns with
+  `server_default="default"`, so every pre-existing partition row automatically
+  references the `default` indexation and retrieval presets — no manual
+  backfill required. Migrations are idempotent (guarded by inspector existence
+  checks), so they are safe to re-run against an already-bootstrapped database.
+- **Defaults derived from your current config:** the default endpoints and
+  presets are seeded *from your existing global `Settings`* (YAML + env vars).
+  For example the default reranker endpoint comes from `RERANKER_ENDPOINT` /
+  `reranker.base_url`, and the default retrieval preset inherits your
+  `reranker.enabled` kill-switch. So the seeded defaults reproduce your current
+  behavior rather than imposing new defaults.
+- **Reranker availability:** if reranking is disabled (`RERANKER_ENABLED=false`
+  / `reranker.enabled: false`), no reranker endpoint is seeded and the default
+  retrieval preset is seeded with `enable_reranker=false`. Deployments without a
+  reachable reranker therefore keep working and do not start failing on
+  retrieval.
+
+---
+
+## Migration steps
+
+### 1. Apply the upgrade
+
+Deploy the new images / pull the new code as usual. Alembic migrations (run at
+app startup) add the new tables and columns idempotently.
+
+### 2. Seed the default endpoints and presets
+
+Seeding reads your current `Settings` and writes one default row per model type
+plus the default indexation/retrieval presets and the `default` partition.
+
+Run the one-time utility **inside the application container** (it needs the
+project venv and reaches the database over the compose network — Postgres is not
+published to the host). From the repo root:
+
+```bash
+# GPU deployment: service "openrag"; CPU deployment: service "openrag-cpu"
+docker compose -f infra/compose/docker-compose.yaml exec openrag uv run python /app/scripts/seed_presets.py
+```
+
+Expected output (counts depend on what is configured in your environment):
+
+```text
+Seeded model endpoints: 1 embedder, 1 reranker, 1 llm, 1 vlm
+Seeded 1 indexation presets, 1 retrieval presets
+Seeded 1 partition(s)
+```
+
+The script is **idempotent** — each phase skips any type/preset/partition that
+already has rows, so it is safe to run more than once (e.g. after adjusting env
+vars and re-seeding only the missing types).
+
+> If a model type has no endpoint configured (e.g. no `VLM_ENDPOINT` and an
+> empty `vlm.base_url`), that type is skipped with a log line and simply has no
+> seeded default — register one later through the admin API when needed.
+
+### 3. Verify
+
+Run `psql` inside the `rdb` container. The database name is
+`partitions_for_collection_<COLLECTION>`, where `<COLLECTION>` is your
+`vectordb.collection_name` (e.g. `partitions_for_collection_vdb`). From the repo
+root:
+
+```bash
+DB=partitions_for_collection_vdb_test   # adjust to your collection name
+
+# Model endpoints — one row per configured type, the seeded one is_default=true
+docker compose -f infra/compose/docker-compose.yaml exec rdb psql -U root -d "$DB" \
+  -c "SELECT name, model_type, endpoint, is_default FROM model_endpoints;"
+
+# Presets — default indexation + retrieval (plus the multiquery / hyde presets)
+docker compose -f infra/compose/docker-compose.yaml exec rdb psql -U root -d "$DB" \
+  -c "SELECT name, preset_type FROM pipeline_presets;"
+
+# Partitions reference presets by name
+docker compose -f infra/compose/docker-compose.yaml exec rdb psql -U root -d "$DB" \
+  -c "SELECT name, embedder, indexation_preset, retrieval_preset FROM partitions;"
+```
+
+---
+
+## After migration
+
+- **Change defaults / add endpoints:** use the admin API
+  (`/admin/model-endpoints`, `/admin/presets`) rather than editing YAML. Changes
+  are persisted to the DB, reloaded into `config.models` / `config.presets`
+  atomically, and the stale cached client is evicted so the next request builds
+  a fresh client.
+- **Per-partition configuration:** create a partition with explicit presets, or
+  update an existing one:
+
+  ```bash
+  curl -X POST http://localhost:8080/partition/research \
+    -H "Authorization: Bearer <AUTH_TOKEN>" -H "Content-Type: application/json" \
+    -d '{"indexation_preset": "default", "retrieval_preset": "hyde"}'
+  ```
+
+- The global YAML config remains the **source of the seed**: it is read once to
+  populate the registries. Ongoing operational changes live in the DB.

openrag/api/error_handlers.py

@@ -100,18 +100,22 @@ async def openrag_exception_handler(request: Request, exc: OpenRAGError) -> JSON
     available — additive, so consumers that ignore the field still parse
     the response correctly.
     """
+    status_code = _status_for(exc)
     logger.error(
-        "OpenRAGError occurred",
+        "OpenRAGError",
         error_code=getattr(exc, "code", type(exc).__name__),
-        status_code=_status_for(exc),
+        status_code=status_code,
+        message=str(exc),
+        method=request.method,
+        path=request.url.path,
     )
     body = exc.to_dict()
     request_id = _get_request_id(request)
     if request_id is not None:
         # Copy so we never mutate the exception's own ``extra`` dict —
         # the same instance may be re-raised / logged elsewhere.
         body["extra"] = {**body.get("extra", {}), "request_id": request_id}
-    return JSONResponse(status_code=_status_for(exc), content=body)
+    return JSONResponse(status_code=status_code, content=body)
 
 
 async def unhandled_exception_handler(request: Request, exc: Exception) -> JSONResponse:
@@ -121,7 +125,13 @@ async def unhandled_exception_handler(request: Request, exc: Exception) -> JSONR
     Robot Framework suite and the existing unit assertions still match
     after the move.
     """
-    logger.exception("Unhandled exception", error_type=type(exc).__name__)
+    logger.exception(
+        "Unhandled exception",
+        error_type=type(exc).__name__,
+        message=str(exc),
+        method=request.method,
+        path=request.url.path,
+    )
     extra: dict[str, object] = {}
     request_id = _get_request_id(request)
     if request_id is not None:

openrag/api/main.py

@@ -40,8 +40,10 @@
 from api.routers.admin.cluster import router as actors_router
 from api.routers.admin.indexing import router as indexer_router
 from api.routers.admin.jobs import router as queue_router
+from api.routers.admin.model_endpoints import router as model_endpoints_router
 from api.routers.admin.monitoring import router as monitoring_router
 from api.routers.admin.partitions import router as partition_router
+from api.routers.admin.presets import router as presets_router
 from api.routers.admin.tools import router as tools_router
 from api.routers.admin.users import router as users_router
 from api.routers.admin.workspaces import router as workspaces_router
@@ -102,12 +104,16 @@
 
 
 class Tags(Enum):
+    """OpenAPI tag labels used by mounted routers."""
+
     VDB = "VectorDB operations"
     INDEXER = "Indexer"
     SEARCH = "Semantic Search"
     OPENAI = "OpenAI Compatible API"
     EXTRACT = "Document extracts"
     PARTITION = "Partitions & files"
+    MODEL_ENDPOINTS = "Model Endpoints"
+    PRESETS = "Presets"
     QUEUE = "Queue management"
     ACTORS = "Ray Actors"
     USERS = "User management"
@@ -219,6 +225,7 @@ async def lifespan(app: FastAPI):
 
 
 def custom_openapi():
+    """Build the OpenAPI schema with global bearer authentication metadata."""
     if app.openapi_schema:
         return app.openapi_schema
     openapi_schema = get_openapi(
@@ -297,6 +304,7 @@ def root_redirect():
 
 @app.get("/config", summary="Get current configuration", tags=["Configuration"], dependencies=[Depends(require_admin)])
 def get_config():
+    """Return the loaded application settings for admins."""
     return settings
 
 
@@ -310,6 +318,8 @@ def get_config():
 app.include_router(extract_router, prefix="/extract", tags=[Tags.EXTRACT])
 app.include_router(search_router, prefix="/search", tags=[Tags.SEARCH])
 app.include_router(partition_router, prefix="/partition", tags=[Tags.PARTITION])
+app.include_router(model_endpoints_router, prefix="/model-endpoints", tags=[Tags.MODEL_ENDPOINTS])
+app.include_router(presets_router, prefix="/presets", tags=[Tags.PRESETS])
 app.include_router(queue_router, prefix="/queue", tags=[Tags.QUEUE])
 app.include_router(actors_router, prefix="/actors", tags=[Tags.ACTORS])
 app.include_router(users_router, prefix="/users", tags=[Tags.USERS])
@@ -338,7 +348,7 @@ def get_config():
         @serve.deployment(num_replicas=settings.ray.serve.num_replicas)
         @serve.ingress(app)
         class OpenRagAPI:
-            pass
+            """Ray Serve deployment wrapper for the FastAPI app."""
 
         serve.start(http_options={"host": settings.ray.serve.host, "port": settings.ray.serve.port})
         if WITH_CHAINLIT_UI:

openrag/api/mcp/server.py

@@ -80,7 +80,12 @@ async def _startup() -> None:
         ray.init(dashboard_host="0.0.0.0", ignore_reinit_error=True)
     ensure_worker_bootstrap()
     container = ServiceContainer(config)
-    await container.initialize()
+    try:
+        await container.initialize()
+    except Exception:
+        logger.exception("MCP server container initialization failed")
+        await container.shutdown()
+        raise
     _container = container
     logger.info("MCP server container initialized")
 

openrag/api/middleware/auth.py

@@ -153,7 +153,11 @@ async def dispatch(self, request: Request, call_next):
 
         user = None
         session = None
-        auth_service = self._get_auth_service(request)
+        try:
+            auth_service = self._get_auth_service(request)
+        except RuntimeError:
+            logger.warning("Auth service unavailable", reason="service_unavailable", path=request.url.path)
+            return JSONResponse(status_code=503, content={"detail": "Service unavailable"})
 
         # --- 1) Cookie session (OIDC UI flow). Gated on oidc mode so the
         #        legacy token-mode contract remains strictly Bearer-only —

openrag/api/routers/admin/model_endpoints.py

@@ -0,0 +1,111 @@
+"""Admin routes for the Phase 14 model endpoint registry.
+
+The router is intentionally transport-only: auth, request validation and
+response shaping live here, while endpoint persistence and validation are
+delegated to the service resolved from the DI container.
+"""
+
+from datetime import UTC, datetime
+
+from api.dependencies.auth import require_admin
+from api.schemas.admin.model_endpoint_schemas import (
+    CreateModelEndpointRequest,
+    ModelEndpointResponse,
+    ModelEndpointType,
+    UpdateModelEndpointRequest,
+    ValidateEndpointResponse,
+)
+from core.config.model_endpoints import ModelEndpointRow
+from di.providers import get_model_endpoint_service
+from fastapi import APIRouter, Depends, Response, status
+
+router = APIRouter(dependencies=[Depends(require_admin)])
+
+
+@router.post(
+    "/",
+    response_model=ModelEndpointResponse,
+    status_code=status.HTTP_201_CREATED,
+)
+async def create_model_endpoint(
+    body: CreateModelEndpointRequest,
+    service=Depends(get_model_endpoint_service),
+):
+    """Register a named inference endpoint."""
+    now = datetime.now(UTC)
+    row = ModelEndpointRow(**body.model_dump(), created_at=now, updated_at=now)
+    return await service.create_model_endpoint(row)
+
+
+@router.get("/", response_model=list[ModelEndpointResponse])
+async def list_model_endpoints(
+    model_type: ModelEndpointType | None = None,
+    service=Depends(get_model_endpoint_service),
+):
+    """List registered inference endpoints, optionally filtered by type."""
+    return await service.list_model_endpoints(model_type=model_type)
+
+
+@router.get("/{model_type}/{name}", response_model=ModelEndpointResponse)
+async def get_model_endpoint(
+    model_type: ModelEndpointType,
+    name: str,
+    service=Depends(get_model_endpoint_service),
+):
+    """Return one registered inference endpoint."""
+    return await service.get_model_endpoint(name=name, model_type=model_type)
+
+
+@router.put("/{model_type}/{name}", response_model=ModelEndpointResponse)
+async def update_model_endpoint(
+    model_type: ModelEndpointType,
+    name: str,
+    body: UpdateModelEndpointRequest,
+    service=Depends(get_model_endpoint_service),
+):
+    """Update a registered inference endpoint."""
+    fields = body.model_dump(exclude_unset=True)
+    if "name" in fields:
+        fields["new_name"] = fields.pop("name")
+    return await service.update_model_endpoint(
+        name=name,
+        model_type=model_type,
+        **fields,
+    )
+
+
+@router.delete("/{model_type}/{name}", status_code=status.HTTP_204_NO_CONTENT)
+async def delete_model_endpoint(
+    model_type: ModelEndpointType,
+    name: str,
+    service=Depends(get_model_endpoint_service),
+):
+    """Delete a registered inference endpoint."""
+    await service.delete_model_endpoint(name=name, model_type=model_type)
+    return Response(status_code=status.HTTP_204_NO_CONTENT)
+
+
+@router.post("/{model_type}/{name}/set-default", response_model=ModelEndpointResponse)
+async def set_default_model_endpoint(
+    model_type: ModelEndpointType,
+    name: str,
+    service=Depends(get_model_endpoint_service),
+):
+    """Promote a registered endpoint to the default for its type."""
+    await service.set_default(model_type=model_type, name=name)
+    return await service.get_model_endpoint(name=name, model_type=model_type)
+
+
+@router.post("/{model_type}/{name}/validate", response_model=ValidateEndpointResponse)
+async def validate_model_endpoint(
+    model_type: ModelEndpointType,
+    name: str,
+    service=Depends(get_model_endpoint_service),
+):
+    """Probe a registered endpoint for reachability and model availability."""
+    endpoint = await service.get_model_endpoint(name=name, model_type=model_type)
+    return await service.validate_endpoint(
+        url=endpoint.endpoint,
+        model_name=endpoint.model_name,
+        api_key=endpoint.extra.get("api_key"),
+    )