Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 8 additions & 9 deletions plugins/nemo-evaluator/src/nemo_evaluator/jobs/evaluate.py
Original file line number Diff line number Diff line change
Expand Up @@ -23,11 +23,11 @@
unresolved_model_refs,
)
from nemo_evaluator.jobs.result_persistence import persist_evaluate_result
from nemo_evaluator.jobs.utils import run_with_isolated_async_sdk
from nemo_evaluator.metric_refs import MetricRefOrInline
from nemo_evaluator.shared.metric_bundles.bundles import unbundle_metric
from nemo_evaluator_sdk import Evaluator
from nemo_evaluator_sdk.execution.config import resolve_params
from nemo_evaluator_sdk.execution.metric_execution import run_sync
from nemo_evaluator_sdk.values import (
Agent,
AgentBase,
Expand Down Expand Up @@ -89,20 +89,19 @@ def _resolve_run_dataset(
return dataset

destination = str(ctx.storage.persistent / "dataset")
if async_sdk is not None:
return run_sync(
lambda: download_dataset(
sdk=async_sdk,
dataset=dataset,
destination=destination,
)
)
# Prefer sync when available; async path isolates httpx so later run_sync calls
# (result persistence) can reuse the injected async_sdk.
if sdk is not None:
return download_dataset_sync(
sdk=sdk,
dataset=dataset,
destination=destination,
)
if async_sdk is not None:
return run_with_isolated_async_sdk(
async_sdk,
lambda sdk: download_dataset(sdk=sdk, dataset=dataset, destination=destination),
)
raise ValueError("FilesetRef datasets require an SDK client for local evaluator job execution.")


Expand Down
11 changes: 6 additions & 5 deletions plugins/nemo-evaluator/src/nemo_evaluator/jobs/publication.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,8 +18,8 @@

from nemo_evaluator.intake.publish import PublishError, PublishReport, publish_to_intake
from nemo_evaluator.jobs.agent_spec import IntakePublicationSpec, Target, target_agent_identity
from nemo_evaluator.jobs.utils import run_with_isolated_async_sdk
from nemo_evaluator_sdk.agent_eval.results import AgentEvalResult
from nemo_evaluator_sdk.execution.metric_execution import run_sync
from nemo_platform import AsyncNeMoPlatform
from nemo_platform._exceptions import NeMoPlatformError, NotFoundError
from nemo_platform_plugin.jobs.schemas import PlatformJobStatus
Expand Down Expand Up @@ -153,15 +153,16 @@ def fail(error: str, report: PublishReport | None = None) -> PublicationOutcome:
workspace,
)
try:
report = run_sync(
lambda: _publish(
report = run_with_isolated_async_sdk(
async_sdk,
lambda sdk: _publish(
result,
platform=async_sdk,
platform=sdk,
spec=spec,
workspace=workspace,
agent_name=agent_name,
model_name=model_name,
)
),
)
except PublishError as error:
return fail(str(error), error.report)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -28,8 +28,8 @@
ModelTarget,
Target,
)
from nemo_evaluator.jobs.utils import run_with_isolated_async_sdk
from nemo_evaluator_sdk.agent_eval.results import AgentEvalResult
from nemo_evaluator_sdk.execution.metric_execution import run_sync
from nemo_evaluator_sdk.values import Agent, AgentBase, Model
from nemo_evaluator_sdk.values.multi_metric_results import BenchmarkEvaluationResult
from nemo_evaluator_sdk.values.results import EvaluationResult
Expand Down Expand Up @@ -107,15 +107,20 @@ def _row_target_fields(target: Model | Agent | None) -> tuple[str | None, str |


def _persist(entity: EntityBase, *, async_sdk: AsyncNeMoPlatform | None) -> None:
client = _entity_client(async_sdk)
if client is None:
if async_sdk is None:
logger.info("No async task SDK injected; skipping result-entity persistence (platformless local run).")
return
# Best-effort: the eval has already succeeded and the full bundle is saved, so a transient
# entity-store error must not fail the job — the record is re-derivable from the bundle. Log
# loudly and move on. (`save` is create-or-update, so a re-run of the same job id is idempotent.)
try:
run_sync(lambda: client.save(entity))

async def _save(sdk: AsyncNeMoPlatform) -> EntityBase:
client = _entity_client(sdk)
assert client is not None
return await client.save(entity)

run_with_isolated_async_sdk(async_sdk, _save)
except Exception:
logger.exception(
"Failed to persist result entity %r in workspace %r; the result bundle is still saved",
Expand Down
32 changes: 32 additions & 0 deletions plugins/nemo-evaluator/src/nemo_evaluator/jobs/utils.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,32 @@
# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0

"""Shared helpers for evaluator plugin jobs."""

from __future__ import annotations

from collections.abc import Awaitable, Callable
from typing import TypeVar

from nemo_evaluator_sdk.execution.metric_execution import run_sync
from nemo_platform import AsyncNeMoPlatform, DefaultAsyncHttpxClient

T = TypeVar("T")


def run_with_isolated_async_sdk(
async_sdk: AsyncNeMoPlatform,
fn: Callable[[AsyncNeMoPlatform], Awaitable[T]],
) -> T:
"""Run ``fn(cloned_sdk)`` via ``run_sync`` without binding ``async_sdk``'s httpx client.

Clones ``async_sdk`` onto a throwaway httpx client for the duration of ``fn`` so the injected
client's transport is not bound to this temporary event loop (and later ``run_sync`` calls can
still reuse ``async_sdk``).
"""

async def _run() -> T:
async with DefaultAsyncHttpxClient() as http_client:
return await fn(async_sdk.copy(http_client=http_client))

return run_sync(_run)
7 changes: 7 additions & 0 deletions plugins/nemo-evaluator/tests/jobs/test_publication.py
Original file line number Diff line number Diff line change
Expand Up @@ -117,13 +117,19 @@ def __init__(
self.atif_calls: list[dict[str, Any]] = []
self.eval_result_calls: list[dict[str, Any]] = []
self.trace_calls: _SessionIds = []
self.copy_calls = 0
self.evaluations = _FakeEvaluations(missing=missing_evaluation, error=preflight_error)
self.intake = SimpleNamespace(
ingest=SimpleNamespace(atif=_FakeIngest(self.atif_calls, error=ingest_error)),
evaluator_results=_FakeIngest(self.eval_result_calls),
traces=_FakeTraces(self.trace_calls),
)

def copy(self, **kwargs: Any) -> _FakeClient:
del kwargs
self.copy_calls += 1
return self


def _client(**kwargs: Any) -> AsyncNeMoPlatform:
return cast(AsyncNeMoPlatform, _FakeClient(**kwargs))
Expand Down Expand Up @@ -267,6 +273,7 @@ def test_publishes_and_reports_what_landed() -> None:
client = _FakeClient()
outcome = _publish(cast(AsyncNeMoPlatform, client))

assert client.copy_calls == 1
assert outcome.status == PlatformJobStatus.COMPLETED
assert outcome.evaluation_id == "eval-1"
assert outcome.trial_count == 1
Expand Down
29 changes: 29 additions & 0 deletions plugins/nemo-evaluator/tests/jobs/test_utils.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,29 @@
# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0

"""Unit tests for evaluator job helpers."""

from __future__ import annotations

from nemo_evaluator.jobs.utils import run_with_isolated_async_sdk
from pytest_mock import MockerFixture


def test_run_with_isolated_async_sdk_passes_cloned_client(mocker: MockerFixture) -> None:
cloned = mocker.Mock(name="cloned_sdk")
async_sdk = mocker.Mock(name="async_sdk")
async_sdk.copy.return_value = cloned
http_client = mocker.AsyncMock(name="http_client")
http_client.__aenter__.return_value = http_client
http_client.__aexit__.return_value = None
mocker.patch("nemo_evaluator.jobs.utils.DefaultAsyncHttpxClient", return_value=http_client)

seen: list[object] = []

async def fn(sdk: object) -> int:
seen.append(sdk)
return 42

assert run_with_isolated_async_sdk(async_sdk, fn) == 42
async_sdk.copy.assert_called_once_with(http_client=http_client)
assert seen == [cloned]
47 changes: 45 additions & 2 deletions plugins/nemo-evaluator/tests/test_evaluate_job.py
Original file line number Diff line number Diff line change
Expand Up @@ -1202,16 +1202,23 @@ def test_downloads_fileset_ref_dataset_and_passes_path_to_sdk_evaluator(
create=True,
)
download_dataset_sync = mocker.patch("nemo_evaluator.jobs.evaluate.download_dataset_sync", create=True)
cloned_sdk = mocker.Mock(name="cloned_async_sdk")
async_sdk = mocker.Mock(name="async_sdk")
async_sdk.copy.return_value = cloned_sdk
http_client = mocker.AsyncMock(name="http_client")
http_client.__aenter__.return_value = http_client
http_client.__aexit__.return_value = None
mocker.patch("nemo_evaluator.jobs.utils.DefaultAsyncHttpxClient", return_value=http_client)
ctx = _make_job_context(tmp_path)
async_sdk = mocker.Mock()
dataset = FilesetRef(root="default/helpsteer2#validation.jsonl")
config = {**_exact_match_spec(), "dataset": dataset}

run_result = EvaluateJob().run(config, ctx=ctx, async_sdk=async_sdk)

_assert_saved_result_artifact(run_result, ctx, result_payload)
async_sdk.copy.assert_called_once_with(http_client=http_client)
download_dataset.assert_awaited_once_with(
sdk=async_sdk,
sdk=cloned_sdk,
dataset=dataset,
destination=str(ctx.storage.persistent / "dataset"),
)
Expand Down Expand Up @@ -1259,6 +1266,42 @@ def test_downloads_fileset_ref_dataset_with_sync_sdk_and_passes_path_to_sdk_eval
assert call_kwargs["target"] is None
assert call_kwargs["prompt_template"] is None

def test_prefers_sync_sdk_for_fileset_ref_when_both_sdks_injected(
self, tmp_path: Path, mocker: MockerFixture
) -> None:
"""Avoid binding async_sdk's httpx client before result-entity persistence."""
result = _empty_evaluation_result()
result_payload = result.model_dump(mode="json")
evaluator = mocker.Mock()
evaluator.run_sync.return_value = result
mocker.patch("nemo_evaluator.jobs.evaluate.Evaluator", return_value=evaluator)
downloaded_path = tmp_path / "persistent" / "dataset" / "default" / "helpsteer2" / "validation.jsonl"
download_dataset = mocker.patch("nemo_evaluator.jobs.evaluate.download_dataset", create=True)
download_dataset_sync = mocker.patch(
"nemo_evaluator.jobs.evaluate.download_dataset_sync",
return_value=downloaded_path,
create=True,
)
persist = mocker.patch("nemo_evaluator.jobs.evaluate.persist_evaluate_result")
ctx = _make_job_context(tmp_path)
sync_sdk = mocker.Mock()
async_sdk = mocker.Mock()
dataset = FilesetRef(root="default/helpsteer2#validation.jsonl")
config = {**_exact_match_spec(), "dataset": dataset}

run_result = EvaluateJob().run(config, ctx=ctx, sdk=sync_sdk, async_sdk=async_sdk)

_assert_saved_result_artifact(run_result, ctx, result_payload)
download_dataset.assert_not_called()
download_dataset_sync.assert_called_once_with(
sdk=sync_sdk,
dataset=dataset,
destination=str(ctx.storage.persistent / "dataset"),
)
persist.assert_called_once()
assert persist.call_args.kwargs["async_sdk"] is async_sdk
assert persist.call_args.kwargs["dataset_ref"] == dataset.root


class TestEvaluateTask:
"""Coverage for the compiled container task entrypoint."""
Expand Down
17 changes: 16 additions & 1 deletion plugins/nemo-evaluator/tests/test_result_persistence.py
Original file line number Diff line number Diff line change
Expand Up @@ -40,9 +40,22 @@
from nemo_platform_plugin.job_results import LocalJobResults
from pytest_mock import MockerFixture


# An opaque stand-in for the async task SDK: every test that reaches the save path patches
# `_entity_client`, so the value is never used as a real client — only its presence matters.
_ASYNC_SDK = cast(AsyncNeMoPlatform, object())
# ``copy`` returns self so ``run_with_isolated_async_sdk`` can isolate a throwaway httpx client.
class _FakeAsyncSdk:
def __init__(self) -> None:
self.copy_calls = 0

def copy(self, **kwargs: object) -> _FakeAsyncSdk:
del kwargs
self.copy_calls += 1
return self


_ASYNC_SDK_FAKE = _FakeAsyncSdk()
_ASYNC_SDK = cast(AsyncNeMoPlatform, _ASYNC_SDK_FAKE)


def test_entity_client_adapts_async_sdk(mocker: MockerFixture) -> None:
Expand Down Expand Up @@ -165,6 +178,7 @@ def _eval_result() -> EvaluationResult:
def test_persist_agent_eval_result_builds_entity_and_saves(tmp_path: Path, mocker: MockerFixture) -> None:
client = _FakeClient()
mocker.patch.object(result_persistence, "_entity_client", return_value=client)
_ASYNC_SDK_FAKE.copy_calls = 0

persist_agent_eval_result(
_agent_result(),
Expand All @@ -174,6 +188,7 @@ def test_persist_agent_eval_result_builds_entity_and_saves(tmp_path: Path, mocke
async_sdk=_ASYNC_SDK,
)

assert _ASYNC_SDK_FAKE.copy_calls == 1
(entity,) = client.saved
assert isinstance(entity, AgentEvalResultEntity)
assert entity.name == "job-1"
Expand Down