feat(evaluation): the EvalRunner Ray actor and its dispatcher

`EvalRunner` executes one run: upload and time each corpus file, shell out
to promptfoo twice, fold the outputs into metrics, persist, drop the
throwaway partition. `RayEvaluationRunner` binds it to the
`EvaluationRunner` port and keeps every Ray concern — actor lookup,
`.remote()`, timeouts, cancellation — on the worker side of the boundary.

The runner drives OpenRAG through its own HTTP API rather than in-process
calls, for two reasons: it is the path a real user's documents take, so the
indexing timings mean something, and it is the same surface promptfoo
talks to, so an eval can never pass against a code path the API does not
expose.

Notes:

- The actor handle is resolved on first use, not in `__init__`. `EvalRunner`
  is detached, so merely building the adapter must not be what spawns it —
  listing datasets should not start a worker process.
- `run()` never raises. The caller dispatched it fire-and-forget and has
  nobody to catch for, so every outcome is written to the run row.
- One bad corpus file is recorded as a failed sample rather than voiding
  the run; every file failing is an error worth surfacing.
- promptfoo exits non-zero when assertions fail, which is a result, not an
  error — the presence of the output file is what decides. Both streams are
  captured, since promptfoo reports config errors on stdout.
- Each run gets its own `PROMPTFOO_CONFIG_DIR`. promptfoo keeps a SQLite
  history under it, defaulting to `$HOME/.promptfoo`, which is not
  guaranteed writable in a container.
- `max_concurrency=4` so `cancel()` and `is_busy()` still land while
  `run()` holds a slot.

Author: EnjoyBacon7

openrag/di/container.py

@@ -60,6 +60,7 @@
     from core.vector_stores import VectorStore
     from services.orchestrators.auth_service import AuthService
     from services.orchestrators.conversion_service import ConversionService
+    from services.orchestrators.evaluation_service import EvaluationService
     from services.orchestrators.indexing_service import IndexingService
     from services.orchestrators.job_service import JobService
     from services.orchestrators.mcp_service import MCPService
@@ -117,6 +118,7 @@ def __init__(self, settings: Settings | None = None) -> None:
         self._partition_service: PartitionService | None = None
         self._model_endpoint_service: ModelEndpointService | None = None
         self._preset_service: PresetService | None = None
+        self._evaluation_service: EvaluationService | None = None
         self._workspace_service: WorkspaceService | None = None
         self._retrieval_service: RetrievalService | None = None
         self._query_service: QueryService | None = None
@@ -464,6 +466,25 @@ def preset_service(self) -> PresetService:
             )
         return self._preset_service
 
+    @property
+    def evaluation_service(self) -> EvaluationService:
+        """EvaluationService — dataset storage and run dispatch."""
+        if self._evaluation_service is None:
+            from services.orchestrators.evaluation_service import EvaluationService
+            from services.workers.eval_dispatcher import from_ray_namespace
+
+            self._evaluation_service = EvaluationService(
+                repo=self.evaluation_repo,
+                # The adapter resolves its detached actor on first use, so
+                # building the service here does not spawn a worker.
+                runner=from_ray_namespace(),
+                user_service=self.user_service,
+                user_repo=self.user_repo,
+                partition_service=self.partition_service,
+                config=self._require_settings(),
+            )
+        return self._evaluation_service
+
     @property
     def workspace_service(self) -> WorkspaceService:
         """WorkspaceService — lazily built, cached for the container's lifetime."""

openrag/di/providers.py

@@ -149,6 +149,11 @@ def get_preset_service(request: Request = None) -> Any:
     return _get_optional_service(_require_initialized(request), "preset_service")
 
 
+def get_evaluation_service(request: Request = None) -> Any:
+    """Resolve the evaluation orchestrator from the active container."""
+    return _get_optional_service(_require_initialized(request), "evaluation_service")
+
+
 def get_config(request: Request = None):
     """Resolve application configuration from the active container."""
     return _require_initialized(request).config
@@ -159,6 +164,7 @@ def get_config(request: Request = None):
     "get_config",
     "get_container",
     "get_conversion_service",
+    "get_evaluation_service",
     "get_indexing_service",
     "get_job_service",
     "get_mcp_service",

openrag/services/workers/eval_dispatcher.py

@@ -0,0 +1,84 @@
+"""Ray adapter for :class:`~core.evaluation.runner.EvaluationRunner`.
+
+Binds the port to the ``EvalRunner`` actor and keeps every Ray concern —
+actor lookup, ``.remote()`` calls, timeout and cancellation handling — on this
+side of the boundary, so ``EvaluationService`` never imports Ray.
+
+The actor handle is resolved on first use rather than in ``__init__``:
+``EvalRunner`` is a *detached* actor, so merely building this adapter must not
+be what spawns it. Listing datasets should not start a worker process.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any
+
+from core.evaluation.runner import EvaluationRunner
+from services.workers.ray_utils import call_ray_actor_with_timeout
+
+if TYPE_CHECKING:
+    from collections.abc import Mapping, Sequence
+
+#: Bound on the calls that are awaited (liveness probe, cancellation).
+#: ``dispatch`` is fire-and-forget and so has nothing to time out.
+DEFAULT_TIMEOUT = 60.0
+
+
+class RayEvaluationRunner(EvaluationRunner):
+    """``EvaluationRunner`` backed by the ``EvalRunner`` Ray actor."""
+
+    def __init__(self, namespace: str = "openrag", timeout: float = DEFAULT_TIMEOUT) -> None:
+        self._namespace = namespace
+        self._timeout = timeout
+        self._actor: Any = None
+
+    def _handle(self) -> Any:
+        """Get-or-create the detached actor, memoised for the process."""
+        if self._actor is None:
+            from services.workers.eval_runner import build_eval_runner
+
+            self._actor = build_eval_runner(namespace=self._namespace)
+        return self._actor
+
+    async def is_busy(self) -> bool:
+        return await call_ray_actor_with_timeout(
+            future=self._handle().is_busy.remote(),
+            timeout=self._timeout,
+            task_description="reaching the evaluation runner",
+        )
+
+    async def dispatch(
+        self,
+        *,
+        run_id: str,
+        partition: str,
+        token: str,
+        api_base_url: str,
+        corpus_dir: str,
+        cases: Sequence[Mapping[str, Any]],
+    ) -> None:
+        # Deliberately not awaited: the worker owns the run from here and
+        # records its own outcome, so the ObjectRef is dropped.
+        self._handle().run.remote(
+            run_id=run_id,
+            partition=partition,
+            token=token,
+            api_base_url=api_base_url,
+            corpus_dir=corpus_dir,
+            cases=[dict(case) for case in cases],
+        )
+
+    async def cancel(self, run_id: str) -> bool:
+        return await call_ray_actor_with_timeout(
+            future=self._handle().cancel.remote(run_id),
+            timeout=self._timeout,
+            task_description=f"cancelling evaluation run {run_id}",
+        )
+
+
+def from_ray_namespace(namespace: str = "openrag", timeout: float = DEFAULT_TIMEOUT) -> RayEvaluationRunner:
+    """Build the adapter bound to the detached ``EvalRunner`` actor."""
+    return RayEvaluationRunner(namespace=namespace, timeout=timeout)
+
+
+__all__ = ["DEFAULT_TIMEOUT", "RayEvaluationRunner", "from_ray_namespace"]