fix: surface ATIF delivery failures consistently

crates/core/src/observability/plugin_component.rs

@@ -63,8 +63,9 @@ use crate::observability::{
 use crate::plugin::{
     ATIF_RUNTIME_DELIVERY_FAILURE_MARKER, ConfigDiagnostic, ConfigPolicy, DiagnosticLevel,
     OTEL_RUNTIME_DELIVERY_FAILURE_MARKER, Plugin, PluginComponentSpec, PluginError,
-    PluginRegistration, PluginRegistrationContext, Result as PluginResult, UnsupportedBehavior,
-    apply_global_config_policy, deregister_plugin, register_builtin_plugin,
+    PluginRegistration, PluginRegistrationCleanupOutcome, PluginRegistrationContext,
+    Result as PluginResult, UnsupportedBehavior, apply_global_config_policy, deregister_plugin,
+    register_builtin_plugin,
 };
 use crate::plugin::{RuntimeDiagnostic, record_active_plugin_runtime_diagnostic};
 
@@ -892,41 +893,62 @@ fn register_atif_dispatcher(
     );
     ctx.register_subscriber("atif", dispatcher)?;
     let shutdown_storage = Arc::clone(&storage);
-    ctx.add_registration(PluginRegistration::new(
+    ctx.add_registration(PluginRegistration::new_with_outcome(
         "observability",
         ctx.qualify_name("atif.shutdown"),
         Box::new(move || {
-            let work = {
-                let mut guard = manager.lock().map_err(|err| {
-                    PluginError::Internal(format!("ATIF dispatcher lock poisoned: {err}"))
-                })?;
-                guard.flush_open_agents()
-            };
-            for (scope_uuid, name) in work.scope_subscribers {
-                deregister_atif_shutdown_subscriber(&scope_uuid, &name)?;
-            }
-            for export in work.exports {
-                let write = prepare_atif_shutdown_file(&export, Arc::clone(&manager))
-                    .map_err(observability_registration_error)?;
-                let agent_uuid = write.agent_uuid;
-                let targets = {
-                    let guard = manager.lock().map_err(|err| {
+            let work = match (|| -> PluginResult<_> {
+                let work = {
+                    let mut guard = manager.lock().map_err(|err| {
                         PluginError::Internal(format!("ATIF dispatcher lock poisoned: {err}"))
                     })?;
-                    guard.sink_targets()
+                    guard.flush_open_agents()
                 };
-                let results = write_atif(&write, shutdown_storage.as_slice(), &targets);
-                let mut guard = manager.lock().map_err(|err| {
-                    PluginError::Internal(format!("ATIF dispatcher lock poisoned: {err}"))
-                })?;
-                let _ = guard.complete_scope_write(agent_uuid, results);
+                for (scope_uuid, name) in &work.scope_subscribers {
+                    deregister_atif_shutdown_subscriber(scope_uuid, name)?;
+                }
+                Ok(work)
+            })() {
+                Ok(work) => work,
+                Err(error) => return PluginRegistrationCleanupOutcome::NotRemoved(error),
+            };
+
+            let delivery = (|| -> PluginResult<()> {
+                for export in work.exports {
+                    let write = prepare_atif_shutdown_file(&export, Arc::clone(&manager))
+                        .map_err(observability_registration_error)?;
+                    let agent_uuid = write.agent_uuid;
+                    let targets = {
+                        let guard = manager.lock().map_err(|err| {
+                            PluginError::Internal(format!("ATIF dispatcher lock poisoned: {err}"))
+                        })?;
+                        guard.sink_targets()
+                    };
+                    let results = write_atif(&write, shutdown_storage.as_slice(), &targets);
+                    let mut guard = manager.lock().map_err(|err| {
+                        PluginError::Internal(format!("ATIF dispatcher lock poisoned: {err}"))
+                    })?;
+                    let _ = guard.complete_scope_write(agent_uuid, results);
+                }
+                Ok(())
+            })();
+            if let Err(error) = delivery {
+                return PluginRegistrationCleanupOutcome::RemovedWithError(error);
+            }
+            let guard = match manager.lock() {
+                Ok(guard) => guard,
+                Err(error) => {
+                    return PluginRegistrationCleanupOutcome::RemovedWithError(
+                        PluginError::Internal(format!("ATIF dispatcher lock poisoned: {error}")),
+                    );
+                }
+            };
+            match guard.last_error_result() {
+                Ok(()) => PluginRegistrationCleanupOutcome::Removed,
+                Err(error) => PluginRegistrationCleanupOutcome::RemovedWithError(
+                    observability_registration_error(error),
+                ),
             }
-            let guard = manager.lock().map_err(|err| {
-                PluginError::Internal(format!("ATIF dispatcher lock poisoned: {err}"))
-            })?;
-            guard
-                .last_error_result()
-                .map_err(observability_registration_error)
         }),
     ));
     Ok(())
@@ -999,10 +1021,20 @@ fn register_opentelemetry(
     // Retain the subscribers as long as the registered fan-out callback exists.
     // Their tracer providers and exporter runtimes must outlive event delivery.
     let delivery_subscribers = subscribers.clone();
-    ctx.add_registration(PluginRegistration::new(
+    ctx.add_registration(PluginRegistration::new_with_outcome(
         "observability",
         ctx.qualify_name("opentelemetry.shutdown"),
-        Box::new(move || shutdown_opentelemetry_subscribers(&subscribers).map_or(Ok(()), Err)),
+        Box::new(
+            move || match shutdown_opentelemetry_subscribers(&subscribers) {
+                None => PluginRegistrationCleanupOutcome::Removed,
+                Some(OpenTelemetryShutdownFailure::Delivery(error)) => {
+                    PluginRegistrationCleanupOutcome::RemovedWithError(error)
+                }
+                Some(OpenTelemetryShutdownFailure::Other(error)) => {
+                    PluginRegistrationCleanupOutcome::NotRemoved(error)
+                }
+            },
+        ),
     ));
     ctx.register_subscriber(
         "opentelemetry",
@@ -1059,9 +1091,14 @@ fn build_opentelemetry_subscribers(
     Ok(subscribers)
 }
 
+enum OpenTelemetryShutdownFailure {
+    Delivery(PluginError),
+    Other(PluginError),
+}
+
 fn shutdown_opentelemetry_subscribers(
     subscribers: &[Arc<OpenTelemetrySubscriber>],
-) -> Option<PluginError> {
+) -> Option<OpenTelemetryShutdownFailure> {
     let mut errors = Vec::new();
     if let Err(error) = flush_subscribers() {
         errors.push(crate::observability::otel::OpenTelemetryError::Core(error));
@@ -1086,7 +1123,12 @@ fn shutdown_opentelemetry_subscribers(
     } else {
         format!("OpenTelemetry shutdown failures: {summary}")
     };
-    Some(PluginError::RegistrationFailed(message))
+    let error = PluginError::RegistrationFailed(message);
+    Some(if all_delivery_failures {
+        OpenTelemetryShutdownFailure::Delivery(error)
+    } else {
+        OpenTelemetryShutdownFailure::Other(error)
+    })
 }
 
 fn shutdown_opentelemetry_providers(
@@ -1624,16 +1666,31 @@ fn render_atif_filename(
             })?;
         let expression = rendered[selector_start..end].to_string();
         let (selector, fallback) = parse_atif_metadata_expression(&expression)?;
-        let value = selector
-            .split('.')
-            .fold(metadata, |value, segment| value?.get(segment))
-            .and_then(Json::as_str)
-            .or(fallback)
-            .ok_or_else(|| {
+        let mut resolved = metadata;
+        for segment in selector.split('.') {
+            resolved = match resolved {
+                Some(Json::Object(object)) => object.get(segment),
+                None | Some(Json::Null) => break,
+                Some(_) => {
+                    return Err(format!(
+                        "filename_template placeholder '{{metadata.{selector}}}' traversed a non-object value"
+                    ));
+                }
+            };
+        }
+        let value = match resolved {
+            Some(Json::String(value)) => value.as_str(),
+            None | Some(Json::Null) => fallback.ok_or_else(|| {
                 format!(
                     "filename_template placeholder '{{metadata.{selector}}}' must resolve to a string"
                 )
-            })?;
+            })?,
+            Some(_) => {
+                return Err(format!(
+                    "filename_template placeholder '{{metadata.{selector}}}' resolved to a non-string value"
+                ));
+            }
+        };
         if !is_safe_atif_metadata_path(value) {
             return Err(format!(
                 "metadata path '{selector}' must be a path-safe relative fragment"