diff --git a/plugins/nemo-evaluator/src/nemo_evaluator/jobs/evaluate.py b/plugins/nemo-evaluator/src/nemo_evaluator/jobs/evaluate.py index 671d7fce70c..8796b6037d0 100644 --- a/plugins/nemo-evaluator/src/nemo_evaluator/jobs/evaluate.py +++ b/plugins/nemo-evaluator/src/nemo_evaluator/jobs/evaluate.py @@ -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, @@ -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.") diff --git a/plugins/nemo-evaluator/src/nemo_evaluator/jobs/publication.py b/plugins/nemo-evaluator/src/nemo_evaluator/jobs/publication.py index a6bc485fa98..5e7fff3cea5 100644 --- a/plugins/nemo-evaluator/src/nemo_evaluator/jobs/publication.py +++ b/plugins/nemo-evaluator/src/nemo_evaluator/jobs/publication.py @@ -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 @@ -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) diff --git a/plugins/nemo-evaluator/src/nemo_evaluator/jobs/result_persistence.py b/plugins/nemo-evaluator/src/nemo_evaluator/jobs/result_persistence.py index 5f71ac3eccb..b3a32b24c0e 100644 --- a/plugins/nemo-evaluator/src/nemo_evaluator/jobs/result_persistence.py +++ b/plugins/nemo-evaluator/src/nemo_evaluator/jobs/result_persistence.py @@ -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 @@ -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", diff --git a/plugins/nemo-evaluator/src/nemo_evaluator/jobs/utils.py b/plugins/nemo-evaluator/src/nemo_evaluator/jobs/utils.py new file mode 100644 index 00000000000..43145757869 --- /dev/null +++ b/plugins/nemo-evaluator/src/nemo_evaluator/jobs/utils.py @@ -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) diff --git a/plugins/nemo-evaluator/tests/jobs/test_publication.py b/plugins/nemo-evaluator/tests/jobs/test_publication.py index 310aa20eda1..2af1160a362 100644 --- a/plugins/nemo-evaluator/tests/jobs/test_publication.py +++ b/plugins/nemo-evaluator/tests/jobs/test_publication.py @@ -117,6 +117,7 @@ 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)), @@ -124,6 +125,11 @@ def __init__( 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)) @@ -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 diff --git a/plugins/nemo-evaluator/tests/jobs/test_utils.py b/plugins/nemo-evaluator/tests/jobs/test_utils.py new file mode 100644 index 00000000000..54dce48972b --- /dev/null +++ b/plugins/nemo-evaluator/tests/jobs/test_utils.py @@ -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] diff --git a/plugins/nemo-evaluator/tests/test_evaluate_job.py b/plugins/nemo-evaluator/tests/test_evaluate_job.py index b37a5bfde80..069d85ccb35 100644 --- a/plugins/nemo-evaluator/tests/test_evaluate_job.py +++ b/plugins/nemo-evaluator/tests/test_evaluate_job.py @@ -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"), ) @@ -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.""" diff --git a/plugins/nemo-evaluator/tests/test_result_persistence.py b/plugins/nemo-evaluator/tests/test_result_persistence.py index 54a2e6839cf..d73ac71886f 100644 --- a/plugins/nemo-evaluator/tests/test_result_persistence.py +++ b/plugins/nemo-evaluator/tests/test_result_persistence.py @@ -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: @@ -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(), @@ -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"