refactor(evaluator)!: unify the backend contract on evaluate/evaluate_dataset

The SDK backend protocol carried two dataset entrypoints that differed only in
metric arity and return type: `evaluate` took one metric and returned an
`EvaluationResult`, `evaluate_benchmark` took a list and returned a
`BenchmarkEvaluationResult`. One metric is the degenerate case of several, so
collapse them into a single `evaluate_dataset(metrics=[...])` and free the
`evaluate` name for the taskset entrypoint.

`Evaluator.run`/`run_sync` lose their single-metric overloads; callers pass a
list. `BenchmarkEvaluationResult` gains the `__str__` that `EvaluationResult`
had, so `print(result)` stays readable through the collapsed path.

Introduce a job-handle contract. `EvaluationJob`/`SyncEvaluationJob` declare
`wait_until_done` and `get_result`, so a backend that runs work elsewhere hands
back a handle and the caller chooses when to wait; in-process execution returns
an already-finished `CompletedJob`. `Evaluator.submit` waits on the caller's
behalf, so the convenience API still returns a result either way.

The plugin's job resources satisfy that contract structurally, so both
`evaluate` and `evaluate_dataset` now return handles and the resource is
consistent. The agent-eval handle carries the taskset it submitted: rebuilding
the result needs the caller's live tasks, because persisted task metrics
serialize as descriptors that cannot be validated back into `Metric` objects,
and holding them on the handle means no caller has to know that. Polling and
bundle reassembly move out of the executor onto the handle, which drops the
duplicated URL, wait, and download logic from both executor flavours and
replaces two hard-coded timeout constants with the three the dataset handles
already take.

Retire plugin-local execution. `client.evaluator.run()` is gone and `submit` is
now `evaluate_dataset`, taking a metric list. That orphaned the executor's local
paths, so remove `run_local`, `evaluate`, `evaluate_benchmark` and
`evaluate_remote` from both executors along with the two metric type-guards
whose only purpose was the single-vs-sequence dispatch.

Reject two silent failures the agent-eval path allowed: a task with no
instruction (the wire schema permits null, so a job would reach the agent with
nothing to do) and run params that do not match the target kind (previously
dropped, so a caller's parallelism vanished without a word).

BREAKING CHANGE: `EvaluationBackend.evaluate` now takes a taskset and returns an
`EvaluationJob`; dataset evaluation moves to `evaluate_dataset` with a metric
list. `Evaluator.run`/`run_sync` no longer accept a bare metric.
`client.evaluator.run()` is removed, `client.evaluator.submit()` is now
`client.evaluator.evaluate_dataset()`, and `client.evaluator.evaluate()` returns
a job handle rather than a completed result.

Signed-off-by: Sandy Chapman <schapman@nvidia.com>

Author: SandyChapman

e2e/test_evaluator_plugin.py

@@ -352,8 +352,8 @@ def evaluator_sdk(sdk: NeMoPlatform, evaluator_workspace: str) -> Iterator[NeMoP
 
 @pytest.fixture(scope="module")
 def completed_offline_job(evaluator_sdk: NeMoPlatform) -> Iterator[EvaluatorJobResource]:
-    job = evaluator_sdk.evaluator.submit(
-        metric=_exact_match_metric(),
+    job = evaluator_sdk.evaluator.evaluate_dataset(
+        metrics=[_exact_match_metric()],
         dataset=_offline_rows(),
         config=RunConfig(parallelism=1),
     )
@@ -479,8 +479,8 @@ def test_fileset_fragment_and_glob_datasets(evaluator_sdk: NeMoPlatform) -> None
             "glob": (f"{workspace}/{fileset_name}#part-*.json", [1.0, 0.0, 1.0]),
         }
         for label, (reference, expected_scores) in cases.items():
-            job = evaluator_sdk.evaluator.submit(
-                metric=_exact_match_metric(),
+            job = evaluator_sdk.evaluator.evaluate_dataset(
+                metrics=[_exact_match_metric()],
                 dataset=FilesetRef(root=reference),
                 config=RunConfig(parallelism=1),
             )
@@ -500,8 +500,8 @@ def test_fileset_fragment_and_glob_datasets(evaluator_sdk: NeMoPlatform) -> None
 
 def test_run_config_limits_samples(evaluator_sdk: NeMoPlatform) -> None:
     rows = [{"expected": str(index), "output": str(index)} for index in range(8)]
-    job = evaluator_sdk.evaluator.submit(
-        metric=_exact_match_metric(),
+    job = evaluator_sdk.evaluator.evaluate_dataset(
+        metrics=[_exact_match_metric()],
         dataset=rows,
         config=RunConfig(limit_samples=3, parallelism=2),
     )
@@ -596,8 +596,8 @@ def test_tool_calling_metric_preserves_structured_references(evaluator_sdk: NeMo
             },
         },
     ]
-    job = evaluator_sdk.evaluator.submit(
-        metric=ToolCallingMetric(reference="{{item.expected_tool_calls}}"),
+    job = evaluator_sdk.evaluator.evaluate_dataset(
+        metrics=[ToolCallingMetric(reference="{{item.expected_tool_calls}}")],
         dataset=rows,
         config=RunConfig(parallelism=2),
     )
@@ -629,8 +629,8 @@ def test_online_evaluate_job_uses_mock_provider(
         format=ModelFormat.OPEN_AI,
     )
 
-    job = evaluator_sdk.evaluator.submit(
-        metric=_exact_match_metric(candidate=None),
+    job = evaluator_sdk.evaluator.evaluate_dataset(
+        metrics=[_exact_match_metric(candidate=None)],
         dataset=[{"question": "What is the capital of France?", "expected": "Paris"}],
         config=RunConfigOnlineModel(
             parallelism=1,
@@ -683,8 +683,8 @@ def test_llm_judge_metric_resolves_model_ref(
             ]
         },
     )
-    job = evaluator_sdk.evaluator.submit(
-        metric=metric,
+    job = evaluator_sdk.evaluator.evaluate_dataset(
+        metrics=[metric],
         dataset=[{"answer": "Paris"}],
         config=RunConfig(parallelism=1),
     )
@@ -701,8 +701,8 @@ def _assert_runtime_input_failure(
     metric: StringCheckMetric | ExactMatchMetric,
     dataset: list[dict[str, object]] | FilesetRef,
 ) -> None:
-    job = evaluator_sdk.evaluator.submit(
-        metric=metric,
+    job = evaluator_sdk.evaluator.evaluate_dataset(
+        metrics=[metric],
         dataset=dataset,
         config=RunConfig(parallelism=1),
     )

packages/nemo_evaluator_sdk/examples/examples.py

@@ -360,7 +360,7 @@ async def run_offline_local_exact_match_example() -> None:
     print("Running offline exact match...")
 
     exact_match_result = await evaluator.run(
-        metrics=exact_match,
+        metrics=[exact_match],
         dataset=OFFLINE_EXACT_MATCH_DATASET,
         config=RunConfig(parallelism=4),
     )
@@ -382,7 +382,7 @@ async def run_online_local_exact_match_example() -> None:
     print("Running local online exact match...")
 
     exact_match_result = await evaluator.run(
-        metrics=exact_match,
+        metrics=[exact_match],
         target=model,
         dataset=ONLINE_EXACT_MATCH_DATASET,
         prompt_template=ONLINE_CHAT_PROMPT_TEMPLATE,
@@ -523,7 +523,7 @@ async def run_local_metric_with_template_failure_example() -> None:
     print("\nRunning local metric evaluation with an invalid metric template...")
     try:
         await evaluator.run(
-            metrics=invalid_metric,
+            metrics=[invalid_metric],
             dataset=dataset,
             config=RunConfig(parallelism=1),
         )
@@ -553,7 +553,7 @@ async def run_offline_local_llm_judge_example() -> None:
     print("\nRunning local LLM judge evaluation...")
 
     llm_judge_result = await evaluator.run(
-        metrics=llm_judge_metric,
+        metrics=[llm_judge_metric],
         dataset=OFFLINE_JUDGE_DATASET,
         config=RunConfig(parallelism=2),
     )
@@ -575,7 +575,7 @@ async def run_online_local_llm_judge_example() -> None:
     print("\nRunning local online LLM judge evaluation...")
 
     llm_judge_result = await evaluator.run(
-        metrics=llm_judge_metric,
+        metrics=[llm_judge_metric],
         target=model_with_custom_headers,
         dataset=ONLINE_JUDGE_DATASET,
         prompt_template=ONLINE_CHAT_PROMPT_TEMPLATE,
@@ -593,7 +593,7 @@ def run_sync_example() -> None:
 
     evaluator = Evaluator()
     result = evaluator.run_sync(
-        metrics=ExactMatchMetric(reference="{{item.reference}}", candidate="{{item.actual}}"),
+        metrics=[ExactMatchMetric(reference="{{item.reference}}", candidate="{{item.actual}}")],
         dataset=OFFLINE_EXACT_MATCH_DATASET[:1],  # Only run the first sample
         config=RunConfig(parallelism=1),
     )

packages/nemo_evaluator_sdk/examples/high_level_evaluate_walkthrough.ipynb

@@ -349,7 +349,7 @@
     "]\n",
     "\n",
     "metric = ExactMatchMetric(reference=\"{{ expected }}\", candidate=\"{{ model_output }}\")\n",
-    "inline_result = evaluator.run_sync(metrics=metric, dataset=inline_rows)\n",
+    "inline_result = evaluator.run_sync(metrics=[metric], dataset=inline_rows)\n",
     "\n",
     "runs[\"Inline list[dict]\"] = inline_result\n",
     "\n",
@@ -383,7 +383,7 @@
     ")\n",
     "\n",
     "metric = ExactMatchMetric(reference=\"{{ expected }}\", candidate=\"{{ model_output }}\")\n",
-    "dataset_rows_result = evaluator.run_sync(metrics=metric, dataset=dataset_rows)\n",
+    "dataset_rows_result = evaluator.run_sync(metrics=[metric], dataset=dataset_rows)\n",
     "\n",
     "runs[\"DatasetRows\"] = dataset_rows_result\n",
     "\n",
@@ -410,7 +410,7 @@
     "arrow_table = pa.Table.from_pylist(base_rows)\n",
     "\n",
     "metric = ExactMatchMetric(reference=\"{{ expected }}\", candidate=\"{{ prediction }}\")\n",
-    "arrow_result = evaluator.run_sync(metrics=metric, dataset=arrow_table)\n",
+    "arrow_result = evaluator.run_sync(metrics=[metric], dataset=arrow_table)\n",
     "\n",
     "runs[\"pyarrow.Table\"] = arrow_result\n",
     "\n",
@@ -435,7 +435,7 @@
    "outputs": [],
    "source": [
     "metric = ExactMatchMetric(reference=\"{{ expected }}\", candidate=\"{{ prediction }}\")\n",
-    "file_result = evaluator.run_sync(metrics=metric, dataset=file_path)\n",
+    "file_result = evaluator.run_sync(metrics=[metric], dataset=file_path)\n",
     "\n",
     "runs[\"File path\"] = file_result\n",
     "\n",
@@ -460,7 +460,7 @@
    "outputs": [],
    "source": [
     "metric = ExactMatchMetric(reference=\"{{ expected }}\", candidate=\"{{ prediction }}\")\n",
-    "directory_result = evaluator.run_sync(metrics=metric, dataset=dataset_dir, pattern=\"part-*.jsonl\")\n",
+    "directory_result = evaluator.run_sync(metrics=[metric], dataset=dataset_dir, pattern=\"part-*.jsonl\")\n",
     "\n",
     "runs[\"Directory + pattern\"] = directory_result\n",
     "\n",

packages/nemo_evaluator_sdk/src/nemo_evaluator_sdk/agent_eval/evaluator.py

@@ -83,6 +83,28 @@
 )
 
 
+def validate_run_inputs(
+    *,
+    tasks: Sequence[AgentEvalTask],
+    trials: Sequence[AgentEvalTrial] | None,
+    target: AgentEvalTarget | None,
+) -> None:
+    """Check the seams a run needs before any work starts.
+
+    Shared with the local backend so that a malformed taskset is rejected when the evaluation is
+    requested, not when it is awaited — the same moment the remote path rejects it, in
+    ``build_spec`` before the job is created.
+
+    Raises:
+        ValueError: If there are no tasks, or if neither or both of ``trials`` and ``target``
+            were supplied.
+    """
+    if not tasks:
+        raise ValueError("at least one task is required")
+    if (trials is None) == (target is None):
+        raise ValueError("provide exactly one of trials or target")
+
+
 class AgentEvaluator:
     """Run stored-trial or live-target agent evaluations.
 
@@ -154,8 +176,7 @@ async def run(
         """
         resolved_config = config or AgentEvalRunConfig()
         task_list = list(tasks)
-        if not task_list:
-            raise ValueError("at least one task is required")
+        validate_run_inputs(tasks=task_list, trials=trials, target=target)
 
         run_id = resolved_config.run_id or _new_run_id()
         runtime_config = resolved_config.model_copy(update={"run_id": run_id})

packages/nemo_evaluator_sdk/src/nemo_evaluator_sdk/execution/README.md

@@ -22,7 +22,7 @@ The execution package exposes a single public entrypoint:
 # Local SDK execution
 evaluator = Evaluator()
 result = await evaluator.run(
-    metrics=ExactMatchMetric(reference="{{item.reference}}"),
+    metrics=[ExactMatchMetric(reference="{{item.reference}}")],
     dataset=[{"reference": "Paris", "output_text": "Paris"}],
 )
 ```

packages/nemo_evaluator_sdk/src/nemo_evaluator_sdk/execution/backends/base.py

@@ -7,8 +7,12 @@
 
 from collections.abc import Sequence
 from pathlib import Path
-from typing import Any, Protocol
+from typing import Any, Protocol, runtime_checkable
 
+from nemo_evaluator_sdk.agent_eval.results import AgentEvalResult
+from nemo_evaluator_sdk.agent_eval.tasks import AgentEvalRunConfig, AgentEvalTask
+from nemo_evaluator_sdk.agent_eval.trials import AgentEvalTarget, AgentEvalTrial
+from nemo_evaluator_sdk.execution.jobs import EvaluationJob, SyncEvaluationJob
 from nemo_evaluator_sdk.inference import PostprocessResponse, PreprocessRequest
 from nemo_evaluator_sdk.metrics.protocol import Metric
 from nemo_evaluator_sdk.values import (
@@ -21,44 +25,50 @@
     RunConfigOnlineModel,
 )
 from nemo_evaluator_sdk.values.multi_metric_results import BenchmarkEvaluationResult
-from nemo_evaluator_sdk.values.results import AggregateFieldName, EvaluationResult
+from nemo_evaluator_sdk.values.results import AggregateFieldName
 
 BackendParams = RunConfig | RunConfigOnline | RunConfigOnlineModel
 
 
+@runtime_checkable
 class EvaluationBackend(Protocol):
     async def evaluate(
         self,
         *,
-        metric: Metric,
-        dataset: DatasetInput | str | Path,
-        params: BackendParams,
-        target: Model | Agent | None = None,
-        field_mapping: FieldMapping | None = None,
-        prompt_template: str | dict[str, Any] | None = None,
-        aggregate_fields: tuple[AggregateFieldName, ...] | None = None,
-        preprocess_hooks: tuple[PreprocessRequest, ...] | None = None,
-        postprocess_hooks: tuple[PostprocessResponse, ...] | None = None,
-    ) -> EvaluationResult:
-        """Evaluate one metric directly and return the completed result.
+        taskset: Sequence[AgentEvalTask],
+        target: AgentEvalTarget | None = None,
+        trials: Sequence[AgentEvalTrial] | None = None,
+        config: AgentEvalRunConfig | None = None,
+    ) -> EvaluationJob[AgentEvalResult]:
+        """Start evaluating a taskset — tasks that each carry their own metrics — and return its job.
+
+        The entrypoint ``evaluate_dataset`` is intended to fold into: a dataset with one shared
+        metric list is a taskset whose metrics have been hoisted. That is not implemented yet — this
+        method cannot express a dataset today — so ``evaluate_dataset`` remains the way to run one,
+        and is not deprecated.
+
+        Returns a job rather than a result so the caller chooses when to wait and can reach the
+        run's identity, partial state, and artifacts meanwhile;
+        :meth:`~nemo_evaluator_sdk.execution.evaluator.Evaluator.submit` waits on the caller's
+        behalf. A backend that runs in-process returns a
+        :class:`~nemo_evaluator_sdk.execution.jobs.LocalJob`, which likewise defers the work to the
+        wait, so the call means the same thing wherever it executed. Implementations may accept extra keyword arguments with defaults (a
+        workspace, a metric packager) without breaking conformance.
 
         Args:
-            metric: Metric to prepare and execute.
-            dataset: Inline dataset rows, a dataset file, or a dataset directory/glob path.
-            params: Validated run configuration for the selected target mode.
-            target: Optional model or agent used to generate candidate responses before scoring.
-            field_mapping: Optional mapping from canonical evaluator fields to dataset columns.
-            prompt_template: Optional prompt template for online target generation.
-            aggregate_fields: Optional aggregate score fields to keep in the returned result.
-            preprocess_hooks: Optional request preprocess hooks for online execution.
-            postprocess_hooks: Optional response postprocess hooks for online execution.
+            taskset: Tasks to evaluate, each carrying its own metrics.
+            target: What generates trials — a model, agent, or runner. Mutually exclusive
+                with ``trials``.
+            trials: Precomputed trials to score instead of generating them. Mutually exclusive
+                with ``target``.
+            config: Run-level execution settings.
 
         Returns:
-            The completed single-metric evaluation result.
+            The job, awaited through its own methods.
         """
         ...
 
-    async def evaluate_benchmark(
+    async def evaluate_dataset(
         self,
         *,
         metrics: Sequence[Metric],
@@ -71,7 +81,7 @@ async def evaluate_benchmark(
         preprocess_hooks: tuple[PreprocessRequest, ...] | None = None,
         postprocess_hooks: tuple[PostprocessResponse, ...] | None = None,
     ) -> BenchmarkEvaluationResult:
-        """Evaluate multiple metrics directly and return the completed result.
+        """Evaluate multiple metrics over a dataset and return the completed result.
 
         Args:
             metrics: Metrics to prepare and execute together.
@@ -90,39 +100,41 @@ async def evaluate_benchmark(
         ...
 
 
+@runtime_checkable
 class SyncEvaluationBackend(Protocol):
     def evaluate(
         self,
         *,
-        metric: Metric,
-        dataset: DatasetInput | str | Path,
-        params: BackendParams,
-        target: Model | Agent | None = None,
-        field_mapping: FieldMapping | None = None,
-        prompt_template: str | dict[str, Any] | None = None,
-        aggregate_fields: tuple[AggregateFieldName, ...] | None = None,
-        preprocess_hooks: tuple[PreprocessRequest, ...] | None = None,
-        postprocess_hooks: tuple[PostprocessResponse, ...] | None = None,
-    ) -> EvaluationResult:
-        """Evaluate one metric directly and return the completed result.
+        taskset: Sequence[AgentEvalTask],
+        target: AgentEvalTarget | None = None,
+        trials: Sequence[AgentEvalTrial] | None = None,
+        config: AgentEvalRunConfig | None = None,
+    ) -> SyncEvaluationJob[AgentEvalResult]:
+        """Start evaluating a taskset — tasks that each carry their own metrics — and return its job.
+
+        The sync counterpart of :meth:`EvaluationBackend.evaluate`.
+
+        The entrypoint ``evaluate_dataset`` is intended to fold into: a dataset with one shared
+        metric list is a taskset whose metrics have been hoisted. That is not implemented yet — this
+        method cannot express a dataset today — so ``evaluate_dataset`` remains the way to run one,
+        and is not deprecated.
+
+        Returns a job rather than a result; see :meth:`EvaluationBackend.evaluate`.
 
         Args:
-            metric: Metric to prepare and execute.
-            dataset: Inline dataset rows, a dataset file, or a dataset directory/glob path.
-            params: Validated run configuration for the selected target mode.
-            target: Optional model or agent used to generate candidate responses before scoring.
-            field_mapping: Optional mapping from canonical evaluator fields to dataset columns.
-            prompt_template: Optional prompt template for online target generation.
-            aggregate_fields: Optional aggregate score fields to keep in the returned result.
-            preprocess_hooks: Optional request preprocess hooks for online execution.
-            postprocess_hooks: Optional response postprocess hooks for online execution.
+            taskset: Tasks to evaluate, each carrying its own metrics.
+            target: What generates trials — a model, agent, or runner. Mutually exclusive
+                with ``trials``.
+            trials: Precomputed trials to score instead of generating them. Mutually exclusive
+                with ``target``.
+            config: Run-level execution settings.
 
         Returns:
-            The completed single-metric evaluation result.
+            The job, awaited through its own methods.
         """
         ...
 
-    def evaluate_benchmark(
+    def evaluate_dataset(
         self,
         *,
         metrics: Sequence[Metric],
@@ -135,7 +147,7 @@ def evaluate_benchmark(
         preprocess_hooks: tuple[PreprocessRequest, ...] | None = None,
         postprocess_hooks: tuple[PostprocessResponse, ...] | None = None,
     ) -> BenchmarkEvaluationResult:
-        """Evaluate multiple metrics directly and return the completed result.
+        """Evaluate multiple metrics over a dataset and return the completed result.
 
         Args:
             metrics: Metrics to prepare and execute together.