From 347e4bac6e89eddc28b906aca34fcd5a67c9cbf1 Mon Sep 17 00:00:00 2001 From: Mike Knepper Date: Fri, 4 Sep 2026 16:11:54 -0500 Subject: [PATCH 01/15] feat(agents): auto-wire Intake telemetry for agent jobs and deployments Signed-off-by: Mike Knepper --- .../src/nemo_agents_plugin/agent_config.py | 6 +- .../src/nemo_agents_plugin/jobs/execute.py | 54 +++++++++ .../runner/deployments_backend.py | 6 + .../telemetry/intake_export.py | 103 +++++++++++++++++ .../tests/unit/test_agent_config.py | 3 +- .../tests/unit/test_execute_job.py | 105 ++++++++++++++++++ 6 files changed, 275 insertions(+), 2 deletions(-) create mode 100644 plugins/nemo-agents/src/nemo_agents_plugin/telemetry/intake_export.py diff --git a/plugins/nemo-agents/src/nemo_agents_plugin/agent_config.py b/plugins/nemo-agents/src/nemo_agents_plugin/agent_config.py index 531a144888..add403e614 100644 --- a/plugins/nemo-agents/src/nemo_agents_plugin/agent_config.py +++ b/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 diff --git a/plugins/nemo-agents/src/nemo_agents_plugin/jobs/execute.py b/plugins/nemo-agents/src/nemo_agents_plugin/jobs/execute.py index e1b9f12ed6..1b891a76a9 100644 --- a/plugins/nemo-agents/src/nemo_agents_plugin/jobs/execute.py +++ b/plugins/nemo-agents/src/nemo_agents_plugin/jobs/execute.py @@ -48,6 +48,7 @@ materialize_agent_workdir, validate_agent_workdir, ) +from nemo_agents_plugin.telemetry.intake_export import configure_intake_atif_export from nemo_platform import AsyncNeMoPlatform, NeMoPlatform from nemo_platform_plugin.entity_client import NemoEntityNotFoundError from nemo_platform_plugin.job import NemoJob @@ -80,6 +81,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 +91,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 +194,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.", @@ -392,6 +403,8 @@ def run(self, config: dict, *, ctx: JobContext, sdk: NeMoPlatform | None = None) logger.info("Executing agent %s (timeout %gs).", agent_ref, step_config.request.timeout_seconds) _validate_agent_config_format(step_config.agent.config_format) + if step_config.request.telemetry: + _configure_intake_telemetry(step_config.agent.config, workspace=ctx.workspace, sdk=sdk) agent_config = _validate_agent_config(step_config.agent.config) fabric_dirs = FabricDirectories.create(agent_config, ctx.storage.ephemeral) @@ -783,6 +796,47 @@ 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 {} + 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": diff --git a/plugins/nemo-agents/src/nemo_agents_plugin/runner/deployments_backend.py b/plugins/nemo-agents/src/nemo_agents_plugin/runner/deployments_backend.py index 40f091871b..09513073a8 100644 --- a/plugins/nemo-agents/src/nemo_agents_plugin/runner/deployments_backend.py +++ b/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,11 @@ 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. + 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) diff --git a/plugins/nemo-agents/src/nemo_agents_plugin/telemetry/intake_export.py b/plugins/nemo-agents/src/nemo_agents_plugin/telemetry/intake_export.py new file mode 100644 index 0000000000..1d51078217 --- /dev/null +++ b/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")) diff --git a/plugins/nemo-agents/tests/unit/test_agent_config.py b/plugins/nemo-agents/tests/unit/test_agent_config.py index f992fdc1e3..dbb39f8c44 100644 --- a/plugins/nemo-agents/tests/unit/test_agent_config.py +++ b/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() diff --git a/plugins/nemo-agents/tests/unit/test_execute_job.py b/plugins/nemo-agents/tests/unit/test_execute_job.py index 2459610b36..0a3e522287 100644 --- a/plugins/nemo-agents/tests/unit/test_execute_job.py +++ b/plugins/nemo-agents/tests/unit/test_execute_job.py @@ -14,6 +14,7 @@ import pytest from fastapi import FastAPI from fastapi.testclient import TestClient +from nemo_agents_plugin.agent_config import AgentConfig from nemo_agents_plugin.entities import ( Agent, AgentComputeSpec, @@ -40,6 +41,7 @@ ExecuteAgentStepConfig, ResolvedAgentConfig, _log_agent_stderr, + _configure_intake_telemetry, ) from nemo_agents_plugin.tasks.execute.workdir import ( AgentWorkdir, @@ -48,6 +50,7 @@ materialize_agent_workdir, validate_agent_workdir, ) +from nemo_platform import NeMoPlatform from nemo_platform_plugin.dependencies import get_entity_client, get_sdk_client from nemo_platform_plugin.entity_client import NemoEntityNotFoundError from nemo_platform_plugin.job_context import JobContext @@ -1236,6 +1239,7 @@ async def _create_job(*, workspace: str, body: object) -> MagicMock: "environment": None, "workdir": {"base_workdir": "source#project", "artifact_mounts": []}, "timeout_seconds": DEFAULT_AGENT_EXECUTION_TIMEOUT_SECONDS, + "telemetry": True, "extension": None, } assert body.spec["workdir"] == {"base_workdir": "default/source#project/", "artifact_mounts": []} @@ -1789,3 +1793,104 @@ def test_log_agent_stderr_refuses_to_follow_a_symlink(tmp_path: Path, caplog: py assert "Could not read agent stderr" in caplog.text assert "some diagnostics" not in caplog.text + + +# --------------------------------------------------------------------------- +# Intake telemetry auto-configuration +# --------------------------------------------------------------------------- + + +def _fabric_agent_config(**overrides: Any) -> dict[str, Any]: + return { + "config_format": "nemo-agents-spec-v1", + "name": "demo-agent", + "default_harness": "h", + "harnesses": {"h": {"kind": "hermes"}}, + "models": {"default": {"provider": "platform", "model": "default/m"}}, + "environment": {"provider": "local"}, + **overrides, + } + + +def test_telemetry_is_pointed_at_the_workspace_intake_ingest(monkeypatch: pytest.MonkeyPatch) -> None: + """Only the task knows the platform URL reachable from its own pod.""" + monkeypatch.setenv("NMP_BASE_URL", "http://nemo-platform-api:8080") + config = _fabric_agent_config() + + _configure_intake_telemetry(config, workspace="team-a", sdk=None) + + telemetry = config["telemetry"] + assert telemetry["enabled"] is True + assert telemetry["provider"] == "relay" + assert telemetry["agent_name"] == "demo-agent" + storage = telemetry["atif"]["storage"][0] + assert storage["type"] == "http" + assert storage["endpoint"] == "http://nemo-platform-api:8080/apis/intake/v2/workspaces/team-a/ingest/atif" + # The wired config still has to be a valid agent config. + assert AgentConfig.model_validate(config).telemetry.enabled is True + + +def test_telemetry_credentials_go_to_the_environment_not_the_config(monkeypatch: pytest.MonkeyPatch) -> None: + """Fabric writes the config into artifacts that are uploaded as a job result.""" + monkeypatch.setenv("NMP_BASE_URL", "http://nemo-platform-api:8080") + sdk = cast(NeMoPlatform, SimpleNamespace(_custom_headers={"X-NMP-Principal-Id": "service:agents"})) + config = _fabric_agent_config() + + _configure_intake_telemetry(config, workspace="default", sdk=sdk) + + storage = config["telemetry"]["atif"]["storage"][0] + assert storage["header_env"] == {"X-NMP-Principal-Id": "NMP_AGENT_TELEMETRY_HEADER_X_NMP_PRINCIPAL_ID"} + assert "headers" not in storage, "an inline header would land in a downloadable artifact" + assert os.environ["NMP_AGENT_TELEMETRY_HEADER_X_NMP_PRINCIPAL_ID"] == "service:agents" + + +def test_an_agent_that_names_its_own_destination_keeps_it(monkeypatch: pytest.MonkeyPatch) -> None: + """An explicit export destination beats an inferred one.""" + monkeypatch.setenv("NMP_BASE_URL", "http://nemo-platform-api:8080") + mine = {"type": "http", "endpoint": "https://elsewhere.example/ingest"} + config = _fabric_agent_config(telemetry={"enabled": True, "atif": {"enabled": True, "storage": [mine]}}) + + _configure_intake_telemetry(config, workspace="default", sdk=None) + + assert config["telemetry"]["atif"]["storage"] == [mine] + + +def test_telemetry_disabled_on_the_agent_is_an_opt_out(monkeypatch: pytest.MonkeyPatch) -> None: + """False means no, as distinct from a config that never mentioned telemetry.""" + monkeypatch.setenv("NMP_BASE_URL", "http://nemo-platform-api:8080") + config = _fabric_agent_config(telemetry={"enabled": False}) + + _configure_intake_telemetry(config, workspace="default", sdk=None) + + assert config["telemetry"] == {"enabled": False} + + +def test_an_agent_that_only_names_itself_is_still_wired(monkeypatch: pytest.MonkeyPatch) -> None: + """The point of the tri-state: naming yourself is not configuring an export.""" + monkeypatch.setenv("NMP_BASE_URL", "http://nemo-platform-api:8080") + config = _fabric_agent_config(telemetry={"agent_name": "my-agent-name"}) + + _configure_intake_telemetry(config, workspace="default", sdk=None) + + assert config["telemetry"]["enabled"] is True + assert config["telemetry"]["agent_name"] == "my-agent-name" + + +def test_telemetry_is_skipped_when_no_platform_url_is_reachable(monkeypatch: pytest.MonkeyPatch) -> None: + """A run untraced beats a run that fails over its own tracing.""" + monkeypatch.delenv("NMP_BASE_URL", raising=False) + config = _fabric_agent_config() + + _configure_intake_telemetry(config, workspace="default", sdk=None) + + assert "telemetry" not in config + + +def test_an_unrecognized_telemetry_section_is_left_alone(monkeypatch: pytest.MonkeyPatch) -> None: + """Deployments never model-validate, so a stray key must not fail the whole config.""" + monkeypatch.setenv("NMP_BASE_URL", "http://nemo-platform-api:8080") + config = _fabric_agent_config(telemetry={"enabled": True, "not_a_real_field": 1}) + + _configure_intake_telemetry(config, workspace="default", sdk=None) + + assert config["telemetry"] == {"enabled": True, "not_a_real_field": 1} From c3c38e6cdc86e1b36fc9410e29701f3375519791 Mon Sep 17 00:00:00 2001 From: Mike Knepper Date: Tue, 8 Sep 2026 09:46:27 -0500 Subject: [PATCH 02/15] test(agents): cover auto-wired Intake telemetry at both tiers Two gaps the unit tests around configure_intake_atif_export cannot close. The translator test puts the wired config through the real translator, so the dict has to be a shape Fabric accepts rather than a plausible one: it must parse into RelayHttpStorageConfig, keep header_env, and leave headers empty. The shared fixture opts out of telemetry, which under the tri-state now means something, so the test drops that key to describe a config that does not. The e2e asserts the point of the feature: a job nobody configured an export for still lands its trajectory in this workspace's Intake. It polls, because ingest is asynchronous -- a job reporting completed does not mean the spans are queryable yet. Co-Authored-By: Claude Opus 5 (1M context) Signed-off-by: Mike Knepper --- e2e/test_nemo_agents_execute_job.py | 30 +++++++++++++++ .../tests/unit/test_fabric_translator.py | 37 +++++++++++++++++++ 2 files changed, 67 insertions(+) diff --git a/e2e/test_nemo_agents_execute_job.py b/e2e/test_nemo_agents_execute_job.py index 285ffdbbf2..f786f49cc2 100644 --- a/e2e/test_nemo_agents_execute_job.py +++ b/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) diff --git a/plugins/nemo-agents/tests/unit/test_fabric_translator.py b/plugins/nemo-agents/tests/unit/test_fabric_translator.py index cd4730d1e4..7c6733d2f3 100644 --- a/plugins/nemo-agents/tests/unit/test_fabric_translator.py +++ b/plugins/nemo-agents/tests/unit/test_fabric_translator.py @@ -13,6 +13,8 @@ from nemo_agents_plugin.agent_config import AgentConfig, load_agent_config from nemo_agents_plugin.fabric.gateway_credentials import PLATFORM_IGW_API_KEY_ENV, PLATFORM_IGW_API_KEY_PLACEHOLDER from nemo_agents_plugin.fabric.translator import FabricTranslationError, translate_agent_config +from nemo_agents_plugin.telemetry.intake_export import configure_intake_atif_export +from nemo_fabric.models import RelayHttpStorageConfig def _example_yaml_config() -> dict[str, Any]: @@ -592,3 +594,38 @@ def test_relay_opentelemetry_export_translates_to_fabric(self) -> None: assert endpoint.resource_attributes == {"deployment.environment": "test"} assert endpoint.service_name == "example-agent" assert opentelemetry.endpoints[1].service_name == "shared-agent-service" + + +def test_auto_wired_intake_telemetry_translates_to_a_relay_http_storage() -> None: + """The wired dict has to be a shape Fabric accepts, not merely a plausible one. + + The unit tests around ``configure_intake_atif_export`` assert its output + structurally; this puts that output through the real translator so a field + Relay does not recognise fails here rather than inside a running job. + """ + payload = _example_yaml_config() + # The shared fixture opts out; this is about a config that does not. + payload.pop("telemetry") + configure_intake_atif_export( + payload, + workspace="team-a", + base_url="http://nemo-platform-api:8080", + header_env={"X-NMP-Principal-Id": "NMP_AGENT_TELEMETRY_HEADER_X_NMP_PRINCIPAL_ID"}, + ) + + fabric_config = translate_agent_config(AgentConfig.model_validate(payload)) + + relay = fabric_config.relay + assert relay is not None + observability = relay.observability + assert observability is not None + atif = observability.atif + assert atif is not None + storage = atif.storage[0] + assert isinstance(storage, RelayHttpStorageConfig) + assert storage.endpoint == "http://nemo-platform-api:8080/apis/intake/v2/workspaces/team-a/ingest/atif" + assert storage.header_env == {"X-NMP-Principal-Id": "NMP_AGENT_TELEMETRY_HEADER_X_NMP_PRINCIPAL_ID"} + assert not storage.headers, "credentials must reach Relay through the environment, not the config" + # Relay identifies the trajectory by these; both come from the agent config. + assert atif.agent_name == payload["name"] + assert atif.model_name From a4487df8ea843db1edf76c749ec8b3a39cacc55e Mon Sep 17 00:00:00 2001 From: Mike Knepper Date: Tue, 8 Sep 2026 10:53:16 -0500 Subject: [PATCH 03/15] lint Signed-off-by: Mike Knepper --- docs/cli/reference.mdx | 1 + plugins/nemo-agents/openapi/openapi.yaml | 7 +++++++ 2 files changed, 8 insertions(+) diff --git a/docs/cli/reference.mdx b/docs/cli/reference.mdx index ea6696b9b1..778b0174b2 100644 --- a/docs/cli/reference.mdx +++ b/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 `: 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:** diff --git a/plugins/nemo-agents/openapi/openapi.yaml b/plugins/nemo-agents/openapi/openapi.yaml index ecdc8e31c2..13df7d7a3a 100644 --- a/plugins/nemo-agents/openapi/openapi.yaml +++ b/plugins/nemo-agents/openapi/openapi.yaml @@ -5707,6 +5707,13 @@ components: title: Timeout Seconds description: Maximum time to wait for Fabric to return an execution result. default: 3600 + telemetry: + type: boolean + title: Telemetry + description: "Export the agent's trajectory to Intake. Set false to run\ + \ untraced, or configure 'telemetry' on the agent yourself \u2014 an agent\ + \ that already declares it is left alone." + default: true extension: allOf: - $ref: '#/components/schemas/ExecuteAgentExtensionConfig' From b81e2aabdd38020bd6d66ba62a234751365c6d58 Mon Sep 17 00:00:00 2001 From: Mike Knepper Date: Tue, 8 Sep 2026 11:03:42 -0500 Subject: [PATCH 04/15] fix(agents): only wire telemetry for adapters that support Relay Auto-wiring assumed every adapter could be instrumented by Relay. Fabric rejects a relay config outright for one that cannot -- "adapter `nvidia.fabric.insights-analyst` does not support `telemetry.providers` value `relay`" -- so the feature turned "this agent cannot be traced" into "this agent cannot run". The insights e2e caught it; a third-party adapter would have hit the same wall. Adapters advertise support in their descriptor's telemetry.providers block, so read it from the plan before wiring. A plan that fails leaves the agent untraced rather than guessing: the invocation reports that failure moments later with its own diagnostics. Co-Authored-By: Claude Opus 5 (1M context) Signed-off-by: Mike Knepper --- .../src/nemo_agents_plugin/jobs/execute.py | 33 +++++++++++++++++-- .../tests/unit/test_execute_job.py | 29 +++++++++++++++- 2 files changed, 59 insertions(+), 3 deletions(-) diff --git a/plugins/nemo-agents/src/nemo_agents_plugin/jobs/execute.py b/plugins/nemo-agents/src/nemo_agents_plugin/jobs/execute.py index 1b891a76a9..1760151f08 100644 --- a/plugins/nemo-agents/src/nemo_agents_plugin/jobs/execute.py +++ b/plugins/nemo-agents/src/nemo_agents_plugin/jobs/execute.py @@ -16,6 +16,7 @@ from pathlib import Path from typing import Any, ClassVar, NamedTuple, cast +import nemo_fabric as fabric from nemo_agents_plugin.agent_config import AgentConfig from nemo_agents_plugin.agent_config_formats import resolve_agent_config_for_deployment from nemo_agents_plugin.config import AgentsConfig @@ -37,6 +38,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, @@ -403,12 +405,14 @@ def run(self, config: dict, *, ctx: JobContext, sdk: NeMoPlatform | None = None) logger.info("Executing agent %s (timeout %gs).", agent_ref, step_config.request.timeout_seconds) _validate_agent_config_format(step_config.agent.config_format) - if step_config.request.telemetry: - _configure_intake_telemetry(step_config.agent.config, workspace=ctx.workspace, sdk=sdk) agent_config = _validate_agent_config(step_config.agent.config) 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) + 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.") @@ -796,6 +800,31 @@ 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.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], *, diff --git a/plugins/nemo-agents/tests/unit/test_execute_job.py b/plugins/nemo-agents/tests/unit/test_execute_job.py index 0a3e522287..2b7ec90e02 100644 --- a/plugins/nemo-agents/tests/unit/test_execute_job.py +++ b/plugins/nemo-agents/tests/unit/test_execute_job.py @@ -40,8 +40,9 @@ ExecuteAgentJobConfig, ExecuteAgentStepConfig, ResolvedAgentConfig, - _log_agent_stderr, + _adapter_supports_relay, _configure_intake_telemetry, + _log_agent_stderr, ) from nemo_agents_plugin.tasks.execute.workdir import ( AgentWorkdir, @@ -1894,3 +1895,29 @@ def test_an_unrecognized_telemetry_section_is_left_alone(monkeypatch: pytest.Mon _configure_intake_telemetry(config, workspace="default", sdk=None) assert config["telemetry"] == {"enabled": True, "not_a_real_field": 1} + + +def test_relay_support_is_read_from_the_adapter_descriptor(tmp_path: Path) -> None: + """The bundled harnesses advertise relay; auto-wiring is allowed to trust that.""" + config = AgentConfig.model_validate(_fabric_agent_config()) + + assert _adapter_supports_relay(config, tmp_path) is True + + +def test_an_adapter_without_relay_support_is_not_wired(tmp_path: Path) -> None: + """Fabric rejects a relay config outright, so wiring one would stop the agent running.""" + config = AgentConfig.model_validate( + _fabric_agent_config(harnesses={"h": {"kind": "nvidia.fabric.insights-analyst"}}) + ) + + assert _adapter_supports_relay(config, tmp_path) is False + + +def test_an_unplannable_config_is_not_wired(tmp_path: Path, caplog: pytest.LogCaptureFixture) -> None: + """We cannot know, so we do not wire; the invocation reports the real problem.""" + config = AgentConfig.model_validate(_fabric_agent_config(harnesses={"h": {"kind": "codex"}})) + + with caplog.at_level(logging.WARNING): + assert _adapter_supports_relay(config, tmp_path) is False + + assert "Could not read adapter telemetry support" in caplog.text From c404a6a2b17cee4bdce68f1301c7dbcc55310d84 Mon Sep 17 00:00:00 2001 From: Mike Knepper Date: Tue, 8 Sep 2026 11:58:54 -0500 Subject: [PATCH 05/15] docs(agents): explain the re-validation after telemetry wiring The second _validate_agent_config call reads as a stray assignment inside a conditional. It is a re-validation: wiring mutates the config mapping, not the model validated above, so without it the export would never reach what Fabric is handed. Co-Authored-By: Claude Opus 5 (1M context) Signed-off-by: Mike Knepper --- plugins/nemo-agents/src/nemo_agents_plugin/jobs/execute.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/plugins/nemo-agents/src/nemo_agents_plugin/jobs/execute.py b/plugins/nemo-agents/src/nemo_agents_plugin/jobs/execute.py index 1760151f08..988bf83091 100644 --- a/plugins/nemo-agents/src/nemo_agents_plugin/jobs/execute.py +++ b/plugins/nemo-agents/src/nemo_agents_plugin/jobs/execute.py @@ -411,6 +411,8 @@ def run(self, config: dict, *, ctx: JobContext, sdk: NeMoPlatform | None = None) 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): From 4a06fc12f5cc77fd182aba487948a9ca45de1fc9 Mon Sep 17 00:00:00 2001 From: Mike Knepper Date: Tue, 8 Sep 2026 13:34:40 -0500 Subject: [PATCH 06/15] feat(insights): carry Analyst telemetry to Intake through Relay The adapter discarded its RuntimeContext, so it never saw the telemetry Fabric prepared for it. Declaring relay support in the descriptor alone had no effect: Relay is opt-in per adapter, and each bundled one integrates differently -- hermes enables a plugin, claude runs a gateway and installs hooks, codex merges the env into its subprocess. The Analyst is a fourth shape, an in-process Nooa agent, and Nooa ships middleware for exactly this. The adapter now activates the plugin config Fabric resolved and names the scope; run_analyst_change_set wraps the agent run in it, because the scope needs the agent object and that only exists there. Nothing about the destination is decided in insights any more: the agents plugin wires the export, Fabric resolves endpoint and credentials into a config file, and the adapter activates it. The direct-to-Intake self-observability path is deliberately untouched, so the two can be compared before either is removed. It stays off for analysis runs. Co-Authored-By: Claude Opus 5 (1M context) Signed-off-by: Mike Knepper --- e2e/test_insights_analysis_run.py | 18 +++ .../src/nemo_agents_plugin/jobs/execute.py | 4 +- .../insights-analyst.fabric-adapter.json | 5 + .../src/nemo_insights_plugin/analyst/run.py | 18 ++- .../nemo_insights_plugin/fabric_adapter.py | 52 +++++++- .../tests/test_analyst_fabric_descriptor.py | 13 ++ .../tests/test_fabric_adapter.py | 125 +++++++++++++++++- 7 files changed, 226 insertions(+), 9 deletions(-) diff --git a/e2e/test_insights_analysis_run.py b/e2e/test_insights_analysis_run.py index 7847133e88..770ca619e9 100644 --- a/e2e/test_insights_analysis_run.py +++ b/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 diff --git a/plugins/nemo-agents/src/nemo_agents_plugin/jobs/execute.py b/plugins/nemo-agents/src/nemo_agents_plugin/jobs/execute.py index 988bf83091..abfe6ea19d 100644 --- a/plugins/nemo-agents/src/nemo_agents_plugin/jobs/execute.py +++ b/plugins/nemo-agents/src/nemo_agents_plugin/jobs/execute.py @@ -16,7 +16,6 @@ from pathlib import Path from typing import Any, ClassVar, NamedTuple, cast -import nemo_fabric as fabric from nemo_agents_plugin.agent_config import AgentConfig from nemo_agents_plugin.agent_config_formats import resolve_agent_config_for_deployment from nemo_agents_plugin.config import AgentsConfig @@ -51,6 +50,7 @@ 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 @@ -812,7 +812,7 @@ def _adapter_supports_relay(agent_config: AgentConfig, base_dir: Path) -> bool: run". """ try: - plan = fabric.Fabric().plan(translate_agent_config(agent_config), base_dir=base_dir) + 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 diff --git a/plugins/nemo-insights/insights-analyst.fabric-adapter.json b/plugins/nemo-insights/insights-analyst.fabric-adapter.json index 9be75c7eab..c7442b4b76 100644 --- a/plugins/nemo-insights/insights-analyst.fabric-adapter.json +++ b/plugins/nemo-insights/insights-analyst.fabric-adapter.json @@ -44,6 +44,11 @@ ], "additionalProperties": false }, + "telemetry": { + "providers": { + "relay": {"outputs": ["atif", "otel", "openinference"]} + } + }, "requirements": {}, "config": { "accepts": [ diff --git a/plugins/nemo-insights/src/nemo_insights_plugin/analyst/run.py b/plugins/nemo-insights/src/nemo_insights_plugin/analyst/run.py index 51caefd396..740ec2292c 100644 --- a/plugins/nemo-insights/src/nemo_insights_plugin/analyst/run.py +++ b/plugins/nemo-insights/src/nemo_insights_plugin/analyst/run.py @@ -117,12 +117,18 @@ async def run_analyst_change_set( evaluation_id: str | None = None, analyst_evaluation: AnalystEvaluationContext | None = None, enable_observability: bool = True, + relay_scope_name: str | None = None, model_refs: ConfiguredModelRefs | None = None, ) -> tuple[AnalystResult, AnalystBackend]: """Build and run the analyst agent without persisting its change-set. The caller owns *client* and is responsible for closing it; this function never does. + + Compared to run_analyst, this function adds the following args: + relay_scope_name: Scope to run the agent under when NeMo Relay is + instrumenting it. ``None`` runs uninstrumented. Independent of + *enable_observability*, which is the older direct-to-Intake path. """ observability = None model_clients: ConfiguredModelClients | None = None @@ -156,7 +162,17 @@ async def run_analyst_change_set( agent=agent, ethos=ethos, ) - result = await _run_agent(analyst, verbose=verbose) + # Relay instruments the agent through Nooa middleware, so the scope + # has to wrap the run and needs the agent object -- which only + # exists here. The caller decides whether Relay is active; it holds + # the Fabric context that says so. + if relay_scope_name is None: + result = await _run_agent(analyst, verbose=verbose) + else: + from nooa.nemo_relay_middleware import nemo_relay_scope + + async with nemo_relay_scope(analyst, relay_scope_name): + result = await _run_agent(analyst, verbose=verbose) return result, backend finally: # *client* is deliberately absent here: it belongs to the caller, who diff --git a/plugins/nemo-insights/src/nemo_insights_plugin/fabric_adapter.py b/plugins/nemo-insights/src/nemo_insights_plugin/fabric_adapter.py index 88282ba5de..8939032996 100644 --- a/plugins/nemo-insights/src/nemo_insights_plugin/fabric_adapter.py +++ b/plugins/nemo-insights/src/nemo_insights_plugin/fabric_adapter.py @@ -5,9 +5,11 @@ from __future__ import annotations +import json import logging import os from datetime import datetime +from pathlib import Path from typing import Any from nemo_fabric_adapter_contract import models as contract @@ -16,9 +18,12 @@ from nemo_platform_plugin.nooa_model_client import ConfiguredModelRefs from nemo_platform_plugin.sdk_provider import get_async_task_sdk from nemo_platform_plugin.tasks.logging_setup import configure_task_logging +from nemo_relay import plugin as relay_plugin logger = logging.getLogger(__name__) +ANALYST_RELAY_SCOPE = "insights-analyst" + class AnalystAdapterConfigError(ValueError): """The Fabric-projected analyst adapter configuration is invalid.""" @@ -41,9 +46,8 @@ async def invoke( request: contract.AgentRunRequest, context: contract.RuntimeContext, ) -> contract.AgentRunResult: - del context try: - result = await self._run_analysis(request) + result = await self._run_with_telemetry(request, context) except Exception as error: # Log the full exception so it reaches this process's stderr, which # is where the job reads a failed run's diagnostics from. @@ -66,7 +70,29 @@ async def invoke( }, ) - async def _run_analysis(self, request: contract.AgentRunRequest): + async def _run_with_telemetry( + self, + request: contract.AgentRunRequest, + context: contract.RuntimeContext, + ): + """Run the analysis, instrumented by Relay when Fabric asked for it. + + Fabric resolves the whole export -- endpoint, credentials, agent name -- + into a config file and points at it through ``telemetry.env``. Nothing + about the destination is decided here; the adapter's job is to activate + the config and let Relay carry the trajectory. + """ + telemetry = context.telemetry + if telemetry is None or not telemetry.relay_enabled: + return await self._run_analysis(request) + + # Relay reads credentials for the export from the environment Fabric + # names, so apply it before the exporter is built. + os.environ.update(telemetry.env) + async with relay_plugin.plugin(_relay_plugin_config(telemetry)): + return await self._run_analysis(request, relay_scope_name=ANALYST_RELAY_SCOPE) + + async def _run_analysis(self, request: contract.AgentRunRequest, *, relay_scope_name: str | None = None): target_agent = _string_setting(self._settings, "agent") or _string_setting(self._settings, "target_agent") if target_agent is None: raise AnalystAdapterConfigError("harness.settings.agent is required for the Insights analyst adapter") @@ -102,6 +128,7 @@ async def _run_analysis(self, request: contract.AgentRunRequest): since=since, evaluation_id=evaluation_id, enable_observability=enable_observability, + relay_scope_name=relay_scope_name, model_refs=model_refs, ) return result @@ -110,6 +137,25 @@ async def stop(self) -> None: self.__init__() +def _relay_plugin_config(telemetry: contract.RuntimeTelemetryContext) -> dict[str, Any]: + """Read the Relay plugin config Fabric resolved for this invocation. + + ``nemo_fabric_adapters.common.load_relay_plugin_config`` does the same from + a raw invocation payload, which a lifecycle adapter never sees -- it is + handed the typed context instead, and ``config_path`` points at the same + file. The helper additionally rebases ATOF file-sink directories, which + matters only for configs this path does not produce: the agents plugin + wires ATIF over HTTP. + """ + if not telemetry.config_path: + raise AnalystAdapterConfigError("Relay is enabled but Fabric supplied no config path") + wrapper = json.loads(Path(telemetry.config_path).read_text(encoding="utf-8")) + config = (wrapper.get("relay") or {}).get("config") or {} + if not config.get("components"): + raise AnalystAdapterConfigError(f"Relay config at {telemetry.config_path} declares no components") + return config + + def _string_setting(settings: dict[str, Any], key: str) -> str | None: value = settings.get(key) if value is None: diff --git a/plugins/nemo-insights/tests/test_analyst_fabric_descriptor.py b/plugins/nemo-insights/tests/test_analyst_fabric_descriptor.py index dce0f474a3..0817a02bee 100644 --- a/plugins/nemo-insights/tests/test_analyst_fabric_descriptor.py +++ b/plugins/nemo-insights/tests/test_analyst_fabric_descriptor.py @@ -152,6 +152,19 @@ def test_descriptor_declares_the_ethos_setting() -> None: assert "agent_spec" not in settings_schema["properties"] +def test_descriptor_declares_relay_telemetry_support() -> None: + """Without this, Fabric rejects the relay config agents auto-wires for every job. + + The Analyst then cannot run at all, rather than merely running untraced -- + the export is configured by the agents layer, which asks the descriptor + first. + """ + telemetry = json.loads(DESCRIPTOR.read_text(encoding="utf-8"))["telemetry"] + + assert "relay" in telemetry["providers"] + assert "atif" in telemetry["providers"]["relay"]["outputs"] + + def test_fabric_plans_a_config_carrying_an_ethos(tmp_path: Path) -> None: """Planning is where a setting the descriptor omits actually blows up.""" _require_installed_descriptor() diff --git a/plugins/nemo-insights/tests/test_fabric_adapter.py b/plugins/nemo-insights/tests/test_fabric_adapter.py index ffeec25bdb..99b651a1f6 100644 --- a/plugins/nemo-insights/tests/test_fabric_adapter.py +++ b/plugins/nemo-insights/tests/test_fabric_adapter.py @@ -5,8 +5,15 @@ from __future__ import annotations +import json +import os +from collections.abc import AsyncIterator +from contextlib import asynccontextmanager +from pathlib import Path from typing import Any, cast +from unittest import mock +import pytest from nemo_fabric_adapter_contract import models as contract from nemo_insights_plugin import fabric_adapter from nemo_insights_plugin.analyst.result import AnalystResult @@ -55,6 +62,23 @@ def factory(service: str) -> _StubClient: return factory +def _runtime_context(telemetry: contract.RuntimeTelemetryContext | None = None) -> contract.RuntimeContext: + """The context Fabric hands every invocation; telemetry is the part we read.""" + return contract.RuntimeContext( + runtime_id="runtime-1", + invocation_id="invocation-1", + request_id="request-1", + environment=contract.EnvironmentHandle( + environment_id="environment-1", + provider="local", + control_location="in_env_control", + ownership="caller_owned", + ), + artifacts=contract.ArtifactManifest(), + telemetry=telemetry, + ) + + def _request(context: dict[str, Any] | None = None) -> contract.AgentRunRequest: return contract.AgentRunRequest(input="Analyze telemetry.", context=context or {}) @@ -85,7 +109,7 @@ async def fake_run_analyst_change_set(**kwargs: Any) -> tuple[AnalystResult, obj } ) - result = await runtime.invoke(_request({"job_workspace": "workspace"}), cast(contract.RuntimeContext, None)) + result = await runtime.invoke(_request({"job_workspace": "workspace"}), _runtime_context()) assert result.status is contract.AgentRunStatus.SUCCEEDED assert result.output == { @@ -123,7 +147,7 @@ async def fail_if_called(**kwargs: Any) -> tuple[AnalystResult, object]: runtime = fabric_adapter.InsightsAnalystRuntime() await runtime.start({"config": _agent_config({"agent": "research-agent", "default_model": " "})}) - result = await runtime.invoke(_request({"job_workspace": "workspace"}), cast(contract.RuntimeContext, None)) + result = await runtime.invoke(_request({"job_workspace": "workspace"}), _runtime_context()) assert result.status is contract.AgentRunStatus.FAILED assert result.error is not None @@ -135,7 +159,7 @@ async def test_fabric_adapter_reports_configuration_failure() -> None: runtime = fabric_adapter.InsightsAnalystRuntime() await runtime.start({"config": _agent_config({})}) - result = await runtime.invoke(_request({"job_workspace": "workspace"}), cast(contract.RuntimeContext, None)) + result = await runtime.invoke(_request({"job_workspace": "workspace"}), _runtime_context()) assert result.status is contract.AgentRunStatus.FAILED assert result.error is not None @@ -194,3 +218,98 @@ async def fail(**kwargs: Any) -> tuple[AnalystResult, object]: assert originating in caplog.text, "root cause was dropped" assert raised in caplog.text + + +async def test_relay_activates_fabrics_config_and_scopes_the_agent( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + """Fabric resolves the whole export; the adapter activates it and names the scope. + + Nothing about the destination is decided here — the endpoint, credentials + and agent name all arrive in the config file Fabric wrote. + """ + relay_config = tmp_path / "relay-config.json" + relay_config.write_text( + json.dumps( + { + "relay": { + "config": { + "version": 1, + "components": [ + { + "kind": "observability", + "enabled": True, + # The shape Fabric writes from the agents plugin's + # auto-wiring: destination and credentials already + # resolved, nothing for the adapter to decide. + "config": { + "version": 3, + "atif": { + "enabled": True, + "agent_name": "insights-analyst", + "storage": [ + { + "type": "http", + "endpoint": "http://platform/apis/intake/v2/workspaces/w/ingest/atif", + "header_env": {"X-NMP-Principal-Id": "NMP_HEADER"}, + } + ], + }, + }, + } + ], + } + } + } + ), + encoding="utf-8", + ) + seen: dict[str, Any] = {} + + async def fake_run_analyst_change_set(**kwargs: Any) -> tuple[AnalystResult, object]: + seen.update(kwargs) + return AnalystResult(summary="done"), object() + + @asynccontextmanager + async def fake_plugin(config: Any) -> AsyncIterator[None]: + seen["plugin_config"] = config + yield + + monkeypatch.setattr(fabric_adapter, "run_analyst_change_set", fake_run_analyst_change_set) + monkeypatch.setattr(fabric_adapter.relay_plugin, "plugin", fake_plugin) + monkeypatch.setattr(fabric_adapter, "get_async_task_sdk", _stub_sdk_factory([])) + + runtime = fabric_adapter.InsightsAnalystRuntime() + await runtime.start({"config": _agent_config({"agent": "research-agent"})}) + telemetry = contract.RuntimeTelemetryContext( + relay_enabled=True, + config_path=str(relay_config), + env={"FABRIC_RELAY_CONFIG_PATH": str(relay_config)}, + ) + + result = await runtime.invoke(_request({"job_workspace": "w"}), _runtime_context(telemetry)) + + assert result.status is contract.AgentRunStatus.SUCCEEDED + assert seen["relay_scope_name"] == fabric_adapter.ANALYST_RELAY_SCOPE + assert seen["plugin_config"]["components"][0]["kind"] == "observability" + assert os.environ["FABRIC_RELAY_CONFIG_PATH"] == str(relay_config) + + +async def test_without_relay_the_agent_runs_unscoped() -> None: + """Relay is opt-in per invocation; Fabric says when.""" + seen: dict[str, Any] = {} + + async def fake_run_analyst_change_set(**kwargs: Any) -> tuple[AnalystResult, object]: + seen.update(kwargs) + return AnalystResult(summary="done"), object() + + runtime = fabric_adapter.InsightsAnalystRuntime() + await runtime.start({"config": _agent_config({"agent": "research-agent"})}) + + with ( + mock.patch.object(fabric_adapter, "run_analyst_change_set", fake_run_analyst_change_set), + mock.patch.object(fabric_adapter, "get_async_task_sdk", _stub_sdk_factory([])), + ): + await runtime.invoke(_request({"job_workspace": "w"}), _runtime_context(None)) + + assert seen["relay_scope_name"] is None From db07a02fb5f65dd4af06e87e59b6a577541a503c Mon Sep 17 00:00:00 2001 From: Mike Knepper Date: Tue, 8 Sep 2026 14:03:42 -0500 Subject: [PATCH 07/15] test(insights): give the adapter's logging tests a real RuntimeContext Two tests arriving with the rebase pass None as the context, which only worked while invoke discarded it. It now reads context.telemetry to decide whether Relay is instrumenting the run, so None raises before the failure they are actually asserting on. They care about error logging, not telemetry, so hand them the same minimal context the other tests build. Co-Authored-By: Claude Opus 5 (1M context) Signed-off-by: Mike Knepper --- plugins/nemo-insights/tests/test_fabric_adapter.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/plugins/nemo-insights/tests/test_fabric_adapter.py b/plugins/nemo-insights/tests/test_fabric_adapter.py index 99b651a1f6..d4775eb923 100644 --- a/plugins/nemo-insights/tests/test_fabric_adapter.py +++ b/plugins/nemo-insights/tests/test_fabric_adapter.py @@ -10,7 +10,7 @@ from collections.abc import AsyncIterator from contextlib import asynccontextmanager from pathlib import Path -from typing import Any, cast +from typing import Any from unittest import mock import pytest @@ -183,7 +183,7 @@ async def fail(**kwargs: Any) -> tuple[AnalystResult, object]: await runtime.start({"config": _agent_config({"agent": "research-agent"})}) with caplog.at_level("ERROR", logger="nemo_insights_plugin.fabric_adapter"): - result = await runtime.invoke(_request({"job_workspace": "workspace"}), cast(contract.RuntimeContext, None)) + result = await runtime.invoke(_request({"job_workspace": "workspace"}), _runtime_context()) assert result.status is contract.AgentRunStatus.FAILED assert result.error is not None @@ -214,7 +214,7 @@ async def fail(**kwargs: Any) -> tuple[AnalystResult, object]: await runtime.start({"config": _agent_config({"agent": "research-agent"})}) with caplog.at_level("ERROR", logger="nemo_insights_plugin.fabric_adapter"): - await runtime.invoke(_request({"job_workspace": "workspace"}), cast(contract.RuntimeContext, None)) + await runtime.invoke(_request({"job_workspace": "workspace"}), _runtime_context()) assert originating in caplog.text, "root cause was dropped" assert raised in caplog.text From 8c7edbf2017cd03ccebec5bf7881be6d469ea970 Mon Sep 17 00:00:00 2001 From: Mike Knepper Date: Tue, 8 Sep 2026 14:47:19 -0500 Subject: [PATCH 08/15] fix(agents,insights): scope telemetry side effects to their caller and run Three fixes from PR review, each verified by reverting the fix and watching a test fail. Deployments: configure_intake_atif_export mutated the config it was handed, which is the caller's deployment entity. Since the wiring keeps an ATIF endpoint it finds already present, a second deployment of the same entity would have exported to the first one's workspace. Copy first, as the adjacent rewrite_fabric_config_base_urls already does. Insights adapter: the invocation's telemetry environment was applied and never removed. The runtime is long-lived, so a leftover FABRIC_RELAY_CONFIG_PATH is exactly the ambient-config hazard the bundled adapters guard against by name. Apply it in a restoring scope. Tests: the header variable written by _configure_intake_telemetry leaked into every later test in the session, because monkeypatch can only restore what it saw first. Register it before the call. Co-Authored-By: Claude Opus 5 (1M context) Signed-off-by: Mike Knepper --- .../runner/deployments_backend.py | 3 ++ .../tests/unit/test_execute_job.py | 6 ++- .../tests/unit/test_runner_deployments.py | 39 +++++++++++++++++++ .../nemo_insights_plugin/fabric_adapter.py | 28 +++++++++++-- .../tests/test_fabric_adapter.py | 8 +++- 5 files changed, 78 insertions(+), 6 deletions(-) diff --git a/plugins/nemo-agents/src/nemo_agents_plugin/runner/deployments_backend.py b/plugins/nemo-agents/src/nemo_agents_plugin/runner/deployments_backend.py index 09513073a8..6d116be3d4 100644 --- a/plugins/nemo-agents/src/nemo_agents_plugin/runner/deployments_backend.py +++ b/plugins/nemo-agents/src/nemo_agents_plugin/runner/deployments_backend.py @@ -673,6 +673,9 @@ async def create_deployment( # 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: diff --git a/plugins/nemo-agents/tests/unit/test_execute_job.py b/plugins/nemo-agents/tests/unit/test_execute_job.py index 2b7ec90e02..bb77882570 100644 --- a/plugins/nemo-agents/tests/unit/test_execute_job.py +++ b/plugins/nemo-agents/tests/unit/test_execute_job.py @@ -1834,6 +1834,10 @@ def test_telemetry_is_pointed_at_the_workspace_intake_ingest(monkeypatch: pytest def test_telemetry_credentials_go_to_the_environment_not_the_config(monkeypatch: pytest.MonkeyPatch) -> None: """Fabric writes the config into artifacts that are uploaded as a job result.""" monkeypatch.setenv("NMP_BASE_URL", "http://nemo-platform-api:8080") + # Registered before the call so pytest unsets it afterwards: the code writes + # this variable directly, and monkeypatch can only restore what it saw first. + header_var = "NMP_AGENT_TELEMETRY_HEADER_X_NMP_PRINCIPAL_ID" + monkeypatch.setenv(header_var, "overwritten-by-the-call") sdk = cast(NeMoPlatform, SimpleNamespace(_custom_headers={"X-NMP-Principal-Id": "service:agents"})) config = _fabric_agent_config() @@ -1842,7 +1846,7 @@ def test_telemetry_credentials_go_to_the_environment_not_the_config(monkeypatch: storage = config["telemetry"]["atif"]["storage"][0] assert storage["header_env"] == {"X-NMP-Principal-Id": "NMP_AGENT_TELEMETRY_HEADER_X_NMP_PRINCIPAL_ID"} assert "headers" not in storage, "an inline header would land in a downloadable artifact" - assert os.environ["NMP_AGENT_TELEMETRY_HEADER_X_NMP_PRINCIPAL_ID"] == "service:agents" + assert os.environ[header_var] == "service:agents" def test_an_agent_that_names_its_own_destination_keeps_it(monkeypatch: pytest.MonkeyPatch) -> None: diff --git a/plugins/nemo-agents/tests/unit/test_runner_deployments.py b/plugins/nemo-agents/tests/unit/test_runner_deployments.py index cec3bba44d..42c4977a72 100644 --- a/plugins/nemo-agents/tests/unit/test_runner_deployments.py +++ b/plugins/nemo-agents/tests/unit/test_runner_deployments.py @@ -1144,6 +1144,45 @@ async def test_create_deployment_fabric_k8s_auth_on_rewrites_to_auth_proxy() -> ) +@pytest.mark.asyncio +async def test_deploying_one_config_twice_does_not_carry_the_first_workspace_over() -> None: + """Telemetry wiring must not mutate the caller's config. + + ``create_deployment`` receives the deployment entity's own config dict, and + the wiring keeps an ATIF storage endpoint it finds already present -- so a + mutated dict would send the second deployment's trajectories to the first + deployment's workspace. + """ + backend = _backend( + default_image="fabric:latest", default_executor="k8s", k8s_internal_base_url="http://nmp-api:8080" + ) + backend._entities = AsyncMock() + config = { + "config_format": "nemo-agents-spec-v1", + "name": "fabric-agent", + "default_harness": "main", + "harnesses": {"main": {"provider": "codex", "model": {"provider": "openai", "model": "m"}}}, + } + + with patch("nemo_agents_plugin.runner.deployments_backend.get_base_url", return_value="http://localhost:8080"): + await backend.create_deployment( + workspace="workspace-a", name="dep-a", config=config, port=0, deployment_mode="k8s" + ) + assert "telemetry" not in config, "the caller's config was mutated" + await backend.create_deployment( + workspace="workspace-b", name="dep-b", config=config, port=0, deployment_mode="k8s" + ) + + # Each deployment creates a DeploymentConfig and a Deployment; only the + # former carries the baked agent config. + configs = [ + call.args[0] for call in backend._entities.create.await_args_list if hasattr(call.args[0], "config_files") + ] + baked = yaml.safe_load(configs[1].config_files[0].content) + endpoint = baked["telemetry"]["atif"]["storage"][0]["endpoint"] + assert "workspace-b" in endpoint and "workspace-a" not in endpoint + + @pytest.mark.asyncio async def test_create_deployment_missing_image_fails() -> None: backend = _backend(default_image="") diff --git a/plugins/nemo-insights/src/nemo_insights_plugin/fabric_adapter.py b/plugins/nemo-insights/src/nemo_insights_plugin/fabric_adapter.py index 8939032996..138cc2dfc5 100644 --- a/plugins/nemo-insights/src/nemo_insights_plugin/fabric_adapter.py +++ b/plugins/nemo-insights/src/nemo_insights_plugin/fabric_adapter.py @@ -8,6 +8,8 @@ import json import logging import os +from collections.abc import Iterator +from contextlib import contextmanager from datetime import datetime from pathlib import Path from typing import Any @@ -87,10 +89,13 @@ async def _run_with_telemetry( return await self._run_analysis(request) # Relay reads credentials for the export from the environment Fabric - # names, so apply it before the exporter is built. - os.environ.update(telemetry.env) - async with relay_plugin.plugin(_relay_plugin_config(telemetry)): - return await self._run_analysis(request, relay_scope_name=ANALYST_RELAY_SCOPE) + # names, so apply it before the exporter is built -- and only for this + # invocation. The runtime is long-lived and serves many; a leftover + # FABRIC_RELAY_CONFIG_PATH is the ambient-config hazard the bundled + # adapters have a named guard against. + with _applied_environment(telemetry.env): + async with relay_plugin.plugin(_relay_plugin_config(telemetry)): + return await self._run_analysis(request, relay_scope_name=ANALYST_RELAY_SCOPE) async def _run_analysis(self, request: contract.AgentRunRequest, *, relay_scope_name: str | None = None): target_agent = _string_setting(self._settings, "agent") or _string_setting(self._settings, "target_agent") @@ -137,6 +142,21 @@ async def stop(self) -> None: self.__init__() +@contextmanager +def _applied_environment(env: dict[str, str]) -> Iterator[None]: + """Apply *env* for the duration of one invocation, then put it back.""" + previous = {name: os.environ.get(name) for name in env} + os.environ.update(env) + try: + yield + finally: + for name, value in previous.items(): + if value is None: + os.environ.pop(name, None) + else: + os.environ[name] = value + + def _relay_plugin_config(telemetry: contract.RuntimeTelemetryContext) -> dict[str, Any]: """Read the Relay plugin config Fabric resolved for this invocation. diff --git a/plugins/nemo-insights/tests/test_fabric_adapter.py b/plugins/nemo-insights/tests/test_fabric_adapter.py index d4775eb923..2c11c32102 100644 --- a/plugins/nemo-insights/tests/test_fabric_adapter.py +++ b/plugins/nemo-insights/tests/test_fabric_adapter.py @@ -273,6 +273,9 @@ async def fake_run_analyst_change_set(**kwargs: Any) -> tuple[AnalystResult, obj @asynccontextmanager async def fake_plugin(config: Any) -> AsyncIterator[None]: seen["plugin_config"] = config + # Relay resolves header_env against the environment while exporting, + # so the variables have to be set for the duration of the run. + seen["env_during_run"] = os.environ.get("FABRIC_RELAY_CONFIG_PATH") yield monkeypatch.setattr(fabric_adapter, "run_analyst_change_set", fake_run_analyst_change_set) @@ -292,7 +295,10 @@ async def fake_plugin(config: Any) -> AsyncIterator[None]: assert result.status is contract.AgentRunStatus.SUCCEEDED assert seen["relay_scope_name"] == fabric_adapter.ANALYST_RELAY_SCOPE assert seen["plugin_config"]["components"][0]["kind"] == "observability" - assert os.environ["FABRIC_RELAY_CONFIG_PATH"] == str(relay_config) + assert seen["env_during_run"] == str(relay_config) + # The runtime serves many invocations; a leftover config path would make + # the next one export against a stale, possibly deleted, config. + assert "FABRIC_RELAY_CONFIG_PATH" not in os.environ async def test_without_relay_the_agent_runs_unscoped() -> None: From 60d7b04039bed1c1ab16605f9a15ccbac768d65c Mon Sep 17 00:00:00 2001 From: Mike Knepper Date: Wed, 9 Sep 2026 10:21:31 -0500 Subject: [PATCH 09/15] fix(agents,insights): close the gaps found in review of telemetry auto-wiring Workload identity: get_forwarding_headers returns only what the SDK was constructed with, and under workload identity that is the internal marker alone -- the bearer is exchanged per request by the SDK's auth layer, which Relay's raw POST to Intake never goes through. Exchange the subject token and pass a bearer instead of exporting unauthenticated on the deployments that enforce auth. The token is resolved once because header_env is a static lookup, so a run outliving its token exports with an expired one; refreshing needs a dynamic-credential hook Relay does not offer. request telemetry=false now disables an export the agent config declares, rather than only skipping the automatic wiring, which is what the field already documented. An explicit atif.enabled=false is honoured: declining ATIF while leaving telemetry on is a real choice, and overriding it contradicts the rule that an explicit declaration wins. The adapter capability guard moves into the shared telemetry module and now covers deployments too -- previously a custom adapter without Relay support became undeployable by default, the same failure the execute-job path already guarded against. It also checks that relay advertises the atif output, since an adapter may support Relay for OpenTelemetry alone. nemo-relay is declared by nemo-insights, which imports it directly and had been relying on it arriving transitively. Co-Authored-By: Claude Opus 5 (1M context) Signed-off-by: Mike Knepper --- .../src/nemo_agents_plugin/jobs/execute.py | 81 ++++++++++------ .../runner/deployments_backend.py | 10 +- .../telemetry/intake_export.py | 56 +++++++++++ .../tests/unit/test_execute_job.py | 94 ++++++++++++++++--- .../tests/unit/test_runner_deployments.py | 6 +- plugins/nemo-insights/pyproject.toml | 2 + uv.lock | 3 +- 7 files changed, 204 insertions(+), 48 deletions(-) diff --git a/plugins/nemo-agents/src/nemo_agents_plugin/jobs/execute.py b/plugins/nemo-agents/src/nemo_agents_plugin/jobs/execute.py index abfe6ea19d..baf292e3a3 100644 --- a/plugins/nemo-agents/src/nemo_agents_plugin/jobs/execute.py +++ b/plugins/nemo-agents/src/nemo_agents_plugin/jobs/execute.py @@ -37,7 +37,6 @@ 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, @@ -49,9 +48,16 @@ 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_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 @@ -199,8 +205,8 @@ class ExecuteAgentJobConfig(BaseModel): 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." + "Export the agent's trajectory to Intake. False runs untraced, overriding any export " + "the agent config declares; an agent that declares its own is otherwise left alone." ), ) extension: ExecuteAgentExtensionConfig | None = Field( @@ -409,7 +415,13 @@ 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): + if not step_config.request.telemetry: + # "Run untraced" has to hold for an agent that configured its own + # export too, or the request-level switch would silently only + # govern the automatic wiring. + _disable_agent_telemetry(step_config.agent.config) + agent_config = _validate_agent_config(step_config.agent.config) + elif 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. @@ -802,29 +814,10 @@ 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 _disable_agent_telemetry(agent_config: dict[str, Any]) -> None: + """Turn off any export the agent config declares, in place.""" + telemetry = agent_config.get("telemetry") + agent_config["telemetry"] = {**telemetry, "enabled": False} if isinstance(telemetry, dict) else {"enabled": False} def _configure_intake_telemetry( @@ -852,6 +845,7 @@ def _configure_intake_telemetry( 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 @@ -863,6 +857,35 @@ def _configure_intake_telemetry( ) +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('-', '_')}" diff --git a/plugins/nemo-agents/src/nemo_agents_plugin/runner/deployments_backend.py b/plugins/nemo-agents/src/nemo_agents_plugin/runner/deployments_backend.py index 6d116be3d4..92021d6454 100644 --- a/plugins/nemo-agents/src/nemo_agents_plugin/runner/deployments_backend.py +++ b/plugins/nemo-agents/src/nemo_agents_plugin/runner/deployments_backend.py @@ -38,7 +38,10 @@ FabricArtifactStagingError, stage_fabric_ethos_config_files, ) -from nemo_agents_plugin.telemetry.intake_export import configure_intake_atif_export +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 ( @@ -675,8 +678,9 @@ async def create_deployment( # 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) + 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) diff --git a/plugins/nemo-agents/src/nemo_agents_plugin/telemetry/intake_export.py b/plugins/nemo-agents/src/nemo_agents_plugin/telemetry/intake_export.py index 1d51078217..f791fe24f7 100644 --- a/plugins/nemo-agents/src/nemo_agents_plugin/telemetry/intake_export.py +++ b/plugins/nemo-agents/src/nemo_agents_plugin/telemetry/intake_export.py @@ -17,8 +17,13 @@ from __future__ import annotations import logging +import tempfile +from collections.abc import Iterator +from contextlib import contextmanager +from pathlib import Path from typing import Any +import nemo_fabric as fabric from nemo_agents_plugin.agent_config import TelemetryConfig from pydantic import ValidationError @@ -27,6 +32,52 @@ INTAKE_ATIF_INGEST_PATH = "/apis/intake/v2/workspaces/{workspace}/ingest/atif" +def supports_intake_atif_export(config: dict[str, Any], *, base_dir: Path | None = None) -> bool: + """Whether the agent's adapter declares that Relay can export ATIF for it. + + Adapters advertise this in their descriptor's ``telemetry.providers``, and + Fabric rejects a relay configuration outright for one that does not -- so + wiring without asking turns "this agent cannot be traced" into "this agent + cannot run". The ``atif`` output is checked too: an adapter may support + Relay for OpenTelemetry alone, and ATIF is what this wires. + + Anything we cannot answer -- a config that will not validate, a plan that + fails -- returns False. Those failures belong to whoever runs the agent + next, reported with their own diagnostics; here they only mean we do not + know enough to wire telemetry. + """ + try: + from nemo_agents_plugin.agent_config import AgentConfig + from nemo_agents_plugin.fabric.translator import translate_agent_config + + agent_config = AgentConfig.model_validate(config) + with _plan_directory(base_dir) as resolved_base: + plan = fabric.Fabric().plan(translate_agent_config(agent_config), base_dir=resolved_base) + descriptor = plan.to_dict().get("adapter_descriptor") or {} + relay = descriptor.get("descriptor", descriptor).get("telemetry", {}).get("providers", {}).get("relay") + except Exception: + logger.warning("Could not read adapter telemetry support; the agent will run untraced.", exc_info=True) + return False + + if relay is None: + logger.info("Adapter does not support Relay telemetry; the agent will run untraced.") + return False + if "atif" not in (relay.get("outputs") or []): + logger.info("Adapter supports Relay but not its ATIF output; the agent will run untraced.") + return False + return True + + +@contextmanager +def _plan_directory(base_dir: Path | None) -> Iterator[Path]: + """Planning needs somewhere to resolve against; callers mid-run already have one.""" + if base_dir is not None: + yield base_dir + return + with tempfile.TemporaryDirectory() as scratch: + yield Path(scratch) + + def configure_intake_atif_export( config: dict[str, Any], *, @@ -83,6 +134,11 @@ def configure_intake_atif_export( storage["header_env"] = dict(header_env) atif = dict(telemetry.atif or {}) + if atif.get("enabled") is False: + # Declining ATIF while leaving telemetry on is a real choice -- an agent + # may want only OTel -- and overriding it would be the opposite of + # preserving an explicit declaration. + return False atif["enabled"] = True atif["storage"] = [storage] diff --git a/plugins/nemo-agents/tests/unit/test_execute_job.py b/plugins/nemo-agents/tests/unit/test_execute_job.py index bb77882570..32bf083a2a 100644 --- a/plugins/nemo-agents/tests/unit/test_execute_job.py +++ b/plugins/nemo-agents/tests/unit/test_execute_job.py @@ -28,6 +28,7 @@ McpFulfillment, ) from nemo_agents_plugin.fabric.runtime import FabricRuntimeResult +from nemo_agents_plugin.jobs import execute as execute_module from nemo_agents_plugin.jobs.execute import ( DEFAULT_AGENT_EXECUTION_TIMEOUT_SECONDS, FABRIC_ERROR_RESULT_NAME, @@ -40,8 +41,8 @@ ExecuteAgentJobConfig, ExecuteAgentStepConfig, ResolvedAgentConfig, - _adapter_supports_relay, _configure_intake_telemetry, + _disable_agent_telemetry, _log_agent_stderr, ) from nemo_agents_plugin.tasks.execute.workdir import ( @@ -51,6 +52,8 @@ materialize_agent_workdir, validate_agent_workdir, ) +from nemo_agents_plugin.telemetry import intake_export +from nemo_agents_plugin.telemetry.intake_export import supports_intake_atif_export from nemo_platform import NeMoPlatform from nemo_platform_plugin.dependencies import get_entity_client, get_sdk_client from nemo_platform_plugin.entity_client import NemoEntityNotFoundError @@ -1902,26 +1905,89 @@ def test_an_unrecognized_telemetry_section_is_left_alone(monkeypatch: pytest.Mon def test_relay_support_is_read_from_the_adapter_descriptor(tmp_path: Path) -> None: - """The bundled harnesses advertise relay; auto-wiring is allowed to trust that.""" - config = AgentConfig.model_validate(_fabric_agent_config()) + """The bundled harnesses advertise relay with an ATIF output.""" + assert supports_intake_atif_export(_fabric_agent_config(), base_dir=tmp_path) is True - assert _adapter_supports_relay(config, tmp_path) is True +def test_an_unplannable_config_is_not_wired(tmp_path: Path, caplog: pytest.LogCaptureFixture) -> None: + """We cannot know, so we do not wire; the invocation reports the real problem.""" + config = _fabric_agent_config(harnesses={"h": {"kind": "codex"}}) -def test_an_adapter_without_relay_support_is_not_wired(tmp_path: Path) -> None: - """Fabric rejects a relay config outright, so wiring one would stop the agent running.""" - config = AgentConfig.model_validate( - _fabric_agent_config(harnesses={"h": {"kind": "nvidia.fabric.insights-analyst"}}) + with caplog.at_level(logging.WARNING): + assert supports_intake_atif_export(config, base_dir=tmp_path) is False + + assert "Could not read adapter telemetry support" in caplog.text + + +def test_an_adapter_without_the_atif_output_is_not_wired( + tmp_path: Path, caplog: pytest.LogCaptureFixture, monkeypatch: pytest.MonkeyPatch +) -> None: + """Relay support alone is not enough: an adapter may offer only OpenTelemetry.""" + otel_only = {"providers": {"relay": {"outputs": ["otel"]}}} + + class _Plan: + @staticmethod + def to_dict() -> dict[str, Any]: + return {"adapter_descriptor": {"descriptor": {"telemetry": otel_only}}} + + monkeypatch.setattr(intake_export.fabric, "Fabric", lambda: SimpleNamespace(plan=lambda *a, **k: _Plan())) + + with caplog.at_level(logging.INFO): + assert supports_intake_atif_export(_fabric_agent_config(), base_dir=tmp_path) is False + + assert "not its ATIF output" in caplog.text + + +def test_a_request_that_declines_telemetry_disables_an_agents_own_export() -> None: + """ "Run untraced" has to beat an export the agent config declared.""" + config = _fabric_agent_config( + telemetry={"enabled": True, "atif": {"enabled": True, "storage": [{"type": "http", "endpoint": "https://x"}]}} ) - assert _adapter_supports_relay(config, tmp_path) is False + _disable_agent_telemetry(config) + assert config["telemetry"]["enabled"] is False -def test_an_unplannable_config_is_not_wired(tmp_path: Path, caplog: pytest.LogCaptureFixture) -> None: - """We cannot know, so we do not wire; the invocation reports the real problem.""" - config = AgentConfig.model_validate(_fabric_agent_config(harnesses={"h": {"kind": "codex"}})) + +def test_workload_identity_jobs_export_with_a_bearer_token(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + """The SDK adds this per request; Relay's raw POST to Intake does not go through it.""" + monkeypatch.setenv("NMP_BASE_URL", "http://nemo-platform-api:8080") + token_file = tmp_path / "subject-token" + token_file.write_text("subject", encoding="utf-8") + monkeypatch.setenv("NMP_WORKLOAD_IDENTITY_TOKEN_FILE", str(token_file)) + monkeypatch.setenv("NMP_AGENT_TELEMETRY_HEADER_AUTHORIZATION", "unset") + monkeypatch.setattr( + execute_module, + "resolve_workload_exchange_provider", + lambda **_kwargs: SimpleNamespace(get_access_token=lambda: "exchanged-token"), + ) + sdk = cast(NeMoPlatform, SimpleNamespace(_custom_headers={"X-NMP-Internal": "true"})) + config = _fabric_agent_config() + + _configure_intake_telemetry(config, workspace="default", sdk=sdk) + + storage = config["telemetry"]["atif"]["storage"][0] + assert storage["header_env"]["Authorization"] == "NMP_AGENT_TELEMETRY_HEADER_AUTHORIZATION" + assert os.environ["NMP_AGENT_TELEMETRY_HEADER_AUTHORIZATION"] == "Bearer exchanged-token" + + +def test_a_failed_token_exchange_still_exports_rather_than_failing_the_run( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path, caplog: pytest.LogCaptureFixture +) -> None: + """Telemetry is not worth failing an agent over.""" + monkeypatch.setenv("NMP_BASE_URL", "http://nemo-platform-api:8080") + token_file = tmp_path / "subject-token" + token_file.write_text("subject", encoding="utf-8") + monkeypatch.setenv("NMP_WORKLOAD_IDENTITY_TOKEN_FILE", str(token_file)) + + def explode(**_kwargs: Any) -> Any: + raise RuntimeError("auth discovery unavailable") + + monkeypatch.setattr(execute_module, "resolve_workload_exchange_provider", explode) + config = _fabric_agent_config() with caplog.at_level(logging.WARNING): - assert _adapter_supports_relay(config, tmp_path) is False + _configure_intake_telemetry(config, workspace="default", sdk=None) - assert "Could not read adapter telemetry support" in caplog.text + assert "Authorization" not in config["telemetry"]["atif"]["storage"][0].get("header_env", {}) + assert "without credentials" in caplog.text diff --git a/plugins/nemo-agents/tests/unit/test_runner_deployments.py b/plugins/nemo-agents/tests/unit/test_runner_deployments.py index 42c4977a72..6e6f66accd 100644 --- a/plugins/nemo-agents/tests/unit/test_runner_deployments.py +++ b/plugins/nemo-agents/tests/unit/test_runner_deployments.py @@ -1161,7 +1161,11 @@ async def test_deploying_one_config_twice_does_not_carry_the_first_workspace_ove "config_format": "nemo-agents-spec-v1", "name": "fabric-agent", "default_harness": "main", - "harnesses": {"main": {"provider": "codex", "model": {"provider": "openai", "model": "m"}}}, + # A config the telemetry guard can actually plan: it validates as + # spec-v1 and names a harness whose adapter advertises Relay ATIF. + "harnesses": {"main": {"kind": "hermes"}}, + "models": {"default": {"provider": "platform", "model": "default/m"}}, + "environment": {"provider": "local"}, } with patch("nemo_agents_plugin.runner.deployments_backend.get_base_url", return_value="http://localhost:8080"): diff --git a/plugins/nemo-insights/pyproject.toml b/plugins/nemo-insights/pyproject.toml index 0cdaff3fd7..34feb53bd0 100644 --- a/plugins/nemo-insights/pyproject.toml +++ b/plugins/nemo-insights/pyproject.toml @@ -11,6 +11,8 @@ dependencies = [ "httpx", "nemo-fabric-adapter-contract>=0.3.0b1", "nemo-fabric-adapters-common>=0.3.0b1", + # Imported by the Fabric adapter to activate the Relay config Fabric resolves. + "nemo-relay>=0.7,<0.8", "nemo-platform", "nemo-platform-plugin", "nooa>=0.0.9", diff --git a/uv.lock b/uv.lock index b7e13dfe90..9345d1a066 100644 --- a/uv.lock +++ b/uv.lock @@ -4838,6 +4838,7 @@ dependencies = [ { name = "nemo-fabric-adapters-common", marker = "(platform_machine == 'arm64' and sys_platform == 'darwin') or (platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux')" }, { name = "nemo-platform", marker = "(platform_machine == 'arm64' and sys_platform == 'darwin') or (platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux')" }, { name = "nemo-platform-plugin", marker = "(platform_machine == 'arm64' and sys_platform == 'darwin') or (platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux')" }, + { name = "nemo-relay", marker = "(platform_machine == 'arm64' and sys_platform == 'darwin') or (platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux')" }, { name = "nooa", marker = "(platform_machine == 'arm64' and sys_platform == 'darwin') or (platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux')" }, { name = "opentelemetry-exporter-otlp", marker = "(platform_machine == 'arm64' and sys_platform == 'darwin') or (platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux')" }, { name = "opentelemetry-sdk", marker = "(platform_machine == 'arm64' and sys_platform == 'darwin') or (platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux')" }, @@ -4855,6 +4856,7 @@ requires-dist = [ { name = "nemo-fabric-adapters-common", specifier = ">=0.3.0b1" }, { name = "nemo-platform", editable = "packages/nemo_platform" }, { name = "nemo-platform-plugin", editable = "packages/nemo_platform_plugin" }, + { name = "nemo-relay", specifier = ">=0.7,<0.8" }, { name = "nooa", specifier = ">=0.0.9" }, { name = "opentelemetry-exporter-otlp", specifier = ">=1.42.1" }, { name = "opentelemetry-sdk", specifier = ">=1.42.1" }, @@ -12362,7 +12364,6 @@ dependencies = [ { name = "wheel", marker = "(platform_machine == 'arm64' and sys_platform == 'darwin') or (platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux')" }, { name = "xformers", marker = "platform_machine == 'x86_64' and sys_platform == 'linux' and 'linux' in sys_platform" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/46/0b/fb9344ea1bc745a2a113f8db502e4d8d7c8111db28ae3196a51228ee5518/unsloth-2026.8.4.tar.gz", hash = "sha256:e34d07f68e66c287769695ab7930508467584b2439a6f639158f44230f70993d", size = 84321638, upload-time = "2026-08-05T07:32:54.93Z" } wheels = [ { url = "https://files.pythonhosted.org/packages/f7/13/a2e9e26ae8e826a6d68e9279e96ca9ba096f9489cfbacf5927f3a08cdc48/unsloth-2026.8.4-py3-none-any.whl", hash = "sha256:c180cc5bae5597f420eaf3d27f92ac8c3fd0bf4d4c51ffed71e23dd36c569bfe", size = 79350425, upload-time = "2026-08-05T07:32:49.984Z" }, ] From 84369694b72bb14c5c388fed47a5899d0b26193a Mon Sep 17 00:00:00 2001 From: Mike Knepper Date: Wed, 9 Sep 2026 10:56:23 -0500 Subject: [PATCH 10/15] fix(agents): leave a config that declares any export destination alone The "already declared" check only looked at ATIF storage, so a config exporting OpenTelemetry to its own collector had an Intake ATIF destination added beside it -- a second export it never asked for, which is the opposite of letting an explicit declaration win. Check every output instead: atif.storage, atof.sinks, opentelemetry.endpoints. The question the helper answers becomes "did the agent name a destination?" rather than one question per output. Co-Authored-By: Claude Opus 5 (1M context) Signed-off-by: Mike Knepper --- .../telemetry/intake_export.py | 18 ++++++++++---- .../tests/unit/test_execute_job.py | 24 +++++++++++++++++++ 2 files changed, 38 insertions(+), 4 deletions(-) diff --git a/plugins/nemo-agents/src/nemo_agents_plugin/telemetry/intake_export.py b/plugins/nemo-agents/src/nemo_agents_plugin/telemetry/intake_export.py index f791fe24f7..6ff231f949 100644 --- a/plugins/nemo-agents/src/nemo_agents_plugin/telemetry/intake_export.py +++ b/plugins/nemo-agents/src/nemo_agents_plugin/telemetry/intake_export.py @@ -123,7 +123,7 @@ def configure_intake_atif_export( if telemetry.enabled is False: return False - if _declares_atif_storage(telemetry): + if _declares_a_destination(telemetry): return False storage: dict[str, object] = { @@ -154,6 +154,16 @@ def configure_intake_atif_export( 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")) +def _declares_a_destination(telemetry: TelemetryConfig) -> bool: + """Whether the config already names anywhere to send telemetry. + + Checked across every output, not just ATIF: a config that exports + OpenTelemetry to its own collector has declared where its telemetry goes, + and adding a second destination it never asked for is the opposite of + letting an explicit declaration win. + """ + return bool( + (isinstance(telemetry.atif, dict) and telemetry.atif.get("storage")) + or (isinstance(telemetry.atof, dict) and telemetry.atof.get("sinks")) + or (isinstance(telemetry.opentelemetry, dict) and telemetry.opentelemetry.get("endpoints")) + ) diff --git a/plugins/nemo-agents/tests/unit/test_execute_job.py b/plugins/nemo-agents/tests/unit/test_execute_job.py index 32bf083a2a..7a17d955d3 100644 --- a/plugins/nemo-agents/tests/unit/test_execute_job.py +++ b/plugins/nemo-agents/tests/unit/test_execute_job.py @@ -1852,6 +1852,30 @@ def test_telemetry_credentials_go_to_the_environment_not_the_config(monkeypatch: assert os.environ[header_var] == "service:agents" +@pytest.mark.parametrize( + ("output", "declaration"), + [ + ("atof", {"enabled": True, "sinks": [{"type": "stream", "url": "https://mine/events"}]}), + ("opentelemetry", {"endpoints": [{"type": "gen_ai", "endpoint": "https://mine/otlp"}]}), + ], +) +def test_a_destination_declared_through_any_output_is_left_alone( + monkeypatch: pytest.MonkeyPatch, output: str, declaration: dict[str, Any] +) -> None: + """Declaring an export is declaring where telemetry goes, whichever output carries it. + + Adding an Intake destination beside one the agent chose would be a second, + unrequested export -- not the "fill in what was left out" this is for. + """ + monkeypatch.setenv("NMP_BASE_URL", "http://nemo-platform-api:8080") + config = _fabric_agent_config(telemetry={"enabled": True, output: declaration}) + + _configure_intake_telemetry(config, workspace="default", sdk=None) + + assert "atif" not in config["telemetry"] + assert config["telemetry"][output] == declaration + + def test_an_agent_that_names_its_own_destination_keeps_it(monkeypatch: pytest.MonkeyPatch) -> None: """An explicit export destination beats an inferred one.""" monkeypatch.setenv("NMP_BASE_URL", "http://nemo-platform-api:8080") From bb147ed18b2d93a36959980e4466b9deeba48e2d Mon Sep 17 00:00:00 2001 From: Mike Knepper Date: Wed, 9 Sep 2026 13:11:26 -0500 Subject: [PATCH 11/15] feat(agents): let a config ask for an Intake destination it does not name Two changes to the same seam. An ATIF block turned on without storage is a request -- "I want a trajectory, you pick where" -- not a declaration, so it is filled even when another output names a destination. Without this, exporting OpenTelemetry to your own collector silently cost you the platform trajectory, recoverable only by hand-writing the endpoint and header names this wiring exists to spare people. Declared ATIF storage is still left alone, and an explicit atif.enabled=false still opts out. request.telemetry becomes request.auto_telemetry, and stops disabling an export the agent config declares. The name now matches what the flag governs: the server filling blanks, not whether the agent is traced. That is the second resolution offered in review for the flag documenting more than it did -- the first, making it force telemetry off, gave the request layer a veto over the agent config in one direction only. Saying a run should not be traced belongs to the agent config, which an inline agent can carry per request. Co-Authored-By: Claude Opus 5 (1M context) Signed-off-by: Mike Knepper --- .../src/nemo_agents_plugin/jobs/execute.py | 24 ++++------ .../telemetry/intake_export.py | 15 +++++- .../tests/unit/test_execute_job.py | 46 +++++++++++++++---- 3 files changed, 59 insertions(+), 26 deletions(-) diff --git a/plugins/nemo-agents/src/nemo_agents_plugin/jobs/execute.py b/plugins/nemo-agents/src/nemo_agents_plugin/jobs/execute.py index baf292e3a3..c5789c6c41 100644 --- a/plugins/nemo-agents/src/nemo_agents_plugin/jobs/execute.py +++ b/plugins/nemo-agents/src/nemo_agents_plugin/jobs/execute.py @@ -202,11 +202,13 @@ class ExecuteAgentJobConfig(BaseModel): gt=0, description="Maximum time to wait for Fabric to return an execution result.", ) - telemetry: bool = Field( + auto_telemetry: bool = Field( default=True, description=( - "Export the agent's trajectory to Intake. False runs untraced, overriding any export " - "the agent config declares; an agent that declares its own is otherwise left alone." + "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( @@ -415,13 +417,9 @@ def run(self, config: dict, *, ctx: JobContext, sdk: NeMoPlatform | None = None) fabric_dirs = FabricDirectories.create(agent_config, ctx.storage.ephemeral) - if not step_config.request.telemetry: - # "Run untraced" has to hold for an agent that configured its own - # export too, or the request-level switch would silently only - # govern the automatic wiring. - _disable_agent_telemetry(step_config.agent.config) - agent_config = _validate_agent_config(step_config.agent.config) - elif supports_intake_atif_export(step_config.agent.config, base_dir=fabric_dirs.base): + 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. @@ -814,12 +812,6 @@ def _validate_agent_config_format(config_format: str) -> None: ) -def _disable_agent_telemetry(agent_config: dict[str, Any]) -> None: - """Turn off any export the agent config declares, in place.""" - telemetry = agent_config.get("telemetry") - agent_config["telemetry"] = {**telemetry, "enabled": False} if isinstance(telemetry, dict) else {"enabled": False} - - def _configure_intake_telemetry( agent_config: dict[str, Any], *, diff --git a/plugins/nemo-agents/src/nemo_agents_plugin/telemetry/intake_export.py b/plugins/nemo-agents/src/nemo_agents_plugin/telemetry/intake_export.py index 6ff231f949..c7e55091f5 100644 --- a/plugins/nemo-agents/src/nemo_agents_plugin/telemetry/intake_export.py +++ b/plugins/nemo-agents/src/nemo_agents_plugin/telemetry/intake_export.py @@ -123,7 +123,7 @@ def configure_intake_atif_export( if telemetry.enabled is False: return False - if _declares_a_destination(telemetry): + if _declares_a_destination(telemetry) and not _asks_for_a_filled_atif(telemetry): return False storage: dict[str, object] = { @@ -154,6 +154,19 @@ def configure_intake_atif_export( return True +def _asks_for_a_filled_atif(telemetry: TelemetryConfig) -> bool: + """Whether the config turns ATIF on without saying where it goes. + + That combination is a request rather than a declaration -- "I want a + trajectory, you pick the destination" -- and it is how a config exporting + OpenTelemetry to its own collector also gets the platform's Intake + trajectory, which it could otherwise only have by hand-writing the endpoint + and header names this exists to spare people. + """ + atif = telemetry.atif + return isinstance(atif, dict) and atif.get("enabled") is True and not atif.get("storage") + + def _declares_a_destination(telemetry: TelemetryConfig) -> bool: """Whether the config already names anywhere to send telemetry. diff --git a/plugins/nemo-agents/tests/unit/test_execute_job.py b/plugins/nemo-agents/tests/unit/test_execute_job.py index 7a17d955d3..48f848f60f 100644 --- a/plugins/nemo-agents/tests/unit/test_execute_job.py +++ b/plugins/nemo-agents/tests/unit/test_execute_job.py @@ -42,7 +42,6 @@ ExecuteAgentStepConfig, ResolvedAgentConfig, _configure_intake_telemetry, - _disable_agent_telemetry, _log_agent_stderr, ) from nemo_agents_plugin.tasks.execute.workdir import ( @@ -1243,7 +1242,7 @@ async def _create_job(*, workspace: str, body: object) -> MagicMock: "environment": None, "workdir": {"base_workdir": "source#project", "artifact_mounts": []}, "timeout_seconds": DEFAULT_AGENT_EXECUTION_TIMEOUT_SECONDS, - "telemetry": True, + "auto_telemetry": True, "extension": None, } assert body.spec["workdir"] == {"base_workdir": "default/source#project/", "artifact_mounts": []} @@ -1962,15 +1961,24 @@ def to_dict() -> dict[str, Any]: assert "not its ATIF output" in caplog.text -def test_a_request_that_declines_telemetry_disables_an_agents_own_export() -> None: - """ "Run untraced" has to beat an export the agent config declared.""" - config = _fabric_agent_config( - telemetry={"enabled": True, "atif": {"enabled": True, "storage": [{"type": "http", "endpoint": "https://x"}]}} - ) +def test_declining_auto_telemetry_submits_the_agent_config_as_written() -> None: + """The request governs server-side filling; the agent config governs the agent. - _disable_agent_telemetry(config) + An agent that declares its own export still exports -- saying a run should + not be traced is the agent config's job, and an inline agent can say it. + """ + declared = {"enabled": True, "atif": {"enabled": True, "storage": [{"type": "http", "endpoint": "https://mine"}]}} + step = ExecuteAgentStepConfig.model_validate( + { + "request": {"agent": "a", "input": "hi", "auto_telemetry": False}, + "agent": {"name": "a", "workspace": "w", "config_format": "nemo-agents-spec-v1", "config": {}}, + } + ) - assert config["telemetry"]["enabled"] is False + assert step.request.auto_telemetry is False + # Nothing in the wiring path runs, so a declared export is untouched. + config = _fabric_agent_config(telemetry=declared) + assert config["telemetry"] == declared def test_workload_identity_jobs_export_with_a_bearer_token(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: @@ -2015,3 +2023,23 @@ def explode(**_kwargs: Any) -> Any: assert "Authorization" not in config["telemetry"]["atif"]["storage"][0].get("header_env", {}) assert "without credentials" in caplog.text + + +def test_an_atif_block_without_a_destination_is_filled_alongside_other_outputs( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Turning ATIF on without a destination asks for one, rather than declaring one. + + Otherwise exporting OpenTelemetry to your own collector would cost you the + platform trajectory, recoverable only by hand-writing the endpoint and + header names this wiring exists to spare people. + """ + monkeypatch.setenv("NMP_BASE_URL", "http://nemo-platform-api:8080") + mine = {"endpoints": [{"type": "gen_ai", "endpoint": "https://mine/otlp"}]} + config = _fabric_agent_config(telemetry={"enabled": True, "atif": {"enabled": True}, "opentelemetry": mine}) + + _configure_intake_telemetry(config, workspace="team-a", sdk=None) + + storage = config["telemetry"]["atif"]["storage"][0] + assert storage["endpoint"] == "http://nemo-platform-api:8080/apis/intake/v2/workspaces/team-a/ingest/atif" + assert config["telemetry"]["opentelemetry"] == mine, "the collector the agent chose is untouched" From 42434d18450c9d6319150d6a159b3591fb878be8 Mon Sep 17 00:00:00 2001 From: Mike Knepper Date: Wed, 9 Sep 2026 13:17:58 -0500 Subject: [PATCH 12/15] fix(agents): decide ATIF wiring from the ATIF block alone An unset atif block is no opinion about ATIF, so the Intake default applies -- the same tri-state reading telemetry.enabled already has. Declaring an OpenTelemetry collector or an ATOF sink is an opinion about that output, and previously it also silently suppressed the trajectory, which is not something the config said. This narrows the "already declared" check back to atif.storage and retires the "asks to be filled" special case: an atif block turned on without a storage is just a block with no storage, and the default applies to it like any other. Declared storage is still kept, and atif.enabled=false still opts out. Co-Authored-By: Claude Opus 5 (1M context) Signed-off-by: Mike Knepper --- .../telemetry/intake_export.py | 34 +++++-------------- .../tests/unit/test_execute_job.py | 15 ++++---- 2 files changed, 17 insertions(+), 32 deletions(-) diff --git a/plugins/nemo-agents/src/nemo_agents_plugin/telemetry/intake_export.py b/plugins/nemo-agents/src/nemo_agents_plugin/telemetry/intake_export.py index c7e55091f5..6057d7d48a 100644 --- a/plugins/nemo-agents/src/nemo_agents_plugin/telemetry/intake_export.py +++ b/plugins/nemo-agents/src/nemo_agents_plugin/telemetry/intake_export.py @@ -123,7 +123,7 @@ def configure_intake_atif_export( if telemetry.enabled is False: return False - if _declares_a_destination(telemetry) and not _asks_for_a_filled_atif(telemetry): + if _declares_atif_storage(telemetry): return False storage: dict[str, object] = { @@ -154,29 +154,13 @@ def configure_intake_atif_export( return True -def _asks_for_a_filled_atif(telemetry: TelemetryConfig) -> bool: - """Whether the config turns ATIF on without saying where it goes. +def _declares_atif_storage(telemetry: TelemetryConfig) -> bool: + """Whether the config says where its trajectory goes. - That combination is a request rather than a declaration -- "I want a - trajectory, you pick the destination" -- and it is how a config exporting - OpenTelemetry to its own collector also gets the platform's Intake - trajectory, which it could otherwise only have by hand-writing the endpoint - and header names this exists to spare people. + Only the ATIF block is consulted. Declaring an OpenTelemetry collector or an + ATOF sink is an opinion about *that* output; leaving ATIF unset is no + opinion about ATIF, so the backend's default applies -- the same tri-state + reading as ``telemetry.enabled``. A config that wants no trajectory says so + with ``atif.enabled: false``. """ - atif = telemetry.atif - return isinstance(atif, dict) and atif.get("enabled") is True and not atif.get("storage") - - -def _declares_a_destination(telemetry: TelemetryConfig) -> bool: - """Whether the config already names anywhere to send telemetry. - - Checked across every output, not just ATIF: a config that exports - OpenTelemetry to its own collector has declared where its telemetry goes, - and adding a second destination it never asked for is the opposite of - letting an explicit declaration win. - """ - return bool( - (isinstance(telemetry.atif, dict) and telemetry.atif.get("storage")) - or (isinstance(telemetry.atof, dict) and telemetry.atof.get("sinks")) - or (isinstance(telemetry.opentelemetry, dict) and telemetry.opentelemetry.get("endpoints")) - ) + return isinstance(telemetry.atif, dict) and bool(telemetry.atif.get("storage")) diff --git a/plugins/nemo-agents/tests/unit/test_execute_job.py b/plugins/nemo-agents/tests/unit/test_execute_job.py index 48f848f60f..1c82254b4b 100644 --- a/plugins/nemo-agents/tests/unit/test_execute_job.py +++ b/plugins/nemo-agents/tests/unit/test_execute_job.py @@ -1858,20 +1858,21 @@ def test_telemetry_credentials_go_to_the_environment_not_the_config(monkeypatch: ("opentelemetry", {"endpoints": [{"type": "gen_ai", "endpoint": "https://mine/otlp"}]}), ], ) -def test_a_destination_declared_through_any_output_is_left_alone( +def test_another_outputs_destination_does_not_speak_for_atif( monkeypatch: pytest.MonkeyPatch, output: str, declaration: dict[str, Any] ) -> None: - """Declaring an export is declaring where telemetry goes, whichever output carries it. + """Declaring a collector is an opinion about that output, not about the trajectory. - Adding an Intake destination beside one the agent chose would be a second, - unrequested export -- not the "fill in what was left out" this is for. + Leaving ATIF unset is no opinion about ATIF, so the Intake default applies + and the agent keeps the destination it did choose. """ monkeypatch.setenv("NMP_BASE_URL", "http://nemo-platform-api:8080") config = _fabric_agent_config(telemetry={"enabled": True, output: declaration}) - _configure_intake_telemetry(config, workspace="default", sdk=None) + _configure_intake_telemetry(config, workspace="team-a", sdk=None) - assert "atif" not in config["telemetry"] + storage = config["telemetry"]["atif"]["storage"][0] + assert storage["endpoint"] == "http://nemo-platform-api:8080/apis/intake/v2/workspaces/team-a/ingest/atif" assert config["telemetry"][output] == declaration @@ -2025,7 +2026,7 @@ def explode(**_kwargs: Any) -> Any: assert "without credentials" in caplog.text -def test_an_atif_block_without_a_destination_is_filled_alongside_other_outputs( +def test_an_atif_block_turned_on_without_a_destination_is_filled( monkeypatch: pytest.MonkeyPatch, ) -> None: """Turning ATIF on without a destination asks for one, rather than declaring one. From f334e4610fb405ef9bf365d1c22f1ebfc5f49f55 Mon Sep 17 00:00:00 2001 From: Mike Knepper Date: Wed, 9 Sep 2026 13:24:35 -0500 Subject: [PATCH 13/15] sort Signed-off-by: Mike Knepper --- plugins/nemo-insights/pyproject.toml | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/plugins/nemo-insights/pyproject.toml b/plugins/nemo-insights/pyproject.toml index 34feb53bd0..dd844c7fb2 100644 --- a/plugins/nemo-insights/pyproject.toml +++ b/plugins/nemo-insights/pyproject.toml @@ -11,10 +11,9 @@ dependencies = [ "httpx", "nemo-fabric-adapter-contract>=0.3.0b1", "nemo-fabric-adapters-common>=0.3.0b1", - # Imported by the Fabric adapter to activate the Relay config Fabric resolves. - "nemo-relay>=0.7,<0.8", "nemo-platform", "nemo-platform-plugin", + "nemo-relay>=0.7,<0.8", "nooa>=0.0.9", "opentelemetry-exporter-otlp>=1.42.1", "opentelemetry-sdk>=1.42.1", From daf95298f61c15bc63d44880fd2860bb8471f1b9 Mon Sep 17 00:00:00 2001 From: Mike Knepper Date: Wed, 9 Sep 2026 13:29:37 -0500 Subject: [PATCH 14/15] lint Signed-off-by: Mike Knepper --- docs/cli/reference.mdx | 2 +- packages/nemo_platform/pyproject.toml | 1 + plugins/nemo-agents/openapi/openapi.yaml | 11 ++++++----- uv.lock | 8 ++++++++ 4 files changed, 16 insertions(+), 6 deletions(-) diff --git a/docs/cli/reference.mdx b/docs/cli/reference.mdx index 778b0174b2..544fa0f4e8 100644 --- a/docs/cli/reference.mdx +++ b/docs/cli/reference.mdx @@ -6828,7 +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 `: 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. +* `--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:** diff --git a/packages/nemo_platform/pyproject.toml b/packages/nemo_platform/pyproject.toml index 42126a8870..543482a449 100644 --- a/packages/nemo_platform/pyproject.toml +++ b/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", diff --git a/plugins/nemo-agents/openapi/openapi.yaml b/plugins/nemo-agents/openapi/openapi.yaml index 13df7d7a3a..b5d8f14fba 100644 --- a/plugins/nemo-agents/openapi/openapi.yaml +++ b/plugins/nemo-agents/openapi/openapi.yaml @@ -5707,12 +5707,13 @@ components: title: Timeout Seconds description: Maximum time to wait for Fabric to return an execution result. default: 3600 - telemetry: + auto_telemetry: type: boolean - title: Telemetry - description: "Export the agent's trajectory to Intake. Set false to run\ - \ untraced, or configure 'telemetry' on the agent yourself \u2014 an agent\ - \ that already declares it is left alone." + title: Auto Telemetry + 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. default: true extension: allOf: diff --git a/uv.lock b/uv.lock index 9345d1a066..f039dc1b53 100644 --- a/uv.lock +++ b/uv.lock @@ -5036,6 +5036,7 @@ all = [ { name = "nemo-optimization-plugin", marker = "(platform_machine == 'arm64' and sys_platform == 'darwin') or (platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux')" }, { name = "nemo-platform-plugin", marker = "(platform_machine == 'arm64' and sys_platform == 'darwin') or (platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux')" }, { name = "nemo-platform-sdk", marker = "(platform_machine == 'arm64' and sys_platform == 'darwin') or (platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux')" }, + { name = "nemo-relay", marker = "(platform_machine == 'arm64' and sys_platform == 'darwin') or (platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux')" }, { name = "nemo-safe-synthesizer", marker = "(platform_machine == 'arm64' and sys_platform == 'darwin') or (platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux')" }, { name = "nemoguardrails", extra = ["tracing"], marker = "(platform_machine == 'arm64' and sys_platform == 'darwin') or (platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux')" }, { name = "ngcsdk", marker = "(platform_machine == 'arm64' and sys_platform == 'darwin') or (platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux')" }, @@ -5402,6 +5403,7 @@ nemo-insights-plugin = [ { name = "nemo-fabric-adapter-contract", marker = "(platform_machine == 'arm64' and sys_platform == 'darwin') or (platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux')" }, { name = "nemo-fabric-adapters-common", marker = "(platform_machine == 'arm64' and sys_platform == 'darwin') or (platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux')" }, { name = "nemo-platform-plugin", marker = "(platform_machine == 'arm64' and sys_platform == 'darwin') or (platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux')" }, + { name = "nemo-relay", marker = "(platform_machine == 'arm64' and sys_platform == 'darwin') or (platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux')" }, { name = "nooa", marker = "(platform_machine == 'arm64' and sys_platform == 'darwin') or (platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux')" }, { name = "opentelemetry-exporter-otlp", marker = "(platform_machine == 'arm64' and sys_platform == 'darwin') or (platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux')" }, { name = "opentelemetry-sdk", marker = "(platform_machine == 'arm64' and sys_platform == 'darwin') or (platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux')" }, @@ -5567,6 +5569,7 @@ plugins = [ { name = "nemo-optimization-plugin", marker = "(platform_machine == 'arm64' and sys_platform == 'darwin') or (platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux')" }, { name = "nemo-platform-plugin", marker = "(platform_machine == 'arm64' and sys_platform == 'darwin') or (platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux')" }, { name = "nemo-platform-sdk", marker = "(platform_machine == 'arm64' and sys_platform == 'darwin') or (platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux')" }, + { name = "nemo-relay", marker = "(platform_machine == 'arm64' and sys_platform == 'darwin') or (platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux')" }, { name = "nemo-safe-synthesizer", marker = "(platform_machine == 'arm64' and sys_platform == 'darwin') or (platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux')" }, { name = "nemoguardrails", extra = ["tracing"], marker = "(platform_machine == 'arm64' and sys_platform == 'darwin') or (platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux')" }, { name = "nmp-automodel", marker = "(platform_machine == 'arm64' and sys_platform == 'darwin') or (platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux')" }, @@ -5671,6 +5674,7 @@ services = [ { name = "nemo-optimization-plugin", marker = "(platform_machine == 'arm64' and sys_platform == 'darwin') or (platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux')" }, { name = "nemo-platform-plugin", marker = "(platform_machine == 'arm64' and sys_platform == 'darwin') or (platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux')" }, { name = "nemo-platform-sdk", marker = "(platform_machine == 'arm64' and sys_platform == 'darwin') or (platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux')" }, + { name = "nemo-relay", marker = "(platform_machine == 'arm64' and sys_platform == 'darwin') or (platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux')" }, { name = "nemo-safe-synthesizer", marker = "(platform_machine == 'arm64' and sys_platform == 'darwin') or (platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux')" }, { name = "nemoguardrails", extra = ["tracing"], marker = "(platform_machine == 'arm64' and sys_platform == 'darwin') or (platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux')" }, { name = "ngcsdk", marker = "(platform_machine == 'arm64' and sys_platform == 'darwin') or (platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux')" }, @@ -6104,7 +6108,11 @@ requires-dist = [ { name = "nemo-platform-sdk", marker = "extra == 'rl-service'", editable = "sdk/python/nemo-platform" }, { name = "nemo-platform-sdk", marker = "extra == 'services'", editable = "sdk/python/nemo-platform" }, { name = "nemo-platform-sdk", marker = "extra == 'unsloth-service'", editable = "sdk/python/nemo-platform" }, + { name = "nemo-relay", marker = "extra == 'all'", specifier = ">=0.7,<0.8" }, { name = "nemo-relay", marker = "extra == 'nemo-evaluator-sdk'", specifier = ">=0.7.3,<0.8" }, + { name = "nemo-relay", marker = "extra == 'nemo-insights-plugin'", specifier = ">=0.7,<0.8" }, + { name = "nemo-relay", marker = "extra == 'plugins'", specifier = ">=0.7,<0.8" }, + { name = "nemo-relay", marker = "extra == 'services'", specifier = ">=0.7,<0.8" }, { name = "nemo-safe-synthesizer", marker = "extra == 'all'", specifier = "==0.1.7" }, { name = "nemo-safe-synthesizer", marker = "extra == 'nemo-safe-synthesizer-plugin'", specifier = "==0.1.7" }, { name = "nemo-safe-synthesizer", marker = "extra == 'plugins'", specifier = "==0.1.7" }, From da337332ed2c7778035f73ca1abbf94923757fd0 Mon Sep 17 00:00:00 2001 From: Mike Knepper Date: Wed, 9 Sep 2026 14:32:26 -0500 Subject: [PATCH 15/15] Shore up a test Signed-off-by: Mike Knepper --- .../tests/unit/test_execute_job.py | 41 +++++++++++++------ 1 file changed, 28 insertions(+), 13 deletions(-) diff --git a/plugins/nemo-agents/tests/unit/test_execute_job.py b/plugins/nemo-agents/tests/unit/test_execute_job.py index 1c82254b4b..0072e3c56d 100644 --- a/plugins/nemo-agents/tests/unit/test_execute_job.py +++ b/plugins/nemo-agents/tests/unit/test_execute_job.py @@ -1962,24 +1962,39 @@ def to_dict() -> dict[str, Any]: assert "not its ATIF output" in caplog.text -def test_declining_auto_telemetry_submits_the_agent_config_as_written() -> None: +def test_declining_auto_telemetry_submits_the_agent_config_as_written( + ctx: JobContext, monkeypatch: pytest.MonkeyPatch +) -> None: """The request governs server-side filling; the agent config governs the agent. An agent that declares its own export still exports -- saying a run should not be traced is the agent config's job, and an inline agent can say it. + Driven through ``run`` so the opt-out branch is what is under test, rather + than the request parsing around it. """ - declared = {"enabled": True, "atif": {"enabled": True, "storage": [{"type": "http", "endpoint": "https://mine"}]}} - step = ExecuteAgentStepConfig.model_validate( - { - "request": {"agent": "a", "input": "hi", "auto_telemetry": False}, - "agent": {"name": "a", "workspace": "w", "config_format": "nemo-agents-spec-v1", "config": {}}, - } - ) - - assert step.request.auto_telemetry is False - # Nothing in the wiring path runs, so a declared export is untouched. - config = _fabric_agent_config(telemetry=declared) - assert config["telemetry"] == declared + monkeypatch.setenv("NMP_BASE_URL", "http://nemo-platform-api:8080") + # ATIF is left unset, so this config *would* be wired -- otherwise the test + # would pass whether or not auto_telemetry was honoured. + declared = {"enabled": True, "opentelemetry": {"endpoints": [{"type": "gen_ai", "endpoint": "https://mine"}]}} + agent = _resolved_agent() + agent.config["telemetry"] = declared + spec = ExecuteAgentStepConfig( + request=ExecuteAgentJobConfig(agent="calc", input="hello", auto_telemetry=False), + agent=agent, + ) + seen: dict[str, Any] = {} + + async def _invoke(request: Any) -> FabricRuntimeResult: + seen["telemetry"] = request.agent_config.telemetry.model_dump(exclude_none=True) + return FabricRuntimeResult(status="succeeded", output={"answer": "done"}) + + with patch("nemo_agents_plugin.jobs.execute.invoke_agent_config_request_once", _invoke): + result = ExecuteAgentJob().run(spec.model_dump(mode="json"), ctx=ctx, sdk=MagicMock()) + + assert result["status"] == "completed" + # Reached Fabric exactly as declared: no Intake destination added beside it. + assert "atif" not in seen["telemetry"] + assert seen["telemetry"]["opentelemetry"] == declared["opentelemetry"] def test_workload_identity_jobs_export_with_a_bearer_token(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: