feat: Auto wire agent telemetry

docs/cli/reference.mdx

@@ -6828,6 +6828,7 @@ nemo agents execute [OPTIONS]
 * `--environment`: AgentEnvironment to run under: a "workspace/name" ref to a stored AgentEnvironment, an inline environment, or None. Its EnvironmentSpec is merged into the agent config and its ComputeSpec/secret refs are snapshotted onto the job step at creation time. This flag accepts the string form only; use --spec or --spec-file for the other union form(s).
 * `--workdir.base-workdir`: Optional Files reference for the initial working directory.
 * `--timeout-seconds <FLOAT>`: Maximum time to wait for Fabric to return an execution result.
+* `--telemetry`: Export the agent's trajectory to Intake. Set false to run untraced, or configure 'telemetry' on the agent yourself — an agent that already declares it is left alone.
 * `--extension.kind`: Trusted extension kind registered by an installed NeMo plugin.
 
 **Spec Source:**

e2e/test_insights_analysis_run.py

@@ -20,6 +20,7 @@
 
 import json
 import re
+import time
 from typing import Any
 
 import pytest
@@ -223,6 +224,17 @@ def _download_job_result(sdk: NeMoPlatform, workspace: str, job_name: str, resul
     return response.text
 
 
+def _wait_for_spans(sdk: NeMoPlatform, *, workspace: str, agent_name: str, timeout: float = 120.0) -> list[Any]:
+    """Poll Intake: Relay posts as the run ends and ingest is asynchronous."""
+    deadline = time.monotonic() + timeout
+    while True:
+        page = sdk.intake.spans.list(workspace=workspace, filter={"agent_name": agent_name}, page_size=50)
+        spans = list(page.data or [])
+        if spans or time.monotonic() >= deadline:
+            return spans
+        time.sleep(2.0)
+
+
 def _created_insight_id(report: str) -> str:
     """Pull the stored insight id out of the report's change log.
 
@@ -284,6 +296,12 @@ def test_analysis_run_persists_insights_and_saves_its_report(sdk: NeMoPlatform,
     # The extension persisted the change-set as a real Insight. The report's
     # change log is what says which one: it is written from what the Insights
     # API returned, so its id only exists if the write landed.
+    # Relay carries the Analyst's own trajectory to Intake. Nothing in this
+    # test configures telemetry: the agents plugin wires the export, Fabric
+    # resolves it, and the adapter activates it.
+    spans = _wait_for_spans(sdk, workspace=workspace, agent_name="insights-analyst")
+    assert spans, "the Analyst ran but its trajectory never reached Intake"
+
     insight_id = _created_insight_id(report)
     filed = sdk.insights.insights.get(workspace=workspace, insight_id=insight_id)
     assert filed.title == INSIGHT_TITLE

e2e/test_nemo_agents_execute_job.py

@@ -8,6 +8,7 @@
 import io
 import json
 import tarfile
+import time
 from typing import Any
 
 import pytest
@@ -67,6 +68,29 @@ def _result_names(results: dict[str, Any]) -> set[str]:
     return {str(result["name"]) for result in results.get("data", [])}
 
 
+def _wait_for_agent_spans(
+    sdk: NeMoPlatform,
+    *,
+    workspace: str,
+    agent_name: str,
+    timeout: float = 120.0,
+    poll_interval: float = 2.0,
+) -> list[Any]:
+    """Poll Intake for the agent's trajectory.
+
+    Ingest is asynchronous: Relay posts the trajectory as the run finishes, and
+    Intake writes it behind the API, so a job reporting ``completed`` does not
+    mean the spans are queryable yet.
+    """
+    deadline = time.monotonic() + timeout
+    while True:
+        page = sdk.intake.spans.list(workspace=workspace, filter={"agent_name": agent_name}, page_size=50)
+        spans = list(page.data or [])
+        if spans or time.monotonic() >= deadline:
+            return spans
+        time.sleep(poll_interval)
+
+
 def _tar_member_names(content: bytes) -> set[str]:
     with tarfile.open(fileobj=io.BytesIO(content), mode="r:gz") as tar:
         return {member.name for member in tar.getmembers()}
@@ -258,6 +282,12 @@ def test_fabric_agent_invocation_job_runs_and_saves_results(sdk: NeMoPlatform, w
         assert "write_file" in run_result_json
         assert "generated-report.md" in run_result_json
         assert TEST_AGENT_RESPONSE in run_result_json
+
+        # Nobody configured an export: the job wires the agent's trajectory to
+        # this workspace's Intake, using the platform URL reachable from the
+        # task pod and the identity the platform gave the job.
+        spans = _wait_for_agent_spans(sdk, workspace=workspace, agent_name=agent_name)
+        assert spans, "the agent ran but no trajectory reached Intake"
     finally:
         delete_agent_if_exists(sdk, workspace=workspace, name=agent_name)
 

plugins/nemo-agents/src/nemo_agents_plugin/agent_config.py

@@ -70,7 +70,11 @@ class RuntimeConfig(BaseModel):
 class TelemetryConfig(BaseModel):
     model_config = ConfigDict(extra="forbid")
 
-    enabled: bool = False
+    # Tri-state so a config can decline telemetry without being mistaken for one
+    # that never mentioned it: unset lets the backend wire an export for this
+    # deployment context, False opts out, True turns it on and still lets the
+    # backend fill in whatever the config left out.
+    enabled: bool | None = None
     provider: str | None = None
     output_dir: str | None = None
     project: str | None = None

plugins/nemo-agents/src/nemo_agents_plugin/jobs/execute.py

@@ -37,6 +37,7 @@
     invoke_agent_config_request_once,
 )
 from nemo_agents_plugin.fabric.runtime import FabricRuntimeTimeoutError
+from nemo_agents_plugin.fabric.translator import translate_agent_config
 from nemo_agents_plugin.jobs.execute_extensions import (
     NOOP_EXECUTE_AGENT_EXTENSION_KIND,
     ExecuteAgentAfterInvokeContext,
@@ -48,6 +49,8 @@
     materialize_agent_workdir,
     validate_agent_workdir,
 )
+from nemo_agents_plugin.telemetry.intake_export import configure_intake_atif_export
+from nemo_fabric import Fabric
 from nemo_platform import AsyncNeMoPlatform, NeMoPlatform
 from nemo_platform_plugin.entity_client import NemoEntityNotFoundError
 from nemo_platform_plugin.job import NemoJob
@@ -80,6 +83,7 @@
 from nemo_platform_plugin.jobs.exceptions import PlatformJobCompilationError
 from nemo_platform_plugin.jobs.image import get_qualified_image
 from nemo_platform_plugin.refs import ENTITY_REF_PATTERN, parse_entity_ref
+from nemo_platform_plugin.sdk_provider import get_forwarding_headers
 from pydantic import BaseModel, Field, field_validator
 
 logger = logging.getLogger(__name__)
@@ -89,6 +93,8 @@
 INPUT_WORKDIR_RESULT_NAME = "input_workdir"
 OUTPUT_WORKDIR_RESULT_NAME = "output_workdir"
 OUTPUT_ARTIFACTS_RESULT_NAME = "output_artifacts"
+NMP_BASE_URL_ENVVAR = "NMP_BASE_URL"
+HEADER_ENVVAR_PREFIX = "NMP_AGENT_TELEMETRY_HEADER_"
 FABRIC_RUN_RESULT_NAME = "fabric_run_result"
 FABRIC_ERROR_RESULT_NAME = "fabric_error"
 FABRIC_RUN_RESULT_FILENAME = "fabric_run_result.json"
@@ -190,6 +196,13 @@ class ExecuteAgentJobConfig(BaseModel):
         gt=0,
         description="Maximum time to wait for Fabric to return an execution result.",
     )
+    telemetry: bool = Field(
+        default=True,
+        description=(
+            "Export the agent's trajectory to Intake. Set false to run untraced, or configure "
+            "'telemetry' on the agent yourself — an agent that already declares it is left alone."
+        ),
+    )
     extension: ExecuteAgentExtensionConfig | None = Field(
         default=None,
         description="Optional trusted plugin extension to run during the execute-agent lifecycle.",
@@ -396,6 +409,12 @@ def run(self, config: dict, *, ctx: JobContext, sdk: NeMoPlatform | None = None)
 
         fabric_dirs = FabricDirectories.create(agent_config, ctx.storage.ephemeral)
 
+        if step_config.request.telemetry and _adapter_supports_relay(agent_config, fabric_dirs.base):
+            _configure_intake_telemetry(step_config.agent.config, workspace=ctx.workspace, sdk=sdk)
+            # Wiring mutates the config mapping, not the model validated above,
+            # so re-validate to carry it into what Fabric is handed.
+            agent_config = _validate_agent_config(step_config.agent.config)
+
         if step_config.workdir is not None and _has_workdir_inputs(step_config.workdir):
             if sdk is None:
                 raise RuntimeError("sdk is required to stage workdir inputs.")
@@ -783,6 +802,72 @@ def _validate_agent_config_format(config_format: str) -> None:
         )
 
 
+def _adapter_supports_relay(agent_config: AgentConfig, base_dir: Path) -> bool:
+    """Whether the agent's adapter declares that Relay can instrument it.
+
+    Adapters advertise this in their descriptor's ``telemetry.providers``; the
+    four bundled harnesses declare ``relay``, and an adapter that does not is
+    rejected outright by Fabric for configuring one. Auto-wiring has to ask
+    first, or it turns "this agent cannot be traced" into "this agent cannot
+    run".
+    """
+    try:
+        plan = Fabric().plan(translate_agent_config(agent_config), base_dir=base_dir)
+        descriptor = plan.to_dict().get("adapter_descriptor") or {}
+        providers = descriptor.get("descriptor", descriptor).get("telemetry", {}).get("providers", {})
+        supported = "relay" in providers
+    except Exception:
+        # Planning failures are the invocation's to report, with its own
+        # diagnostics; here they only mean we cannot know, so do not wire.
+        logger.warning("Could not read adapter telemetry support; the agent will run untraced.", exc_info=True)
+        return False
+
+    if not supported:
+        logger.info("Adapter does not support Relay telemetry; the agent will run untraced.")
+    return supported
+
+
+def _configure_intake_telemetry(
+    agent_config: dict[str, Any],
+    *,
+    workspace: str,
+    sdk: NeMoPlatform | None,
+) -> None:
+    """Wire the agent's trajectory export to Intake for this job.
+
+    Runs here rather than at create time because only the task knows both
+    halves: ``NMP_BASE_URL`` is the platform URL reachable from *this* pod (the
+    Jobs service rewrites it per runtime), and the task's own SDK carries the
+    identity the platform gave this job -- the same ``service:agents`` principal
+    and on-behalf-of delegation a deployment gets from its auth-proxy sidecar.
+
+    Credentials go in the process environment and the config names them.
+    Fabric writes the resolved agent config into the run's artifacts, and those
+    are uploaded as a job result, so an inline header would be a downloadable
+    one.
+    """
+    base_url = os.environ.get(NMP_BASE_URL_ENVVAR)
+    if not base_url:
+        logger.warning("%s is not set; the agent will run untraced.", NMP_BASE_URL_ENVVAR)
+        return
+
+    headers = get_forwarding_headers(sdk) if sdk is not None else {}
+    for name, value in headers.items():
+        os.environ[_header_envvar(name)] = value
+
+    configure_intake_atif_export(
+        agent_config,
+        workspace=workspace,
+        base_url=base_url,
+        header_env={name: _header_envvar(name) for name in headers},
+    )
+
+
+def _header_envvar(header_name: str) -> str:
+    """Environment variable the exporter reads one outbound header value from."""
+    return f"{HEADER_ENVVAR_PREFIX}{header_name.upper().replace('-', '_')}"
+
+
 def _validate_agent_config(config: dict) -> AgentConfig:
     agent_config = AgentConfig.model_validate(config)
     if agent_config.environment.provider != "local":

plugins/nemo-agents/src/nemo_agents_plugin/runner/deployments_backend.py

@@ -38,6 +38,7 @@
     FabricArtifactStagingError,
     stage_fabric_ethos_config_files,
 )
+from nemo_agents_plugin.telemetry.intake_export import configure_intake_atif_export
 from nemo_agents_plugin.utils import get_base_url, get_internal_base_url
 from nemo_deployments_plugin.auth_proxy import auth_proxy_port
 from nemo_deployments_plugin.entities import (
@@ -668,6 +669,14 @@ async def create_deployment(
             rewrite_target = gateway
 
         if is_fabric:
+            # Wire the trajectory export before the rebase below, so the
+            # endpoint it produces is rewritten for reachability alongside the
+            # inference URLs. No header_env here: the auth-proxy sidecar stamps
+            # identity on the way out, which is why the workload needs none.
+            # config is the caller's deployment entity; rewrite_fabric_config_base_urls
+            # deep-copies for the same reason, and wiring runs before it.
+            config = copy.deepcopy(config)
+            configure_intake_atif_export(config, workspace=workspace, base_url=rewrite_target)
             config = rewrite_fabric_config_base_urls(config, rewrite_target)
         else:
             config = rewrite_config_base_urls(config, rewrite_target)

plugins/nemo-agents/src/nemo_agents_plugin/telemetry/intake_export.py

@@ -0,0 +1,103 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Auto-wire an agent's Relay ATIF export to the platform's Intake ingest.
+
+Filling in an export destination by hand is a poor thing to ask of anyone
+writing an agent config: the reachable platform URL differs per deployment
+context, and the same config should work whether it is deployed or run as a
+job. So the backend wires it, and the config carries at most a name.
+
+The two contexts differ only in how identity reaches Intake. A deployment
+routes through a loopback auth-proxy sidecar that stamps the principal on the
+way out. A job has one creator for its whole life and is handed that principal
+directly, so it names environment variables the exporter reads instead.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from nemo_agents_plugin.agent_config import TelemetryConfig
+from pydantic import ValidationError
+
+logger = logging.getLogger(__name__)
+
+INTAKE_ATIF_INGEST_PATH = "/apis/intake/v2/workspaces/{workspace}/ingest/atif"
+
+
+def configure_intake_atif_export(
+    config: dict[str, Any],
+    *,
+    workspace: str,
+    base_url: str,
+    header_env: dict[str, str] | None = None,
+) -> bool:
+    """Point *config*'s ATIF export at *workspace*'s Intake ingest.
+
+    Mutates *config* in place. Returns whether telemetry was wired.
+
+    Takes the config as a mapping rather than an ``AgentConfig`` because the
+    deployments path reaches this point with a config already resolved for its
+    runtime, whose harnesses no longer round-trip through the model. The
+    telemetry section does round-trip, so it is manipulated as a
+    :class:`TelemetryConfig` rather than by poking at keys.
+
+    ``telemetry.enabled`` is tri-state: unset means "wire it for me", ``False``
+    is an explicit opt-out, and ``True`` turns it on while still letting the
+    backend fill in anything the config left out. An agent that already names
+    its own ATIF storage keeps it — an explicit destination beats an inferred
+    one.
+
+    Args:
+        config: Agent config to wire, modified in place.
+        workspace: Workspace whose Intake receives the trajectory.
+        base_url: Platform URL reachable from wherever the agent will run.
+        header_env: Header name to environment variable name, for contexts
+            with no auth proxy to stamp identity. The variables must exist in
+            the agent process; the values deliberately never enter the config,
+            which is written into the run's artifacts.
+    """
+    section = config.get("telemetry")
+    try:
+        telemetry = TelemetryConfig.model_validate(section if isinstance(section, dict) else {})
+    except ValidationError as exc:
+        # Leave a section we do not understand exactly as we found it. The jobs
+        # path validates the whole config moments later and will report this
+        # properly; deployments do not, and a telemetry key is no reason to
+        # fail one.
+        logger.warning("Leaving an unrecognized telemetry section unwired: %s", exc)
+        return False
+
+    if telemetry.enabled is False:
+        return False
+    if _declares_atif_storage(telemetry):
+        return False
+
+    storage: dict[str, object] = {
+        "type": "http",
+        "endpoint": f"{base_url.rstrip('/')}{INTAKE_ATIF_INGEST_PATH.format(workspace=workspace)}",
+    }
+    if header_env:
+        storage["header_env"] = dict(header_env)
+
+    atif = dict(telemetry.atif or {})
+    atif["enabled"] = True
+    atif["storage"] = [storage]
+
+    wired = telemetry.model_copy(
+        update={
+            "enabled": True,
+            "provider": telemetry.provider or "relay",
+            "agent_name": telemetry.agent_name or config.get("name"),
+            "atif": atif,
+        }
+    )
+    config["telemetry"] = wired.model_dump(exclude_none=True)
+    return True
+
+
+def _declares_atif_storage(telemetry: TelemetryConfig) -> bool:
+    """Whether the config already names somewhere to send trajectories."""
+    return isinstance(telemetry.atif, dict) and bool(telemetry.atif.get("storage"))

plugins/nemo-agents/tests/unit/test_agent_config.py

@@ -129,7 +129,8 @@ def test_defaults_fill_optional_sections(self) -> None:
         assert config.environment.provider == "local"
         assert config.environment.workspace == "./workspace"
         assert config.environment.artifacts == "./artifacts"
-        assert config.telemetry.enabled is False
+        # Tri-state: unset, so the backend may wire an export for the context.
+        assert config.telemetry.enabled is None
 
     def test_shared_capability_sections_validate(self) -> None:
         payload = _example_yaml_config()