refactor(evaluator)!: drop the plugin's local execution path

`SKILL.md` already told callers not to build on `client.evaluator.run()` because
it was being retired. This removes it, along with the executor methods that
existed only to serve it: `run_local`, `evaluate_remote`, `evaluate`, and
`evaluate_benchmark` on both the sync and async executors.

Nothing in production called it. The plugin's remote path is `submit`, and
callers wanting local execution use the standalone `nemo_evaluator_sdk.Evaluator`
directly, which is what the skill already recommends.

Removing it orphans a supporting cast, so that goes too: `sdk/fs_utils.py`
entirely, since `local_artifact_path` was only reachable from
`local_result_path` and `EvaluatorLocalRunResult` only from tests of the removed
methods; and `filter_evaluation_result`/`filter_benchmark_result` from
`sdk/utils.py`, which keeps `filter_aggregate_scores` for `job_resources`.

Two `create` tests patched `_executor.asyncio.to_thread` to prove creation never
bridges through a thread. With local execution gone the module has no asyncio
import at all, so they now assert its absence — the property holds by
construction rather than by one observed call.

The CLI `run` verb is unaffected: it comes from the job framework's
`NemoJobScheduler.run_local`, which is a different path from the executor method
of the same name.

BREAKING CHANGE: `client.evaluator.run()` is removed from both the sync and
async plugin resources. Use `submit` for platform evaluation, or
`nemo_evaluator_sdk.Evaluator` for local execution.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Signed-off-by: Sandy Chapman <schapman@nvidia.com>

Author: SandyChapman

plugins/nemo-evaluator/src/nemo_evaluator/sdk/_executor.py

@@ -5,7 +5,6 @@
 
 from __future__ import annotations
 
-import asyncio
 from collections.abc import Sequence
 from typing import Any, TypeAlias, cast
 
@@ -15,21 +14,18 @@
 from nemo_evaluator.jobs.evaluate import EvaluateInputSpec, EvaluateJob, EvaluateSpec, TargetSpec
 from nemo_evaluator.resolvers import PlatformModelResolver
 from nemo_evaluator.sdk import http_utils
-from nemo_evaluator.sdk.fs_utils import EvaluatorLocalRunResult, local_result_path
 from nemo_evaluator.sdk.job_resources import (
     AsyncEvaluatorJobResource,
     EvaluatorJob,
     EvaluatorJobResource,
 )
 from nemo_evaluator.sdk.types import PluginDatasetInput
-from nemo_evaluator.sdk.utils import filter_benchmark_result, filter_evaluation_result
 from nemo_evaluator.shared.metric_bundles.bundles import (
     MetricBundle,
     MetricBundlePackager,
     MetricBundlePackagerPolicyError,
     bundle_metric,
 )
-from nemo_evaluator.shared.metric_bundles.defaults import resolve_default_metric_bundle_packager
 from nemo_evaluator_sdk.datasets.loader import prepare_dataset_rows
 from nemo_evaluator_sdk.execution.config import resolve_params
 from nemo_evaluator_sdk.execution.metric_execution import run_sync
@@ -44,10 +40,7 @@
     RunConfigOnline,
     RunConfigOnlineModel,
 )
-from nemo_evaluator_sdk.values.multi_metric_results import BenchmarkEvaluationResult
-from nemo_evaluator_sdk.values.results import AggregateFieldName, EvaluationResult
 from nemo_platform import AsyncNeMoPlatform, NeMoPlatform
-from nemo_platform_plugin.scheduler import NemoJobScheduler
 
 _DEFAULT_POLL_INTERVAL_SECONDS = 10.0
 _DEFAULT_JOB_TIMEOUT_SECONDS = 3600.0
@@ -247,90 +240,6 @@ def create(
             )
         return job_resource
 
-    def run_local(self, *, spec: EvaluateRequestSpec, workspace: str | None = None) -> EvaluatorLocalRunResult:
-        """Run an evaluator plugin job locally with a sync platform client."""
-        resolved_workspace = http_utils.resolve_workspace(self._platform, workspace)
-        canonical_spec = _resolve_sync_local_spec(
-            spec,
-            platform=self._platform,
-            workspace=resolved_workspace,
-        )
-        payload = NemoJobScheduler().run_local(
-            EvaluateJob,
-            canonical_spec.model_dump(mode="json"),
-            workspace=resolved_workspace,
-            sdk=self._platform,
-        )
-
-        return EvaluatorLocalRunResult.model_validate(payload)
-
-    def evaluate_remote(
-        self,
-        *,
-        metric: Metric,
-        dataset: PluginDatasetInput,
-        params: RunConfig | RunConfigOnline | RunConfigOnlineModel,
-        target: Model | Agent | None = None,
-        field_mapping: FieldMapping | None = None,
-        prompt_template: str | dict[str, Any] | None = None,
-        aggregate_fields: tuple[AggregateFieldName, ...] | None = None,
-        metric_bundle_packager: MetricBundlePackager | None = None,
-    ) -> EvaluationResult:
-        """Submit, poll, and download a remote evaluator plugin metric job."""
-        normalized_params = resolve_params(params, target)
-        spec = _build_evaluate_spec(
-            metrics=metric,
-            dataset=dataset,
-            params=normalized_params,
-            target=target,
-            field_mapping=field_mapping,
-            prompt_template=prompt_template,
-            metric_bundle_packager=metric_bundle_packager,
-        )
-
-        job = self.create(
-            spec=spec, workspace=http_utils.resolve_workspace(self._platform, self._workspace, strict=True)
-        )
-        job.wait_until_done(
-            poll_interval_seconds=self._poll_interval_seconds,
-            job_timeout_seconds=self._job_timeout_seconds,
-            pending_timeout_seconds=self._pending_timeout_seconds,
-        )
-
-        return job.get_result(aggregate_fields=aggregate_fields)
-
-    def evaluate(
-        self,
-        *,
-        metric: Metric,
-        dataset: PluginDatasetInput,
-        params: RunConfig | RunConfigOnline | RunConfigOnlineModel | None = None,
-        target: Model | Agent | None = None,
-        field_mapping: FieldMapping | None = None,
-        prompt_template: str | dict[str, Any] | None = None,
-        aggregate_fields: tuple[AggregateFieldName, ...] | None = None,
-    ) -> EvaluationResult:
-        """Evaluate one metric through local plugin job execution."""
-        normalized_params = resolve_params(params, target)
-        spec = _build_evaluate_spec(
-            metrics=metric,
-            dataset=dataset,
-            params=normalized_params,
-            target=target,
-            field_mapping=field_mapping,
-            prompt_template=prompt_template,
-            metric_bundle_packager=resolve_default_metric_bundle_packager(
-                metric, None, allow_cloudpickle_fallback=True, action="Running"
-            ),
-        )
-        payload = self.run_local(
-            spec=spec,
-            workspace=http_utils.resolve_workspace(self._platform, self._workspace, strict=True),
-        )
-        result_path = local_result_path(payload)
-        result = EvaluationResult.model_validate_json(result_path.read_text(encoding="utf-8"))
-        return filter_evaluation_result(result, aggregate_fields)
-
     def submit(
         self,
         *,
@@ -361,38 +270,6 @@ def submit(
 
         return job
 
-    def evaluate_benchmark(
-        self,
-        *,
-        metrics: Sequence[Metric],
-        dataset: PluginDatasetInput,
-        params: RunConfig | RunConfigOnline | RunConfigOnlineModel,
-        target: Model | Agent | None = None,
-        field_mapping: FieldMapping | None = None,
-        prompt_template: str | dict[str, Any] | None = None,
-        aggregate_fields: tuple[AggregateFieldName, ...] | None = None,
-    ) -> BenchmarkEvaluationResult:
-        """Evaluate multiple metrics through local plugin job execution."""
-        normalized_params = resolve_params(params, target)
-        spec = _build_evaluate_spec(
-            metrics=metrics,
-            dataset=dataset,
-            params=normalized_params,
-            target=target,
-            field_mapping=field_mapping,
-            prompt_template=prompt_template,
-            metric_bundle_packager=resolve_default_metric_bundle_packager(
-                metrics, None, allow_cloudpickle_fallback=True, action="Running"
-            ),
-        )
-        payload = self.run_local(
-            spec=spec,
-            workspace=http_utils.resolve_workspace(self._platform, self._workspace, strict=True),
-        )
-        result_path = local_result_path(payload)
-        result = BenchmarkEvaluationResult.model_validate_json(result_path.read_text(encoding="utf-8"))
-        return filter_benchmark_result(result, aggregate_fields)
-
 
 class _AsyncEvaluatorPluginExecutor:
     """Async evaluator plugin executor used by the async SDK resource."""
@@ -449,26 +326,6 @@ async def create(
             )
         return job_resource
 
-    async def run_local(self, *, spec: EvaluateRequestSpec, workspace: str | None = None) -> EvaluatorLocalRunResult:
-        """Run an evaluator plugin job locally without blocking the event loop."""
-        resolved_workspace = http_utils.resolve_workspace(self._platform, workspace)
-        canonical_spec = await _resolve_async_local_spec(
-            spec,
-            platform=self._platform,
-            workspace=resolved_workspace,
-        )
-        scheduler = NemoJobScheduler()
-        # Leverages programmatic dispatch as described in
-        # packages/nemo_platform_plugin/src/nemo_platform_plugin/docs/ARCHITECTURE.md#job-entry-point-keys
-        payload = await asyncio.to_thread(
-            scheduler.run_local,
-            EvaluateJob,
-            canonical_spec.model_dump(mode="json"),
-            workspace=resolved_workspace,
-            async_sdk=self._platform,
-        )
-        return EvaluatorLocalRunResult.model_validate(payload)
-
     async def submit(
         self,
         *,
@@ -499,107 +356,6 @@ async def submit(
 
         return job
 
-    async def evaluate_remote(
-        self,
-        *,
-        metric: Metric,
-        dataset: PluginDatasetInput,
-        params: RunConfig | RunConfigOnline | RunConfigOnlineModel,
-        target: Model | Agent | None = None,
-        field_mapping: FieldMapping | None = None,
-        prompt_template: str | dict[str, Any] | None = None,
-        aggregate_fields: tuple[AggregateFieldName, ...] | None = None,
-        metric_bundle_packager: MetricBundlePackager | None = None,
-    ) -> EvaluationResult:
-        """Submit, poll, and download a remote evaluator plugin metric job."""
-        normalized_params = resolve_params(params, target)
-        spec = _build_evaluate_spec(
-            metrics=metric,
-            dataset=dataset,
-            params=normalized_params,
-            target=target,
-            field_mapping=field_mapping,
-            prompt_template=prompt_template,
-            metric_bundle_packager=metric_bundle_packager,
-        )
-
-        job = await self.create(
-            spec=spec, workspace=http_utils.resolve_workspace(self._platform, self._workspace, strict=True)
-        )
-        await job.wait_until_done(
-            poll_interval_seconds=self._poll_interval_seconds,
-            job_timeout_seconds=self._job_timeout_seconds,
-            pending_timeout_seconds=self._pending_timeout_seconds,
-        )
-
-        return await job.get_result(aggregate_fields=aggregate_fields)
-
-    async def evaluate(
-        self,
-        *,
-        metric: Metric,
-        dataset: PluginDatasetInput,
-        params: RunConfig | RunConfigOnline | RunConfigOnlineModel | None = None,
-        target: Model | Agent | None = None,
-        field_mapping: FieldMapping | None = None,
-        prompt_template: str | dict[str, Any] | None = None,
-        aggregate_fields: tuple[AggregateFieldName, ...] | None = None,
-    ) -> EvaluationResult:
-        """Evaluate one metric through local plugin job execution."""
-        normalized_params = resolve_params(params, target)
-        spec = _build_evaluate_spec(
-            metrics=metric,
-            dataset=dataset,
-            params=normalized_params,
-            target=target,
-            field_mapping=field_mapping,
-            prompt_template=prompt_template,
-            metric_bundle_packager=resolve_default_metric_bundle_packager(
-                metric, None, allow_cloudpickle_fallback=True, action="Running"
-            ),
-        )
-        payload = await self.run_local(
-            spec=spec,
-            workspace=http_utils.resolve_workspace(self._platform, self._workspace, strict=True),
-        )
-        result_path = local_result_path(payload)
-        result_text = await asyncio.to_thread(result_path.read_text, encoding="utf-8")
-        result = EvaluationResult.model_validate_json(result_text)
-        return filter_evaluation_result(result, aggregate_fields)
-
-    async def evaluate_benchmark(
-        self,
-        *,
-        metrics: Sequence[Metric],
-        dataset: PluginDatasetInput,
-        params: RunConfig | RunConfigOnline | RunConfigOnlineModel,
-        target: Model | Agent | None = None,
-        field_mapping: FieldMapping | None = None,
-        prompt_template: str | dict[str, Any] | None = None,
-        aggregate_fields: tuple[AggregateFieldName, ...] | None = None,
-    ) -> BenchmarkEvaluationResult:
-        """Evaluate multiple metrics through local plugin job execution."""
-        normalized_params = resolve_params(params, target)
-        spec = _build_evaluate_spec(
-            metrics=metrics,
-            dataset=dataset,
-            params=normalized_params,
-            target=target,
-            field_mapping=field_mapping,
-            prompt_template=prompt_template,
-            metric_bundle_packager=resolve_default_metric_bundle_packager(
-                metrics, None, allow_cloudpickle_fallback=True, action="Running"
-            ),
-        )
-        payload = await self.run_local(
-            spec=spec,
-            workspace=http_utils.resolve_workspace(self._platform, self._workspace, strict=True),
-        )
-        result_path = local_result_path(payload)
-        result_text = await asyncio.to_thread(result_path.read_text, encoding="utf-8")
-        result = BenchmarkEvaluationResult.model_validate_json(result_text)
-        return filter_benchmark_result(result, aggregate_fields)
-
 
 def bundle_metrics_for_spec(
     metrics: Metric | Sequence[Metric], *, metric_bundle_packager: MetricBundlePackager