From 4ee3fd3d9b1f1d9b665a1407e1d3a8fcc65e91a3 Mon Sep 17 00:00:00 2001 From: Will Killian Date: Thu, 16 Jul 2026 15:54:58 -0400 Subject: [PATCH] docs: document multi-sink ATOF export Signed-off-by: Will Killian --- docs/configure-plugins/observability/atif.mdx | 4 +- docs/configure-plugins/observability/atof.mdx | 275 ++++++++++-------- .../observability/configuration.mdx | 257 ++++++++-------- .../plugin-configuration-files.mdx | 46 ++- scripts/test-claude-plugin-e2e.sh | 5 +- scripts/test-codex-plugin-e2e.sh | 5 +- scripts/test-hermes-mcp-e2e.sh | 5 +- .../references/cli-try-now.md | 5 +- .../references/atof.md | 31 +- 9 files changed, 366 insertions(+), 267 deletions(-) diff --git a/docs/configure-plugins/observability/atif.mdx b/docs/configure-plugins/observability/atif.mdx index 1063f1d33..096282b50 100644 --- a/docs/configure-plugins/observability/atif.mdx +++ b/docs/configure-plugins/observability/atif.mdx @@ -28,7 +28,7 @@ kind = "observability" enabled = true [components.config] -version = 1 +version = 2 [components.config.atif] enabled = true @@ -300,7 +300,7 @@ void (async () => { version: 1, components: [ observability.ComponentSpec({ - version: 1, + version: 2, atif: observability.atifConfig({ enabled: true, agent_name: "Planner", diff --git a/docs/configure-plugins/observability/atof.mdx b/docs/configure-plugins/observability/atof.mdx index cd3f41b0c..f3013e665 100644 --- a/docs/configure-plugins/observability/atof.mdx +++ b/docs/configure-plugins/observability/atof.mdx @@ -1,20 +1,19 @@ --- title: "Agent Trajectory Observability Format (ATOF)" -description: "Configure raw ATOF event export to JSONL files and streaming endpoints." +description: "Configure raw ATOF event export to JSONL files and stream sinks." position: 4 --- {/* SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. SPDX-License-Identifier: Apache-2.0 */} -Use the `atof` section when you want the raw Agent Trajectory Observability -Format (ATOF) `0.1` event stream written as JSONL or streamed to raw-event -collectors. +Use the `atof` section to write raw Agent Trajectory Observability Format (ATOF) +`0.1` events to JSONL files, send them to raw-event collectors, or do both. -ATOF JSONL export is useful for local debugging, offline inspection, and -preserving the canonical event stream before it is translated into another -format. Streaming endpoints are useful when a collector wants the same raw -event shape in near real time. +File sinks are useful for local debugging, offline inspection, and preserving +the canonical event stream before another exporter translates it. Stream sinks +send the same events to collectors in near real time. They can also transform +dotted object keys before delivery. ## `plugins.toml` Example @@ -46,11 +45,19 @@ url = "http://localhost:8080/events" transport = "http_post" timeout_millis = 3000 field_name_policy = "replace_dots" + +[components.config.atof.sinks.header_env] +authorization = "NEMO_RELAY_ATOF_AUTH_HEADER" ``` -This configuration registers the plugin-managed ATOF exporter, writes one JSON -object per lifecycle event to `logs/events.jsonl`, and sends each raw event to -the configured stream sink. +Before you validate or activate this example, set +`NEMO_RELAY_ATOF_AUTH_HEADER` to the complete authorization header value, such +as `Bearer `. NeMo Relay rejects a missing or blank value. + +This configuration registers the plugin-managed ATOF exporter and writes one +JSON object per lifecycle event to `logs/events.jsonl`. It also applies the +`replace_dots` field-name policy and sends each raw event to the named `archive` +stream sink. ## Fields @@ -59,47 +66,51 @@ The following table describes the top-level ATOF settings: | Field | Default | Notes | |---|---|---| | `enabled` | `false` | Must be `true` to export events. | -| `sinks` | `[]` | File and stream destinations. An enabled ATOF section requires at least one sink. | +| `sinks` | `[]` | File and stream destinations. An enabled ATOF section requires at least one sink, and every sink receives each event. | ## File Sinks -The following table describes each file sink: +Each file sink writes the raw event stream to one JSONL file. The following +table describes each file sink: | Field | Default | Notes | |---|---|---| -| `type` | Required | Must be `file`. | +| `type` | Required | Sink variant. Must be `file`. | | `output_directory` | Current working directory | Directory containing the JSONL file. | | `filename` | Timestamped `nemo-relay-events-*.jsonl` | Output filename. | | `mode` | `append` | `append` or `overwrite`. | -## Streaming Sinks +## Stream Sinks -Each stream sink receives the same raw ATOF JSON object that file sinks write as -one JSONL line. Sinks are independent: a failed stream sink is skipped or -retried without blocking file output or other stream sinks. +Each stream sink receives the same canonical ATOF event that file sinks write. +Before sending it, NeMo Relay applies the configured field-name policy and then +uses the transport-specific encoding. WebSocket sends a JSON text message, +while the HTTP transports send a JSONL record. Each stream runs independently, +so a failed connection does not block file output or other streams. The following table describes each stream sink: | Field | Default | Notes | |---|---|---| -| `type` | Required | Must be `stream`. | -| `name` | None | Optional unique name that another component can use to reference this stream sink. Names must be non-empty and cannot have leading or trailing whitespace. | -| `url` | Required | Endpoint URL. | +| `type` | Required | Sink variant. Must be `stream`. | +| `name` | Omitted | Optional stable name for cross-component references. Stream sink names must be unique, non-empty, and have no leading or trailing whitespace. | +| `url` | Required | Stream URL. Use `http` or `https` for `http_post` and `ndjson`, and `ws` or `wss` for `websocket`. | | `transport` | `http_post` | `http_post`, `websocket`, or `ndjson`. | | `headers` | `{}` | String-to-string headers for requests or handshakes. | -| `header_env` | `{}` | Map of header names to environment-variable names. Referenced values must be present and non-blank. | +| `header_env` | `{}` | Maps each header name to an environment variable that contains the full header value. Each variable must be set and nonblank, and the same header cannot also appear in `headers`. | | `timeout_millis` | `3000` | Per-sink timeout. Must be greater than `0`. | -| `field_name_policy` | `preserve` | Field-name handling before Relay sends stream sink events. Accepted values are `preserve` and `replace_dots`. | +| `field_name_policy` | `preserve` | Field-name handling before Relay sends stream events. Accepted values are `preserve` and `replace_dots`. | -Do not define the same header in both `headers` and `header_env`. Use -`header_env` for secrets so configuration stores the environment-variable name -instead of the secret value. +Use `header_env` for secrets so the configuration stores the environment +variable name instead of the secret value. `preserve` sends canonical ATOF field names unchanged. `replace_dots` replaces -dots in JSON object keys with underscores recursively. If replacement produces -a collision, Relay keeps both values and deterministically appends `_2`, `_3`, -and later suffixes as needed. Validation rejects any other -`sinks[i].field_name_policy` value. +dots in JSON object keys with underscores recursively. If that creates a +duplicate key, NeMo Relay keeps both values and appends `_2`, `_3`, and later +suffixes in a deterministic order. Validation rejects any other +`sinks[i].field_name_policy` value for a stream sink. + +The following transports are supported: - `http_post` sends each event as one JSONL record in an HTTP `POST` request with `Content-Type: application/x-ndjson`. Any `2xx` response is treated as @@ -109,14 +120,32 @@ and later suffixes as needed. Validation rejects any other - `ndjson` opens one long-lived HTTP upload and writes each event as one newline-delimited JSON record. -`force_flush()` flushes file output and drains queued stream sink events without -closing streaming connections. `shutdown()` is terminal: it flushes pending -work, closes streaming connections, and makes later events no-op. +`force_flush()` flushes file output and drains queued stream events while +keeping stream connections open. `shutdown()` is terminal: it flushes pending +work, closes the connections, and ignores events that arrive later. + +## Migrate From Version 1 + +Observability component configuration now requires `version = 2`. NeMo Relay +does not translate the removed ATOF fields from version 1. + +Make the following changes when you migrate a plugin configuration: + +- Move `output_directory`, `filename`, and `mode` from + `[components.config.atof]` into a sink with `type = "file"`. +- Replace each `[[components.config.atof.endpoints]]` entry with a + `[[components.config.atof.sinks]]` entry that has `type = "stream"`. +- Keep static request headers in `headers`, or use `header_env` to read complete + header values from environment variables at activation time. + +The plugin sends every event to every configured sink. The manual +`AtofExporter` API is different: each exporter owns one file or stream sink. ## Expected Output -Each emitted scope, tool, LLM, middleware, or mark event is written as one ATOF -JSON object per line. For event field semantics, refer to +Each file sink writes an emitted scope, tool, LLM, middleware, or mark event as +one ATOF JSON object per line. Stream sinks encode the event as described in +[Stream Sinks](#stream-sinks). For event field semantics, refer to [Events](/about-nemo-relay/concepts/events). Coding-agent sessions emit a `session.start` mark with the logical @@ -136,14 +165,14 @@ event-only encoding fails, Relay retains the complete event input and annotation. The request used for provider execution remains unchanged. Refer to [Events](/about-nemo-relay/concepts/events) for the complete freshness contract. -Register the plugin before instrumented work starts and clear it during -shutdown so file handles flush. +Register the plugin before instrumented work starts. Clear it during shutdown +so every sink flushes pending events and stream connections close cleanly. ## Plugin Configuration -Use plugin configuration when the application should let NeMo Relay own the ATOF -exporter lifecycle. The following examples configure and activate the ATOF -exporter through each supported language binding. +Use plugin configuration when you want NeMo Relay to manage the ATOF exporter +lifecycle. The following examples configure and activate the exporter through +each supported language binding. `validate()` checks only the supplied in-memory object. `initialize()` also layers discovered `plugins.toml` configuration. For effective file-backed @@ -179,7 +208,7 @@ config = plugin.PluginConfig( AtofStreamSinkConfig( url="http://localhost:8080/events", transport="http_post", - ) + ), ], ) ) @@ -211,28 +240,28 @@ const observability = require("nemo-relay-node/observability"); void (async () => { await plugin.initialize({ - version: 1, - components: [ - observability.ComponentSpec({ - version: 2, - atof: observability.atofConfig({ - enabled: true, - sinks: [ - { - type: "file", - output_directory: "logs", - filename: "events.jsonl", - mode: "overwrite", - }, - { - type: "stream", - url: "http://localhost:8080/events", - transport: "http_post", - }, - ], + version: 1, + components: [ + observability.ComponentSpec({ + version: 2, + atof: observability.atofConfig({ + enabled: true, + sinks: [ + { + type: "file", + output_directory: "logs", + filename: "events.jsonl", + mode: "overwrite", + }, + { + type: "stream", + url: "http://localhost:8080/events", + transport: "http_post", + }, + ], + }), }), - }), - ], + ], }); try { @@ -250,8 +279,6 @@ void (async () => { ```rust -#[tokio::main] -async fn main() -> Result<(), Box> { use nemo_relay::observability::plugin_component::{ AtofFileSinkSectionConfig, AtofSectionConfig, AtofSinkSectionConfig, AtofStreamSinkSectionConfig, ComponentSpec, ObservabilityConfig, @@ -260,44 +287,46 @@ use nemo_relay::plugin::{ clear_plugin_configuration, initialize_plugins, validate_plugin_config, PluginConfig, }; -let component = ComponentSpec::new(ObservabilityConfig { - atof: Some(AtofSectionConfig { - enabled: true, - sinks: vec![ - AtofSinkSectionConfig::File(AtofFileSinkSectionConfig { - output_directory: Some("logs".into()), - filename: Some("events.jsonl".into()), - mode: "overwrite".into(), - }), - AtofSinkSectionConfig::Stream(AtofStreamSinkSectionConfig { - url: "http://localhost:8080/events".into(), - transport: "http_post".into(), - headers: Default::default(), - header_env: Default::default(), - timeout_millis: 3000, - field_name_policy: "replace_dots".into(), - name: Some("archive".into()), - }), - ], - }), - ..ObservabilityConfig::default() -}); - -let config = PluginConfig { - version: 1, - components: vec![component.into()], - policy: Default::default(), -}; - -let report = validate_plugin_config(&config); -assert!(!report.has_errors()); - -let _active = initialize_plugins(config).await?; +#[tokio::main] +async fn main() -> Result<(), Box> { + let component = ComponentSpec::new(ObservabilityConfig { + atof: Some(AtofSectionConfig { + enabled: true, + sinks: vec![ + AtofSinkSectionConfig::File(AtofFileSinkSectionConfig { + output_directory: Some("logs".into()), + filename: Some("events.jsonl".into()), + mode: "overwrite".into(), + }), + AtofSinkSectionConfig::Stream(AtofStreamSinkSectionConfig { + url: "http://localhost:8080/events".into(), + transport: "http_post".into(), + headers: Default::default(), + header_env: Default::default(), + timeout_millis: 3000, + field_name_policy: "replace_dots".into(), + name: Some("archive".into()), + }), + ], + }), + ..ObservabilityConfig::default() + }); + + let config = PluginConfig { + version: 1, + components: vec![component.into()], + policy: Default::default(), + }; + + let report = validate_plugin_config(&config); + assert!(!report.has_errors()); + + let _active = initialize_plugins(config).await?; -// Run instrumented application work here. + // Run instrumented application work here. -clear_plugin_configuration()?; -Ok(()) + clear_plugin_configuration()?; + Ok(()) } ``` @@ -308,20 +337,19 @@ Ok(()) ## Manual API Use the manual `AtofExporter` API when a test or script needs a custom -subscriber name or explicit registration window. +subscriber name or explicit registration window. A manual exporter owns one +sink. To send the same events to several destinations, create and register one +exporter for each sink or use the observability plugin. ```python -from nemo_relay import AtofEndpointConfig, AtofExporter, AtofExporterConfig, AtofExporterMode +from nemo_relay import AtofExporter, AtofExporterConfig config = AtofExporterConfig() -config.output_directory = "logs" -config.filename = "events.jsonl" -config.mode = AtofExporterMode.Overwrite -config.endpoints = [ - AtofEndpointConfig("http://localhost:8080/events", transport="http_post") -] +config.sink_type = "stream" +config.url = "http://localhost:8080/events" +config.transport = "http_post" exporter = AtofExporter(config) exporter.register("atof-exporter") @@ -340,10 +368,9 @@ exporter.shutdown() const { AtofExporter } = require("nemo-relay-node"); const exporter = new AtofExporter({ - outputDirectory: "logs", - filename: "events.jsonl", - mode: "overwrite", - endpoints: [{ url: "http://localhost:8080/events", transport: "http_post" }], + type: "stream", + url: "http://localhost:8080/events", + transport: "http_post", }); exporter.register("atof-exporter"); @@ -362,18 +389,14 @@ try { ```rust use nemo_relay::observability::atof::{ - AtofEndpointConfig, AtofEndpointTransport, AtofExporter, AtofExporterConfig, AtofExporterMode, + AtofEndpointTransport, AtofExporter, AtofExporterConfig, AtofStreamSinkConfig, }; fn main() -> Result<(), Box> { - let config = AtofExporterConfig::new() - .with_output_directory("logs") - .with_filename("events.jsonl") - .with_mode(AtofExporterMode::Overwrite) - .with_endpoint(AtofEndpointConfig::new( - "http://localhost:8080/events", - AtofEndpointTransport::HttpPost, - )); + let config = AtofExporterConfig::new().with_stream_sink(AtofStreamSinkConfig::new( + "http://localhost:8080/events", + AtofEndpointTransport::HttpPost, + )); let exporter = AtofExporter::new(config)?; exporter.register("atof-exporter")?; @@ -392,12 +415,14 @@ fn main() -> Result<(), Box> { ## Common Configuration and Runtime Issues +Common configuration and runtime issues include: + - ATOF is enabled without at least one plugin-configured sink. -- `mode` is not `append` or `overwrite`. -- A stream sink URL is empty, its transport is not supported, or its timeout is - `0`. +- A file sink's `mode` is not `append` or `overwrite`. +- A stream sink has an empty `url`, an unsupported `transport`, or a + `timeout_millis` value of `0`. - The output directory is not writable at runtime. - `nemo-relay doctor` cannot deliver its synthetic ATOF mark probe to a - configured endpoint. + configured stream sink. - A file sink is configured in a target that cannot access the native filesystem. diff --git a/docs/configure-plugins/observability/configuration.mdx b/docs/configure-plugins/observability/configuration.mdx index 176cec729..27f5e280f 100644 --- a/docs/configure-plugins/observability/configuration.mdx +++ b/docs/configure-plugins/observability/configuration.mdx @@ -35,9 +35,9 @@ only when it includes `enabled = true`. The following table lists the runtime behavior that each section installs: -| Section | Runtime behavior | +| Section | Runtime Behavior | |---|---| -| `atof` | Registers a global Agent Trajectory Observability Format (ATOF) JSONL exporter for raw lifecycle events. | +| `atof` | Registers a global Agent Trajectory Observability Format (ATOF) exporter with file and stream sinks for raw lifecycle events. | | `atif` | Registers one Agent Trajectory Interchange Format (ATIF) dispatcher that writes one trajectory file for each top-level Agent scope or supported coding-agent turn scope. | | `opentelemetry` | Registers a global OpenTelemetry OTLP subscriber. | | `openinference` | Registers a global OpenInference OTLP subscriber. | @@ -82,8 +82,8 @@ url = "http://localhost:8080/events" transport = "http_post" timeout_millis = 3000 -[components.config.atof.sinks.headers] -authorization = "Bearer " +[components.config.atof.sinks.header_env] +authorization = "NEMO_RELAY_ATOF_AUTH_HEADER" [components.config.atif] enabled = true @@ -130,6 +130,13 @@ unknown_field = "warn" unsupported_value = "error" ``` +Observability component configuration requires `version = 2`. To update an +older ATOF configuration, follow [Migrate From Version 1](/configure-plugins/observability/atof#migrate-from-version-1). + +This example reads its ATOF authorization header from +`NEMO_RELAY_ATOF_AUTH_HEADER`. Set that variable to the complete header value +before validation or activation; NeMo Relay rejects missing or blank values. + Include only the sections you want to configure. In layered `plugins.toml` files, omission inherits lower-precedence values; write `enabled = false` to disable an inherited section. @@ -147,7 +154,7 @@ The following table describes how each failure affects application work: | Failure | Behavior | |---|---| -| Invalid `plugins.toml`, duplicate component kinds, malformed component shapes, unsupported values, unavailable exporter features, ATOF file-open failures, invalid ATOF sink config, unavailable ATOF streaming support, or ATOF stream worker startup failures | Validation or initialization fails. If a previous plugin configuration was active, NeMo Relay attempts to restore it after a failed replacement. | +| Invalid `plugins.toml`, duplicate component kinds, malformed component shapes, unsupported values, unavailable exporter features, ATOF file-open failures, invalid ATOF stream sink configuration, unavailable ATOF streaming support, or ATOF stream worker startup failures | Validation or initialization fails. If a previous plugin configuration was active, NeMo Relay attempts to restore it after a failed replacement. | | ATOF event serialization or file write/flush failure after activation | Application work continues. The exporter stores the failure, stops accepting later events for that file, and returns the stored error from `force_flush()` or `shutdown()`. | | ATOF stream sink connection or send failure after activation | File output and other already-started sinks continue. Stream sink failures are logged with the sink index; flush and close timeouts are logged instead of blocking shutdown indefinitely. | | ATIF local file write, HTTP storage, or S3-compatible storage failure | Application work continues. The ATIF exporter records the failed sink as unhealthy and skips it for later trajectories. Other configured sinks continue to receive writes. | @@ -171,7 +178,7 @@ create the output directory and that teardown calls `plugin.clear()` or ## Per-Language Plugin Configuration -The following examples configure and activate the Observability component +The following examples configure and activate the observability component through each supported language binding: `validate()` checks only the supplied in-memory object. `initialize()` also @@ -189,6 +196,7 @@ from nemo_relay.observability import ( AtifConfig, AtofConfig, AtofFileSinkConfig, + AtofStreamSinkConfig, ComponentSpec, ObservabilityConfig, OtlpConfig, @@ -205,7 +213,15 @@ config = plugin.PluginConfig( output_directory="logs", filename="events.jsonl", mode="overwrite", - ) + ), + AtofStreamSinkConfig( + name="archive", + url="http://localhost:8080/events", + transport="http_post", + header_env={ + "authorization": "NEMO_RELAY_ATOF_AUTH_HEADER" + }, + ), ], ), atif=AtifConfig( @@ -257,50 +273,59 @@ const observability = require("nemo-relay-node/observability"); void (async () => { await plugin.initialize({ - version: 1, - components: [ - observability.ComponentSpec({ - version: 2, - atof: observability.atofConfig({ - enabled: true, - sinks: [ - { - type: "file", - output_directory: "logs", - filename: "events.jsonl", - mode: "overwrite", + version: 1, + components: [ + observability.ComponentSpec({ + version: 2, + atof: observability.atofConfig({ + enabled: true, + sinks: [ + { + type: "file", + output_directory: "logs", + filename: "events.jsonl", + mode: "overwrite", + }, + { + type: "stream", + name: "archive", + url: "http://localhost:8080/events", + transport: "http_post", + header_env: { + authorization: "NEMO_RELAY_ATOF_AUTH_HEADER", + }, + }, + ], + }), + atif: observability.atifConfig({ + enabled: true, + output_directory: "logs", + filename_template: "trajectory-{session_id}.json", + }), + opentelemetry: observability.otlpConfig({ + enabled: true, + endpoint: "http://localhost:4318/v1/traces", + service_name: "nemo-relay", + service_namespace: "agent", + service_version: "0.6.0", + instrumentation_scope: "nemo-relay-observability", + resource_attributes: { + "deployment.environment": "dev", }, - ], - }), - atif: observability.atifConfig({ - enabled: true, - output_directory: "logs", - filename_template: "trajectory-{session_id}.json", - }), - opentelemetry: observability.otlpConfig({ - enabled: true, - endpoint: "http://localhost:4318/v1/traces", - service_name: "nemo-relay", - service_namespace: "agent", - service_version: "0.6.0", - instrumentation_scope: "nemo-relay-observability", - resource_attributes: { - "deployment.environment": "dev", - }, - }), - openinference: observability.otlpConfig({ - enabled: true, - endpoint: "http://localhost:6006/v1/traces", - service_name: "nemo-relay", - service_namespace: "agent", - service_version: "0.6.0", - instrumentation_scope: "nemo-relay-openinference", - resource_attributes: { - "deployment.environment": "dev", - }, + }), + openinference: observability.otlpConfig({ + enabled: true, + endpoint: "http://localhost:6006/v1/traces", + service_name: "nemo-relay", + service_namespace: "agent", + service_version: "0.6.0", + instrumentation_scope: "nemo-relay-openinference", + resource_attributes: { + "deployment.environment": "dev", + }, + }), }), - }), - ], + ], }); try { @@ -318,8 +343,6 @@ void (async () => { ```rust -#[tokio::main] -async fn main() -> Result<(), Box> { use nemo_relay::observability::plugin_component::{ AtifSectionConfig, AtofFileSinkSectionConfig, AtofSectionConfig, AtofSinkSectionConfig, AtofStreamSinkSectionConfig, ComponentSpec, @@ -329,70 +352,76 @@ use nemo_relay::plugin::{ clear_plugin_configuration, initialize_plugins, validate_plugin_config, PluginConfig, }; -let component = ComponentSpec::new(ObservabilityConfig { - atof: Some(AtofSectionConfig { - enabled: true, - sinks: vec![ - AtofSinkSectionConfig::File(AtofFileSinkSectionConfig { - output_directory: Some("logs".into()), - filename: Some("events.jsonl".into()), - mode: "overwrite".into(), - }), - AtofSinkSectionConfig::Stream(AtofStreamSinkSectionConfig { - url: "http://localhost:8080/events".into(), - transport: "http_post".into(), - headers: [("authorization".into(), "Bearer ".into())].into(), - header_env: Default::default(), - timeout_millis: 3000, - field_name_policy: "preserve".into(), - name: Some("archive".into()), - }), - ], - }), - atif: Some(AtifSectionConfig { - enabled: true, - output_directory: Some("logs".into()), - filename_template: "trajectory-{session_id}.json".into(), - ..AtifSectionConfig::default() - }), - opentelemetry: Some(OtlpSectionConfig { - enabled: true, - endpoint: Some("http://localhost:4318/v1/traces".into()), - service_name: "nemo-relay".into(), - service_namespace: Some("agent".into()), - service_version: Some("0.6.0".into()), - instrumentation_scope: Some("nemo-relay-observability".into()), - resource_attributes: [("deployment.environment".into(), "dev".into())].into(), - ..OtlpSectionConfig::default() - }), - openinference: Some(OtlpSectionConfig { - enabled: true, - endpoint: Some("http://localhost:6006/v1/traces".into()), - service_name: "nemo-relay".into(), - service_namespace: Some("agent".into()), - service_version: Some("0.6.0".into()), - instrumentation_scope: Some("nemo-relay-openinference".into()), - resource_attributes: [("deployment.environment".into(), "dev".into())].into(), - ..OtlpSectionConfig::default() - }), - ..ObservabilityConfig::default() -}); - -let config = PluginConfig { - version: 1, - components: vec![component.into()], - policy: Default::default(), -}; - -let report = validate_plugin_config(&config); -assert!(!report.has_errors()); - -let _active = initialize_plugins(config).await?; +#[tokio::main] +async fn main() -> Result<(), Box> { + let component = ComponentSpec::new(ObservabilityConfig { + atof: Some(AtofSectionConfig { + enabled: true, + sinks: vec![ + AtofSinkSectionConfig::File(AtofFileSinkSectionConfig { + output_directory: Some("logs".into()), + filename: Some("events.jsonl".into()), + mode: "overwrite".into(), + }), + AtofSinkSectionConfig::Stream(AtofStreamSinkSectionConfig { + url: "http://localhost:8080/events".into(), + transport: "http_post".into(), + headers: Default::default(), + header_env: [( + "authorization".into(), + "NEMO_RELAY_ATOF_AUTH_HEADER".into(), + )] + .into(), + timeout_millis: 3000, + field_name_policy: "preserve".into(), + name: Some("archive".into()), + }), + ], + }), + atif: Some(AtifSectionConfig { + enabled: true, + output_directory: Some("logs".into()), + filename_template: "trajectory-{session_id}.json".into(), + ..AtifSectionConfig::default() + }), + opentelemetry: Some(OtlpSectionConfig { + enabled: true, + endpoint: Some("http://localhost:4318/v1/traces".into()), + service_name: "nemo-relay".into(), + service_namespace: Some("agent".into()), + service_version: Some("0.6.0".into()), + instrumentation_scope: Some("nemo-relay-observability".into()), + resource_attributes: [("deployment.environment".into(), "dev".into())].into(), + ..OtlpSectionConfig::default() + }), + openinference: Some(OtlpSectionConfig { + enabled: true, + endpoint: Some("http://localhost:6006/v1/traces".into()), + service_name: "nemo-relay".into(), + service_namespace: Some("agent".into()), + service_version: Some("0.6.0".into()), + instrumentation_scope: Some("nemo-relay-openinference".into()), + resource_attributes: [("deployment.environment".into(), "dev".into())].into(), + ..OtlpSectionConfig::default() + }), + ..ObservabilityConfig::default() + }); + + let config = PluginConfig { + version: 1, + components: vec![component.into()], + policy: Default::default(), + }; + + let report = validate_plugin_config(&config); + assert!(!report.has_errors()); + + let _active = initialize_plugins(config).await?; -// Run instrumented application work here. + // Run instrumented application work here. -clear_plugin_configuration()?; -Ok(()) + clear_plugin_configuration()?; + Ok(()) } ``` @@ -409,7 +438,7 @@ filename templates, unknown fields according to policy, and enabled exporters that are unavailable in the current build or target. Call `plugin.clear()` or `clear_plugin_configuration()` during teardown. -Clearing the plugin config deregisters inferred subscribers, flushes file +Clearing the plugin configuration deregisters inferred subscribers, flushes file exporters, drains and closes ATOF stream sinks, and shuts down owned OTLP subscribers. diff --git a/docs/configure-plugins/plugin-configuration-files.mdx b/docs/configure-plugins/plugin-configuration-files.mdx index 2d5d7e81c..226ffb77c 100644 --- a/docs/configure-plugins/plugin-configuration-files.mdx +++ b/docs/configure-plugins/plugin-configuration-files.mdx @@ -39,10 +39,13 @@ kind = "observability" enabled = true [components.config] -version = 1 +version = 2 [components.config.atof] enabled = true + +[[components.config.atof.sinks]] +type = "file" output_directory = "logs" filename = "events.jsonl" mode = "append" @@ -97,10 +100,13 @@ kind = "observability" enabled = true [components.config] -version = 1 +version = 2 [components.config.atof] enabled = true + +[[components.config.atof.sinks]] +type = "file" output_directory = "logs" filename = "events.jsonl" mode = "append" @@ -254,20 +260,29 @@ When more than one `plugins.toml` file is discovered, later files have higher precedence. User config overrides project config, and project config overrides system config. -TOML tables merge recursively: +TOML tables merge recursively, while a higher-precedence array replaces the +lower-precedence array. For example, the following system configuration enables +one ATOF file sink: ```toml # system plugins.toml [[components]] kind = "observability" +[components.config] +version = 2 + [components.config.atof] enabled = true + +[[components.config.atof.sinks]] +type = "file" output_directory = "/var/log/nemo-relay" mode = "append" ``` -The following user configuration overrides only the exporter mode: +To change the file mode at user scope, repeat the complete sink because +`sinks` is an array: ```toml # user plugins.toml @@ -275,12 +290,16 @@ The following user configuration overrides only the exporter mode: kind = "observability" [components.config.atof] + +[[components.config.atof.sinks]] +type = "file" +output_directory = "/var/log/nemo-relay" mode = "overwrite" ``` -The effective Agent Trajectory Observability Format (ATOF) config keeps -`enabled` and `output_directory` from the system file and uses -`mode = "overwrite"` from the user file. +The effective Agent Trajectory Observability Format (ATOF) configuration keeps +`version` and `enabled` from the system file. The user file replaces the +complete `sinks` array with the file sink that uses `mode = "overwrite"`. The top-level `components` array is special. Relay matches components by `kind` across files. A higher-precedence component with the same `kind` merges into the @@ -344,16 +363,15 @@ kind = "observability" [components.config.atof] enabled = false -mode = "append" ``` -The merged config can still contain inherited ATOF sibling fields, such as -`output_directory`, but the runtime ignores the section because `enabled = -false`. +The merged configuration can still contain an inherited ATOF `sinks` array, but +the runtime ignores the section because `enabled = false`. To override an inherited non-default field with its default value, write the -default explicitly in the higher-precedence file. For example, use -`mode = "append"` to override a lower-precedence `mode = "overwrite"`. +default explicitly in the higher-precedence file. For a field inside an array, +repeat the complete array entry. For example, repeat an inherited file sink and +set its `mode` to `append`. There is no tombstone syntax for deleting an inherited nested field while keeping the rest of the lower-precedence component. To remove inherited settings @@ -382,7 +400,7 @@ Common validation failures include: Use `nemo-relay doctor` to inspect the resolved gateway configuration and plugin diagnostics. For Observability, doctor also reports enabled exporter sections, checks writable file exporter directories, probes configured ATOF -streaming endpoints, and checks reachable OTLP endpoints when those settings +stream sinks, and checks reachable OTLP endpoints when those settings are present. For model pricing, doctor validates enabled file and inline sources and fails when a source is unreadable or the catalog schema is invalid. For dynamic-plugin validation, lifecycle state, and trust diagnostics, use diff --git a/scripts/test-claude-plugin-e2e.sh b/scripts/test-claude-plugin-e2e.sh index a3a28f156..b26d2559a 100755 --- a/scripts/test-claude-plugin-e2e.sh +++ b/scripts/test-claude-plugin-e2e.sh @@ -95,10 +95,13 @@ kind = "observability" enabled = true [components.config] -version = 1 +version = 2 [components.config.atof] enabled = true + +[[components.config.atof.sinks]] +type = "file" output_directory = "$work/atof" filename = "events.jsonl" mode = "append" diff --git a/scripts/test-codex-plugin-e2e.sh b/scripts/test-codex-plugin-e2e.sh index 9a4e28fb5..26a20ff63 100755 --- a/scripts/test-codex-plugin-e2e.sh +++ b/scripts/test-codex-plugin-e2e.sh @@ -143,10 +143,13 @@ kind = "observability" enabled = true [components.config] -version = 1 +version = 2 [components.config.atof] enabled = true + +[[components.config.atof.sinks]] +type = "file" output_directory = "atof" filename = "events.jsonl" mode = "append" diff --git a/scripts/test-hermes-mcp-e2e.sh b/scripts/test-hermes-mcp-e2e.sh index cd542d6b2..950b1cfe2 100755 --- a/scripts/test-hermes-mcp-e2e.sh +++ b/scripts/test-hermes-mcp-e2e.sh @@ -122,10 +122,13 @@ kind = "observability" enabled = true [components.config] -version = 1 +version = 2 [components.config.atof] enabled = true + +[[components.config.atof.sinks]] +type = "file" output_directory = "$work/atof" filename = "events.jsonl" mode = "append" diff --git a/skills/nemo-relay-get-started/references/cli-try-now.md b/skills/nemo-relay-get-started/references/cli-try-now.md index e8823b850..cce18b298 100644 --- a/skills/nemo-relay-get-started/references/cli-try-now.md +++ b/skills/nemo-relay-get-started/references/cli-try-now.md @@ -112,10 +112,13 @@ kind = "observability" enabled = true [components.config] -version = 1 +version = 2 [components.config.atof] enabled = true + +[[components.config.atof.sinks]] +type = "file" output_directory = ".nemo-relay/atof" filename = "events.jsonl" mode = "append" diff --git a/skills/nemo-relay-plugin-observability/references/atof.md b/skills/nemo-relay-plugin-observability/references/atof.md index cc8c7e846..de50b0817 100644 --- a/skills/nemo-relay-plugin-observability/references/atof.md +++ b/skills/nemo-relay-plugin-observability/references/atof.md @@ -22,21 +22,36 @@ kind = "observability" enabled = true [components.config] -version = 1 +version = 2 [components.config.atof] enabled = true + +[[components.config.atof.sinks]] +type = "file" output_directory = "logs" filename = "events.jsonl" mode = "append" + +[[components.config.atof.sinks]] +type = "stream" +url = "http://localhost:8080/events" +transport = "http_post" +header_env = { authorization = "NEMO_RELAY_ATOF_AUTH_HEADER" } ``` Use `overwrite` for an isolated one-run artifact and `append` for repeated local -runs. File output remains active when optional streaming endpoints are added. +runs. Add a `stream` sink when the same events should also be delivered +remotely. Use `header_env` to map stream header names to environment variables, +keeping credentials out of configuration files. Before activation, set each +named variable to the complete header value; validation rejects missing or +blank values. Use the manual `AtofExporter` API only when the caller needs a custom subscriber -name or explicit registration window. The lifecycle is: create, register, run -instrumented work, force flush, deregister, then shut down. +name or explicit registration window. Each manual exporter owns one sink, so +register one exporter per destination when you need fan-out. The lifecycle is: +create, register, run instrumented work, force flush, deregister, then shut +down. ## Verify @@ -47,11 +62,11 @@ Verify the export with the following checks: - Check UUID and parent UUID relationships instead of relying only on event order. - Confirm sensitive fields are absent before retaining or transmitting output. -- For streaming endpoints, verify file output separately from endpoint delivery. +- For stream sinks, verify file output separately from remote delivery. Common failures include an unwritable output directory, an invalid mode, an -empty endpoint URL, an unsupported endpoint transport, or shutdown occurring -before pending events flush. +empty stream URL, an unsupported stream transport, abrupt process termination, +or interruption before `shutdown()` finishes flushing pending events. -For the complete exporter configuration, see +For the complete exporter configuration, refer to [ATOF observability](https://docs.nvidia.com/nemo/relay/dev/configure-plugins/observability/atof).