feat: use Relay v2 ATOF sinks

adapters/deepagents/README.md

@@ -181,6 +181,7 @@ CLI flags are involved:
 from nemo_fabric import (
     RelayAtifConfig,
     RelayAtofConfig,
+    RelayAtofFileSinkConfig,
     RelayObservabilityConfig,
 )
 from examples.code_review_agent import deepagents_config
@@ -192,9 +193,13 @@ config.enable_relay(
     observability=RelayObservabilityConfig(
         atof=RelayAtofConfig(
             enabled=True,
-            output_directory="./artifacts/relay",
-            filename="events.atof.jsonl",
-            mode="overwrite",
+            sinks=[
+                RelayAtofFileSinkConfig(
+                    output_directory="./artifacts/relay",
+                    filename="events.atof.jsonl",
+                    mode="overwrite",
+                )
+            ],
         ),
         atif=RelayAtifConfig(
             enabled=True,

adapters/deepagents/src/nemo_fabric_adapters/deepagents/adapter.py

@@ -480,7 +480,7 @@ def __init__(self) -> None:
         self._relay_plugin: Any = None
         self._relay_scope: Any = None
         self._relay_scope_type: Any = None
-        self._relay_api_config: Any = None
+        self._relay_plugin_config: dict[str, Any] | None = None
         self._callback_handler_type: Any = None
 
     async def start(self, payload: dict[str, Any]) -> None:
@@ -549,9 +549,7 @@ def _configure_observability(self, agent_kwargs: dict[str, Any]) -> dict[str, An
             raise _relay_dependency_error() from exc
 
         assert self._observability is not None
-        self._relay_api_config = common_utils.relay_api_plugin_config(
-            self._observability.plugin_config
-        )
+        self._relay_plugin_config = self._observability.plugin_config
         self._relay_plugin = plugin
         self._relay_scope = scope
         self._relay_scope_type = ScopeType
@@ -590,7 +588,7 @@ async def invoke(self, invocation: dict[str, Any]) -> dict[str, Any]:
         try:
             if self._observability is not None:
                 callback_handler = self._callback_handler_type()
-                async with self._relay_plugin.plugin(self._relay_api_config):
+                async with self._relay_plugin.plugin(self._relay_plugin_config):
                     with self._relay_scope.scope(
                         "deepagents-request", self._relay_scope_type.Agent
                     ):
@@ -663,7 +661,7 @@ async def stop(self) -> None:
         self._relay_plugin = None
         self._relay_scope = None
         self._relay_scope_type = None
-        self._relay_api_config = None
+        self._relay_plugin_config = None
         self._callback_handler_type = None
         self._started = False
         if checkpointer is not None:

adapters/hermes/src/nemo_fabric_adapters/hermes/adapter.py

@@ -238,12 +238,9 @@ async def start(self, payload: dict[str, Any]) -> None:
                 self._relay_plugin_config = common_utils.load_relay_plugin_config(
                     payload
                 )
-                relay_api_config = common_utils.relay_api_plugin_config(
-                    self._relay_plugin_config
-                )
                 from nemo_relay import plugin
 
-                self._relay_context = plugin.plugin(relay_api_config)
+                self._relay_context = plugin.plugin(self._relay_plugin_config)
                 await self._relay_context.__aenter__()
                 self._relay_context_entered = True
 

crates/fabric-core/src/config.rs

@@ -625,43 +625,59 @@ pub struct RelayAtofConfig {
     /// Whether ATOF export is enabled.
     #[serde(default)]
     pub enabled: bool,
-    /// Directory used for ATOF files.
-    #[serde(default, skip_serializing_if = "Option::is_none")]
-    pub output_directory: Option<PathBuf>,
-    /// ATOF file name.
-    #[serde(default, skip_serializing_if = "Option::is_none")]
-    pub filename: Option<String>,
-    /// File write mode.
-    #[serde(default)]
-    pub mode: RelayAtofMode,
-    /// Optional remote ATOF endpoints.
+    /// ATOF file and stream sinks.
     #[serde(default, skip_serializing_if = "Vec::is_empty")]
-    pub endpoints: Vec<RelayAtofEndpointConfig>,
+    pub sinks: Vec<RelayAtofSinkConfig>,
     /// Additive ATOF fields.
     #[serde(default, flatten)]
     pub extensions: BTreeMap<String, Value>,
 }
 
-/// Relay ATOF endpoint configuration.
+/// Relay ATOF sink configuration.
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
-pub struct RelayAtofEndpointConfig {
-    /// Endpoint URL.
-    pub url: String,
-    /// Endpoint transport.
-    #[serde(default)]
-    pub transport: RelayAtofEndpointTransport,
-    /// Endpoint headers.
-    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
-    pub headers: BTreeMap<String, String>,
-    /// Request timeout in milliseconds.
-    #[serde(default = "default_relay_timeout_millis")]
-    pub timeout_millis: u64,
-    /// Field-name handling policy.
-    #[serde(default)]
-    pub field_name_policy: RelayAtofEndpointFieldNamePolicy,
-    /// Additive endpoint fields.
-    #[serde(default, flatten)]
-    pub extensions: BTreeMap<String, Value>,
+#[serde(tag = "type", rename_all = "snake_case")]
+pub enum RelayAtofSinkConfig {
+    /// Write ATOF records to a local file.
+    File {
+        /// Directory used for ATOF files.
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        output_directory: Option<PathBuf>,
+        /// ATOF file name.
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        filename: Option<String>,
+        /// File write mode.
+        #[serde(default)]
+        mode: RelayAtofMode,
+        /// Additive file sink fields.
+        #[serde(default, flatten)]
+        extensions: BTreeMap<String, Value>,
+    },
+    /// Send ATOF records to a remote stream.
+    Stream {
+        /// Stream URL.
+        url: String,
+        /// Stream transport.
+        #[serde(default)]
+        transport: RelayAtofStreamTransport,
+        /// Static stream headers.
+        #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
+        headers: BTreeMap<String, String>,
+        /// Environment-variable-backed stream headers.
+        #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
+        header_env: BTreeMap<String, String>,
+        /// Request timeout in milliseconds.
+        #[serde(default = "default_relay_timeout_millis")]
+        timeout_millis: u64,
+        /// Field-name handling policy.
+        #[serde(default)]
+        field_name_policy: RelayAtofStreamFieldNamePolicy,
+        /// Optional stream sink name.
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        name: Option<String>,
+        /// Additive stream sink fields.
+        #[serde(default, flatten)]
+        extensions: BTreeMap<String, Value>,
+    },
 }
 
 /// Relay ATIF export configuration.
@@ -874,10 +890,10 @@ pub enum RelayAtofMode {
     Overwrite,
 }
 
-/// Relay ATOF endpoint transport.
+/// Relay ATOF stream transport.
 #[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
 #[serde(rename_all = "snake_case")]
-pub enum RelayAtofEndpointTransport {
+pub enum RelayAtofStreamTransport {
     /// HTTP POST transport.
     #[default]
     HttpPost,
@@ -887,10 +903,10 @@ pub enum RelayAtofEndpointTransport {
     Ndjson,
 }
 
-/// Relay ATOF endpoint field-name policy.
+/// Relay ATOF stream field-name policy.
 #[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
 #[serde(rename_all = "snake_case")]
-pub enum RelayAtofEndpointFieldNamePolicy {
+pub enum RelayAtofStreamFieldNamePolicy {
     /// Preserve field names.
     #[default]
     Preserve,
@@ -933,7 +949,7 @@ impl TelemetryProvider {
 }
 
 fn default_relay_config_version() -> u32 {
-    1
+    2
 }
 
 fn default_enabled() -> bool {
@@ -1643,6 +1659,36 @@ mod tests {
         PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../..")
     }
 
+    #[test]
+    fn relay_observability_uses_v2_typed_atof_sinks() {
+        let observability: RelayObservabilityConfig = serde_json::from_value(serde_json::json!({
+            "atof": {
+                "enabled": true,
+                "sinks": [
+                    {
+                        "type": "file",
+                        "output_directory": "artifacts/relay",
+                        "filename": "events.atof.jsonl",
+                        "mode": "overwrite"
+                    },
+                    {
+                        "type": "stream",
+                        "url": "http://localhost:4319/events",
+                        "transport": "ndjson",
+                        "header_env": {"authorization": "RELAY_AUTHORIZATION"},
+                        "name": "live-events"
+                    }
+                ]
+            }
+        }))
+        .expect("Relay v2 observability config");
+
+        let value = serde_json::to_value(observability).expect("serialized observability");
+        assert_eq!(value["version"], 2);
+        assert_eq!(value["atof"]["sinks"][0]["type"], "file");
+        assert_eq!(value["atof"]["sinks"][1]["type"], "stream");
+    }
+
     #[test]
     fn resolves_complete_typed_config_with_explicit_base_dir() {
         let base_dir = repository_root();

docs/reference/api/python-library-reference/index.md

@@ -31,7 +31,8 @@ SPDX-License-Identifier: Apache-2.0 */}
 - [`models.ModelConfig`](./nemo_fabric.models.md#class-modelconfig): Model alias configuration.
 - [`models.RelayAtifConfig`](./nemo_fabric.models.md#class-relayatifconfig): NeMo Relay ATIF export configuration.
 - [`models.RelayAtofConfig`](./nemo_fabric.models.md#class-relayatofconfig): NeMo Relay ATOF export configuration.
-- [`models.RelayAtofEndpointConfig`](./nemo_fabric.models.md#class-relayatofendpointconfig): NeMo Relay ATOF remote endpoint configuration.
+- [`models.RelayAtofFileSinkConfig`](./nemo_fabric.models.md#class-relayatoffilesinkconfig): NeMo Relay ATOF file sink configuration.
+- [`models.RelayAtofStreamSinkConfig`](./nemo_fabric.models.md#class-relayatofstreamsinkconfig): NeMo Relay ATOF stream sink configuration.
 - [`models.RelayComponentConfig`](./nemo_fabric.models.md#class-relaycomponentconfig): Generic NeMo Relay plugin component configuration.
 - [`models.RelayConfig`](./nemo_fabric.models.md#class-relayconfig): First-class NeMo Relay integration configuration.
 - [`models.RelayConfigPolicy`](./nemo_fabric.models.md#class-relayconfigpolicy): NeMo Relay config validation policy.

docs/reference/api/python-library-reference/nemo_fabric.models.md

@@ -668,8 +668,68 @@ Return a detached JSON-compatible mapping for Rust/core calls.
 ---
 
 
-## <kbd>class</kbd> `RelayAtofEndpointConfig`
-NeMo Relay ATOF remote endpoint configuration.
+## <kbd>class</kbd> `RelayAtofFileSinkConfig`
+NeMo Relay ATOF file sink configuration.
+
+
+---
+
+### <kbd>property</kbd> extra_fields
+
+Return fields preserved by the extension point for this model.
+
+---
+
+### <kbd>property</kbd> model_extra
+
+Get extra fields set during validation.
+
+
+
+**Returns:**
+  A dictionary of extra fields, or `None` if `config.extra` is not set to `"allow"`.
+
+---
+
+### <kbd>property</kbd> model_fields_set
+
+Returns the set of fields that have been explicitly set on this model instance.
+
+
+
+**Returns:**
+  A set of strings representing the fields that have been set,  i.e. that were not filled from defaults.
+
+
+
+---
+
+
+### <kbd>classmethod</kbd> `from_mapping`
+
+```python
+from_mapping(value: 'Mapping[str, Any]') → Self
+```
+
+Validate a mapping using this Pydantic model.
+
+---
+
+
+### <kbd>method</kbd> `to_mapping`
+
+```python
+to_mapping() → dict[str, Any]
+```
+
+Return a detached JSON-compatible mapping for Rust/core calls.
+
+
+---
+
+
+## <kbd>class</kbd> `RelayAtofStreamSinkConfig`
+NeMo Relay ATOF stream sink configuration.
 
 
 ---

docs/reference/api/rust-library-reference/nemo-fabric-core/config/enum-capabilitykind.mdx

@@ -2,7 +2,7 @@
 title: "Enum Capability Kind"
 sidebar-title: "CapabilityKind"
 description: "Capability kind."
-position: 39
+position: 38
 ---
 {/* SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 SPDX-License-Identifier: Apache-2.0 */}

docs/reference/api/rust-library-reference/nemo-fabric-core/config/enum-capabilitytarget.mdx

@@ -2,7 +2,7 @@
 title: "Enum Capability Target"
 sidebar-title: "CapabilityTarget"
 description: "Capability routing target."
-position: 40
+position: 39
 ---
 {/* SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 SPDX-License-Identifier: Apache-2.0 */}

docs/reference/api/rust-library-reference/nemo-fabric-core/config/enum-relayatifstorageconfig.mdx

@@ -2,7 +2,7 @@
 title: "Enum Relay Atif Storage Config"
 sidebar-title: "RelayAtifStorageConfig"
 description: "Relay ATIF remote storage configuration."
-position: 44
+position: 43
 ---
 {/* SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 SPDX-License-Identifier: Apache-2.0 */}

docs/reference/api/rust-library-reference/nemo-fabric-core/config/enum-relayatofmode.mdx

@@ -2,7 +2,7 @@
 title: "Enum Relay Atof Mode"
 sidebar-title: "RelayAtofMode"
 description: "Relay ATOF file mode."
-position: 47
+position: 44
 ---
 {/* SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 SPDX-License-Identifier: Apache-2.0 */}