feat(plugin): validate named Switchyard ATOF endpoints

crates/core/src/observability/plugin_component.rs

@@ -215,6 +215,9 @@ pub struct AtofStreamSinkSectionConfig {
     /// Field name policy applied before sending events: `preserve` or `replace_dots`.
     #[serde(default = "default_atof_endpoint_field_name_policy")]
     pub field_name_policy: String,
+    /// Optional stable name used by other components to reference this endpoint.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub name: Option<String>,
 }
 
 /// Per-trajectory ATIF exporter config.
@@ -503,6 +506,7 @@ crate::editor_config! {
         header_env => { label: "header_env", kind: StringMap },
         timeout_millis => { label: "timeout_millis", kind: Integer },
         field_name_policy => { label: "field_name_policy", kind: Enum, values: ["preserve", "replace_dots"] },
+        name => { label: "name", kind: String, optional: true },
     }
 }
 
@@ -1902,6 +1906,7 @@ fn validate_atof_values(
             "ATOF requires at least one configured sink when enabled".to_string(),
         );
     }
+    let mut stream_sink_names = HashSet::new();
     for (index, sink) in section.sinks.iter().enumerate() {
         match sink {
             AtofSinkSectionConfig::File(file) => {
@@ -1917,6 +1922,30 @@ fn validate_atof_values(
                 }
             }
             AtofSinkSectionConfig::Stream(stream) => {
+                if let Some(name) = stream.name.as_deref() {
+                    let trimmed = name.trim();
+                    let message = if trimmed.is_empty() {
+                        Some(format!("ATOF sinks[{index}].name must be non-empty"))
+                    } else if name != trimmed {
+                        Some(format!(
+                            "ATOF sinks[{index}].name must not have leading or trailing whitespace"
+                        ))
+                    } else if !stream_sink_names.insert(name) {
+                        Some(format!("ATOF stream sink name {name:?} must be unique"))
+                    } else {
+                        None
+                    };
+                    if let Some(message) = message {
+                        push_policy_diag(
+                            diagnostics,
+                            policy.unsupported_value,
+                            "observability.unsupported_value",
+                            Some("atof".to_string()),
+                            Some(format!("sinks[{index}].name")),
+                            message,
+                        );
+                    }
+                }
                 validate_atof_stream_sink_values(diagnostics, policy, index, stream);
             }
         }
@@ -2450,11 +2479,11 @@ fn default_atof_mode() -> String {
 }
 
 fn default_atof_endpoint_transport() -> String {
-    "http_post".to_string()
+    AtofEndpointTransport::default().as_str().to_string()
 }
 
 fn default_atof_endpoint_field_name_policy() -> String {
-    "preserve".to_string()
+    AtofEndpointFieldNamePolicy::default().as_str().to_string()
 }
 
 fn default_agent_name() -> String {

crates/core/tests/unit/observability/plugin_component_tests.rs

@@ -266,12 +266,13 @@ fn default_config_and_component_conversion_cover_public_shape() {
     assert!(atof.sinks.is_empty());
 
     let parsed_atof: AtofSectionConfig = serde_json::from_value(json!({
-        "sinks": [{"type": "stream", "url": "http://localhost/events"}]
+        "sinks": [{"type": "stream", "name": "switchyard", "url": "http://localhost/events"}]
     }))
     .unwrap();
     let AtofSinkSectionConfig::Stream(stream) = &parsed_atof.sinks[0] else {
         panic!("expected stream sink");
     };
+    assert_eq!(stream.name.as_deref(), Some("switchyard"));
     assert_eq!(stream.transport, "http_post");
     assert_eq!(stream.field_name_policy, "preserve");
 
@@ -368,6 +369,7 @@ fn schema_contains_every_supported_observability_option() {
         "output_directory",
         "filename",
         "mode",
+        "name",
         "sinks",
         "type",
         "url",
@@ -639,7 +641,10 @@ fn atof_endpoint_validation_rejects_bad_values() {
                 {"type": "stream", "url": "not a url", "transport": "http_post"},
                 {"type": "stream", "url": "http://localhost/events", "transport": "http_post", "field_name_policy": "bogus"},
                 {"type": "stream", "url": "http://localhost/events", "transport": "websocket"},
-                {"type": "stream", "url": "http://localhost/events", "headers": {"invalid header": "value", "x-api-key": "value"}, "header_env": {"X-Api-Key": "NEMO_RELAY_TEST_MISSING_ATOF_HEADER_ENV"}}
+                {"type": "stream", "url": "http://localhost/events", "headers": {"invalid header": "value", "x-api-key": "value"}, "header_env": {"X-Api-Key": "NEMO_RELAY_TEST_MISSING_ATOF_HEADER_ENV"}},
+                {"type": "stream", "name": "switchyard", "url": "http://localhost/first"},
+                {"type": "stream", "name": "switchyard", "url": "http://localhost/second"},
+                {"type": "stream", "name": " ", "url": "http://localhost/blank"}
             ]
         }
     })));
@@ -695,6 +700,18 @@ fn atof_endpoint_validation_rejects_bad_values() {
             .iter()
             .any(|diag| { diag.field.as_deref() == Some("sinks[6].headers.invalid header") })
     );
+    assert!(
+        report
+            .diagnostics
+            .iter()
+            .any(|diag| { diag.field.as_deref() == Some("sinks[8].name") })
+    );
+    assert!(
+        report
+            .diagnostics
+            .iter()
+            .any(|diag| { diag.field.as_deref() == Some("sinks[9].name") })
+    );
 }
 
 #[test]
@@ -704,6 +721,7 @@ fn build_atof_sink_config_maps_headers_timeout_and_rejects_transport() {
     let config = build_atof_sink_config(
         2,
         AtofSinkSectionConfig::Stream(AtofStreamSinkSectionConfig {
+            name: Some("switchyard".into()),
             url: "ws://127.0.0.1:47632/events".into(),
             transport: "websocket".into(),
             headers: headers.clone(),
@@ -739,6 +757,7 @@ fn build_atof_sink_config_maps_headers_timeout_and_rejects_transport() {
     let error = build_atof_sink_config(
         3,
         AtofSinkSectionConfig::Stream(AtofStreamSinkSectionConfig {
+            name: None,
             url: "http://127.0.0.1:47632/events".into(),
             transport: "smtp".into(),
             headers: std::collections::HashMap::new(),
@@ -753,6 +772,7 @@ fn build_atof_sink_config_maps_headers_timeout_and_rejects_transport() {
     let error = build_atof_sink_config(
         4,
         AtofSinkSectionConfig::Stream(AtofStreamSinkSectionConfig {
+            name: None,
             url: "http://127.0.0.1:47632/events".into(),
             transport: "http_post".into(),
             headers: std::collections::HashMap::new(),

crates/node/observability.d.ts

@@ -28,6 +28,7 @@ export interface AtofStreamSinkConfig {
   header_env?: Record<string, string>;
   timeout_millis?: number;
   field_name_policy?: 'preserve' | 'replace_dots' | string;
+  name?: string;
 }
 
 /** @deprecated Use AtofStreamSinkConfig. */

crates/node/tests/observability_plugin_tests.mjs

@@ -56,14 +56,18 @@ describe('observability plugin helpers', () => {
         }),
       ],
     });
-    assert.deepEqual(report.diagnostics.map((diagnostic) => diagnostic.field).sort(), ['filename_template', 'sinks[0].mode']);
+    assert.deepEqual(report.diagnostics.map((diagnostic) => diagnostic.field).sort(), [
+      'filename_template',
+      'sinks[0].mode',
+    ]);
   });
 
   it('serializes ATOF stream sinks', () => {
     const config = observability.atofConfig({
       sinks: [
         {
           type: 'stream',
+          name: 'switchyard',
           url: 'http://localhost:8080/events',
           transport: 'http_post',
           headers: { 'X-Test': 'yes' },
@@ -76,6 +80,7 @@ describe('observability plugin helpers', () => {
     assert.deepEqual(config.sinks, [
       {
         type: 'stream',
+        name: 'switchyard',
         url: 'http://localhost:8080/events',
         transport: 'http_post',
         headers: { 'X-Test': 'yes' },

crates/switchyard/src/component.rs

@@ -21,6 +21,7 @@ use nemo_relay::codec::optimization::{
     LlmOptimizationModelTransition,
 };
 use nemo_relay::error::{FlowError, Result as FlowResult};
+use nemo_relay::observability::atof::{AtofEndpointFieldNamePolicy, AtofEndpointTransport};
 use nemo_relay::plugin::{
     ConfigDiagnostic, DiagnosticLevel, Plugin, PluginComponentSpec, PluginConfig, PluginError,
     PluginRegistrationContext, Result as PluginResult, deregister_plugin, register_plugin,
@@ -217,9 +218,9 @@ pub struct SwitchyardConfig {
     pub targets: BTreeMap<String, TargetBinding>,
     /// Trusted per-protocol fallbacks.
     pub default_targets: ProtocolDefaults,
-    /// Optional explicit ATOF endpoint used by CLI cross-validation.
+    /// Named observability ATOF endpoint used by history-backed profiles.
     #[serde(default)]
-    pub atof_endpoint_url: Option<String>,
+    pub atof_endpoint_name: Option<String>,
 }
 
 impl Default for SwitchyardConfig {
@@ -244,7 +245,7 @@ impl Default for SwitchyardConfig {
                 openai_responses: String::new(),
                 anthropic_messages: String::new(),
             },
-            atof_endpoint_url: None,
+            atof_endpoint_name: None,
         }
     }
 }
@@ -269,7 +270,7 @@ nemo_relay::editor_config! {
         enabled_inbound_profiles => { label: "Enabled inbound profiles", kind: Json },
         targets => { label: "Backend target bindings", kind: Json },
         default_targets => { label: "Trusted protocol defaults", kind: Json },
-        atof_endpoint_url => { label: "ATOF endpoint URL", kind: String, optional: true }
+        atof_endpoint_name => { label: "ATOF endpoint name", kind: String, optional: true }
     }
 }
 
@@ -401,67 +402,74 @@ pub fn validate_switchyard_atof_configuration(config: &PluginConfig) -> Result<(
     if switchyard.context_mode != ContextMode::AtofRequired {
         return Ok(());
     }
-    let required_url = match &switchyard.atof_endpoint_url {
-        Some(url) => url.clone(),
-        None => derived_atof_url(&switchyard.decision_api_url)?,
-    };
+    let required_name = validate_atof_endpoint_name(switchyard.atof_endpoint_name.as_deref())?
+        .ok_or_else(|| {
+            "atof_required Switchyard profiles require atof_endpoint_name".to_string()
+        })?;
     let observability = config
         .components
         .iter()
         .find(|component| component.enabled && component.kind == "observability")
         .ok_or_else(|| "atof_required Switchyard profiles require observability".to_string())?;
-    let endpoints = observability
+    let sinks = observability
         .config
         .get("atof")
         .filter(|atof| atof.get("enabled").and_then(Json::as_bool) == Some(true))
-        .and_then(|atof| atof.get("endpoints"))
+        .and_then(|atof| atof.get("sinks"))
         .and_then(Json::as_array)
         .ok_or_else(|| {
             "atof_required Switchyard profiles require an enabled ATOF endpoint".to_string()
         })?;
-    let endpoint = endpoints
+    let matching_sinks = sinks
         .iter()
-        .find(|endpoint| {
-            endpoint.get("url").and_then(Json::as_str) == Some(required_url.as_str())
-                && endpoint
-                    .get("transport")
-                    .and_then(Json::as_str)
-                    .unwrap_or("http_post")
-                    == "http_post"
+        .filter(|sink| {
+            sink.get("type").and_then(Json::as_str) == Some("stream")
+                && sink.get("name").and_then(Json::as_str) == Some(required_name)
         })
-        .ok_or_else(|| {
-            format!("atof_required Switchyard profile requires HTTP ATOF endpoint {required_url}")
-        })?;
-    if endpoint
-        .get("field_name_policy")
-        .and_then(Json::as_str)
-        .unwrap_or("preserve")
-        != "preserve"
-    {
-        return Err("Switchyard ATOF endpoint must use field_name_policy = preserve".into());
+        .collect::<Vec<_>>();
+    let endpoint = match matching_sinks.as_slice() {
+        [sink] => *sink,
+        [] => {
+            return Err(format!(
+                "atof_required Switchyard profile requires named ATOF endpoint {required_name:?}"
+            ));
+        }
+        _ => {
+            return Err(format!(
+                "ATOF endpoint name {required_name:?} must resolve to exactly one endpoint"
+            ));
+        }
+    };
+    let transport = endpoint.get("transport").map_or_else(
+        || Some(AtofEndpointTransport::default()),
+        |value| value.as_str().and_then(AtofEndpointTransport::parse),
+    );
+    if transport != Some(AtofEndpointTransport::HttpPost) {
+        return Err(format!(
+            "Switchyard ATOF endpoint {required_name:?} must use transport = http_post"
+        ));
+    }
+    let field_name_policy = endpoint.get("field_name_policy").map_or_else(
+        || Some(AtofEndpointFieldNamePolicy::default()),
+        |value| value.as_str().and_then(AtofEndpointFieldNamePolicy::parse),
+    );
+    if field_name_policy != Some(AtofEndpointFieldNamePolicy::Preserve) {
+        return Err(format!(
+            "Switchyard ATOF endpoint {required_name:?} must use field_name_policy = preserve"
+        ));
     }
     if endpoint
         .get("header_env")
         .and_then(Json::as_object)
         .is_none_or(Map::is_empty)
     {
-        return Err(
-            "Switchyard ATOF endpoint authentication must use at least one environment-referenced header"
-                .into(),
-        );
+        return Err(format!(
+            "Switchyard ATOF endpoint {required_name:?} authentication must use at least one environment-referenced header"
+        ));
     }
     Ok(())
 }
 
-fn derived_atof_url(decision_api_url: &str) -> Result<String, String> {
-    let mut url = reqwest::Url::parse(decision_api_url)
-        .map_err(|error| format!("decision_api_url is invalid: {error}"))?;
-    url.set_path("/v1/atof/events");
-    url.set_query(None);
-    url.set_fragment(None);
-    Ok(url.to_string().trim_end_matches('/').to_string())
-}
-
 fn parse_config(config: &Map<String, Json>) -> Result<SwitchyardConfig, String> {
     serde_json::from_value(Json::Object(config.clone()))
         .map_err(|error| format!("invalid Switchyard plugin config: {error}"))
@@ -1218,6 +1226,18 @@ impl SwitchyardRuntime {
     }
 }
 
+fn validate_atof_endpoint_name(name: Option<&str>) -> Result<Option<&str>, String> {
+    if let Some(name) = name {
+        if name.trim().is_empty() {
+            return Err("atof_endpoint_name must be non-empty when configured".into());
+        }
+        if name != name.trim() {
+            return Err("atof_endpoint_name must not have leading or trailing whitespace".into());
+        }
+    }
+    Ok(name)
+}
+
 fn validate_config(config: &SwitchyardConfig) -> Result<(), String> {
     if config.version != 1 {
         return Err(format!(
@@ -1237,6 +1257,10 @@ fn validate_config(config: &SwitchyardConfig) -> Result<(), String> {
     if config.recent_message_count == 0 {
         return Err("recent_message_count must be greater than zero".into());
     }
+    let atof_endpoint_name = validate_atof_endpoint_name(config.atof_endpoint_name.as_deref())?;
+    if config.context_mode == ContextMode::AtofRequired && atof_endpoint_name.is_none() {
+        return Err("atof_required Switchyard profiles require atof_endpoint_name".into());
+    }
     let url = reqwest::Url::parse(&config.decision_api_url)
         .map_err(|error| format!("decision_api_url is invalid: {error}"))?;
     if !matches!(url.scheme(), "http" | "https") {