diff --git a/docs/cli/reference.mdx b/docs/cli/reference.mdx index ea6696b9b1..544fa0f4e8 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. +* `--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/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/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/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 ecdc8e31c2..b5d8f14fba 100644 --- a/plugins/nemo-agents/openapi/openapi.yaml +++ b/plugins/nemo-agents/openapi/openapi.yaml @@ -5707,6 +5707,14 @@ components: title: Timeout Seconds description: Maximum time to wait for Fabric to return an execution result. default: 3600 + auto_telemetry: + type: boolean + 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: - $ref: '#/components/schemas/ExecuteAgentExtensionConfig' 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..c5789c6c41 100644 --- a/plugins/nemo-agents/src/nemo_agents_plugin/jobs/execute.py +++ b/plugins/nemo-agents/src/nemo_agents_plugin/jobs/execute.py @@ -48,7 +48,16 @@ materialize_agent_workdir, validate_agent_workdir, ) +from nemo_agents_plugin.telemetry.intake_export import ( + configure_intake_atif_export, + supports_intake_atif_export, +) from nemo_platform import AsyncNeMoPlatform, NeMoPlatform +from nemo_platform_plugin.client.constants import ( + WORKLOAD_IDENTITY_TOKEN_FILE_ENVVAR, + is_workload_identity_token_file_set, +) +from nemo_platform_plugin.client.oidc_factory import resolve_workload_exchange_provider from nemo_platform_plugin.entity_client import NemoEntityNotFoundError from nemo_platform_plugin.job import NemoJob from nemo_platform_plugin.job_context import JobContext @@ -80,6 +89,7 @@ from nemo_platform_plugin.jobs.exceptions import PlatformJobCompilationError from nemo_platform_plugin.jobs.image import get_qualified_image from nemo_platform_plugin.refs import ENTITY_REF_PATTERN, parse_entity_ref +from nemo_platform_plugin.sdk_provider import get_forwarding_headers from pydantic import BaseModel, Field, field_validator logger = logging.getLogger(__name__) @@ -89,6 +99,8 @@ INPUT_WORKDIR_RESULT_NAME = "input_workdir" OUTPUT_WORKDIR_RESULT_NAME = "output_workdir" OUTPUT_ARTIFACTS_RESULT_NAME = "output_artifacts" +NMP_BASE_URL_ENVVAR = "NMP_BASE_URL" +HEADER_ENVVAR_PREFIX = "NMP_AGENT_TELEMETRY_HEADER_" FABRIC_RUN_RESULT_NAME = "fabric_run_result" FABRIC_ERROR_RESULT_NAME = "fabric_error" FABRIC_RUN_RESULT_FILENAME = "fabric_run_result.json" @@ -190,6 +202,15 @@ class ExecuteAgentJobConfig(BaseModel): gt=0, description="Maximum time to wait for Fabric to return an execution result.", ) + auto_telemetry: bool = Field( + default=True, + description=( + "Let the server fill in the agent's telemetry export -- an Intake destination for a " + "config that asks for one and does not say where. False submits the agent config as " + "written, which still exports if the config says to; an agent config is the place to " + "say a run should not be traced." + ), + ) extension: ExecuteAgentExtensionConfig | None = Field( default=None, description="Optional trusted plugin extension to run during the execute-agent lifecycle.", @@ -396,6 +417,14 @@ def run(self, config: dict, *, ctx: JobContext, sdk: NeMoPlatform | None = None) fabric_dirs = FabricDirectories.create(agent_config, ctx.storage.ephemeral) + if step_config.request.auto_telemetry and supports_intake_atif_export( + step_config.agent.config, base_dir=fabric_dirs.base + ): + _configure_intake_telemetry(step_config.agent.config, workspace=ctx.workspace, sdk=sdk) + # Wiring mutates the config mapping, not the model validated above, + # so re-validate to carry it into what Fabric is handed. + agent_config = _validate_agent_config(step_config.agent.config) + if step_config.workdir is not None and _has_workdir_inputs(step_config.workdir): if sdk is None: raise RuntimeError("sdk is required to stage workdir inputs.") @@ -783,6 +812,77 @@ def _validate_agent_config_format(config_format: str) -> None: ) +def _configure_intake_telemetry( + agent_config: dict[str, Any], + *, + workspace: str, + sdk: NeMoPlatform | None, +) -> None: + """Wire the agent's trajectory export to Intake for this job. + + Runs here rather than at create time because only the task knows both + halves: ``NMP_BASE_URL`` is the platform URL reachable from *this* pod (the + Jobs service rewrites it per runtime), and the task's own SDK carries the + identity the platform gave this job -- the same ``service:agents`` principal + and on-behalf-of delegation a deployment gets from its auth-proxy sidecar. + + Credentials go in the process environment and the config names them. + Fabric writes the resolved agent config into the run's artifacts, and those + are uploaded as a job result, so an inline header would be a downloadable + one. + """ + base_url = os.environ.get(NMP_BASE_URL_ENVVAR) + if not base_url: + logger.warning("%s is not set; the agent will run untraced.", NMP_BASE_URL_ENVVAR) + return + + headers = get_forwarding_headers(sdk) if sdk is not None else {} + headers.update(_workload_identity_headers(base_url)) + for name, value in headers.items(): + os.environ[_header_envvar(name)] = value + + configure_intake_atif_export( + agent_config, + workspace=workspace, + base_url=base_url, + header_env={name: _header_envvar(name) for name in headers}, + ) + + +def _workload_identity_headers(base_url: str) -> dict[str, str]: + """Bearer credentials for Relay when the job runs under workload identity. + + ``get_forwarding_headers`` returns only what the SDK was *constructed* with. + Under workload identity that is the internal marker alone -- the bearer is + exchanged per request by the SDK's own auth layer, which Relay's raw POST to + Intake does not go through. Without this the export would be unauthenticated + on exactly the deployments that enforce auth. + + The token is resolved once and read from the environment at export time, so + a run outliving its token exports with an expired one. Relay resolves + ``header_env`` statically, so refreshing needs a dynamic-credential hook it + does not offer today. + """ + if not is_workload_identity_token_file_set(): + return {} + token_file = os.environ[WORKLOAD_IDENTITY_TOKEN_FILE_ENVVAR] + try: + provider = resolve_workload_exchange_provider(base_url=base_url, subject_token_file=Path(token_file)) + return {"Authorization": f"Bearer {provider.get_access_token()}"} + except Exception: + logger.warning( + "Could not exchange the workload identity token for telemetry export; " + "the trajectory will be posted without credentials.", + exc_info=True, + ) + return {} + + +def _header_envvar(header_name: str) -> str: + """Environment variable the exporter reads one outbound header value from.""" + return f"{HEADER_ENVVAR_PREFIX}{header_name.upper().replace('-', '_')}" + + def _validate_agent_config(config: dict) -> AgentConfig: agent_config = AgentConfig.model_validate(config) if agent_config.environment.provider != "local": 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..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,6 +38,10 @@ FabricArtifactStagingError, stage_fabric_ethos_config_files, ) +from nemo_agents_plugin.telemetry.intake_export import ( + configure_intake_atif_export, + supports_intake_atif_export, +) from nemo_agents_plugin.utils import get_base_url, get_internal_base_url from nemo_deployments_plugin.auth_proxy import auth_proxy_port from nemo_deployments_plugin.entities import ( @@ -668,6 +672,15 @@ async def create_deployment( rewrite_target = gateway if is_fabric: + # Wire the trajectory export before the rebase below, so the + # endpoint it produces is rewritten for reachability alongside the + # inference URLs. No header_env here: the auth-proxy sidecar stamps + # identity on the way out, which is why the workload needs none. + # config is the caller's deployment entity; rewrite_fabric_config_base_urls + # deep-copies for the same reason, and wiring runs before it. + if supports_intake_atif_export(config): + config = copy.deepcopy(config) + configure_intake_atif_export(config, workspace=workspace, base_url=rewrite_target) config = rewrite_fabric_config_base_urls(config, rewrite_target) else: config = rewrite_config_base_urls(config, rewrite_target) 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..6057d7d48a --- /dev/null +++ b/plugins/nemo-agents/src/nemo_agents_plugin/telemetry/intake_export.py @@ -0,0 +1,166 @@ +# 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 +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 + +logger = logging.getLogger(__name__) + +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], + *, + 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 {}) + 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] + + 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 says where its trajectory goes. + + 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``. + """ + 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..0072e3c56d 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, @@ -27,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, @@ -39,6 +41,7 @@ ExecuteAgentJobConfig, ExecuteAgentStepConfig, ResolvedAgentConfig, + _configure_intake_telemetry, _log_agent_stderr, ) from nemo_agents_plugin.tasks.execute.workdir import ( @@ -48,6 +51,9 @@ 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 from nemo_platform_plugin.job_context import JobContext @@ -1236,6 +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, + "auto_telemetry": True, "extension": None, } assert body.spec["workdir"] == {"base_workdir": "default/source#project/", "artifact_mounts": []} @@ -1789,3 +1796,266 @@ 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") + # 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() + + _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[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_another_outputs_destination_does_not_speak_for_atif( + monkeypatch: pytest.MonkeyPatch, output: str, declaration: dict[str, Any] +) -> None: + """Declaring a collector is an opinion about that output, not about the trajectory. + + 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="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"][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") + 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} + + +def test_relay_support_is_read_from_the_adapter_descriptor(tmp_path: Path) -> None: + """The bundled harnesses advertise relay with an ATIF output.""" + assert supports_intake_atif_export(_fabric_agent_config(), base_dir=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"}}) + + 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_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. + """ + 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: + """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): + _configure_intake_telemetry(config, workspace="default", sdk=None) + + assert "Authorization" not in config["telemetry"]["atif"]["storage"][0].get("header_env", {}) + assert "without credentials" in caplog.text + + +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. + + 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" 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 diff --git a/plugins/nemo-agents/tests/unit/test_runner_deployments.py b/plugins/nemo-agents/tests/unit/test_runner_deployments.py index cec3bba44d..6e6f66accd 100644 --- a/plugins/nemo-agents/tests/unit/test_runner_deployments.py +++ b/plugins/nemo-agents/tests/unit/test_runner_deployments.py @@ -1144,6 +1144,49 @@ 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", + # 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"): + 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/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/pyproject.toml b/plugins/nemo-insights/pyproject.toml index 0cdaff3fd7..dd844c7fb2 100644 --- a/plugins/nemo-insights/pyproject.toml +++ b/plugins/nemo-insights/pyproject.toml @@ -13,6 +13,7 @@ dependencies = [ "nemo-fabric-adapters-common>=0.3.0b1", "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", 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..138cc2dfc5 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,13 @@ from __future__ import annotations +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 from nemo_fabric_adapter_contract import models as contract @@ -16,9 +20,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 +48,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 +72,32 @@ 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 -- 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") if target_agent is None: raise AnalystAdapterConfigError("harness.settings.agent is required for the Insights analyst adapter") @@ -102,6 +133,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 +142,40 @@ 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. + + ``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..2c11c32102 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 -from typing import Any, cast - +import json +import os +from collections.abc import AsyncIterator +from contextlib import asynccontextmanager +from pathlib import Path +from typing import Any +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 @@ -159,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 @@ -190,7 +214,108 @@ 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 + + +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 + # 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) + 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 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: + """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 diff --git a/uv.lock b/uv.lock index b7e13dfe90..f039dc1b53 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" }, @@ -5034,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')" }, @@ -5400,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')" }, @@ -5565,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')" }, @@ -5669,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')" }, @@ -6102,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" }, @@ -12362,7 +12372,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" }, ]