feat(evaluator): write both eval and publish durations to intake

plugins/nemo-evaluator/src/nemo_evaluator/jobs/publication.py

@@ -19,7 +19,8 @@
 from __future__ import annotations
 
 import logging
-from datetime import datetime
+import time
+from datetime import UTC, datetime
 
 from nemo_evaluator.intake.publish import PublishError, PublishReport, publish_to_intake
 from nemo_evaluator.intake.row_adapter import RowIdentityError, row_result_to_agent_eval_result
@@ -33,11 +34,18 @@
 from nemo_evaluator_sdk.values.results import EvaluationResult
 from nemo_platform import AsyncNeMoPlatform
 from nemo_platform._exceptions import NeMoPlatformError, NotFoundError
+from nemo_platform.types.evaluations import EvaluationResponse
 from nemo_platform_plugin.jobs.schemas import PlatformJobStatus
 from pydantic import BaseModel, ConfigDict, Field
 
 logger = logging.getLogger(__name__)
 
+#: Evaluation metadata keys carrying how long the run took and how long publishing it took, both as
+#: string-encoded seconds (the API takes ``dict[str, str]``). Two keys rather than one ``duration``
+#: because a reader cannot tell which of the two a single key means.
+EVAL_DURATION_KEY = "eval_duration_sec"
+PUBLISH_DURATION_KEY = "publish_duration_sec"
+
 
 class PublicationOutcome(BaseModel):
     """What publication did, as reported in the job output.
@@ -99,6 +107,40 @@ def _failed(evaluation_id: str, error: str, report: PublishReport | None) -> Pub
     )
 
 
+def _eval_duration_sec(result: AgentEvalResult) -> float | None:
+    """Wall-clock seconds the run itself took, or ``None`` when it cannot be known."""
+    if result.metadata.duration_sec is not None:
+        return result.metadata.duration_sec
+    if result.metadata.started_at is not None:
+        # A row-eval result carries only a start time (see ``intake.row_adapter``), so its length is
+        # measured here instead. Call this before publishing, not after, or the run duration swallows
+        # the publish it is meant to be reported alongside. It still slightly overcounts: it covers
+        # persisting the bundle between the run finishing and this call.
+        return (datetime.now(UTC) - result.metadata.started_at).total_seconds()
+    return None
+
+
+async def _record_durations(
+    platform: AsyncNeMoPlatform,
+    *,
+    evaluation: EvaluationResponse,
+    spec: IntakePublicationSpec,
+    workspace: str,
+    eval_duration_sec: float | None,
+    publish_duration_sec: float,
+) -> None:
+    """Stamp the two durations onto the Evaluation's metadata.
+
+    PATCH replaces the metadata dict wholesale, so the merge with what is already there is
+    mandatory — a blind write would drop producer keys such as ``eval_config_fileset``.
+    """
+    metadata = dict(evaluation.metadata or {})
+    if eval_duration_sec is not None:
+        metadata[EVAL_DURATION_KEY] = f"{eval_duration_sec:.1f}"
+    metadata[PUBLISH_DURATION_KEY] = f"{publish_duration_sec:.1f}"
+    await platform.evaluations.patch(spec.evaluation_id, workspace=workspace, metadata=metadata)
+
+
 async def _publish(
     result: AgentEvalResult,
     *,
@@ -108,13 +150,17 @@ async def _publish(
     agent_name: str,
     model_name: str | None,
 ) -> PublishReport:
-    """Check the Evaluation exists, then publish under it."""
+    """Check the Evaluation exists, publish under it, then record how long both took."""
+    # Measured before the publish so the two durations stay disjoint — for a result that carries only
+    # a start time, reading this afterwards would fold the whole publish into the run's own length.
+    eval_duration_sec = _eval_duration_sec(result)
     # 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(
+    evaluation = await platform.evaluations.retrieve(spec.evaluation_id, workspace=workspace)
+    publish_started = time.monotonic()
+    report = await publish_to_intake(
         result,
         platform=platform,
         experiment_id=spec.evaluation_id,
@@ -123,6 +169,26 @@ async def _publish(
         agent_version=spec.agent_version,
         model_name=model_name,
     )
+    publish_duration_sec = time.monotonic() - publish_started
+    try:
+        await _record_durations(
+            platform,
+            evaluation=evaluation,
+            spec=spec,
+            workspace=workspace,
+            eval_duration_sec=eval_duration_sec,
+            publish_duration_sec=publish_duration_sec,
+        )
+    except Exception:
+        # The durations are informational, and the publish already succeeded. Every caller-side
+        # handler turns an exception here into a failed job (and a raise when `required`), so letting
+        # one escape would report a successful publish as a failure.
+        logger.warning(
+            "Published to Intake but could not record durations on evaluation %r",
+            spec.evaluation_id,
+            exc_info=True,
+        )
+    return report
 
 
 def publish_agent_eval_result(

plugins/nemo-evaluator/tests/jobs/test_publication.py

@@ -32,7 +32,12 @@
     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 import (
+    EVAL_DURATION_KEY,
+    PUBLISH_DURATION_KEY,
+    PublicationFailedError,
+    publish_agent_eval_result,
+)
 from nemo_evaluator.jobs.publication_spec import (
     IntakePublicationSpec,
     PublicationSpec,
@@ -96,28 +101,46 @@ async def _gen() -> AsyncIterator[object]:
 
 
 class _FakeEvaluations:
-    def __init__(self, *, missing: bool = False, error: Exception | None = None) -> None:
+    def __init__(
+        self,
+        *,
+        missing: bool = False,
+        error: Exception | None = None,
+        patch_error: Exception | None = None,
+    ) -> None:
         self.missing = missing
         self.error = error
+        self.patch_error = patch_error
         self.retrieved: _SessionIds = []
+        self.patched: list[dict[str, Any]] = []
+        self.metadata: dict[str, str] = {"eval_config_fileset": "fs-1"}
 
     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, metadata=dict(self.metadata))
+
+    async def patch(self, name: str, *, workspace: str | None = None, **kwargs: Any) -> object:
+        if self.patch_error is not None:
+            raise self.patch_error
+        self.patched.append({"name": name, **kwargs})
         return SimpleNamespace(name=name)
 
 
 class _FakeIngest:
-    def __init__(self, calls: list[dict[str, Any]], *, error: Exception | None = None) -> None:
+    def __init__(self, calls: list[dict[str, Any]], *, error: Exception | None = None, delay_sec: float = 0.0) -> None:
         self._calls = calls
         self._error = error
+        self._delay_sec = delay_sec
         self.loop: asyncio.AbstractEventLoop | None = None
 
     async def create(self, **kwargs: Any) -> None:
         self.loop = asyncio.get_running_loop()
+        if self._delay_sec:
+            await asyncio.sleep(self._delay_sec)
         if self._error is not None:
             raise self._error
         self._calls.append(kwargs)
@@ -132,15 +155,17 @@ def __init__(
         missing_evaluation: bool = False,
         ingest_error: Exception | None = None,
         preflight_error: Exception | None = None,
+        patch_error: Exception | None = None,
+        ingest_delay_sec: float = 0.0,
     ) -> None:
         self.workspace = "default"
         self.atif_calls: list[dict[str, Any]] = []
         self.eval_result_calls: list[dict[str, Any]] = []
         self.trace_calls: _SessionIds = []
         self.copy_calls = 0
-        self.evaluations = _FakeEvaluations(missing=missing_evaluation, error=preflight_error)
+        self.evaluations = _FakeEvaluations(missing=missing_evaluation, error=preflight_error, patch_error=patch_error)
         self.intake = SimpleNamespace(
-            ingest=SimpleNamespace(atif=_FakeIngest(self.atif_calls, error=ingest_error)),
+            ingest=SimpleNamespace(atif=_FakeIngest(self.atif_calls, error=ingest_error, delay_sec=ingest_delay_sec)),
             evaluator_results=_FakeIngest(self.eval_result_calls),
             traces=_FakeTraces(self.trace_calls),
         )
@@ -317,6 +342,52 @@ def test_publishes_and_reports_what_landed() -> None:
     assert client.evaluations.retrieved == ["eval-1"]
 
 
+def test_durations_are_stamped_without_dropping_existing_metadata() -> None:
+    client = _FakeClient()
+    _publish(cast(AsyncNeMoPlatform, client))
+
+    (patched,) = client.evaluations.patched
+    metadata = patched["metadata"]
+    # PATCH replaces the metadata dict wholesale, so stamping the durations has to merge with what
+    # the producer already wrote. A blind write would silently drop `eval_config_fileset`.
+    assert metadata["eval_config_fileset"] == "fs-1"
+    assert float(metadata[EVAL_DURATION_KEY]) > 0
+    assert float(metadata[PUBLISH_DURATION_KEY]) >= 0
+
+
+def test_a_started_at_only_result_does_not_count_publish_time_as_run_time() -> None:
+    # A row-eval result carries no `duration_sec`, so the run's length is derived from `started_at`.
+    # Deriving it after the publish instead of before would fold the publish into the run.
+    publish_sec = 0.3
+    client = _FakeClient(ingest_delay_sec=publish_sec)
+    result = _result()
+    result.metadata.started_at = datetime.now(UTC)
+
+    publish_agent_eval_result(
+        result,
+        spec=IntakePublicationSpec(evaluation_id="eval-1", agent_name="a", required=True),
+        target=None,
+        workspace="default",
+        async_sdk=cast(AsyncNeMoPlatform, client),
+    )
+
+    (patched,) = client.evaluations.patched
+    metadata = patched["metadata"]
+    assert float(metadata[PUBLISH_DURATION_KEY]) >= publish_sec
+    assert float(metadata[EVAL_DURATION_KEY]) < publish_sec / 2
+
+
+def test_a_failed_duration_stamp_does_not_fail_the_publish() -> None:
+    client = _FakeClient(patch_error=APIConnectionError(request=httpx.Request("PATCH", "http://x")))
+    outcome = _publish(cast(AsyncNeMoPlatform, client), required=True)
+
+    # The durations are informational and the trials already landed, so losing them must not turn a
+    # successful publish into a failed job — which is what every caller-side handler would do.
+    assert outcome.status == PlatformJobStatus.COMPLETED
+    assert outcome.trial_count == 1
+    assert outcome.error is None
+
+
 def test_outcome_does_not_leak_experiment_id() -> None:
     outcome = _publish(_client())
     assert "experiment_id" not in outcome.model_dump()

web/packages/studio/src/api/evaluation/utils.ts

@@ -74,6 +74,20 @@ export const EVAL_JOB_KIND_LABEL: Record<EvalJobKind, string> = {
   dataset: 'Dataset-Driven',
 };
 
+/** Metadata key the evaluator stamps the run's wall-clock seconds under, at publish time. */
+export const EVAL_DURATION_METADATA_KEY = 'eval_duration_sec';
+
+/** How long the published run took, in milliseconds, or undefined when it was never recorded.
+ *  Metadata values are strings server-side, so a non-numeric one reads the same as an absent one. */
+export const evalDurationMs = (metadata?: Record<string, string> | null): number | undefined => {
+  const raw = metadata?.[EVAL_DURATION_METADATA_KEY];
+  // Metadata is free-form and hand-editable, so treat anything that is not a non-negative number as
+  // absent. `Number('')` is 0, which would otherwise render as a confident "0ms".
+  if (raw == null || raw.trim() === '') return undefined;
+  const seconds = Number(raw);
+  return Number.isFinite(seconds) && seconds >= 0 ? seconds * 1000 : undefined;
+};
+
 export const publishedEvaluationName = (job: PlatformJobResponse): string | null => {
   const intake = asRecord(asRecord(specOf(job).publication)?.intake);
   return asNonEmptyString(intake?.evaluation_id) ?? null;

web/packages/studio/src/components/dataViews/ExperimentDataView/index.tsx

@@ -21,6 +21,7 @@ import { formatDurationMs } from '@nemo/common/src/utils/date';
 import { formatEvaluatorScore, snakeCaseToTitleCase } from '@nemo/common/src/utils/formatters';
 import type { EvaluationFilter, ExperimentResponse } from '@nemo/sdk/generated/platform/schema';
 import { Button, Text, Tooltip } from '@nvidia/foundations-react-core';
+import { EVAL_DURATION_METADATA_KEY, evalDurationMs } from '@studio/api/evaluation/utils';
 import { ChangesetBadge } from '@studio/components/ChangesetBadge';
 import { ExperimentParetoChart } from '@studio/components/charts/ExperimentParetoChart';
 import { AddToGroupModal } from '@studio/components/dataViews/ExperimentDataView/AddToGroupModal';
@@ -230,13 +231,17 @@ export const ExperimentDataView: FC<ExperimentDataViewProps> = ({ group, paretoV
 
   // One column per metadata key: keys are lowercased so case variants (e.g. "status"
   // and "Status") collapse into one column rather than producing duplicate headers.
+  // The run duration is excluded because the Duration column below already renders it, formatted;
+  // publish duration is deliberately left in, which is what surfaces publish latency at all.
   const metadataKeys = useMemo(
     () =>
       [
         ...new Set(
           orderedData.flatMap((e) => Object.keys(e.metadata ?? {}).map((k) => k.toLowerCase()))
         ),
-      ].sort(),
+      ]
+        .filter((key) => key !== EVAL_DURATION_METADATA_KEY)
+        .sort(),
     [orderedData]
   );
 
@@ -481,6 +486,15 @@ export const ExperimentDataView: FC<ExperimentDataViewProps> = ({ group, paretoV
         enableSorting: true,
         cell: ({ row }) => <Text>{String(row.original.test_case_count ?? 0)}</Text>,
       }),
+      // Written as metadata at publish time, so it is a string and only present for runs that
+      // published successfully. Not sortable: the API sorts metadata lexically, which would order
+      // "9" after "10", and the list pages at 100 so a client-side sort would lie across pages.
+      accessor((original) => evalDurationMs(original.metadata), {
+        id: 'eval_duration',
+        header: 'Duration',
+        enableSorting: false,
+        cell: ({ getValue }) => <Text>{formatDurationMs(getValue<number | undefined>())}</Text>,
+      }),
       accessor('created_at', {
         header: 'Created',
         size: 200,

web/packages/studio/src/components/evaluation/Jobs/DetailsPanel.tsx

@@ -9,18 +9,16 @@ import { StatusBadge } from '@nemo/common/src/components/StatusBadge';
 import { PlatformJobTerminalStatuses } from '@nemo/common/src/constants/query';
 import { useLiveSeconds } from '@nemo/common/src/hooks/useLiveSeconds';
 import { useToast } from '@nemo/common/src/providers/toast/useToast';
-import {
-  formatTimeInSeconds,
-  getDifferenceInMilliseconds,
-  utcToLocalDate,
-} from '@nemo/common/src/utils/date';
+import { formatDurationMs, formatTimeInSeconds, utcToLocalDate } from '@nemo/common/src/utils/date';
 import { logger } from '@nemo/common/src/utils/logger';
 import {
   getEvaluatorGetEvaluateJobQueryKey,
   useEvaluatorCancelEvaluateJob,
 } from '@nemo/sdk/generated/evaluator/api';
 import type { EvaluateJob } from '@nemo/sdk/generated/evaluator/schema';
+import { useGetEvaluation } from '@nemo/sdk/generated/platform/api';
 import { Banner, Button, Flex, Modal, Panel, Stack, Text } from '@nvidia/foundations-react-core';
+import { evalDurationMs } from '@studio/api/evaluation/utils';
 import { ButtonLaunchEvaluation } from '@studio/components/evaluation/ButtonLaunchEvaluation';
 import { useWorkspaceFromPath } from '@studio/hooks/useWorkspaceFromPath';
 import { getFilesetRoute } from '@studio/routes/utils';
@@ -64,20 +62,21 @@ export const DetailsPanel = ({ evaluationJob, error }: DetailsPanelProps) => {
     }
   };
 
-  const differenceInMilliseconds = getDifferenceInMilliseconds(
-    evaluationJob?.created_at,
-    evaluationJob?.updated_at
-  );
-  const elapsedSeconds = differenceInMilliseconds
-    ? Math.floor(differenceInMilliseconds / 1000)
-    : undefined;
   const isTerminalStatus =
     evaluationJob?.status && PlatformJobTerminalStatuses.includes(evaluationJob.status);
   const canCancelJob = evaluationJob?.status && !isTerminalStatus;
   const liveSeconds = useLiveSeconds({
     startDate: !isTerminalStatus ? utcToLocalDate(evaluationJob?.created_at) : undefined,
   });
 
+  // A finished job has no end time of its own — `updated_at` is stamped at create and on rerun,
+  // never on a status change — so the run's length comes from the evaluation it published under.
+  const publishedEvaluation = evaluationJob?.spec.publication?.intake?.evaluation_id;
+  const { data: evaluation } = useGetEvaluation(workspace, publishedEvaluation ?? '', {
+    query: { enabled: !!workspace && !!publishedEvaluation && !!isTerminalStatus },
+  });
+  const durationMs = evalDurationMs(evaluation?.metadata);
+
   if (error || !evaluationJob) {
     return (
       <ErrorMessage
@@ -165,9 +164,9 @@ export const DetailsPanel = ({ evaluationJob, error }: DetailsPanelProps) => {
               status ? (
                 <Flex align="center" gap="2">
                   <StatusBadge status={status} />
-                  {formatTimeInSeconds(
-                    PlatformJobTerminalStatuses.includes(status) ? elapsedSeconds : liveSeconds
-                  )}
+                  {PlatformJobTerminalStatuses.includes(status)
+                    ? durationMs !== undefined && formatDurationMs(durationMs)
+                    : formatTimeInSeconds(liveSeconds)}
                 </Flex>
               ) : (
                 <Text kind="body/semibold/sm">Detail not available</Text>