saddle_runtime/

application.rs

use std::{
    future::Future,
    sync::Arc,
    sync::atomic::{AtomicBool, Ordering},
    time::{Duration, Instant},
};

use saddle_core::{ComponentLifecycle, ErrorKind, Result, SaddleError};
use saddle_observability::root_diagnostic::RecordedComponentLifecycle;

use crate::{RecordedComponentBoundary, RequestLifecycle};

enum ApplicationComponent {
    Legacy(Arc<dyn ComponentLifecycle>),
    Recorded(Arc<dyn RecordedComponentLifecycle>),
}
type PrevalidatedLegacyComponent = Arc<dyn ComponentLifecycle>;
impl ApplicationComponent {
    fn name(&self) -> &'static str {
        match self {
            Self::Legacy(component) => component.name(),
            Self::Recorded(component) => component.name(),
        }
    }
}

static RUNTIME_STARTED: AtomicBool = AtomicBool::new(false);
const WORKER_THREADS: usize = 2;
const MAX_IO_EVENTS_PER_TICK: usize = 5;
const DEFAULT_START_TIMEOUT: Duration = Duration::from_secs(30);
const DEFAULT_SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(30);

/// Fixed wall-clock limits for the managed component lifecycle.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct LifecycleTimeouts {
    start: Duration,
    shutdown: Duration,
}

impl LifecycleTimeouts {
    pub fn from_millis(start_ms: u64, shutdown_ms: u64) -> Option<Self> {
        if start_ms == 0 || shutdown_ms == 0 {
            return None;
        }
        Some(Self {
            start: Duration::from_millis(start_ms),
            shutdown: Duration::from_millis(shutdown_ms),
        })
    }
}

impl Default for LifecycleTimeouts {
    fn default() -> Self {
        Self {
            start: DEFAULT_START_TIMEOUT,
            shutdown: DEFAULT_SHUTDOWN_TIMEOUT,
        }
    }
}

/// A complete Saddle application hosted by the process-wide async runtime.
///
/// This is an assembly API, not a general-purpose async executor: it exposes no
/// Tokio handle, task spawning, runtime configuration, or arbitrary `block_on`.
pub struct Application {
    components: Vec<ApplicationComponent>,
    recorded_boundary: Option<RecordedComponentBoundary>,
    requests: RequestLifecycle,
    deployment_resource_budget: Option<saddle_admission::DeploymentResourceBudget>,
    ingress_bridge_issued: AtomicBool,
    lifecycle_timeouts: LifecycleTimeouts,
    shutdown_deadline: Arc<std::sync::Mutex<Option<Instant>>>,
    lifecycle_observer: Option<(saddle_observability::Observer, String)>,
    #[cfg(all(target_arch = "x86_64", target_os = "linux"))]
    pending_driver_finalizer: crate::post_driver::PendingDriverFinalizerSlot,
}

impl Application {
    pub(crate) fn shutdown_signal_source(&self) -> Option<(saddle_core::ContextLabel, saddle_observability::EmergencyDiagnosticHandle)> {
        self.requests.shutdown_signal_source()
    }
    /// Creates an empty application assembly.
    pub fn new() -> Self {
        Self {
            components: Vec::new(),
            recorded_boundary: None,
            requests: RequestLifecycle::new(),
            deployment_resource_budget: None,
            ingress_bridge_issued: AtomicBool::new(false),
            lifecycle_timeouts: LifecycleTimeouts::default(),
            shutdown_deadline: Arc::new(std::sync::Mutex::new(None)),
            lifecycle_observer: None,
            #[cfg(all(target_arch = "x86_64", target_os = "linux"))]
            pending_driver_finalizer: crate::post_driver::PendingDriverFinalizerSlot::new(),
        }
    }

    /// Installs the frozen process lifecycle policy before component startup.
    #[doc(hidden)]
    pub fn set_lifecycle_timeouts(&mut self, timeouts: LifecycleTimeouts) {
        self.lifecycle_timeouts = timeouts;
    }

    #[doc(hidden)]
    pub fn install_lifecycle_observer(
        &mut self,
        observer: saddle_observability::Observer,
        application: &str,
    ) {
        self.lifecycle_observer = Some((observer, application.to_owned()));
    }

    /// Installs the same application identity with the selected source output
    /// before any internal request admission owner can run.
    #[doc(hidden)]
    pub fn install_lifecycle_observer_with_output(
        &mut self,
        observer: saddle_observability::Observer,
        application: &str,
        output: Option<saddle_observability::EmergencyDiagnosticHandle>,
    ) {
        assert!(self.requests.install_admission_source(application.into(), output),
            "one Application installs one admission source binding");
        self.lifecycle_observer = Some((observer, application.to_owned()));
    }

    /// Formal process assembly accepts only the output selected and opened
    /// during checked startup. The legacy optional installation stays outside
    /// this capability boundary.
    #[doc(hidden)]
    pub fn install_lifecycle_observer_with_source_output(
        &mut self,
        observer: saddle_observability::Observer,
        application: &str,
        output: saddle_observability::SourceOutput,
    ) {
        assert!(self.components.is_empty(),
            "formal source output must precede component registration");
        let label = saddle_core::ContextLabel::checked(application)
            .expect("formal Application requires checked identity");
        assert!(self.requests.install_admission_source(
            application.into(), Some(output.handle().clone())),
            "one Application installs one admission source binding");
        self.recorded_boundary = Some(RecordedComponentBoundary::new(label, output));
        self.lifecycle_observer = Some((observer, application.to_owned()));
    }

    #[cfg(all(target_arch = "x86_64", target_os = "linux"))]
    pub(crate) fn install_prevalidated_components(
        &mut self,
        components: Vec<PrevalidatedLegacyComponent>,
    ) {
        debug_assert!(self.components.is_empty());
        assert!(self.recorded_boundary.is_none(),
            "prevalidated legacy components cannot replace formal recorded components");
        self.components = components.into_iter().map(ApplicationComponent::Legacy).collect();
    }

    #[cfg(all(target_arch = "x86_64", target_os = "linux"))]
    #[doc(hidden)]
    pub fn pending_driver_finalizer(&self) -> crate::post_driver::PendingDriverFinalizerSlot {
        self.pending_driver_finalizer.clone()
    }

    #[cfg(all(test, target_arch = "x86_64", target_os = "linux"))]
    pub(crate) fn post_driver_is_unarmed_for_test(&self) -> bool {
        self.pending_driver_finalizer.is_unarmed_for_test()
    }

    #[cfg(all(target_arch = "x86_64", target_os = "linux"))]
    #[doc(hidden)]
    #[allow(clippy::result_large_err)]
    pub fn commit_post_driver_install(
        &self,
        binding: saddle_admission::VerifiedPostDriverInstallBinding,
    ) {
        self.pending_driver_finalizer
            .commit_verified_install(binding)
    }

    pub(crate) fn reserved_post_driver_submit(
        &self,
    ) -> crate::post_driver::MustSubmitDriverFinalizer {
        self.pending_driver_finalizer.reserved_submit_handle()
    }

    /// Returns the request lifecycle shared with Saddle's Service adapter.
    pub fn request_lifecycle(&self) -> RequestLifecycle {
        self.requests.clone()
    }

    /// Returns a read-only observer of the framework's unique lifecycle
    /// state. The observer cannot admit requests or mutate readiness.
    pub fn health(&self) -> crate::ApplicationHealth {
        self.requests.health()
    }

    /// Reserves the fixed alpha.1 Ingress execution bridge attached to this
    /// application's existing 0.2 request lifecycle.
    #[doc(hidden)]
    pub fn managed_ingress_bridge(
        &self,
        capacity: usize,
    ) -> Option<crate::alpha1_ingress::ManagedIngressBridge> {
        if capacity == 0 {
            return None;
        }
        if self
            .ingress_bridge_issued
            .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
            .is_err()
        {
            return None;
        }
        crate::alpha1_ingress::ManagedIngressBridge::new(self.requests.clone(), capacity)
    }

    /// Registers a framework component for managed startup and shutdown.
    ///
    /// Components start in registration order and stop in reverse order.
    pub fn register<C>(&mut self, component: C) -> Result<()>
    where
        C: ComponentLifecycle + 'static,
    {
        self.register_shared(Arc::new(component))
    }

    /// Registers an already shared framework component.
    pub fn register_shared(&mut self, component: Arc<dyn ComponentLifecycle>) -> Result<()> {
        if self.recorded_boundary.is_some() {
            return Err(SaddleError::new(ErrorKind::Infrastructure,
                "runtime.recorded_component_required",
                "formal Application requires a recorded component"));
        }
        if self
            .components
            .iter()
            .any(|registered| registered.name() == component.name())
        {
            return Err(SaddleError::new(
                ErrorKind::Conflict,
                "runtime.duplicate_component",
                format!("component '{}' is already registered", component.name()),
            ));
        }
        self.components.push(ApplicationComponent::Legacy(component));
        Ok(())
    }

    /// Formal registration requires a selected source output before startup.
    pub fn register_recorded<C>(&mut self, component: C) -> Result<()>
    where C: RecordedComponentLifecycle + 'static {
        self.register_recorded_shared(Arc::new(component))
    }

    /// Shares a recorded component with a lifecycle fault owner without
    /// reopening the legacy registration path.
    pub fn register_recorded_shared(
        &mut self, component: Arc<dyn RecordedComponentLifecycle>,
    ) -> Result<()> {
        if self.recorded_boundary.is_none() {
            return Err(SaddleError::new(ErrorKind::Infrastructure,
                "runtime.source_output_required",
                "formal component output has not been selected").with_unconfirmed_source());
        }
        if self.components.iter().any(|registered| registered.name() == component.name()) {
            return Err(SaddleError::new(ErrorKind::Conflict,
                "runtime.duplicate_component",
                format!("component '{}' is already registered", component.name())));
        }
        self.components.push(ApplicationComponent::Recorded(component));
        Ok(())
    }

    /// Runs the application on Saddle's single process-wide async runtime.
    ///
    /// The call blocks the process entry thread until SIGINT or, on Unix,
    /// SIGTERM. Shutdown first closes request admission, then waits for every
    /// admitted request, and finally stops components in reverse order.
    pub fn run(self) -> Result<()> {
        Self::run_with(|| async move { Ok(self) })
    }

    /// Creates the application inside Saddle's process-wide async runtime and
    /// then runs it until shutdown.
    ///
    /// This is the framework assembly path for components whose initialization
    /// performs async I/O. Business code is not given a runtime handle or an
    /// executor through this API.
    pub fn run_with<F, Fut>(bootstrap: F) -> Result<()>
    where
        F: FnOnce() -> Fut + Send + 'static,
        Fut: Future<Output = Result<Self>> + Send + 'static,
    {
        if RUNTIME_STARTED
            .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
            .is_err()
        {
            return Err(SaddleError::new(
                ErrorKind::Conflict,
                "runtime.already_started",
                "the Saddle runtime has already started in this process",
            ));
        }

        let runtime = build_runtime()?;

        #[cfg(all(target_arch = "x86_64", target_os = "linux"))]
        {
            Self::run_with_owned_runtime(runtime, bootstrap)
        }

        #[cfg(not(all(target_arch = "x86_64", target_os = "linux")))]
        runtime.block_on(async {
            let signal = ShutdownSignal::register()?;
            bootstrap_and_run(bootstrap, signal.wait()).await
        })
    }

    /// Runs the formal process while retaining its one frozen deployment
    /// budget inside Runtime assembly. No read or replacement surface escapes.
    #[doc(hidden)]
    pub fn run_with_deployment_resource_budget<F, Fut>(
        budget: saddle_admission::DeploymentResourceBudget,
        bootstrap: F,
    ) -> Result<()>
    where
        F: FnOnce() -> Fut + Send + 'static,
        Fut: Future<Output = Result<Self>> + Send + 'static,
    {
        Self::run_with(move || async move {
            let mut application = bootstrap().await?;
            if application.deployment_resource_budget.is_some() {
                return Err(SaddleError::new(
                    ErrorKind::Conflict,
                    "runtime.deployment_resource_budget_already_installed",
                    "the deployment resource budget was already installed",
                ));
            }
            application.deployment_resource_budget = Some(budget);
            Ok(application)
        })
    }

    #[cfg(all(target_arch = "x86_64", target_os = "linux"))]
    pub(crate) fn claim_process_runtime() -> Result<()> {
        if RUNTIME_STARTED
            .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
            .is_err()
        {
            return Err(SaddleError::new(
                ErrorKind::Conflict,
                "runtime.already_started",
                "the Saddle runtime has already started in this process",
            ));
        }
        Ok(())
    }

    #[cfg(all(target_arch = "x86_64", target_os = "linux"))]
    pub(crate) fn run_with_owned_runtime<F, Fut>(
        runtime: tokio::runtime::Runtime,
        bootstrap: F,
    ) -> Result<()>
    where
        F: FnOnce() -> Fut,
        Fut: Future<Output = Result<Self>>,
    {
        let outcome = runtime.block_on(async {
            let signal = ShutdownSignal::register()?;
            let application = bootstrap().await?;
            let finalizer = application.pending_driver_finalizer();
            let shutdown_deadline = Arc::clone(&application.shutdown_deadline);
            let lifecycle_observer = application.lifecycle_observer_handle();
            let source = application.shutdown_signal_source();
            let result = application.run_until_shutdown(signal.wait_observed(source)).await;
            Ok::<_, SaddleError>((finalizer, shutdown_deadline, lifecycle_observer, result))
        });
        match outcome {
            Ok((finalizer, shutdown_deadline, lifecycle_observer, result)) => {
                let deadline = *shutdown_deadline
                    .lock()
                    .unwrap_or_else(|poisoned| poisoned.into_inner());
                finalizer.finish(runtime, result, deadline, lifecycle_observer)
            }
            Err(error) => {
                drop(runtime);
                Err(error)
            }
        }
    }

    pub(crate) async fn run_until_shutdown<F>(self, shutdown: F) -> Result<()>
    where
        F: Future<Output = Result<()>>,
    {
        tokio::pin!(shutdown);
        let signal_before_start = tokio::select! {
            biased;
            signal_result = &mut shutdown => Some(signal_result),
            _ = std::future::ready(()) => None,
        };
        if let Some(signal_result) = signal_before_start {
            self.requests.begin_draining();
            self.requests.wait_until_drained().await;
            self.requests.mark_stopped();
            return signal_result.map_err(|error| crate::diagnostics::shutdown_signal_source(
                error, self.lifecycle_observer.as_ref().map(|(_, application)| application.as_str()),
            ));
        }

        let mut started = 0;

        for component in &self.components {
            let start_started = Instant::now();
            let start = crate::diagnostics::task(
                async {
                    match component {
                        ApplicationComponent::Legacy(legacy) => legacy.start().await,
                        ApplicationComponent::Recorded(component) => self.recorded_boundary
                            .as_ref().expect("recorded component requires selected output")
                            .start(component.as_ref()).await.map_err(|failure| failure.into_safe()),
                    }
                },
                saddle_core::DiagnosticStage::StartupListener,
                "runtime.component_start",
            );
            tokio::pin!(start);
            let mut shutdown_during_start = None;
            let start_result = tokio::select! {
                biased;
                signal_result = &mut shutdown => {
                    let deadline = Instant::now() + self.lifecycle_timeouts.shutdown;
                    self.set_shutdown_deadline(deadline);
                    shutdown_during_start = Some((signal_result, deadline));
                    // A component may have partially initialized before its
                    // start future yielded. Bound that in-progress start by
                    // both lifecycle budgets, then include it in rollback.
                    let start_deadline = std::cmp::min(
                        start_started + self.lifecycle_timeouts.start,
                        deadline,
                    );
                    match tokio::time::timeout_at(start_deadline.into(), start).await {
                        Ok(result) => result,
                        Err(elapsed) => Err(match component {
                            ApplicationComponent::Legacy(_) => lifecycle_timeout_error("component_start"),
                            ApplicationComponent::Recorded(_) => self.recorded_boundary.as_ref()
                                .expect("selected output").timeout_failure(elapsed,
                                    saddle_observability::root_diagnostic::ComponentSourceKind::Start,
                                    None).into_safe(),
                        }),
                    }
                }
                start_result = tokio::time::timeout(self.lifecycle_timeouts.start, &mut start) => {
                    start_result.unwrap_or_else(|elapsed| Err(match component {
                        ApplicationComponent::Legacy(_) => lifecycle_timeout_error("component_start"),
                        ApplicationComponent::Recorded(_) => self.recorded_boundary.as_ref()
                            .expect("selected output").timeout_failure(elapsed,
                                saddle_observability::root_diagnostic::ComponentSourceKind::Start,
                                None).into_safe(),
                    }))
                },
            };

            if let Err(error) = start_result {
                // Fix cleanup's budget BEFORE any diagnostic capture/submit.
                let deadline = shutdown_during_start
                    .as_ref()
                    .map(|(_, deadline)| *deadline)
                    .unwrap_or_else(|| Instant::now() + self.lifecycle_timeouts.shutdown);
                self.set_shutdown_deadline(deadline);
                let error = if matches!(component, ApplicationComponent::Recorded(_)) {
                    error
                } else {
                    crate::diagnostics::component_start_source(error,
                        self.lifecycle_observer.as_ref().map(|(_, application)| application.as_str()))
                };
                self.record_timeout(&error, start_started.elapsed());
                self.requests.begin_draining();
                let cleanup_count = if error.code() == "runtime.lifecycle_timeout.component_start" {
                    started + 1
                } else {
                    started
                };
                let drain_result = timeout_at(
                    deadline,
                    self.requests.wait_until_drained(),
                    "request_drain",
                )
                .await;
                let cleanup_source_unavailable = match drain_result {
                    Ok(()) => self
                        .shutdown_components(cleanup_count, deadline, Some(&error))
                        .await
                        .err()
                        .is_some_and(|cleanup| cleanup.cleanup_source_unavailable()),
                    Err(cleanup) => {
                        let cleanup = crate::diagnostics::request_drain_source(
                            cleanup,
                            Some(&error),
                            self.lifecycle_observer.as_ref().map(|(_, application)| application.as_str()),
                        );
                        cleanup.cleanup_source_unavailable()
                    }
                };
                self.requests.mark_stopped();
                return Err(if cleanup_source_unavailable {
                    error.with_unconfirmed_cleanup_source()
                } else { error });
            }
            started += 1;

            if let Some((signal_result, deadline)) = shutdown_during_start {
                self.requests.begin_draining();
                let drain_result = timeout_at(
                    deadline,
                    self.requests.wait_until_drained(),
                    "request_drain",
                )
                .await;
                let signal_result = signal_result.map_err(|e| {
                    crate::diagnostics::shutdown_signal_source(
                        e, self.lifecycle_observer.as_ref().map(|(_, application)| application.as_str()),
                    )
                });
                let shutdown_result = if drain_result.is_ok() {
                    self.shutdown_components(started, deadline, signal_result.as_ref().err())
                        .await
                } else {
                    drain_result.map_err(|e| {
                        crate::diagnostics::request_drain_source(
                            e,
                            signal_result.as_ref().err(),
                            self.lifecycle_observer.as_ref().map(|(_, application)| application.as_str()),
                        )
                    })
                };
                self.requests.mark_stopped();
                return combine_lifecycle_results(signal_result, shutdown_result);
            }
        }

        self.requests.mark_ready();
        let signal_result = shutdown.await;
        let deadline = Instant::now() + self.lifecycle_timeouts.shutdown;
        self.set_shutdown_deadline(deadline);
        self.requests.observe_shutdown_decision(signal_result.is_ok());
        self.requests.begin_draining();
        let signal_result = signal_result.map_err(|e| {
            crate::diagnostics::shutdown_signal_source(
                e, self.lifecycle_observer.as_ref().map(|(_, application)| application.as_str()),
            )
        });
        let drain_result = timeout_at(
            deadline,
            self.requests.wait_until_drained(),
            "request_drain",
        )
        .await;
        if let Err(error) = &drain_result {
            self.record_timeout(error, self.lifecycle_timeouts.shutdown);
        }
        let shutdown_result = if drain_result.is_ok() {
            self.shutdown_components(started, deadline, signal_result.as_ref().err())
                .await
        } else {
            drain_result.map_err(|e| {
                crate::diagnostics::request_drain_source(
                    e,
                    signal_result.as_ref().err(),
                    self.lifecycle_observer.as_ref().map(|(_, application)| application.as_str()),
                )
            })
        };
        self.requests.mark_stopped();

        combine_lifecycle_results(signal_result, shutdown_result)
    }

    fn set_shutdown_deadline(&self, deadline: Instant) {
        *self
            .shutdown_deadline
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner()) = Some(deadline);
    }

    pub(crate) fn shutdown_deadline_handle(&self) -> Arc<std::sync::Mutex<Option<Instant>>> {
        Arc::clone(&self.shutdown_deadline)
    }

    pub(crate) fn lifecycle_observer_handle(
        &self,
    ) -> Option<(saddle_observability::Observer, String)> {
        self.lifecycle_observer.clone()
    }

    fn record_timeout(&self, error: &SaddleError, elapsed: Duration) {
        let stage = match error.code() {
            "runtime.lifecycle_timeout.component_start" => {
                saddle_observability::LifecycleTimeoutStage::ComponentStart
            }
            "runtime.lifecycle_timeout.request_drain" => {
                saddle_observability::LifecycleTimeoutStage::RequestDrain
            }
            "runtime.lifecycle_timeout.component_shutdown" => {
                saddle_observability::LifecycleTimeoutStage::ComponentShutdown
            }
            _ => return,
        };
        if let Some((observer, application)) = &self.lifecycle_observer {
            observer.record_lifecycle_timeout(
                application.as_str(),
                stage,
                u64::try_from(elapsed.as_millis()).unwrap_or(u64::MAX),
            );
        }
    }

    async fn shutdown_components(
        &self,
        started: usize,
        deadline: Instant,
        primary: Option<&SaddleError>,
    ) -> Result<()> {
        let mut first_error = None;
        let mut cleanup_source_unavailable = false;
        for component in self.components[..started].iter().rev() {
            let result = match tokio::time::timeout_at(
                deadline.into(),
                crate::diagnostics::task(
                    async {
                        let parent = primary.or(first_error.as_ref())
                            .and_then(|error| error.diagnostic().map(|d| d.occurrence()));
                        match component {
                            ApplicationComponent::Legacy(legacy) =>
                                legacy.shutdown_with_primary(parent).await,
                            ApplicationComponent::Recorded(component) => self.recorded_boundary
                                .as_ref().expect("recorded component requires selected output")
                                .shutdown(component.as_ref(), parent).await
                                .map_err(|failure| {
                                    let safe = failure.into_safe();
                                    if safe.source_unavailable() {
                                        safe.with_unconfirmed_cleanup_source()
                                    } else { safe }
                                }),
                        }
                    },
                    saddle_core::DiagnosticStage::ShutdownComponent,
                    "runtime.component_shutdown",
                ),
            )
            .await
            {
                Ok(result) => result,
                Err(elapsed) => Err(match component {
                    ApplicationComponent::Legacy(_) => lifecycle_timeout_error("component_shutdown"),
                    ApplicationComponent::Recorded(_) => {
                        let safe = self.recorded_boundary.as_ref()
                            .expect("selected output").timeout_failure(elapsed,
                                saddle_observability::root_diagnostic::ComponentSourceKind::Cleanup,
                                primary.or(first_error.as_ref())
                                    .and_then(|error| error.diagnostic().map(|d| d.occurrence())))
                            .into_safe();
                        if safe.source_unavailable() { safe.with_unconfirmed_cleanup_source() }
                        else { safe }
                    }
                }),
            };
            if let Err(error) = result {
                let error = if matches!(component, ApplicationComponent::Recorded(_)) {
                    error
                } else {
                    crate::diagnostics::component_cleanup_source(error,
                        primary.or(first_error.as_ref()),
                        self.lifecycle_observer.as_ref().map(|(_, application)| application.as_str()))
                };
                cleanup_source_unavailable |= error.cleanup_source_unavailable();
                self.record_timeout(&error, self.lifecycle_timeouts.shutdown);
                if first_error.is_none() {
                    first_error = Some(error);
                }
                if Instant::now() >= deadline {
                    break;
                }
            }
        }
        first_error.map_or(Ok(()), |error| Err(if cleanup_source_unavailable {
            error.with_unconfirmed_cleanup_source()
        } else { error }))
    }
}

pub(crate) fn combine_lifecycle_results(primary: Result<()>, cleanup: Result<()>) -> Result<()> {
    match (primary, cleanup) {
        (Err(primary), Err(cleanup)) if cleanup.cleanup_source_unavailable() => {
            Err(primary.with_unconfirmed_cleanup_source())
        }
        (Err(primary), _) => Err(primary),
        (Ok(()), result) => result,
    }
}

async fn timeout_at<T>(
    deadline: Instant,
    future: impl Future<Output = T>,
    stage: &'static str,
) -> Result<T> {
    let remaining = deadline.saturating_duration_since(Instant::now());
    tokio::time::timeout(remaining, future)
        .await
        .map_err(|_| lifecycle_timeout_error(stage))
}

fn lifecycle_timeout_error(stage: &'static str) -> SaddleError {
    SaddleError::new(
        ErrorKind::Infrastructure,
        match stage {
            "component_start" => "runtime.lifecycle_timeout.component_start",
            "request_drain" => "runtime.lifecycle_timeout.request_drain",
            "component_shutdown" => "runtime.lifecycle_timeout.component_shutdown",
            _ => "runtime.lifecycle_timeout.post_driver",
        },
        format!("managed lifecycle stage '{stage}' exceeded its wall-clock deadline"),
    )
}

#[cfg(any(test, not(all(target_arch = "x86_64", target_os = "linux"))))]
async fn bootstrap_and_run<F, Fut, S>(bootstrap: F, shutdown: S) -> Result<()>
where
    F: FnOnce() -> Fut,
    Fut: Future<Output = Result<Application>>,
    S: Future<Output = Result<()>>,
{
    let application = bootstrap().await?;
    application.run_until_shutdown(shutdown).await
}

impl Default for Application {
    fn default() -> Self {
        Self::new()
    }
}

fn build_runtime() -> Result<tokio::runtime::Runtime> {
    tokio::runtime::Builder::new_multi_thread()
        .worker_threads(WORKER_THREADS)
        .max_io_events_per_tick(MAX_IO_EVENTS_PER_TICK)
        .enable_all()
        .build()
        .map_err(|_| {
            SaddleError::new(
                ErrorKind::Infrastructure,
                "runtime.initialization_failed",
                "failed to initialize the Saddle async runtime",
            )
        })
}

#[cfg(all(target_arch = "x86_64", target_os = "linux"))]
pub(crate) fn claim_owned_runtime() -> Result<tokio::runtime::Runtime> {
    Application::claim_process_runtime()?;
    build_runtime()
}

#[cfg(unix)]
pub(crate) struct ShutdownSignal {
    interrupt: tokio::signal::unix::Signal,
    terminate: tokio::signal::unix::Signal,
}

#[cfg(unix)]
impl ShutdownSignal {
    /// Registers both listeners synchronously before any component starts.
    pub(crate) fn register() -> Result<Self> {
        Ok(Self {
            interrupt: tokio::signal::unix::signal(tokio::signal::unix::SignalKind::interrupt())
                .map_err(|_| signal_error())?,
            terminate: tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())
                .map_err(|_| signal_error())?,
        })
    }

    pub(crate) async fn wait(mut self) -> Result<()> {
        tokio::select! {
            _ = self.interrupt.recv() => Ok(()),
            _ = self.terminate.recv() => Ok(()),
        }
    }

    pub(crate) async fn wait_observed(mut self, source: Option<(saddle_core::ContextLabel, saddle_observability::EmergencyDiagnosticHandle)>) -> Result<()> {
        let (kind, received) = tokio::select! {
            value = self.interrupt.recv() => ("SIGINT", value.is_some()),
            value = self.terminate.recv() => ("SIGTERM", value.is_some()),
        };
        if let Some((application, output)) = source {
            output.record_shutdown_signal_received(application, kind, received);
        }
        Ok(())
    }
}

#[cfg(windows)]
impl ShutdownSignal {
    pub(crate) async fn wait_observed(self, _source: Option<(saddle_core::ContextLabel, saddle_observability::EmergencyDiagnosticHandle)>) -> Result<()> {
        // Unix signal-kind receipts are unavailable on the Windows adapter.
        self.wait().await
    }
}

#[cfg(windows)]
struct ShutdownSignal {
    ctrl_c: tokio::signal::windows::CtrlC,
    ctrl_break: tokio::signal::windows::CtrlBreak,
}

#[cfg(windows)]
impl ShutdownSignal {
    /// Registers both listeners synchronously before any component starts.
    fn register() -> Result<Self> {
        Ok(Self {
            ctrl_c: tokio::signal::windows::ctrl_c().map_err(|_| signal_error())?,
            ctrl_break: tokio::signal::windows::ctrl_break().map_err(|_| signal_error())?,
        })
    }

    async fn wait(mut self) -> Result<()> {
        tokio::select! {
            _ = self.ctrl_c.recv() => Ok(()),
            _ = self.ctrl_break.recv() => Ok(()),
        }
    }
}

fn signal_error() -> SaddleError {
    SaddleError::new(
        ErrorKind::Infrastructure,
        "runtime.signal_registration_failed",
        "failed to register the application shutdown signal",
    )
}

#[cfg(test)]
mod tests {
    use std::sync::Mutex;

    use saddle_core::LifecycleFuture;

    use super::*;
    use crate::ApplicationPhase;

    struct RecordedStartFailure;
    struct RecordedPendingStart;
    struct RecordedPendingCleanup;
    impl RecordedComponentLifecycle for RecordedPendingCleanup {
        fn name(&self) -> &'static str { "recorded-pending-cleanup" }
        fn start<'a>(&'a self, _: &'a saddle_core::ContextLabel,
            _: &'a saddle_observability::SourceOutput)
            -> saddle_observability::root_diagnostic::RecordedLifecycleFuture<'a> {
            Box::pin(async { Ok(()) })
        }
        fn shutdown<'a>(&'a self, _: &'a saddle_core::ContextLabel,
            _: &'a saddle_observability::SourceOutput)
            -> saddle_observability::root_diagnostic::RecordedLifecycleFuture<'a> {
            Box::pin(std::future::pending())
        }
        fn shutdown_with_primary<'a>(&'a self, _: &'a saddle_core::ContextLabel,
            _: &'a saddle_observability::SourceOutput,
            _: Option<saddle_core::DiagnosticOccurrence>)
            -> saddle_observability::root_diagnostic::RecordedLifecycleFuture<'a> {
            Box::pin(std::future::pending())
        }
    }
    impl RecordedComponentLifecycle for RecordedPendingStart {
        fn name(&self) -> &'static str { "recorded-pending-start" }
        fn start<'a>(&'a self, _: &'a saddle_core::ContextLabel,
            _: &'a saddle_observability::SourceOutput)
            -> saddle_observability::root_diagnostic::RecordedLifecycleFuture<'a> {
            Box::pin(std::future::pending())
        }
        fn shutdown<'a>(&'a self, _: &'a saddle_core::ContextLabel,
            _: &'a saddle_observability::SourceOutput)
            -> saddle_observability::root_diagnostic::RecordedLifecycleFuture<'a> {
            Box::pin(async { Ok(()) })
        }
        fn shutdown_with_primary<'a>(&'a self, _: &'a saddle_core::ContextLabel,
            _: &'a saddle_observability::SourceOutput,
            _: Option<saddle_core::DiagnosticOccurrence>)
            -> saddle_observability::root_diagnostic::RecordedLifecycleFuture<'a> {
            Box::pin(async { Ok(()) })
        }
    }
    impl RecordedComponentLifecycle for RecordedStartFailure {
        fn name(&self) -> &'static str { "recorded-start-failure" }
        fn start<'a>(&'a self, application: &'a saddle_core::ContextLabel,
            output: &'a saddle_observability::SourceOutput)
            -> saddle_observability::root_diagnostic::RecordedLifecycleFuture<'a> {
            Box::pin(async move {
                saddle_observability::root_diagnostic::RecordedSaddleError::component_result::<(), _>(
                    Err(std::io::Error::other("M_ACTUAL_START_ORIGINAL")), application, output,
                    saddle_observability::root_diagnostic::ComponentSourceKind::Start, None,
                    ErrorKind::Infrastructure, "runtime.recorded_start_test", "safe start failure")
            })
        }
        fn shutdown<'a>(&'a self, _: &'a saddle_core::ContextLabel,
            _: &'a saddle_observability::SourceOutput)
            -> saddle_observability::root_diagnostic::RecordedLifecycleFuture<'a> {
            Box::pin(async { Ok(()) })
        }
        fn shutdown_with_primary<'a>(&'a self, _: &'a saddle_core::ContextLabel,
            _: &'a saddle_observability::SourceOutput,
            _: Option<saddle_core::DiagnosticOccurrence>)
            -> saddle_observability::root_diagnostic::RecordedLifecycleFuture<'a> {
            Box::pin(async { Ok(()) })
        }
    }

    #[test]
    fn formal_application_requires_output_and_reads_original_before_return() {
        let mut absent = Application::new();
        assert_eq!(absent.register_recorded(RecordedStartFailure).unwrap_err().code(),
            "runtime.source_output_required");
        let root = std::env::temp_dir().join(format!("saddle-m-recorded-{}-{}",
            std::process::id(), std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH).unwrap().as_nanos()));
        std::fs::create_dir(&root).unwrap();
        let mut output = saddle_observability::EmergencyDiagnostics::start_checked(
            &saddle_observability::FileLoggingConfig::new(root.clone(),
                saddle_observability::Rotation::Daily)).unwrap();
        let target = output.target().to_owned();
        let observer = saddle_observability::Observer::with_writer(
            Default::default(), std::io::sink()).unwrap();
        let mut app = Application::new();
        app.install_lifecycle_observer_with_source_output(observer, "m-formal-app",
            output.source_output().unwrap());
        assert_eq!(app.register(RecordingComponent {
            name: "legacy", events: Arc::new(Mutex::new(Vec::new())),
            start_error: false, shutdown_error: false,
        }).unwrap_err().code(), "runtime.recorded_component_required");
        app.register_recorded(RecordedStartFailure).unwrap();
        let failed = test_runtime().block_on(app.run_until_shutdown(
            std::future::pending::<Result<()>>())).unwrap_err();
        assert_eq!(failed.code(), "runtime.recorded_start_test");
        let id = failed.diagnostic().unwrap().id().to_string();
        let raw = std::fs::read_to_string(&target).unwrap();
        assert!(raw.contains("M_ACTUAL_START_ORIGINAL"));
        assert!(raw.contains(&id));
        assert!(!failed.source_unavailable());
        while output.shutdown() == saddle_observability::DiagnosticShutdown::Pending {
            std::thread::yield_now();
        }
        drop(output);
        std::fs::remove_file(target).unwrap();
        std::fs::remove_dir(root).unwrap();
    }

    #[test]
    fn formal_application_timeout_records_actual_elapsed_before_return() {
        let root = std::env::temp_dir().join(format!("saddle-m-timeout-{}-{}",
            std::process::id(), std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH).unwrap().as_nanos()));
        std::fs::create_dir(&root).unwrap();
        let mut output = saddle_observability::EmergencyDiagnostics::start_checked(
            &saddle_observability::FileLoggingConfig::new(root.clone(),
                saddle_observability::Rotation::Daily)).unwrap();
        let target = output.target().to_owned();
        let observer = saddle_observability::Observer::with_writer(
            Default::default(), std::io::sink()).unwrap();
        let mut app = Application::new();
        app.set_lifecycle_timeouts(LifecycleTimeouts::from_millis(20, 100).unwrap());
        app.install_lifecycle_observer_with_source_output(observer, "m-timeout-app",
            output.source_output().unwrap());
        app.register_recorded(RecordedPendingStart).unwrap();
        let failed = test_runtime().block_on(app.run_until_shutdown(
            std::future::pending::<Result<()>>())).unwrap_err();
        assert_eq!(failed.code(), "runtime.lifecycle_timeout.component_start");
        assert!(!failed.source_unavailable());
        let raw = std::fs::read_to_string(&target).unwrap();
        assert!(raw.contains("deadline has elapsed"));
        assert!(raw.contains(&failed.diagnostic().unwrap().id().to_string()));
        while output.shutdown() == saddle_observability::DiagnosticShutdown::Pending {
            std::thread::yield_now();
        }
        drop(output);
        std::fs::remove_file(target).unwrap();
        std::fs::remove_dir(root).unwrap();
    }

    #[test]
    fn formal_application_cleanup_timeout_records_actual_elapsed() {
        let root = std::env::temp_dir().join(format!("saddle-m-cleanup-{}-{}",
            std::process::id(), std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH).unwrap().as_nanos()));
        std::fs::create_dir(&root).unwrap();
        let mut output = saddle_observability::EmergencyDiagnostics::start_checked(
            &saddle_observability::FileLoggingConfig::new(root.clone(),
                saddle_observability::Rotation::Daily)).unwrap();
        let target = output.target().to_owned();
        let observer = saddle_observability::Observer::with_writer(
            Default::default(), std::io::sink()).unwrap();
        let mut app = Application::new();
        app.set_lifecycle_timeouts(LifecycleTimeouts::from_millis(100, 20).unwrap());
        app.install_lifecycle_observer_with_source_output(observer, "m-cleanup-app",
            output.source_output().unwrap());
        app.register_recorded(RecordedPendingCleanup).unwrap();
        let shutdown = shutdown_when_ready(app.request_lifecycle());
        let failed = test_runtime().block_on(app.run_until_shutdown(shutdown)).unwrap_err();
        assert_eq!(failed.code(), "runtime.lifecycle_timeout.component_shutdown");
        assert!(!failed.cleanup_source_unavailable());
        let raw = std::fs::read_to_string(&target).unwrap();
        assert!(raw.contains("deadline has elapsed"));
        assert!(raw.contains(&failed.diagnostic().unwrap().id().to_string()));
        while output.shutdown() == saddle_observability::DiagnosticShutdown::Pending {
            std::thread::yield_now();
        }
        drop(output);
        std::fs::remove_file(target).unwrap();
        std::fs::remove_dir(root).unwrap();
    }

    struct RecordingComponent {
        name: &'static str,
        events: Arc<Mutex<Vec<String>>>,
        start_error: bool,
        shutdown_error: bool,
    }

    struct BlockingStartComponent {
        events: Arc<Mutex<Vec<String>>>,
        started: Mutex<Option<tokio::sync::oneshot::Sender<()>>>,
        release: Mutex<Option<tokio::sync::oneshot::Receiver<()>>>,
    }

    struct BlockingShutdownComponent {
        events: Arc<Mutex<Vec<String>>>,
    }

    struct HealthAwareListener {
        health: crate::ApplicationHealth,
        events: Arc<Mutex<Vec<String>>>,
    }

    impl ComponentLifecycle for RecordingComponent {
        fn name(&self) -> &'static str {
            self.name
        }

        fn start(&self) -> LifecycleFuture<'_> {
            Box::pin(async move {
                self.events
                    .lock()
                    .unwrap()
                    .push(format!("start:{}", self.name));
                if self.start_error {
                    Err(test_error("start failed"))
                } else {
                    Ok(())
                }
            })
        }

        fn shutdown(&self) -> LifecycleFuture<'_> {
            Box::pin(async move {
                self.events
                    .lock()
                    .unwrap()
                    .push(format!("shutdown:{}", self.name));
                if self.shutdown_error {
                    Err(test_error("shutdown failed"))
                } else {
                    Ok(())
                }
            })
        }
    }

    impl ComponentLifecycle for BlockingStartComponent {
        fn name(&self) -> &'static str {
            "blocking"
        }

        fn start(&self) -> LifecycleFuture<'_> {
            Box::pin(async move {
                self.events
                    .lock()
                    .unwrap()
                    .push("start:blocking".to_owned());
                let started = self.started.lock().unwrap().take().unwrap();
                let release = self.release.lock().unwrap().take().unwrap();
                started.send(()).unwrap();
                release.await.unwrap();
                Ok(())
            })
        }

        fn shutdown(&self) -> LifecycleFuture<'_> {
            Box::pin(async move {
                self.events
                    .lock()
                    .unwrap()
                    .push("shutdown:blocking".to_owned());
                Ok(())
            })
        }
    }

    impl ComponentLifecycle for BlockingShutdownComponent {
        fn name(&self) -> &'static str {
            "blocking-shutdown"
        }

        fn start(&self) -> LifecycleFuture<'_> {
            Box::pin(async move {
                self.events
                    .lock()
                    .unwrap()
                    .push("start:blocking-shutdown".into());
                Ok(())
            })
        }

        fn shutdown(&self) -> LifecycleFuture<'_> {
            Box::pin(async move {
                self.events
                    .lock()
                    .unwrap()
                    .push("shutdown:blocking-shutdown".into());
                std::future::pending().await
            })
        }
    }

    impl ComponentLifecycle for HealthAwareListener {
        fn name(&self) -> &'static str {
            "health-aware-listener"
        }

        fn start(&self) -> LifecycleFuture<'_> {
            Box::pin(async move {
                let snapshot = self.health.snapshot();
                assert!(snapshot.is_live());
                assert!(!snapshot.is_ready());
                assert_eq!(snapshot.phase(), ApplicationPhase::Starting);
                self.events
                    .lock()
                    .unwrap()
                    .push("listener:accepting".into());
                Ok(())
            })
        }

        fn shutdown(&self) -> LifecycleFuture<'_> {
            Box::pin(async move {
                let snapshot = self.health.snapshot();
                assert!(snapshot.is_live());
                assert!(!snapshot.is_ready());
                assert_eq!(snapshot.phase(), ApplicationPhase::Draining);
                self.events.lock().unwrap().push("listener:stopped".into());
                Ok(())
            })
        }
    }

    fn component(name: &'static str, events: &Arc<Mutex<Vec<String>>>) -> RecordingComponent {
        RecordingComponent {
            name,
            events: Arc::clone(events),
            start_error: false,
            shutdown_error: false,
        }
    }

    fn test_error(message: &'static str) -> SaddleError {
        SaddleError::new(ErrorKind::Infrastructure, "test.failure", message)
    }

    fn test_runtime() -> tokio::runtime::Runtime {
        tokio::runtime::Builder::new_current_thread()
            .enable_time()
            .build()
            .expect("test runtime must build")
    }

    async fn shutdown_when_ready(requests: RequestLifecycle) -> Result<()> {
        while requests.phase() != crate::ApplicationPhase::Ready {
            tokio::task::yield_now().await;
        }
        Ok(())
    }

    #[test]
    fn components_start_in_order_and_shutdown_in_reverse() {
        let events = Arc::new(Mutex::new(Vec::new()));
        let mut application = Application::new();
        application.register(component("db", &events)).unwrap();
        application.register(component("service", &events)).unwrap();
        let shutdown = shutdown_when_ready(application.request_lifecycle());

        test_runtime()
            .block_on(application.run_until_shutdown(shutdown))
            .unwrap();

        assert_eq!(
            *events.lock().unwrap(),
            [
                "start:db",
                "start:service",
                "shutdown:service",
                "shutdown:db"
            ]
        );
    }

    #[test]
    fn health_uses_the_unique_lifecycle_and_clears_ready_before_listener_shutdown() {
        test_runtime().block_on(async {
            let events = Arc::new(Mutex::new(Vec::new()));
            let mut application = Application::new();
            let health = application.health();
            let initial = health.snapshot();
            assert!(initial.is_live());
            assert!(!initial.is_ready());
            assert_eq!(initial.phase(), ApplicationPhase::Starting);

            application
                .register(HealthAwareListener {
                    health: health.clone(),
                    events: Arc::clone(&events),
                })
                .unwrap();
            let shutdown_health = health.clone();
            application
                .run_until_shutdown(async move {
                    loop {
                        let snapshot = shutdown_health.snapshot();
                        if snapshot.is_ready() {
                            assert!(snapshot.is_live());
                            assert_eq!(snapshot.phase(), ApplicationPhase::Ready);
                            return Ok(());
                        }
                        tokio::task::yield_now().await;
                    }
                })
                .await
                .unwrap();

            let stopped = health.snapshot();
            assert!(!stopped.is_live());
            assert!(!stopped.is_ready());
            assert_eq!(stopped.phase(), ApplicationPhase::Stopped);
            assert_eq!(
                *events.lock().unwrap(),
                ["listener:accepting", "listener:stopped"]
            );
        });
    }

    #[test]
    fn async_bootstrap_runs_before_early_shutdown_prevents_component_start() {
        let events = Arc::new(Mutex::new(Vec::new()));
        let bootstrap_events = Arc::clone(&events);

        test_runtime()
            .block_on(bootstrap_and_run(
                move || async move {
                    bootstrap_events
                        .lock()
                        .unwrap()
                        .push("bootstrap".to_owned());
                    let mut application = Application::new();
                    application.register(component("component", &bootstrap_events))?;
                    Ok(application)
                },
                std::future::ready(Ok(())),
            ))
            .unwrap();

        assert_eq!(*events.lock().unwrap(), ["bootstrap"]);
    }

    #[test]
    fn failed_async_bootstrap_does_not_start_components() {
        let error = test_runtime()
            .block_on(bootstrap_and_run(
                || async { Err(test_error("bootstrap failed")) },
                std::future::pending(),
            ))
            .unwrap_err();
        assert_eq!(error.message(), "bootstrap failed");
    }

    #[test]
    fn startup_failure_rolls_back_only_started_components() {
        let events = Arc::new(Mutex::new(Vec::new()));
        let mut application = Application::new();
        application.register(component("first", &events)).unwrap();
        let mut failing = component("failing", &events);
        failing.start_error = true;
        application.register(failing).unwrap();
        application.register(component("never", &events)).unwrap();
        let shutdown = shutdown_when_ready(application.request_lifecycle());

        let error = test_runtime()
            .block_on(application.run_until_shutdown(shutdown))
            .unwrap_err();

        assert_eq!(error.message(), "start failed");
        assert_eq!(
            *events.lock().unwrap(),
            ["start:first", "start:failing", "shutdown:first"]
        );
    }

    #[test]
    fn startup_primary_retains_unconfirmed_cleanup_fact() {
        let events = Arc::new(Mutex::new(Vec::new()));
        let mut application = Application::new();
        let mut cleanup = component("cleanup", &events);
        cleanup.shutdown_error = true;
        application.register(cleanup).unwrap();
        let mut startup = component("startup", &events);
        startup.start_error = true;
        application.register(startup).unwrap();
        let primary = test_runtime().block_on(application.run_until_shutdown(
            std::future::pending::<Result<()>>(),
        )).unwrap_err();
        assert_eq!(primary.code(), "test.failure");
        assert_eq!(primary.message(), "start failed");
        assert!(primary.source_unavailable());
        assert!(primary.cleanup_source_unavailable());
    }

    #[test]
    fn shutdown_continues_after_a_component_error() {
        let events = Arc::new(Mutex::new(Vec::new()));
        let mut application = Application::new();
        application.register(component("first", &events)).unwrap();
        let mut failing = component("second", &events);
        failing.shutdown_error = true;
        application.register(failing).unwrap();
        let shutdown = shutdown_when_ready(application.request_lifecycle());

        let error = test_runtime()
            .block_on(application.run_until_shutdown(shutdown))
            .unwrap_err();

        assert_eq!(error.message(), "shutdown failed");
        assert_eq!(
            *events.lock().unwrap(),
            [
                "start:first",
                "start:second",
                "shutdown:second",
                "shutdown:first"
            ]
        );
    }

    #[test]
    fn component_cleanup_original_precedes_output_close() {
        const CHILD: &str = "SADDLE_COMPONENT_CLEANUP_ORIGINAL_CHILD";
        if let Some(path) = std::env::var_os(CHILD) {
            use saddle_observability::{EmergencyDiagnostics, FileLoggingConfig, Rotation};
            let output = EmergencyDiagnostics::start(&FileLoggingConfig::new(
                std::path::PathBuf::from(&path), Rotation::Daily,
            )).unwrap();
            assert!(crate::diagnostics::install_output(output.handle()).is_ok());
            let events = Arc::new(Mutex::new(Vec::new()));
            let mut application = Application::new();
            application.install_lifecycle_observer(
                saddle_observability::Observer::with_writer(Default::default(), std::io::sink()).unwrap(),
                "test-app",
            );
            let mut cleanup = component("cleanup", &events);
            cleanup.shutdown_error = true;
            application.register(cleanup).unwrap();
            let mut startup = component("startup", &events);
            startup.start_error = true;
            application.register(startup).unwrap();
            let primary = test_runtime().block_on(application.run_until_shutdown(
                std::future::pending::<Result<()>>(),
            )).unwrap_err();
            let primary_id = primary.diagnostic().unwrap().id();
            assert!(!primary.source_unavailable());
            assert!(!primary.cleanup_source_unavailable());
            let mut signal_application = Application::new();
            signal_application.install_lifecycle_observer(
                saddle_observability::Observer::with_writer(Default::default(), std::io::sink()).unwrap(),
                "test-app",
            );
            let signal = test_runtime().block_on(signal_application.run_until_shutdown(
                async { Err(signal_error()) },
            )).unwrap_err();
            assert!(!signal.source_unavailable());
            let signal_id = signal.diagnostic().unwrap().id();
            let mut drain_application = Application::new();
            drain_application.set_lifecycle_timeouts(LifecycleTimeouts::from_millis(100, 10).unwrap());
            drain_application.install_lifecycle_observer(
                saddle_observability::Observer::with_writer(Default::default(), std::io::sink()).unwrap(),
                "test-app",
            );
            let requests = drain_application.request_lifecycle();
            let drain = test_runtime().block_on(drain_application.run_until_shutdown(async move {
                while requests.phase() != crate::ApplicationPhase::Ready {
                    tokio::task::yield_now().await;
                }
                let request = requests.try_accept().unwrap();
                tokio::spawn(async move {
                    tokio::time::sleep(Duration::from_millis(50)).await;
                    drop(request);
                });
                Ok(())
            })).unwrap_err();
            assert_eq!(drain.code(), "runtime.lifecycle_timeout.request_drain");
            assert!(!drain.source_unavailable());
            let drain_id = drain.diagnostic().unwrap().id();
            let exit = crate::diagnostics::close_output(
                output, Some(Instant::now() + Duration::from_secs(2)),
            );
            assert_eq!(exit.shutdown, saddle_observability::DiagnosticShutdown::Finished);
            let text = std::fs::read_to_string(
                std::path::PathBuf::from(path).join("saddle.emergency.log"),
            ).unwrap();
            let rows: Vec<serde_json::Value> = text.lines()
                .map(|line| serde_json::from_str(line).unwrap()).collect();
            let field = |rows: &[serde_json::Value], occurrence: &serde_json::Value, channel: &str| -> Option<String> {
                let segments: Vec<_> = rows.iter().filter(|row| row["event"] == "request_error_original"
                    && row["occurrence"] == *occurrence && row["channel"] == channel
                    && row["cause_depth"] == 0).collect();
                if segments.last()?["state"] != "field_end" { return None; }
                Some(segments.iter().map(|row| row["payload"].as_str().unwrap()).collect::<String>())
            };
            let signal_header = rows.iter().find(|row| row["event"] == "request_error_original"
                && row["channel"] == "context"
                && row["occurrence"]["diagnostic_id"] == signal_id)
                .expect("signal original written before output close");
            let signal_context: serde_json::Value = serde_json::from_str(
                signal_header["payload"].as_str().unwrap(),
            ).unwrap();
            assert_eq!(signal_context["stage"], "shutdown_signal");
            assert_eq!(signal_context["context"]["application"]["value"], "test-app");
            assert_eq!(field(&rows, &signal_header["occurrence"], "description").unwrap(),
                "runtime.signal_registration_failed: failed to register the application shutdown signal");
            let drain_header = rows.iter().find(|row| row["event"] == "request_error_original"
                && row["channel"] == "context"
                && row["occurrence"]["diagnostic_id"] == drain_id)
                .expect("drain original written before output close");
            let drain_context: serde_json::Value = serde_json::from_str(
                drain_header["payload"].as_str().unwrap(),
            ).unwrap();
            assert_eq!(drain_context["stage"], "request_drain");
            assert_eq!(drain_context["context"]["application"]["value"], "test-app");
            assert_eq!(field(&rows, &drain_header["occurrence"], "description").unwrap(),
                "runtime.lifecycle_timeout.request_drain: managed lifecycle stage 'request_drain' exceeded its wall-clock deadline");
            let startup_header = rows.iter().find(|row| row["event"] == "request_error_original"
                && row["channel"] == "context"
                && row["occurrence"]["diagnostic_id"] == primary_id)
                .expect("startup original written before cleanup");
            let startup_context: serde_json::Value = serde_json::from_str(
                startup_header["payload"].as_str().unwrap(),
            ).unwrap();
            assert_eq!(startup_context["stage"], "component_start");
            assert_eq!(startup_context["context"]["application"]["value"], "test-app");
            assert_eq!(startup_context["context"]["request"]["state"], "not_established");
            let startup_occurrence = &startup_header["occurrence"];
            assert_eq!(field(&rows, startup_occurrence, "description").unwrap(), "test.failure: start failed");
            assert!(field(&rows, startup_occurrence, "debug").unwrap().contains("start failed"));
            assert!(rows.iter().any(|row| row["occurrence"] == *startup_occurrence
                && row["channel"] == "terminal" && row["state"] == "exposed_chain_complete"));
            let mut missing_startup = rows.clone();
            missing_startup.retain(|row| !(row["occurrence"] == *startup_occurrence
                && row["channel"] == "description" && row["state"] == "field_end"));
            assert!(field(&missing_startup, startup_occurrence, "description").is_none());
            let header = rows.iter().find(|row| row["event"] == "request_error_original"
                && row["channel"] == "context"
                && row["occurrence"]["primary_diagnostic_id"] == primary_id)
                .expect("cleanup original linked to startup primary");
            let context: serde_json::Value = serde_json::from_str(header["payload"].as_str().unwrap()).unwrap();
            assert_eq!(context["context"]["application"]["state"], "present");
            assert_eq!(context["context"]["application"]["value"], "test-app");
            assert_eq!(context["stage"], "component_cleanup");
            let occurrence = &header["occurrence"];
            assert_eq!(field(&rows, occurrence, "description").unwrap(), "test.failure: shutdown failed");
            let debug = field(&rows, occurrence, "debug").unwrap();
            assert!(debug.contains("test.failure"));
            assert!(debug.contains("shutdown failed"));
            let mut missing_debug = rows.clone();
            missing_debug.retain(|row| !(row["occurrence"] == *occurrence
                && row["channel"] == "debug" && row["state"] == "field_end"));
            assert!(field(&missing_debug, occurrence, "debug").is_none());
            assert!(rows.iter().any(|row| row["occurrence"] == *occurrence
                && row["channel"] == "terminal" && row["state"] == "exposed_chain_complete"));
            return;
        }
        let path = std::env::temp_dir().join(format!(
            "saddle-component-cleanup-{}", std::process::id(),
        ));
        std::fs::create_dir(&path).unwrap();
        let mut child = std::process::Command::new(std::env::current_exe().unwrap())
            .args(["--exact", "application::tests::component_cleanup_original_precedes_output_close"])
            .env(CHILD, &path).spawn().unwrap();
        let started = Instant::now();
        let status = loop {
            if let Some(status) = child.try_wait().unwrap() { break status; }
            if started.elapsed() > Duration::from_secs(10) {
                child.kill().unwrap();
                child.wait().unwrap();
                panic!("component cleanup child exceeded watchdog");
            }
            std::thread::sleep(Duration::from_millis(10));
        };
        assert!(status.success());
        std::fs::remove_file(path.join("saddle.emergency.log")).unwrap();
        std::fs::remove_dir(path).unwrap();
    }

    #[test]
    fn duplicate_component_names_are_rejected() {
        let events = Arc::new(Mutex::new(Vec::new()));
        let mut application = Application::new();
        application.register(component("db", &events)).unwrap();

        let error = application.register(component("db", &events)).unwrap_err();
        assert_eq!(error.code(), "runtime.duplicate_component");
    }

    #[test]
    fn application_shutdown_waits_for_an_admitted_request() {
        test_runtime().block_on(async {
            let application = Application::new();
            let requests = application.request_lifecycle();
            let (release, released) = tokio::sync::oneshot::channel();

            let shutdown = async move {
                shutdown_when_ready(requests.clone()).await?;
                let request = requests
                    .try_accept()
                    .expect("application is ready before waiting for shutdown");
                tokio::spawn(async move {
                    released.await.unwrap();
                    drop(request);
                });
                Ok(())
            };
            let running = tokio::spawn(application.run_until_shutdown(shutdown));

            tokio::task::yield_now().await;
            assert!(!running.is_finished());
            release.send(()).unwrap();
            running.await.unwrap().unwrap();
        });
    }

    #[test]
    fn signal_failure_before_start_prevents_component_startup() {
        let events = Arc::new(Mutex::new(Vec::new()));
        let mut application = Application::new();
        application.register(component("service", &events)).unwrap();

        let error = test_runtime()
            .block_on(application.run_until_shutdown(async { Err(signal_error()) }))
            .unwrap_err();

        assert_eq!(error.code(), "runtime.signal_registration_failed");
        assert!(error.source_unavailable());
        assert!(events.lock().unwrap().is_empty());
    }

    #[test]
    fn signal_primary_keeps_independent_cleanup_failure_fact() {
        let signal = test_error("signal failed").with_unconfirmed_source();
        let cleanup = test_error("cleanup failed").with_unconfirmed_cleanup_source();
        let result = super::combine_lifecycle_results(Err(signal), Err(cleanup)).unwrap_err();
        assert_eq!(result.message(), "signal failed");
        assert!(result.source_unavailable());
        assert!(result.cleanup_source_unavailable());
    }

    #[test]
    fn shutdown_during_startup_stops_starting_and_rolls_back() {
        test_runtime().block_on(async {
            let events = Arc::new(Mutex::new(Vec::new()));
            let (started_tx, started_rx) = tokio::sync::oneshot::channel();
            let (release_tx, release_rx) = tokio::sync::oneshot::channel();
            let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel();
            let mut application = Application::new();
            application
                .register(BlockingStartComponent {
                    events: Arc::clone(&events),
                    started: Mutex::new(Some(started_tx)),
                    release: Mutex::new(Some(release_rx)),
                })
                .unwrap();
            application.register(component("never", &events)).unwrap();

            let running = tokio::spawn(application.run_until_shutdown(async move {
                shutdown_rx.await.unwrap();
                Ok(())
            }));
            started_rx.await.unwrap();
            shutdown_tx.send(()).unwrap();
            tokio::task::yield_now().await;
            release_tx.send(()).unwrap();

            running.await.unwrap().unwrap();
            assert_eq!(
                *events.lock().unwrap(),
                ["start:blocking", "shutdown:blocking"]
            );
        });
    }

    #[test]
    fn managed_runtime_provides_an_async_io_driver() {
        build_runtime()
            .unwrap()
            .block_on(async {
                tokio::net::TcpListener::bind((std::net::Ipv4Addr::LOCALHOST, 0)).await
            })
            .expect("service listeners require the managed async I/O driver");
    }

    #[test]
    fn blocked_component_start_times_out_and_rolls_back_started_components() {
        test_runtime().block_on(async {
            let events = Arc::new(Mutex::new(Vec::new()));
            let (started_tx, started_rx) = tokio::sync::oneshot::channel();
            let (_release_tx, release_rx) = tokio::sync::oneshot::channel();
            let mut application = Application::new();
            application.set_lifecycle_timeouts(LifecycleTimeouts::from_millis(10, 100).unwrap());
            application.register(component("first", &events)).unwrap();
            application
                .register(BlockingStartComponent {
                    events: Arc::clone(&events),
                    started: Mutex::new(Some(started_tx)),
                    release: Mutex::new(Some(release_rx)),
                })
                .unwrap();

            let running = tokio::spawn(application.run_until_shutdown(std::future::pending()));
            started_rx.await.unwrap();
            let error = running.await.unwrap().unwrap_err();
            assert_eq!(error.code(), "runtime.lifecycle_timeout.component_start");
            assert_eq!(
                *events.lock().unwrap(),
                [
                    "start:first",
                    "start:blocking",
                    "shutdown:blocking",
                    "shutdown:first"
                ]
            );
        });
    }

    #[test]
    fn blocked_component_shutdown_uses_one_total_deadline_and_is_not_clean() {
        test_runtime().block_on(async {
            let events = Arc::new(Mutex::new(Vec::new()));
            let mut application = Application::new();
            application.set_lifecycle_timeouts(LifecycleTimeouts::from_millis(100, 10).unwrap());
            application
                .register(BlockingShutdownComponent {
                    events: Arc::clone(&events),
                })
                .unwrap();
            let shutdown = shutdown_when_ready(application.request_lifecycle());
            let error = application.run_until_shutdown(shutdown).await.unwrap_err();
            assert_eq!(error.code(), "runtime.lifecycle_timeout.component_shutdown");
            assert_eq!(
                *events.lock().unwrap(),
                ["start:blocking-shutdown", "shutdown:blocking-shutdown"]
            );
        });
    }
}

#[cfg(all(test, unix))]
mod actual_signal_receipt_tests {
    #[test]
    fn isolated_sigterm_selected_recv_writes_owned_original() {
        const KEY: &str = "SADDLE_RESOURCE_SIGNAL_CHILD_PATH";
        let path = match std::env::var_os(KEY) {
            Some(path) => std::path::PathBuf::from(path),
            None => {
                let path = std::env::temp_dir().join(format!("resource-signal-{}-{}", std::process::id(),
                    std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_nanos()));
                std::fs::create_dir(&path).unwrap();
                let mut child = std::process::Command::new(std::env::current_exe().unwrap())
                    .args(["--exact", "application::actual_signal_receipt_tests::isolated_sigterm_selected_recv_writes_owned_original", "--nocapture"])
                    .env(KEY, &path).spawn().unwrap();
                let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10);
                while !path.join("ready").exists() {
                    if let Some(status) = child.try_wait().unwrap() { panic!("child exited before ready: {status}"); }
                    if std::time::Instant::now() >= deadline { child.kill().unwrap(); child.wait().unwrap(); panic!("signal child readiness timeout"); }
                    std::thread::sleep(std::time::Duration::from_millis(10));
                }
                assert!(std::process::Command::new("kill").args(["-TERM", &child.id().to_string()]).status().unwrap().success());
                loop {
                    if let Some(status) = child.try_wait().unwrap() { assert!(status.success()); break; }
                    if std::time::Instant::now() >= deadline { child.kill().unwrap(); child.wait().unwrap(); panic!("signal child exit timeout"); }
                    std::thread::sleep(std::time::Duration::from_millis(10));
                }
                let raw = std::fs::read_to_string(path.join("saddle.emergency.log")).unwrap();
                let rows: Vec<serde_json::Value> = raw.lines().map(|line| serde_json::from_str(line).unwrap()).collect();
                let receipts: Vec<_> = rows.iter().filter(|r| r["schema"] == "process-signal-receipt/1").collect();
                assert_eq!(receipts.len(), 1);
                assert_eq!(receipts[0]["signal"], "SIGTERM");
                assert_eq!(receipts[0]["received"], true);
                assert_eq!(receipts[0]["application"], "isolated-signal-app");
                assert_eq!(receipts[0]["pid"], child.id());
                println!("actual signal evidence {}", path.display());
                return;
            }
        };
        let mut output = saddle_observability::EmergencyDiagnostics::start(
            &saddle_observability::FileLoggingConfig::new(path.clone(), saddle_observability::Rotation::Daily)).unwrap();
        let runtime = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap();
        runtime.block_on(async {
            let signal = super::ShutdownSignal::register().unwrap();
            std::fs::write(path.join("ready"), b"registered").unwrap();
            tokio::time::timeout(std::time::Duration::from_secs(8), signal.wait_observed(Some((
                saddle_core::ContextLabel::checked("isolated-signal-app").unwrap(), output.handle())))).await.unwrap().unwrap();
        });
        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(2);
        while output.shutdown() == saddle_observability::DiagnosticShutdown::Pending {
            assert!(std::time::Instant::now() < deadline);
            std::thread::sleep(std::time::Duration::from_millis(5));
        }
        assert_eq!(output.snapshot().dropped, 0);
        assert_eq!(output.snapshot().enqueued, output.snapshot().written);
    }
}