feat(evaluation): eval_datasets and eval_runs persistence

The `EvaluationRepository` port, its asyncpg implementation, the two
tables, and the idempotent migration that creates them.

`eval_runs` carries the metric payloads as JSONB rather than columns: the
metric set is expected to grow, and the dataclasses in
`core.models.evaluation` already define their shape.

The store is what enforces one run at a time. The partial unique index
`ux_eval_runs_single_active` admits a single row in an active status, so
`create_run` raises `ConflictError` on the loser of a race rather than the
orchestrator reading and then inserting. That matters because starting a
run regenerates a shared service user's token: two racing starts that both
passed a read check would revoke each other's credentials.

The migration is idempotent per this repo's alembic rules —
`Base.metadata.create_all()` runs at startup, so a freshly bootstrapped
database already has these tables before alembic sees them.

Author: EnjoyBacon7

openrag/core/ports/catalog_store.py

@@ -13,6 +13,7 @@
 from .conversation_repo import ConversationRepository
 from .document_repo import DocumentRepository
 from .entity_repo import EntityRepository
+from .evaluation_repo import EvaluationRepository
 from .idempotency_repo import IdempotencyRepository
 from .job_repo import JobRepository
 from .model_endpoint_repo import ModelEndpointRepository
@@ -70,6 +71,10 @@ def model_endpoint_repo(self) -> ModelEndpointRepository: ...
     @abstractmethod
     def preset_repo(self) -> PresetRepository: ...
 
+    @property
+    @abstractmethod
+    def evaluation_repo(self) -> EvaluationRepository: ...
+
     @property
     @abstractmethod
     def chunk_repo(self) -> ChunkRepository: ...

openrag/core/ports/evaluation_repo.py

@@ -0,0 +1,63 @@
+"""Port for evaluation dataset and run persistence."""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+
+from core.models.evaluation import EvalDataset, EvalRun, EvalRunStatus
+
+
+class EvaluationRepository(ABC):
+    """Storage contract for evaluation datasets and runs."""
+
+    @abstractmethod
+    async def create_dataset(self, dataset: EvalDataset) -> EvalDataset:
+        """Persist a new dataset row."""
+
+    @abstractmethod
+    async def list_datasets(self) -> list[EvalDataset]:
+        """All datasets, newest first."""
+
+    @abstractmethod
+    async def get_dataset(self, dataset_id: str) -> EvalDataset | None:
+        """One dataset, or ``None`` when it does not exist."""
+
+    @abstractmethod
+    async def delete_dataset(self, dataset_id: str) -> bool:
+        """Delete a dataset. Returns ``False`` when nothing was deleted."""
+
+    @abstractmethod
+    async def create_run(self, run: EvalRun) -> EvalRun:
+        """Persist a queued run.
+
+        Raises:
+            ConflictError: A run is already active. At most one may be, and the
+                store is what enforces it.
+        """
+
+    @abstractmethod
+    async def list_runs(self, limit: int = 50) -> list[EvalRun]:
+        """Recent runs, newest first."""
+
+    @abstractmethod
+    async def get_run(self, run_id: str) -> EvalRun | None:
+        """One run with its metrics, or ``None``."""
+
+    @abstractmethod
+    async def active_run(self) -> EvalRun | None:
+        """The run currently occupying the runner, if any.
+
+        Advisory only — mutual exclusion between starts is enforced by the
+        store's single-active-run index, not by reading this.
+        """
+
+    @abstractmethod
+    async def update_run_status(self, run_id: str, status: EvalRunStatus, *, error: str | None = None) -> None:
+        """Move a run to a new status, stamping ``finished_at`` when terminal."""
+
+    @abstractmethod
+    async def save_run_results(self, run: EvalRun) -> None:
+        """Write the metric payloads and terminal status of a finished run."""
+
+
+__all__ = ["EvaluationRepository"]

openrag/di/container.py

@@ -45,6 +45,7 @@
     from core.ports.conversation_repo import ConversationRepository
     from core.ports.document_repo import DocumentRepository
     from core.ports.entity_repo import EntityRepository
+    from core.ports.evaluation_repo import EvaluationRepository
     from core.ports.idempotency_repo import IdempotencyRepository
     from core.ports.job_repo import JobRepository
     from core.ports.model_endpoint_repo import ModelEndpointRepository
@@ -360,6 +361,10 @@ def model_endpoint_repo(self) -> ModelEndpointRepository:
     def preset_repo(self) -> PresetRepository:
         return self.catalog_store.preset_repo
 
+    @property
+    def evaluation_repo(self) -> EvaluationRepository:
+        return self.catalog_store.evaluation_repo
+
     # ------------------------------------------------------------------
     # Orchestrators (Phase 8)
     # ------------------------------------------------------------------

openrag/services/persistence/evaluation_repo.py

@@ -0,0 +1,210 @@
+"""asyncpg-backed :class:`EvaluationRepository`.
+
+Backs the ``eval_datasets`` and ``eval_runs`` tables. Metric payloads round-trip
+as JSONB; the dataclasses in ``core.models.evaluation`` define their shape.
+"""
+
+from __future__ import annotations
+
+import json
+from collections.abc import Callable
+from dataclasses import asdict
+from typing import TYPE_CHECKING, Any
+
+from core.models.evaluation import (
+    AnswerMetrics,
+    EvalCaseResult,
+    EvalDataset,
+    EvalRun,
+    EvalRunStatus,
+    FileIndexingSample,
+    IndexingMetrics,
+    RetrievalMetrics,
+)
+from core.ports.evaluation_repo import EvaluationRepository
+from core.utils.exceptions import ConflictError
+
+if TYPE_CHECKING:
+    import asyncpg
+
+_ACTIVE_STATUSES = ("QUEUED", "INDEXING", "EVALUATING")
+
+
+def _dump(payload: Any) -> str | None:
+    """Serialise a metrics dataclass — or a list of them — for a JSONB column."""
+    if payload is None:
+        return None
+    if isinstance(payload, list):
+        return json.dumps([asdict(item) for item in payload])
+    return json.dumps(asdict(payload) if hasattr(payload, "__dataclass_fields__") else payload)
+
+
+def _load(raw: Any) -> Any:
+    """asyncpg returns JSONB as ``str`` unless a codec is registered."""
+    if raw is None:
+        return None
+    return json.loads(raw) if isinstance(raw, str | bytes) else raw
+
+
+class PgEvaluationRepository(EvaluationRepository):
+    """asyncpg-backed implementation of :class:`EvaluationRepository`."""
+
+    def __init__(self, pool_getter: Callable[[], asyncpg.Pool]) -> None:
+        self._pool_getter = pool_getter
+
+    @property
+    def pool(self) -> asyncpg.Pool:
+        return self._pool_getter()
+
+    # ── datasets ─────────────────────────────────────────────────────
+
+    async def create_dataset(self, dataset: EvalDataset) -> EvalDataset:
+        row = await self.pool.fetchrow(
+            """
+            INSERT INTO eval_datasets (id, name, corpus_file_count,
+                                       testset_row_count, created_by)
+            VALUES ($1, $2, $3, $4, $5)
+            RETURNING *
+            """,
+            dataset.id,
+            dataset.name,
+            dataset.corpus_file_count,
+            dataset.testset_row_count,
+            dataset.created_by,
+        )
+        return self._row_to_dataset(row)
+
+    async def list_datasets(self) -> list[EvalDataset]:
+        rows = await self.pool.fetch("SELECT * FROM eval_datasets ORDER BY created_at DESC")
+        return [self._row_to_dataset(row) for row in rows]
+
+    async def get_dataset(self, dataset_id: str) -> EvalDataset | None:
+        row = await self.pool.fetchrow("SELECT * FROM eval_datasets WHERE id = $1", dataset_id)
+        return self._row_to_dataset(row) if row else None
+
+    async def delete_dataset(self, dataset_id: str) -> bool:
+        result = await self.pool.execute("DELETE FROM eval_datasets WHERE id = $1", dataset_id)
+        return result.endswith(" 1")
+
+    # ── runs ─────────────────────────────────────────────────────────
+
+    async def create_run(self, run: EvalRun) -> EvalRun:
+        import asyncpg
+
+        try:
+            row = await self.pool.fetchrow(
+                """
+                INSERT INTO eval_runs (id, dataset_id, status, created_by)
+                VALUES ($1, $2, $3, $4)
+                RETURNING *
+                """,
+                run.id,
+                run.dataset_id,
+                run.status.value,
+                run.created_by,
+            )
+        except asyncpg.UniqueViolationError as exc:
+            # ux_eval_runs_single_active: a run is already in flight.
+            raise ConflictError("An evaluation run is already in progress.") from exc
+        return self._row_to_run(row)
+
+    async def list_runs(self, limit: int = 50) -> list[EvalRun]:
+        rows = await self.pool.fetch("SELECT * FROM eval_runs ORDER BY started_at DESC LIMIT $1", limit)
+        return [self._row_to_run(row) for row in rows]
+
+    async def get_run(self, run_id: str) -> EvalRun | None:
+        row = await self.pool.fetchrow("SELECT * FROM eval_runs WHERE id = $1", run_id)
+        return self._row_to_run(row) if row else None
+
+    async def active_run(self) -> EvalRun | None:
+        row = await self.pool.fetchrow(
+            """
+            SELECT * FROM eval_runs
+            WHERE status = ANY($1::text[])
+            ORDER BY started_at DESC
+            LIMIT 1
+            """,
+            list(_ACTIVE_STATUSES),
+        )
+        return self._row_to_run(row) if row else None
+
+    async def update_run_status(self, run_id: str, status: EvalRunStatus, *, error: str | None = None) -> None:
+        await self.pool.execute(
+            """
+            UPDATE eval_runs
+               SET status = $2,
+                   error = COALESCE($3, error),
+                   finished_at = CASE WHEN $4 THEN now() ELSE finished_at END
+             WHERE id = $1
+            """,
+            run_id,
+            status.value,
+            error,
+            status.is_terminal,
+        )
+
+    async def save_run_results(self, run: EvalRun) -> None:
+        await self.pool.execute(
+            """
+            UPDATE eval_runs
+               SET status = $2,
+                   indexing = $3::jsonb,
+                   retrieval = $4::jsonb,
+                   answer = $5::jsonb,
+                   cases = $6::jsonb,
+                   error = $7,
+                   finished_at = now()
+             WHERE id = $1
+            """,
+            run.id,
+            run.status.value,
+            _dump(run.indexing),
+            _dump(run.retrieval),
+            _dump(run.answer),
+            _dump(run.cases),
+            run.error,
+        )
+
+    # ── row mapping ──────────────────────────────────────────────────
+
+    @staticmethod
+    def _row_to_dataset(row: Any) -> EvalDataset:
+        return EvalDataset(
+            id=row["id"],
+            name=row["name"],
+            corpus_file_count=row["corpus_file_count"],
+            testset_row_count=row["testset_row_count"],
+            created_at=row["created_at"],
+            created_by=row["created_by"],
+        )
+
+    @staticmethod
+    def _row_to_run(row: Any) -> EvalRun:
+        indexing = _load(row["indexing"])
+        retrieval = _load(row["retrieval"])
+        answer = _load(row["answer"])
+        cases = _load(row["cases"]) or []
+        samples = indexing.pop("samples", []) if indexing else []
+        return EvalRun(
+            id=row["id"],
+            dataset_id=row["dataset_id"],
+            status=EvalRunStatus(row["status"]),
+            started_at=row["started_at"],
+            finished_at=row["finished_at"],
+            indexing=(
+                IndexingMetrics(
+                    **indexing,
+                    samples=[FileIndexingSample(**sample) for sample in samples],
+                )
+                if indexing
+                else None
+            ),
+            retrieval=RetrievalMetrics(**retrieval) if retrieval else None,
+            answer=AnswerMetrics(**answer) if answer else None,
+            cases=[EvalCaseResult(**case) for case in cases],
+            error=row["error"],
+            created_by=row["created_by"],
+        )
+
+
+__all__ = ["PgEvaluationRepository"]