From ac8ce99e92caf03bdd177534ce911c511c249931 Mon Sep 17 00:00:00 2001 From: Octavian Drulea Date: Thu, 6 Aug 2026 16:05:33 -0700 Subject: [PATCH 1/9] feat(evaluator): model eval publish_to_intake api Signed-off-by: Octavian Drulea --- plugins/nemo-evaluator/openapi/openapi.yaml | 68 ++++++- .../src/nemo_evaluator/intake/row_adapter.py | 133 ++++++++++++++ .../src/nemo_evaluator/jobs/agent_spec.py | 62 ++----- .../src/nemo_evaluator/jobs/evaluate.py | 55 +++++- .../src/nemo_evaluator/jobs/publication.py | 59 +++++- .../nemo_evaluator/jobs/publication_spec.py | 85 +++++++++ .../src/nemo_evaluator/sdk/_executor.py | 15 +- .../src/nemo_evaluator/sdk/http_utils.py | 7 - .../tests/intake/test_row_adapter.py | 171 ++++++++++++++++++ .../integration/test_publish_to_intake.py | 68 +++++++ .../tests/jobs/test_publication.py | 164 ++++++++++++++++- plugins/nemo-evaluator/tests/test_sdk.py | 26 ++- 12 files changed, 836 insertions(+), 77 deletions(-) create mode 100644 plugins/nemo-evaluator/src/nemo_evaluator/intake/row_adapter.py create mode 100644 plugins/nemo-evaluator/src/nemo_evaluator/jobs/publication_spec.py create mode 100644 plugins/nemo-evaluator/tests/intake/test_row_adapter.py diff --git a/plugins/nemo-evaluator/openapi/openapi.yaml b/plugins/nemo-evaluator/openapi/openapi.yaml index d9d602b24a..6742341a23 100644 --- a/plugins/nemo-evaluator/openapi/openapi.yaml +++ b/plugins/nemo-evaluator/openapi/openapi.yaml @@ -3113,6 +3113,11 @@ components: - $ref: '#/components/schemas/FieldMapping' description: Optional mapping from canonical evaluator fields to dataset columns. + publication: + allOf: + - $ref: '#/components/schemas/RowPublicationSpec' + description: Where the completed run publishes its results, beyond its own + result artifacts. Omit to publish nowhere. metrics: items: anyOf: @@ -3444,6 +3449,11 @@ components: - $ref: '#/components/schemas/FieldMapping' description: Optional mapping from canonical evaluator fields to dataset columns. + publication: + allOf: + - $ref: '#/components/schemas/RowPublicationSpec' + description: Where the completed run publishes its results, beyond its own + result artifacts. Omit to publish nowhere. metrics: items: $ref: '#/components/schemas/MetricInline' @@ -4796,8 +4806,7 @@ components: additionalProperties: false type: object title: PublicationSpec - description: Where a completed run publishes its results, beyond its own result - bundle. + description: Where a completed agent-evaluation run publishes its results. ReasoningParams: properties: end_token: @@ -4913,6 +4922,61 @@ components: required: - data title: RevisionsPage + RowIntakePublicationSpec: + properties: + evaluation_id: + type: string + minLength: 1 + title: Evaluation Id + description: Name of the existing Evaluation to publish under. Must already + exist; the job does not create it. + agent_name: + title: Agent Name + description: Agent name recorded on each published trajectory. Derived from + the target when it names one; required otherwise. + type: string + agent_version: + type: string + title: Agent Version + description: Agent version recorded on each published trajectory. Neither + a Model nor an Agent carries a version, so this defaults to 'unknown' + unless the submitter supplies one. + default: unknown + required: + type: boolean + title: Required + description: Fail the job when publication fails. Defaults to True so a + run that asked to publish does not report success with nothing in Experiments. + The result bundle is saved before publication runs, so a failed job still + leaves the results intact to re-publish. Set False to keep the job successful + and report the failure in its output instead. + default: true + test_case_id_field: + title: Test Case Id Field + description: "Dataset column identifying each row, recorded as the published\ + \ test case id. Defaults to the row's position in the run, which is only\ + \ stable for a single-file dataset evaluated in full \u2014 a multi-file\ + \ or glob dataset is concatenated in filesystem order, so positions shift\ + \ between runs and re-published rows would not line up. Name a column\ + \ here when the dataset has a real identifier." + type: string + additionalProperties: false + type: object + required: + - evaluation_id + title: RowIntakePublicationSpec + description: Intake publication for a dataset-driven evaluation, where a trial + is a dataset row. + RowPublicationSpec: + properties: + intake: + allOf: + - $ref: '#/components/schemas/RowIntakePublicationSpec' + description: Publish scored rows to Intake. Omit to publish nowhere. + additionalProperties: false + type: object + title: RowPublicationSpec + description: Where a completed dataset-driven evaluation publishes its results. RubricScoreStat: properties: label: diff --git a/plugins/nemo-evaluator/src/nemo_evaluator/intake/row_adapter.py b/plugins/nemo-evaluator/src/nemo_evaluator/intake/row_adapter.py new file mode 100644 index 0000000000..636201786f --- /dev/null +++ b/plugins/nemo-evaluator/src/nemo_evaluator/intake/row_adapter.py @@ -0,0 +1,133 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Adapt a dataset-driven (row) evaluation result into the shape ``publish_to_intake`` consumes. + +Row evaluation and agent evaluation converge at the Intake boundary: a published trajectory is a +single step carrying the final output text (see ``mapping.trial_to_atif_ingest``), and a row's +``sample`` is already ``{"output_text": ..., "response": ...}`` — the field names of ``AgentOutput``. +So rather than a second mapping and a second publish loop, a row result is adapted to an +``AgentEvalResult`` and goes through the same publisher, inheriting its idempotency guarantees. + +The row vocabulary maps as: one row -> one trial, one (row, metric key) -> one score. +""" + +from __future__ import annotations + +from datetime import datetime + +from nemo_evaluator_sdk.agent_eval.results import AgentEvalResult, AgentEvalSummary, RunMetadata +from nemo_evaluator_sdk.agent_eval.scores import ( + AgentEvalDiagnostic, + AgentEvalDiagnosticSeverity, + AgentEvalScoreStatus, + AgentEvalTaskScore, +) +from nemo_evaluator_sdk.agent_eval.trials import AgentEvalTrial, AgentEvalTrialStatus, AgentOutput +from nemo_evaluator_sdk.values.multi_metric_results import BenchmarkEvaluationResult +from nemo_evaluator_sdk.values.results import EvaluationResult, RowScore + +#: Key ``sample`` carries when generation itself failed, rather than the metric. +_INFERENCE_ERROR = "inference_error" + + +class RowIdentityError(ValueError): + """A configured ``test_case_id_field`` is missing from a row.""" + + +def _test_case_id(row: RowScore, index: int, test_case_id_field: str | None) -> str: + """Stable identity for a row, used as both trial id and published test case id.""" + if test_case_id_field is not None: + # Deliberately not falling back to the positional id: the whole point of naming a column is + # that positions are not stable, so a silent fallback would publish rows that never line up + # with the previous run and give no indication of why. + if test_case_id_field not in row.item: + raise RowIdentityError( + f"Row {index} has no {test_case_id_field!r} column; " + f"available columns: {sorted(row.item)}. " + "Fix `publication.intake.test_case_id_field` or remove it to use row position." + ) + return str(row.item[test_case_id_field]) + return f"row-{row.row_index if row.row_index is not None else index}" + + +def _output(row: RowScore) -> AgentOutput | None: + """The row's generated output, or ``None`` when generation failed or produced nothing.""" + if row.sample.get(_INFERENCE_ERROR): + return None + output_text = row.sample.get("output_text") + response = row.sample.get("response") + if output_text is None and not response: + return None + return AgentOutput(output_text=output_text, response=response) + + +def _scores(row: RowScore, *, run_id: str, test_case_id: str) -> list[AgentEvalTaskScore]: + """One score per metric key on the row; ``metrics`` values are already ``MetricOutput``.""" + errors = row.metric_errors or {} + diagnostics = row.metric_diagnostics or {} + scores: list[AgentEvalTaskScore] = [] + for metric_key, outputs in row.metrics.items(): + error = errors.get(metric_key) + # Error first: `score_to_evaluator_results` publishes `diagnostics[0].message` as the row's + # comment, and the failure is what a reader needs to see there. + row_diagnostics = [ + AgentEvalDiagnostic(severity=AgentEvalDiagnosticSeverity.WARNING, message=item.message) + for item in diagnostics.get(metric_key, []) + ] + if error: + row_diagnostics.insert(0, AgentEvalDiagnostic(severity=AgentEvalDiagnosticSeverity.ERROR, message=error)) + scores.append( + AgentEvalTaskScore( + id=f"{run_id}:{test_case_id}:{metric_key}", + run_id=run_id, + task_id=test_case_id, + trial_id=test_case_id, + metric_type=metric_key, + status=AgentEvalScoreStatus.FAILED if error else AgentEvalScoreStatus.COMPLETED, + outputs=list(outputs), + diagnostics=row_diagnostics, + ) + ) + return scores + + +def row_result_to_agent_eval_result( + result: EvaluationResult | BenchmarkEvaluationResult, + *, + run_id: str, + started_at: datetime, + test_case_id_field: str | None = None, +) -> AgentEvalResult: + """Adapt a row evaluation result for ``publish_to_intake``. + + ``run_id`` and ``started_at`` come from the job: a row result carries neither, and both must be + stable across a re-publish or the trajectory lands as a duplicate span rather than replacing the + previous one. + + Reads the top-level ``row_scores`` only. ``BenchmarkEvaluationResult`` repeats every row under + ``per_metric[key].row_scores`` as well; walking those would publish each row once per metric. + """ + trials: list[AgentEvalTrial] = [] + scores: list[AgentEvalTaskScore] = [] + for index, row in enumerate(result.row_scores): + test_case_id = _test_case_id(row, index, test_case_id_field) + output = _output(row) + trials.append( + AgentEvalTrial( + id=test_case_id, + task_id=test_case_id, + status=AgentEvalTrialStatus.COMPLETED if output is not None else AgentEvalTrialStatus.FAILED, + output=output, + ) + ) + scores.extend(_scores(row, run_id=run_id, test_case_id=test_case_id)) + + return AgentEvalResult( + run_id=run_id, + tasks=[], + trials=trials, + scores=scores, + summary=AgentEvalSummary(scores=result.aggregate_scores), + metadata=RunMetadata(started_at=started_at), + ) diff --git a/plugins/nemo-evaluator/src/nemo_evaluator/jobs/agent_spec.py b/plugins/nemo-evaluator/src/nemo_evaluator/jobs/agent_spec.py index 8e28f50e7b..f8fb5e0e93 100644 --- a/plugins/nemo-evaluator/src/nemo_evaluator/jobs/agent_spec.py +++ b/plugins/nemo-evaluator/src/nemo_evaluator/jobs/agent_spec.py @@ -19,13 +19,14 @@ import nemo_evaluator.shared.metric_bundles.cloudpickle # noqa: F401 import nemo_evaluator.shared.metric_bundles.inline # noqa: F401 from nemo_evaluator.api.schemas import MetricInline, TaskInputs, TaskMetadataList, TasksetRef -from nemo_evaluator.intake.mapping import DEFAULT_AGENT_VERSION from nemo_evaluator.jobs.metric_resolution import to_runtime_bundle, unresolved_model_refs +from nemo_evaluator.jobs.publication_spec import PublicationSpec from nemo_evaluator.metric_refs import MetricRefOrInline from nemo_evaluator.shared.metric_bundles.bundles import unbundle_metric from nemo_evaluator_sdk.agent_eval.tasks import SemanticView from nemo_evaluator_sdk.agent_eval.trials import AgentEvalTrial from nemo_evaluator_sdk.values import Agent, Model, RunConfigOnline, RunConfigOnlineModel +from nemo_evaluator_sdk.values.agents import AgentBase from pydantic import BaseModel, ConfigDict, Field, model_validator @@ -153,7 +154,7 @@ class HarborRunnerTarget(BaseModel): Target: TypeAlias = ModelTarget | AgentTarget | AgentRunnerTarget -def target_agent_identity(target: Target | None) -> tuple[str | None, str | None]: +def target_agent_identity(target: Target | Model | AgentBase | None) -> tuple[str | None, str | None]: """``(agent_name, model_name)`` derivable from a target, for publishing to Intake. Only targets that carry a real name yield one — nothing here invents an identity, because a @@ -161,6 +162,10 @@ def target_agent_identity(target: Target | None) -> tuple[str | None, str | None has a model but no agent; the runners other than Harbor name a harness, not an agent. Those cases return ``None`` and the spec must carry ``publication.intake.agent_name``. + Accepts both unions: agent-eval passes its ``Target`` spec wrappers, while the dataset-driven + eval's ``TargetSpec`` is the bare ``Model``/``Agent`` SDK value. Without the bare branches a row + target falls through to ``(None, None)`` and publishes under an empty agent name. + Distinct from ``result_persistence._agent_target_fields``, which flattens the same targets to ``(kind, name, url)`` filter traits and folds runner *models* into its ``name`` slot. """ @@ -172,57 +177,14 @@ def target_agent_identity(target: Target | None) -> tuple[str | None, str | None return None, target.model.name if isinstance(target, CodexRunnerTarget | FabricRunnerTarget): return None, target.model + # Bare SDK values, as carried by the dataset-driven eval spec. + if isinstance(target, AgentBase): + return target.name, None + if isinstance(target, Model): + return None, target.name return None, None -class IntakePublicationSpec(BaseModel): - """Publish this run's trials and scores to Intake, under an Evaluation that already exists. - - ``evaluation_id`` is the *name* of a ``client.evaluations`` record. Intake stores that record as - its ``Experiment`` entity and the SDK's ``publish_to_intake`` calls the argument - ``experiment_id``, but the value is the same one either way — the parent ``client.experiments`` - group is a different resource and is not what goes here. The job never creates the Evaluation: a - missing one is an error, because nothing in an eval spec can supply the dataset identity - ``evaluations.create`` requires. - """ - - model_config = ConfigDict(extra="forbid") - - evaluation_id: str = Field( - min_length=1, - description="Name of the existing Evaluation to publish under. Must already exist; the job does not create it.", - ) - agent_name: str | None = Field( - default=None, - min_length=1, - description="Agent name recorded on each published trajectory. Derived from the target when " - "it names one; required otherwise.", - ) - agent_version: str = Field( - default=DEFAULT_AGENT_VERSION, - min_length=1, - description="Agent version recorded on each published trajectory. Neither a Model nor an " - "Agent carries a version, so this defaults to 'unknown' unless the submitter supplies one.", - ) - required: bool = Field( - default=True, - description="Fail the job when publication fails. Defaults to True so a run that asked to " - "publish does not report success with nothing in Experiments. The result bundle is saved " - "before publication runs, so a failed job still leaves the results intact to re-publish. " - "Set False to keep the job successful and report the failure in its output instead.", - ) - - -class PublicationSpec(BaseModel): - """Where a completed run publishes its results, beyond its own result bundle.""" - - model_config = ConfigDict(extra="forbid") - - intake: IntakePublicationSpec | None = Field( - default=None, description="Publish trials and scores to Intake. Omit to publish nowhere." - ) - - class _AgentEvalTaskCommon(BaseModel): """Fields shared by the submitter and canonical task DTOs (everything but ``metrics``). diff --git a/plugins/nemo-evaluator/src/nemo_evaluator/jobs/evaluate.py b/plugins/nemo-evaluator/src/nemo_evaluator/jobs/evaluate.py index 8796b6037d..87ea58f37c 100644 --- a/plugins/nemo-evaluator/src/nemo_evaluator/jobs/evaluate.py +++ b/plugins/nemo-evaluator/src/nemo_evaluator/jobs/evaluate.py @@ -8,6 +8,7 @@ import json import logging from dataclasses import dataclass +from datetime import UTC, datetime from pathlib import Path from typing import Annotated, Any, ClassVar, Self, TypeAlias, cast @@ -17,11 +18,14 @@ import nemo_evaluator.shared.metric_bundles.inline # noqa: F401 from nemo_evaluator.api.schemas import MetricInline from nemo_evaluator.filesets import FilesetRef, download_dataset, download_dataset_sync +from nemo_evaluator.jobs.agent_spec import target_agent_identity from nemo_evaluator.jobs.metric_resolution import ( resolve_metrics_to_inline, to_runtime_bundle, unresolved_model_refs, ) +from nemo_evaluator.jobs.publication import publish_row_eval_result +from nemo_evaluator.jobs.publication_spec import RowPublicationSpec 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 @@ -129,12 +133,33 @@ class _EvaluateSpecCommon(BaseModel): field_mapping: FieldMapping | None = Field( default=None, description="Optional mapping from canonical evaluator fields to dataset columns." ) + publication: RowPublicationSpec | None = Field( + default=None, + description="Where the completed run publishes its results, beyond its own result artifacts. " + "Omit to publish nowhere.", + ) @model_validator(mode="after") def validate_params_for_target(self) -> Self: self.params = resolve_params(self.params, self.target) return self + @model_validator(mode="after") + def _require_resolvable_publication_identity(self) -> Self: + # Publishing needs an agent name and only some targets carry one. Rejecting here makes it a + # 422 on submit rather than a failure discovered after the evaluation has already run — and + # without it a target that names nothing publishes every trajectory under an empty name. + intake = self.publication.intake if self.publication is not None else None + if intake is None or intake.agent_name is not None: + return self + if target_agent_identity(self.target)[0] is None: + source = "an offline evaluation" if self.target is None else f"a {type(self.target).__name__} target" + raise ValueError( + f"`publication.intake.agent_name` is required: it cannot be derived from {source}. " + "Supply the name the published trajectories should be recorded under." + ) + return self + class EvaluateInputSpec(_EvaluateSpecCommon): """Submitter-facing SDK evaluation input for the evaluator plugin job.""" @@ -219,7 +244,10 @@ async def to_spec( *, workspace: str, entity_client: object, - async_sdk: AsyncNeMoPlatform | None, + # Widened from the base signature: `resolve_metrics_to_inline` documents that it takes + # either client, and the local-run path (`_executor._resolve_sync_local_spec`) forwards the + # sync one. Contravariant, so overriding with a wider parameter stays substitutable. + async_sdk: AsyncNeMoPlatform | NeMoPlatform | None, is_local: bool, ) -> BaseModel: """Resolve submitter-facing model and metric references into the canonical evaluation spec.""" @@ -243,6 +271,7 @@ async def to_spec( target=submit_spec.target, prompt_template=submit_spec.prompt_template, field_mapping=submit_spec.field_mapping, + publication=submit_spec.publication, ) def run( @@ -255,6 +284,10 @@ def run( ) -> dict: """Run the evaluator job locally and persist its result artifact.""" spec = EvaluateSpec.model_validate(config) + # Stamped here because the row evaluator records no timing at all and `EvaluationResult` has + # nowhere to put it. Publication needs a start time that is a function of the run, not of + # when it was published, or re-ingest duplicates spans instead of replacing them. + started_at = datetime.now(UTC) evaluator = Evaluator() params = resolve_params(spec.params, spec.target) metrics = [unbundle_metric(to_runtime_bundle(metric)) for metric in spec.metrics] @@ -334,7 +367,25 @@ def run( # status="completed", # ) - return { + output = { "status": "completed", "artifact": artifact.model_dump(), } + + # Publication runs last, after the artifacts and the queryable record are both durable, so a + # failed publish costs a re-publish rather than a re-run. It is also the only step here that + # can fail the job (when `required`). + intake = spec.publication.intake if spec.publication is not None else None + if intake is not None: + outcome = publish_row_eval_result( + result, + spec=intake, + target=spec.target, + run_id=ctx.job_id, + started_at=started_at, + workspace=ctx.workspace, + async_sdk=async_sdk, + ) + output["publication"] = outcome.model_dump(exclude_none=True) + + return output diff --git a/plugins/nemo-evaluator/src/nemo_evaluator/jobs/publication.py b/plugins/nemo-evaluator/src/nemo_evaluator/jobs/publication.py index 5e7fff3cea..88495189ad 100644 --- a/plugins/nemo-evaluator/src/nemo_evaluator/jobs/publication.py +++ b/plugins/nemo-evaluator/src/nemo_evaluator/jobs/publication.py @@ -15,11 +15,18 @@ from __future__ import annotations import logging +from datetime import datetime 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.intake.row_adapter import RowIdentityError, row_result_to_agent_eval_result +from nemo_evaluator.jobs.agent_spec import Target, target_agent_identity +from nemo_evaluator.jobs.publication_spec import IntakePublicationSpec, RowIntakePublicationSpec from nemo_evaluator.jobs.utils import run_with_isolated_async_sdk from nemo_evaluator_sdk.agent_eval.results import AgentEvalResult +from nemo_evaluator_sdk.values import Model +from nemo_evaluator_sdk.values.agents import AgentBase +from nemo_evaluator_sdk.values.multi_metric_results import BenchmarkEvaluationResult +from nemo_evaluator_sdk.values.results import EvaluationResult from nemo_platform import AsyncNeMoPlatform from nemo_platform._exceptions import NeMoPlatformError, NotFoundError from nemo_platform_plugin.jobs.schemas import PlatformJobStatus @@ -118,7 +125,7 @@ def publish_agent_eval_result( result: AgentEvalResult, *, spec: IntakePublicationSpec, - target: Target | None, + target: Target | Model | AgentBase | None, workspace: str, async_sdk: AsyncNeMoPlatform | None, ) -> PublicationOutcome: @@ -188,3 +195,51 @@ def fail(error: str, report: PublishReport | None = None) -> PublicationOutcome: spec.evaluation_id, ) return _completed(spec.evaluation_id, report) + + +def publish_row_eval_result( + result: EvaluationResult | BenchmarkEvaluationResult, + *, + spec: RowIntakePublicationSpec, + target: Model | AgentBase | None, + run_id: str | None, + started_at: datetime, + workspace: str, + async_sdk: AsyncNeMoPlatform | None, +) -> PublicationOutcome: + """Publish a finished dataset-driven run to Intake and describe what happened. + + Adapts the row result to the shape the publisher consumes, then hands off to + :func:`publish_agent_eval_result` — the failure semantics, the outcome shape, and the + ``required`` behaviour are all the same. + + ``run_id`` is the job id and is ``None`` on a platformless local run. Unlike agent eval, whose + result always carries a generated run id, a row result has none, so there is nothing stable to + key published sessions on and the run cannot be published. + """ + if run_id is None: + outcome = _failed( + spec.evaluation_id, + "No job id to publish under (platformless local run); a dataset-driven evaluation takes " + "its run identity from the job.", + None, + ) + if spec.required: + raise PublicationFailedError(outcome) + return outcome + + try: + adapted = row_result_to_agent_eval_result( + result, + run_id=run_id, + started_at=started_at, + test_case_id_field=spec.test_case_id_field, + ) + except RowIdentityError as error: + outcome = _failed(spec.evaluation_id, str(error), None) + if spec.required: + raise PublicationFailedError(outcome) from error + logger.warning("Publication to Intake failed for evaluation %r: %s", spec.evaluation_id, error) + return outcome + + return publish_agent_eval_result(adapted, spec=spec, target=target, workspace=workspace, async_sdk=async_sdk) diff --git a/plugins/nemo-evaluator/src/nemo_evaluator/jobs/publication_spec.py b/plugins/nemo-evaluator/src/nemo_evaluator/jobs/publication_spec.py new file mode 100644 index 0000000000..a47ba81b82 --- /dev/null +++ b/plugins/nemo-evaluator/src/nemo_evaluator/jobs/publication_spec.py @@ -0,0 +1,85 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Where a completed evaluation publishes its results, beyond its own result bundle. + +Shared by both evaluation jobs — the agent-eval spec and the dataset-driven (row) eval spec — so it +lives here rather than in either one. The row variants add the one field that only makes sense for a +dataset: which column identifies a test case. +""" + +from __future__ import annotations + +from nemo_evaluator.intake.mapping import DEFAULT_AGENT_VERSION +from pydantic import BaseModel, ConfigDict, Field + + +class IntakePublicationSpec(BaseModel): + """Publish this run's trials and scores to Intake, under an Evaluation that already exists. + + ``evaluation_id`` is the *name* of a ``client.evaluations`` record. Intake stores that record as + its ``Experiment`` entity and the SDK's ``publish_to_intake`` calls the argument + ``experiment_id``, but the value is the same one either way — the parent ``client.experiments`` + group is a different resource and is not what goes here. The job never creates the Evaluation: a + missing one is an error, because nothing in an eval spec can supply the dataset identity + ``evaluations.create`` requires. + """ + + model_config = ConfigDict(extra="forbid") + + evaluation_id: str = Field( + min_length=1, + description="Name of the existing Evaluation to publish under. Must already exist; the job does not create it.", + ) + agent_name: str | None = Field( + default=None, + min_length=1, + description="Agent name recorded on each published trajectory. Derived from the target when " + "it names one; required otherwise.", + ) + agent_version: str = Field( + default=DEFAULT_AGENT_VERSION, + min_length=1, + description="Agent version recorded on each published trajectory. Neither a Model nor an " + "Agent carries a version, so this defaults to 'unknown' unless the submitter supplies one.", + ) + required: bool = Field( + default=True, + description="Fail the job when publication fails. Defaults to True so a run that asked to " + "publish does not report success with nothing in Experiments. The result bundle is saved " + "before publication runs, so a failed job still leaves the results intact to re-publish. " + "Set False to keep the job successful and report the failure in its output instead.", + ) + + +class PublicationSpec(BaseModel): + """Where a completed agent-evaluation run publishes its results.""" + + model_config = ConfigDict(extra="forbid") + + intake: IntakePublicationSpec | None = Field( + default=None, description="Publish trials and scores to Intake. Omit to publish nowhere." + ) + + +class RowIntakePublicationSpec(IntakePublicationSpec): + """Intake publication for a dataset-driven evaluation, where a trial is a dataset row.""" + + test_case_id_field: str | None = Field( + default=None, + description="Dataset column identifying each row, recorded as the published test case id. " + "Defaults to the row's position in the run, which is only stable for a single-file dataset " + "evaluated in full — a multi-file or glob dataset is concatenated in filesystem order, so " + "positions shift between runs and re-published rows would not line up. Name a column here " + "when the dataset has a real identifier.", + ) + + +class RowPublicationSpec(BaseModel): + """Where a completed dataset-driven evaluation publishes its results.""" + + model_config = ConfigDict(extra="forbid") + + intake: RowIntakePublicationSpec | None = Field( + default=None, description="Publish scored rows to Intake. Omit to publish nowhere." + ) diff --git a/plugins/nemo-evaluator/src/nemo_evaluator/sdk/_executor.py b/plugins/nemo-evaluator/src/nemo_evaluator/sdk/_executor.py index e6b48a10d5..f1afd7624d 100644 --- a/plugins/nemo-evaluator/src/nemo_evaluator/sdk/_executor.py +++ b/plugins/nemo-evaluator/src/nemo_evaluator/sdk/_executor.py @@ -57,6 +57,17 @@ SubmitTargetSpec = TargetSpec | ModelRef +def create_job_payload(spec: EvaluateInputSpec) -> dict[str, dict[str, Any]]: + """Serialize an evaluator job creation request body. + + Lives here rather than in ``http_utils`` because it is the only thing there that needed + ``EvaluateInputSpec``, and that import made ``http_utils`` — and so everything importing it, + including ``intake.publish`` — pull in ``jobs.evaluate``, which cycles for any module that + ``jobs.evaluate`` itself imports. + """ + return {"spec": spec.model_dump(mode="json")} + + def _require_metric_bundle_packager(metric_bundle_packager: MetricBundlePackager | None) -> MetricBundlePackager: if metric_bundle_packager is None: raise MetricBundlePackagerPolicyError( @@ -212,7 +223,7 @@ def create( resolved_workspace = http_utils.resolve_workspace(self._platform, workspace) response = self._http_client.post( http_utils.url(self._platform, "/v2/workspaces/{workspace}/evaluate/jobs", resolved_workspace), - json=http_utils.create_job_payload(spec), + json=create_job_payload(spec), headers=http_utils.platform_default_headers(self._platform), timeout=self._platform.timeout, ) @@ -414,7 +425,7 @@ async def create( resolved_workspace = http_utils.resolve_workspace(self._platform, workspace) response = await self._http_client.post( http_utils.url(self._platform, "/v2/workspaces/{workspace}/evaluate/jobs", resolved_workspace), - json=http_utils.create_job_payload(spec), + json=create_job_payload(spec), headers=http_utils.platform_default_headers(self._platform), timeout=self._platform.timeout, ) diff --git a/plugins/nemo-evaluator/src/nemo_evaluator/sdk/http_utils.py b/plugins/nemo-evaluator/src/nemo_evaluator/sdk/http_utils.py index 75e6a06c75..ac2d72739d 100644 --- a/plugins/nemo-evaluator/src/nemo_evaluator/sdk/http_utils.py +++ b/plugins/nemo-evaluator/src/nemo_evaluator/sdk/http_utils.py @@ -5,10 +5,8 @@ from __future__ import annotations -from typing import Any from urllib.parse import quote, urljoin -from nemo_evaluator.jobs.evaluate import EvaluateInputSpec from nemo_platform import AsyncNeMoPlatform, NeMoPlatform PlatformClient = NeMoPlatform | AsyncNeMoPlatform @@ -63,11 +61,6 @@ def platform_default_headers(platform: PlatformClient) -> dict[str, str]: return {str(key): value for key, value in platform.default_headers.items() if isinstance(value, str)} -def create_job_payload(spec: EvaluateInputSpec) -> dict[str, dict[str, Any]]: - """Serialize an evaluator job creation request body.""" - return {"spec": spec.model_dump(mode="json")} - - def job_route_base_url(*, raw_base_url: str, workspace: str, job_name: str) -> str: """Build the stable evaluator plugin URL prefix for one submitted job.""" encoded_workspace = quote(workspace, safe="") diff --git a/plugins/nemo-evaluator/tests/intake/test_row_adapter.py b/plugins/nemo-evaluator/tests/intake/test_row_adapter.py new file mode 100644 index 0000000000..49b229bf36 --- /dev/null +++ b/plugins/nemo-evaluator/tests/intake/test_row_adapter.py @@ -0,0 +1,171 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Unit tests for adapting a dataset-driven eval result into the publisher's shape.""" + +from __future__ import annotations + +from datetime import UTC, datetime +from typing import Any + +import pytest +from nemo_evaluator.intake.row_adapter import RowIdentityError, row_result_to_agent_eval_result +from nemo_evaluator_sdk.agent_eval.scores import AgentEvalScoreStatus +from nemo_evaluator_sdk.agent_eval.trials import AgentEvalTrialStatus +from nemo_evaluator_sdk.metrics.protocol import MetricOutput +from nemo_evaluator_sdk.values.multi_metric_results import BenchmarkEvaluationResult +from nemo_evaluator_sdk.values.results import AggregatedMetricResult, EvaluationResult, RowScore + +RUN_ID = "job-1" +STARTED_AT = datetime(2026, 1, 2, 3, 4, 5, tzinfo=UTC) + + +def _row( + *, + row_index: int | None = 0, + item: dict[str, Any] | None = None, + sample: dict[str, Any] | None = None, + metrics: dict[str, list[MetricOutput]] | None = None, + metric_errors: dict[str, str] | None = None, + metric_diagnostics: dict[str, list[Any]] | None = None, +) -> RowScore: + return RowScore( + row_index=row_index, + item=item if item is not None else {"question": "2+2?"}, + sample=sample if sample is not None else {"output_text": "4", "response": {"choices": []}}, + metrics=metrics if metrics is not None else {"exact_match": [MetricOutput(name="score", value=1.0)]}, + requests=[], + metric_errors=metric_errors, + metric_diagnostics=metric_diagnostics, + ) + + +def _result(rows: list[RowScore]) -> EvaluationResult: + return EvaluationResult(row_scores=rows, aggregate_scores=AggregatedMetricResult(scores=[])) + + +def _adapt(rows: list[RowScore], **kwargs: Any) -> Any: + return row_result_to_agent_eval_result(_result(rows), run_id=RUN_ID, started_at=STARTED_AT, **kwargs) + + +# --- shape ------------------------------------------------------------------ + + +def test_row_becomes_a_trial_carrying_its_sample() -> None: + result = _adapt([_row()]) + + assert result.run_id == RUN_ID + assert result.metadata.started_at == STARTED_AT + assert len(result.trials) == 1 + trial = result.trials[0] + assert trial.status == AgentEvalTrialStatus.COMPLETED + assert trial.output is not None + assert trial.output.output_text == "4" + assert trial.output.response == {"choices": []} + + +def test_each_metric_key_becomes_its_own_score() -> None: + result = _adapt( + [ + _row( + metrics={ + "exact_match": [MetricOutput(name="score", value=1.0)], + "judge": [MetricOutput(name="verdict", value="correct")], + } + ) + ] + ) + + assert {score.metric_type for score in result.scores} == {"exact_match", "judge"} + assert all(score.trial_id == result.trials[0].id for score in result.scores) + assert all(score.run_id == RUN_ID for score in result.scores) + # Score ids must be distinct or Intake would collapse them onto one row. + assert len({score.id for score in result.scores}) == 2 + + +def test_aggregate_scores_carry_into_the_summary() -> None: + aggregates = AggregatedMetricResult(scores=[]) + result = row_result_to_agent_eval_result( + EvaluationResult(row_scores=[_row()], aggregate_scores=aggregates), + run_id=RUN_ID, + started_at=STARTED_AT, + ) + assert result.summary.scores == aggregates + + +def test_benchmark_result_rows_are_not_published_once_per_metric() -> None: + # BenchmarkEvaluationResult repeats every row under per_metric; only the top-level list counts. + row = _row() + single = _result([row]) + result = row_result_to_agent_eval_result( + BenchmarkEvaluationResult( + row_scores=[row], + aggregate_scores=AggregatedMetricResult(scores=[]), + per_metric={"exact_match": single, "judge": single}, + ), + run_id=RUN_ID, + started_at=STARTED_AT, + ) + assert len(result.trials) == 1 + + +# --- identity --------------------------------------------------------------- + + +def test_defaults_to_row_position() -> None: + result = _adapt([_row(row_index=0), _row(row_index=1)]) + assert [trial.id for trial in result.trials] == ["row-0", "row-1"] + assert [trial.task_id for trial in result.trials] == ["row-0", "row-1"] + + +def test_falls_back_to_enumeration_when_row_index_is_absent() -> None: + result = _adapt([_row(row_index=None), _row(row_index=None)]) + assert [trial.id for trial in result.trials] == ["row-0", "row-1"] + + +def test_test_case_id_field_overrides_position() -> None: + result = _adapt( + [_row(item={"qid": "q-42"}), _row(row_index=1, item={"qid": "q-7"})], + test_case_id_field="qid", + ) + assert [trial.id for trial in result.trials] == ["q-42", "q-7"] + + +def test_non_string_id_column_is_coerced() -> None: + result = _adapt([_row(item={"qid": 42})], test_case_id_field="qid") + assert result.trials[0].id == "42" + + +def test_missing_id_column_raises_instead_of_falling_back() -> None: + # A silent fallback to row position would reinstate exactly the instability the field exists to + # remove, and the caller would have no way to notice. + with pytest.raises(RowIdentityError, match="no 'qid' column"): + _adapt([_row(item={"question": "2+2?"})], test_case_id_field="qid") + + +# --- failures --------------------------------------------------------------- + + +def test_inference_failure_becomes_a_failed_trial() -> None: + result = _adapt([_row(sample={"output_text": None, "response": {}, "inference_error": "boom"})]) + assert result.trials[0].status == AgentEvalTrialStatus.FAILED + assert result.trials[0].output is None + + +def test_empty_sample_becomes_a_failed_trial() -> None: + result = _adapt([_row(sample={"output_text": None, "response": {}})]) + assert result.trials[0].status == AgentEvalTrialStatus.FAILED + + +def test_metric_error_becomes_a_failed_score_reporting_why() -> None: + result = _adapt([_row(metric_errors={"exact_match": "judge timed out"})]) + score = result.scores[0] + assert score.status == AgentEvalScoreStatus.FAILED + # Publish surfaces diagnostics[0].message as the row comment, so the error must lead. + assert score.diagnostics[0].message == "judge timed out" + + +def test_a_metric_error_does_not_fail_the_trial_itself() -> None: + # The agent answered; only scoring failed. The trajectory is still worth publishing. + result = _adapt([_row(metric_errors={"exact_match": "judge timed out"})]) + assert result.trials[0].status == AgentEvalTrialStatus.COMPLETED diff --git a/plugins/nemo-evaluator/tests/integration/test_publish_to_intake.py b/plugins/nemo-evaluator/tests/integration/test_publish_to_intake.py index cc191341dd..e5da9b3b92 100644 --- a/plugins/nemo-evaluator/tests/integration/test_publish_to_intake.py +++ b/plugins/nemo-evaluator/tests/integration/test_publish_to_intake.py @@ -34,10 +34,12 @@ import pytest from nemo_evaluator.intake.publish import PublishReport, publish_to_intake +from nemo_evaluator.intake.row_adapter import row_result_to_agent_eval_result from nemo_evaluator_sdk.agent_eval.results import AgentEvalResult, AgentEvalSummary, RunMetadata from nemo_evaluator_sdk.agent_eval.scores import AgentEvalScoreStatus, AgentEvalTaskScore from nemo_evaluator_sdk.agent_eval.trials import AgentEvalTrial, AgentEvalTrialStatus, AgentOutput from nemo_evaluator_sdk.metrics.protocol import MetricOutput +from nemo_evaluator_sdk.values.results import AggregatedMetricResult, EvaluationResult, RowScore from nemo_platform import AsyncNeMoPlatform from nemo_platform.types.intake.trace_filter_param import TraceFilterParam @@ -53,6 +55,8 @@ NAN_RUN_ID = "intake-it-nan-run" IDEMPOTENCY_EXPERIMENT_NAME = "intake-it-idempotency-exp" IDEMPOTENCY_RUN_ID = "intake-it-idempotency-run" +ROW_EXPERIMENT_NAME = "intake-it-row-exp" +ROW_RUN_ID = "intake-it-row-run" #: Recent, not fixed: a trajectory's start_time is a real timestamp, and Intake's trace queries #: only look back a bounded window — a hardcoded past date ingests fine but reads back empty. STARTED_AT = datetime.now(UTC) - timedelta(minutes=5) @@ -415,3 +419,67 @@ async def publish() -> PublishReport: rows = await client.intake.spans.evaluator_results.list(second.published_trials[0].span_id, workspace=WORKSPACE) assert [row.name for row in rows] == ["accuracy.score"] + + +def _row_result() -> EvaluationResult: + """One scored row — the smallest dataset-driven result exercising both write paths.""" + return EvaluationResult( + row_scores=[ + RowScore( + row_index=0, + item={"question": "capital of France?", "qid": "q-1"}, + sample={"output_text": "Paris", "response": {}}, + metrics={"exact_match": [MetricOutput(name="score", value=1.0)]}, + requests=[], + ) + ], + aggregate_scores=AggregatedMetricResult(scores=[]), + ) + + +async def test_row_result_publishes_and_is_idempotent(platform_base_url: str) -> None: + # The dataset-driven path adapts rows into the publisher's shape rather than using a second + # mapping, so it inherits the same idempotency guarantee: re-publishing replaces rather than + # duplicating. Row identity comes from the configured column, not the row's position. + async with AsyncNeMoPlatform(base_url=platform_base_url, max_retries=2) as client: + group = await client.experiments.create(workspace=WORKSPACE, name=GROUP_NAME, exist_ok=True) + await client.evaluations.create( + workspace=WORKSPACE, + name=ROW_EXPERIMENT_NAME, + experiment_ids=[group.id], + dataset_name="intake-it-row-dataset", + dataset_version="v1", + exist_ok=True, + ) + + async def publish() -> PublishReport: + adapted = row_result_to_agent_eval_result( + _row_result(), + run_id=ROW_RUN_ID, + started_at=STARTED_AT, + test_case_id_field="qid", + ) + return await publish_to_intake( + adapted, + platform=client, + experiment_id=ROW_EXPERIMENT_NAME, + workspace=WORKSPACE, + agent_name="intake-it-row-agent", + ) + + first = await publish() + second = await publish() + + assert first.trial_count == second.trial_count == 1 + session_id = second.published_trials[0].session_id + assert session_id == f"{ROW_RUN_ID}:q-1" + assert first.published_trials[0].span_id == second.published_trials[0].span_id + + trace_filter: TraceFilterParam = {"session_id": session_id} + traces = [trace async for trace in client.intake.traces.list(workspace=WORKSPACE, filter=trace_filter)] + assert len(traces) == 1, "re-publish duplicated the row instead of replacing it" + assert traces[0].evaluation_context is not None + assert traces[0].evaluation_context.test_case_id == "q-1" + + rows = await client.intake.spans.evaluator_results.list(second.published_trials[0].span_id, workspace=WORKSPACE) + assert [row.name for row in rows] == ["exact_match.score"] diff --git a/plugins/nemo-evaluator/tests/jobs/test_publication.py b/plugins/nemo-evaluator/tests/jobs/test_publication.py index 2af1160a36..db4d4ebd38 100644 --- a/plugins/nemo-evaluator/tests/jobs/test_publication.py +++ b/plugins/nemo-evaluator/tests/jobs/test_publication.py @@ -14,6 +14,7 @@ import httpx import pytest +from nemo_evaluator.api.schemas import MetricInline from nemo_evaluator.jobs.agent_evaluate import AgentEvalJob from nemo_evaluator.jobs.agent_spec import ( AgentEvalInputSpec, @@ -24,21 +25,31 @@ CodexRunnerTarget, FabricRunnerTarget, HarborRunnerTarget, - IntakePublicationSpec, ModelTarget, - PublicationSpec, Target, target_agent_identity, ) +from nemo_evaluator.jobs.evaluate import EvaluateInputSpec, EvaluateJob, EvaluateSpec from nemo_evaluator.jobs.publication import PublicationFailedError, publish_agent_eval_result +from nemo_evaluator.jobs.publication_spec import ( + IntakePublicationSpec, + PublicationSpec, + RowIntakePublicationSpec, + RowPublicationSpec, +) +from nemo_evaluator.shared.metric_bundles.bundles import bundle_metric +from nemo_evaluator.shared.metric_bundles.cloudpickle import CloudpickleMetricBundlePackager from nemo_evaluator_sdk.agent_eval.results import AgentEvalResult, AgentEvalSummary, RunMetadata from nemo_evaluator_sdk.agent_eval.scores import AgentEvalScoreStatus, AgentEvalTaskScore from nemo_evaluator_sdk.agent_eval.tasks import AgentEvalTask from nemo_evaluator_sdk.agent_eval.trials import AgentEvalTrial, AgentEvalTrialStatus, AgentOutput from nemo_evaluator_sdk.execution.metric_execution import run_sync +from nemo_evaluator_sdk.metrics.exact_match import ExactMatchMetric from nemo_evaluator_sdk.metrics.protocol import MetricOutput -from nemo_evaluator_sdk.values import Model +from nemo_evaluator_sdk.values import Model, RunConfigOnline, RunConfigOnlineModel from nemo_evaluator_sdk.values.agents import NemoAgentToolkitAgent +from nemo_evaluator_sdk.values.multi_metric_results import BenchmarkEvaluationResult # noqa: F401 +from nemo_evaluator_sdk.values.results import AggregatedMetricResult, EvaluationResult, RowScore from nemo_platform import AsyncNeMoPlatform from nemo_platform._exceptions import APIConnectionError, NotFoundError from nemo_platform_plugin.job_context import JobContext, StoragePaths @@ -49,6 +60,13 @@ STARTED_AT = datetime(2026, 1, 2, 3, 4, 5, tzinfo=UTC) +_INLINE_METRIC = MetricInline.model_validate( + bundle_metric( + ExactMatchMetric(reference="{{item.question}}", candidate="{{item.question}}"), + CloudpickleMetricBundlePackager(), + ).model_dump(mode="json") +) + #: Aliased because `_FakeTraces` defines a `list` method, which shadows the builtin in class scope. _SessionIds = list[str] @@ -388,7 +406,7 @@ async def _run(self, tasks: Sequence[AgentEvalTask]) -> AgentEvalResult: ) -def _job_context(tmp_path: Path) -> JobContext: +def _job_context(tmp_path: Path, *, job_id: str | None = None) -> JobContext: storage = StoragePaths(ephemeral=tmp_path / "ephemeral", persistent=tmp_path / "persistent") storage.ephemeral.mkdir() storage.persistent.mkdir() @@ -396,6 +414,7 @@ def _job_context(tmp_path: Path) -> JobContext: workspace="default", storage=storage, results=LocalJobResults(root=storage.persistent / "results"), + job_id=job_id, ) @@ -494,3 +513,140 @@ def test_job_publication_requires_a_run_start_time(tmp_path: Path, mocker: Mocke assert result["publication"]["status"] == PlatformJobStatus.ERROR assert "started_at" in result["publication"]["error"] + + +# --- dataset-driven (row) eval wiring --------------------------------------- + + +class _FakeRowEvaluator: + """Stand-in for the row Evaluator, returning one scored row.""" + + def run_sync(self, **kwargs: Any) -> EvaluationResult: + return EvaluationResult( + row_scores=[ + RowScore( + row_index=0, + item={"question": "2+2?", "qid": "q-1"}, + sample={"output_text": "4", "response": {}}, + metrics={"exact_match": [MetricOutput(name="score", value=1.0)]}, + requests=[], + ) + ], + aggregate_scores=AggregatedMetricResult(scores=[]), + ) + + +def _evaluate_spec(*, required: bool = True, **intake: Any) -> EvaluateSpec: + return EvaluateSpec( + metrics=[_INLINE_METRIC], + dataset=[{"question": "2+2?", "qid": "q-1"}], + target=Model(name="gpt-4o", url="http://model"), + params=RunConfigOnlineModel(), + publication=RowPublicationSpec( + intake=RowIntakePublicationSpec(evaluation_id="eval-1", agent_name="a", required=required, **intake) + ), + ) + + +def test_evaluate_job_does_not_publish_without_a_publication_spec(tmp_path: Path, mocker: MockerFixture) -> None: + mocker.patch("nemo_evaluator.jobs.evaluate.Evaluator", return_value=_FakeRowEvaluator()) + client = _FakeClient() + + spec = EvaluateSpec(metrics=[_INLINE_METRIC], dataset=[{"question": "2+2?"}]) + result = EvaluateJob().run( + spec.model_dump(), ctx=_job_context(tmp_path, job_id="job-1"), async_sdk=cast(AsyncNeMoPlatform, client) + ) + + assert "publication" not in result + assert client.atif_calls == [] + + +def test_evaluate_job_publishes_rows_through_the_real_sync_bridge(tmp_path: Path, mocker: MockerFixture) -> None: + # `run` has already driven one event loop via `evaluator.run_sync`; publication drives another + # through `run_sync` on the same injected SDK. Nothing patched out, so a loop-binding regression + # surfaces as "Event loop is closed". + mocker.patch("nemo_evaluator.jobs.evaluate.Evaluator", return_value=_FakeRowEvaluator()) + client = _FakeClient() + + result = EvaluateJob().run( + _evaluate_spec().model_dump(), + ctx=_job_context(tmp_path, job_id="job-1"), + async_sdk=cast(AsyncNeMoPlatform, client), + ) + + assert result["publication"]["status"] == PlatformJobStatus.COMPLETED + assert result["publication"]["trial_count"] == 1 + assert len(client.atif_calls) == 1 + # The run identity is the job id, so re-publishing the same job replaces rather than duplicates. + assert client.atif_calls[0]["session_id"] == "job-1:row-0" + + +def test_evaluate_job_uses_the_configured_test_case_id_column(tmp_path: Path, mocker: MockerFixture) -> None: + mocker.patch("nemo_evaluator.jobs.evaluate.Evaluator", return_value=_FakeRowEvaluator()) + client = _FakeClient() + + EvaluateJob().run( + _evaluate_spec(test_case_id_field="qid").model_dump(), + ctx=_job_context(tmp_path, job_id="job-1"), + async_sdk=cast(AsyncNeMoPlatform, client), + ) + + assert client.atif_calls[0]["session_id"] == "job-1:q-1" + assert client.atif_calls[0]["evaluation_context"]["test_case_id"] == "q-1" + + +def test_evaluate_job_without_a_job_id_cannot_publish(tmp_path: Path, mocker: MockerFixture) -> None: + # A row result carries no run id of its own, so a platformless local run has nothing stable to + # key sessions on. + mocker.patch("nemo_evaluator.jobs.evaluate.Evaluator", return_value=_FakeRowEvaluator()) + client = _FakeClient() + + result = EvaluateJob().run( + _evaluate_spec(required=False).model_dump(), + ctx=_job_context(tmp_path, job_id=None), + async_sdk=cast(AsyncNeMoPlatform, client), + ) + + assert result["publication"]["status"] == PlatformJobStatus.ERROR + assert "job id" in result["publication"]["error"] + assert client.atif_calls == [] + + +def test_evaluate_job_reports_a_bad_test_case_id_column(tmp_path: Path, mocker: MockerFixture) -> None: + mocker.patch("nemo_evaluator.jobs.evaluate.Evaluator", return_value=_FakeRowEvaluator()) + client = _FakeClient() + + result = EvaluateJob().run( + _evaluate_spec(required=False, test_case_id_field="missing").model_dump(), + ctx=_job_context(tmp_path, job_id="job-1"), + async_sdk=cast(AsyncNeMoPlatform, client), + ) + + assert result["publication"]["status"] == PlatformJobStatus.ERROR + assert "missing" in result["publication"]["error"] + assert client.atif_calls == [] + + +def test_row_target_without_a_derivable_agent_name_is_rejected_at_submit() -> None: + # A Model target names a model, not an agent. Without this the run would publish every + # trajectory under an empty agent name. + with pytest.raises(ValidationError, match="agent_name` is required"): + EvaluateInputSpec( + metrics=[_INLINE_METRIC], + dataset=[{"question": "2+2?"}], + target=Model(name="gpt-4o", url="http://model"), + params=RunConfigOnlineModel(), + publication=RowPublicationSpec(intake=RowIntakePublicationSpec(evaluation_id="eval-1")), + ) + + +def test_row_agent_target_derives_its_name() -> None: + spec = EvaluateInputSpec( + metrics=[_INLINE_METRIC], + dataset=[{"question": "2+2?"}], + target=NemoAgentToolkitAgent(name="my-agent", url="http://agent"), + params=RunConfigOnline(), + prompt_template="{{item.question}}", + publication=RowPublicationSpec(intake=RowIntakePublicationSpec(evaluation_id="eval-1")), + ) + assert spec.publication is not None diff --git a/plugins/nemo-evaluator/tests/test_sdk.py b/plugins/nemo-evaluator/tests/test_sdk.py index df2427cdc6..828b88a3d4 100644 --- a/plugins/nemo-evaluator/tests/test_sdk.py +++ b/plugins/nemo-evaluator/tests/test_sdk.py @@ -11,8 +11,10 @@ import httpx import pytest +from nemo_evaluator.api.schemas import MetricInline from nemo_evaluator.filesets import FilesetRef from nemo_evaluator.jobs.evaluate import EvaluateInputSpec, EvaluateJob, EvaluateSpec +from nemo_evaluator.metric_refs import MetricRefOrInline from nemo_evaluator.sdk import http_utils from nemo_evaluator.sdk._executor import ( MetricBundlePackagerPolicyError, @@ -20,12 +22,12 @@ _build_evaluate_spec, _SyncEvaluatorPluginExecutor, bundle_metrics_for_spec, + create_job_payload, ) from nemo_evaluator.sdk.fs_utils import EvaluatorLocalRunResult from nemo_evaluator.sdk.job_resources import AsyncEvaluatorJobResource, EvaluatorJobResource from nemo_evaluator.sdk.resources import AsyncEvaluator, Evaluator from nemo_evaluator.shared.metric_bundles.bundles import ( - MetricBundle, MetricBundlePackager, MetricBundlePayload, bundle_metric, @@ -65,11 +67,20 @@ _EXACT_MATCH_EVALUATE_INPUT_SPEC_JSON = _EXACT_MATCH_EVALUATE_INPUT_SPEC.model_dump(mode="json") -def _single_metric(spec: EvaluateInputSpec | EvaluateSpec) -> MetricBundle: +def _single_metric(spec: EvaluateInputSpec | EvaluateSpec) -> MetricInline: """Return the single metric from an evaluator job spec.""" if len(spec.metrics) != 1: raise AssertionError("Expected a single metric spec.") - return spec.metrics[0] + metric = spec.metrics[0] + # `EvaluateInputSpec.metrics` also admits `MetricRef`; every caller here builds inline metrics. + assert isinstance(metric, MetricInline) + return metric + + +def _metric_type(metric: MetricRefOrInline) -> str: + """Metric type of an inline metric, narrowing away the `MetricRef` arm of the union.""" + assert isinstance(metric, MetricInline) + return metric.metric_type def _local_run_result(tmp_path: Path, result: EvaluationResult) -> EvaluatorLocalRunResult: @@ -118,7 +129,8 @@ class _SyncPlatform: def __init__(self) -> None: self.base_url = "http://test:8000" self.workspace = "platform-ws" - self.default_headers = {"Authorization": "Bearer sync-platform-token"} + # Deliberately untyped values: one test sets a non-str header to prove they get filtered. + self.default_headers: dict[str, Any] = {"Authorization": "Bearer sync-platform-token"} self.timeout = httpx.Timeout(42.0) self._client = MagicMock(spec=httpx.Client) @@ -159,9 +171,7 @@ def test_http_utils_builds_evaluator_job_creation_request_parts() -> None: "x-trace-id": 123, } - assert http_utils.create_job_payload(_EXACT_MATCH_EVALUATE_INPUT_SPEC) == { - "spec": _EXACT_MATCH_EVALUATE_INPUT_SPEC_JSON - } + assert create_job_payload(_EXACT_MATCH_EVALUATE_INPUT_SPEC) == {"spec": _EXACT_MATCH_EVALUATE_INPUT_SPEC_JSON} assert http_utils.platform_default_headers(cast(NeMoPlatform, platform)) == { "Authorization": "Bearer sync-platform-token" } @@ -284,7 +294,7 @@ def test_build_evaluate_spec_uses_selected_packager_for_all_runtime_metrics() -> ) assert packager.metrics == [metric_a, metric_b] - assert [metric.metric_type for metric in spec.metrics] == ["exact-match", "exact-match"] + assert [_metric_type(metric) for metric in spec.metrics] == ["exact-match", "exact-match"] def test_build_evaluate_spec_excludes_aggregate_fields() -> None: From 2ae84c919a1fd3ece12b7ba2101c61e0b78d4088 Mon Sep 17 00:00:00 2001 From: Octavian Drulea Date: Thu, 6 Aug 2026 21:59:05 -0700 Subject: [PATCH 2/9] fix(evaluator): duplicate ids silent-corruption fix Signed-off-by: Octavian Drulea --- .../src/nemo_evaluator/intake/row_adapter.py | 19 ++++++++++++++++++- .../tests/intake/test_row_adapter.py | 14 ++++++++++++++ 2 files changed, 32 insertions(+), 1 deletion(-) diff --git a/plugins/nemo-evaluator/src/nemo_evaluator/intake/row_adapter.py b/plugins/nemo-evaluator/src/nemo_evaluator/intake/row_adapter.py index 636201786f..dfbca4a640 100644 --- a/plugins/nemo-evaluator/src/nemo_evaluator/intake/row_adapter.py +++ b/plugins/nemo-evaluator/src/nemo_evaluator/intake/row_adapter.py @@ -32,7 +32,7 @@ class RowIdentityError(ValueError): - """A configured ``test_case_id_field`` is missing from a row.""" + """A row's published identity is unusable — missing, or shared with another row.""" def _test_case_id(row: RowScore, index: int, test_case_id_field: str | None) -> str: @@ -110,8 +110,25 @@ def row_result_to_agent_eval_result( """ trials: list[AgentEvalTrial] = [] scores: list[AgentEvalTaskScore] = [] + # Published session ids are `{run_id}:{trial id}`, so two rows sharing an id are one session: + # the second trajectory replaces the first and its scores land on the same span. That loses a row + # with nothing to show for it, so refuse the whole run instead — a column that is not unique is a + # misconfiguration, and publishing 1 of 1000 rows silently is the worst way to find out. + seen: dict[str, int] = {} for index, row in enumerate(result.row_scores): test_case_id = _test_case_id(row, index, test_case_id_field) + if test_case_id in seen: + source = ( + f"column {test_case_id_field!r}" + if test_case_id_field is not None + else "row position (rows carry inconsistent `row_index` values)" + ) + raise RowIdentityError( + f"Rows {seen[test_case_id]} and {index} both resolve to test case id " + f"{test_case_id!r} from {source}. Ids must be unique or the rows overwrite each " + "other in Intake." + ) + seen[test_case_id] = index output = _output(row) trials.append( AgentEvalTrial( diff --git a/plugins/nemo-evaluator/tests/intake/test_row_adapter.py b/plugins/nemo-evaluator/tests/intake/test_row_adapter.py index 49b229bf36..4d0cd12125 100644 --- a/plugins/nemo-evaluator/tests/intake/test_row_adapter.py +++ b/plugins/nemo-evaluator/tests/intake/test_row_adapter.py @@ -136,6 +136,20 @@ def test_non_string_id_column_is_coerced() -> None: assert result.trials[0].id == "42" +def test_duplicate_id_column_values_are_rejected() -> None: + # Two rows sharing an id publish as one session, the second silently replacing the first. A + # non-unique column is a misconfiguration, so fail the run rather than lose rows. + with pytest.raises(RowIdentityError, match="both resolve to test case id 'q-1'"): + _adapt([_row(item={"qid": "q-1"}), _row(row_index=1, item={"qid": "q-1"})], test_case_id_field="qid") + + +def test_duplicate_positional_ids_are_rejected() -> None: + # `row_index` is optional, so a mix of set and unset values can collide against the enumeration + # fallback: here row 0 has no index (-> "row-0") and row 1 carries row_index=0 (-> "row-0"). + with pytest.raises(RowIdentityError, match="both resolve to test case id 'row-0'"): + _adapt([_row(row_index=None), _row(row_index=0)]) + + def test_missing_id_column_raises_instead_of_falling_back() -> None: # A silent fallback to row position would reinstate exactly the instability the field exists to # remove, and the caller would have no way to notice. From 1aec95913c6295fc39be511a5cff5f823c7dbb7f Mon Sep 17 00:00:00 2001 From: Octavian Drulea Date: Thu, 6 Aug 2026 22:03:51 -0700 Subject: [PATCH 3/9] fix(evaluator): cover optional params not null edge case Signed-off-by: Octavian Drulea --- .../src/nemo_evaluator/jobs/publication_spec.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/plugins/nemo-evaluator/src/nemo_evaluator/jobs/publication_spec.py b/plugins/nemo-evaluator/src/nemo_evaluator/jobs/publication_spec.py index a47ba81b82..f712a4f22b 100644 --- a/plugins/nemo-evaluator/src/nemo_evaluator/jobs/publication_spec.py +++ b/plugins/nemo-evaluator/src/nemo_evaluator/jobs/publication_spec.py @@ -31,6 +31,10 @@ class IntakePublicationSpec(BaseModel): min_length=1, description="Name of the existing Evaluation to publish under. Must already exist; the job does not create it.", ) + # Both are `min_length=1` because an empty string is worse than omitting them: `agent_name=""` + # is not None, so it satisfies the identity validator and skips the derivation it exists to + # require, then resolves back to "" at publish time — the empty agent name the validator was + # written to prevent. Omit the field to get derivation; there is no use for a blank one. agent_name: str | None = Field( default=None, min_length=1, From 3731048681694c8e733676ee066f1939774618f2 Mon Sep 17 00:00:00 2001 From: Octavian Drulea Date: Fri, 7 Aug 2026 20:58:20 -0700 Subject: [PATCH 4/9] fix(evaluator): identify eval rows by content hash, not position Signed-off-by: Octavian Drulea --- plugins/nemo-evaluator/openapi/openapi.yaml | 14 ++- .../src/nemo_evaluator/intake/row_adapter.py | 92 +++++++++++-------- .../nemo_evaluator/jobs/publication_spec.py | 10 +- .../tests/intake/test_row_adapter.py | 58 ++++++++---- .../tests/jobs/test_publication.py | 5 +- 5 files changed, 113 insertions(+), 66 deletions(-) diff --git a/plugins/nemo-evaluator/openapi/openapi.yaml b/plugins/nemo-evaluator/openapi/openapi.yaml index 6742341a23..b4452c5e8b 100644 --- a/plugins/nemo-evaluator/openapi/openapi.yaml +++ b/plugins/nemo-evaluator/openapi/openapi.yaml @@ -4935,8 +4935,10 @@ components: description: Agent name recorded on each published trajectory. Derived from the target when it names one; required otherwise. type: string + minLength: 1 agent_version: type: string + minLength: 1 title: Agent Version description: Agent version recorded on each published trajectory. Neither a Model nor an Agent carries a version, so this defaults to 'unknown' @@ -4954,12 +4956,14 @@ components: test_case_id_field: title: Test Case Id Field description: "Dataset column identifying each row, recorded as the published\ - \ test case id. Defaults to the row's position in the run, which is only\ - \ stable for a single-file dataset evaluated in full \u2014 a multi-file\ - \ or glob dataset is concatenated in filesystem order, so positions shift\ - \ between runs and re-published rows would not line up. Name a column\ - \ here when the dataset has a real identifier." + \ test case id. Defaults to a hash of the row's content, which keeps a\ + \ row's id stable across dataset reorderings and revisions so the same\ + \ test case can be compared run over run; editing a row makes it a new\ + \ test case. Name a column here when the dataset has a real identifier\ + \ \u2014 it reads better and survives content edits. Values must be unique\ + \ per row; the run is rejected if they are not." type: string + minLength: 1 additionalProperties: false type: object required: diff --git a/plugins/nemo-evaluator/src/nemo_evaluator/intake/row_adapter.py b/plugins/nemo-evaluator/src/nemo_evaluator/intake/row_adapter.py index dfbca4a640..6dc95e9748 100644 --- a/plugins/nemo-evaluator/src/nemo_evaluator/intake/row_adapter.py +++ b/plugins/nemo-evaluator/src/nemo_evaluator/intake/row_adapter.py @@ -9,11 +9,15 @@ So rather than a second mapping and a second publish loop, a row result is adapted to an ``AgentEvalResult`` and goes through the same publisher, inheriting its idempotency guarantees. -The row vocabulary maps as: one row -> one trial, one (row, metric key) -> one score. +The row vocabulary maps as: one row -> one trial, one (row, metric key) -> one score. A row's +test case identity is its content hash by default, so repeated rows become repeated trials of a +single test case — which is what the trial/task split already expresses. """ from __future__ import annotations +import hashlib +import json from datetime import datetime from nemo_evaluator_sdk.agent_eval.results import AgentEvalResult, AgentEvalSummary, RunMetadata @@ -24,31 +28,45 @@ AgentEvalTaskScore, ) from nemo_evaluator_sdk.agent_eval.trials import AgentEvalTrial, AgentEvalTrialStatus, AgentOutput +from nemo_evaluator_sdk.values.dataset_schemas import _KNOWN_BINDING_FIELDS from nemo_evaluator_sdk.values.multi_metric_results import BenchmarkEvaluationResult from nemo_evaluator_sdk.values.results import EvaluationResult, RowScore #: Key ``sample`` carries when generation itself failed, rather than the metric. _INFERENCE_ERROR = "inference_error" +#: Canonical evaluator fields that ``field_mapping`` copies into a row alongside its own columns. +_CANONICAL_FIELDS = frozenset(_KNOWN_BINDING_FIELDS) + class RowIdentityError(ValueError): """A row's published identity is unusable — missing, or shared with another row.""" -def _test_case_id(row: RowScore, index: int, test_case_id_field: str | None) -> str: - """Stable identity for a row, used as both trial id and published test case id.""" - if test_case_id_field is not None: - # Deliberately not falling back to the positional id: the whole point of naming a column is - # that positions are not stable, so a silent fallback would publish rows that never line up - # with the previous run and give no indication of why. - if test_case_id_field not in row.item: - raise RowIdentityError( - f"Row {index} has no {test_case_id_field!r} column; " - f"available columns: {sorted(row.item)}. " - "Fix `publication.intake.test_case_id_field` or remove it to use row position." - ) - return str(row.item[test_case_id_field]) - return f"row-{row.row_index if row.row_index is not None else index}" +def _canonical_row_hash(row: RowScore) -> str: + """Stable ``sha256`` of a row's dataset content, excluding field-mapping's canonical aliases. + + Mirrors ``gym_runtime._canonical_row_hash``: identity is the row content alone, so a row keeps + its id across dataset revisions and reorderings and a changed row becomes a new test case. The + aliases are excluded because they duplicate values already in the row, so hashing them would + change the id whenever only the ``field_mapping`` changed. + """ + payload = {key: value for key, value in row.item.items() if key not in _CANONICAL_FIELDS} + encoded = json.dumps(payload, sort_keys=True, separators=(",", ":"), default=str) + return hashlib.sha256(encoded.encode("utf-8")).hexdigest() + + +def _task_id(row: RowScore, index: int, test_case_id_field: str | None) -> str: + """The row's test case identity — stable run over run, so rollups line up.""" + if test_case_id_field is None: + return _canonical_row_hash(row) + if test_case_id_field not in row.item: + raise RowIdentityError( + f"Row {index} has no {test_case_id_field!r} column; " + f"available columns: {sorted(row.item)}. " + "Fix `publication.intake.test_case_id_field` or remove it to identify rows by content." + ) + return str(row.item[test_case_id_field]) def _output(row: RowScore) -> AgentOutput | None: @@ -62,7 +80,7 @@ def _output(row: RowScore) -> AgentOutput | None: return AgentOutput(output_text=output_text, response=response) -def _scores(row: RowScore, *, run_id: str, test_case_id: str) -> list[AgentEvalTaskScore]: +def _scores(row: RowScore, *, run_id: str, task_id: str, trial_id: str) -> list[AgentEvalTaskScore]: """One score per metric key on the row; ``metrics`` values are already ``MetricOutput``.""" errors = row.metric_errors or {} diagnostics = row.metric_diagnostics or {} @@ -79,10 +97,10 @@ def _scores(row: RowScore, *, run_id: str, test_case_id: str) -> list[AgentEvalT row_diagnostics.insert(0, AgentEvalDiagnostic(severity=AgentEvalDiagnosticSeverity.ERROR, message=error)) scores.append( AgentEvalTaskScore( - id=f"{run_id}:{test_case_id}:{metric_key}", + id=f"{run_id}:{trial_id}:{metric_key}", run_id=run_id, - task_id=test_case_id, - trial_id=test_case_id, + task_id=task_id, + trial_id=trial_id, metric_type=metric_key, status=AgentEvalScoreStatus.FAILED if error else AgentEvalScoreStatus.COMPLETED, outputs=list(outputs), @@ -110,35 +128,33 @@ def row_result_to_agent_eval_result( """ trials: list[AgentEvalTrial] = [] scores: list[AgentEvalTaskScore] = [] - # Published session ids are `{run_id}:{trial id}`, so two rows sharing an id are one session: - # the second trajectory replaces the first and its scores land on the same span. That loses a row - # with nothing to show for it, so refuse the whole run instead — a column that is not unique is a - # misconfiguration, and publishing 1 of 1000 rows silently is the worst way to find out. - seen: dict[str, int] = {} + first_seen: dict[str, int] = {} + occurrences: dict[str, int] = {} for index, row in enumerate(result.row_scores): - test_case_id = _test_case_id(row, index, test_case_id_field) - if test_case_id in seen: - source = ( - f"column {test_case_id_field!r}" - if test_case_id_field is not None - else "row position (rows carry inconsistent `row_index` values)" - ) + task_id = _task_id(row, index, test_case_id_field) + repeat = occurrences.get(task_id, 0) + # A named column that repeats is a misconfiguration — the submitter said it identifies rows. + # Identical content under the content hash is just the same test case evaluated twice. + if repeat and test_case_id_field is not None: raise RowIdentityError( - f"Rows {seen[test_case_id]} and {index} both resolve to test case id " - f"{test_case_id!r} from {source}. Ids must be unique or the rows overwrite each " - "other in Intake." + f"Rows {first_seen[task_id]} and {index} share test case id {task_id!r} from column " + f"{test_case_id_field!r}. Name a column whose values are unique per row." ) - seen[test_case_id] = index + occurrences[task_id] = repeat + 1 + first_seen.setdefault(task_id, index) + # Session ids are `{run_id}:{trial id}`, so repeats need distinct trial ids or the second + # trajectory would replace the first. They keep one `task_id`, which is what rollups group on. + trial_id = task_id if repeat == 0 else f"{task_id}#{repeat + 1}" output = _output(row) trials.append( AgentEvalTrial( - id=test_case_id, - task_id=test_case_id, + id=trial_id, + task_id=task_id, status=AgentEvalTrialStatus.COMPLETED if output is not None else AgentEvalTrialStatus.FAILED, output=output, ) ) - scores.extend(_scores(row, run_id=run_id, test_case_id=test_case_id)) + scores.extend(_scores(row, run_id=run_id, task_id=task_id, trial_id=trial_id)) return AgentEvalResult( run_id=run_id, diff --git a/plugins/nemo-evaluator/src/nemo_evaluator/jobs/publication_spec.py b/plugins/nemo-evaluator/src/nemo_evaluator/jobs/publication_spec.py index f712a4f22b..94212acf1a 100644 --- a/plugins/nemo-evaluator/src/nemo_evaluator/jobs/publication_spec.py +++ b/plugins/nemo-evaluator/src/nemo_evaluator/jobs/publication_spec.py @@ -71,11 +71,13 @@ class RowIntakePublicationSpec(IntakePublicationSpec): test_case_id_field: str | None = Field( default=None, + min_length=1, description="Dataset column identifying each row, recorded as the published test case id. " - "Defaults to the row's position in the run, which is only stable for a single-file dataset " - "evaluated in full — a multi-file or glob dataset is concatenated in filesystem order, so " - "positions shift between runs and re-published rows would not line up. Name a column here " - "when the dataset has a real identifier.", + "Defaults to a hash of the row's content, which keeps a row's id stable across dataset " + "reorderings and revisions so the same test case can be compared run over run; editing a " + "row makes it a new test case. Name a column here when the dataset has a real identifier — " + "it reads better and survives content edits. Values must be unique per row; the run is " + "rejected if they are not.", ) diff --git a/plugins/nemo-evaluator/tests/intake/test_row_adapter.py b/plugins/nemo-evaluator/tests/intake/test_row_adapter.py index 4d0cd12125..86c57e5f47 100644 --- a/plugins/nemo-evaluator/tests/intake/test_row_adapter.py +++ b/plugins/nemo-evaluator/tests/intake/test_row_adapter.py @@ -112,18 +112,35 @@ def test_benchmark_result_rows_are_not_published_once_per_metric() -> None: # --- identity --------------------------------------------------------------- -def test_defaults_to_row_position() -> None: - result = _adapt([_row(row_index=0), _row(row_index=1)]) - assert [trial.id for trial in result.trials] == ["row-0", "row-1"] - assert [trial.task_id for trial in result.trials] == ["row-0", "row-1"] +def test_identity_defaults_to_a_content_hash() -> None: + result = _adapt([_row(item={"qid": "a"}), _row(item={"qid": "b"})]) + ids = [trial.task_id for trial in result.trials] + assert ids[0] != ids[1] + assert all(len(i) == 64 for i in ids) -def test_falls_back_to_enumeration_when_row_index_is_absent() -> None: - result = _adapt([_row(row_index=None), _row(row_index=None)]) - assert [trial.id for trial in result.trials] == ["row-0", "row-1"] +def test_identity_ignores_row_position() -> None: + # The whole point of hashing: a reordered or renumbered dataset keeps its ids. + first = _adapt([_row(row_index=0, item={"qid": "a"}), _row(row_index=1, item={"qid": "b"})]) + reordered = _adapt([_row(row_index=7, item={"qid": "b"}), _row(row_index=9, item={"qid": "a"})]) + assert {t.task_id for t in first.trials} == {t.task_id for t in reordered.trials} -def test_test_case_id_field_overrides_position() -> None: +def test_identity_ignores_field_mapping_aliases() -> None: + # `field_mapping` copies columns into canonical keys on `item`; hashing them would churn ids + # whenever only the mapping changed. + bare = _adapt([_row(item={"question": "2+2?"})]) + mapped = _adapt([_row(item={"question": "2+2?", "input": "2+2?", "reference": "4"})]) + assert bare.trials[0].task_id == mapped.trials[0].task_id + + +def test_changed_content_becomes_a_new_test_case() -> None: + before = _adapt([_row(item={"question": "2+2?"})]) + after = _adapt([_row(item={"question": "3+3?"})]) + assert before.trials[0].task_id != after.trials[0].task_id + + +def test_test_case_id_field_overrides_the_hash() -> None: result = _adapt( [_row(item={"qid": "q-42"}), _row(row_index=1, item={"qid": "q-7"})], test_case_id_field="qid", @@ -137,22 +154,27 @@ def test_non_string_id_column_is_coerced() -> None: def test_duplicate_id_column_values_are_rejected() -> None: - # Two rows sharing an id publish as one session, the second silently replacing the first. A - # non-unique column is a misconfiguration, so fail the run rather than lose rows. - with pytest.raises(RowIdentityError, match="both resolve to test case id 'q-1'"): + # A named column that repeats is a misconfiguration: the submitter said it identifies rows. + with pytest.raises(RowIdentityError, match="share test case id 'q-1'"): _adapt([_row(item={"qid": "q-1"}), _row(row_index=1, item={"qid": "q-1"})], test_case_id_field="qid") -def test_duplicate_positional_ids_are_rejected() -> None: - # `row_index` is optional, so a mix of set and unset values can collide against the enumeration - # fallback: here row 0 has no index (-> "row-0") and row 1 carries row_index=0 (-> "row-0"). - with pytest.raises(RowIdentityError, match="both resolve to test case id 'row-0'"): - _adapt([_row(row_index=None), _row(row_index=0)]) +def test_identical_rows_are_trials_of_one_test_case() -> None: + # Repeated rows are the same test case evaluated twice, which the model already expresses as + # N trials per task. Distinct trial ids keep their sessions apart; the shared task_id groups them. + result = _adapt([_row(item={"qid": "a"}), _row(row_index=1, item={"qid": "a"})]) + task_ids = [trial.task_id for trial in result.trials] + trial_ids = [trial.id for trial in result.trials] + assert task_ids[0] == task_ids[1] + assert trial_ids[0] != trial_ids[1] + assert trial_ids[1] == f"{task_ids[0]}#2" + assert [score.trial_id for score in result.scores] == trial_ids + assert len({score.id for score in result.scores}) == 2 def test_missing_id_column_raises_instead_of_falling_back() -> None: - # A silent fallback to row position would reinstate exactly the instability the field exists to - # remove, and the caller would have no way to notice. + # Falling back to the hash would silently ignore an explicit request and give no indication why + # the expected ids never appeared. with pytest.raises(RowIdentityError, match="no 'qid' column"): _adapt([_row(item={"question": "2+2?"})], test_case_id_field="qid") diff --git a/plugins/nemo-evaluator/tests/jobs/test_publication.py b/plugins/nemo-evaluator/tests/jobs/test_publication.py index db4d4ebd38..4270999387 100644 --- a/plugins/nemo-evaluator/tests/jobs/test_publication.py +++ b/plugins/nemo-evaluator/tests/jobs/test_publication.py @@ -578,7 +578,10 @@ def test_evaluate_job_publishes_rows_through_the_real_sync_bridge(tmp_path: Path assert result["publication"]["trial_count"] == 1 assert len(client.atif_calls) == 1 # The run identity is the job id, so re-publishing the same job replaces rather than duplicates. - assert client.atif_calls[0]["session_id"] == "job-1:row-0" + # The trial half is the row content hash, stable across runs and dataset reorderings. + session_id = client.atif_calls[0]["session_id"] + assert session_id.startswith("job-1:") + assert len(session_id.removeprefix("job-1:")) == 64 def test_evaluate_job_uses_the_configured_test_case_id_column(tmp_path: Path, mocker: MockerFixture) -> None: From 8e0da121ccf12383edc76126683faf8736e3e83f Mon Sep 17 00:00:00 2001 From: Octavian Drulea Date: Fri, 7 Aug 2026 21:09:09 -0700 Subject: [PATCH 5/9] fix(evaluator): log row publication failures that were silent Signed-off-by: Octavian Drulea --- .../nemo-evaluator/src/nemo_evaluator/jobs/publication.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/plugins/nemo-evaluator/src/nemo_evaluator/jobs/publication.py b/plugins/nemo-evaluator/src/nemo_evaluator/jobs/publication.py index 88495189ad..1bb058d863 100644 --- a/plugins/nemo-evaluator/src/nemo_evaluator/jobs/publication.py +++ b/plugins/nemo-evaluator/src/nemo_evaluator/jobs/publication.py @@ -218,14 +218,14 @@ def publish_row_eval_result( key published sessions on and the run cannot be published. """ if run_id is None: - outcome = _failed( - spec.evaluation_id, + error = ( "No job id to publish under (platformless local run); a dataset-driven evaluation takes " - "its run identity from the job.", - None, + "its run identity from the job." ) + outcome = _failed(spec.evaluation_id, error, None) if spec.required: raise PublicationFailedError(outcome) + logger.warning("Publication to Intake failed for evaluation %r: %s", spec.evaluation_id, error) return outcome try: From 2e21da30ab26d8d2e8f408becc6187c1cd9b3da9 Mon Sep 17 00:00:00 2001 From: Octavian Drulea Date: Fri, 7 Aug 2026 21:14:31 -0700 Subject: [PATCH 6/9] fix(evaluator): persist row run identity so re-publish can reuse it Signed-off-by: Octavian Drulea --- .../src/nemo_evaluator/jobs/evaluate.py | 23 ++++++++++++++++--- .../tests/jobs/test_publication.py | 18 +++++++++++++++ 2 files changed, 38 insertions(+), 3 deletions(-) diff --git a/plugins/nemo-evaluator/src/nemo_evaluator/jobs/evaluate.py b/plugins/nemo-evaluator/src/nemo_evaluator/jobs/evaluate.py index 87ea58f37c..06f04fd5c2 100644 --- a/plugins/nemo-evaluator/src/nemo_evaluator/jobs/evaluate.py +++ b/plugins/nemo-evaluator/src/nemo_evaluator/jobs/evaluate.py @@ -68,6 +68,7 @@ ROW_SCORES_RESULT_NAME = "row-scores" AGGREGATE_SCORES_FILE_NAME = "aggregate-scores.json" ROW_SCORES_FILE_NAME = "row-scores.jsonl" +RUN_METADATA_FILE_NAME = "run-metadata.json" RESULT_IGNORE_PATTERNS = ["cache.db", "cache/"] @@ -78,6 +79,7 @@ class EvaluationResultFiles: full_result: Path aggregate_scores: Path row_scores: Path + run_metadata: Path artifacts_dir: Path @@ -215,8 +217,10 @@ async def compile( return compile_evaluate_job(canonical_spec, profile=profile) @staticmethod - def _write_result_files(result: EvaluationArtifactResult, persistent_dir: Path) -> EvaluationResultFiles: - """Write full, aggregate, and row-level evaluator artifacts.""" + def _write_result_files( + result: EvaluationArtifactResult, persistent_dir: Path, *, run_id: str | None, started_at: datetime + ) -> EvaluationResultFiles: + """Write full, aggregate, row-level and run-metadata evaluator artifacts.""" result_payload = result.model_dump(mode="json") full_result_path = persistent_dir / DEFAULT_FILE_NAME full_result_path.write_text(json.dumps(result_payload, indent=2), encoding="utf-8") @@ -230,10 +234,21 @@ def _write_result_files(result: EvaluationArtifactResult, persistent_dir: Path) for row_score in result.row_scores: f.write(row_score.model_dump_json() + "\n") + # `EvaluationResult` has nowhere to carry timings, so the run identity Intake publishes + # under is written beside the scores. A re-publish must reuse these: `session_id` is + # `{run_id}:{trial id}` and the span key includes `start_time`, so minting either afresh + # writes a second trajectory instead of replacing the first. + run_metadata_path = artifacts_dir / RUN_METADATA_FILE_NAME + run_metadata_path.write_text( + json.dumps({"run_id": run_id, "started_at": started_at.isoformat()}, indent=2), + encoding="utf-8", + ) + return EvaluationResultFiles( full_result=full_result_path, aggregate_scores=aggregate_path, row_scores=row_scores_path, + run_metadata=run_metadata_path, artifacts_dir=artifacts_dir, ) @@ -333,7 +348,9 @@ def run( field_mapping=spec.field_mapping, prompt_template=None, ) - result_files = self._write_result_files(result, ctx.storage.persistent) + result_files = self._write_result_files( + result, ctx.storage.persistent, run_id=ctx.job_id, started_at=started_at + ) artifact = ctx.results.save(DEFAULT_RESULT_NAME, result_files.full_result) ctx.results.save(AGGREGATE_SCORES_RESULT_NAME, result_files.aggregate_scores) ctx.results.save(ROW_SCORES_RESULT_NAME, result_files.row_scores) diff --git a/plugins/nemo-evaluator/tests/jobs/test_publication.py b/plugins/nemo-evaluator/tests/jobs/test_publication.py index 4270999387..d36915a91a 100644 --- a/plugins/nemo-evaluator/tests/jobs/test_publication.py +++ b/plugins/nemo-evaluator/tests/jobs/test_publication.py @@ -6,6 +6,7 @@ from __future__ import annotations import asyncio +import json from collections.abc import AsyncIterator, Sequence from datetime import UTC, datetime from pathlib import Path @@ -561,6 +562,23 @@ def test_evaluate_job_does_not_publish_without_a_publication_spec(tmp_path: Path assert client.atif_calls == [] +def test_evaluate_job_persists_the_run_identity_it_published_under(tmp_path: Path, mocker: MockerFixture) -> None: + # `EvaluationResult` carries no timings, so without this artifact a re-publish would have to mint + # a new `started_at` — a different span `start_time` for the same session, which writes a second + # trajectory rather than replacing the first. + mocker.patch("nemo_evaluator.jobs.evaluate.Evaluator", return_value=_FakeRowEvaluator()) + client = _FakeClient() + ctx = _job_context(tmp_path, job_id="job-1") + + EvaluateJob().run(_evaluate_spec().model_dump(), ctx=ctx, async_sdk=cast(AsyncNeMoPlatform, client)) + + persisted = json.loads((ctx.storage.persistent / "artifacts" / "run-metadata.json").read_text()) + assert persisted["run_id"] == "job-1" + published_session = client.atif_calls[0]["session_id"] + assert published_session.startswith(f"{persisted['run_id']}:") + assert client.atif_calls[0]["steps"][0]["timestamp"] == datetime.fromisoformat(persisted["started_at"]) + + def test_evaluate_job_publishes_rows_through_the_real_sync_bridge(tmp_path: Path, mocker: MockerFixture) -> None: # `run` has already driven one event loop via `evaluator.run_sync`; publication drives another # through `run_sync` on the same injected SDK. Nothing patched out, so a loop-binding regression From 6be806052aff6961d46715399e72f2c613c2bf37 Mon Sep 17 00:00:00 2001 From: Octavian Drulea Date: Fri, 7 Aug 2026 21:30:14 -0700 Subject: [PATCH 7/9] fix(evaluator): address review comments Signed-off-by: Octavian Drulea --- .../src/nemo_evaluator/intake/row_adapter.py | 2 +- .../src/nemo_evaluator/jobs/evaluate.py | 2 ++ .../tests/intake/test_row_adapter.py | 15 +++++++++- .../tests/jobs/test_publication.py | 30 ++++++++++++++++--- 4 files changed, 43 insertions(+), 6 deletions(-) diff --git a/plugins/nemo-evaluator/src/nemo_evaluator/intake/row_adapter.py b/plugins/nemo-evaluator/src/nemo_evaluator/intake/row_adapter.py index 6dc95e9748..424b07d6e8 100644 --- a/plugins/nemo-evaluator/src/nemo_evaluator/intake/row_adapter.py +++ b/plugins/nemo-evaluator/src/nemo_evaluator/intake/row_adapter.py @@ -75,7 +75,7 @@ def _output(row: RowScore) -> AgentOutput | None: return None output_text = row.sample.get("output_text") response = row.sample.get("response") - if output_text is None and not response: + if output_text is None and response is None: return None return AgentOutput(output_text=output_text, response=response) diff --git a/plugins/nemo-evaluator/src/nemo_evaluator/jobs/evaluate.py b/plugins/nemo-evaluator/src/nemo_evaluator/jobs/evaluate.py index 06f04fd5c2..8ec8cf750f 100644 --- a/plugins/nemo-evaluator/src/nemo_evaluator/jobs/evaluate.py +++ b/plugins/nemo-evaluator/src/nemo_evaluator/jobs/evaluate.py @@ -66,6 +66,7 @@ ARTIFACTS_RESULT_NAME = "artifacts" AGGREGATE_SCORES_RESULT_NAME = "aggregate-scores" ROW_SCORES_RESULT_NAME = "row-scores" +RUN_METADATA_RESULT_NAME = "run-metadata" AGGREGATE_SCORES_FILE_NAME = "aggregate-scores.json" ROW_SCORES_FILE_NAME = "row-scores.jsonl" RUN_METADATA_FILE_NAME = "run-metadata.json" @@ -354,6 +355,7 @@ def run( artifact = ctx.results.save(DEFAULT_RESULT_NAME, result_files.full_result) ctx.results.save(AGGREGATE_SCORES_RESULT_NAME, result_files.aggregate_scores) ctx.results.save(ROW_SCORES_RESULT_NAME, result_files.row_scores) + ctx.results.save(RUN_METADATA_RESULT_NAME, result_files.run_metadata) ctx.results.save(ARTIFACTS_RESULT_NAME, result_files.artifacts_dir, ignore_patterns=RESULT_IGNORE_PATTERNS) # Persist the queryable result record (aggregate scores); per-row detail lives in the fileset diff --git a/plugins/nemo-evaluator/tests/intake/test_row_adapter.py b/plugins/nemo-evaluator/tests/intake/test_row_adapter.py index 86c57e5f47..c190c81ef5 100644 --- a/plugins/nemo-evaluator/tests/intake/test_row_adapter.py +++ b/plugins/nemo-evaluator/tests/intake/test_row_adapter.py @@ -189,8 +189,21 @@ def test_inference_failure_becomes_a_failed_trial() -> None: def test_empty_sample_becomes_a_failed_trial() -> None: - result = _adapt([_row(sample={"output_text": None, "response": {}})]) + # The generation path omits `output_text`/`response` entirely when falsy, so a row that produced + # nothing carries neither key. + result = _adapt([_row(sample={})]) assert result.trials[0].status == AgentEvalTrialStatus.FAILED + assert result.trials[0].output is None + + +@pytest.mark.parametrize("response", [False, 0, [], {}]) +def test_falsy_response_is_still_an_output(response: object) -> None: + # `AgentOutput.response` is any JSON value, so a falsy one is data, not absence — publishing it + # as a failed trial would drop a completed row's output. + result = _adapt([_row(sample={"output_text": None, "response": response})]) + assert result.trials[0].status == AgentEvalTrialStatus.COMPLETED + assert result.trials[0].output is not None + assert result.trials[0].output.response == response def test_metric_error_becomes_a_failed_score_reporting_why() -> None: diff --git a/plugins/nemo-evaluator/tests/jobs/test_publication.py b/plugins/nemo-evaluator/tests/jobs/test_publication.py index d36915a91a..899d32b8f2 100644 --- a/plugins/nemo-evaluator/tests/jobs/test_publication.py +++ b/plugins/nemo-evaluator/tests/jobs/test_publication.py @@ -522,7 +522,16 @@ def test_job_publication_requires_a_run_start_time(tmp_path: Path, mocker: Mocke class _FakeRowEvaluator: """Stand-in for the row Evaluator, returning one scored row.""" + def __init__(self) -> None: + self.loop: asyncio.AbstractEventLoop | None = None + def run_sync(self, **kwargs: Any) -> EvaluationResult: + # Drives a real loop to completion the way the row `Evaluator` does, so publication + # afterwards runs against an already-closed loop. + return run_sync(self._run) + + async def _run(self) -> EvaluationResult: + self.loop = asyncio.get_running_loop() return EvaluationResult( row_scores=[ RowScore( @@ -572,6 +581,11 @@ def test_evaluate_job_persists_the_run_identity_it_published_under(tmp_path: Pat EvaluateJob().run(_evaluate_spec().model_dump(), ctx=ctx, async_sdk=cast(AsyncNeMoPlatform, client)) + # Registered as its own artifact, like the sibling score files — a re-publish should not have to + # unpack the artifacts directory to recover the identity it must reuse. + registered = ctx.storage.persistent / "results" / "run-metadata" + assert registered.exists() + persisted = json.loads((ctx.storage.persistent / "artifacts" / "run-metadata.json").read_text()) assert persisted["run_id"] == "job-1" published_session = client.atif_calls[0]["session_id"] @@ -580,10 +594,8 @@ def test_evaluate_job_persists_the_run_identity_it_published_under(tmp_path: Pat def test_evaluate_job_publishes_rows_through_the_real_sync_bridge(tmp_path: Path, mocker: MockerFixture) -> None: - # `run` has already driven one event loop via `evaluator.run_sync`; publication drives another - # through `run_sync` on the same injected SDK. Nothing patched out, so a loop-binding regression - # surfaces as "Event loop is closed". - mocker.patch("nemo_evaluator.jobs.evaluate.Evaluator", return_value=_FakeRowEvaluator()) + evaluator = _FakeRowEvaluator() + mocker.patch("nemo_evaluator.jobs.evaluate.Evaluator", return_value=evaluator) client = _FakeClient() result = EvaluateJob().run( @@ -592,6 +604,16 @@ def test_evaluate_job_publishes_rows_through_the_real_sync_bridge(tmp_path: Path async_sdk=cast(AsyncNeMoPlatform, client), ) + # The evaluator drove a loop to completion first; publication then ran on a different one, + # reusing the same injected SDK. That crossing is what raises "Event loop is closed" when the + # client is bound to a dead loop. It does not distinguish `run_sync` from a bare `asyncio.run` — + # no loop is running at this point, so both behave the same here. + ingest_loop = client.intake.ingest.atif.loop + assert evaluator.loop is not None + assert evaluator.loop.is_closed() + assert ingest_loop is not None + assert ingest_loop is not evaluator.loop + assert result["publication"]["status"] == PlatformJobStatus.COMPLETED assert result["publication"]["trial_count"] == 1 assert len(client.atif_calls) == 1 From b50ad05d702575c2772d9bedab5d0748a31c4c09 Mon Sep 17 00:00:00 2001 From: Octavian Drulea Date: Wed, 12 Aug 2026 13:04:32 -0700 Subject: [PATCH 8/9] fix(evaluator): merge conflicts from main, fix docstrings Signed-off-by: Octavian Drulea --- .../src/nemo_evaluator/jobs/publication.py | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/plugins/nemo-evaluator/src/nemo_evaluator/jobs/publication.py b/plugins/nemo-evaluator/src/nemo_evaluator/jobs/publication.py index 1bb058d863..a43a524377 100644 --- a/plugins/nemo-evaluator/src/nemo_evaluator/jobs/publication.py +++ b/plugins/nemo-evaluator/src/nemo_evaluator/jobs/publication.py @@ -1,15 +1,19 @@ # SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 -"""Publish a finished agent-eval run to Intake, when the spec asked for it. +"""Publish a finished evaluation run to Intake, when the spec asked for it. + +Covers both shapes: ``publish_agent_eval_result`` for an agent-eval run, and +``publish_row_eval_result`` for a dataset-driven one, which is adapted to the same +``AgentEvalResult`` shape first (see ``intake.row_adapter``). ``publish_to_intake`` is deliberately not a side effect of ``AgentEvaluator.run()`` (AALGO-290): optionality is structural, you call it or you don't. ``spec.publication.intake`` keeps that shape — absent means no publish, and nothing here runs — while giving the job API a way to request it, which is what Studio needs to get evaluation runs into Experiments. -``run`` is synchronous and the publisher is async, so the call goes through the same ``run_sync`` -bridge ``result_persistence`` uses for the entity write. +``run`` is synchronous and the publisher is async, so the call goes through the same +``run_with_isolated_async_sdk`` bridge ``result_persistence`` uses for the entity write. """ from __future__ import annotations From 2335f4de0132e8f19dfe1702e658178c0ed1c8ad Mon Sep 17 00:00:00 2001 From: Octavian Drulea Date: Wed, 12 Aug 2026 13:39:09 -0700 Subject: [PATCH 9/9] fix(evaluator): fix one test assertion Signed-off-by: Octavian Drulea --- plugins/nemo-evaluator/tests/jobs/test_publication.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/plugins/nemo-evaluator/tests/jobs/test_publication.py b/plugins/nemo-evaluator/tests/jobs/test_publication.py index 899d32b8f2..e19b316fae 100644 --- a/plugins/nemo-evaluator/tests/jobs/test_publication.py +++ b/plugins/nemo-evaluator/tests/jobs/test_publication.py @@ -692,4 +692,4 @@ def test_row_agent_target_derives_its_name() -> None: prompt_template="{{item.question}}", publication=RowPublicationSpec(intake=RowIntakePublicationSpec(evaluation_id="eval-1")), ) - assert spec.publication is not None + assert target_agent_identity(spec.target) == ("my-agent", None)