diff --git a/plugins/nemo-evaluator/openapi/openapi.yaml b/plugins/nemo-evaluator/openapi/openapi.yaml
index 3c0e0c5a2a..130dd09cc2 100644
--- a/plugins/nemo-evaluator/openapi/openapi.yaml
+++ b/plugins/nemo-evaluator/openapi/openapi.yaml
@@ -2132,6 +2132,11 @@ components:
           title: Labels
           description: Caller-supplied tags recorded on the run's metadata (e.g. benchmark,
             mode, backend).
+        publication:
+          allOf:
+          - $ref: '#/components/schemas/PublicationSpec'
+          description: Where the completed run publishes its results, beyond its own
+            result bundle. Omit to publish nowhere.
         tasks:
           anyOf:
           - $ref: '#/components/schemas/TasksetRef'
@@ -2298,6 +2303,11 @@ components:
           title: Labels
           description: Caller-supplied tags recorded on the run's metadata (e.g. benchmark,
             mode, backend).
+        publication:
+          allOf:
+          - $ref: '#/components/schemas/PublicationSpec'
+          description: Where the completed run publishes its results, beyond its own
+            result bundle. Omit to publish nowhere.
         tasks:
           items:
             $ref: '#/components/schemas/AgentEvalTaskSpec'
@@ -2665,9 +2675,12 @@ components:
           title: Name
           description: Name of the score.
         count:
-          type: integer
           title: Count
-          description: Number of samples evaluated (excluding NaN).
+          description: "Number of samples evaluated (excluding NaN). None when the\
+            \ sample size is unknown \u2014 e.g. a figure imported from a backend\
+            \ that reports statistics without the n behind them. Distinct from 0,\
+            \ which asserts that nothing was evaluated."
+          type: integer
         nan_count:
           type: integer
           title: Nan Count
@@ -2688,13 +2701,33 @@ components:
           title: Max
           description: Maximum score value.
           type: number
+        median:
+          title: Median
+          description: Median score value. Equal to percentiles.p50 when a percentile
+            distribution is also present; carried separately because a backend may
+            report a median without one.
+          type: number
         std_dev:
           title: Std Dev
-          description: Standard deviation of the scores.
+          description: Population standard deviation of the scores (divides by n).
+            Describes the spread of the values actually evaluated. See sample_std_dev
+            to estimate the spread of the wider process.
           type: number
         variance:
           title: Variance
-          description: Variance of the scores.
+          description: Population variance of the scores (divides by n). See sample_variance.
+          type: number
+        sample_std_dev:
+          title: Sample Std Dev
+          description: "Sample standard deviation of the scores (Bessel-corrected,\
+            \ divides by n-1). Estimates the spread of the process the values were\
+            \ drawn from \u2014 the right choice when repeated trials sample a stochastic\
+            \ system. None when fewer than two values (undefined, not zero)."
+          type: number
+        sample_variance:
+          title: Sample Variance
+          description: Sample variance of the scores (Bessel-corrected, divides by
+            n-1). None when fewer than two values.
           type: number
         score_type:
           type: string
@@ -2714,7 +2747,6 @@ components:
       type: object
       required:
       - name
-      - count
       - nan_count
       title: AggregateRangeScore
       description: Aggregated statistics for a range-type score with percentiles and
@@ -2726,9 +2758,12 @@ components:
           title: Name
           description: Name of the score.
         count:
-          type: integer
           title: Count
-          description: Number of samples evaluated (excluding NaN).
+          description: "Number of samples evaluated (excluding NaN). None when the\
+            \ sample size is unknown \u2014 e.g. a figure imported from a backend\
+            \ that reports statistics without the n behind them. Distinct from 0,\
+            \ which asserts that nothing was evaluated."
+          type: integer
         nan_count:
           type: integer
           title: Nan Count
@@ -2749,13 +2784,33 @@ components:
           title: Max
           description: Maximum score value.
           type: number
+        median:
+          title: Median
+          description: Median score value. Equal to percentiles.p50 when a percentile
+            distribution is also present; carried separately because a backend may
+            report a median without one.
+          type: number
         std_dev:
           title: Std Dev
-          description: Standard deviation of the scores.
+          description: Population standard deviation of the scores (divides by n).
+            Describes the spread of the values actually evaluated. See sample_std_dev
+            to estimate the spread of the wider process.
           type: number
         variance:
           title: Variance
-          description: Variance of the scores.
+          description: Population variance of the scores (divides by n). See sample_variance.
+          type: number
+        sample_std_dev:
+          title: Sample Std Dev
+          description: "Sample standard deviation of the scores (Bessel-corrected,\
+            \ divides by n-1). Estimates the spread of the process the values were\
+            \ drawn from \u2014 the right choice when repeated trials sample a stochastic\
+            \ system. None when fewer than two values (undefined, not zero)."
+          type: number
+        sample_variance:
+          title: Sample Variance
+          description: Sample variance of the scores (Bessel-corrected, divides by
+            n-1). None when fewer than two values.
           type: number
         score_type:
           type: string
@@ -2777,11 +2832,105 @@ components:
       type: object
       required:
       - name
-      - count
       - nan_count
       - rubric_distribution
       title: AggregateRubricScore
       description: Aggregated statistics for a rubric-type score with category distribution.
+    AggregateScalarScore:
+      properties:
+        name:
+          type: string
+          title: Name
+          description: Name of the score.
+        count:
+          title: Count
+          description: "Number of samples evaluated (excluding NaN). None when the\
+            \ sample size is unknown \u2014 e.g. a figure imported from a backend\
+            \ that reports statistics without the n behind them. Distinct from 0,\
+            \ which asserts that nothing was evaluated."
+          type: integer
+        nan_count:
+          type: integer
+          title: Nan Count
+          description: Number of samples that produced NaN scores.
+        sum:
+          title: Sum
+          description: Sum of all score values.
+          type: number
+        mean:
+          title: Mean
+          description: Mean score value.
+          type: number
+        min:
+          title: Min
+          description: Minimum score value.
+          type: number
+        max:
+          title: Max
+          description: Maximum score value.
+          type: number
+        median:
+          title: Median
+          description: Median score value. Equal to percentiles.p50 when a percentile
+            distribution is also present; carried separately because a backend may
+            report a median without one.
+          type: number
+        std_dev:
+          title: Std Dev
+          description: Population standard deviation of the scores (divides by n).
+            Describes the spread of the values actually evaluated. See sample_std_dev
+            to estimate the spread of the wider process.
+          type: number
+        variance:
+          title: Variance
+          description: Population variance of the scores (divides by n). See sample_variance.
+          type: number
+        sample_std_dev:
+          title: Sample Std Dev
+          description: "Sample standard deviation of the scores (Bessel-corrected,\
+            \ divides by n-1). Estimates the spread of the process the values were\
+            \ drawn from \u2014 the right choice when repeated trials sample a stochastic\
+            \ system. None when fewer than two values (undefined, not zero)."
+          type: number
+        sample_variance:
+          title: Sample Variance
+          description: Sample variance of the scores (Bessel-corrected, divides by
+            n-1). None when fewer than two values.
+          type: number
+        score_type:
+          type: string
+          const: scalar
+          title: Score Type
+          description: Type of score.
+          default: scalar
+        value:
+          type: number
+          title: Value
+          description: The reported value.
+      additionalProperties: false
+      type: object
+      required:
+      - name
+      - nan_count
+      - value
+      title: AggregateScalarScore
+      description: 'A single pre-computed value with no underlying distribution available.
+
+
+        For figures a backend reports as one number (e.g. an environment''s own ``pass@1``
+        or Elo) rather
+
+        than a set of per-sample values the SDK could aggregate itself. ``value``
+        carries the number;
+
+        ``mean``/``min``/``max`` are left unset because there is no sample to describe.
+        Distinct from
+
+        :class:`AggregateRangeScore` so a reader can tell "this is the whole story"
+        from "this summarizes
+
+        ``count`` samples", instead of seeing a range score with a suspicious ``count``
+        of 1.'
     AggregatedMetricResult:
       properties:
         scores:
@@ -2789,6 +2938,7 @@ components:
             anyOf:
             - $ref: '#/components/schemas/AggregateRangeScore'
             - $ref: '#/components/schemas/AggregateRubricScore'
+            - $ref: '#/components/schemas/AggregateScalarScore'
           type: array
           title: Scores
           description: The list of aggregated scores.
@@ -3736,6 +3886,50 @@ components:
         so no code is shipped or executed on load. Used for platform-recognized
 
         built-in metric types.'
+    IntakePublicationSpec:
+      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
+          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'
+            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
+      additionalProperties: false
+      type: object
+      required:
+      - evaluation_id
+      title: IntakePublicationSpec
+      description: "Publish this run's trials and scores to Intake, under an Evaluation\
+        \ that already exists.\n\n``evaluation_id`` is the *name* of a ``client.evaluations``\
+        \ record. Intake stores that record as\nits ``Experiment`` entity and the\
+        \ SDK's ``publish_to_intake`` calls the argument\n``experiment_id``, but the\
+        \ value is the same one either way \u2014 the parent ``client.experiments``\n\
+        group is a different resource and is not what goes here. The job never creates\
+        \ the Evaluation: a\nmissing one is an error, because nothing in an eval spec\
+        \ can supply the dataset identity\n``evaluations.create`` requires."
     JsonValue:
       title: JsonValue
     MetadataItem:
@@ -4483,6 +4677,17 @@ components:
       - created_at
       - updated_at
       title: PlatformJobTaskStatusResponse
+    PublicationSpec:
+      properties:
+        intake:
+          allOf:
+          - $ref: '#/components/schemas/IntakePublicationSpec'
+          description: Publish trials and scores to Intake. Omit to publish nowhere.
+      additionalProperties: false
+      type: object
+      title: PublicationSpec
+      description: Where a completed run publishes its results, beyond its own result
+        bundle.
     ReasoningParams:
       properties:
         end_token:
diff --git a/plugins/nemo-evaluator/src/nemo_evaluator/intake/mapping.py b/plugins/nemo-evaluator/src/nemo_evaluator/intake/mapping.py
index 5f8de82d5f..281812d3f4 100644
--- a/plugins/nemo-evaluator/src/nemo_evaluator/intake/mapping.py
+++ b/plugins/nemo-evaluator/src/nemo_evaluator/intake/mapping.py
@@ -31,6 +31,7 @@
 
 import math
 from dataclasses import dataclass
+from datetime import datetime
 from typing import Literal
 
 from nemo_evaluator_sdk.agent_eval.scores import AgentEvalScoreStatus, AgentEvalTaskScore
@@ -85,6 +86,7 @@ def trial_to_atif_ingest(
     run_id: str,
     experiment_id: str,
     agent_name: str,
+    started_at: datetime,
     agent_version: str = DEFAULT_AGENT_VERSION,
     model_name: str | None = None,
     final_metrics: AtifFinalMetricsParam | None = None,
@@ -95,12 +97,24 @@ def trial_to_atif_ingest(
     a minimal single-step trajectory carrying the trial's final output text, so
     the session/score path works end to end. Real ``steps[]`` reconstructed from
     ``trial.evidence`` arrive with D2.
+
+    ``started_at`` (the run's start time) is stamped on the step because it is what
+    makes re-ingest idempotent. Intake's ``spans`` table is a ``ReplacingMergeTree``
+    keyed on ``(workspace, session_id, start_time, id)``, and a step with no timestamp
+    falls back to the server's per-request ingest clock — so the same trajectory sent
+    twice lands as two rows that never collapse. An explicit timestamp makes the root
+    span's ``start_time`` a function of the run, not of when it was published.
     """
     output_text = trial.output.output_text if trial.output is not None else None
     agent: AtifAgentParam = {"name": agent_name, "version": agent_version}
     if model_name is not None:
         agent["model_name"] = model_name
-    step: AtifStepAgentParam = {"source": "agent", "step_id": 1, "message": output_text or ""}
+    step: AtifStepAgentParam = {
+        "source": "agent",
+        "step_id": 1,
+        "message": output_text or "",
+        "timestamp": started_at,
+    }
 
     body: AtifCreateParams = {
         "schema_version": ATIF_SCHEMA_VERSION,
diff --git a/plugins/nemo-evaluator/src/nemo_evaluator/intake/publish.py b/plugins/nemo-evaluator/src/nemo_evaluator/intake/publish.py
index 11a2acf05b..be5bf60709 100644
--- a/plugins/nemo-evaluator/src/nemo_evaluator/intake/publish.py
+++ b/plugins/nemo-evaluator/src/nemo_evaluator/intake/publish.py
@@ -115,7 +115,14 @@ async def publish_to_intake(
     collected and raised together as a :class:`PublishError` (carrying the partial
     report). The evaluation's local bundle is the system of record and is never
     touched, so the caller can re-run ``publish_to_intake`` once the issue is fixed
-    to publish the remaining trials. (Re-publish is not yet idempotent — see ask X1.)
+    to publish the remaining trials.
+
+    Re-publish is **idempotent**: the session id is derived from the run and trial
+    (``mapping.session_id_for``), the step timestamp from the run's ``started_at``,
+    and each evaluator-result id from its target — so every key Intake replaces on is
+    a function of the result, not of when it was published. Re-sending a trial that
+    already landed replaces its rows instead of duplicating them, which is what makes
+    a worker retry after a partial publish safe.
 
     ``experiment_id`` must reference an Experiment that already exists — ATIF ingest
     rejects unknown experiments with HTTP 400. Creating the Experiment/group is a
@@ -127,6 +134,17 @@ async def publish_to_intake(
     """
     resolved_workspace = http_utils.resolve_workspace(platform, workspace, strict=True)
 
+    # Required, not defaulted to "now": a fallback would silently reintroduce the
+    # publish-time clock that makes re-ingest duplicate rows (see the ingest note on
+    # ``mapping.trial_to_atif_ingest``). A real run always sets it; a hand-built result
+    # must say when it ran.
+    started_at = result.metadata.started_at
+    if started_at is None:
+        raise PublishError(
+            f"Cannot publish run {result.run_id!r}: metadata.started_at is unset, and publishing "
+            "without it would write trajectories that duplicate on re-publish."
+        )
+
     scores_by_trial: dict[str, list[AgentEvalTaskScore]] = defaultdict(list)
     for score in result.scores:
         scores_by_trial[score.trial_id].append(score)
@@ -141,6 +159,7 @@ async def _publish_trial(trial: AgentEvalTrial) -> PublishedTrial:
                 run_id=result.run_id,
                 experiment_id=experiment_id,
                 agent_name=agent_name,
+                started_at=started_at,
                 agent_version=agent_version,
                 model_name=model_name,
             )
diff --git a/plugins/nemo-evaluator/src/nemo_evaluator/jobs/agent_evaluate.py b/plugins/nemo-evaluator/src/nemo_evaluator/jobs/agent_evaluate.py
index 8e0d8e9d54..b1acb2ac3b 100644
--- a/plugins/nemo-evaluator/src/nemo_evaluator/jobs/agent_evaluate.py
+++ b/plugins/nemo-evaluator/src/nemo_evaluator/jobs/agent_evaluate.py
@@ -38,6 +38,7 @@
     Target,
 )
 from nemo_evaluator.jobs.metric_resolution import resolve_metrics_to_inline, to_runtime_bundle
+from nemo_evaluator.jobs.publication import publish_agent_eval_result
 from nemo_evaluator.jobs.result_persistence import persist_agent_eval_result
 from nemo_evaluator.shared.metric_bundles.bundles import unbundle_metric
 from nemo_evaluator.task_refs import resolve_agent_eval_tasks
@@ -197,6 +198,7 @@ async def to_spec(
             max_concurrent_tasks=submit_spec.max_concurrent_tasks,
             fail_fast=submit_spec.fail_fast,
             labels=submit_spec.labels,
+            publication=submit_spec.publication,
         )
 
     @classmethod
@@ -418,4 +420,20 @@ def run(
                 exc_info=True,
             )
 
-        return {"status": "completed", "artifact": artifact.model_dump()}
+        output = {"status": "completed", "artifact": artifact.model_dump()}
+
+        # Publication runs last, after the bundle 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`), which is why nothing depends on its result.
+        intake = spec.publication.intake if spec.publication is not None else None
+        if intake is not None:
+            outcome = publish_agent_eval_result(
+                result,
+                spec=intake,
+                target=spec.target,
+                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/agent_spec.py b/plugins/nemo-evaluator/src/nemo_evaluator/jobs/agent_spec.py
index a073bbbc96..8e28f50e7b 100644
--- a/plugins/nemo-evaluator/src/nemo_evaluator/jobs/agent_spec.py
+++ b/plugins/nemo-evaluator/src/nemo_evaluator/jobs/agent_spec.py
@@ -19,6 +19,7 @@
 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.metric_refs import MetricRefOrInline
 from nemo_evaluator.shared.metric_bundles.bundles import unbundle_metric
@@ -152,6 +153,76 @@ class HarborRunnerTarget(BaseModel):
 Target: TypeAlias = ModelTarget | AgentTarget | AgentRunnerTarget
 
 
+def target_agent_identity(target: Target | 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``.
+
+    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.
+    """
+    if isinstance(target, AgentTarget):
+        return target.agent.name, None
+    if isinstance(target, HarborRunnerTarget):
+        return target.agent_import_path or target.agent_name, target.agent_model_name
+    if isinstance(target, ModelTarget):
+        return None, target.model.name
+    if isinstance(target, CodexRunnerTarget | FabricRunnerTarget):
+        return None, target.model
+    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``).
 
@@ -236,6 +307,26 @@ class _AgentEvalSpecCommon(BaseModel):
         default_factory=dict,
         description="Caller-supplied tags recorded on the run's metadata (e.g. benchmark, mode, backend).",
     )
+    publication: PublicationSpec | None = Field(
+        default=None,
+        description="Where the completed run publishes its results, beyond its own result bundle. "
+        "Omit to publish nowhere.",
+    )
+
+    @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.
+        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 = "the precomputed `trials`" if self.target is None else f"a `{self.target.kind}` 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
 
     @model_validator(mode="after")
     def _require_exactly_one_trial_source(self) -> Self:
diff --git a/plugins/nemo-evaluator/src/nemo_evaluator/jobs/publication.py b/plugins/nemo-evaluator/src/nemo_evaluator/jobs/publication.py
new file mode 100644
index 0000000000..a6bc485fa9
--- /dev/null
+++ b/plugins/nemo-evaluator/src/nemo_evaluator/jobs/publication.py
@@ -0,0 +1,189 @@
+# 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_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.
+"""
+
+from __future__ import annotations
+
+import logging
+
+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_sdk.agent_eval.results import AgentEvalResult
+from nemo_evaluator_sdk.execution.metric_execution import run_sync
+from nemo_platform import AsyncNeMoPlatform
+from nemo_platform._exceptions import NeMoPlatformError, NotFoundError
+from nemo_platform_plugin.jobs.schemas import PlatformJobStatus
+from pydantic import BaseModel, ConfigDict, Field
+
+logger = logging.getLogger(__name__)
+
+
+class PublicationOutcome(BaseModel):
+    """What publication did, as reported in the job output.
+
+    Deliberately not ``PublishReport``: that is the publisher's internal shape and names the
+    Evaluation ``experiment_id``, which contradicts the ``evaluation_id`` the job API accepts. This
+    is the public contract, so it uses the API's vocabulary and omits the rest.
+    """
+
+    model_config = ConfigDict(extra="forbid")
+
+    status: PlatformJobStatus = Field(
+        description="Platform job status vocabulary: COMPLETED when everything published, ERROR "
+        "otherwise. Only those two are ever emitted here.",
+    )
+    evaluation_id: str = Field(description="Evaluation the results were published under.")
+    trial_count: int = Field(default=0, description="Trials actually published (partial on failure).")
+    evaluator_result_count: int = Field(default=0, description="Evaluator-result rows written.")
+    skipped: list[str] = Field(
+        default_factory=list,
+        description="Score outputs Intake cannot represent, as 'trial_id: name (reason)'.",
+    )
+    error: str | None = Field(default=None, description="Why publication failed; absent on success.")
+
+
+class PublicationFailedError(RuntimeError):
+    """Publication failed and the spec marked it required, so the job fails with it.
+
+    Carries the outcome so the caller can still report what landed before the failure.
+    """
+
+    def __init__(self, outcome: PublicationOutcome) -> None:
+        super().__init__(outcome.error or "publication failed")
+        self.outcome = outcome
+
+
+def _skipped_lines(report: PublishReport) -> list[str]:
+    return [f"{item.trial_id}: {item.name} ({item.reason})" for item in report.skipped]
+
+
+def _completed(evaluation_id: str, report: PublishReport) -> PublicationOutcome:
+    return PublicationOutcome(
+        status=PlatformJobStatus.COMPLETED,
+        evaluation_id=evaluation_id,
+        trial_count=report.trial_count,
+        evaluator_result_count=report.evaluator_result_count,
+        skipped=_skipped_lines(report),
+    )
+
+
+def _failed(evaluation_id: str, error: str, report: PublishReport | None) -> PublicationOutcome:
+    return PublicationOutcome(
+        status=PlatformJobStatus.ERROR,
+        evaluation_id=evaluation_id,
+        trial_count=report.trial_count if report is not None else 0,
+        evaluator_result_count=report.evaluator_result_count if report is not None else 0,
+        skipped=_skipped_lines(report) if report is not None else [],
+        error=error,
+    )
+
+
+async def _publish(
+    result: AgentEvalResult,
+    *,
+    platform: AsyncNeMoPlatform,
+    spec: IntakePublicationSpec,
+    workspace: str,
+    agent_name: str,
+    model_name: str | None,
+) -> PublishReport:
+    """Check the Evaluation exists, then publish under it."""
+    # The Evaluation must pre-exist — ATIF ingest rejects an unknown one per trial, so without this
+    # a typo would surface as N failed writes after a partial publish instead of one clear stop.
+    # This reads the entity store, so it says nothing about whether Intake's span storage is up;
+    # that surfaces on the first ingest below, and re-publish is idempotent, so it needs no probe.
+    await platform.evaluations.retrieve(spec.evaluation_id, workspace=workspace)
+    return await publish_to_intake(
+        result,
+        platform=platform,
+        experiment_id=spec.evaluation_id,
+        workspace=workspace,
+        agent_name=agent_name,
+        agent_version=spec.agent_version,
+        model_name=model_name,
+    )
+
+
+def publish_agent_eval_result(
+    result: AgentEvalResult,
+    *,
+    spec: IntakePublicationSpec,
+    target: Target | None,
+    workspace: str,
+    async_sdk: AsyncNeMoPlatform | None,
+) -> PublicationOutcome:
+    """Publish a finished run to Intake and describe what happened.
+
+    Raises :class:`PublicationFailedError` when publication fails and ``spec.required`` is set;
+    otherwise returns a failed outcome for the job output. Either way the result bundle has already
+    been saved by the caller, so nothing is lost — a failure costs a re-publish, not a re-run.
+    """
+    derived_agent_name, model_name = target_agent_identity(target)
+    # Spec validation guarantees one of these is set (see `_require_resolvable_publication_identity`).
+    agent_name = spec.agent_name or derived_agent_name or ""
+
+    def fail(error: str, report: PublishReport | None = None) -> PublicationOutcome:
+        outcome = _failed(spec.evaluation_id, error, report)
+        if spec.required:
+            raise PublicationFailedError(outcome)
+        logger.warning(
+            "Publication to Intake failed for evaluation %r but was not required; continuing: %s",
+            spec.evaluation_id,
+            error,
+        )
+        return outcome
+
+    if async_sdk is None:
+        return fail("No platform client available to publish with (platformless local run).")
+
+    logger.info(
+        "Publishing %d trial(s) to Intake under evaluation %r in workspace %r",
+        len(result.trials),
+        spec.evaluation_id,
+        workspace,
+    )
+    try:
+        report = run_sync(
+            lambda: _publish(
+                result,
+                platform=async_sdk,
+                spec=spec,
+                workspace=workspace,
+                agent_name=agent_name,
+                model_name=model_name,
+            )
+        )
+    except PublishError as error:
+        return fail(str(error), error.report)
+    except NotFoundError:
+        return fail(
+            f"Evaluation {spec.evaluation_id!r} does not exist in workspace {workspace!r}. "
+            "Create it before submitting the job; the evaluation does not create it."
+        )
+    except NeMoPlatformError as error:
+        return fail(f"{type(error).__name__}: {error}")
+    except Exception as error:
+        # `required=False` promises the evaluation survives a failed publish. Letting an unforeseen
+        # error escape would break that promise for exactly the failures nobody anticipated, so the
+        # catch-all is the point rather than an oversight. Logged with a traceback because, unlike
+        # the handlers above, there is no known cause to report.
+        logger.exception("Unexpected error publishing to Intake for evaluation %r", spec.evaluation_id)
+        return fail(f"Unexpected {type(error).__name__}: {error}")
+
+    logger.info(
+        "Published %d trial(s) and %d evaluator result(s) to Intake under evaluation %r",
+        report.trial_count,
+        report.evaluator_result_count,
+        spec.evaluation_id,
+    )
+    return _completed(spec.evaluation_id, report)
diff --git a/plugins/nemo-evaluator/tests/intake/test_mapping.py b/plugins/nemo-evaluator/tests/intake/test_mapping.py
index 4e5cb70e19..2abd1e15f3 100644
--- a/plugins/nemo-evaluator/tests/intake/test_mapping.py
+++ b/plugins/nemo-evaluator/tests/intake/test_mapping.py
@@ -6,6 +6,7 @@
 from __future__ import annotations
 
 import math
+from datetime import UTC, datetime
 
 import pytest
 from nemo_evaluator.intake.mapping import (
@@ -32,6 +33,8 @@
 )
 from nemo_platform.types.intake.evaluator_result_create_params import EvaluatorResultCreateParams
 
+STARTED_AT = datetime(2026, 1, 2, 3, 4, 5, tzinfo=UTC)
+
 
 def _trial(*, trial_id: str = "trial-1", task_id: str = "task-1", output_text: str | None = "hello") -> AgentEvalTrial:
     output = AgentOutput(output_text=output_text) if output_text is not None else None
@@ -89,25 +92,28 @@ def test_trial_to_atif_ingest_shape() -> None:
         run_id="run-1",
         experiment_id="exp-1",
         agent_name="my-agent",
+        started_at=STARTED_AT,
         model_name="gpt-4o",
     )
     assert body["schema_version"] == ATIF_SCHEMA_VERSION
     assert body["session_id"] == "run-1:t-1"
     assert body["agent"] == {"name": "my-agent", "version": DEFAULT_AGENT_VERSION, "model_name": "gpt-4o"}
-    assert body["steps"] == [{"source": "agent", "step_id": 1, "message": "final answer"}]
+    assert body["steps"] == [{"source": "agent", "step_id": 1, "message": "final answer", "timestamp": STARTED_AT}]
     assert body["evaluation_context"] == {"evaluation_id": "exp-1", "test_case_id": "task-1"}
     assert "final_metrics" not in body
 
 
 def test_trial_to_atif_ingest_defaults_version_and_omits_model_name() -> None:
-    body = trial_to_atif_ingest(_trial(), run_id="run-1", experiment_id="exp-1", agent_name="a")
+    body = trial_to_atif_ingest(_trial(), run_id="run-1", experiment_id="exp-1", agent_name="a", started_at=STARTED_AT)
     assert body["agent"] == {"name": "a", "version": "unknown"}
     assert "model_name" not in body["agent"]
 
 
 def test_trial_to_atif_ingest_handles_missing_output() -> None:
-    body = trial_to_atif_ingest(_trial(output_text=None), run_id="run-1", experiment_id="exp-1", agent_name="a")
-    assert body["steps"] == [{"source": "agent", "step_id": 1, "message": ""}]
+    body = trial_to_atif_ingest(
+        _trial(output_text=None), run_id="run-1", experiment_id="exp-1", agent_name="a", started_at=STARTED_AT
+    )
+    assert body["steps"] == [{"source": "agent", "step_id": 1, "message": "", "timestamp": STARTED_AT}]
 
 
 def test_trial_to_atif_ingest_includes_final_metrics_when_given() -> None:
@@ -116,6 +122,7 @@ def test_trial_to_atif_ingest_includes_final_metrics_when_given() -> None:
         run_id="run-1",
         experiment_id="exp-1",
         agent_name="a",
+        started_at=STARTED_AT,
         final_metrics={"total_prompt_tokens": 10},
     )
     assert body["final_metrics"] == {"total_prompt_tokens": 10}
diff --git a/plugins/nemo-evaluator/tests/intake/test_publish.py b/plugins/nemo-evaluator/tests/intake/test_publish.py
index bc616359b5..3090ebd4e9 100644
--- a/plugins/nemo-evaluator/tests/intake/test_publish.py
+++ b/plugins/nemo-evaluator/tests/intake/test_publish.py
@@ -7,12 +7,13 @@
 
 import math
 from collections.abc import AsyncIterator
+from datetime import UTC, datetime
 from types import SimpleNamespace
 from typing import Any, cast
 
 import pytest
 from nemo_evaluator.intake.publish import PublishError, publish_to_intake
-from nemo_evaluator_sdk.agent_eval.results import AgentEvalResult, AgentEvalSummary
+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
@@ -113,6 +114,9 @@ def _score(
     )
 
 
+STARTED_AT = datetime(2026, 1, 2, 3, 4, 5, tzinfo=UTC)
+
+
 def _result(trials: list[AgentEvalTrial], scores: list[AgentEvalTaskScore]) -> AgentEvalResult:
     return AgentEvalResult(
         run_id="run-1",
@@ -120,6 +124,7 @@ def _result(trials: list[AgentEvalTrial], scores: list[AgentEvalTaskScore]) -> A
         trials=trials,
         scores=scores,
         summary=AgentEvalSummary(),
+        metadata=RunMetadata(started_at=STARTED_AT),
     )
 
 
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 17bb02350c..cc191341dd 100644
--- a/plugins/nemo-evaluator/tests/integration/test_publish_to_intake.py
+++ b/plugins/nemo-evaluator/tests/integration/test_publish_to_intake.py
@@ -12,7 +12,10 @@
 
     uv run pytest plugins/nemo-evaluator/tests/integration/test_publish_to_intake.py -v
 
-Requires Docker (Intake is ClickHouse-backed) and a free :8080 / :8123.
+Requires Docker (Intake is ClickHouse-backed) and a free :8123. The platform binds the port from
+``NMP_BASE_URL`` (default :8080), so set it to run alongside a local dev platform::
+
+    NMP_BASE_URL=http://localhost:8096 uv run pytest ...
 """
 
 from __future__ import annotations
@@ -24,12 +27,14 @@
 import time
 import urllib.request
 from collections.abc import Iterator
+from datetime import UTC, datetime, timedelta
 from importlib.util import find_spec
 from pathlib import Path
+from urllib.parse import urlsplit
 
 import pytest
-from nemo_evaluator.intake.publish import publish_to_intake
-from nemo_evaluator_sdk.agent_eval.results import AgentEvalResult, AgentEvalSummary
+from nemo_evaluator.intake.publish import PublishReport, publish_to_intake
+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
@@ -46,6 +51,11 @@
 RUN_ID = "intake-it-run"
 NAN_EXPERIMENT_NAME = "intake-it-nan-exp"
 NAN_RUN_ID = "intake-it-nan-run"
+IDEMPOTENCY_EXPERIMENT_NAME = "intake-it-idempotency-exp"
+IDEMPOTENCY_RUN_ID = "intake-it-idempotency-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)
 
 
 def _docker_available() -> bool:
@@ -118,8 +128,13 @@ def _clickhouse(tmp_path_factory: pytest.TempPathFactory) -> Iterator[None]:
 
 @pytest.fixture(scope="session")
 def platform_base_url(_clickhouse: None) -> Iterator[str]:
+    # Bind the port from BASE_URL rather than letting `services run` fall back to its 8080 default:
+    # NMP_BASE_URL is client-side only, so without this the suite silently requires 8080 to be free
+    # and cannot run alongside a local dev platform. Mirrors the sibling fixtures in conftest, which
+    # each take their own port for the same reason.
+    port = urlsplit(BASE_URL).port or 8080
     process = subprocess.Popen(
-        ["uv", "run", "nemo", "services", "run", "--services", "auth,entities,intake"],
+        ["uv", "run", "nemo", "services", "run", "--services", "auth,entities,intake", "--port", str(port)],
         cwd=REPO_ROOT,
         env={
             **os.environ,
@@ -182,7 +197,14 @@ def _result() -> AgentEvalResult:
             outputs=[MetricOutput(name="score", value=0.0), MetricOutput(name="passed", value=False)],
         ),
     ]
-    return AgentEvalResult(run_id=RUN_ID, tasks=[], trials=trials, scores=scores, summary=AgentEvalSummary())
+    return AgentEvalResult(
+        run_id=RUN_ID,
+        tasks=[],
+        trials=trials,
+        scores=scores,
+        summary=AgentEvalSummary(),
+        metadata=RunMetadata(started_at=STARTED_AT),
+    )
 
 
 async def test_publish_to_intake_round_trip(platform_base_url: str) -> None:
@@ -280,7 +302,14 @@ def _nan_result() -> AgentEvalResult:
             outputs=[MetricOutput(name="verdict", value=math.nan)],
         ),
     ]
-    return AgentEvalResult(run_id=NAN_RUN_ID, tasks=[], trials=[trial], scores=scores, summary=AgentEvalSummary())
+    return AgentEvalResult(
+        run_id=NAN_RUN_ID,
+        tasks=[],
+        trials=[trial],
+        scores=scores,
+        summary=AgentEvalSummary(),
+        metadata=RunMetadata(started_at=STARTED_AT),
+    )
 
 
 async def test_publish_skips_nan_and_failed_scores(platform_base_url: str) -> None:
@@ -315,3 +344,74 @@ async def test_publish_skips_nan_and_failed_scores(platform_base_url: str) -> No
             ("accuracy.broken", "non-finite value"),
             ("judge.verdict", "scoring failed"),
         }
+
+
+def _idempotency_result() -> AgentEvalResult:
+    """One trial with one score — the smallest result that exercises both write paths."""
+    return AgentEvalResult(
+        run_id=IDEMPOTENCY_RUN_ID,
+        tasks=[],
+        trials=[
+            AgentEvalTrial(
+                id="trial-1",
+                task_id="task-1",
+                status=AgentEvalTrialStatus.COMPLETED,
+                output=AgentOutput(output_text="answer"),
+            )
+        ],
+        scores=[
+            AgentEvalTaskScore(
+                id="score-1",
+                run_id=IDEMPOTENCY_RUN_ID,
+                task_id="task-1",
+                trial_id="trial-1",
+                metric_type="accuracy",
+                status=AgentEvalScoreStatus.COMPLETED,
+                outputs=[MetricOutput(name="score", value=1.0)],
+            )
+        ],
+        summary=AgentEvalSummary(),
+        metadata=RunMetadata(started_at=STARTED_AT),
+    )
+
+
+async def test_republishing_the_same_result_is_idempotent(platform_base_url: str) -> None:
+    # A job worker can publish successfully and die before recording completion, so a retry must not
+    # double-count. Intake's spans table is a ReplacingMergeTree keyed on start_time, which is only
+    # stable because the trajectory carries the run's started_at (see mapping.trial_to_atif_ingest);
+    # without it each publish lands a second, uncollapsible row per trial.
+    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=IDEMPOTENCY_EXPERIMENT_NAME,
+            experiment_ids=[group.id],
+            dataset_name="intake-it-idempotency-dataset",
+            dataset_version="v1",
+            exist_ok=True,
+        )
+
+        async def publish() -> PublishReport:
+            return await publish_to_intake(
+                _idempotency_result(),
+                platform=client,
+                experiment_id=IDEMPOTENCY_EXPERIMENT_NAME,
+                workspace=WORKSPACE,
+                agent_name="intake-it-agent",
+            )
+
+        first = await publish()
+        second = await publish()
+
+        # Same identities both times — nothing is minted per-publish.
+        assert first.trial_count == second.trial_count == 1
+        assert first.published_trials[0].session_id == second.published_trials[0].session_id
+        assert first.published_trials[0].span_id == second.published_trials[0].span_id
+
+        session_id = second.published_trials[0].session_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 trajectory instead of replacing it"
+
+        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"]
diff --git a/plugins/nemo-evaluator/tests/jobs/test_publication.py b/plugins/nemo-evaluator/tests/jobs/test_publication.py
new file mode 100644
index 0000000000..310aa20eda
--- /dev/null
+++ b/plugins/nemo-evaluator/tests/jobs/test_publication.py
@@ -0,0 +1,489 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Unit tests for the agent-eval job's Intake publication step."""
+
+from __future__ import annotations
+
+import asyncio
+from collections.abc import AsyncIterator, Sequence
+from datetime import UTC, datetime
+from pathlib import Path
+from types import SimpleNamespace
+from typing import Any, cast
+
+import httpx
+import pytest
+from nemo_evaluator.jobs.agent_evaluate import AgentEvalJob
+from nemo_evaluator.jobs.agent_spec import (
+    AgentEvalInputSpec,
+    AgentEvalSpec,
+    AgentEvalTaskInput,
+    AgentEvalTaskSpec,
+    AgentTarget,
+    CodexRunnerTarget,
+    FabricRunnerTarget,
+    HarborRunnerTarget,
+    IntakePublicationSpec,
+    ModelTarget,
+    PublicationSpec,
+    Target,
+    target_agent_identity,
+)
+from nemo_evaluator.jobs.publication import PublicationFailedError, publish_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.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.protocol import MetricOutput
+from nemo_evaluator_sdk.values import Model
+from nemo_evaluator_sdk.values.agents import NemoAgentToolkitAgent
+from nemo_platform import AsyncNeMoPlatform
+from nemo_platform._exceptions import APIConnectionError, NotFoundError
+from nemo_platform_plugin.job_context import JobContext, StoragePaths
+from nemo_platform_plugin.job_results import LocalJobResults
+from nemo_platform_plugin.jobs.schemas import PlatformJobStatus
+from pydantic import ValidationError
+from pytest_mock import MockerFixture
+
+STARTED_AT = datetime(2026, 1, 2, 3, 4, 5, tzinfo=UTC)
+
+#: Aliased because `_FakeTraces` defines a `list` method, which shadows the builtin in class scope.
+_SessionIds = list[str]
+
+
+def _response(status: int) -> httpx.Response:
+    """A real httpx response — the SDK errors read ``response.request`` in their constructor."""
+    return httpx.Response(status, request=httpx.Request("GET", "http://platform/evaluations"))
+
+
+# --- fakes ------------------------------------------------------------------
+
+
+class _FakeTraces:
+    def __init__(self, calls: _SessionIds) -> None:
+        self._calls = calls
+
+    def list(self, *, workspace: str, filter: dict[str, Any]) -> AsyncIterator[object]:  # noqa: A002
+        session_id = filter["session_id"]
+        self._calls.append(session_id)
+
+        async def _gen() -> AsyncIterator[object]:
+            yield SimpleNamespace(session_id=session_id, root_span_id=f"span:{session_id}")
+
+        return _gen()
+
+
+class _FakeEvaluations:
+    def __init__(self, *, missing: bool = False, error: Exception | None = None) -> None:
+        self.missing = missing
+        self.error = error
+        self.retrieved: _SessionIds = []
+
+    async def retrieve(self, name: str, *, workspace: str | None = None) -> object:
+        self.retrieved.append(name)
+        if self.error is not None:
+            raise self.error
+        if self.missing:
+            raise NotFoundError("not found", response=_response(404), body=None)
+        return SimpleNamespace(name=name)
+
+
+class _FakeIngest:
+    def __init__(self, calls: list[dict[str, Any]], *, error: Exception | None = None) -> None:
+        self._calls = calls
+        self._error = error
+        self.loop: asyncio.AbstractEventLoop | None = None
+
+    async def create(self, **kwargs: Any) -> None:
+        self.loop = asyncio.get_running_loop()
+        if self._error is not None:
+            raise self._error
+        self._calls.append(kwargs)
+
+
+class _FakeClient:
+    """Minimal stand-in for the bits of ``AsyncNeMoPlatform`` publication touches."""
+
+    def __init__(
+        self,
+        *,
+        missing_evaluation: bool = False,
+        ingest_error: Exception | None = None,
+        preflight_error: Exception | None = None,
+    ) -> None:
+        self.workspace = "default"
+        self.atif_calls: list[dict[str, Any]] = []
+        self.eval_result_calls: list[dict[str, Any]] = []
+        self.trace_calls: _SessionIds = []
+        self.evaluations = _FakeEvaluations(missing=missing_evaluation, error=preflight_error)
+        self.intake = SimpleNamespace(
+            ingest=SimpleNamespace(atif=_FakeIngest(self.atif_calls, error=ingest_error)),
+            evaluator_results=_FakeIngest(self.eval_result_calls),
+            traces=_FakeTraces(self.trace_calls),
+        )
+
+
+def _client(**kwargs: Any) -> AsyncNeMoPlatform:
+    return cast(AsyncNeMoPlatform, _FakeClient(**kwargs))
+
+
+def _result() -> AgentEvalResult:
+    return AgentEvalResult(
+        run_id="run-1",
+        tasks=[],
+        trials=[
+            AgentEvalTrial(
+                id="trial-1",
+                task_id="task-1",
+                status=AgentEvalTrialStatus.COMPLETED,
+                output=AgentOutput(output_text="answer"),
+            )
+        ],
+        scores=[
+            AgentEvalTaskScore(
+                id="score-1",
+                run_id="run-1",
+                task_id="task-1",
+                trial_id="trial-1",
+                metric_type="accuracy",
+                status=AgentEvalScoreStatus.COMPLETED,
+                outputs=[MetricOutput(name="score", value=1.0)],
+            )
+        ],
+        summary=AgentEvalSummary(),
+        metadata=RunMetadata(started_at=STARTED_AT),
+    )
+
+
+def _publish(client: AsyncNeMoPlatform | None, *, required: bool = True, agent_name: str | None = "a") -> Any:
+    return publish_agent_eval_result(
+        _result(),
+        spec=IntakePublicationSpec(evaluation_id="eval-1", agent_name=agent_name, required=required),
+        target=None,
+        workspace="default",
+        async_sdk=client,
+    )
+
+
+# --- identity resolution ----------------------------------------------------
+
+
+@pytest.mark.parametrize(
+    ("target", "expected"),
+    [
+        (AgentTarget(agent=NemoAgentToolkitAgent(name="my-agent", url="http://agent")), ("my-agent", None)),
+        (ModelTarget(model=Model(name="gpt-4o", url="http://model")), (None, "gpt-4o")),
+        (HarborRunnerTarget(agent_name="oracle", agent_model_name="m"), ("oracle", "m")),
+        (HarborRunnerTarget(agent_name="oracle", agent_import_path="pkg:Agent"), ("pkg:Agent", None)),
+        (CodexRunnerTarget(model="gpt-5.5"), (None, "gpt-5.5")),
+        (FabricRunnerTarget(config={}, model="p/m"), (None, "p/m")),
+        (None, (None, None)),
+    ],
+)
+def test_target_agent_identity(target: Target | None, expected: tuple[str | None, str | None]) -> None:
+    assert target_agent_identity(target) == expected
+
+
+# --- submit-time validation -------------------------------------------------
+
+
+def _input_spec(target: Target | None, publication: PublicationSpec | None, **kwargs: Any) -> AgentEvalInputSpec:
+    trials = None if target is not None else [_result().trials[0]]
+    return AgentEvalInputSpec(
+        tasks=[AgentEvalTaskInput(id="task-1", intent="do it")],
+        target=target,
+        trials=trials,
+        publication=publication,
+        **kwargs,
+    )
+
+
+def test_publication_is_optional() -> None:
+    spec = _input_spec(ModelTarget(model=Model(name="m", url="http://m")), None)
+    assert spec.publication is None
+
+
+def test_publication_defaults_to_required() -> None:
+    spec = _input_spec(
+        AgentTarget(agent=NemoAgentToolkitAgent(name="a", url="http://a")),
+        PublicationSpec(intake=IntakePublicationSpec(evaluation_id="eval-1")),
+    )
+    assert spec.publication is not None
+    assert spec.publication.intake is not None
+    assert spec.publication.intake.required is True
+
+
+def test_agent_name_derived_from_agent_target_needs_no_override() -> None:
+    spec = _input_spec(
+        AgentTarget(agent=NemoAgentToolkitAgent(name="derived", url="http://a")),
+        PublicationSpec(intake=IntakePublicationSpec(evaluation_id="eval-1")),
+    )
+    assert spec.publication is not None
+    assert spec.publication.intake is not None
+    assert spec.publication.intake.agent_name is None
+    assert target_agent_identity(spec.target)[0] == "derived"
+
+
+@pytest.mark.parametrize(
+    "target",
+    [
+        ModelTarget(model=Model(name="gpt-4o", url="http://model")),
+        CodexRunnerTarget(model="gpt-5.5"),
+        FabricRunnerTarget(config={}),
+        None,
+    ],
+)
+def test_undeducible_agent_name_is_rejected_at_submit(target: Target | None) -> None:
+    with pytest.raises(ValidationError, match="agent_name` is required"):
+        _input_spec(target, PublicationSpec(intake=IntakePublicationSpec(evaluation_id="eval-1")))
+
+
+def test_blank_identity_fields_are_rejected() -> None:
+    # An empty `agent_name` satisfies `is not None` in the identity validator, so it would skip the
+    # derivation that validator exists to require and resolve back to "" at publish time.
+    with pytest.raises(ValidationError):
+        IntakePublicationSpec(evaluation_id="eval-1", agent_name="")
+    with pytest.raises(ValidationError):
+        IntakePublicationSpec(evaluation_id="eval-1", agent_version="")
+
+
+@pytest.mark.parametrize(
+    "target",
+    [ModelTarget(model=Model(name="gpt-4o", url="http://model")), CodexRunnerTarget(model="gpt-5.5"), None],
+)
+def test_explicit_agent_name_satisfies_undeducible_targets(target: Target | None) -> None:
+    spec = _input_spec(
+        target, PublicationSpec(intake=IntakePublicationSpec(evaluation_id="eval-1", agent_name="explicit"))
+    )
+    assert spec.publication is not None
+
+
+# --- publishing -------------------------------------------------------------
+
+
+def test_publishes_and_reports_what_landed() -> None:
+    client = _FakeClient()
+    outcome = _publish(cast(AsyncNeMoPlatform, client))
+
+    assert outcome.status == PlatformJobStatus.COMPLETED
+    assert outcome.evaluation_id == "eval-1"
+    assert outcome.trial_count == 1
+    assert outcome.evaluator_result_count == 1
+    assert outcome.error is None
+    assert client.evaluations.retrieved == ["eval-1"]
+
+
+def test_outcome_does_not_leak_experiment_id() -> None:
+    outcome = _publish(_client())
+    assert "experiment_id" not in outcome.model_dump()
+
+
+def test_stamps_the_run_start_time_so_republish_is_idempotent() -> None:
+    client = _FakeClient()
+    _publish(cast(AsyncNeMoPlatform, client))
+    assert client.atif_calls[0]["steps"][0]["timestamp"] == STARTED_AT
+
+
+def test_missing_evaluation_fails_before_any_ingest() -> None:
+    client = _FakeClient(missing_evaluation=True)
+    with pytest.raises(PublicationFailedError) as excinfo:
+        _publish(cast(AsyncNeMoPlatform, client))
+
+    assert client.atif_calls == []
+    outcome = excinfo.value.outcome
+    assert outcome.status == PlatformJobStatus.ERROR
+    assert "does not exist" in (outcome.error or "")
+
+
+def test_required_failure_raises_with_partial_outcome() -> None:
+    client = _FakeClient(ingest_error=APIConnectionError(request=httpx.Request("POST", "http://platform/intake")))
+    with pytest.raises(PublicationFailedError) as excinfo:
+        _publish(cast(AsyncNeMoPlatform, client))
+
+    outcome = excinfo.value.outcome
+    assert outcome.status == PlatformJobStatus.ERROR
+    assert outcome.trial_count == 0
+
+
+def test_optional_failure_returns_outcome_instead_of_raising() -> None:
+    client = _FakeClient(ingest_error=APIConnectionError(request=httpx.Request("POST", "http://platform/intake")))
+    outcome = _publish(cast(AsyncNeMoPlatform, client), required=False)
+
+    assert outcome.status == PlatformJobStatus.ERROR
+    assert outcome.error
+
+
+def test_unexpected_failure_still_honours_required_false() -> None:
+    # `required=False` promises the evaluation survives a failed publish. An error outside the known
+    # taxonomy — a bug, a transport quirk, anything unforeseen — must not be the one case that
+    # escapes and fails the job anyway.
+    client = _FakeClient(preflight_error=ValueError("something nobody planned for"))
+    outcome = _publish(cast(AsyncNeMoPlatform, client), required=False)
+
+    assert outcome.status == PlatformJobStatus.ERROR
+    assert "ValueError" in (outcome.error or "")
+    assert "something nobody planned for" in (outcome.error or "")
+
+
+def test_unexpected_failure_fails_the_job_when_required() -> None:
+    client = _FakeClient(preflight_error=ValueError("something nobody planned for"))
+    with pytest.raises(PublicationFailedError) as excinfo:
+        _publish(cast(AsyncNeMoPlatform, client))
+
+    assert excinfo.value.outcome.status == PlatformJobStatus.ERROR
+
+
+def test_platformless_run_is_a_failure_not_a_crash() -> None:
+    outcome = _publish(None, required=False)
+    assert outcome.status == PlatformJobStatus.ERROR
+    assert "platformless" in (outcome.error or "")
+
+
+def test_platformless_run_fails_the_job_when_required() -> None:
+    with pytest.raises(PublicationFailedError):
+        _publish(None)
+
+
+# --- job wiring -------------------------------------------------------------
+
+
+class _FakeEvaluator:
+    """Stand-in for AgentEvaluator returning one completed trial per task."""
+
+    def __init__(self, *, started_at: datetime | None = STARTED_AT) -> None:
+        self._started_at = started_at
+        self.loop: asyncio.AbstractEventLoop | None = None
+
+    def run_sync(self, *, tasks: Sequence[AgentEvalTask], **kwargs: Any) -> AgentEvalResult:
+        # Drives a real loop to completion the way `AgentEvaluator.run_sync` does, so publication
+        # afterwards runs against an already-closed loop.
+        return run_sync(lambda: self._run(tasks))
+
+    async def _run(self, tasks: Sequence[AgentEvalTask]) -> AgentEvalResult:
+        self.loop = asyncio.get_running_loop()
+        return AgentEvalResult(
+            run_id="run-1",
+            tasks=list(tasks),
+            trials=[
+                AgentEvalTrial(
+                    id=f"{task.id}:trial",
+                    task_id=task.id,
+                    status=AgentEvalTrialStatus.COMPLETED,
+                    output=AgentOutput(output_text="4"),
+                )
+                for task in tasks
+            ],
+            scores=[],
+            summary=AgentEvalSummary(),
+            metadata=RunMetadata(started_at=self._started_at),
+        )
+
+
+def _job_context(tmp_path: Path) -> JobContext:
+    storage = StoragePaths(ephemeral=tmp_path / "ephemeral", persistent=tmp_path / "persistent")
+    storage.ephemeral.mkdir()
+    storage.persistent.mkdir()
+    return JobContext(
+        workspace="default",
+        storage=storage,
+        results=LocalJobResults(root=storage.persistent / "results"),
+    )
+
+
+def _job_spec(*, required: bool = True) -> AgentEvalSpec:
+    return AgentEvalSpec(
+        tasks=[AgentEvalTaskSpec(id="task-1", intent="Answer.")],
+        target=CodexRunnerTarget(model="gpt-5.5"),
+        publication=PublicationSpec(
+            intake=IntakePublicationSpec(evaluation_id="eval-1", agent_name="a", required=required)
+        ),
+    )
+
+
+def test_job_does_not_publish_without_a_publication_spec(tmp_path: Path, mocker: MockerFixture) -> None:
+    mocker.patch.object(AgentEvalJob, "_build_evaluator", return_value=_FakeEvaluator())
+    client = _FakeClient()
+
+    spec = AgentEvalSpec(tasks=[AgentEvalTaskSpec(id="task-1", intent="Answer.")], target=CodexRunnerTarget())
+    result = AgentEvalJob().run(
+        spec.model_dump(), ctx=_job_context(tmp_path), async_sdk=cast(AsyncNeMoPlatform, client)
+    )
+
+    assert "publication" not in result
+    assert client.atif_calls == []
+
+
+def test_job_publishes_through_the_real_sync_bridge(tmp_path: Path, mocker: MockerFixture) -> None:
+    evaluator = _FakeEvaluator()
+    mocker.patch.object(AgentEvalJob, "_build_evaluator", return_value=evaluator)
+    client = _FakeClient()
+    ctx = _job_context(tmp_path)
+
+    result = AgentEvalJob().run(_job_spec().model_dump(), ctx=ctx, 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 (cf. nmp-1hr.2). 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["status"] == PlatformJobStatus.COMPLETED
+    assert result["publication"] == {
+        "status": PlatformJobStatus.COMPLETED,
+        "evaluation_id": "eval-1",
+        "trial_count": 1,
+        "evaluator_result_count": 0,
+        "skipped": [],
+    }
+    assert len(client.atif_calls) == 1
+
+
+def test_job_keeps_the_bundle_when_required_publication_fails(tmp_path: Path, mocker: MockerFixture) -> None:
+    # Publication is the only step that can fail the job, so it runs last: the bundle and summary
+    # artifacts must survive for a later re-publish.
+    mocker.patch.object(AgentEvalJob, "_build_evaluator", return_value=_FakeEvaluator())
+    client = _FakeClient(missing_evaluation=True)
+    ctx = _job_context(tmp_path)
+
+    with pytest.raises(PublicationFailedError):
+        AgentEvalJob().run(_job_spec().model_dump(), ctx=ctx, async_sdk=cast(AsyncNeMoPlatform, client))
+
+    assert (ctx.storage.persistent / "agent-eval" / "trials.jsonl").exists()
+    assert (ctx.storage.persistent / "results" / "agent-eval-results").exists()
+
+
+def test_job_completes_when_optional_publication_fails(tmp_path: Path, mocker: MockerFixture) -> None:
+    mocker.patch.object(AgentEvalJob, "_build_evaluator", return_value=_FakeEvaluator())
+    client = _FakeClient(missing_evaluation=True)
+
+    result = AgentEvalJob().run(
+        _job_spec(required=False).model_dump(),
+        ctx=_job_context(tmp_path),
+        async_sdk=cast(AsyncNeMoPlatform, client),
+    )
+
+    assert result["status"] == PlatformJobStatus.COMPLETED
+    assert result["publication"]["status"] == PlatformJobStatus.ERROR
+    assert "does not exist" in result["publication"]["error"]
+
+
+def test_job_publication_requires_a_run_start_time(tmp_path: Path, mocker: MockerFixture) -> None:
+    # Without `started_at` the trajectory would fall back to Intake's per-request ingest clock, and
+    # re-publishing would duplicate spans instead of replacing them. Refuse rather than write rows
+    # that can never be collapsed.
+    mocker.patch.object(AgentEvalJob, "_build_evaluator", return_value=_FakeEvaluator(started_at=None))
+
+    result = AgentEvalJob().run(
+        _job_spec(required=False).model_dump(),
+        ctx=_job_context(tmp_path),
+        async_sdk=cast(AsyncNeMoPlatform, _FakeClient()),
+    )
+
+    assert result["publication"]["status"] == PlatformJobStatus.ERROR
+    assert "started_at" in result["publication"]["error"]
diff --git a/web/packages/studio/src/api/evaluation/agent-evaluations.ts b/web/packages/studio/src/api/evaluation/agent-evaluations.ts
index 11bbf46942..b26fef404f 100644
--- a/web/packages/studio/src/api/evaluation/agent-evaluations.ts
+++ b/web/packages/studio/src/api/evaluation/agent-evaluations.ts
@@ -12,6 +12,7 @@ import {
 import type {
   AggregateRangeScore,
   AggregateRubricScore,
+  AggregateScalarScore,
   AgentEvaluateJob,
   AgentEvaluateJobRequest,
   AgentEvalResult,
@@ -22,8 +23,11 @@ import { filesDownloadFile } from '@nemo/sdk/generated/platform/api';
 
 const PAGE_SIZE = 50;
 
-/** Aggregate score — numeric range or rubric category distribution. */
-export type AgentEvalAggregateScore = AggregateRangeScore | AggregateRubricScore;
+/** Aggregate score — numeric range, rubric category distribution, or a single reported value. */
+export type AgentEvalAggregateScore =
+  | AggregateRangeScore
+  | AggregateRubricScore
+  | AggregateScalarScore;
 
 /** Re-export so callers continue to import AgentEvalResult from this module. */
 export type { AgentEvalResult };