fix(otel): emit one correlated signal set and shut providers down cleanly

agent-identity: unknown
agent-persona: generalist
agent-supervisor: unavailable
agent-tool: OMP
agent-tool-version: 18.0.3
agent-runtime: OMP 18.0.3
tooling-profile: dotfiles@e4789b0

Author: schickling-assistant

src/run.rs

@@ -1800,7 +1800,6 @@ fn finish_reconcile_pass(span: &tracing::Span, report: &UpReport) {
 pub fn up_once(root: &Path, this_host: &str, runner: &dyn Runner) -> anyhow::Result<UpReport> {
     let task_context = TaskCompileContext::current(root.to_path_buf())?;
     let mut debounce = LivenessDebounce::new(DEBOUNCE_GRACE);
-    let pass_span = crate::telemetry::PassSpan::start(this_host);
     let started = Instant::now();
     let span = reconcile_span(this_host);
     let report = {
@@ -1817,7 +1816,6 @@ pub fn up_once(root: &Path, this_host: &str, runner: &dyn Runner) -> anyhow::Res
         finish_reconcile_pass(&span, &report);
         report
     };
-    pass_span.finish(report.crash_loops.len(), report.unparked.len());
     crate::metrics::record_reconcile_pass(started.elapsed(), !report.errors.is_empty());
     Ok(report)
 }
@@ -1895,7 +1893,6 @@ pub(crate) fn reconcile_pass_specs_with_sessions(
     presentation_cursor: &mut PresentationPatchCursor,
 ) -> UpReport {
     let started = Instant::now();
-    let pass_span = crate::telemetry::PassSpan::start(this_host);
     let span = reconcile_span(this_host);
     let mut report = UpReport::default();
     {
@@ -1917,7 +1914,6 @@ pub(crate) fn reconcile_pass_specs_with_sessions(
         }
         finish_reconcile_pass(&span, &report);
     }
-    pass_span.finish(report.crash_loops.len(), report.unparked.len());
     crate::metrics::record_reconcile_pass(started.elapsed(), !report.errors.is_empty());
     report
 }
@@ -1960,7 +1956,8 @@ pub fn up_once_selected(
     this_host: &str,
     runner: &dyn Runner,
 ) -> anyhow::Result<UpReport> {
-    let pass_span = crate::telemetry::PassSpan::start(this_host);
+    let span = reconcile_span(this_host);
+    let _entered = span.enter();
     let _catalog_lock = crate::CatalogLock::shared(catalog_root)
         .context("acquire shared catalog-authoring lock for selected reconcile")?;
     let found = crate::discovery::discover(catalog_root);
@@ -2007,7 +2004,7 @@ pub fn up_once_selected(
         || Ok(()),
     )?;
     report.absorb(execution);
-    pass_span.finish(report.crash_loops.len(), report.unparked.len());
+    finish_reconcile_pass(&span, &report);
     Ok(report)
 }
 
@@ -2322,7 +2319,6 @@ fn up_loop_until(
         let mut pre = UpReport::default();
         park_channel.grant_requests(&mut cap, &mut pre);
         let mut report = {
-            let pass_span = crate::telemetry::PassSpan::start(this_host);
             let started = Instant::now();
             let span = reconcile_span(this_host);
             let pass = {
@@ -2339,7 +2335,6 @@ fn up_loop_until(
                 finish_reconcile_pass(&span, &pass);
                 pass
             };
-            pass_span.finish(pass.crash_loops.len(), pass.unparked.len());
             crate::metrics::record_reconcile_pass(started.elapsed(), !pass.errors.is_empty());
             pass
         };

src/telemetry.rs

@@ -13,11 +13,9 @@
 //! carry its trace/span ids.
 
 use std::io::IsTerminal as _;
-use std::sync::atomic::{AtomicBool, Ordering};
 
-use opentelemetry::trace::{Span as _, Tracer as _};
-use opentelemetry::KeyValue;
 use opentelemetry::trace::TracerProvider as _;
+use opentelemetry::KeyValue;
 use opentelemetry_appender_tracing::layer::OpenTelemetryTracingBridge;
 use opentelemetry_otlp::{LogExporter, MetricExporter, SpanExporter, WithExportConfig};
 use opentelemetry_sdk::logs::SdkLoggerProvider;
@@ -26,15 +24,6 @@ use opentelemetry_sdk::trace::SdkTracerProvider;
 use tracing_opentelemetry::OpenTelemetryLayer;
 use tracing_subscriber::layer::{Layer as _, SubscriberExt};
 
-static ENABLED: AtomicBool = AtomicBool::new(false);
-
-/// Process-wide export gate. Spans and other instrumented work check this before allocating
-/// anything, so the unset-endpoint path stays allocation-free (the provider guard alone cannot
-/// remove no-op span construction on the supervisor hot path).
-pub fn enabled() -> bool {
-    ENABLED.load(Ordering::Relaxed)
-}
-
 /// Seconds-scale explicit bucket boundaries for the duration histograms. The SDK's default
 /// boundaries are millisecond-tuned (`[0, 5, 10, …, 10000]`), so sub-second reconcile passes
 /// and session spawns would collapse into the lowest buckets and be indistinguishable.
@@ -57,9 +46,9 @@ fn duration_view(instrument: &Instrument) -> Option<Stream> {
     }
 }
 
-/// Guard holding the tracer, meter, and logger providers for a process lifetime. Dropping it
-/// flushes and shuts all three exporters down so short-lived CLI invocations still deliver
-/// their spans, metric points, and log records.
+/// Guard holding the tracer, meter, and logger providers for a process lifetime. Explicit
+/// shutdown delivers pending telemetry from short-lived CLI invocations; the process-global
+/// log bridge requires slightly different logger-provider lifetime handling (see [`Self::shutdown`]).
 pub struct Telemetry {
     tracer_provider: Option<SdkTracerProvider>,
     meter_provider: Option<SdkMeterProvider>,
@@ -133,8 +122,8 @@ impl Telemetry {
         // best-effort: if its exporter fails to build (e.g. malformed
         // `OTEL_EXPORTER_OTLP_METRICS_ENDPOINT`), metrics are disabled but span and log
         // export must continue. The periodic reader's default interval only governs
-        // background collection — `shutdown` below force-flushes, so short-lived CLI runs
-        // still deliver their points.
+        // background collection — provider shutdown below performs the final collection, so
+        // short-lived CLI runs still deliver their points.
         let meter_provider = match build_metric_exporter() {
             Ok(exporter) => {
                 let reader = opentelemetry_sdk::metrics::PeriodicReader::builder(exporter).build();
@@ -145,7 +134,6 @@ impl Telemetry {
                     .build();
                 opentelemetry::global::set_meter_provider(meter_provider.clone());
                 crate::metrics::set_enabled(true);
-                ENABLED.store(true, Ordering::Relaxed);
                 Some(meter_provider)
             }
             Err(err) => {
@@ -154,7 +142,6 @@ impl Telemetry {
             }
         };
 
-
         // Log records share endpoint, protocol, and resource too; the appender bridge maps
         // `tracing` events onto OTLP logs and stamps the current span context onto them.
         let logger_provider = match build_log_exporter() {
@@ -189,23 +176,28 @@ impl Telemetry {
         self.tracer_provider.is_some()
     }
 
-    /// Flush pending spans, metric points, and log records, then stop all exporters. Safe to
-    /// call multiple times.
+    /// Deliver pending spans, metric points, and log records, then stop bounded exporter
+    /// workers. Safe to call multiple times.
     pub fn shutdown(&mut self) {
+        // `PeriodicReader::shutdown` performs a final collect-and-export itself. Calling
+        // `force_flush` first would export the same cumulative counter and histogram snapshot
+        // twice for every short-lived process.
         if let Some(provider) = self.meter_provider.take() {
-            let _ = provider.force_flush();
             let _ = provider.shutdown();
             crate::metrics::set_enabled(false);
         }
-        if let Some(provider) = self.logger_provider.take() {
-            let _ = provider.force_flush();
+        if let Some(provider) = self.tracer_provider.take() {
             let _ = provider.shutdown();
         }
-        if let Some(provider) = self.tracer_provider.take() {
+        // The OpenTelemetry log bridge is installed in the process-global tracing subscriber
+        // and retains a logger from this provider. That subscriber cannot be uninstalled, so
+        // shutting the provider down here exposes a stopped BatchLogProcessor to subsequent
+        // events (including HTTP-client events produced by exporter shutdown). Force-flush last
+        // to deliver the correlated completion log, then leave the provider alive through the
+        // global bridge until process exit.
+        if let Some(provider) = self.logger_provider.take() {
             let _ = provider.force_flush();
-            let _ = provider.shutdown();
         }
-        ENABLED.store(false, Ordering::Relaxed);
     }
 }
 
@@ -215,49 +207,6 @@ impl Drop for Telemetry {
     }
 }
 
-/// A root `st2.reconcile_pass` span for one bounded pass, or nothing when telemetry is
-/// disabled (construction is skipped entirely — see [`enabled`]). Each pass gets its own
-/// trace; the supervisor loop never holds an endless root open.
-pub struct PassSpan(Option<opentelemetry::global::BoxedSpan>);
-
-impl PassSpan {
-    pub fn start(this_host: &str) -> Self {
-        if !enabled() {
-            return Self(None);
-        }
-        let tracer = opentelemetry::global::tracer("st2");
-        let span = tracer
-            .span_builder("st2.reconcile_pass")
-            .with_attributes(vec![KeyValue::new("st2.host", this_host.to_string())])
-            .start(&tracer);
-        Self(Some(span))
-    }
-
-    /// Record pass outcomes and end the span. Early-drop paths end it without attributes.
-    pub fn finish(mut self, crash_loops: usize, unparked: usize) {
-        if let Some(span) = self.0.as_mut() {
-            let to_i64 = |n: usize| i64::try_from(n).unwrap_or(i64::MAX);
-            span.set_attribute(KeyValue::new("st2.crash_loops", to_i64(crash_loops)));
-            span.set_attribute(KeyValue::new("st2.unparked", to_i64(unparked)));
-        }
-        self.end();
-    }
-}
-
-impl Drop for PassSpan {
-    fn drop(&mut self) {
-        self.end();
-    }
-}
-
-impl PassSpan {
-    fn end(&mut self) {
-        if let Some(mut span) = self.0.take() {
-            span.end();
-        }
-    }
-}
-
 /// Install the global subscriber once: stderr fmt (always), span exporter layer, and log-record
 /// bridge behind the endpoint guard. A second `Telemetry::init` in the same process cannot
 /// replace it (`set_global_default` errors), which is fine: init runs once per entrypoint.