lightshuttle_runtime/lifecycle/

manager.rs

//! Coordinated startup, supervision, and shutdown of every resource in a
//! [`crate::LifecyclePlan`].
//!
//! The main type, [`LifecycleManager`], is generic over any
//! [`crate::ContainerRuntime`] implementation. It spawns one `tokio` task per
//! resource; each task waits for its dependencies to reach a ready state before
//! calling `start` on the runtime. Status transitions are published on a
//! `tokio::sync::watch` channel (consumed by peer tasks for ordering) and on a
//! `tokio::sync::broadcast` channel (consumed by the CLI, dashboard, and tests
//! via [`LifecycleManager::subscribe_events`]).
//!
//! ## Startup sequence (per resource)
//!
//! 1. Wait for every dependency to reach [`crate::NodeStatus::Running`] or
//!    [`crate::NodeStatus::Healthy`].
//! 2. Collect dependency outputs and resolve `${resources.*}` interpolations.
//! 3. Inject `LSH_<DEP>_<PROPERTY>` environment variables automatically.
//! 4. Remove any stale container with the same name.
//! 5. Call [`crate::ContainerRuntime::start`].
//! 6. Poll [`crate::ContainerRuntime::wait_healthy`] until healthy or timeout.
//!
//! ## Teardown
//!
//! Resources are stopped in reverse topological order. Each stop sends
//! `SIGTERM` and waits up to the configured grace window before issuing
//! `SIGKILL`. After all containers are removed, the per-project bridge network
//! is torn down.

use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use std::time::{Duration, SystemTime};

use lightshuttle_manifest::{InterpolationContext, Interpolator};
use tokio::sync::{broadcast, watch};
use tracing::{Instrument, debug, info, info_span, instrument, warn};

/// Buffer size for the broadcast event channel. Slow subscribers that
/// fall behind by more than this number of events will see lagged
/// messages and have to resynchronise.
const EVENT_CHANNEL_CAPACITY: usize = 256;

use crate::error::RuntimeError;
use crate::lifecycle::error::LifecycleError;
use crate::lifecycle::plan::LifecyclePlan;
use crate::lifecycle::status::{LifecycleEvent, NodeStatus};
use crate::runtime::{ContainerId, ContainerRuntime};
use lightshuttle_spec::{ContainerSpec, ResourceOutputs};

/// Default healthcheck timeout, applied when the manifest does not
/// provide one of its own. Kept conservative for v0.1.
const DEFAULT_HEALTHCHECK_TIMEOUT: Duration = Duration::from_secs(60);

/// Per-resource shared state.
#[derive(Clone)]
struct NodeHandle {
    status_tx: Arc<watch::Sender<NodeStatus>>,
    status_rx: watch::Receiver<NodeStatus>,
    outputs_tx: Arc<watch::Sender<Option<ResourceOutputs>>>,
    outputs_rx: watch::Receiver<Option<ResourceOutputs>>,
    container_id: Arc<Mutex<Option<ContainerId>>>,
    started_at: Arc<Mutex<Option<SystemTime>>>,
}

/// Point-in-time snapshot of one managed resource, consumed by the
/// control plane via [`super::handle::ManagerHandle`].
pub(super) struct NodeSnapshot {
    /// Lifecycle status at the moment of the snapshot.
    pub(super) status: NodeStatus,
    /// Wall-clock time at which the runtime accepted the start request.
    pub(super) started_at: Option<SystemTime>,
    /// Container identifier returned by the runtime, when known.
    pub(super) container_id: Option<ContainerId>,
}

/// Coordinates the startup, supervision, and shutdown of every resource
/// declared in a [`LifecyclePlan`].
///
/// Construct with [`LifecycleManager::new`], optionally inject extra
/// environment variables with [`LifecycleManager::with_env`], then call one of:
///
/// - [`LifecycleManager::start_all`]: start all resources and return.
/// - [`LifecycleManager::run_until_signal`]: start all resources, block until
///   `SIGINT` or `SIGTERM`, then stop cleanly (the typical `lightshuttle up`
///   entry point).
///
/// # Example
///
/// ```rust,no_run
/// use std::time::Duration;
///
/// use lightshuttle_manifest::Manifest;
/// use lightshuttle_runtime::{LifecyclePlan, LifecycleManager};
/// use lightshuttle_runtime::testkit::MockRuntime;
///
/// # #[tokio::main]
/// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
/// let manifest = Manifest::parse(
///     "project:\n  name: app\nresources:\n  db:\n    postgres:\n      version: \"16\"\n"
/// )?;
/// let plan = LifecyclePlan::from_manifest(&manifest)?;
/// let (manager, mut events) = LifecycleManager::new(plan, MockRuntime::new());
///
/// manager.start_all().await?;
/// manager.stop_all(Duration::from_secs(5)).await?;
/// # Ok(())
/// # }
/// ```
pub struct LifecycleManager<R: ContainerRuntime + 'static> {
    plan: Arc<LifecyclePlan>,
    runtime: Arc<R>,
    nodes: HashMap<String, NodeHandle>,
    event_tx: broadcast::Sender<LifecycleEvent>,
    extra_env: Arc<HashMap<String, String>>,
}

impl<R: ContainerRuntime + 'static> LifecycleManager<R> {
    /// Build a manager bound to `plan` and `runtime`. Returns a fresh
    /// event subscriber alongside; further subscribers can be obtained
    /// from [`Self::subscribe_events`].
    #[must_use]
    pub fn new(plan: LifecyclePlan, runtime: R) -> (Self, broadcast::Receiver<LifecycleEvent>) {
        let (event_tx, event_rx) = broadcast::channel(EVENT_CHANNEL_CAPACITY);
        let mut nodes: HashMap<String, NodeHandle> = HashMap::new();
        for node in plan.nodes() {
            let (status_tx, status_rx) = watch::channel(NodeStatus::Pending);
            let (outputs_tx, outputs_rx) = watch::channel(None);
            nodes.insert(
                node.name.clone(),
                NodeHandle {
                    status_tx: Arc::new(status_tx),
                    status_rx,
                    outputs_tx: Arc::new(outputs_tx),
                    outputs_rx,
                    container_id: Arc::new(Mutex::new(None)),
                    started_at: Arc::new(Mutex::new(None)),
                },
            );
        }
        let manager = Self {
            plan: Arc::new(plan),
            runtime: Arc::new(runtime),
            nodes,
            event_tx,
            extra_env: Arc::new(HashMap::new()),
        };
        (manager, event_rx)
    }

    /// Merge additional environment variables into the interpolation context
    /// used for every resource.
    ///
    /// Variables provided here take precedence over same-named variables from
    /// the ambient process environment. The typical use-case is forwarding the
    /// contents of a `.env` file so that `${env.NAME}` references in the
    /// manifest resolve to the file values. Call before [`Self::start_all`].
    ///
    /// Returns `self` for method chaining.
    #[must_use]
    pub fn with_env(mut self, env: HashMap<String, String>) -> Self {
        self.extra_env = Arc::new(env);
        self
    }

    /// Scan every resource spec for `${env.VAR}` references that cannot be
    /// resolved and return a single error listing all missing names.
    ///
    /// Delegates to [`LifecyclePlan::env_report`] so the fail-fast preflight
    /// and the `lightshuttle secrets check` diagnostic command share one source
    /// of truth. Call before [`Self::start_all`] to surface missing variables
    /// before any container is started, which avoids a partial stack start
    /// followed by an immediate rollback.
    ///
    /// # Errors
    ///
    /// Returns [`crate::LifecycleError::MissingEnvVars`] with a sorted,
    /// deduplicated list of every missing variable name.
    pub fn check_required_env(&self) -> Result<(), LifecycleError> {
        let report = self.plan.env_report(&self.extra_env);
        if report.has_missing() {
            Err(LifecycleError::MissingEnvVars {
                names: report.missing(),
            })
        } else {
            Ok(())
        }
    }

    /// Start every resource in topological order, with independent branches
    /// starting in parallel.
    ///
    /// Each resource waits for its dependencies to become ready (i.e. reach
    /// [`crate::NodeStatus::Running`] or [`crate::NodeStatus::Healthy`]) before
    /// calling [`crate::ContainerRuntime::start`]. Readiness is gate-kept by the
    /// healthcheck: a container with a declared healthcheck must report
    /// [`crate::ContainerStatus::Healthy`] before its dependents may proceed.
    ///
    /// On the first failure, every resource that has already started is stopped
    /// automatically (best-effort, 10-second grace) before the error is
    /// returned.
    ///
    /// # Errors
    ///
    /// Returns the first [`crate::LifecycleError`] encountered. Secondary
    /// failures from the automatic rollback are logged but not returned.
    pub async fn start_all(&self) -> Result<(), LifecycleError> {
        let mut handles: Vec<tokio::task::JoinHandle<Result<(), LifecycleError>>> =
            Vec::with_capacity(self.plan.nodes().len());

        for node in self.plan.nodes() {
            let mut dep_status_rxs: HashMap<String, watch::Receiver<NodeStatus>> = HashMap::new();
            let mut dep_outputs_rxs: HashMap<String, watch::Receiver<Option<ResourceOutputs>>> =
                HashMap::new();
            for dep in &node.depends_on {
                let handle = self
                    .nodes
                    .get(dep)
                    .ok_or_else(|| LifecycleError::ResourceNotFound(dep.clone()))?;
                dep_status_rxs.insert(dep.clone(), handle.status_rx.clone());
                dep_outputs_rxs.insert(dep.clone(), handle.outputs_rx.clone());
            }

            let node_handle = self.nodes[&node.name].clone();
            let spec = node.spec.clone();
            let own_outputs = node.outputs.clone();
            let name = node.name.clone();
            let runtime = Arc::clone(&self.runtime);
            let event_tx = self.event_tx.clone();
            let extra_env = Arc::clone(&self.extra_env);

            let task = tokio::spawn(async move {
                start_one(
                    name,
                    spec,
                    own_outputs,
                    runtime,
                    node_handle,
                    dep_status_rxs,
                    dep_outputs_rxs,
                    event_tx,
                    extra_env,
                )
                .await
            });
            handles.push(task);
        }

        let mut first_error: Option<LifecycleError> = None;
        for handle in handles {
            match handle.await {
                Ok(Ok(())) => {}
                Ok(Err(err)) => {
                    if first_error.is_none() {
                        first_error = Some(err);
                    }
                }
                Err(join_err) => {
                    if first_error.is_none() {
                        first_error = Some(LifecycleError::Start {
                            resource: "<panicked task>".to_owned(),
                            source: RuntimeError::InvalidSpec(join_err.to_string()),
                        });
                    }
                }
            }
        }

        if let Some(err) = first_error {
            warn!(error = %err, "start_all failed; rolling back");
            let _ = self.stop_all(Duration::from_secs(10)).await;
            return Err(err);
        }

        let _ = self.event_tx.send(LifecycleEvent::StackStarted);
        info!(
            "stack started: {} resource(s) healthy",
            self.plan.nodes().len()
        );
        Ok(())
    }

    /// Stop every resource in reverse topological order.
    ///
    /// Each resource receives `SIGTERM`. After `grace` elapses, the runtime
    /// sends `SIGKILL` to any container that has not exited yet. Resources are
    /// stopped in the reverse of startup order (dependents before their
    /// dependencies). After all containers are removed, the per-project bridge
    /// network is torn down (failure is logged but does not abort the call).
    ///
    /// # Errors
    ///
    /// Returns the first [`crate::LifecycleError::Stop`] encountered. Other
    /// stop failures are logged but not propagated.
    #[instrument(skip_all, fields(resources = self.plan.nodes().len()))]
    pub async fn stop_all(&self, grace: Duration) -> Result<(), LifecycleError> {
        let _ = self.event_tx.send(LifecycleEvent::StackStopping);

        let mut errors: Vec<(String, RuntimeError)> = Vec::new();
        for node in self.plan.nodes().iter().rev() {
            let Some(handle) = self.nodes.get(&node.name) else {
                continue;
            };
            let id = {
                let guard = handle
                    .container_id
                    .lock()
                    .expect("container_id mutex poisoned");
                guard.clone()
            };
            let Some(id) = id else { continue };
            let stop_span = info_span!("stop", resource = %node.name);
            match self.runtime.stop(&id, grace).instrument(stop_span).await {
                Ok(()) => {
                    let _ = handle.status_tx.send(NodeStatus::Stopped);
                    let _ = self.event_tx.send(LifecycleEvent::ResourceStopped {
                        name: node.name.clone(),
                    });
                }
                Err(e) => errors.push((node.name.clone(), e)),
            }
        }

        let _ = self.event_tx.send(LifecycleEvent::StackStopped);

        // Remove the per-project bridge network. Containers that failed
        // to stop may still hold endpoints, causing Docker to reject the
        // request: log the failure and continue so callers always see
        // the primary stop errors, not a secondary network error.
        if let Some(project) = self.plan.nodes().first().map(|n| n.spec.project.as_str()) {
            if let Err(e) = self.runtime.teardown_project_network(project).await {
                warn!(error = %e, "could not remove project network");
            }
        }

        if let Some((resource, source)) = errors.into_iter().next() {
            return Err(LifecycleError::Stop { resource, source });
        }
        Ok(())
    }

    /// Start the stack, wait for `SIGINT` or `SIGTERM`, then stop cleanly.
    ///
    /// This is the opinionated entry point for the `lightshuttle up` command.
    /// It calls [`Self::start_all`], blocks until a shutdown signal is received,
    /// then calls [`Self::stop_all`] with the provided `grace` window.
    ///
    /// On Unix, both `SIGINT` (Ctrl+C) and `SIGTERM` trigger the teardown.
    /// On Windows, only `Ctrl+C` is intercepted.
    ///
    /// # Example
    ///
    /// ```rust,no_run
    /// use std::time::Duration;
    ///
    /// use lightshuttle_manifest::Manifest;
    /// use lightshuttle_runtime::{DockerRuntime, LifecyclePlan, LifecycleManager};
    ///
    /// # #[tokio::main]
    /// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
    /// let manifest = Manifest::parse(
    ///     "project:\n  name: app\nresources:\n  db:\n    postgres:\n      version: \"16\"\n"
    /// )?;
    /// let plan = LifecyclePlan::from_manifest(&manifest)?;
    /// let runtime = DockerRuntime::connect()?;
    /// let (manager, _events) = LifecycleManager::new(plan, runtime);
    ///
    /// // Blocks until Ctrl+C or SIGTERM.
    /// manager.run_until_signal(Duration::from_secs(30)).await?;
    /// # Ok(())
    /// # }
    /// ```
    ///
    /// # Errors
    ///
    /// Propagates errors from [`Self::start_all`] or [`Self::stop_all`].
    pub async fn run_until_signal(&self, grace: Duration) -> Result<(), LifecycleError> {
        self.start_all().await?;
        wait_for_shutdown_signal().await;
        self.stop_all(grace).await
    }

    /// Restart a single resource without touching its dependents.
    ///
    /// The target is stopped via `SIGTERM` (10-second grace window), its
    /// container id and started-at timestamp are cleared, then the full
    /// `start_one` cycle is re-run from the same cached spec. Three events are
    /// emitted on the lifecycle channel in order: [`crate::LifecycleEvent::ResourceStopped`],
    /// [`crate::LifecycleEvent::ResourceStarted`], [`crate::LifecycleEvent::ResourceHealthy`].
    ///
    /// Dependents keep running. Their internal `watch` channels observe the
    /// target's status transition through `Stopped` -> `Pending` -> `Starting`
    /// -> `Running` -> `Healthy`, so upstream processes that hold a watch
    /// receiver can pause themselves locally until the dependency is healthy
    /// again.
    ///
    /// # Errors
    ///
    /// Returns [`crate::LifecycleError::ResourceNotFound`] when `resource` is
    /// not part of the plan, or a [`crate::LifecycleError::Start`] /
    /// [`crate::LifecycleError::Stop`] variant on runtime failure.
    #[instrument(skip(self), fields(resource = %resource))]
    pub async fn restart_one(&self, resource: &str) -> Result<(), LifecycleError> {
        let node = self
            .plan
            .nodes()
            .iter()
            .find(|n| n.name == resource)
            .ok_or_else(|| LifecycleError::ResourceNotFound(resource.to_owned()))?;
        let handle = self
            .nodes
            .get(resource)
            .ok_or_else(|| LifecycleError::ResourceNotFound(resource.to_owned()))?;

        // Stop the running container if any.
        let id = {
            let guard = handle
                .container_id
                .lock()
                .expect("container_id mutex poisoned");
            guard.clone()
        };
        if let Some(id) = id {
            self.runtime
                .stop(&id, Duration::from_secs(10))
                .await
                .map_err(|source| LifecycleError::Stop {
                    resource: resource.to_owned(),
                    source,
                })?;
            *handle
                .container_id
                .lock()
                .expect("container_id mutex poisoned") = None;
            *handle.started_at.lock().expect("started_at mutex poisoned") = None;
            let _ = handle.status_tx.send(NodeStatus::Stopped);
            let _ = self.event_tx.send(LifecycleEvent::ResourceStopped {
                name: resource.to_owned(),
            });
        }

        // Reset to Pending so start_one drives the full restart cycle.
        let _ = handle.status_tx.send(NodeStatus::Pending);

        // Collect dependency watch receivers. Deps are already Healthy,
        // so start_one's wait loop returns instantly.
        let mut dep_status_rxs: HashMap<String, watch::Receiver<NodeStatus>> = HashMap::new();
        let mut dep_outputs_rxs: HashMap<String, watch::Receiver<Option<ResourceOutputs>>> =
            HashMap::new();
        for dep in &node.depends_on {
            let dep_handle = self
                .nodes
                .get(dep)
                .ok_or_else(|| LifecycleError::ResourceNotFound(dep.clone()))?;
            dep_status_rxs.insert(dep.clone(), dep_handle.status_rx.clone());
            dep_outputs_rxs.insert(dep.clone(), dep_handle.outputs_rx.clone());
        }

        start_one(
            resource.to_owned(),
            node.spec.clone(),
            node.outputs.clone(),
            Arc::clone(&self.runtime),
            handle.clone(),
            dep_status_rxs,
            dep_outputs_rxs,
            self.event_tx.clone(),
            Arc::clone(&self.extra_env),
        )
        .await
    }

    /// Open a new subscription on the lifecycle event broadcast.
    ///
    /// Multiple subscribers can read concurrently. Subscribers that
    /// fall more than 256 events behind (the broadcast channel capacity)
    /// observe a `RecvError::Lagged` and have to resynchronise.
    #[must_use]
    pub fn subscribe_events(&self) -> broadcast::Receiver<LifecycleEvent> {
        self.event_tx.subscribe()
    }

    /// Shared reference to the underlying execution plan.
    pub(super) fn plan_arc(&self) -> &Arc<LifecyclePlan> {
        &self.plan
    }

    /// Shared reference to the underlying container runtime.
    pub(super) fn runtime_arc(&self) -> &Arc<R> {
        &self.runtime
    }

    /// Point-in-time snapshot of one resource, or `None` when the name
    /// is not part of the plan.
    pub(super) fn snapshot(&self, name: &str) -> Option<NodeSnapshot> {
        let handle = self.nodes.get(name)?;
        let status = handle.status_rx.borrow().clone();
        let started_at = *handle.started_at.lock().expect("started_at mutex poisoned");
        let container_id = handle
            .container_id
            .lock()
            .expect("container_id mutex poisoned")
            .clone();
        Some(NodeSnapshot {
            status,
            started_at,
            container_id,
        })
    }
}

#[allow(clippy::too_many_arguments, clippy::too_many_lines)]
#[instrument(name = "start", skip_all, fields(resource = %name))]
async fn start_one<R: ContainerRuntime + 'static>(
    name: String,
    spec: ContainerSpec,
    own_outputs: ResourceOutputs,
    runtime: Arc<R>,
    handle: NodeHandle,
    dep_status_rxs: HashMap<String, watch::Receiver<NodeStatus>>,
    mut dep_outputs_rxs: HashMap<String, watch::Receiver<Option<ResourceOutputs>>>,
    event_tx: broadcast::Sender<LifecycleEvent>,
    extra_env: Arc<HashMap<String, String>>,
) -> Result<(), LifecycleError> {
    // 1. Wait for every dependency to become ready.
    for (dep_name, mut rx) in dep_status_rxs {
        loop {
            let status = rx.borrow_and_update().clone();
            if status.is_ready() {
                debug!(node = %name, dep = %dep_name, "dependency ready");
                break;
            }
            if let NodeStatus::Failed { reason } = status {
                let _ = handle.status_tx.send(NodeStatus::Failed {
                    reason: format!("dependency `{dep_name}` failed: {reason}"),
                });
                return Err(LifecycleError::DependencyFailed {
                    resource: name,
                    dependency: dep_name,
                    reason,
                });
            }
            if rx.changed().await.is_err() {
                let reason = format!("dependency `{dep_name}` watch channel closed");
                let _ = handle.status_tx.send(NodeStatus::Failed {
                    reason: reason.clone(),
                });
                return Err(LifecycleError::DependencyFailed {
                    resource: name,
                    dependency: dep_name,
                    reason,
                });
            }
        }
    }

    // 2. Collect dependency outputs.
    let mut dep_outputs: HashMap<String, ResourceOutputs> = HashMap::new();
    for (dep_name, rx) in &mut dep_outputs_rxs {
        loop {
            if let Some(out) = rx.borrow_and_update().clone() {
                dep_outputs.insert(dep_name.clone(), out);
                break;
            }
            if rx.changed().await.is_err() {
                let reason = format!("dependency `{dep_name}` outputs channel closed");
                let _ = handle.status_tx.send(NodeStatus::Failed {
                    reason: reason.clone(),
                });
                return Err(LifecycleError::DependencyFailed {
                    resource: name,
                    dependency: dep_name.clone(),
                    reason,
                });
            }
        }
    }

    // 3. Resolve interpolations and inject LSH_<DEP>_<PROP> env vars.
    let resolved_spec = match interpolate_and_inject(spec, &dep_outputs, &extra_env) {
        Ok(s) => s,
        Err(reason) => {
            let _ = handle.status_tx.send(NodeStatus::Failed {
                reason: reason.clone(),
            });
            return Err(LifecycleError::Start {
                resource: name,
                source: RuntimeError::InvalidSpec(reason),
            });
        }
    };

    // 4. Remove any container left over from a previous run so the
    //    create call below never collides with a stale name.
    let _ = handle.status_tx.send(NodeStatus::Starting);
    if let Err(source) = runtime.remove(&resolved_spec.name).await {
        let _ = handle.status_tx.send(NodeStatus::Failed {
            reason: source.to_string(),
        });
        let _ = event_tx.send(LifecycleEvent::ResourceFailed {
            name: name.clone(),
            error: source.to_string(),
        });
        return Err(LifecycleError::Start {
            resource: name,
            source,
        });
    }

    // 5. Start the container.
    let id = match runtime.start(&resolved_spec).await {
        Ok(id) => id,
        Err(source) => {
            let _ = handle.status_tx.send(NodeStatus::Failed {
                reason: source.to_string(),
            });
            let _ = event_tx.send(LifecycleEvent::ResourceFailed {
                name: name.clone(),
                error: source.to_string(),
            });
            return Err(LifecycleError::Start {
                resource: name,
                source,
            });
        }
    };

    {
        let mut guard = handle
            .container_id
            .lock()
            .expect("container_id mutex poisoned");
        *guard = Some(id.clone());
    }
    {
        let mut guard = handle.started_at.lock().expect("started_at mutex poisoned");
        *guard = Some(SystemTime::now());
    }
    let _ = handle.status_tx.send(NodeStatus::Running);
    let _ = event_tx.send(LifecycleEvent::ResourceStarted {
        name: name.clone(),
        container_id: id.to_string(),
    });

    // 6. Wait for the healthcheck.
    let wait_span = info_span!("wait_healthy", resource = %name);
    match runtime
        .wait_healthy(&id, DEFAULT_HEALTHCHECK_TIMEOUT)
        .instrument(wait_span)
        .await
    {
        Ok(()) => {
            let _ = handle.outputs_tx.send(Some(own_outputs));
            let _ = handle.status_tx.send(NodeStatus::Healthy);
            let _ = event_tx.send(LifecycleEvent::ResourceHealthy { name: name.clone() });
            Ok(())
        }
        Err(RuntimeError::Timeout { .. }) => {
            let reason = format!("healthcheck timed out after {DEFAULT_HEALTHCHECK_TIMEOUT:?}");
            let _ = handle.status_tx.send(NodeStatus::Failed {
                reason: reason.clone(),
            });
            let _ = event_tx.send(LifecycleEvent::ResourceFailed {
                name: name.clone(),
                error: reason,
            });
            Err(LifecycleError::HealthcheckTimeout {
                resource: name,
                timeout: DEFAULT_HEALTHCHECK_TIMEOUT,
            })
        }
        Err(source) => {
            let _ = handle.status_tx.send(NodeStatus::Failed {
                reason: source.to_string(),
            });
            let _ = event_tx.send(LifecycleEvent::ResourceFailed {
                name: name.clone(),
                error: source.to_string(),
            });
            Err(LifecycleError::Start {
                resource: name,
                source,
            })
        }
    }
}

/// Apply two-pass interpolation to `spec`: resolve every
/// `${resources.<name>.<property>}` against `dep_outputs`, then inject
/// `LSH_<DEP>_<PROPERTY>` automatic environment variables.
///
/// Returns the resolved spec or a human-readable diagnostic when an
/// interpolation references an unknown resource or property.
fn interpolate_and_inject(
    mut spec: ContainerSpec,
    dep_outputs: &HashMap<String, ResourceOutputs>,
    extra_env: &HashMap<String, String>,
) -> std::result::Result<ContainerSpec, String> {
    let mut ctx = InterpolationContext::from_env()
        .with_env(extra_env.iter().map(|(k, v)| (k.clone(), v.clone())));
    for (name, outputs) in dep_outputs {
        ctx = ctx.with_resource(name.clone(), outputs.clone());
    }
    let interpolator = Interpolator::new(&ctx);

    // Resolve env values.
    let mut resolved_env = std::collections::HashMap::with_capacity(spec.env.len());
    for (k, v) in spec.env.drain() {
        let resolved = interpolator.resolve(&v).map_err(|e| e.to_string())?;
        resolved_env.insert(k, resolved);
    }

    // Inject LSH_<DEP>_<PROPERTY> variables.
    for (dep_name, outputs) in dep_outputs {
        let dep_upper = dep_name.to_uppercase().replace('-', "_");
        for (prop, value) in outputs {
            let prop_upper = prop.to_uppercase().replace('-', "_");
            let key = format!("LSH_{dep_upper}_{prop_upper}");
            resolved_env.entry(key).or_insert_with(|| value.clone());
        }
    }
    spec.env = resolved_env;

    // Resolve command arguments.
    if let Some(args) = spec.command.as_mut() {
        for arg in args.iter_mut() {
            *arg = interpolator.resolve(arg).map_err(|e| e.to_string())?;
        }
    }

    Ok(spec)
}

#[cfg(unix)]
async fn wait_for_shutdown_signal() {
    use tokio::signal::unix::{SignalKind, signal};
    let mut sigterm = match signal(SignalKind::terminate()) {
        Ok(s) => s,
        Err(e) => {
            warn!("failed to install SIGTERM handler: {e}");
            let _ = tokio::signal::ctrl_c().await;
            return;
        }
    };
    tokio::select! {
        _ = tokio::signal::ctrl_c() => info!("received SIGINT"),
        _ = sigterm.recv() => info!("received SIGTERM"),
    }
}

#[cfg(windows)]
async fn wait_for_shutdown_signal() {
    let _ = tokio::signal::ctrl_c().await;
    info!("received Ctrl+C");
}