feat(evaluation): run start, dispatch and cancellation

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.

Author: EnjoyBacon7

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 ``<data_dir>/eval/<dataset_id>/``, 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,16 +61,35 @@
 _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."""
 
     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",
+]