diff --git a/crates/core/src/observability/plugin_component.rs b/crates/core/src/observability/plugin_component.rs index 116460fa4..4374e7da7 100644 --- a/crates/core/src/observability/plugin_component.rs +++ b/crates/core/src/observability/plugin_component.rs @@ -22,7 +22,7 @@ use std::borrow::Cow; use std::collections::{HashMap, HashSet}; use std::future::Future; use std::panic::{AssertUnwindSafe, catch_unwind}; -use std::path::PathBuf; +use std::path::{Component, Path, PathBuf}; use std::pin::Pin; #[cfg(feature = "object-store")] use std::sync::atomic::{AtomicU8, Ordering}; @@ -64,9 +64,12 @@ use crate::plugin::{ PluginRegistration, PluginRegistrationContext, Result as PluginResult, UnsupportedBehavior, apply_global_config_policy, deregister_plugin, register_builtin_plugin, }; +use crate::plugin::{RuntimeDiagnostic, record_active_plugin_runtime_diagnostic}; /// The plugin kind registered by the core crate. pub const OBSERVABILITY_PLUGIN_KIND: &str = "observability"; +/// Identifies teardown errors caused by recoverable ATIF delivery failures. +pub(crate) const ATIF_RUNTIME_DELIVERY_FAILURE_MARKER: &str = "ATIF runtime delivery failures"; /// Top-level observability component wrapper. /// @@ -1093,9 +1096,7 @@ struct AtifDispatcher { /// that cannot be isolated to a single sink. Once set, the dispatcher stops /// observing further events. fatal_error: Option, - /// Per-sink last error. A sink that recorded an error is skipped on - /// subsequent trajectories; other sinks continue to receive writes. - sink_errors: HashMap, + runtime_failures: Vec, validated_sinks: HashSet, } @@ -1195,7 +1196,7 @@ impl AtifDispatcher { scope_owners: HashMap::new(), scope_subscribers: HashMap::new(), fatal_error: None, - sink_errors: HashMap::new(), + runtime_failures: Vec::new(), validated_sinks: HashSet::new(), } } @@ -1227,6 +1228,12 @@ impl AtifDispatcher { let (filename, local_path) = match self.prepare_destination(&session_id, event.metadata()) { Ok(destination) => destination, Err(error) => { + self.record_runtime_failure( + "atif.destination_render_failed", + Some("filename_template".into()), + error.clone(), + Some(session_id.clone()), + ); log::warn!( target: "nemo_relay.observability", event = "atif_destination_render_failed", @@ -1330,6 +1337,9 @@ impl AtifDispatcher { agent_uuid: Uuid, results: Vec<(SinkLabel, std::io::Result<()>)>, ) -> Option<(Uuid, String)> { + let is_remote_fallback = results + .iter() + .any(|(label, _)| matches!(label, SinkLabel::Remote(_))); for (label, result) in results { if result.is_ok() && label == SinkLabel::Local @@ -1345,6 +1355,10 @@ impl AtifDispatcher { "ATIF storage access validated" ); } else if let Err(err) = result { + let (field, message) = match &label { + SinkLabel::Local => ("output_directory".to_string(), err.to_string()), + SinkLabel::Remote(index) => (format!("storage[{index}]"), err.to_string()), + }; match &label { SinkLabel::Local => log::warn!( target: "nemo_relay.observability", @@ -1356,9 +1370,25 @@ impl AtifDispatcher { reason = "write_failed"; "ATIF storage access failed" ), - SinkLabel::Remote(_) => {} + SinkLabel::Remote(index) => log::warn!( + target: "nemo_relay.observability", + event = "atif_remote_delivery_failed", + plugin_kind = OBSERVABILITY_PLUGIN_KIND, + exporter = "atif", + storage_index = *index; + "ATIF remote storage upload failed" + ), } - self.sink_errors.insert(label, err.to_string()); + self.record_runtime_failure( + match &label { + SinkLabel::Local if is_remote_fallback => "atif.local_fallback_failed", + SinkLabel::Local => "atif.local_write_failed", + SinkLabel::Remote(_) => "atif.remote_delivery_failed", + }, + Some(field), + message, + Some(agent_uuid.to_string()), + ); } } if let Some(agent) = self.agents.get_mut(&agent_uuid) { @@ -1413,6 +1443,16 @@ impl AtifDispatcher { if let Some(message) = &self.fatal_error { return Err(std::io::Error::other(message.clone())); } + if !self.runtime_failures.is_empty() { + return Err(std::io::Error::other(format!( + "{ATIF_RUNTIME_DELIVERY_FAILURE_MARKER}: {}", + self.runtime_failures + .iter() + .map(|diagnostic| format!("{} ({})", diagnostic.code, diagnostic.count)) + .collect::>() + .join(", ") + ))); + } Ok(()) } @@ -1431,10 +1471,8 @@ impl AtifDispatcher { session_id: &str, metadata: Option<&Json>, ) -> Result<(String, Option), String> { + validate_atif_filename_template(&self.config.filename_template)?; let filename = render_atif_filename(&self.config.filename_template, session_id, metadata)?; - if !self.config.storage.is_empty() { - return Ok((filename, None)); - } let directory = self .config .output_directory @@ -1446,18 +1484,42 @@ impl AtifDispatcher { fn sink_targets(&self) -> Vec { if self.config.storage.is_empty() { - if self.sink_errors.contains_key(&SinkLabel::Local) { - Vec::new() - } else { - vec![SinkLabel::Local] - } + vec![SinkLabel::Local] } else { (0..self.config.storage.len()) .map(SinkLabel::Remote) - .filter(|label| !self.sink_errors.contains_key(label)) .collect() } } + + fn record_runtime_failure( + &mut self, + code: &str, + field: Option, + message: String, + session_id: Option, + ) { + let diagnostic = RuntimeDiagnostic { + code: code.to_string(), + component: OBSERVABILITY_PLUGIN_KIND.to_string(), + field, + message, + session_id, + count: 1, + }; + if let Some(existing) = self + .runtime_failures + .iter_mut() + .find(|existing| existing.code == diagnostic.code && existing.field == diagnostic.field) + { + existing.message = diagnostic.message.clone(); + existing.session_id = diagnostic.session_id.clone(); + existing.count += 1; + } else { + self.runtime_failures.push(diagnostic.clone()); + } + record_active_plugin_runtime_diagnostic(diagnostic); + } } fn is_valid_atif_metadata_selector(selector: &str) -> bool { @@ -1489,6 +1551,23 @@ fn validate_atif_filename_template(template: &str) -> Result<(), String> { return Err("ATIF filename_template must contain '{session_id}'".to_string()); } + let literal_path = template + .replace("{session_id}", "session") + .replace("{metadata.", "metadata."); + if Path::new(&literal_path).is_absolute() + || Path::new(&literal_path).components().any(|component| { + matches!( + component, + Component::ParentDir + | Component::CurDir + | Component::RootDir + | Component::Prefix(_) + ) + }) + { + return Err("ATIF filename_template must be a path-safe relative path".to_string()); + } + let mut cursor = 0; while let Some(relative_start) = template[cursor..].find(PREFIX) { let selector_start = cursor + relative_start + PREFIX.len(); @@ -1713,7 +1792,7 @@ fn write_atif( storage: &[Arc], targets: &[SinkLabel], ) -> Vec<(SinkLabel, std::io::Result<()>)> { - targets + let mut results = targets .iter() .map(|label| { let result = match label { @@ -1727,7 +1806,22 @@ fn write_atif( }; (label.clone(), result) }) - .collect() + .collect::>(); + if !targets.is_empty() + && targets + .iter() + .all(|label| matches!(label, SinkLabel::Remote(_))) + && results.iter().all(|(_, result)| result.is_err()) + { + let fallback = match &write.local_path { + Some(path) => write_atif_local(path, &write.payload), + None => Err(std::io::Error::other( + "ATIF local fallback has no output path", + )), + }; + results.push((SinkLabel::Local, fallback)); + } + results } fn write_atif_local(path: &PathBuf, payload: &[u8]) -> std::io::Result<()> { diff --git a/crates/core/src/plugin.rs b/crates/core/src/plugin.rs index 601cb5559..d7b21be38 100644 --- a/crates/core/src/plugin.rs +++ b/crates/core/src/plugin.rs @@ -44,6 +44,7 @@ use crate::api::runtime::{ ToolExecutionFn, ToolInterceptFn, ToolSanitizeFn, }; use crate::api::subscriber::{deregister_subscriber, register_subscriber}; +use crate::observability::plugin_component::ATIF_RUNTIME_DELIVERY_FAILURE_MARKER; pub use nemo_relay_types::plugin::{ConfigDiagnostic, DiagnosticLevel}; pub mod dynamic; @@ -73,6 +74,8 @@ pub(crate) enum PluginDeregistrationOutcome { static PLUGIN_HANDLERS: LazyLock> = LazyLock::new(|| RwLock::new(HashMap::new())); static ACTIVE_PLUGIN_CONFIGURATION: LazyLock>> = LazyLock::new(|| Mutex::new(None)); +static LAST_FAILED_RUNTIME_DIAGNOSTICS_REPORT: LazyLock>> = + LazyLock::new(|| Mutex::new(None)); static PLUGIN_MUTATION_OWNER: LazyLock> = LazyLock::new(|| Mutex::new(PluginMutationOwner::Idle)); static NEXT_PLUGIN_REGISTRATION_ID: AtomicU64 = AtomicU64::new(1); @@ -186,6 +189,29 @@ pub struct ConfigReport { /// Validation and compatibility diagnostics in evaluation order. #[serde(default)] pub diagnostics: Vec, + /// Runtime delivery diagnostics recorded after activation. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub runtime_diagnostics: Vec, +} + +/// Bounded aggregate for a runtime failure observed by an active plugin. +#[derive(Debug, Clone, Serialize, Deserialize)] +#[cfg_attr(feature = "schema", derive(schemars::JsonSchema))] +pub struct RuntimeDiagnostic { + /// Stable failure classification. + pub code: String, + /// Plugin component that reported the failure. + pub component: String, + /// Optional configuration field associated with the failure. + #[serde(skip_serializing_if = "Option::is_none")] + pub field: Option, + /// Latest human-readable failure detail. + pub message: String, + /// Latest affected trajectory session identifier. + #[serde(skip_serializing_if = "Option::is_none")] + pub session_id: Option, + /// Number of failures aggregated into this entry. + pub count: u64, } impl ConfigReport { @@ -1508,7 +1534,10 @@ async fn initialize_plugins_exact_inner( component_count = enabled_component_count; "Plugin configuration activation started" ); - let mut report = ConfigReport { diagnostics }; + let mut report = ConfigReport { + diagnostics, + ..ConfigReport::default() + }; report .diagnostics .extend(validate_plugin_config(&config).diagnostics); @@ -2045,7 +2074,7 @@ fn clear_plugin_configuration_inner() -> PluginHostClearOutcome { let flush_error = crate::api::runtime::subscriber_dispatcher::flush_queued_subscribers() .err() .map(|error| error.to_string()); - let previous = { + let mut registrations = { let mut guard = match ACTIVE_PLUGIN_CONFIGURATION.lock() { Ok(guard) => guard, Err(err) => { @@ -2057,13 +2086,34 @@ fn clear_plugin_configuration_inner() -> PluginHostClearOutcome { }; } }; - guard.take() + guard + .as_mut() + .map(|state| std::mem::take(&mut state.registrations)) }; - let deregistration_errors = previous - .map(|mut previous_state| rollback_registrations_checked(&mut previous_state.registrations)) + // Keep the report installed while callbacks run so runtime diagnostics + // emitted by teardown work can be recorded against it. + let deregistration_errors = registrations + .as_mut() + .map(rollback_registrations_checked) .unwrap_or_default(); - let callbacks_cleared = deregistration_errors.is_empty(); - let deregistration_error = (!callbacks_cleared).then(|| { + let teardown_report = match ACTIVE_PLUGIN_CONFIGURATION.lock() { + Ok(mut guard) => guard.take().map(|state| state.report), + Err(err) => { + return PluginHostClearOutcome { + result: Err(PluginError::Internal(format!( + "active plugin configuration lock poisoned: {err}" + ))), + callbacks_cleared: false, + }; + } + }; + // Runtime delivery failures are reported by an otherwise successful + // deregistration callback. They must propagate without treating callback + // removal itself as unsafe. + let callbacks_cleared = deregistration_errors + .iter() + .all(|error| error.contains(ATIF_RUNTIME_DELIVERY_FAILURE_MARKER)); + let deregistration_error = (!deregistration_errors.is_empty()).then(|| { PluginError::RegistrationFailed(format!( "plugin teardown failed: {}", deregistration_errors.join("; ") @@ -2077,6 +2127,16 @@ fn clear_plugin_configuration_inner() -> PluginHostClearOutcome { "{deregister}; subscriber flush also failed: {flush}" ))), }; + if result.is_ok() { + if let Ok(mut guard) = LAST_FAILED_RUNTIME_DIAGNOSTICS_REPORT.lock() { + *guard = None; + } + } else if let Some(report) = + teardown_report.filter(|report| !report.runtime_diagnostics.is_empty()) + && let Ok(mut guard) = LAST_FAILED_RUNTIME_DIAGNOSTICS_REPORT.lock() + { + *guard = Some(report); + } PluginHostClearOutcome { result, callbacks_cleared, @@ -2183,22 +2243,55 @@ fn plugin_mutation_conflict(owner: PluginMutationOwner) -> PluginError { PluginError::Conflict(message.into()) } -/// Returns the last successfully configured plugin report. +/// Returns the active plugin report or a report retained after a failed teardown. /// -/// `None` indicates that no plugin configuration is currently active. +/// `None` indicates that no plugin configuration is active and no failed +/// teardown report is retained. /// /// # Returns -/// The last successful [`ConfigReport`], or `None` when no configuration is -/// active. +/// The active [`ConfigReport`], or the report containing runtime diagnostics +/// from the last failed teardown. /// /// # Notes /// This is a snapshot of the last successful activation and does not re-run /// validation. pub fn active_plugin_report() -> Option { - ACTIVE_PLUGIN_CONFIGURATION + let active_report = ACTIVE_PLUGIN_CONFIGURATION .lock() .ok() - .and_then(|guard| guard.as_ref().map(|state| state.report.clone())) + .and_then(|guard| guard.as_ref().map(|state| state.report.clone())); + active_report.or_else(|| { + LAST_FAILED_RUNTIME_DIAGNOSTICS_REPORT + .lock() + .ok() + .and_then(|guard| guard.clone()) + }) +} + +/// Record a bounded runtime diagnostic against the active plugin report. +pub fn record_active_plugin_runtime_diagnostic(diagnostic: RuntimeDiagnostic) { + let Ok(mut guard) = ACTIVE_PLUGIN_CONFIGURATION.lock() else { + return; + }; + let Some(state) = guard.as_mut() else { + return; + }; + if let Some(existing) = state + .report + .runtime_diagnostics + .iter_mut() + .find(|existing| { + existing.code == diagnostic.code + && existing.component == diagnostic.component + && existing.field == diagnostic.field + }) + { + existing.message = diagnostic.message; + existing.session_id = diagnostic.session_id; + existing.count += 1; + } else { + state.report.runtime_diagnostics.push(diagnostic); + } } /// Rolls back registrations in reverse order, ignoring rollback failures. @@ -2368,6 +2461,9 @@ fn store_active_plugin_configuration( report, registrations, }); + if let Ok(mut guard) = LAST_FAILED_RUNTIME_DIAGNOSTICS_REPORT.lock() { + *guard = None; + } Ok(()) } diff --git a/crates/core/tests/integration/atif_storage_tests.rs b/crates/core/tests/integration/atif_storage_tests.rs index 524905835..09d3d9f04 100644 --- a/crates/core/tests/integration/atif_storage_tests.rs +++ b/crates/core/tests/integration/atif_storage_tests.rs @@ -494,7 +494,7 @@ fn atif_storage_posts_trajectory_to_http_endpoints() { } #[test] -fn atif_storage_http_non_2xx_marks_sink_unhealthy() { +fn atif_storage_http_non_2xx_retries_on_the_next_trajectory() { let _guard = PLUGIN_TEST_LOCK.lock().unwrap(); reset_runtime(); let mut server = start_http_server(2, vec![("/fail", 500)]); @@ -532,11 +532,31 @@ fn atif_storage_http_non_2xx_marks_sink_unhealthy() { server.stop(); { let requests = server.received.lock().unwrap(); - assert_eq!(requests.len(), 1); - assert_eq!(requests[0].method, "POST"); - assert_eq!(requests[0].path, "/fail"); + assert_eq!(requests.len(), 2); + assert!( + requests + .iter() + .all(|request| request.method == "POST" && request.path == "/fail") + ); + let request_session_ids = requests + .iter() + .map(|request| { + request + .headers + .get("x-nemo-relay-atif-session-id") + .expect("ATIF HTTP requests should identify their trajectory") + .clone() + }) + .collect::>(); + assert!(request_session_ids.contains(&handle.uuid.to_string())); + assert!(request_session_ids.contains(&second.uuid.to_string())); } - clear_plugin_configuration().expect("plugin teardown should ignore unhealthy sink errors"); + let teardown = + clear_plugin_configuration().expect_err("plugin teardown should report failed uploads"); + assert!( + teardown.to_string().contains("atif.remote_delivery_failed"), + "teardown should identify the failed remote destination: {teardown}" + ); // SAFETY: cleanup of test-only env var. unsafe { std::env::remove_var("NEMO_RELAY_ATIF_HTTP_TEST_TOKEN"); diff --git a/crates/core/tests/unit/observability/plugin_component_tests.rs b/crates/core/tests/unit/observability/plugin_component_tests.rs index 745be8c04..494b7b731 100644 --- a/crates/core/tests/unit/observability/plugin_component_tests.rs +++ b/crates/core/tests/unit/observability/plugin_component_tests.rs @@ -706,7 +706,7 @@ fn atof_stream_header_validation_reports_invalid_values_and_environment_names() } #[test] -fn atif_dispatcher_surfaces_fatal_and_disabled_local_sink_states() { +fn atif_dispatcher_surfaces_fatal_and_runtime_failure_states() { let mut dispatcher = AtifDispatcher::new(AtifSectionConfig::default()); assert_eq!(dispatcher.sink_targets(), vec![SinkLabel::Local]); dispatcher.fatal_error = Some("fatal export failure".into()); @@ -719,10 +719,14 @@ fn atif_dispatcher_surfaces_fatal_and_disabled_local_sink_states() { ); dispatcher.fatal_error = None; - dispatcher - .sink_errors - .insert(SinkLabel::Local, "local write failed".into()); - assert!(dispatcher.sink_targets().is_empty()); + dispatcher.record_runtime_failure( + "atif.local_fallback_failed", + Some("output_directory".into()), + "local write failed".into(), + Some("session".into()), + ); + assert_eq!(dispatcher.sink_targets(), vec![SinkLabel::Local]); + assert!(dispatcher.last_error_result().is_err()); } #[test] @@ -1828,7 +1832,20 @@ fn atif_filename_template_routes_by_metadata_and_skips_invalid_paths() { .unwrap(); pop(&valid); - clear_plugin_configuration().unwrap(); + flush_subscribers().unwrap(); + assert!( + crate::plugin::active_plugin_report() + .unwrap() + .runtime_diagnostics + .iter() + .any(|diagnostic| diagnostic.code == "atif.destination_render_failed") + ); + let teardown = clear_plugin_configuration().unwrap_err(); + assert!( + teardown + .to_string() + .contains("atif.destination_render_failed") + ); let invalid_filename = format!("trajectory-{}.json", invalid.uuid); assert!( !dir.join(&invalid_filename).exists() @@ -1842,6 +1859,16 @@ fn atif_filename_template_routes_by_metadata_and_skips_invalid_paths() { )) .exists() ); + + futures::executor::block_on(initialize_plugins_exact(plugin_config(json!({ + "atif": { + "enabled": true, + "output_directory": dir, + "filename_template": "trajectory-{session_id}.json" + } + })))) + .expect("ATIF teardown errors should not block later activation"); + clear_plugin_configuration().expect("later ATIF teardown should succeed"); } #[test] @@ -2222,7 +2249,7 @@ fn atif_completed_top_level_agent_is_evicted_after_write() { let dispatcher = manager.lock().unwrap(); assert!(dispatcher.fatal_error.is_none()); - assert!(dispatcher.sink_errors.is_empty()); + assert!(dispatcher.runtime_failures.is_empty()); assert!(!dispatcher.agents.contains_key(&agent.uuid)); assert!(!dispatcher.scope_subscribers.contains_key(&agent.uuid)); assert!(path.exists()); @@ -2270,14 +2297,12 @@ fn atif_dispatcher_records_failed_agent_writes() { vec![(SinkLabel::Local, Err(std::io::Error::other("disk full")))], ); assert!(scope_subscriber.is_some()); + assert_eq!(dispatcher.runtime_failures.len(), 1); assert_eq!( - dispatcher - .sink_errors - .get(&SinkLabel::Local) - .map(String::as_str), - Some("disk full") + dispatcher.runtime_failures[0].code, + "atif.local_write_failed" ); - assert!(dispatcher.last_error_result().is_ok()); + assert!(dispatcher.last_error_result().is_err()); drop(dispatcher); pop(&agent); } @@ -2311,6 +2336,28 @@ fn write_atif_reports_missing_local_path_and_unregistered_remote_sink() { assert!(remote_error.contains("ATIF storage support is not enabled in this build")); } +#[test] +fn write_atif_spills_to_local_when_all_remote_sinks_fail() { + let dir = temp_dir("observability-atif-remote-fallback"); + let path = dir.join("trajectory.json"); + let agent_uuid = Uuid::now_v7(); + let write = PendingAtifWrite { + agent_uuid, + session_id: agent_uuid.to_string(), + filename: "trajectory.json".into(), + local_path: Some(path.clone()), + payload: b"{}".to_vec(), + }; + + let results = write_atif(&write, &[], &[SinkLabel::Remote(0)]); + + assert_eq!(results.len(), 2); + assert!(results[0].1.is_err()); + assert_eq!(results[1].0, SinkLabel::Local); + assert!(results[1].1.is_ok()); + assert_eq!(fs::read(path).unwrap(), b"{}"); +} + #[test] fn atif_dispatcher_default_output_path_uses_current_directory() { let dispatcher = AtifDispatcher::new(AtifSectionConfig::default()); @@ -2377,6 +2424,8 @@ fn atif_metadata_template_values_must_be_safe_path_fragments() { ); for template in [ + "/tmp/trajectory-{session_id}.json", + "../trajectory-{session_id}.json", "{metadata.}/trajectory-{session_id}.json", "{metadata.tenant..id}/trajectory-{session_id}.json", "{metadata.tenant/trajectory-{session_id}.json", @@ -2495,6 +2544,47 @@ fn atif_explicit_options_and_open_agent_teardown_are_written() { pop(&agent); } +#[test] +#[cfg(feature = "object-store")] +fn atif_open_agent_teardown_failure_retains_runtime_diagnostic_report() { + let _guard = crate::observability::test_mutex().lock().unwrap(); + reset_runtime(); + let dir = temp_dir("observability-atif-open-agent-delivery-failure"); + let (endpoint, server) = start_http_status_server("500 Internal Server Error"); + let config = plugin_config(json!({ + "atif": { + "enabled": true, + "output_directory": dir, + "filename_template": "trajectory-{session_id}.json", + "storage": [{"type": "http", "endpoint": endpoint}] + } + })); + futures::executor::block_on(initialize_plugins_exact(config)).unwrap(); + let agent = push_agent("open-agent-delivery-failure"); + + let teardown = clear_plugin_configuration().unwrap_err(); + assert!(teardown.to_string().contains("atif.remote_delivery_failed")); + server.join().unwrap().unwrap(); + + let report = crate::plugin::active_plugin_report() + .expect("failed teardown should retain its runtime diagnostics"); + let diagnostic = report + .runtime_diagnostics + .iter() + .find(|diagnostic| diagnostic.code == "atif.remote_delivery_failed") + .expect("remote failure should be retained in the report"); + assert_eq!(diagnostic.field.as_deref(), Some("storage[0]")); + assert_eq!( + diagnostic.session_id.as_deref(), + Some(agent.uuid.to_string().as_str()) + ); + assert!(!diagnostic.message.is_empty()); + + pop(&agent); + clear_plugin_configuration().unwrap(); + assert!(crate::plugin::active_plugin_report().is_none()); +} + #[test] fn atif_rejects_unsafe_template_and_ignores_non_top_level_agents() { let _guard = crate::observability::test_mutex().lock().unwrap(); diff --git a/crates/core/tests/unit/plugin_tests.rs b/crates/core/tests/unit/plugin_tests.rs index b39f6f934..13fd05f52 100644 --- a/crates/core/tests/unit/plugin_tests.rs +++ b/crates/core/tests/unit/plugin_tests.rs @@ -764,6 +764,7 @@ fn test_config_report_has_errors() { field: None, message: "boom".into(), }], + ..ConfigReport::default() }; assert!(report.has_errors()); } @@ -1048,7 +1049,13 @@ fn test_plugin_helper_defaults_and_policy_diagnostics() { assert_eq!(diagnostics[0].component.as_deref(), Some("warn.plugin")); assert_eq!(diagnostics[0].field.as_deref(), Some("field")); assert_eq!(diagnostics[1].level, DiagnosticLevel::Error); - assert_eq!(join_error_messages(&ConfigReport { diagnostics }), "error"); + assert_eq!( + join_error_messages(&ConfigReport { + diagnostics, + ..ConfigReport::default() + }), + "error" + ); reset_global(); } @@ -1498,6 +1505,50 @@ fn test_checked_teardown_reports_unremoved_registrations() { reset_global(); } +#[test] +fn test_teardown_runtime_diagnostics_remain_in_the_plugin_report() { + let _guard = lock_runtime_owner(); + reset_global(); + store_active_plugin_configuration( + PluginConfig::default(), + ConfigReport::default(), + vec![PluginRegistration::new( + "fixture", + "atif-shutdown", + Box::new(|| { + record_active_plugin_runtime_diagnostic(RuntimeDiagnostic { + code: "atif.remote_delivery_failed".into(), + component: "observability".into(), + field: Some("storage[0]".into()), + message: "HTTP 500".into(), + session_id: Some("session-123".into()), + count: 1, + }); + Err(PluginError::RegistrationFailed(format!( + "{}: atif.remote_delivery_failed (1)", + crate::observability::plugin_component::ATIF_RUNTIME_DELIVERY_FAILURE_MARKER + ))) + }), + )], + ) + .unwrap(); + + let outcome = clear_plugin_configuration_inner(); + assert!(outcome.callbacks_cleared); + assert!(outcome.result.is_err()); + let report = active_plugin_report().expect("failed teardown should retain its report"); + assert_eq!(report.runtime_diagnostics.len(), 1); + let diagnostic = &report.runtime_diagnostics[0]; + assert_eq!(diagnostic.code, "atif.remote_delivery_failed"); + assert_eq!(diagnostic.field.as_deref(), Some("storage[0]")); + assert_eq!(diagnostic.message, "HTTP 500"); + assert_eq!(diagnostic.session_id.as_deref(), Some("session-123")); + + clear_plugin_configuration_inner(); + assert!(active_plugin_report().is_none()); + reset_global(); +} + #[test] fn test_legacy_clear_retains_mutation_owner_after_incomplete_teardown() { let _guard = lock_runtime_owner(); diff --git a/crates/node/plugin.d.ts b/crates/node/plugin.d.ts index 4d05ba448..d4b77b92e 100644 --- a/crates/node/plugin.d.ts +++ b/crates/node/plugin.d.ts @@ -49,6 +49,17 @@ export interface ConfigDiagnostic { /** Validation or activation report for a plugin configuration. */ export interface ConfigReport { diagnostics: ConfigDiagnostic[]; + runtime_diagnostics?: RuntimeDiagnostic[]; +} + +/** One bounded aggregate of a runtime plugin failure. */ +export interface RuntimeDiagnostic { + code: string; + component: string; + field?: string; + message: string; + session_id?: string; + count: number; } /** One top-level plugin component. */ diff --git a/crates/node/src/api/mod.rs b/crates/node/src/api/mod.rs index a290b86a8..9daec5dad 100644 --- a/crates/node/src/api/mod.rs +++ b/crates/node/src/api/mod.rs @@ -5020,7 +5020,7 @@ pub fn clear_plugin_configuration() -> napi::Result<()> { clear_plugin_configuration_impl().map_err(|e| napi::Error::from_reason(e.to_string())) } -/// Return the last successfully configured plugin report. +/// Return the active plugin report or one retained after a teardown failure with runtime diagnostics. #[napi] pub fn active_plugin_report() -> napi::Result> { active_plugin_report_impl() diff --git a/docs/configure-plugins/observability/atif.mdx b/docs/configure-plugins/observability/atif.mdx index 817874ad8..dec1ad05c 100644 --- a/docs/configure-plugins/observability/atif.mdx +++ b/docs/configure-plugins/observability/atif.mdx @@ -67,9 +67,9 @@ The following table describes the top-level ATIF settings: | `model_name` | `unknown` | Default model metadata when no call-level model is present. | | `tool_definitions` | Omitted | Optional ATIF tool metadata. | | `extra` | Omitted | Optional ATIF agent metadata. | -| `output_directory` | Current working directory | Directory containing trajectory files. Ignored when `storage` is non-empty. | +| `output_directory` | Current working directory | Directory containing trajectory files and the recovery copy when every configured remote destination fails. | | `filename_template` | `nemo-relay-atif-{session_id}.json` | Must contain `{session_id}`. Can contain `{metadata.}` placeholders for metadata-based routing. | -| `storage` | Omitted | Optional list of remote storage destinations. When non-empty, trajectories are uploaded to every configured backend instead of being written locally. Refer to [Remote Storage](#remote-storage). | +| `storage` | Omitted | Optional list of remote storage destinations. When non-empty, trajectories are uploaded to every configured backend. If all fail for one trajectory, Relay writes a local recovery copy. Refer to [Remote Storage](#remote-storage). | ### Metadata-Based Paths @@ -97,9 +97,10 @@ relative path fragment. Slash-separated segments can contain ASCII letters, digits, `-`, `_`, `.`, and `~`. Relay rejects empty segments, `.` and `..` segments, absolute paths, backslashes, spaces, and other characters. If a placeholder has no fallback and is missing or non-string, or if its resolved -value is unsafe, Relay skips that trajectory and logs -`atif_destination_render_failed`. The rendered filename applies to local, S3, -and HTTP storage in the same way as a static filename. +value is unsafe, Relay skips that trajectory. It records a runtime diagnostic +in `plugin.report()` and causes plugin teardown to fail. Literal template text +must also be a relative, traversal-free path. The rendered filename applies to +local, S3, and HTTP storage in the same way as a static filename. The CLI gateway parses `x-nemo-relay-session-metadata` as JSON and merges it into the top-level scope metadata: @@ -113,7 +114,9 @@ x-nemo-relay-session-metadata: {"atif_prefix":"tenant-a/session-123"} Use `storage` when local trace files are not durable across sessions, such as in sandboxed runtimes. When `storage` is non-empty, Relay uploads each completed trajectory to every configured backend instead of writing a local file. -`output_directory` is ignored. +`output_directory` is ignored for successful remote writes, but Relay uses it +for a local recovery copy when every configured remote destination fails for a +trajectory. Each storage entry is tagged with a `type` discriminator so additional backends can be added without breaking existing configs. S3-compatible object @@ -162,8 +165,10 @@ also sets `x-nemo-relay-atif-filename` and identity that local files and S3 uploads use. HTTP `2xx` responses are treated as success. Any non-`2xx` response or -transport error records the endpoint as unhealthy and skips it for later -trajectories. Other configured destinations continue to receive writes. +transport error records a runtime delivery diagnostic. Relay attempts every +configured destination again for each later trajectory, while other configured +destinations continue to receive writes. If every remote destination fails for +a trajectory, Relay writes a local recovery copy under `output_directory`. The following table describes HTTP storage settings: @@ -235,11 +240,15 @@ renders `filename_template` the same way it does for local files, so a local→remote transition keeps object names stable. Relay adds a trailing `/` to `key_prefix` when one is missing. -If an upload fails for a destination, the ATIF exporter records that destination -as unhealthy and skips it for later trajectories. Other destinations continue -to receive writes. The exporter reports fatal dispatcher failures, such as -trajectory serialization failures, during plugin teardown. It isolates -per-destination sink failures to the failed sink. +If an upload fails for a destination, the ATIF exporter records a runtime +delivery diagnostic and retries that destination for each later trajectory. +Other destinations continue to receive writes. When every remote destination +fails for one trajectory, Relay writes a local recovery copy under +`output_directory`; a successful remote destination does not create that copy. +The diagnostic remains visible in `plugin.report()` and is retained there after +a failed teardown. Teardown also reports the degraded delivery, even when the +local recovery write succeeds. Fatal dispatcher failures, such as trajectory +serialization failures, are also reported during teardown. ## Expected Output diff --git a/go/nemo_relay/plugin.go b/go/nemo_relay/plugin.go index 8a7f1942e..1de3de731 100644 --- a/go/nemo_relay/plugin.go +++ b/go/nemo_relay/plugin.go @@ -218,7 +218,18 @@ type ConfigDiagnostic struct { // ConfigReport is the validation or activation report for a plugin config. type ConfigReport struct { - Diagnostics []ConfigDiagnostic `json:"diagnostics,omitempty"` + Diagnostics []ConfigDiagnostic `json:"diagnostics,omitempty"` + RuntimeDiagnostics []RuntimeDiagnostic `json:"runtime_diagnostics,omitempty"` +} + +// RuntimeDiagnostic is one bounded aggregate of a runtime plugin failure. +type RuntimeDiagnostic struct { + Code string `json:"code"` + Component string `json:"component"` + Field *string `json:"field,omitempty"` + Message string `json:"message"` + SessionID *string `json:"session_id,omitempty"` + Count uint64 `json:"count"` } // PluginComponentSpec is one top-level plugin component. @@ -494,9 +505,10 @@ func ClearPluginConfiguration() error { return checkStatus(C.nemo_relay_clear_plugin_configuration()) } -// ActivePluginReport returns the last successfully activated plugin report. +// ActivePluginReport returns the active plugin report or one retained after a +// teardown failure with runtime diagnostics. // -// A nil report means no plugin configuration is currently active. +// A nil report means neither report is available. func ActivePluginReport() (*ConfigReport, error) { raw, err := activePluginReportJSON() if err != nil { diff --git a/python/nemo_relay/_native.pyi b/python/nemo_relay/_native.pyi index e5ed22496..43f2fa5f1 100644 --- a/python/nemo_relay/_native.pyi +++ b/python/nemo_relay/_native.pyi @@ -2397,11 +2397,11 @@ def clear_plugin_configuration_async() -> Awaitable[None]: ... def active_plugin_report() -> Optional[_JsonObject]: - """Return the active plugin report. + """Return the active plugin report or a failed-teardown diagnostic report. Returns: - Report JSON object for the last active configuration, or ``None`` if no - plugin configuration is active. + Report JSON object for the active configuration or a failed teardown + with runtime diagnostics, or ``None`` if neither exists. """ ... diff --git a/python/nemo_relay/plugin.py b/python/nemo_relay/plugin.py index ce9e0b430..dbd807083 100644 --- a/python/nemo_relay/plugin.py +++ b/python/nemo_relay/plugin.py @@ -80,10 +80,25 @@ class ConfigDiagnostic(_ConfigDiagnosticRequired, total=False): field: str +class _RuntimeDiagnosticRequired(TypedDict): + code: str + component: str + message: str + count: int + + +class RuntimeDiagnostic(_RuntimeDiagnosticRequired, total=False): + """One aggregated runtime failure from an active plugin.""" + + field: str + session_id: str + + class ConfigReport(TypedDict): """Validation or activation report for a plugin config.""" diagnostics: list[ConfigDiagnostic] + runtime_diagnostics: list[RuntimeDiagnostic] DynamicPluginKind = Literal["rust_dynamic", "worker"] @@ -520,15 +535,16 @@ async def plugin(config: PluginConfig | JsonObject, *, clear_on_exit: bool = Tru def report() -> ConfigReport | None: - """Return the last successful plugin report. + """Return the active plugin report or a failed-teardown diagnostic report. Returns: - The active `ConfigReport`, or `None` when no plugin configuration is - currently active. + The active `ConfigReport`, a report retained after a teardown failure + with runtime diagnostics, or `None` when neither exists. Behavior: - This reports the last successfully activated configuration snapshot. It - does not revalidate plugin state or inspect pending registrations. + A retained teardown report is cleared by the next successful + activation or clean clear. This function does not revalidate plugin + state or inspect pending registrations. """ return cast(ConfigReport | None, _active_plugin_report()) @@ -582,6 +598,7 @@ def deregister(plugin_kind: str) -> bool: __all__ = [ "ComponentSpec", "ConfigDiagnostic", + "RuntimeDiagnostic", "ConfigPolicy", "ConfigReport", "DynamicPluginActivationSpec", diff --git a/python/nemo_relay/plugin.pyi b/python/nemo_relay/plugin.pyi index 174b7f921..39bc464d5 100644 --- a/python/nemo_relay/plugin.pyi +++ b/python/nemo_relay/plugin.pyi @@ -33,8 +33,19 @@ class ConfigDiagnostic(_ConfigDiagnosticRequired, total=False): component: str field: str +class _RuntimeDiagnosticRequired(TypedDict): + code: str + component: str + message: str + count: int + +class RuntimeDiagnostic(_RuntimeDiagnosticRequired, total=False): + field: str + session_id: str + class ConfigReport(TypedDict): diagnostics: list[ConfigDiagnostic] + runtime_diagnostics: list[RuntimeDiagnostic] class PluginContext(Protocol): def register_subscriber(self, name: str, callback: Callable[[Event], None]) -> None: ... diff --git a/skills/nemo-relay-plugin-observability/references/atif.md b/skills/nemo-relay-plugin-observability/references/atif.md index 6e8f0226a..1ab0f98fb 100644 --- a/skills/nemo-relay-plugin-observability/references/atif.md +++ b/skills/nemo-relay-plugin-observability/references/atif.md @@ -68,7 +68,13 @@ live OTLP spans. - Metadata values must be relative path fragments using ASCII letters, digits, `-`, `_`, `.`, `~`, and safe `/`-separated segments. - Missing, non-string, or unsafe metadata values skip only the affected - trajectory and produce an `atif_destination_render_failed` warning. + trajectory, record an `atif.destination_render_failed` runtime diagnostic, + and cause plugin teardown to fail. Literal template text must be relative + and traversal-free. +- Remote storage failures are retried for each later trajectory. If every + remote destination fails for a trajectory, Relay writes a local recovery + copy under `output_directory`; the failure remains visible in + `plugin.report()` and teardown still fails. ## Checklist