fix: recover failed ATIF remote deliveries

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,6 +64,7 @@ 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";
@@ -1093,9 +1094,7 @@ struct AtifDispatcher {
     /// that cannot be isolated to a single sink. Once set, the dispatcher stops
     /// observing further events.
     fatal_error: Option<String>,
-    /// Per-sink last error. A sink that recorded an error is skipped on
-    /// subsequent trajectories; other sinks continue to receive writes.
-    sink_errors: HashMap<SinkLabel, String>,
+    runtime_failures: Vec<RuntimeDiagnostic>,
     validated_sinks: HashSet<SinkLabel>,
 }
 
@@ -1195,7 +1194,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 +1226,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",
@@ -1345,6 +1350,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 +1365,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(
+                    if matches!(label, SinkLabel::Local) {
+                        "atif.local_fallback_failed"
+                    } else {
+                        "atif.remote_delivery_failed"
+                    },
+                    Some(field),
+                    message,
+                    Some(agent_uuid.to_string()),
+                );
             }
         }
         if let Some(agent) = self.agents.get_mut(&agent_uuid) {
@@ -1413,6 +1438,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 failures: {}",
+                self.runtime_failures
+                    .iter()
+                    .map(|diagnostic| format!("{} ({})", diagnostic.code, diagnostic.count))
+                    .collect::<Vec<_>>()
+                    .join(", ")
+            )));
+        }
         Ok(())
     }
 
@@ -1431,10 +1466,8 @@ impl AtifDispatcher {
         session_id: &str,
         metadata: Option<&Json>,
     ) -> Result<(String, Option<PathBuf>), 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 +1479,42 @@ impl AtifDispatcher {
 
     fn sink_targets(&self) -> Vec<SinkLabel> {
         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<String>,
+        message: String,
+        session_id: Option<String>,
+    ) {
+        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 +1546,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 +1787,7 @@ fn write_atif(
     storage: &[Arc<AtifRemoteStorage>],
     targets: &[SinkLabel],
 ) -> Vec<(SinkLabel, std::io::Result<()>)> {
-    targets
+    let mut results = targets
         .iter()
         .map(|label| {
             let result = match label {
@@ -1727,7 +1801,22 @@ fn write_atif(
             };
             (label.clone(), result)
         })
-        .collect()
+        .collect::<Vec<_>>();
+    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<()> {

crates/core/src/plugin.rs

@@ -186,6 +186,29 @@ pub struct ConfigReport {
     /// Validation and compatibility diagnostics in evaluation order.
     #[serde(default)]
     pub diagnostics: Vec<ConfigDiagnostic>,
+    /// Runtime delivery diagnostics recorded after activation.
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub runtime_diagnostics: Vec<RuntimeDiagnostic>,
+}
+
+/// 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<String>,
+    /// Latest human-readable failure detail.
+    pub message: String,
+    /// Latest affected trajectory session identifier.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub session_id: Option<String>,
+    /// Number of failures aggregated into this entry.
+    pub count: u64,
 }
 
 impl ConfigReport {
@@ -1508,7 +1531,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);
@@ -2025,8 +2051,13 @@ fn clear_plugin_configuration_inner() -> PluginHostClearOutcome {
     let deregistration_errors = previous
         .map(|mut previous_state| rollback_registrations_checked(&mut previous_state.registrations))
         .unwrap_or_default();
-    let callbacks_cleared = deregistration_errors.is_empty();
-    let deregistration_error = (!callbacks_cleared).then(|| {
+    // 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 failures"));
+    let deregistration_error = (!deregistration_errors.is_empty()).then(|| {
         PluginError::RegistrationFailed(format!(
             "plugin teardown failed: {}",
             deregistration_errors.join("; ")
@@ -2164,6 +2195,32 @@ pub fn active_plugin_report() -> Option<ConfigReport> {
         .and_then(|guard| guard.as_ref().map(|state| state.report.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.
 ///
 /// This is used internally during failed initialization and by

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,14 @@ 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")
+        );
     }
-    clear_plugin_configuration().expect("plugin teardown should ignore unhealthy sink errors");
+    clear_plugin_configuration().expect_err("plugin teardown should report failed uploads");
     // SAFETY: cleanup of test-only env var.
     unsafe {
         std::env::remove_var("NEMO_RELAY_ATIF_HTTP_TEST_TOKEN");