diff --git a/conf/config.yaml b/conf/config.yaml index 11d605a73..e11b849bb 100644 --- a/conf/config.yaml +++ b/conf/config.yaml @@ -122,9 +122,14 @@ verbose: level: INFO # --- Server --- -# Env: PREFERRED_URL_SCHEME +# Env: PREFERRED_URL_SCHEME, OPENRAG_INTERNAL_URL server: preferred_url_scheme: null + # internal_url — how out-of-process workers (e.g. the evaluation runner) reach + # the API from inside the deployment. Left unset here so it defaults to + # http://openrag:$APP_iPORT, following the port uvicorn actually binds. Set + # OPENRAG_INTERNAL_URL if the API is not reachable under the compose service + # name. # --- LLM Context --- # Env: MAX_LLM_CONTEXT_SIZE, MAX_OUTPUT_TOKENS @@ -363,3 +368,19 @@ mcp: similarity_threshold: 0.8 download_timeout: 30.0 max_download_bytes: 104857600 # 100 MiB + +# --- Evaluation --- +# Env: PROMPTFOO_BIN, EVAL_MAX_CORPUS_MB, EVAL_MAX_TESTSET_MB, +# EVAL_MAX_TESTSET_ROWS, EVAL_TOP_K, EVAL_TASK_TIMEOUT, +# EVAL_TASK_POLL_INTERVAL, EVAL_HTTP_TIMEOUT, EVAL_PROMPTFOO_TIMEOUT +# The runner reaches the API via server.internal_url. +evaluation: + promptfoo_bin: promptfoo + max_corpus_mb: 512 + max_testset_mb: 5 + max_testset_rows: 500 + top_k: 5 + task_timeout_seconds: 1800.0 + task_poll_seconds: 1.0 + http_timeout_seconds: 300.0 + promptfoo_timeout_seconds: 3600.0 diff --git a/docs/content/docs/documentation/env_vars.md b/docs/content/docs/documentation/env_vars.md index bced96799..f11887b76 100644 --- a/docs/content/docs/documentation/env_vars.md +++ b/docs/content/docs/documentation/env_vars.md @@ -572,6 +572,7 @@ The following environment variables configure the FastAPI server and control acc | `DEFAULT_FILE_QUOTA` | `int` | `-1` | Default per-user file quota. `<0` disables quotas globally; `>=0` sets the default limit when a user has no explicit quota. | | `PREFERRED_URL_SCHEME` | `string` | `null` | URL scheme (`http` or `https`) used when generating URLs in API responses (e.g., `task_status_url`). When running behind a reverse proxy that terminates SSL, set this to `https` to ensure generated URLs use the correct scheme. If unset, the scheme from the incoming request is used. | | `CORS_EXTRA_ORIGINS` | `string` | _(unset)_ | Semicolon-separated list of additional origins allowed by CORS (e.g. `https://app.example.com;https://other.example.com`). Extends the default list without replacing it. | +| `OPENRAG_INTERNAL_URL` | `string` | `http://openrag:$APP_iPORT` | Overrides `server.internal_url`: the base URL out-of-process workers use to reach the API from inside the deployment. Used by the evaluation runner, which uploads the corpus and drives promptfoo over HTTP from the Ray container. The default already follows `APP_iPORT`, the container-internal port uvicorn binds; set this when the API is reachable under a different host than the compose service name. The bundled [Helm chart](/openrag/documentation/kubernetes/) sets it for you, to `http://-openrag:`. | | `UVICORN_FORWARDED_ALLOW_IPS` | `string` | `127.0.0.1` | Comma-separated CIDRs/IPs (or `*`) whose `X-Forwarded-*` headers uvicorn trusts. **Required when OpenRAG runs behind a reverse proxy that lives outside loopback** (typical docker-compose / k8s — including the bundled admin-ui proxy). Otherwise `X-Forwarded-Proto` is dropped and OIDC cookies ship with `Secure=False` even over HTTPS, and `X-Forwarded-For` is dropped so per-user rate limits collapse onto the proxy's single IP. **Set this to your proxy's subnet, not `*`** — see the proxy-trust caution under [Rate Limiting](#rate-limiting) for why `*` can be spoofed. | | `MAX_UPLOAD_SIZE_MB` | `int` | `1024` | Maximum accepted upload size, in MB. `0` or a negative value means unlimited. | | `MAX_PARTITIONS_PER_USER` | `int` | `100` | Maximum number of partitions a non-admin user may own. `-1` disables the cap (unlimited). Admin users always bypass it. | @@ -673,6 +674,24 @@ OpenRAG ships a standalone [Model Context Protocol](https://modelcontextprotocol | `OPENRAG_MCP_DOWNLOAD_TIMEOUT` | `float` | `30.0` | Timeout (seconds) for the server-side `index_url` fetch (SSRF/DoS hardening). | | `OPENRAG_MCP_MAX_DOWNLOAD_BYTES` | `int` | `104857600` | Maximum bytes downloaded by an `index_url` fetch. Default is 100 MiB. | +### Evaluation + +On-demand benchmarking from the admin **System → Evaluation** tab. A run indexes a stored corpus into a throwaway partition, replays the test set through [promptfoo](https://promptfoo.dev/), and reports indexing, retrieval and answer metrics. + +These are operational limits — every value is validated as strictly positive at config load, so a typo fails at startup rather than at run time. The reserved partition prefix, the test-set CSV column names and the `file_id` alphabet are deliberately *not* configurable: they are contracts with datasets already on disk. The base URL the runner calls back on is [`OPENRAG_INTERNAL_URL`](#fastapi--access-control), a server setting rather than an evaluation one. + +| Variable | Type | Default | Description | +|----------|------|---------|-------------| +| `PROMPTFOO_BIN` | `string` | `promptfoo` | Executable the evaluation runner shells out to. The API and Ray images both install a pinned promptfoo on `PATH`; set this only for a custom location. | +| `EVAL_MAX_CORPUS_MB` | `int` | `512` | Maximum total size of one dataset's corpus upload, in MB. Enforced while streaming to disk, so an inflated `Content-Length` cannot get past it. A dataset is re-indexed on every run, so an oversized corpus costs far more than the upload itself. Also bounded by the global [`MAX_UPLOAD_SIZE_MB`](#fastapi--access-control) — raise both to go past 1024. | +| `EVAL_MAX_TESTSET_MB` | `int` | `5` | Maximum size of the test-set CSV upload, in MB. | +| `EVAL_MAX_TESTSET_ROWS` | `int` | `500` | Maximum number of questions in a test set. Each row costs one retrieval call plus one LLM-graded generation per run. | +| `EVAL_TOP_K` | `int` | `5` | Chunks retrieved per question when measuring retrieval quality. Raising it makes hit rate and recall more forgiving, so compare runs only at a fixed value. | +| `EVAL_TASK_TIMEOUT` | `float` | `1800` | Seconds to wait for one corpus file's indexing task before the run gives up on it. Raise it for slow parsers (large scanned PDFs through Marker). | +| `EVAL_TASK_POLL_INTERVAL` | `float` | `1.0` | Seconds between polls of a file's indexing task status. | +| `EVAL_HTTP_TIMEOUT` | `float` | `300` | Per-request timeout for the runner's own HTTP calls to the API. | +| `EVAL_PROMPTFOO_TIMEOUT` | `float` | `3600` | Seconds allowed for one `promptfoo eval` invocation. Every row is graded by an LLM, so raise it for a slow grader or a large test set. | + ### Advanced & Legacy Variables #### Model-endpoint seed overrides (legacy aliases) diff --git a/docs/content/docs/documentation/kubernetes.md b/docs/content/docs/documentation/kubernetes.md index 758e850e2..c73b732ae 100644 --- a/docs/content/docs/documentation/kubernetes.md +++ b/docs/content/docs/documentation/kubernetes.md @@ -55,6 +55,8 @@ This guide explains how to deploy the **OpenRAG** stack on a Kubernetes cluster - Ensure your GPU nodes have the correct NVIDIA drivers and `nvidia` `RuntimeClass` configured. +- `OPENRAG_INTERNAL_URL` is set for you, to `http://-openrag:`. Out-of-process workers — the evaluation runner, which drives its corpus upload and promptfoo over HTTP — run in the RayCluster pod and use it to reach the API. The built-in default is the compose service name and does not resolve here. Override it only if you front the API with a different in-cluster Service. + ## Managed PostgreSQL The chart can run against a database that is provisioned outside OpenRAG, which is the recommended setup on OpenShift or cloud-managed PostgreSQL. diff --git a/infra/charts/openrag-stack/values.yaml b/infra/charts/openrag-stack/values.yaml index 6e36a781d..642bd6143 100644 --- a/infra/charts/openrag-stack/values.yaml +++ b/infra/charts/openrag-stack/values.yaml @@ -334,6 +334,12 @@ env: ENABLE_RAY_SERVE: "true" RAY_SERVE_NUM_REPLICAS: "4" RAY_SERVE_PORT: "80" + # How out-of-process workers reach the API. The built-in default is + # http://openrag:$APP_iPORT, which is the compose service name and does + # not resolve here — this chart's Service is -openrag, and the + # RayCluster that runs those workers is a separate pod. Follows + # openrag.service.port, the same value that opens the container port. + OPENRAG_INTERNAL_URL: "http://{{ .Release.Name }}-openrag:{{ .Values.openrag.service.port }}" WITH_CHAINLIT_UI: "false" SAVE_UPLOADED_FILES: "false" diff --git a/openrag/api/routers/admin/partitions.py b/openrag/api/routers/admin/partitions.py index 1f098b78d..68a0bdbb9 100644 --- a/openrag/api/routers/admin/partitions.py +++ b/openrag/api/routers/admin/partitions.py @@ -72,7 +72,7 @@ async def list_existant_partitions( # partitions_with_details. Gate the all-expansion on the caller actually being # an admin, so a (legacy) partition literally named ``all`` owned by a regular # user cannot leak every partition. New ``all`` partitions are already rejected - # at creation (_RESERVED_PARTITION_NAMES). + # at creation (``RESERVED_PARTITION_NAMES``, core.models.partition). is_admin = bool(request.state.user.get("is_admin")) summaries = await service.list_partition_summaries() if is_admin and len(partitions) == 1 and partitions[0]["partition"] == "all": diff --git a/openrag/core/config/evaluation.py b/openrag/core/config/evaluation.py new file mode 100644 index 000000000..476830fe3 --- /dev/null +++ b/openrag/core/config/evaluation.py @@ -0,0 +1,60 @@ +"""Configuration for the admin evaluation feature. + +Operational limits live here rather than as constants in the code that uses +them, so a deployment can be retuned without a rebuild. + +Domain contracts stay out of this file: the reserved partition prefix, the CSV +column names and the ``file_id`` alphabet are not settings, and changing them +would invalidate stored datasets. +""" + +from __future__ import annotations + +from pydantic import Field + +from .base import ConfigMixin + + +class EvaluationConfig(ConfigMixin): + """Limits and timeouts for evaluation datasets and runs. + + Every field is bounded: these are all reachable from the environment, and a + non-positive limit does not degrade gracefully — it reaches the runner as a + cap that rejects every upload, or as a timeout that expires instantly. A + typo should fail at config load, where the message names the field. + """ + + #: Executable the runner shells out to; the images install it on PATH. + promptfoo_bin: str = Field(default="promptfoo", min_length=1) + + #: Upload caps. A dataset is re-indexed on every run, so an oversized + #: corpus costs far more than the upload itself. + max_corpus_mb: int = Field(default=512, gt=0) + max_testset_mb: int = Field(default=5, gt=0) + #: Each test-set row costs one retrieval call plus one graded generation. + max_testset_rows: int = Field(default=500, gt=0) + + #: Chunks retrieved per question by the retrieval config. Bounded like the + #: retrieval pipeline's own ``top_k``. + top_k: int = Field(default=5, gt=0, le=1000) + + #: How long to wait for one file's indexing task, and how often to poll it. + task_timeout_seconds: float = Field(default=1800.0, gt=0) + task_poll_seconds: float = Field(default=1.0, gt=0) + + #: Per-request timeout for the runner's own HTTP calls. + http_timeout_seconds: float = Field(default=300.0, gt=0) + + #: promptfoo grades every row with an LLM, so allow for a slow grader. + promptfoo_timeout_seconds: float = Field(default=3600.0, gt=0) + + @property + def max_corpus_bytes(self) -> int: + return self.max_corpus_mb * 1024 * 1024 + + @property + def max_testset_bytes(self) -> int: + return self.max_testset_mb * 1024 * 1024 + + +__all__ = ["EvaluationConfig"] diff --git a/openrag/core/config/infrastructure.py b/openrag/core/config/infrastructure.py index ecc4052dd..32d754c31 100644 --- a/openrag/core/config/infrastructure.py +++ b/openrag/core/config/infrastructure.py @@ -2,6 +2,7 @@ from __future__ import annotations +import os from pathlib import Path from pydantic import Field @@ -96,8 +97,27 @@ class PathsConfig(ConfigMixin): # --------------------------------------------------------------------------- +def _default_internal_url() -> str: + """Base URL under which the API reaches itself from inside the deployment. + + Used by out-of-process workers (e.g. the evaluation runner, which uploads a + corpus and drives promptfoo over HTTP): they run in their own container and + cannot reuse whatever host the admin's browser happened to use. + + The port follows ``APP_iPORT``, the container-internal port uvicorn binds + (``infra/scripts/entrypoint.sh``), so moving it does not silently leave + workers calling 8080. ``OPENRAG_INTERNAL_URL`` overrides the whole URL. + + ``or`` rather than a ``get`` default, to match the ``${APP_iPORT:-8080}`` + in entrypoint.sh and docker-compose.yaml: a bare ``APP_iPORT=`` line in an + env file is empty, not absent, and would otherwise yield ``http://openrag:``. + """ + return f"http://openrag:{os.environ.get('APP_iPORT') or '8080'}" + + class ServerConfig(ConfigMixin): preferred_url_scheme: str | None = None + internal_url: str = Field(default_factory=_default_internal_url) # --------------------------------------------------------------------------- diff --git a/openrag/core/config/loader.py b/openrag/core/config/loader.py index 0e0dcfe1c..1463c6dff 100644 --- a/openrag/core/config/loader.py +++ b/openrag/core/config/loader.py @@ -86,6 +86,7 @@ ("LOG_LEVEL", "verbose.level", str), # Server ("PREFERRED_URL_SCHEME", "server.preferred_url_scheme", str), + ("OPENRAG_INTERNAL_URL", "server.internal_url", str), # LLM Context ("MAX_LLM_CONTEXT_SIZE", "llm_context.max_llm_context_size", int), ("MAX_OUTPUT_TOKENS", "llm_context.max_output_tokens", int), @@ -183,6 +184,16 @@ ("OPENRAG_MCP_SIMILARITY_THRESHOLD", "mcp.similarity_threshold", float), ("OPENRAG_MCP_DOWNLOAD_TIMEOUT", "mcp.download_timeout", float), ("OPENRAG_MCP_MAX_DOWNLOAD_BYTES", "mcp.max_download_bytes", int), + # Evaluation + ("PROMPTFOO_BIN", "evaluation.promptfoo_bin", str), + ("EVAL_MAX_CORPUS_MB", "evaluation.max_corpus_mb", int), + ("EVAL_MAX_TESTSET_MB", "evaluation.max_testset_mb", int), + ("EVAL_MAX_TESTSET_ROWS", "evaluation.max_testset_rows", int), + ("EVAL_TOP_K", "evaluation.top_k", int), + ("EVAL_TASK_TIMEOUT", "evaluation.task_timeout_seconds", float), + ("EVAL_TASK_POLL_INTERVAL", "evaluation.task_poll_seconds", float), + ("EVAL_HTTP_TIMEOUT", "evaluation.http_timeout_seconds", float), + ("EVAL_PROMPTFOO_TIMEOUT", "evaluation.promptfoo_timeout_seconds", float), ] _AUDIO_EXTENSIONS = ("mp3", "flac", "ogg", "aac", "flv", "wma", "mp4") diff --git a/openrag/core/config/root.py b/openrag/core/config/root.py index 73adf8e72..a8dcf67ea 100644 --- a/openrag/core/config/root.py +++ b/openrag/core/config/root.py @@ -14,6 +14,7 @@ SemaphoreConfig, VLMConfig, ) +from .evaluation import EvaluationConfig from .indexation import LoaderConfig from .infrastructure import ( PathsConfig, @@ -66,6 +67,22 @@ class Settings(ConfigMixin): rag: RAGConfig = Field(default_factory=RAGConfig) websearch: WebSearchConfig = Field(default_factory=StaanWebSearchConfig) mcp: MCPServerConfig = Field(default_factory=MCPServerConfig) + evaluation: EvaluationConfig = Field(default_factory=EvaluationConfig) models: ModelsConfig = Field(default_factory=ModelsConfig) presets: PresetsConfig = Field(default_factory=PresetsConfig) partitions: dict[str, PartitionConfig] = Field(default_factory=dict) + + def resolved_rdb(self) -> RDBConfig: + """``rdb`` with its database name filled in. + + ``rdb.database`` is optional: historically the name is derived from the + Milvus collection. Any process opening its own Postgres connection — + the API's catalog store, or a Ray worker such as ``EvalRunner`` — must + resolve it the same way, so the derivation lives here rather than in + the callers. + """ + if self.rdb.database is not None: + return self.rdb + return self.rdb.model_copy( + update={"database": f"partitions_for_collection_{self.vectordb.collection_name}"}, + ) diff --git a/openrag/core/models/__init__.py b/openrag/core/models/__init__.py index dd6bd1a64..f02ebb5b0 100644 --- a/openrag/core/models/__init__.py +++ b/openrag/core/models/__init__.py @@ -1,4 +1,9 @@ -"""Domain models — pure Pydantic, no infrastructure imports.""" +"""Domain models — plain Pydantic models or dataclasses, no infrastructure imports. + +Types that are validated at a boundary (parsed input, stored rows) are Pydantic; +purely internal value objects may be dataclasses. Either way nothing here may +import from ``services`` or ``api``. +""" from .catalog import TERMINAL_TASK_STATES, DocumentRecord, DocumentStatus, IndexationJob, JobStatus from .chunk import Chunk, ChunkType diff --git a/openrag/core/models/evaluation.py b/openrag/core/models/evaluation.py new file mode 100644 index 000000000..1c7382273 --- /dev/null +++ b/openrag/core/models/evaluation.py @@ -0,0 +1,172 @@ +"""Domain models for the evaluation feature. + +An *evaluation dataset* pairs a corpus (the files to index) with a test set +(the questions to ask). An *evaluation run* indexes that corpus into a +throwaway partition, replays the test set against the live API through +promptfoo, and records three families of metrics: indexing speed, retrieval +quality, and answer quality. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from datetime import datetime +from enum import Enum + +#: Runs index their corpus into a throwaway partition named from the run id. +#: These are an implementation detail of an eval and are filtered out of the +#: partition listings so they never show up as user-facing collections. +EVAL_PARTITION_PREFIX = "__eval_" + + +def is_eval_partition(partition: str) -> bool: + """True for the throwaway partition a run creates.""" + return partition.startswith(EVAL_PARTITION_PREFIX) + + +class EvalRunStatus(str, Enum): + """Lifecycle of a single evaluation run. + + Mirrors the indexing task vocabulary (``services.workers.task_state``) so + the admin UI can reuse the same status styling. + """ + + QUEUED = "QUEUED" + INDEXING = "INDEXING" + EVALUATING = "EVALUATING" + COMPLETED = "COMPLETED" + FAILED = "FAILED" + CANCELLED = "CANCELLED" + + @property + def is_terminal(self) -> bool: + return self in (EvalRunStatus.COMPLETED, EvalRunStatus.FAILED, EvalRunStatus.CANCELLED) + + +@dataclass(frozen=True) +class EvalTestCase: + """One row of the uploaded test set CSV. + + ``expected_file_ids`` is optional: rows without it still contribute to the + answer-quality metrics, but are excluded from hit rate / MRR / recall + rather than being counted as misses. + """ + + query: str + expected_answer: str + expected_file_ids: tuple[str, ...] = () + + @property + def has_ground_truth_sources(self) -> bool: + return bool(self.expected_file_ids) + + +@dataclass +class EvalDataset: + """A stored corpus + test set pair.""" + + id: str + name: str + corpus_file_count: int + testset_row_count: int + created_at: datetime | None = None + created_by: int | None = None + + +@dataclass +class FileIndexingSample: + """Wall-clock cost of indexing one corpus file.""" + + filename: str + size_bytes: int + duration_seconds: float + failed: bool = False + + +@dataclass +class IndexingMetrics: + """Aggregate indexing speed over a run's corpus.""" + + files_total: int = 0 + files_failed: int = 0 + bytes_total: int = 0 + wall_seconds: float = 0.0 + files_per_minute: float = 0.0 + megabytes_per_second: float = 0.0 + p50_seconds: float = 0.0 + p95_seconds: float = 0.0 + by_extension: dict[str, dict[str, float]] = field(default_factory=dict) + samples: list[FileIndexingSample] = field(default_factory=list) + + +@dataclass +class RetrievalMetrics: + """Ranking quality of the retrieved chunks. + + ``scored_cases`` counts the test rows that carried ``expected_file_ids``; + ``skipped_cases`` counts those that did not. Both are reported so a + near-empty ground truth can never masquerade as a perfect score. + """ + + scored_cases: int = 0 + skipped_cases: int = 0 + hit_rate: float = 0.0 + mrr: float = 0.0 + recall: float = 0.0 + context_relevance: float | None = None + + +@dataclass +class AnswerMetrics: + """LLM-graded quality of the generated answers.""" + + scored_cases: int = 0 + pass_rate: float = 0.0 + factuality: float | None = None + rubric_score: float | None = None + + +@dataclass +class EvalCaseResult: + """Per-question detail surfaced in the run detail table.""" + + query: str + retrieved_file_ids: list[str] = field(default_factory=list) + expected_file_ids: list[str] = field(default_factory=list) + hit: bool | None = None + reciprocal_rank: float | None = None + answer: str | None = None + answer_passed: bool | None = None + grader_reason: str | None = None + + +@dataclass +class EvalRun: + """One evaluation execution against a dataset.""" + + id: str + dataset_id: str + status: EvalRunStatus = EvalRunStatus.QUEUED + started_at: datetime | None = None + finished_at: datetime | None = None + indexing: IndexingMetrics | None = None + retrieval: RetrievalMetrics | None = None + answer: AnswerMetrics | None = None + cases: list[EvalCaseResult] = field(default_factory=list) + error: str | None = None + created_by: int | None = None + + +__all__ = [ + "EVAL_PARTITION_PREFIX", + "AnswerMetrics", + "EvalCaseResult", + "EvalDataset", + "EvalRun", + "EvalRunStatus", + "EvalTestCase", + "FileIndexingSample", + "IndexingMetrics", + "RetrievalMetrics", + "is_eval_partition", +] diff --git a/openrag/core/models/partition.py b/openrag/core/models/partition.py new file mode 100644 index 000000000..f99fd2ca1 --- /dev/null +++ b/openrag/core/models/partition.py @@ -0,0 +1,68 @@ +"""Partition naming rules — which names exist, and which are spoken for. + +Kept out of any one feature's module because the callers are generic: +``PartitionService`` decides what may be created and what a listing shows, +``RetrievalService`` decides what ``all`` expands to. None of them should have +to name the subsystem that owns a namespace in order to ask the question. + +Adding an internal namespace is a line in ``INTERNAL_PARTITION_PREFIXES``. The +call sites do not change. +""" + +from __future__ import annotations + +from .evaluation import EVAL_PARTITION_PREFIX + +#: Names that collide with a cross-partition sentinel (``openrag-all``, +#: ``?partitions=all``). A real partition named ``all`` would make the admin +#: partition-list route expand to *every* partition. Never creatable, by any +#: caller — no internal caller has a reason to want it either. +RESERVED_PARTITION_NAMES = frozenset({"all"}) + +#: Prefixes owned by an internal subsystem. A partition under one of these is +#: real but not user-facing: hidden from the listings and from the ``all`` +#: fan-out, and creatable only by the subsystem that owns the namespace. +#: ``__eval_`` belongs to an evaluation run — see ``core.models.evaluation``. +INTERNAL_PARTITION_PREFIXES = (EVAL_PARTITION_PREFIX,) + + +def is_internal_partition(partition: str) -> bool: + """True for a partition in a namespace an internal subsystem owns. + + Hiding one of these from a listing is only half the rule: see + ``is_reserved_partition_name`` for why it must also be unwritable. + """ + return partition.startswith(INTERNAL_PARTITION_PREFIXES) + + +def is_reserved_partition_name(partition: str, *, allow_internal: bool = False) -> bool: + """True if ``partition`` may not be created by this caller. + + Compared on the stripped, lowercased name, so neither ``" all "`` nor + ``__EVAL_x`` can be spelled to sit just outside the check — the latter + would otherwise be a real, quota-consuming partition that no listing, and + therefore no admin audit, can see. + + Reserving the internal prefixes is also what keeps the wildcard honest. + They are hidden from the listings and from ``RetrievalService``'s ``all`` + fan-out, but *not* from a SUPER_ADMIN_MODE admin's raw + ``GET /search?partitions=all``, which is an intentionally unscoped Milvus + query with no partition clause to narrow. Making the names unwritable is + what guarantees the only rows there are a live subsystem's own. + + ``allow_internal`` lets a subsystem claim its own namespace; it is never + reachable from the HTTP surface. It does not unlock + ``RESERVED_PARTITION_NAMES``. + """ + normalized = partition.strip().lower() + if normalized in RESERVED_PARTITION_NAMES: + return True + return is_internal_partition(normalized) and not allow_internal + + +__all__ = [ + "INTERNAL_PARTITION_PREFIXES", + "RESERVED_PARTITION_NAMES", + "is_internal_partition", + "is_reserved_partition_name", +] diff --git a/openrag/di/repositories.py b/openrag/di/repositories.py index c39c4b090..c53ceb819 100644 --- a/openrag/di/repositories.py +++ b/openrag/di/repositories.py @@ -36,13 +36,7 @@ def create_catalog_store( ``settings.vectordb.collection_name`` so the new adapter targets the same Postgres database the legacy actor has always used. """ - rdb = settings.rdb - if rdb.database is None: - rdb = rdb.model_copy( - update={ - "database": f"partitions_for_collection_{settings.vectordb.collection_name}", - }, - ) + rdb = settings.resolved_rdb() if run_migrations is None: run_migrations = rdb.run_migrations return PostgresStore(rdb, run_migrations=run_migrations) diff --git a/openrag/services/orchestrators/partition_service.py b/openrag/services/orchestrators/partition_service.py index b0204a9e4..f405d161b 100644 --- a/openrag/services/orchestrators/partition_service.py +++ b/openrag/services/orchestrators/partition_service.py @@ -34,6 +34,7 @@ from core.config.indexation_pipeline import IndexationPipelineConfig from core.config.retrieval_pipeline import RetrievalPipelineConfig from core.indexing.validators import validate_partition_name +from core.models.partition import is_internal_partition, is_reserved_partition_name from core.models.preset import PartitionConfig from core.utils.conts import is_internal_metadata_key from core.utils.exceptions import ( @@ -56,12 +57,6 @@ logger = get_logger() -# Names reserved as cross-partition sentinels (e.g. ``openrag-all`` / -# ``?partitions=all``). A real partition named ``all`` collides with the sentinel -# and the admin partition-list route would expand it to *every* partition — see -# ``list_existant_partitions``. Matched case-insensitively. -_RESERVED_PARTITION_NAMES = frozenset({"all"}) - # Columns where an explicit ``None`` in a PATCH is a real value (SQL NULL = # "reset to default"), not the omitted-field sentinel that the None-filter # in ``update_partition`` gives every other column. @@ -242,7 +237,8 @@ async def _ensure_partition_for_operation(self, partition: str, *, operation: An raise PartitionNotFoundError(f"Partition '{partition}' does not exist.") async def list_partitions(self) -> list[dict]: - return await self._partition_repo.list_partitions() + rows = await self._partition_repo.list_partitions() + return [row for row in rows if not is_internal_partition(str(row.get("partition", "")))] async def file_counts_by_partition(self) -> dict[str, int]: """Return a ``{partition: document_count}`` map for all partitions (one query).""" @@ -262,6 +258,10 @@ async def list_partition_summaries(self) -> dict[str, dict]: summaries: dict[str, dict] = {} for r in rows: name = r["partition"] + # Internal partitions are hidden here as well as in + # list_partitions: this is what GET /partition/ responds from. + if is_internal_partition(str(name)): + continue created = r.get("created_at") summaries[name] = { "partition": name, @@ -282,6 +282,7 @@ async def create_partition( partition: str, user_id: int, *, + allow_reserved: bool = False, max_owned: int | None = None, description: str = "", embedder: str = "default", @@ -300,11 +301,17 @@ async def create_partition( are validated *before* the row is written (so a bad preset name fails fast and atomically), the non-default config columns are persisted, and the in-memory partition cache is re-resolved. + + ``allow_reserved`` is the internal escape hatch for callers that own a + reserved namespace — currently only an evaluation run creating its own + ``__eval_``. It is never reachable from the HTTP surface, and + it does not unlock the ``all`` sentinel. """ # Reserved-name check first so a name that normalises to a reserved # sentinel (e.g. " all ") returns the specific RESERVED_PARTITION_NAME - # error rather than the generic identifier-allowlist rejection. - if partition.strip().lower() in _RESERVED_PARTITION_NAMES: + # error rather than the generic identifier-allowlist rejection. What is + # reserved, and why, lives in ``core.models.partition``. + if is_reserved_partition_name(partition, allow_internal=allow_reserved): raise ValidationError( f"Partition name '{partition}' is reserved.", status_code=400, diff --git a/openrag/services/orchestrators/retrieval_service.py b/openrag/services/orchestrators/retrieval_service.py index 0fcbf4862..1eb0213ff 100644 --- a/openrag/services/orchestrators/retrieval_service.py +++ b/openrag/services/orchestrators/retrieval_service.py @@ -28,6 +28,7 @@ from collections.abc import Callable from typing import TYPE_CHECKING, Any +from core.models.partition import is_internal_partition from core.prompts import load_template_by_key from core.retrieval.pipeline import RetrieverPipeline from core.retrieval.retriever import ( @@ -227,8 +228,16 @@ def _pipeline_groups_for_partitions( # layer post-authorization (a SUPER_ADMIN_MODE admin; regular users are # already expanded to their memberships upstream), so every hydrated # partition is in scope. + # + # Every *user-facing* one, that is. An internal partition (a run's + # throwaway ``__eval_*``) is hydrated like any other — it has to be, + # because the run measures retrieval by searching it by name — but it + # is filtered out of the partition listings, so letting the wildcard + # put it back would feed an admin's ``openrag-all`` chat context from a + # partition no listing shows and no detail view can reach. Named access + # is unaffected; only the meaning of "all" is narrowed. if "all" in partitions and configs: - partitions = list(configs.keys()) + partitions = [name for name in configs if not is_internal_partition(name)] elif not partitions or not configs: # Nothing to expand (no partitions exist yet) — keep the single # legacy pipeline; there is no per-partition config to honour. diff --git a/openrag/services/persistence/migrations/run.py b/openrag/services/persistence/migrations/run.py index e2634ead0..c8100acae 100644 --- a/openrag/services/persistence/migrations/run.py +++ b/openrag/services/persistence/migrations/run.py @@ -5,24 +5,11 @@ import asyncio from core.config import load_config -from core.config.infrastructure import RDBConfig -from core.config.root import Settings from services.persistence.connection import ConnectionManager -def _rdb_config_for_migrations(settings: Settings) -> RDBConfig: - rdb = settings.rdb - if rdb.database is not None: - return rdb - return rdb.model_copy( - update={ - "database": f"partitions_for_collection_{settings.vectordb.collection_name}", - } - ) - - async def _run() -> None: - manager = ConnectionManager(_rdb_config_for_migrations(load_config())) + manager = ConnectionManager(load_config().resolved_rdb()) await manager.run_migrations() diff --git a/openrag/services/workers/indexer_pool.py b/openrag/services/workers/indexer_pool.py index c8fedb012..a67827992 100644 --- a/openrag/services/workers/indexer_pool.py +++ b/openrag/services/workers/indexer_pool.py @@ -8,11 +8,10 @@ import ray from core.config.model_endpoints import CONTROL_EXTRA_KEYS +from core.config.root import Settings from core.models.catalog import CONTENT_CLAIM_TOKEN_METADATA_KEY from services.workers.indexer_actor import IndexerWorker, delete_uploaded_file -from openrag.core.config.root import Settings - # The indexer reloads the DB-backed model-endpoint registry at most once per # this window (and on a miss), bounding both staleness and DB load regardless # of indexing throughput. @@ -30,14 +29,6 @@ def _indexer_worker_actor_name(index: int) -> str: return f"IndexerWorker-{_INDEXER_ACTOR_PROTOCOL_VERSION}-{index}" -def _catalog_rdb_config(settings: Settings) -> Any: - if settings.rdb.database is not None: - return settings.rdb - return settings.rdb.model_copy( - update={"database": f"partitions_for_collection_{settings.vectordb.collection_name}"} - ) - - @ray.remote class IndexerWorkerActor: """Thin Ray actor wrapping ``IndexerWorker`` — one instance per pool slot. @@ -110,7 +101,7 @@ def __init__(self) -> None: topic_tagger_factory=topic_tagger_factory, defer_replace_cleanup=True, ) - self._catalog_store = PostgresStore(_catalog_rdb_config(cfg), run_migrations=False) + self._catalog_store = PostgresStore(cfg.resolved_rdb(), run_migrations=False) self._catalog_initialized = False self._catalog_init_lock = asyncio.Lock() # Model-endpoint registry hydration. Unlike the API process, the indexer @@ -444,7 +435,7 @@ async def _claim_document_repo(self) -> Any: from services.storage.postgres_store import PostgresStore cfg = load_config() - store = PostgresStore(_catalog_rdb_config(cfg), run_migrations=False) + store = PostgresStore(cfg.resolved_rdb(), run_migrations=False) await store.initialize() self._claim_store = store return self._claim_store.document_repo diff --git a/tests/unit/core/config/test_evaluation_config.py b/tests/unit/core/config/test_evaluation_config.py new file mode 100644 index 000000000..03027c10c --- /dev/null +++ b/tests/unit/core/config/test_evaluation_config.py @@ -0,0 +1,86 @@ +"""Tests for EvaluationConfig — env plumbing and the bounds on every field. + +The env mapping is a hand-maintained table in ``core/config/loader.py`` and the +field names do not match the variable names (``EVAL_TASK_TIMEOUT`` sets +``task_timeout_seconds``), so a typo there is invisible until a deployment +retunes a limit and nothing happens. +""" + +from __future__ import annotations + +import pytest +from core.config import load_config +from core.config.evaluation import EvaluationConfig +from pydantic import ValidationError + +# (env var, config attribute, value to set, expected parsed value) +_ENV_OVERRIDES = [ + ("PROMPTFOO_BIN", "promptfoo_bin", "/opt/promptfoo/bin/promptfoo", "/opt/promptfoo/bin/promptfoo"), + ("EVAL_MAX_CORPUS_MB", "max_corpus_mb", "64", 64), + ("EVAL_MAX_TESTSET_MB", "max_testset_mb", "2", 2), + ("EVAL_MAX_TESTSET_ROWS", "max_testset_rows", "50", 50), + ("EVAL_TOP_K", "top_k", "10", 10), + ("EVAL_TASK_TIMEOUT", "task_timeout_seconds", "60", 60.0), + ("EVAL_TASK_POLL_INTERVAL", "task_poll_seconds", "0.25", 0.25), + ("EVAL_HTTP_TIMEOUT", "http_timeout_seconds", "30", 30.0), + ("EVAL_PROMPTFOO_TIMEOUT", "promptfoo_timeout_seconds", "120", 120.0), +] + + +@pytest.mark.parametrize(("env_var", "attribute", "raw", "expected"), _ENV_OVERRIDES) +def test_every_eval_setting_is_reachable_from_the_environment(monkeypatch, tmp_path, env_var, attribute, raw, expected): + (tmp_path / "config.yaml").write_text("retriever:\n type: single\n", encoding="utf-8") + monkeypatch.setenv(env_var, raw) + + settings = load_config(config_path=tmp_path) + + assert getattr(settings.evaluation, attribute) == expected + + +def test_the_env_table_covers_every_field(): + """A field added without an ``EVAL_*`` entry is a setting no deployment can + actually reach — the whole reason these limits are config.""" + mapped = {attribute for _, attribute, _, _ in _ENV_OVERRIDES} + + assert mapped == set(EvaluationConfig.model_fields) + + +@pytest.mark.parametrize( + "field", + [ + "max_corpus_mb", + "max_testset_mb", + "max_testset_rows", + "top_k", + "task_timeout_seconds", + "task_poll_seconds", + "http_timeout_seconds", + "promptfoo_timeout_seconds", + ], +) +def test_non_positive_limits_are_rejected(field): + """A zero or negative limit does not degrade gracefully: it reaches the + runner as a cap that rejects every upload, or a timeout that has already + expired. Fail at config load, where the error names the field.""" + with pytest.raises(ValidationError): + EvaluationConfig(**{field: 0}) + + with pytest.raises(ValidationError): + EvaluationConfig(**{field: -1}) + + +def test_an_empty_promptfoo_bin_is_rejected(): + with pytest.raises(ValidationError): + EvaluationConfig(promptfoo_bin="") + + +def test_top_k_is_bounded_like_the_retrieval_pipeline(): + with pytest.raises(ValidationError): + EvaluationConfig(top_k=1001) + + +def test_byte_properties_convert_from_megabytes(): + settings = EvaluationConfig(max_corpus_mb=3, max_testset_mb=2) + + assert settings.max_corpus_bytes == 3 * 1024 * 1024 + assert settings.max_testset_bytes == 2 * 1024 * 1024 diff --git a/tests/unit/core/config/test_resolved_rdb.py b/tests/unit/core/config/test_resolved_rdb.py new file mode 100644 index 000000000..5ef4ad220 --- /dev/null +++ b/tests/unit/core/config/test_resolved_rdb.py @@ -0,0 +1,42 @@ +"""Tests for Settings.resolved_rdb(). + +``rdb.database`` is optional in config; every process that opens its own +Postgres connection has to derive the same name. A Ray worker that skipped +this derivation died in its constructor with "RDBConfig.database is required", +which surfaced only as a run stuck in QUEUED — hence the coverage. +""" + +from __future__ import annotations + +from core.config.infrastructure import RDBConfig, VectorDBConfig +from core.config.root import Settings + + +def _settings(**rdb_fields) -> Settings: + return Settings( + rdb=RDBConfig(**rdb_fields), + vectordb=VectorDBConfig(collection_name="my_collection"), + ) + + +def test_derives_the_database_name_from_the_collection_when_unset(): + assert _settings(database=None).resolved_rdb().database == "partitions_for_collection_my_collection" + + +def test_keeps_an_explicit_database_name(): + assert _settings(database="explicit_db").resolved_rdb().database == "explicit_db" + + +def test_does_not_mutate_the_original_config(): + settings = _settings(database=None) + + settings.resolved_rdb() + + assert settings.rdb.database is None + + +def test_preserves_the_other_connection_fields(): + resolved = _settings(database=None, host="db.internal", port=6543).resolved_rdb() + + assert resolved.host == "db.internal" + assert resolved.port == 6543 diff --git a/tests/unit/core/config/test_server_internal_url.py b/tests/unit/core/config/test_server_internal_url.py new file mode 100644 index 000000000..e15155cde --- /dev/null +++ b/tests/unit/core/config/test_server_internal_url.py @@ -0,0 +1,63 @@ +"""The workers' API base URL must follow the port uvicorn actually binds.""" + +from __future__ import annotations + +from pathlib import Path + +from core.config import load_config +from core.config.infrastructure import ServerConfig + +# tests/unit/core/config/ -> repository root +_CONF_DIR = Path(__file__).resolve().parents[4] / "conf" + + +def test_internal_url_defaults_to_the_container_internal_port(monkeypatch): + """APP_iPORT is what entrypoint.sh passes to uvicorn, so a deployment that + moves it must not leave workers calling 8080.""" + monkeypatch.setenv("APP_iPORT", "9000") + assert ServerConfig().internal_url == "http://openrag:9000" + + +def test_internal_url_falls_back_to_8080(monkeypatch): + monkeypatch.delenv("APP_iPORT", raising=False) + assert ServerConfig().internal_url == "http://openrag:8080" + + +def test_an_empty_app_iport_falls_back_too(monkeypatch): + """``APP_iPORT=`` in an env file is empty, not absent. entrypoint.sh and + docker-compose.yaml both spell it ``${APP_iPORT:-8080}``, which falls back + for either — so a bare assignment must not yield ``http://openrag:``.""" + monkeypatch.setenv("APP_iPORT", "") + assert ServerConfig().internal_url == "http://openrag:8080" + + +def test_an_explicit_internal_url_wins(monkeypatch): + """OPENRAG_INTERNAL_URL resolves onto this field, so an explicit value has + to survive the default factory.""" + monkeypatch.setenv("APP_iPORT", "9000") + assert ServerConfig(internal_url="http://api.internal:1234").internal_url == "http://api.internal:1234" + + +def test_the_shipped_config_does_not_pin_internal_url(monkeypatch): + """Through the real loader and the real conf/config.yaml. + + A literal ``internal_url:`` re-added to the YAML would silently win over + the default factory and pin workers to whatever port was written there — + the exact regression this default exists to prevent. Only a load against + the shipped file can catch it. + """ + monkeypatch.setenv("APP_iPORT", "9999") + monkeypatch.delenv("OPENRAG_INTERNAL_URL", raising=False) + + settings = load_config(config_path=_CONF_DIR) + + assert settings.server.internal_url == "http://openrag:9999" + + +def test_openrag_internal_url_overrides_the_derived_default(monkeypatch): + monkeypatch.setenv("APP_iPORT", "9999") + monkeypatch.setenv("OPENRAG_INTERNAL_URL", "https://api.internal") + + settings = load_config(config_path=_CONF_DIR) + + assert settings.server.internal_url == "https://api.internal" diff --git a/tests/unit/core/models/test_partition_naming.py b/tests/unit/core/models/test_partition_naming.py new file mode 100644 index 000000000..5fc205836 --- /dev/null +++ b/tests/unit/core/models/test_partition_naming.py @@ -0,0 +1,60 @@ +"""Which partition names are spoken for. + +The two rules read alike but are not the same: ``all`` collides with a sentinel +and is closed to everyone, while an internal prefix is a namespace whose owner +is allowed in. Conflating them either hands users an invisible partition or +stops a run from creating its own. +""" + +from __future__ import annotations + +import pytest +from core.models.evaluation import EVAL_PARTITION_PREFIX +from core.models.partition import ( + INTERNAL_PARTITION_PREFIXES, + is_internal_partition, + is_reserved_partition_name, +) + + +def test_the_eval_namespace_is_registered_as_internal(): + """The registry is what the generic call sites consult; a namespace missing + from it is one no listing filters and no creation path reserves.""" + assert EVAL_PARTITION_PREFIX in INTERNAL_PARTITION_PREFIXES + + +@pytest.mark.parametrize("name", ["__eval_deadbeef", "__eval_"]) +def test_internal_partitions_are_recognised(name): + assert is_internal_partition(name) + + +@pytest.mark.parametrize("name", ["p1", "eval_x", "_eval_x", "all", "__evaluation"]) +def test_ordinary_partitions_are_not_internal(name): + assert not is_internal_partition(name) + + +@pytest.mark.parametrize("name", ["all", "ALL", " all "]) +def test_the_all_sentinel_is_reserved_however_it_is_spelled(name): + assert is_reserved_partition_name(name) + + +@pytest.mark.parametrize("name", ["all", "ALL", " all "]) +def test_the_all_sentinel_stays_reserved_for_internal_callers(name): + """It would expand a listing to every partition — wanted by no caller.""" + assert is_reserved_partition_name(name, allow_internal=True) + + +@pytest.mark.parametrize("name", ["__eval_mine", "__EVAL_mine", " __eval_mine "]) +def test_the_internal_prefix_is_reserved_against_ordinary_callers(name): + """Case and whitespace are normalised, so a name cannot be spelled to sit + just outside the check and become a partition no listing shows.""" + assert is_reserved_partition_name(name) + + +def test_a_namespace_owner_may_claim_its_own_prefix(): + assert not is_reserved_partition_name("__eval_deadbeef", allow_internal=True) + + +@pytest.mark.parametrize("name", ["p1", "eval_x", "my-partition"]) +def test_ordinary_names_are_not_reserved(name): + assert not is_reserved_partition_name(name) diff --git a/tests/unit/services/orchestrators/test_partition_preset_resolution.py b/tests/unit/services/orchestrators/test_partition_preset_resolution.py index 8f156292e..0d7a223f6 100644 --- a/tests/unit/services/orchestrators/test_partition_preset_resolution.py +++ b/tests/unit/services/orchestrators/test_partition_preset_resolution.py @@ -53,6 +53,9 @@ async def get_partition_row(self, name: str) -> dict | None: async def list_partition_rows(self) -> list[dict]: return list(self._store.values()) + async def list_partitions(self) -> list[dict]: + return list(self._store.values()) + async def update_partition(self, name: str, **fields) -> dict | None: self.calls.append(("update_partition", (name,))) row = self._store.get(name) @@ -474,6 +477,72 @@ async def test_list_partition_summaries_has_counts_and_no_pipelines(): assert "retrieval_pipeline" not in summaries["p1"] +@pytest.mark.asyncio +async def test_list_partition_summaries_hides_throwaway_eval_partitions(): + """GET /partition/ responds from here, so an orphaned __eval_ would + otherwise surface as a user-facing collection.""" + repo = _FakePartitionRepo(rows=[_full_row("p1"), _full_row("__eval_deadbeef")]) + svc = _make_service(repo) + + summaries = await svc.list_partition_summaries() + + assert set(summaries) == {"p1"} + + +@pytest.mark.asyncio +async def test_list_partitions_hides_throwaway_eval_partitions(): + repo = _FakePartitionRepo(rows=[_full_row("p1"), _full_row("__eval_deadbeef")]) + svc = _make_service(repo) + + names = [row["partition"] for row in await svc.list_partitions()] + + assert names == ["p1"] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("name", ["__eval_mine", "__EVAL_mine", " __eval_mine "]) +async def test_create_partition_rejects_the_reserved_eval_prefix(name): + """Hiding a name from the listings without reserving it at creation hands + users an invisible partition: real, quota-consuming, and absent from every + admin listing and from ``partitions=all``.""" + from core.utils.exceptions import ValidationError + + repo = _FakePartitionRepo() + svc = _make_service(repo) + + with pytest.raises(ValidationError) as excinfo: + await svc.create_partition(name, user_id=1) + + assert excinfo.value.code == "RESERVED_PARTITION_NAME" + assert repo.calls == [] + + +@pytest.mark.asyncio +async def test_a_run_may_create_its_own_eval_partition(): + """The escape hatch the evaluation service uses; not reachable over HTTP.""" + repo = _FakePartitionRepo() + svc = _make_service(repo) + + await svc.create_partition("__eval_deadbeef", user_id=1, allow_reserved=True) + + assert ("create_partition", ("__eval_deadbeef", 1, None)) in repo.calls + + +@pytest.mark.asyncio +async def test_allow_reserved_does_not_unlock_the_all_sentinel(): + """``all`` collides with the cross-partition sentinel and would expand a + listing to every partition — worse than being hidden, and wanted by no + internal caller.""" + from core.utils.exceptions import ValidationError + + svc = _make_service(_FakePartitionRepo()) + + with pytest.raises(ValidationError) as excinfo: + await svc.create_partition("all", user_id=1, allow_reserved=True) + + assert excinfo.value.code == "RESERVED_PARTITION_NAME" + + @pytest.mark.asyncio async def test_get_partition_config_missing_raises_404(): from core.utils.exceptions import PartitionNotFoundError diff --git a/tests/unit/services/orchestrators/test_retrieval_service.py b/tests/unit/services/orchestrators/test_retrieval_service.py index 217f737b4..bda8c4d60 100644 --- a/tests/unit/services/orchestrators/test_retrieval_service.py +++ b/tests/unit/services/orchestrators/test_retrieval_service.py @@ -329,6 +329,83 @@ def factory(name: str) -> FakeSearcher: assert {c.id for c in out} == {"embed-1-hit", "embed-2-hit"} +@pytest.mark.asyncio +async def test_retrieve_all_skips_throwaway_eval_partitions(): + """A run's `__eval_*` partition is hydrated like any other — the run has to + search it by name to measure retrieval — but it is filtered out of the + partition listings. The wildcard must not put it back, or a super-admin's + `openrag-all` chat draws context from a partition no listing shows. + """ + per_partition: dict[str, FakeSearcher] = {} + + def factory(name: str) -> FakeSearcher: + s = per_partition.setdefault(name, FakeSearcher()) + s.search_result = [_chunk(f"{name}-hit")] + return s + + cfg = _config() + cfg.partitions = { + "p1": _partition(name="p1", embedder="embed-1"), + "__eval_deadbeef": _partition(name="__eval_deadbeef", embedder="embed-eval"), + } + svc = RetrievalService( + searcher=FakeSearcher(), + reranker=None, + llm=None, + config=cfg, + searcher_factory=factory, + ) + + out = await svc.retrieve(partitions=["all"], query=Query(query="hello")) + + assert set(per_partition) == {"embed-1"} + assert {c.id for c in out} == {"embed-1-hit"} + + +@pytest.mark.asyncio +async def test_retrieve_names_an_eval_partition_explicitly(): + """Narrowing "all" must not cost the run its own retrieval: the eval + partition stays reachable by name, which is how the run is measured.""" + s = FakeSearcher() + s.search_result = [_chunk("eval-hit")] + cfg = _config() + cfg.partitions = {"__eval_deadbeef": _partition(name="__eval_deadbeef")} + svc = RetrievalService( + searcher=s, + reranker=None, + llm=None, + config=cfg, + searcher_factory=lambda name: s, + ) + + out = await svc.retrieve(partitions=["__eval_deadbeef"], query=Query(query="hello")) + + assert s.search_calls[0]["partition"] == ["__eval_deadbeef"] + assert {c.id for c in out} == {"eval-hit"} + + +@pytest.mark.asyncio +async def test_retrieve_all_with_only_eval_partitions_returns_nothing(): + """Fail closed rather than falling back to the unscoped legacy pipeline: + with nothing user-facing to search, "all" matches nothing.""" + s = FakeSearcher() + s.search_result = [_chunk("eval-hit")] + cfg = _config() + cfg.partitions = {"__eval_deadbeef": _partition(name="__eval_deadbeef")} + svc = RetrievalService( + searcher=s, + reranker=None, + llm=None, + config=cfg, + searcher_factory=lambda name: s, + ) + + out = await svc.retrieve(partitions=["all"], query=Query(query="hello")) + + assert out == [] + assert s.search_calls == [] + + @pytest.mark.asyncio async def test_retrieve_all_applies_partition_top_n(): """The reranker top_n was dropped on the `all` path (default_top_k was None). diff --git a/tests/unit/services/persistence/test_migration_entrypoint.py b/tests/unit/services/persistence/test_migration_entrypoint.py index 249eab8bb..5b8878a4a 100644 --- a/tests/unit/services/persistence/test_migration_entrypoint.py +++ b/tests/unit/services/persistence/test_migration_entrypoint.py @@ -3,45 +3,22 @@ from core.config.infrastructure import RDBConfig, VectorDBConfig from core.config.root import Settings +# The derivation itself is covered by tests/unit/core/config/test_resolved_rdb.py. +# What this file pins is that the standalone entrypoint goes *through* it: run +# against a different database than the API opens, and the migrations silently +# upgrade an empty one. -def test_rdb_config_for_migrations_derives_database_from_collection() -> None: - from services.persistence.migrations.run import _rdb_config_for_migrations - settings = Settings( - rdb=RDBConfig(password="x", database=None), - vectordb=VectorDBConfig(collection_name="managed"), - ) - - rdb = _rdb_config_for_migrations(settings) - - assert rdb.database == "partitions_for_collection_managed" - - -def test_rdb_config_for_migrations_preserves_explicit_database() -> None: - from services.persistence.migrations.run import _rdb_config_for_migrations - - settings = Settings( - rdb=RDBConfig(password="x", database="openrag_catalog"), - vectordb=VectorDBConfig(collection_name="managed"), - ) - - rdb = _rdb_config_for_migrations(settings) - - assert rdb.database == "openrag_catalog" - - -def test_migration_entrypoint_runs_alembic_without_initializing_pool(monkeypatch) -> None: +def _connection_manager_arg(monkeypatch, settings: Settings) -> RDBConfig: + """Drive ``main()`` with a stub manager, returning the RDBConfig it was handed.""" import services.persistence.migrations.run as migration_run + received: list[RDBConfig] = [] calls: list[str] = [] - settings = Settings( - rdb=RDBConfig(password="x", database="openrag_catalog"), - vectordb=VectorDBConfig(collection_name="managed"), - ) class FakeConnectionManager: def __init__(self, rdb): - assert rdb.database == "openrag_catalog" + received.append(rdb) async def initialize(self): # pragma: no cover - should never be called raise AssertionError("migration entrypoint must not open the application pool") @@ -54,3 +31,24 @@ async def run_migrations(self): assert migration_run.main() == 0 assert calls == ["run_migrations"] + return received[0] + + +def _settings(database: str | None) -> Settings: + return Settings( + rdb=RDBConfig(password="x", database=database), + vectordb=VectorDBConfig(collection_name="managed"), + ) + + +def test_migration_entrypoint_derives_database_from_collection(monkeypatch) -> None: + assert _connection_manager_arg(monkeypatch, _settings(None)).database == "partitions_for_collection_managed" + + +def test_migration_entrypoint_preserves_explicit_database(monkeypatch) -> None: + assert _connection_manager_arg(monkeypatch, _settings("openrag_catalog")).database == "openrag_catalog" + + +def test_migration_entrypoint_runs_alembic_without_initializing_pool(monkeypatch) -> None: + """The stub's ``initialize`` raises, so reaching it fails the test.""" + _connection_manager_arg(monkeypatch, _settings("openrag_catalog")) diff --git a/tests/unit/services/workers/test_indexer_pool.py b/tests/unit/services/workers/test_indexer_pool.py index b728a448a..ed0451380 100644 --- a/tests/unit/services/workers/test_indexer_pool.py +++ b/tests/unit/services/workers/test_indexer_pool.py @@ -1033,10 +1033,17 @@ async def _settle_pool_release_tasks(pool: object, *futures: asyncio.Future[obje async def test_claim_repo_preserves_configured_catalog_database(monkeypatch: pytest.MonkeyPatch) -> None: import core.config import services.storage.postgres_store as postgres_store + from core.config.infrastructure import RDBConfig, VectorDBConfig + from core.config.root import Settings pool = _bare_pool([_FakeWorker()]) - rdb = SimpleNamespace(database="custom_catalog") - cfg = SimpleNamespace(rdb=rdb, vectordb=SimpleNamespace(collection_name="ignored_collection")) + # A real Settings, so this exercises resolved_rdb() rather than a stub of + # it: an explicit database must survive, not be replaced by the name + # derived from the collection. + cfg = Settings( + rdb=RDBConfig(password="x", database="custom_catalog"), + vectordb=VectorDBConfig(collection_name="ignored_collection"), + ) repo = object() calls = [] @@ -1052,7 +1059,8 @@ async def initialize(self) -> None: monkeypatch.setattr(postgres_store, "PostgresStore", Store) assert await pool._claim_document_repo() is repo - assert calls == [(rdb, False), "initialized"] + assert calls == [(cfg.rdb, False), "initialized"] + assert calls[0][0].database == "custom_catalog" def test_pool_requires_positive_pool_size() -> None: @@ -1278,11 +1286,10 @@ def test_indexer_pool_wires_contextualizer_factory(monkeypatch: pytest.MonkeyPat topic_tagger_factory = object() vlm_factory = object() - class RDBConfig: - database = "custom_catalog" - - def model_copy(self, *, update): - return SimpleNamespace(**update) + # The actor asks Settings for an already-resolved RDBConfig rather than + # deriving the catalog database name itself; the derivation's own cases are + # covered by tests/unit/core/config/test_resolved_rdb.py. + catalog_rdb = SimpleNamespace(database="custom_catalog") cfg = SimpleNamespace( embedder=SimpleNamespace( @@ -1296,7 +1303,7 @@ def model_copy(self, *, update): ), loader=SimpleNamespace(parse_timeout=3600, save_uploaded_files=True), vectordb=SimpleNamespace(collection_name="vdb_test"), - rdb=RDBConfig(), + resolved_rdb=lambda: catalog_rdb, ) class Store: @@ -1346,7 +1353,7 @@ def fake_get_actor(*args, **kwargs): assert captured["contextualizer_factory"] is contextualizer_factory assert captured["topic_tagger_factory"] is topic_tagger_factory assert captured["vlm_factory"] is vlm_factory - assert captured["catalog_config"] is cfg.rdb + assert captured["catalog_config"] is catalog_rdb assert captured["catalog_config"].database == "custom_catalog" assert captured["catalog_run_migrations"] is False @@ -1366,12 +1373,6 @@ def test_indexer_pool_loads_caption_prompt_without_global_vlm_default(monkeypatc captured = {} - class RDBConfig: - database = None - - def model_copy(self, *, update): - return SimpleNamespace(**update) - cfg = SimpleNamespace( embedder=SimpleNamespace( base_url="http://embedder/v1", @@ -1384,7 +1385,9 @@ def model_copy(self, *, update): ), loader=SimpleNamespace(parse_timeout=3600, save_uploaded_files=True), vectordb=SimpleNamespace(collection_name="vdb_test"), - rdb=RDBConfig(), + # Only here so the actor's constructor can build its catalog store; + # this test is about the caption prompt. + resolved_rdb=lambda: SimpleNamespace(database="partitions_for_collection_vdb_test"), ) class Store: