From 04d832f44513f4e94777c3cdfb54023b209f15d8 Mon Sep 17 00:00:00 2001 From: Bryan Bednarski Date: Tue, 14 Jul 2026 10:13:08 -0600 Subject: [PATCH 01/11] feat(plugin): validate named Switchyard ATOF endpoints Signed-off-by: Bryan Bednarski --- .../src/observability/plugin_component.rs | 36 ++++++++ .../observability/plugin_component_tests.rs | 24 +++++- crates/switchyard/src/component.rs | 85 +++++++++++-------- .../switchyard/tests/unit/component_tests.rs | 40 +++++++-- .../switchyard/hermes-ollama-plugins.toml | 2 + examples/switchyard/plugins.toml | 2 + examples/switchyard/real-e2e-plugins.toml | 2 + go/nemo_relay/observability_plugin.go | 1 + go/nemo_relay/observability_plugin_test.go | 7 +- 9 files changed, 155 insertions(+), 44 deletions(-) diff --git a/crates/core/src/observability/plugin_component.rs b/crates/core/src/observability/plugin_component.rs index f90de36f5..34cf59429 100644 --- a/crates/core/src/observability/plugin_component.rs +++ b/crates/core/src/observability/plugin_component.rs @@ -187,6 +187,9 @@ impl Default for AtofSectionConfig { #[derive(Debug, Clone, Serialize, Deserialize)] #[cfg_attr(feature = "schema", derive(schemars::JsonSchema))] pub struct AtofEndpointSectionConfig { + /// Optional stable name used by other components to reference this endpoint. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub name: Option, /// Endpoint URL. pub url: String, /// Transport: `http_post`, `websocket`, or `ndjson`. @@ -1790,7 +1793,40 @@ fn validate_atof_values( "ATOF mode must be 'append' or 'overwrite'".to_string(), ); } + let mut endpoint_names = HashSet::new(); for (index, endpoint) in section.endpoints.iter().enumerate() { + if let Some(name) = endpoint.name.as_deref() { + if name.trim().is_empty() { + push_policy_diag( + diagnostics, + policy.unsupported_value, + "observability.unsupported_value", + Some("atof".to_string()), + Some(format!("endpoints[{index}].name")), + format!("ATOF endpoints[{index}].name must be non-empty"), + ); + } else if name != name.trim() { + push_policy_diag( + diagnostics, + policy.unsupported_value, + "observability.unsupported_value", + Some("atof".to_string()), + Some(format!("endpoints[{index}].name")), + format!( + "ATOF endpoints[{index}].name must not have leading or trailing whitespace" + ), + ); + } else if !endpoint_names.insert(name) { + push_policy_diag( + diagnostics, + policy.unsupported_value, + "observability.unsupported_value", + Some("atof".to_string()), + Some(format!("endpoints[{index}].name")), + format!("ATOF endpoint name {name:?} must be unique"), + ); + } + } validate_atof_endpoint_values(diagnostics, policy, index, endpoint); } } diff --git a/crates/core/tests/unit/observability/plugin_component_tests.rs b/crates/core/tests/unit/observability/plugin_component_tests.rs index 318b1b417..18c7393fb 100644 --- a/crates/core/tests/unit/observability/plugin_component_tests.rs +++ b/crates/core/tests/unit/observability/plugin_component_tests.rs @@ -205,9 +205,10 @@ fn default_config_and_component_conversion_cover_public_shape() { assert!(atof.filename.is_none()); let parsed_atof: AtofSectionConfig = serde_json::from_value(json!({ - "endpoints": [{"url": "http://localhost/events"}] + "endpoints": [{"name": "switchyard", "url": "http://localhost/events"}] })) .unwrap(); + assert_eq!(parsed_atof.endpoints[0].name.as_deref(), Some("switchyard")); assert_eq!(parsed_atof.endpoints[0].transport, "http_post"); assert_eq!(parsed_atof.endpoints[0].field_name_policy, "preserve"); @@ -305,6 +306,7 @@ fn schema_contains_every_supported_observability_option() { "filename", "mode", "endpoints", + "name", "agent_name", "agent_version", "model_name", @@ -570,7 +572,10 @@ fn atof_endpoint_validation_rejects_bad_values() { {"url": "http://localhost/events", "transport": "bogus"}, {"url": "http://localhost/events", "transport": "ndjson", "timeout_millis": 0}, {"url": "not a url", "transport": "http_post"}, - {"url": "http://localhost/events", "transport": "http_post", "field_name_policy": "bogus"} + {"url": "http://localhost/events", "transport": "http_post", "field_name_policy": "bogus"}, + {"name": "switchyard", "url": "http://localhost/first"}, + {"name": "switchyard", "url": "http://localhost/second"}, + {"name": " ", "url": "http://localhost/blank"} ] } }))); @@ -607,6 +612,18 @@ fn atof_endpoint_validation_rejects_bad_values() { .iter() .any(|diag| { diag.field.as_deref() == Some("endpoints[4].field_name_policy") }) ); + assert!( + report + .diagnostics + .iter() + .any(|diag| { diag.field.as_deref() == Some("endpoints[6].name") }) + ); + assert!( + report + .diagnostics + .iter() + .any(|diag| { diag.field.as_deref() == Some("endpoints[7].name") }) + ); } #[test] @@ -616,6 +633,7 @@ fn build_atof_endpoint_config_maps_headers_timeout_and_rejects_transport() { let config = build_atof_endpoint_config( 2, AtofEndpointSectionConfig { + name: Some("switchyard".into()), url: "ws://127.0.0.1:47632/events".into(), transport: "websocket".into(), headers: headers.clone(), @@ -648,6 +666,7 @@ fn build_atof_endpoint_config_maps_headers_timeout_and_rejects_transport() { let error = build_atof_endpoint_config( 3, AtofEndpointSectionConfig { + name: None, url: "http://127.0.0.1:47632/events".into(), transport: "smtp".into(), headers: std::collections::HashMap::new(), @@ -662,6 +681,7 @@ fn build_atof_endpoint_config_maps_headers_timeout_and_rejects_transport() { let error = build_atof_endpoint_config( 4, AtofEndpointSectionConfig { + name: None, url: "http://127.0.0.1:47632/events".into(), transport: "http_post".into(), headers: std::collections::HashMap::new(), diff --git a/crates/switchyard/src/component.rs b/crates/switchyard/src/component.rs index ab48e258b..f7d9e364f 100644 --- a/crates/switchyard/src/component.rs +++ b/crates/switchyard/src/component.rs @@ -217,9 +217,9 @@ pub struct SwitchyardConfig { pub targets: BTreeMap, /// 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, + pub atof_endpoint_name: Option, } impl Default for SwitchyardConfig { @@ -244,7 +244,7 @@ impl Default for SwitchyardConfig { openai_responses: String::new(), anthropic_messages: String::new(), }, - atof_endpoint_url: None, + atof_endpoint_name: None, } } } @@ -269,7 +269,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,10 +401,9 @@ 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 = switchyard.atof_endpoint_name.as_deref().ok_or_else(|| { + "atof_required Switchyard profiles require atof_endpoint_name".to_string() + })?; let observability = config .components .iter() @@ -419,49 +418,55 @@ pub fn validate_switchyard_atof_configuration(config: &PluginConfig) -> Result<( .ok_or_else(|| { "atof_required Switchyard profiles require an enabled ATOF endpoint".to_string() })?; - let endpoint = endpoints + let matching_endpoints = endpoints .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" - }) - .ok_or_else(|| { - format!("atof_required Switchyard profile requires HTTP ATOF endpoint {required_url}") - })?; + .filter(|endpoint| endpoint.get("name").and_then(Json::as_str) == Some(required_name)) + .collect::>(); + let endpoint = match matching_endpoints.as_slice() { + [endpoint] => *endpoint, + [] => { + 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" + )); + } + }; + if endpoint + .get("transport") + .and_then(Json::as_str) + .unwrap_or("http_post") + != "http_post" + { + return Err(format!( + "Switchyard ATOF endpoint {required_name:?} must use transport = http_post" + )); + } 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()); + 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 { - 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) -> Result { serde_json::from_value(Json::Object(config.clone())) .map_err(|error| format!("invalid Switchyard plugin config: {error}")) @@ -1237,6 +1242,18 @@ 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 = 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()); + } + if let Some(name) = atof_endpoint_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()); + } + } 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") { diff --git a/crates/switchyard/tests/unit/component_tests.rs b/crates/switchyard/tests/unit/component_tests.rs index a6e7d5e92..d02d37802 100644 --- a/crates/switchyard/tests/unit/component_tests.rs +++ b/crates/switchyard/tests/unit/component_tests.rs @@ -254,6 +254,12 @@ fn configuration_validation_rejects_unsafe_or_ambiguous_bindings() { let mut candidate = config(url.clone()); candidate.recent_message_count = 0; assert_invalid(candidate, "recent_message_count"); + let mut candidate = config(url.clone()); + candidate.context_mode = ContextMode::AtofRequired; + assert_invalid(candidate, "atof_endpoint_name"); + let mut candidate = config(url.clone()); + candidate.atof_endpoint_name = Some(" switchyard ".into()); + assert_invalid(candidate, "leading or trailing whitespace"); let candidate = config("file:///tmp/decision".into()); assert_invalid(candidate, "must use http or https"); let mut candidate = config(url.clone()); @@ -340,7 +346,7 @@ fn atof_cross_component_validation_reports_each_activation_mismatch() { let mut switchyard = config("http://switchyard.test/v1/routing/decision".into()); switchyard.context_mode = ContextMode::AtofRequired; - switchyard.atof_endpoint_url = Some("http://events.test/v1/atof/events".into()); + switchyard.atof_endpoint_name = Some("switchyard".into()); let mut plugin_config = PluginConfig { components: vec![switchyard.into()], ..PluginConfig::default() @@ -349,7 +355,8 @@ fn atof_cross_component_validation_reports_each_activation_mismatch() { kind: "observability".into(), enabled: true, config: json!({"atof": {"enabled": true, "endpoints": [{ - "url": "http://wrong.test/v1/atof/events", + "name": "other", + "url": "http://events.test/v1/atof/events", "header_env": {"authorization": "TOKEN"} }]}}) .as_object() @@ -359,11 +366,31 @@ fn atof_cross_component_validation_reports_each_activation_mismatch() { assert!( validate_switchyard_atof_configuration(&plugin_config) .unwrap_err() - .contains("requires HTTP ATOF endpoint") + .contains("requires named ATOF endpoint") ); - plugin_config.components[1].config["atof"]["endpoints"][0]["url"] = - json!("http://events.test/v1/atof/events"); + plugin_config.components[1].config["atof"]["endpoints"][0]["name"] = json!("switchyard"); + plugin_config.components[1].config["atof"]["endpoints"][0]["transport"] = json!("websocket"); + assert!( + validate_switchyard_atof_configuration(&plugin_config) + .unwrap_err() + .contains("transport = http_post") + ); + plugin_config.components[1].config["atof"]["endpoints"][0]["transport"] = json!("http_post"); + let duplicate = plugin_config.components[1].config["atof"]["endpoints"][0].clone(); + plugin_config.components[1].config["atof"]["endpoints"] + .as_array_mut() + .unwrap() + .push(duplicate); + assert!( + validate_switchyard_atof_configuration(&plugin_config) + .unwrap_err() + .contains("exactly one endpoint") + ); + plugin_config.components[1].config["atof"]["endpoints"] + .as_array_mut() + .unwrap() + .pop(); plugin_config.components[1].config["atof"]["endpoints"][0]["field_name_policy"] = json!("snake_case"); assert!( @@ -677,6 +704,7 @@ fn identity_policy_requires_stable_request_scope_only_for_atof_profiles() { assert_eq!(synthetic.identity.quality, "synthetic"); config.context_mode = ContextMode::AtofRequired; + config.atof_endpoint_name = Some("switchyard".into()); let atof_runtime = SwitchyardRuntime::new(config).unwrap(); assert!( atof_runtime @@ -808,6 +836,7 @@ fn routing_decision_mark_has_canonical_shape_and_mirrored_identity() { fn atof_required_cross_component_validation_is_context_sensitive() { let mut switchyard = config("http://switchyard.test:8080/v1/routing/decision".into()); switchyard.context_mode = ContextMode::AtofRequired; + switchyard.atof_endpoint_name = Some("switchyard".into()); let mut plugin_config = PluginConfig { components: vec![switchyard.into()], ..PluginConfig::default() @@ -819,6 +848,7 @@ fn atof_required_cross_component_validation_is_context_sensitive() { config: json!({"atof": { "enabled": true, "endpoints": [{ + "name": "switchyard", "url": "http://switchyard.test:8080/v1/atof/events", "transport": "http_post", "field_name_policy": "preserve", diff --git a/examples/switchyard/hermes-ollama-plugins.toml b/examples/switchyard/hermes-ollama-plugins.toml index 6a7d128b0..85ce6fa88 100644 --- a/examples/switchyard/hermes-ollama-plugins.toml +++ b/examples/switchyard/hermes-ollama-plugins.toml @@ -13,6 +13,7 @@ decision_api_url = "http://127.0.0.1:4000/v1/routing/decision" decision_profile_id = "hermes-ollama-stage-router" request_materialization = "full_body" context_mode = "atof_required" +atof_endpoint_name = "switchyard" decision_timeout_millis = 120000 max_retries = 3 recent_message_count = 8 @@ -60,6 +61,7 @@ output_directory = "." filename = "trajectory.atof.jsonl" [[components.config.atof.endpoints]] +name = "switchyard" url = "http://127.0.0.1:4000/v1/atof/events" transport = "http_post" field_name_policy = "preserve" diff --git a/examples/switchyard/plugins.toml b/examples/switchyard/plugins.toml index f99d9e4c3..109bdabb9 100644 --- a/examples/switchyard/plugins.toml +++ b/examples/switchyard/plugins.toml @@ -14,6 +14,7 @@ decision_api_url = "http://127.0.0.1:4000/v1/routing/decision" decision_profile_id = "smart-stage-router" request_materialization = "recent_message_window" context_mode = "atof_required" +atof_endpoint_name = "switchyard" decision_timeout_millis = 25 max_retries = 3 recent_message_count = 8 @@ -60,6 +61,7 @@ enabled = true mode = "append" [[components.config.atof.endpoints]] +name = "switchyard" url = "http://127.0.0.1:4000/v1/atof/events" transport = "http_post" field_name_policy = "preserve" diff --git a/examples/switchyard/real-e2e-plugins.toml b/examples/switchyard/real-e2e-plugins.toml index 9c6351a6d..1b5da7063 100644 --- a/examples/switchyard/real-e2e-plugins.toml +++ b/examples/switchyard/real-e2e-plugins.toml @@ -13,6 +13,7 @@ decision_api_url = "http://127.0.0.1:4000/v1/routing/decision" decision_profile_id = "remote-stage-router" request_materialization = "summary_only" context_mode = "atof_required" +atof_endpoint_name = "switchyard" decision_timeout_millis = 1000 max_retries = 3 recent_message_count = 8 @@ -58,6 +59,7 @@ enabled = true mode = "append" [[components.config.atof.endpoints]] +name = "switchyard" url = "http://127.0.0.1:4000/v1/atof/events" transport = "http_post" field_name_policy = "preserve" diff --git a/go/nemo_relay/observability_plugin.go b/go/nemo_relay/observability_plugin.go index c79488919..0d79e2611 100644 --- a/go/nemo_relay/observability_plugin.go +++ b/go/nemo_relay/observability_plugin.go @@ -41,6 +41,7 @@ type ObservabilityAtofConfig struct { // ObservabilityAtofEndpoint configures one streaming destination for raw ATOF events. type ObservabilityAtofEndpoint struct { + Name string `json:"name,omitempty"` URL string `json:"url"` Transport string `json:"transport,omitempty"` Headers map[string]string `json:"headers,omitempty"` diff --git a/go/nemo_relay/observability_plugin_test.go b/go/nemo_relay/observability_plugin_test.go index 7c94c33b9..f09cfd7cf 100644 --- a/go/nemo_relay/observability_plugin_test.go +++ b/go/nemo_relay/observability_plugin_test.go @@ -36,6 +36,7 @@ func TestObservabilityConfigHelpers(t *testing.T) { t.Fatalf("unexpected ATOF defaults: %#v", atof) } atof.Endpoints = []ObservabilityAtofEndpoint{{ + Name: "archive", URL: "http://localhost:8080/events", Transport: "http_post", Headers: map[string]string{"X-Test": "yes"}, @@ -85,15 +86,15 @@ func TestObservabilityConfigHelpers(t *testing.T) { t.Fatalf("expected serialized ATOF endpoints, got %#v", atofConfig) } firstEndpoint, ok := endpoints[0].(map[string]any) - if !ok || firstEndpoint["field_name_policy"] != "replace_dots" { + if !ok || firstEndpoint["name"] != "archive" || firstEndpoint["field_name_policy"] != "replace_dots" { t.Fatalf("expected serialized ATOF endpoint field name policy, got %#v", endpoints) } serialized, err := json.Marshal(wrapped) if err != nil { t.Fatalf("marshal observability component failed: %v", err) } - if !strings.Contains(string(serialized), `"field_name_policy":"replace_dots"`) { - t.Fatalf("expected field_name_policy in serialized component, got %s", serialized) + if !strings.Contains(string(serialized), `"name":"archive"`) || !strings.Contains(string(serialized), `"field_name_policy":"replace_dots"`) { + t.Fatalf("expected named ATOF endpoint in serialized component, got %s", serialized) } assertWrappedAtifStorageConfig(t, wrapped.Config["atif"].(map[string]any)) if wrapped.Config["opentelemetry"].(map[string]any)["mark_projection"] != "tool" { From 42a21af85694432d6eb473bc39d760f1a7da660d Mon Sep 17 00:00:00 2001 From: Bryan Bednarski Date: Tue, 14 Jul 2026 11:09:12 -0600 Subject: [PATCH 02/11] fix: expose named ATOF endpoints in plugin helpers Signed-off-by: Bryan Bednarski --- crates/node/observability.d.ts | 1 + crates/node/tests/observability_plugin_tests.mjs | 2 ++ python/nemo_relay/observability.py | 2 ++ python/nemo_relay/observability.pyi | 1 + python/tests/test_observability_plugin.py | 3 +++ 5 files changed, 9 insertions(+) diff --git a/crates/node/observability.d.ts b/crates/node/observability.d.ts index eb15be134..5ed1a5d3c 100644 --- a/crates/node/observability.d.ts +++ b/crates/node/observability.d.ts @@ -15,6 +15,7 @@ export interface AtofConfig { } export interface AtofEndpointConfig { + name?: string; url: string; transport?: 'http_post' | 'websocket' | 'ndjson' | string; headers?: Record; diff --git a/crates/node/tests/observability_plugin_tests.mjs b/crates/node/tests/observability_plugin_tests.mjs index 94b825031..e13b58284 100644 --- a/crates/node/tests/observability_plugin_tests.mjs +++ b/crates/node/tests/observability_plugin_tests.mjs @@ -63,6 +63,7 @@ describe('observability plugin helpers', () => { const config = observability.atofConfig({ endpoints: [ { + name: 'switchyard', url: 'http://localhost:8080/events', transport: 'http_post', headers: { 'X-Test': 'yes' }, @@ -74,6 +75,7 @@ describe('observability plugin helpers', () => { assert.deepEqual(config.endpoints, [ { + name: 'switchyard', url: 'http://localhost:8080/events', transport: 'http_post', headers: { 'X-Test': 'yes' }, diff --git a/python/nemo_relay/observability.py b/python/nemo_relay/observability.py index ad837a0a1..b524da536 100644 --- a/python/nemo_relay/observability.py +++ b/python/nemo_relay/observability.py @@ -59,6 +59,7 @@ class AtofEndpointConfig: """Streaming destination for raw ATOF events.""" url: str + name: str | None = None transport: Literal["http_post", "websocket", "ndjson"] = "http_post" headers: dict[str, str] = field(default_factory=dict) timeout_millis: int = 3000 @@ -68,6 +69,7 @@ def to_dict(self) -> JsonObject: """Serialize this ATOF endpoint config to the canonical JSON object shape.""" return _normalize_object( { + "name": self.name, "url": self.url, "transport": self.transport, "headers": self.headers, diff --git a/python/nemo_relay/observability.pyi b/python/nemo_relay/observability.pyi index f562a10a1..0bed1f800 100644 --- a/python/nemo_relay/observability.pyi +++ b/python/nemo_relay/observability.pyi @@ -22,6 +22,7 @@ class ConfigPolicy: @dataclass(slots=True) class AtofEndpointConfig: url: str = ... + name: str | None = ... transport: Literal["http_post", "websocket", "ndjson"] = ... headers: dict[str, str] = field(default_factory=dict) timeout_millis: int = ... diff --git a/python/tests/test_observability_plugin.py b/python/tests/test_observability_plugin.py index e536fcaf8..6cf5e528e 100644 --- a/python/tests/test_observability_plugin.py +++ b/python/tests/test_observability_plugin.py @@ -103,12 +103,14 @@ def test_s3_storage_config_serializes_credential_fields(self): def test_atof_endpoint_config_serializes_streaming_fields(self): endpoint = AtofEndpointConfig( url="http://localhost:8080/events", + name="switchyard", transport="http_post", headers={"X-Test": "yes"}, timeout_millis=1000, field_name_policy="replace_dots", ) assert endpoint.to_dict() == { + "name": "switchyard", "url": "http://localhost:8080/events", "transport": "http_post", "headers": {"X-Test": "yes"}, @@ -116,6 +118,7 @@ def test_atof_endpoint_config_serializes_streaming_fields(self): "field_name_policy": "replace_dots", } assert AtofConfig(endpoints=[endpoint]).to_dict()["endpoints"] == [endpoint.to_dict()] + assert "name" not in AtofEndpointConfig(url="http://localhost:8080/events").to_dict() def test_http_storage_config_serializes_headers(self): s3 = S3StorageConfig(bucket="archive") From 26076d5b55da61ed1a163e23de21e503efb71615 Mon Sep 17 00:00:00 2001 From: Bryan Bednarski Date: Tue, 14 Jul 2026 11:24:30 -0600 Subject: [PATCH 03/11] refactor: share ATOF endpoint defaults Signed-off-by: Bryan Bednarski --- .../src/observability/plugin_component.rs | 4 ++-- crates/switchyard/src/component.rs | 23 +++++++++---------- .../switchyard/tests/unit/component_tests.rs | 1 + 3 files changed, 14 insertions(+), 14 deletions(-) diff --git a/crates/core/src/observability/plugin_component.rs b/crates/core/src/observability/plugin_component.rs index 34cf59429..77583885b 100644 --- a/crates/core/src/observability/plugin_component.rs +++ b/crates/core/src/observability/plugin_component.rs @@ -2215,11 +2215,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 { diff --git a/crates/switchyard/src/component.rs b/crates/switchyard/src/component.rs index f7d9e364f..c43ed1eee 100644 --- a/crates/switchyard/src/component.rs +++ b/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, @@ -435,22 +436,20 @@ pub fn validate_switchyard_atof_configuration(config: &PluginConfig) -> Result<( )); } }; - if endpoint - .get("transport") - .and_then(Json::as_str) - .unwrap_or("http_post") - != "http_post" - { + 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" )); } - if endpoint - .get("field_name_policy") - .and_then(Json::as_str) - .unwrap_or("preserve") - != "preserve" - { + 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" )); diff --git a/crates/switchyard/tests/unit/component_tests.rs b/crates/switchyard/tests/unit/component_tests.rs index d02d37802..6cfa19c67 100644 --- a/crates/switchyard/tests/unit/component_tests.rs +++ b/crates/switchyard/tests/unit/component_tests.rs @@ -370,6 +370,7 @@ fn atof_cross_component_validation_reports_each_activation_mismatch() { ); plugin_config.components[1].config["atof"]["endpoints"][0]["name"] = json!("switchyard"); + assert!(validate_switchyard_atof_configuration(&plugin_config).is_ok()); plugin_config.components[1].config["atof"]["endpoints"][0]["transport"] = json!("websocket"); assert!( validate_switchyard_atof_configuration(&plugin_config) From 2bad4f6c2da12eea979fdb425c4d1d6bdf49e437 Mon Sep 17 00:00:00 2001 From: Bryan Bednarski Date: Tue, 14 Jul 2026 12:57:47 -0600 Subject: [PATCH 04/11] style(python): simplify ATOF sink annotation Signed-off-by: Bryan Bednarski --- python/nemo_relay/observability.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/python/nemo_relay/observability.py b/python/nemo_relay/observability.py index c8a459161..05c9f9076 100644 --- a/python/nemo_relay/observability.py +++ b/python/nemo_relay/observability.py @@ -87,7 +87,7 @@ class AtofConfig: """Multi-sink raw ATOF export settings.""" enabled: bool = False - sinks: list["AtofFileSinkConfig | AtofStreamSinkConfig"] | None = None + sinks: list[AtofFileSinkConfig | AtofStreamSinkConfig] | None = None def to_dict(self) -> JsonObject: """Serialize this ATOF config to the canonical JSON object shape.""" From 3d646c9e44055808e805d0dc397f9ac54dab079e Mon Sep 17 00:00:00 2001 From: Bryan Bednarski Date: Tue, 14 Jul 2026 13:02:31 -0600 Subject: [PATCH 05/11] fix(python): preserve ATOF sink constructor order Signed-off-by: Bryan Bednarski --- python/nemo_relay/observability.py | 2 +- python/nemo_relay/observability.pyi | 2 +- python/tests/test_observability_plugin.py | 6 ++++++ 3 files changed, 8 insertions(+), 2 deletions(-) diff --git a/python/nemo_relay/observability.py b/python/nemo_relay/observability.py index 05c9f9076..1fc60eae4 100644 --- a/python/nemo_relay/observability.py +++ b/python/nemo_relay/observability.py @@ -59,12 +59,12 @@ class AtofStreamSinkConfig: """Stream sink for raw ATOF events.""" url: str - name: str | None = None transport: Literal["http_post", "websocket", "ndjson"] = "http_post" headers: dict[str, str] = field(default_factory=dict) header_env: dict[str, str] = field(default_factory=dict) timeout_millis: int = 3000 field_name_policy: Literal["preserve", "replace_dots"] = "preserve" + name: str | None = None def to_dict(self) -> JsonObject: """Serialize this ATOF stream sink to the canonical JSON object shape.""" diff --git a/python/nemo_relay/observability.pyi b/python/nemo_relay/observability.pyi index 501cc1da8..a1c421632 100644 --- a/python/nemo_relay/observability.pyi +++ b/python/nemo_relay/observability.pyi @@ -22,12 +22,12 @@ class ConfigPolicy: @dataclass(slots=True) class AtofStreamSinkConfig: url: str = ... - name: str | None = ... transport: Literal["http_post", "websocket", "ndjson"] = ... headers: dict[str, str] = field(default_factory=dict) header_env: dict[str, str] = field(default_factory=dict) timeout_millis: int = ... field_name_policy: Literal["preserve", "replace_dots"] = ... + name: str | None = ... def to_dict(self) -> JsonObject: ... @dataclass(slots=True) diff --git a/python/tests/test_observability_plugin.py b/python/tests/test_observability_plugin.py index 42e109061..77a527c49 100644 --- a/python/tests/test_observability_plugin.py +++ b/python/tests/test_observability_plugin.py @@ -15,6 +15,7 @@ OBSERVABILITY_PLUGIN_KIND, AtifConfig, AtofConfig, + AtofEndpointConfig, AtofFileSinkConfig, AtofStreamSinkConfig, ComponentSpec, @@ -124,6 +125,11 @@ def test_atof_sink_config_serializes_streaming_fields(self): assert AtofConfig(sinks=[sink]).to_dict()["sinks"] == [sink.to_dict()] assert "name" not in AtofStreamSinkConfig(url="http://localhost:8080/events").to_dict() + def test_atof_endpoint_alias_preserves_positional_transport(self): + endpoint = AtofEndpointConfig("http://localhost:8080/events", "websocket") + assert endpoint.transport == "websocket" + assert endpoint.name is None + def test_http_storage_config_serializes_headers(self): s3 = S3StorageConfig(bucket="archive") http = HttpStorageConfig( From dc52e2035c9cc3bd8679e434353ad180db407cf7 Mon Sep 17 00:00:00 2001 From: Bryan Bednarski Date: Tue, 14 Jul 2026 13:12:38 -0600 Subject: [PATCH 06/11] fix(switchyard): validate ATOF endpoint names early Signed-off-by: Bryan Bednarski --- crates/switchyard/src/component.rs | 29 +++++++++++-------- .../switchyard/tests/unit/component_tests.rs | 24 +++++++++++++++ 2 files changed, 41 insertions(+), 12 deletions(-) diff --git a/crates/switchyard/src/component.rs b/crates/switchyard/src/component.rs index 483bb0fa8..56cdc8cfa 100644 --- a/crates/switchyard/src/component.rs +++ b/crates/switchyard/src/component.rs @@ -402,9 +402,10 @@ pub fn validate_switchyard_atof_configuration(config: &PluginConfig) -> Result<( if switchyard.context_mode != ContextMode::AtofRequired { return Ok(()); } - let required_name = switchyard.atof_endpoint_name.as_deref().ok_or_else(|| { - "atof_required Switchyard profiles require atof_endpoint_name".to_string() - })?; + 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() @@ -1225,6 +1226,18 @@ impl SwitchyardRuntime { } } +fn validate_atof_endpoint_name(name: Option<&str>) -> Result, 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!( @@ -1244,18 +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 = config.atof_endpoint_name.as_deref(); + 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()); } - if let Some(name) = atof_endpoint_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()); - } - } 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") { diff --git a/crates/switchyard/tests/unit/component_tests.rs b/crates/switchyard/tests/unit/component_tests.rs index 78205de9d..de0267073 100644 --- a/crates/switchyard/tests/unit/component_tests.rs +++ b/crates/switchyard/tests/unit/component_tests.rs @@ -334,6 +334,30 @@ fn disabled_protocol_defaults_are_optional() { SwitchyardRuntime::new(candidate).expect("Anthropic-only configuration should validate"); } +#[test] +fn atof_cross_component_validation_rejects_invalid_endpoint_names_before_lookup() { + for (name, expected) in [ + (" ", "atof_endpoint_name must be non-empty when configured"), + ( + " switchyard ", + "atof_endpoint_name must not have leading or trailing whitespace", + ), + ] { + let mut switchyard = config("http://switchyard.test/v1/routing/decision".into()); + switchyard.context_mode = ContextMode::AtofRequired; + switchyard.atof_endpoint_name = Some(name.into()); + let plugin_config = PluginConfig { + components: vec![switchyard.into()], + ..PluginConfig::default() + }; + + assert_eq!( + validate_switchyard_atof_configuration(&plugin_config).unwrap_err(), + expected + ); + } +} + #[test] fn atof_cross_component_validation_reports_each_activation_mismatch() { assert!(validate_switchyard_atof_configuration(&PluginConfig::default()).is_ok()); From 240fd809de2ba6c0e922c5b98a0b2abbfe8a013a Mon Sep 17 00:00:00 2001 From: Bryan Bednarski Date: Tue, 14 Jul 2026 14:06:07 -0600 Subject: [PATCH 07/11] fix(go): preserve ATOF endpoint compatibility Signed-off-by: Bryan Bednarski --- go/nemo_relay/observability_plugin.go | 4 ++++ go/nemo_relay/observability_plugin_test.go | 2 +- 2 files changed, 5 insertions(+), 1 deletion(-) diff --git a/go/nemo_relay/observability_plugin.go b/go/nemo_relay/observability_plugin.go index 459619067..c19d21689 100644 --- a/go/nemo_relay/observability_plugin.go +++ b/go/nemo_relay/observability_plugin.go @@ -81,6 +81,10 @@ func (config ObservabilityAtofStreamSinkConfig) MarshalJSON() ([]byte, error) { }{Type: "stream", alias: alias(config)}) } +// ObservabilityAtofEndpoint is the deprecated name for an ATOF stream sink. +// Deprecated: Use ObservabilityAtofStreamSinkConfig. +type ObservabilityAtofEndpoint = ObservabilityAtofStreamSinkConfig + // ObservabilityAtifConfig configures per-top-level-agent ATIF file export. type ObservabilityAtifConfig struct { Enabled bool `json:"enabled,omitempty"` diff --git a/go/nemo_relay/observability_plugin_test.go b/go/nemo_relay/observability_plugin_test.go index c0de7418c..c30ac9fa2 100644 --- a/go/nemo_relay/observability_plugin_test.go +++ b/go/nemo_relay/observability_plugin_test.go @@ -35,7 +35,7 @@ func TestObservabilityConfigHelpers(t *testing.T) { if atof.Enabled || len(atof.Sinks) != 0 { t.Fatalf("unexpected ATOF defaults: %#v", atof) } - atof.Sinks = []ObservabilityAtofSinkConfigurer{ObservabilityAtofStreamSinkConfig{ + atof.Sinks = []ObservabilityAtofSinkConfigurer{ObservabilityAtofEndpoint{ Name: "archive", URL: "http://localhost:8080/events", Transport: "http_post", From 6a38eadbcdbb359e94c08b87f40a198de209e550 Mon Sep 17 00:00:00 2001 From: Bryan Bednarski Date: Tue, 14 Jul 2026 17:16:51 -0600 Subject: [PATCH 08/11] style(node): preserve ATOF sink field order Signed-off-by: Bryan Bednarski --- crates/node/observability.d.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/node/observability.d.ts b/crates/node/observability.d.ts index 6cb5a67e3..0d93f8fa1 100644 --- a/crates/node/observability.d.ts +++ b/crates/node/observability.d.ts @@ -22,13 +22,13 @@ export interface AtofFileSinkConfig { export interface AtofStreamSinkConfig { type: 'stream'; - name?: string; url: string; transport?: 'http_post' | 'websocket' | 'ndjson' | string; headers?: Record; header_env?: Record; timeout_millis?: number; field_name_policy?: 'preserve' | 'replace_dots' | string; + name?: string; } /** @deprecated Use AtofStreamSinkConfig. */ From 13e1ac2df30b0e124593a29a10093f8a68b61b44 Mon Sep 17 00:00:00 2001 From: Bryan Bednarski Date: Tue, 14 Jul 2026 17:19:43 -0600 Subject: [PATCH 09/11] style(go): preserve ATOF sink field order Signed-off-by: Bryan Bednarski --- go/nemo_relay/observability_plugin.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/go/nemo_relay/observability_plugin.go b/go/nemo_relay/observability_plugin.go index c19d21689..2885eaad3 100644 --- a/go/nemo_relay/observability_plugin.go +++ b/go/nemo_relay/observability_plugin.go @@ -61,13 +61,13 @@ func (config ObservabilityAtofFileSinkConfig) MarshalJSON() ([]byte, error) { // ObservabilityAtofStreamSinkConfig configures one remote ATOF destination. type ObservabilityAtofStreamSinkConfig struct { - Name string `json:"name,omitempty"` URL string `json:"url"` Transport string `json:"transport,omitempty"` Headers map[string]string `json:"headers,omitempty"` HeaderEnv map[string]string `json:"header_env,omitempty"` TimeoutMillis uint64 `json:"timeout_millis,omitempty"` FieldNamePolicy string `json:"field_name_policy,omitempty"` + Name string `json:"name,omitempty"` } func (ObservabilityAtofStreamSinkConfig) atofSinkConfig() {} From 1d41388ee2cf9cd2b1b13b0cb120802f9d67a276 Mon Sep 17 00:00:00 2001 From: Bryan Bednarski Date: Tue, 14 Jul 2026 17:28:45 -0600 Subject: [PATCH 10/11] refactor(observability): deduplicate sink name diagnostics Signed-off-by: Bryan Bednarski --- .../src/observability/plugin_component.rs | 34 +++++++------------ 1 file changed, 13 insertions(+), 21 deletions(-) diff --git a/crates/core/src/observability/plugin_component.rs b/crates/core/src/observability/plugin_component.rs index 548048f5b..5e324fc02 100644 --- a/crates/core/src/observability/plugin_component.rs +++ b/crates/core/src/observability/plugin_component.rs @@ -1923,34 +1923,26 @@ fn validate_atof_values( } AtofSinkSectionConfig::Stream(stream) => { if let Some(name) = stream.name.as_deref() { - if name.trim().is_empty() { - push_policy_diag( - diagnostics, - policy.unsupported_value, - "observability.unsupported_value", - Some("atof".to_string()), - Some(format!("sinks[{index}].name")), - format!("ATOF sinks[{index}].name must be non-empty"), - ); - } else if name != name.trim() { - push_policy_diag( - diagnostics, - policy.unsupported_value, - "observability.unsupported_value", - Some("atof".to_string()), - Some(format!("sinks[{index}].name")), - format!( - "ATOF sinks[{index}].name must not have leading or trailing whitespace" - ), - ); + 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")), - format!("ATOF stream sink name {name:?} must be unique"), + message, ); } } From e626dd3fe8f5c6cf235b6c6f01c29017b57aa85c Mon Sep 17 00:00:00 2001 From: Bryan Bednarski Date: Tue, 14 Jul 2026 17:45:55 -0600 Subject: [PATCH 11/11] style(observability): order required sink fields first Signed-off-by: Bryan Bednarski --- crates/core/src/observability/plugin_component.rs | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/crates/core/src/observability/plugin_component.rs b/crates/core/src/observability/plugin_component.rs index 5e324fc02..a32eb317f 100644 --- a/crates/core/src/observability/plugin_component.rs +++ b/crates/core/src/observability/plugin_component.rs @@ -194,9 +194,6 @@ pub struct AtofFileSinkSectionConfig { #[derive(Debug, Clone, Serialize, Deserialize)] #[cfg_attr(feature = "schema", derive(schemars::JsonSchema))] pub struct AtofStreamSinkSectionConfig { - /// Optional stable name used by other components to reference this endpoint. - #[serde(default, skip_serializing_if = "Option::is_none")] - pub name: Option, /// Endpoint URL. pub url: String, /// Transport: `http_post`, `websocket`, or `ndjson`. @@ -218,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, } /// Per-trajectory ATIF exporter config. @@ -500,13 +500,13 @@ crate::editor_config! { crate::editor_config! { impl AtofStreamSinkSectionConfig { - name => { label: "name", kind: String, optional: true }, url => { label: "url", kind: String }, transport => { label: "transport", kind: Enum, values: ["http_post", "websocket", "ndjson"] }, headers => { label: "headers", kind: StringMap }, 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 }, } }