From 3468f20f1f5bba1e58f6765fce76a373c2e466e1 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 1/2] feat(evaluation): run start, dispatch and cancellation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The run half of `EvaluationService`, dispatched through the `EvaluationRunner` port so the orchestrator stays Ray-free. Setup and teardown of a run's identity live here rather than in the worker, because creating users and partitions is orchestration the API layer already owns. The worker receives a partition it may write to and a token it may use, and nothing else about the system. Ordering in `start_run` is deliberate and each step is regression-tested: 1. Ping the runner first. Dispatch is fire-and-forget, so an unreachable worker would otherwise strand the run in QUEUED with a partition and a token provisioned for nobody. It surfaces as 503. 2. Insert the run row before provisioning anything. The partial unique index is the mutual exclusion; a read-then-insert would let two racing requests both regenerate the shared eval user's token, the second revoking the credentials the first is still indexing with. The loser gets a 409 before touching anything. 3. On a failed provision, release the row and drop the partition. A run left in an active status holds the lock and would block every later run. Runs authenticate as one long-lived non-admin service user (`__openrag_eval__`) whose token is regenerated at the start of every run, so no usable plaintext token is ever stored at rest. Cancellation asks the worker first. A worker that owns the run writes its own terminal status, including metrics. `False` means nobody owns it — the run was orphaned by an actor restart — so the row is reaped here instead, otherwise it would block every subsequent run forever. --- .../orchestrators/evaluation_service.py | 233 +++++++++++++++- .../orchestrators/test_evaluation_service.py | 256 +++++++++++++++++- 2 files changed, 476 insertions(+), 13 deletions(-) diff --git a/openrag/services/orchestrators/evaluation_service.py b/openrag/services/orchestrators/evaluation_service.py index 5e5aa1766..92a1c053a 100644 --- a/openrag/services/orchestrators/evaluation_service.py +++ b/openrag/services/orchestrators/evaluation_service.py @@ -1,8 +1,19 @@ """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. +Setup and teardown of a run's *identity* live here rather than in the worker, +because creating users and partitions is orchestration the API layer already +owns. The worker receives a partition it may write to and a token it may use, +and nothing else about the system. + +Dispatch goes through the :class:`~core.evaluation.runner.EvaluationRunner` +port, so this orchestrator stays Ray-free; the Ray actor lives behind the +adapter in ``services/workers/eval_dispatcher.py``. + +The bearer token handed to the worker belongs to a single long-lived service +user (``__openrag_eval__``) whose token is **regenerated at the start of every +run**. That keeps exactly one non-admin service account in the database while +ensuring no usable plaintext token is ever stored at rest — the previous one +stops working the moment a new run starts. """ from __future__ import annotations @@ -14,15 +25,34 @@ 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 +from core.models.evaluation import ( + EVAL_PARTITION_PREFIX, + EvalDataset, + EvalRun, + EvalRunStatus, + EvalTestCase, + is_eval_partition, +) +from core.models.user import UserCreate +from core.utils.exceptions import ConflictError, NotFoundError, OpenRAGError, ValidationError +from core.utils.logging import get_logger if TYPE_CHECKING: from collections.abc import Sequence from typing import IO from core.config.root import Settings + from core.evaluation.runner import EvaluationRunner from core.ports.evaluation_repo import EvaluationRepository + from core.ports.user_repo import UserRepository + from services.orchestrators.partition_service import PartitionService + from services.orchestrators.user_service import UserService + +logger = get_logger() + +#: Stable identity of the service account runs authenticate as. +EVAL_USER_EXTERNAL_ID = "__openrag_eval__" +EVAL_USER_DISPLAY_NAME = "OpenRAG Evaluation" TESTSET_FILENAME = "testset.csv" CORPUS_DIRNAME = "corpus" @@ -31,6 +61,17 @@ _COPY_CHUNK_BYTES = 1024 * 1024 +def eval_partition_name(run_id: str) -> str: + return f"{EVAL_PARTITION_PREFIX}{run_id}" + + +class EvaluationRunnerUnavailableError(OpenRAGError): + """The runner actor could not be reached. Maps to HTTP 503.""" + + def __init__(self, message: str) -> None: + super().__init__(message, code="EVAL_RUNNER_UNAVAILABLE", status_code=503) + + class EvaluationService: """Dataset storage plus run dispatch for the admin evaluation page.""" @@ -38,9 +79,17 @@ def __init__( self, *, repo: EvaluationRepository, + runner: EvaluationRunner, + user_service: UserService, + user_repo: UserRepository, + partition_service: PartitionService, config: Settings, ) -> None: self._repo = repo + self._runner = runner + self._user_service = user_service + self._user_repo = user_repo + self._partition_service = partition_service self._config = config self._settings = config.evaluation self._root = Path(config.paths.data_dir) / "eval" @@ -170,5 +219,177 @@ async def delete_dataset(self, dataset_id: str) -> None: raise NotFoundError(f"Evaluation dataset '{dataset_id}' not found") await asyncio.to_thread(shutil.rmtree, self._dataset_dir(dataset_id), True) + # ── runs ───────────────────────────────────────────────────────── + + async def list_runs(self, limit: int = 50) -> list[EvalRun]: + return await self._repo.list_runs(limit) + + async def get_run(self, run_id: str) -> EvalRun: + run = await self._repo.get_run(run_id) + if run is None: + raise NotFoundError(f"Evaluation run '{run_id}' not found") + return run + + async def start_run(self, dataset_id: str, user_id: int | None) -> EvalRun: + """Provision a run's partition and token, then dispatch it. + + The run row is inserted before anything is provisioned: the partial + unique index ``ux_eval_runs_single_active`` makes that insert the mutual + exclusion between concurrent starts. A read-then-insert would let two + racing requests both regenerate the shared eval user's token, the second + revoking the credentials the first is still indexing with. + + Raises: + NotFoundError: The dataset does not exist. + ConflictError: Another run is already in flight — the runner + executes one at a time so timings stay comparable. + """ + dataset = await self._repo.get_dataset(dataset_id) + if dataset is None: + raise NotFoundError(f"Evaluation dataset '{dataset_id}' not found") + + directory = self._dataset_dir(dataset_id) + testset_path = directory / TESTSET_FILENAME + if not testset_path.exists(): + raise NotFoundError(f"Test set for dataset '{dataset_id}' is missing on disk") + cases = parse_testset(testset_path.read_bytes(), max_rows=self._settings.max_testset_rows) + + # Reach the runner before claiming the slot: dispatch is + # fire-and-forget, so an unreachable worker would otherwise strand the + # run in QUEUED with a partition and token provisioned for nobody. + await self._ping_runner() + + run_id = uuid.uuid4().hex + partition = eval_partition_name(run_id) + run = await self._repo.create_run( + EvalRun( + id=run_id, + dataset_id=dataset_id, + status=EvalRunStatus.QUEUED, + created_by=user_id, + ) + ) + + try: + eval_user_id = await self._ensure_eval_user() + token = (await self._user_service.regenerate_token(eval_user_id))["token"] + await self._partition_service.create_partition(partition, user_id=eval_user_id) + await self._dispatch(run_id, partition, token, directory, cases) + except Exception as exc: + # The run row is the lock; leaving it active would block every + # later run. + logger.exception(f"Could not start evaluation run {run_id}: {exc}") + await self._repo.update_run_status( + run_id, + EvalRunStatus.FAILED, + error=f"Could not start the run: {exc}", + ) + await self._drop_orphaned_partition(run_id) + raise + + logger.bind(run_id=run_id, dataset_id=dataset_id).info("Dispatched evaluation run") + return run + + async def _dispatch( + self, + run_id: str, + partition: str, + token: str, + directory: Path, + cases: Sequence[EvalTestCase], + ) -> None: + """Hand the run to the worker. + + Fire and forget: the worker owns the run from here and records its own + outcome. + """ + await self._runner.dispatch( + run_id=run_id, + partition=partition, + token=token, + api_base_url=self._config.server.internal_url, + corpus_dir=str(directory / CORPUS_DIRNAME), + cases=[ + { + "query": case.query, + "expected_answer": case.expected_answer, + "expected_file_ids": list(case.expected_file_ids), + } + for case in cases + ], + ) + + async def cancel_run(self, run_id: str) -> EvalRun: + """Ask the worker to stop, or reap the run if no worker owns it. + + The worker writes the terminal status for a run it is executing. When + it disowns the run — it restarted, or died before picking the run up — + nothing else would ever move that row out of an active status, and it + would block every subsequent run. Cancelling reaps it instead. + """ + run = await self.get_run(run_id) + if run.status.is_terminal: + raise ConflictError(f"Evaluation run '{run_id}' has already finished.") + + owned = False + try: + owned = await self._runner.cancel(run_id) + except Exception as exc: # noqa: BLE001 — an unreachable runner still has to be reaped + logger.warning(f"Evaluation runner unreachable while cancelling {run_id}: {exc}") + + if not owned: + await self._repo.update_run_status( + run_id, + EvalRunStatus.CANCELLED, + error="No runner owns this run — it was orphaned and has been reaped.", + ) + await self._drop_orphaned_partition(run_id) + return await self.get_run(run_id) + + async def _drop_orphaned_partition(self, run_id: str) -> None: + """Best-effort cleanup of the throwaway partition of a reaped run.""" + try: + await self._partition_service.delete_partition(eval_partition_name(run_id)) + except Exception as exc: # noqa: BLE001 — it may never have been created + logger.debug(f"No eval partition to drop for run {run_id}: {exc}") + + # ── internals ──────────────────────────────────────────────────── + + async def _ping_runner(self) -> None: + """Fail fast when the runner cannot be reached. + + Raises: + OpenRAGError: The worker is unreachable — surfaced to the caller + instead of being discovered as a run that never leaves QUEUED. + """ + try: + await self._runner.is_busy() + except Exception as exc: + logger.exception(f"Evaluation runner is unavailable: {exc}") + raise EvaluationRunnerUnavailableError(f"The evaluation runner could not be reached: {exc}") from exc + + async def _ensure_eval_user(self) -> int: + """Get-or-create the non-admin service user runs authenticate as.""" + existing = await self._user_repo.get_user_by_external_id(EVAL_USER_EXTERNAL_ID) + if existing is not None: + return int(existing.id) + created = await self._user_service.create_user( + UserCreate( + display_name=EVAL_USER_DISPLAY_NAME, + external_user_id=EVAL_USER_EXTERNAL_ID, + is_admin=False, + # A corpus is uploaded on every run, so a quota would fail the + # second one for reasons unrelated to the eval. + file_quota=-1, + ) + ) + return int(created["id"]) + -__all__ = ["EvaluationService"] +__all__ = [ + "EVAL_PARTITION_PREFIX", + "EVAL_USER_EXTERNAL_ID", + "EvaluationService", + "eval_partition_name", + "is_eval_partition", +] diff --git a/tests/unit/services/orchestrators/test_evaluation_service.py b/tests/unit/services/orchestrators/test_evaluation_service.py index bb8c2d69c..e25c6d416 100644 --- a/tests/unit/services/orchestrators/test_evaluation_service.py +++ b/tests/unit/services/orchestrators/test_evaluation_service.py @@ -1,10 +1,21 @@ -"""Tests for EvaluationService dataset storage.""" +"""Tests for EvaluationService run dispatch and cancellation. + +Both behaviours here were written after a real deployment produced a run that +sat in QUEUED forever: the runner actor had died in its constructor, dispatch +is fire-and-forget so nothing noticed, and cancelling could not clear the row +because no actor claimed it — which blocked every later run. +""" from __future__ import annotations import pytest +from core.evaluation.runner import EvaluationRunner from core.models.evaluation import EvalDataset, EvalRun, EvalRunStatus -from services.orchestrators.evaluation_service import EvaluationService +from core.utils.exceptions import ConflictError +from services.orchestrators.evaluation_service import ( + EvaluationRunnerUnavailableError, + EvaluationService, +) DATASET_ID = "ds1" @@ -14,6 +25,17 @@ 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 + self.status_updates: list[tuple[str, EvalRunStatus, str | None]] = [] + + async def get_dataset(self, dataset_id): + return self.dataset if dataset_id == DATASET_ID else None + + async def create_run(self, run): + self.run = run + return run + + async def get_run(self, run_id): + return self.run async def active_run(self): if self.run is not None and not self.run.status.is_terminal: @@ -24,14 +46,178 @@ async def delete_dataset(self, dataset_id): self.deleted_datasets.append(dataset_id) return True + async def update_run_status(self, run_id, status, *, error=None): + self.status_updates.append((run_id, status, error)) + if self.run is not None: + self.run.status = status + self.run.error = error + + +class FakeRunner(EvaluationRunner): + """In-memory ``EvaluationRunner`` — no Ray, no actor, no worker process.""" + + def __init__(self, *, busy_error: Exception | None = None, owns: bool = True) -> None: + self._busy_error = busy_error + self._owns = owns + self.dispatched: dict | None = None + + async def is_busy(self) -> bool: + if self._busy_error: + raise self._busy_error + return False + + async def dispatch(self, **kwargs) -> None: + self.dispatched = kwargs + + async def cancel(self, run_id: str) -> bool: + return self._owns + + +class FakePartitionService: + def __init__(self, create_error: Exception | None = None) -> None: + self.deleted: list[str] = [] + self.created: list[str] = [] + self._create_error = create_error + + async def delete_partition(self, partition): + self.deleted.append(partition) + + async def create_partition(self, partition, user_id=None): + if self._create_error: + raise self._create_error + self.created.append(partition) + + +class FakeUserRepo: + """Serves only what the ``UserRepository`` port declares.""" + + def __init__(self, user=None) -> None: + self.user = user + + async def get_user_by_external_id(self, external_id): + return self.user + -def _service(repo, tmp_path=None, settings=None): +class FakeUserService: + """Counts token regeneration — the side effect the run lock protects.""" + + def __init__(self) -> None: + self.regenerated = 0 + + async def regenerate_token(self, user_id): + self.regenerated += 1 + return {"token": "or-testtoken"} + + +def _service( + repo, + runner, + partition_service=None, + tmp_path=None, + settings=None, + user_repo=None, + user_service=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) + return EvaluationService( + repo=repo, + runner=runner, + user_service=user_service or FakeUserService(), + user_repo=user_repo if user_repo is not None else FakeUserRepo(), + partition_service=partition_service or FakePartitionService(), + config=settings, + ) + + +@pytest.mark.asyncio +async def test_start_run_refuses_when_the_runner_cannot_be_reached(tmp_path): + """A dead actor must surface as an error, not as a run stuck in QUEUED.""" + 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") + + repo = FakeRepo() + runner = FakeRunner(busy_error=RuntimeError("actor died in __init__")) + service = _service(repo, runner, tmp_path=tmp_path) + + with pytest.raises(EvaluationRunnerUnavailableError): + await service.start_run(DATASET_ID, user_id=1) + + assert runner.dispatched is None + # Nothing was provisioned and no run row was left behind. + assert repo.run is None + + +@pytest.mark.asyncio +async def test_dispatch_uses_the_configured_internal_url(tmp_path): + """The worker's API base URL comes from Settings, not from the environment.""" + from core.config.root import Settings + from core.models.user import User + + 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") + + settings = Settings() + settings = settings.model_copy( + update={"server": settings.server.model_copy(update={"internal_url": "http://api.internal:9000"})} + ) + runner = FakeRunner() + service = _service( + FakeRepo(), + runner, + tmp_path=tmp_path, + settings=settings, + # The eval service user already exists, so it is resolved through the + # port's ``get_user_by_external_id`` rather than being created. + user_repo=FakeUserRepo(User(id=7, external_user_id="__openrag_eval__")), + ) + + await service.start_run(DATASET_ID, user_id=1) + + assert runner.dispatched is not None + assert runner.dispatched["api_base_url"] == "http://api.internal:9000" + assert runner.dispatched["cases"] == [{"query": "q", "expected_answer": "a", "expected_file_ids": []}] + + +@pytest.mark.asyncio +async def test_cancel_reaps_a_run_no_runner_owns(): + """Otherwise the orphaned row blocks every subsequent run forever.""" + run = EvalRun(id="r1", dataset_id=DATASET_ID, status=EvalRunStatus.QUEUED) + repo = FakeRepo(run) + partitions = FakePartitionService() + service = _service(repo, FakeRunner(owns=False), partition_service=partitions) + + result = await service.cancel_run("r1") + + assert result.status is EvalRunStatus.CANCELLED + assert "orphaned" in (result.error or "") + assert partitions.deleted == ["__eval_r1"] + + +@pytest.mark.asyncio +async def test_cancel_leaves_an_owned_run_for_the_worker_to_finalise(): + """The worker writes its own terminal status, including the metrics.""" + run = EvalRun(id="r1", dataset_id=DATASET_ID, status=EvalRunStatus.EVALUATING) + repo = FakeRepo(run) + service = _service(repo, FakeRunner(owns=True)) + + await service.cancel_run("r1") + + assert repo.status_updates == [] + + +@pytest.mark.asyncio +async def test_cancel_rejects_an_already_finished_run(): + run = EvalRun(id="r1", dataset_id=DATASET_ID, status=EvalRunStatus.COMPLETED) + service = _service(FakeRepo(run), FakeRunner()) + + with pytest.raises(ConflictError): + await service.cancel_run("r1") def _dataset_on_disk(tmp_path): @@ -40,6 +226,62 @@ def _dataset_on_disk(tmp_path): (dataset_dir / "testset.csv").write_text("question,expected_answer\nq,a\n", encoding="utf-8") +@pytest.mark.asyncio +async def test_a_second_start_is_refused_before_the_token_is_regenerated(tmp_path): + """The run row is the lock, so the 409 has to land before provisioning. + + Regenerating the shared eval user's token is what makes a lost race + destructive: it revokes the credentials the in-flight run is indexing with. + """ + from core.utils.exceptions import ConflictError + + _dataset_on_disk(tmp_path) + + class BusyRepo(FakeRepo): + async def create_run(self, run): + raise ConflictError("An evaluation run is already in progress.") + + repo = BusyRepo() + runner = FakeRunner() + users = FakeUserService() + partitions = FakePartitionService() + service = _service(repo, runner, partition_service=partitions, tmp_path=tmp_path, user_service=users) + + with pytest.raises(ConflictError): + await service.start_run(DATASET_ID, user_id=1) + + assert users.regenerated == 0, "the loser must not touch the in-flight run's token" + assert partitions.created == [] + assert runner.dispatched is None + + +@pytest.mark.asyncio +async def test_a_failed_provision_releases_the_run_lock(tmp_path): + """A run left in an active status would block every later run.""" + _dataset_on_disk(tmp_path) + + from core.models.user import User + + repo = FakeRepo() + runner = FakeRunner() + partitions = FakePartitionService(create_error=RuntimeError("milvus unreachable")) + service = _service( + repo, + runner, + partition_service=partitions, + tmp_path=tmp_path, + user_repo=FakeUserRepo(User(id=7, external_user_id="__openrag_eval__")), + ) + + with pytest.raises(RuntimeError): + await service.start_run(DATASET_ID, user_id=1) + + assert runner.dispatched is None + statuses = [status for _, status, _ in repo.status_updates] + assert EvalRunStatus.FAILED in statuses, "the lock must be released" + assert partitions.deleted, "the throwaway partition must not leak" + + @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 @@ -49,7 +291,7 @@ async def test_deleting_a_dataset_in_use_is_refused(tmp_path): _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) + service = _service(repo, FakeRunner(), tmp_path=tmp_path) with pytest.raises(ConflictError): await service.delete_dataset(DATASET_ID) @@ -64,7 +306,7 @@ async def test_deleting_a_dataset_an_idle_run_used_is_allowed(tmp_path): _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) + service = _service(repo, FakeRunner(), tmp_path=tmp_path) await service.delete_dataset(DATASET_ID) @@ -79,7 +321,7 @@ async def test_an_oversized_test_set_is_rejected_without_buffering_it_all(tmp_pa from core.utils.exceptions import ValidationError - service = _service(FakeRepo(), tmp_path=tmp_path) + service = _service(FakeRepo(), FakeRunner(), tmp_path=tmp_path) cap = service._settings.max_testset_bytes oversized = io.BytesIO(b"x" * (cap + 5000)) From cc3263cfba7c73b2736419835e11bbc2b41db057 Mon Sep 17 00:00:00 2001 From: EnjoyBacon7 <59032058+EnjoyBacon7@users.noreply.github.com> Date: Mon, 27 Jul 2026 14:44:52 +0000 Subject: [PATCH 2/2] fix(evaluation): claim the reserved eval partition explicitly `__eval_*` is now rejected on the public creation path (part 1/15), so a run has to opt in to the namespace it owns. Without this the first run fails at `create_partition` with RESERVED_PARTITION_NAME. The unit tests did not catch it: `FakePartitionService` stands in for the whole service, so it accepted a name the real one refuses. The fake now asserts the flag, which is the only thing that keeps the two in step. --- openrag/services/orchestrators/evaluation_service.py | 5 ++++- tests/unit/services/orchestrators/test_evaluation_service.py | 5 ++++- 2 files changed, 8 insertions(+), 2 deletions(-) diff --git a/openrag/services/orchestrators/evaluation_service.py b/openrag/services/orchestrators/evaluation_service.py index 92a1c053a..3558f4da4 100644 --- a/openrag/services/orchestrators/evaluation_service.py +++ b/openrag/services/orchestrators/evaluation_service.py @@ -273,7 +273,10 @@ async def start_run(self, dataset_id: str, user_id: int | None) -> EvalRun: try: eval_user_id = await self._ensure_eval_user() token = (await self._user_service.regenerate_token(eval_user_id))["token"] - await self._partition_service.create_partition(partition, user_id=eval_user_id) + # ``__eval_*`` is rejected on the public creation path, so that a + # user cannot mint a partition the listings hide. A run owns the + # namespace and is the one caller allowed through. + await self._partition_service.create_partition(partition, user_id=eval_user_id, allow_reserved=True) await self._dispatch(run_id, partition, token, directory, cases) except Exception as exc: # The run row is the lock; leaving it active would block every diff --git a/tests/unit/services/orchestrators/test_evaluation_service.py b/tests/unit/services/orchestrators/test_evaluation_service.py index e25c6d416..fb1e62fbe 100644 --- a/tests/unit/services/orchestrators/test_evaluation_service.py +++ b/tests/unit/services/orchestrators/test_evaluation_service.py @@ -82,9 +82,12 @@ def __init__(self, create_error: Exception | None = None) -> None: async def delete_partition(self, partition): self.deleted.append(partition) - async def create_partition(self, partition, user_id=None): + async def create_partition(self, partition, user_id=None, *, allow_reserved=False): if self._create_error: raise self._create_error + # The real service rejects the ``__eval_`` prefix without this; a fake + # that quietly accepted either way would hide a run that cannot start. + assert allow_reserved, "an eval run must claim its reserved partition explicitly" self.created.append(partition)