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/__init__.py

@@ -33,6 +33,11 @@
     from nemo_evaluator_sdk.datasets import DatasetLoadError, load_dataset, load_dataset_as_dicts
     from nemo_evaluator_sdk.execution.backends.local.backend import LocalBackend
     from nemo_evaluator_sdk.execution.evaluator import Evaluator
+    from nemo_evaluator_sdk.execution.jobs import (
+        EvaluationJob,
+        LocalJob,
+        SyncEvaluationJob,
+    )
     from nemo_evaluator_sdk.execution.values import (
         EvaluationError,
         EvaluationPhase,
@@ -116,6 +121,9 @@
     "load_dataset_as_dicts": ".datasets",
     "LocalBackend": ".execution.backends.local.backend",
     "Evaluator": ".execution.evaluator",
+    "EvaluationJob": ".execution.jobs",
+    "SyncEvaluationJob": ".execution.jobs",
+    "LocalJob": ".execution.jobs",
     "EvaluationError": ".execution.values",
     "EvaluationPhase": ".execution.values",
     "BLEUMetric": ".metrics.bleu",
@@ -189,6 +197,9 @@
     "RunConfigOnlineModel",
     "EvaluationResult",
     "Evaluator",
+    "EvaluationJob",
+    "SyncEvaluationJob",
+    "LocalJob",
     "ExactMatchMetric",
     "F1Metric",
     "FieldMapping",

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.
 
@@ -140,6 +162,24 @@ def __init__(
         self.client = client
         self.default_headers = default_headers
 
+    @overload
+    async def run(
+        self,
+        *,
+        tasks: Sequence[AgentEvalTask],
+        target: AgentEvalTarget,
+        config: AgentEvalRunConfig | None = None,
+    ) -> AgentEvalResult: ...
+
+    @overload
+    async def run(
+        self,
+        *,
+        tasks: Sequence[AgentEvalTask],
+        trials: Sequence[AgentEvalTrial],
+        config: AgentEvalRunConfig | None = None,
+    ) -> AgentEvalResult: ...
+
     async def run(
         self,
         *,
@@ -150,12 +190,13 @@ async def run(
     ) -> AgentEvalResult:
         """Evaluate imported trials or generate live trials before scoring.
 
-        Exactly one of ``trials`` or ``target`` must be provided.
+        Exactly one of ``trials`` or ``target`` must be provided; the overloads above say so to the
+        type checker, and :func:`validate_run_inputs` still says it at runtime for callers that
+        assemble their arguments dynamically.
         """
         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})
@@ -199,6 +240,24 @@ async def run(
 
         return result
 
+    @overload
+    def run_sync(
+        self,
+        *,
+        tasks: Sequence[AgentEvalTask],
+        target: AgentEvalTarget,
+        config: AgentEvalRunConfig | None = None,
+    ) -> AgentEvalResult: ...
+
+    @overload
+    def run_sync(
+        self,
+        *,
+        tasks: Sequence[AgentEvalTask],
+        trials: Sequence[AgentEvalTrial],
+        config: AgentEvalRunConfig | None = None,
+    ) -> AgentEvalResult: ...
+
     def run_sync(
         self,
         *,
@@ -207,8 +266,17 @@ def run_sync(
         target: AgentEvalTarget | None = None,
         config: AgentEvalRunConfig | None = None,
     ) -> AgentEvalResult:
-        """Synchronous bridge for :meth:`run`."""
-        return run_sync(lambda: self.run(tasks=tasks, trials=trials, target=target, config=config))
+        """Synchronous bridge for :meth:`run`.
+
+        Branches on which seam was supplied because the overloads keep the two apart; the final
+        raise is what narrows, and is unreachable once one of them is set.
+        """
+        validate_run_inputs(tasks=tasks, trials=trials, target=target)
+        if trials is not None:
+            return run_sync(lambda: self.run(tasks=tasks, trials=trials, config=config))
+        if target is not None:
+            return run_sync(lambda: self.run(tasks=tasks, target=target, config=config))
+        raise ValueError("provide exactly one of trials or target")
 
     async def _score_trials(
         self,

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"}],
 )
 ```