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.
+* `--auto-telemetry`: Let the server fill in the agent's telemetry export -- an Intake destination for a config that asks for one and does not say where. False submits the agent config as written, which still exports if the config says to; an agent config is the place to say a run should not be traced.
 * `--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)
 

packages/nemo_platform/pyproject.toml

@@ -395,6 +395,7 @@ nemo-insights-plugin = [
   "nemo-fabric-adapter-contract>=0.3.0b1",
   "nemo-fabric-adapters-common>=0.3.0b1",
   "nemo-platform-plugin",
+  "nemo-relay>=0.7,<0.8",
   "nooa>=0.0.9",
   "opentelemetry-exporter-otlp>=1.42.1",
   "opentelemetry-sdk>=1.42.1",

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

@@ -48,7 +48,16 @@
     materialize_agent_workdir,
     validate_agent_workdir,
 )
+from nemo_agents_plugin.telemetry.intake_export import (
+    configure_intake_atif_export,
+    supports_intake_atif_export,
+)
 from nemo_platform import AsyncNeMoPlatform, NeMoPlatform
+from nemo_platform_plugin.client.constants import (
+    WORKLOAD_IDENTITY_TOKEN_FILE_ENVVAR,
+    is_workload_identity_token_file_set,
+)
+from nemo_platform_plugin.client.oidc_factory import resolve_workload_exchange_provider
 from nemo_platform_plugin.entity_client import NemoEntityNotFoundError
 from nemo_platform_plugin.job import NemoJob
 from nemo_platform_plugin.job_context import JobContext
@@ -80,6 +89,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 +99,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 +202,15 @@ class ExecuteAgentJobConfig(BaseModel):
         gt=0,
         description="Maximum time to wait for Fabric to return an execution result.",
     )
+    auto_telemetry: bool = Field(
+        default=True,
+        description=(
+            "Let the server fill in the agent's telemetry export -- an Intake destination for a "
+            "config that asks for one and does not say where. False submits the agent config as "
+            "written, which still exports if the config says to; an agent config is the place to "
+            "say a run should not be traced."
+        ),
+    )
     extension: ExecuteAgentExtensionConfig | None = Field(
         default=None,
         description="Optional trusted plugin extension to run during the execute-agent lifecycle.",
@@ -396,6 +417,14 @@ def run(self, config: dict, *, ctx: JobContext, sdk: NeMoPlatform | None = None)
 
         fabric_dirs = FabricDirectories.create(agent_config, ctx.storage.ephemeral)
 
+        if step_config.request.auto_telemetry and supports_intake_atif_export(
+            step_config.agent.config, base_dir=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 +812,77 @@ def _validate_agent_config_format(config_format: str) -> None:
         )
 
 
+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 {}
+    headers.update(_workload_identity_headers(base_url))
+    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 _workload_identity_headers(base_url: str) -> dict[str, str]:
+    """Bearer credentials for Relay when the job runs under workload identity.
+
+    ``get_forwarding_headers`` returns only what the SDK was *constructed* with.
+    Under workload identity that is the internal marker alone -- the bearer is
+    exchanged per request by the SDK's own auth layer, which Relay's raw POST to
+    Intake does not go through. Without this the export would be unauthenticated
+    on exactly the deployments that enforce auth.
+
+    The token is resolved once and read from the environment at export time, so
+    a run outliving its token exports with an expired one. Relay resolves
+    ``header_env`` statically, so refreshing needs a dynamic-credential hook it
+    does not offer today.
+    """
+    if not is_workload_identity_token_file_set():
+        return {}
+    token_file = os.environ[WORKLOAD_IDENTITY_TOKEN_FILE_ENVVAR]
+    try:
+        provider = resolve_workload_exchange_provider(base_url=base_url, subject_token_file=Path(token_file))
+        return {"Authorization": f"Bearer {provider.get_access_token()}"}
+    except Exception:
+        logger.warning(
+            "Could not exchange the workload identity token for telemetry export; "
+            "the trajectory will be posted without credentials.",
+            exc_info=True,
+        )
+        return {}
+
+
+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,10 @@
     FabricArtifactStagingError,
     stage_fabric_ethos_config_files,
 )
+from nemo_agents_plugin.telemetry.intake_export import (
+    configure_intake_atif_export,
+    supports_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 +672,15 @@ 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.
+            if supports_intake_atif_export(config):
+                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)