From 54e7a034be40d218aad91e9cafd3586f3841d9d9 Mon Sep 17 00:00:00 2001 From: EnjoyBacon7 <59032058+EnjoyBacon7@users.noreply.github.com> Date: Mon, 27 Jul 2026 14:30:07 +0000 Subject: [PATCH] feat(evaluation): dataset upload, storage and deletion MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `EvaluationService` gains the dataset half: an admin uploads a corpus plus a test set, both land under `/eval//`, and a row records what is there. - The CSV is parsed at upload, so a malformed test set is rejected while the admin is still looking at the dialog rather than after a run has already spent minutes indexing the corpus. - Uploads arrive as open binary streams, not `bytes`. Each is copied to disk in fixed-size blocks under a running byte budget, so a 512 MB corpus is a disk cost rather than a RAM one. The blocking I/O runs on a worker thread so it cannot stall the event loop. - The cap is enforced by reading one byte past it, not by trusting a client-supplied `Content-Length`. - Duplicate corpus basenames are rejected rather than silently overwriting — that would leave `corpus_file_count` overstating what a run will actually index. Path components in a filename are flattened. - A partial write is cleaned up: any failure after the directory is created removes it before the error propagates. - Deleting a dataset an active run is using is refused. The runner reads the corpus off disk for the whole indexing phase, so removing it mid-run would surface as a FileNotFoundError instead of a clear conflict. --- .../orchestrators/evaluation_service.py | 174 ++++++++++++++++++ .../orchestrators/test_evaluation_service.py | 94 ++++++++++ 2 files changed, 268 insertions(+) create mode 100644 openrag/services/orchestrators/evaluation_service.py create mode 100644 tests/unit/services/orchestrators/test_evaluation_service.py diff --git a/openrag/services/orchestrators/evaluation_service.py b/openrag/services/orchestrators/evaluation_service.py new file mode 100644 index 000000000..5e5aa1766 --- /dev/null +++ b/openrag/services/orchestrators/evaluation_service.py @@ -0,0 +1,174 @@ +"""EvaluationService — datasets on disk, runs dispatched to the worker layer. + +This slice covers dataset storage: an admin uploads a corpus plus a test set, +both land under ``/eval//``, and a row records what is +there. Run dispatch follows. +""" + +from __future__ import annotations + +import asyncio +import shutil +import uuid +from pathlib import Path +from typing import TYPE_CHECKING + +from core.evaluation import parse_testset +from core.models.evaluation import EvalDataset +from core.utils.exceptions import ConflictError, NotFoundError, ValidationError + +if TYPE_CHECKING: + from collections.abc import Sequence + from typing import IO + + from core.config.root import Settings + from core.ports.evaluation_repo import EvaluationRepository + +TESTSET_FILENAME = "testset.csv" +CORPUS_DIRNAME = "corpus" + +#: Block size for streaming an upload to disk. +_COPY_CHUNK_BYTES = 1024 * 1024 + + +class EvaluationService: + """Dataset storage plus run dispatch for the admin evaluation page.""" + + def __init__( + self, + *, + repo: EvaluationRepository, + config: Settings, + ) -> None: + self._repo = repo + self._config = config + self._settings = config.evaluation + self._root = Path(config.paths.data_dir) / "eval" + + # ── datasets ───────────────────────────────────────────────────── + + def _dataset_dir(self, dataset_id: str) -> Path: + return self._root / dataset_id + + async def create_dataset( + self, + *, + name: str, + corpus: Sequence[tuple[str, IO[bytes]]], + testset: IO[bytes], + user_id: int | None, + ) -> EvalDataset: + """Validate and store a corpus + test set. + + The CSV is parsed here so a malformed test set is rejected at upload + rather than after a run has already spent minutes indexing. + + Uploads arrive as open binary streams rather than ``bytes``: each is + read under a size cap and copied to disk in fixed-size blocks, so a + large corpus is never held in memory. The blocking file I/O runs on a + worker thread so it cannot stall the event loop. + """ + if not name.strip(): + raise ValidationError("Dataset name is required.", status_code=400) + if not corpus: + raise ValidationError("At least one corpus file is required.", status_code=400) + + testset_csv = await asyncio.to_thread( + self._read_capped, + testset, + self._settings.max_testset_bytes, + f"Test set exceeds the {self._settings.max_testset_mb} MB limit.", + ) + cases = parse_testset(testset_csv, max_rows=self._settings.max_testset_rows) + + dataset_id = uuid.uuid4().hex + directory = self._dataset_dir(dataset_id) + try: + written = await asyncio.to_thread(self._store_upload, directory, corpus, testset_csv) + return await self._repo.create_dataset( + EvalDataset( + id=dataset_id, + name=name.strip(), + corpus_file_count=written, + testset_row_count=len(cases), + created_by=user_id, + ) + ) + except Exception: + await asyncio.to_thread(shutil.rmtree, directory, True) + raise + + @staticmethod + def _read_capped(stream: IO[bytes], limit: int, message: str) -> bytes: + """Read a stream, refusing anything past ``limit``. + + Reads one byte beyond the cap rather than trusting a client-supplied + length, so an inflated ``Content-Length`` cannot get past it. + """ + stream.seek(0) + payload = stream.read(limit + 1) + if len(payload) > limit: + raise ValidationError(message, status_code=413) + return payload + + def _store_upload( + self, + directory: Path, + corpus: Sequence[tuple[str, IO[bytes]]], + testset_csv: bytes, + ) -> int: + """Write the corpus and test set to disk. Blocking; call in a thread.""" + corpus_dir = directory / CORPUS_DIRNAME + corpus_dir.mkdir(parents=True, exist_ok=True) + + written = 0 + budget = self._settings.max_corpus_bytes + for filename, stream in corpus: + # Flatten any path components a browser may have sent. + target = corpus_dir / Path(filename).name + if target.exists(): + raise ValidationError( + f"Corpus contains more than one file named '{target.name}'.", + status_code=400, + ) + budget -= self._copy_within_budget(stream, target, budget) + written += 1 + + (directory / TESTSET_FILENAME).write_bytes(testset_csv) + return written + + def _copy_within_budget(self, stream: IO[bytes], target: Path, budget: int) -> int: + """Copy ``stream`` into ``target``, refusing to exceed ``budget``.""" + stream.seek(0) + written = 0 + with target.open("wb") as handle: + while chunk := stream.read(_COPY_CHUNK_BYTES): + written += len(chunk) + if written > budget: + raise ValidationError( + f"Corpus exceeds the {self._settings.max_corpus_mb} MB limit.", + status_code=413, + ) + handle.write(chunk) + return written + + async def list_datasets(self) -> list[EvalDataset]: + return await self._repo.list_datasets() + + async def delete_dataset(self, dataset_id: str) -> None: + """Delete a dataset and its stored files. + + Refused while a run is using it: the runner reads the corpus from disk + for the whole indexing phase, so removing it mid-run would surface as a + confusing FileNotFoundError instead of a clear conflict. + """ + active = await self._repo.active_run() + if active is not None and active.dataset_id == dataset_id: + raise ConflictError(f"Evaluation run '{active.id}' is still using this dataset.") + + if not await self._repo.delete_dataset(dataset_id): + raise NotFoundError(f"Evaluation dataset '{dataset_id}' not found") + await asyncio.to_thread(shutil.rmtree, self._dataset_dir(dataset_id), True) + + +__all__ = ["EvaluationService"] diff --git a/tests/unit/services/orchestrators/test_evaluation_service.py b/tests/unit/services/orchestrators/test_evaluation_service.py new file mode 100644 index 000000000..bb8c2d69c --- /dev/null +++ b/tests/unit/services/orchestrators/test_evaluation_service.py @@ -0,0 +1,94 @@ +"""Tests for EvaluationService dataset storage.""" + +from __future__ import annotations + +import pytest +from core.models.evaluation import EvalDataset, EvalRun, EvalRunStatus +from services.orchestrators.evaluation_service import EvaluationService + +DATASET_ID = "ds1" + + +class FakeRepo: + def __init__(self, run: EvalRun | None = None) -> None: + self.deleted_datasets: list[str] = [] + self.dataset = EvalDataset(id=DATASET_ID, name="d", corpus_file_count=1, testset_row_count=1) + self.run = run + + async def active_run(self): + if self.run is not None and not self.run.status.is_terminal: + return self.run + return None + + async def delete_dataset(self, dataset_id): + self.deleted_datasets.append(dataset_id) + return True + + +def _service(repo, tmp_path=None, settings=None): + from core.config.root import Settings + + settings = settings or Settings() + if tmp_path is not None: + settings = settings.model_copy(update={"paths": settings.paths.model_copy(update={"data_dir": str(tmp_path)})}) + return EvaluationService(repo=repo, config=settings) + + +def _dataset_on_disk(tmp_path): + dataset_dir = tmp_path / "eval" / DATASET_ID + (dataset_dir / "corpus").mkdir(parents=True) + (dataset_dir / "testset.csv").write_text("question,expected_answer\nq,a\n", encoding="utf-8") + + +@pytest.mark.asyncio +async def test_deleting_a_dataset_in_use_is_refused(tmp_path): + """The runner reads the corpus off disk for the whole indexing phase, so + removing it mid-run would surface as a FileNotFoundError.""" + from core.utils.exceptions import ConflictError + + _dataset_on_disk(tmp_path) + run = EvalRun(id="run-1", dataset_id=DATASET_ID, status=EvalRunStatus.INDEXING) + repo = FakeRepo(run=run) + service = _service(repo, tmp_path=tmp_path) + + with pytest.raises(ConflictError): + await service.delete_dataset(DATASET_ID) + + assert repo.deleted_datasets == [] + assert (tmp_path / "eval" / DATASET_ID).exists(), "files must survive a refused delete" + + +@pytest.mark.asyncio +async def test_deleting_a_dataset_an_idle_run_used_is_allowed(tmp_path): + """Only an *active* run blocks deletion; history keeps its results.""" + _dataset_on_disk(tmp_path) + run = EvalRun(id="run-1", dataset_id=DATASET_ID, status=EvalRunStatus.COMPLETED) + repo = FakeRepo(run=run) + service = _service(repo, tmp_path=tmp_path) + + await service.delete_dataset(DATASET_ID) + + assert repo.deleted_datasets == [DATASET_ID] + assert not (tmp_path / "eval" / DATASET_ID).exists() + + +@pytest.mark.asyncio +async def test_an_oversized_test_set_is_rejected_without_buffering_it_all(tmp_path): + """The stream is read to one byte past the cap, not to its end.""" + import io + + from core.utils.exceptions import ValidationError + + service = _service(FakeRepo(), tmp_path=tmp_path) + cap = service._settings.max_testset_bytes + oversized = io.BytesIO(b"x" * (cap + 5000)) + + with pytest.raises(ValidationError) as excinfo: + await service.create_dataset( + name="d", + corpus=[("a.txt", io.BytesIO(b"hello"))], + testset=oversized, + user_id=1, + ) + assert excinfo.value.status_code == 413 + assert oversized.tell() <= cap + 1, "must stop reading once the cap is exceeded"