feat(evaluator): model evaluate wire publish_to_intake() to API

plugins/nemo-evaluator/src/nemo_evaluator/intake/row_adapter.py

@@ -0,0 +1,166 @@
+# 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. 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
+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.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 _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:
+    """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 response is None:
+        return None
+    return AgentOutput(output_text=output_text, response=response)
+
+
+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 {}
+    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}:{trial_id}:{metric_key}",
+                run_id=run_id,
+                task_id=task_id,
+                trial_id=trial_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] = []
+    first_seen: dict[str, int] = {}
+    occurrences: dict[str, int] = {}
+    for index, row in enumerate(result.row_scores):
+        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 {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."
+            )
+        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=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, task_id=task_id, trial_id=trial_id))
+
+    return AgentEvalResult(
+        run_id=run_id,
+        tasks=[],
+        trials=trials,
+        scores=scores,
+        summary=AgentEvalSummary(scores=result.aggregate_scores),
+        metadata=RunMetadata(started_at=started_at),
+    )

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,14 +154,18 @@ 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
     made-up agent name is worse than an explicit one the submitter had to supply. A ``ModelTarget``
     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``).