pitchfork_cli/supervisor/

lifecycle.rs

//! Daemon lifecycle management - start/stop operations
//!
//! Contains the core `run()`, `run_once()`, and `stop()` methods for daemon process management.

use super::hooks::{self, HookType, fire_hook};
use super::{SUPERVISOR, Supervisor};
use crate::config_types::OneshotWait;
use crate::daemon::RunOptions;
use crate::daemon_id::DaemonId;
use crate::daemon_status::DaemonStatus;
use crate::error::PortError;
use crate::ipc::IpcResponse;
use crate::log_store::LogStore;
use crate::log_store::sqlite::LOG_STORE;
use crate::pitchfork_toml::{ReadyCmd, ReadyHttp, ReadyOutput, ReadyPort};
use crate::procs::PROCS;
use crate::settings::{resolve_shell, settings};
use crate::shell::{HideConsoleWindow, Shell, ShellScript};
use crate::supervisor::state::UpsertDaemonOpts;
use crate::{Result, env};
use indexmap::IndexMap;
use miette::IntoDiagnostic;
use once_cell::sync::Lazy;
use regex::Regex;
use std::collections::HashMap;
#[cfg(unix)]
use std::ffi::CString;
use std::sync::{Arc, atomic};
use std::time::Duration;
use tokio::select;
use tokio::sync::oneshot;
use tokio::time;

/// Cache for compiled regex patterns to avoid recompilation on daemon restarts
static REGEX_CACHE: Lazy<std::sync::Mutex<HashMap<String, Regex>>> =
    Lazy::new(|| std::sync::Mutex::new(HashMap::new()));

fn resolve_configured_ready_port(
    configured_port: u16,
    expected_ports: &[u16],
    resolved_ports: &[u16],
) -> u16 {
    let bump_offset = resolved_ports
        .first()
        .unwrap_or(&0)
        .saturating_sub(*expected_ports.first().unwrap_or(&0));
    if expected_ports.contains(&configured_port) && bump_offset > 0 {
        configured_port
            .checked_add(bump_offset)
            .unwrap_or(configured_port)
    } else {
        configured_port
    }
}

fn active_port_from_ready_port(ready_port: u16, resolved_ports: &[u16]) -> Option<u16> {
    resolved_ports
        .first()
        .copied()
        .filter(|&primary_port| primary_port == ready_port)
}

#[cfg(unix)]
#[derive(Clone, Debug, PartialEq, Eq)]
enum RunIdentity {
    Inherit,
    Switch {
        uid: nix::unistd::Uid,
        gid: nix::unistd::Gid,
        username: Option<CString>,
        /// Home directory from the user's passwd entry.
        home: Option<std::path::PathBuf>,
    },
}

/// Get or compile a regex pattern, caching the result for future use
pub(crate) fn get_or_compile_regex(pattern: &str) -> Option<Regex> {
    let mut cache = REGEX_CACHE.lock().unwrap_or_else(|e| e.into_inner());
    if let Some(re) = cache.get(pattern) {
        return Some(re.clone());
    }
    match Regex::new(pattern) {
        Ok(re) => {
            cache.insert(pattern.to_string(), re.clone());
            Some(re)
        }
        Err(e) => {
            error!("invalid regex pattern '{pattern}': {e}");
            None
        }
    }
}

/// Handle for an in-flight readiness command probe.
///
/// The spawned task owns the `tokio::process::Child` and waits for either the
/// process to exit or the cancel signal. Dropping the handle without cancelling
/// leaves the task running, but the child is started with `kill_on_drop(true)`
/// so it will still be terminated when the task ends.
pub(crate) struct CmdProbe {
    pub(crate) cancel_tx: tokio::sync::oneshot::Sender<()>,
    pub(crate) result_rx: tokio::sync::oneshot::Receiver<std::io::Result<std::process::ExitStatus>>,
}

/// Spawn a readiness command probe and return a handle that can be used to wait
/// for the exit status or cancel the probe.
///
/// The probe is started with `kill_on_drop(true)` as a cancellation fallback. The
/// spawned task waits for the process to exit; if cancellation is requested, it
/// kills the child and waits for it to reap before reporting the result.
fn apply_runtime_env(
    command: &mut tokio::process::Command,
    id: &DaemonId,
    retry_count: u32,
    daemon_env: Option<&IndexMap<String, String>>,
    resolved_ports: &[u16],
) {
    if let Some(ref path) = *env::ORIGINAL_PATH {
        command.env("PATH", path);
    }
    if let Some(env_vars) = daemon_env {
        command.envs(env_vars);
    }
    command
        .env("PITCHFORK_DAEMON_ID", id.qualified())
        .env("PITCHFORK_DAEMON_NAMESPACE", id.namespace())
        .env("PITCHFORK_RETRY_COUNT", retry_count.to_string());
    if let Some(port) = resolved_ports.first() {
        command.env("PORT", port.to_string());
        for (index, port) in resolved_ports.iter().enumerate() {
            command.env(format!("PORT{index}"), port.to_string());
        }
    }
}

pub(crate) fn spawn_cmd_probe(
    id: &DaemonId,
    cmd: &str,
    dir: &std::path::Path,
    retry_count: u32,
    daemon_env: Option<&IndexMap<String, String>>,
    resolved_ports: &[u16],
) -> CmdProbe {
    // Use the same shell as daemon run and hooks. A probe is not worth failing
    // the daemon over, so an unparseable setting degrades to the platform's own
    // shell here rather than propagating; run_once has already rejected the
    // start by then, so this only fires for a daemon whose settings changed
    // under it.
    let mut command = match resolve_shell() {
        Ok(parts) => {
            let (program, args) = parts.split_first().unwrap();
            let mut c = tokio::process::Command::new(program);
            c.shell_script(program, args, cmd);
            c
        }
        Err(e) => {
            warn!("daemon {id}: {e}; using the platform shell for this probe");
            Shell::default_for_platform().command(cmd)
        }
    };
    command
        .current_dir(dir)
        .stdout(std::process::Stdio::null())
        .stderr(std::process::Stdio::null())
        .kill_on_drop(true)
        .hide_console_window();
    apply_runtime_env(&mut command, id, retry_count, daemon_env, resolved_ports);
    let mut child = match command.spawn() {
        Ok(child) => child,
        Err(e) => {
            warn!("daemon {id}: failed to spawn command probe: {e}");
            // Return a probe whose result channel is already closed. The caller will
            // treat this the same as a probe that exited non-zero and respawn after
            // the ready_check_interval, preserving the existing retry behaviour.
            let (cancel_tx, _) = tokio::sync::oneshot::channel();
            let (_, result_rx) = tokio::sync::oneshot::channel();
            return CmdProbe {
                cancel_tx,
                result_rx,
            };
        }
    };

    let (cancel_tx, mut cancel_rx) = tokio::sync::oneshot::channel();
    let (result_tx, result_rx) = tokio::sync::oneshot::channel();

    tokio::spawn(async move {
        let status = tokio::select! {
            status = child.wait() => status,
            _ = &mut cancel_rx => {
                let mut child = child;
                let _ = child.kill().await;
                child.wait().await
            }
        };
        let _ = result_tx.send(status);
    });

    CmdProbe {
        cancel_tx,
        result_rx,
    }
}

/// Cancel an active command probe and clear its handle.
fn stop_cmd_probe_state(probe: &mut Option<CmdProbe>) {
    if let Some(p) = probe.take() {
        let _ = p.cancel_tx.send(());
    }
}

/// Spawn a detached task that kills a daemon's process group after its
/// readiness checks are exhausted, logging a failed kill instead of
/// discarding it. The returned handle is awaited before the readiness
/// failure is reported so the process group is down by then.
fn spawn_ready_fail_kill(
    id: DaemonId,
    pid: u32,
    stop_cfg: crate::config_types::StopConfig,
) -> tokio::task::JoinHandle<()> {
    tokio::spawn(async move {
        if let Err(e) = PROCS
            .kill_process_group_async(pid, stop_cfg.signal.into(), stop_cfg.timeout)
            .await
        {
            error!("daemon {id}: failed to kill pid {pid} after readiness failure: {e}");
        }
    })
}

/// Returns true if any configured readiness check can still succeed.
/// A check with no timeout is unbounded; a timed check can still succeed until its
/// deadline fires. `ready_delay` is only used as a fallback when no other check is
/// configured, so it is not counted here.
#[allow(clippy::too_many_arguments)]
fn any_ready_check_remaining(
    ready_output: Option<&ReadyOutput>,
    output_exhausted: bool,
    ready_port: Option<&ReadyPort>,
    port_exhausted: bool,
    ready_http: Option<&ReadyHttp>,
    http_exhausted: bool,
    ready_cmd: Option<&ReadyCmd>,
    cmd_exhausted: bool,
) -> bool {
    ready_output.is_some_and(|o| o.timeout.is_none() || !output_exhausted)
        || ready_port.is_some_and(|p| p.timeout.is_none() || !port_exhausted)
        || ready_http.is_some_and(|h| h.timeout.is_none() || !http_exhausted)
        || ready_cmd.is_some_and(|c| c.timeout.is_none() || !cmd_exhausted)
}

fn delay_readiness_succeeded(
    ready_notified: bool,
    has_other_ready_check: bool,
    process_exited: bool,
    process_running: bool,
) -> bool {
    !ready_notified && !has_other_ready_check && !process_exited && process_running
}

/// Terminal state recorded for a daemon run that has ended, and whether that
/// ending counts as a successful exit.
///
/// A `oneshot` daemon's whole job is to finish, so a clean exit of its own
/// accord is `Completed` rather than `Stopped` — that is what makes it
/// distinguishable from a service that is merely not running, and what lets
/// `depends` treat it as satisfied. An explicit stop is still a stop: the task
/// was interrupted, not completed.
fn terminal_exit_state(
    exit_reason: &str,
    oneshot: bool,
    exit_code: i32,
    exited_cleanly: bool,
) -> (DaemonStatus, bool) {
    match exit_reason {
        "exit" if oneshot => (DaemonStatus::Completed, true),
        "stop" | "exit" => (DaemonStatus::Stopped, exited_cleanly),
        _ => (DaemonStatus::Errored(exit_code), false),
    }
}

/// Whether a stop that arrived after a run's process was already gone should
/// leave the status its monitor settled on alone.
///
/// Only a completed task is left alone: it had already done its work, so there
/// was nothing for the stop to interrupt, and overwriting it would report a
/// failure to anyone waiting on it. Anything else — a failure above all — is
/// replaced by the stop, so the retry checker does not carry on with a task
/// the user has stopped.
/// Why the argv form of `run` cannot be started, if it cannot.
///
/// Config load checks the array as written, but a template can still render
/// the program to nothing, or to `exec`.
fn invalid_argv_program(id: &DaemonId, argv: &[String]) -> Option<String> {
    match argv.first().map(String::as_str) {
        None | Some("") => Some(format!(
            "daemon {id} has no program to run: its run array starts with an empty value"
        )),
        Some("exec") => Some(format!(
            "daemon {id} starts its run array with \"exec\"; a run array starts the program directly, so remove \"exec\""
        )),
        Some(_) => None,
    }
}

/// Program and arguments to spawn for `words` — the daemon's program followed
/// by its arguments — run through `mise x` when `mise_bin` is given.
///
/// Each word stays one argument: mise hands everything after `--` to the
/// program as it received it.
fn launch_command(words: Vec<String>, mise_bin: Option<&std::path::Path>) -> (String, Vec<String>) {
    match mise_bin {
        Some(mise_bin) => {
            let mut args = vec!["x".to_string(), "--".to_string()];
            args.extend(words);
            (mise_bin.to_string_lossy().to_string(), args)
        }
        None => {
            let mut words = words.into_iter();
            // Never empty: a shell resolves to at least its program, and an
            // empty argv is refused before this is reached.
            let program = words.next().unwrap_or_default();
            (program, words.collect())
        }
    }
}

/// Text of one line read from a daemon's output, its line ending removed.
///
/// Decoded as the log sink decodes lines, so output that is not UTF-8 is
/// logged rather than ending the read. A PTY slave's ONLCR turns `\n` into
/// `\r\n`, so every trailing `\r` goes, as the sink also strips them.
fn output_line_text(line: &[u8]) -> String {
    let line = line.strip_suffix(b"\n").unwrap_or(line);
    crate::cli::log_sink::decode_line(line)
        .trim_end_matches('\r')
        .to_string()
}

/// Send each line of `reader` to `tx` until the output ends or nobody is
/// listening.
///
/// Reading stops only there, never on a line's content: a reader that gave up
/// would stop draining the daemon's pipe or PTY, and the daemon would block,
/// or fail, on its next write.
async fn forward_output_lines<R>(mut reader: R, tx: tokio::sync::mpsc::Sender<super::OutputLine>)
where
    R: tokio::io::AsyncBufRead + Unpin,
{
    use tokio::io::AsyncBufReadExt;
    let mut line = Vec::new();
    loop {
        line.clear();
        // A read error ends the output too: a Linux PTY master reports the
        // slave closing, once the daemon is gone, as EIO rather than end of
        // file. Whatever of a last, unterminated line arrived before it is
        // still sent.
        let last = match reader.read_until(b'\n', &mut line).await {
            Ok(0) => break,
            Ok(_) => false,
            Err(_) if line.is_empty() => break,
            Err(_) => true,
        };
        let sent = tx
            .send(super::OutputLine {
                text: output_line_text(&line),
                source: super::OutputSource::Local,
            })
            .await;
        if sent.is_err() || last {
            break;
        }
    }
}

fn stop_keeps_finalized_status(status: &DaemonStatus) -> bool {
    status.is_completed()
}

/// How long a failed start waits for the daemon's output to become queryable
/// before reporting. Typically satisfied in a few dozen milliseconds; a daemon
/// that failed without printing anything waits the whole of it, so keep it
/// short.
const SINK_OUTPUT_TIMEOUT: Duration = Duration::from_millis(400);

/// Marks a daemon as having its retries managed by a foreground `run` for as
/// long as this value lives, so the background checker does not start an
/// attempt out from under it. Released on every exit from the retry loop,
/// including the early returns.
/// Counts a stop of this daemon once the stop is done, while its lock is
/// still held. See `Supervisor::stop_epochs`.
struct StopEpochGuard(DaemonId);

impl Drop for StopEpochGuard {
    fn drop(&mut self) {
        SUPERVISOR.bump_stop_epoch(&self.0);
    }
}

pub(crate) struct RetryingGuard {
    id: DaemonId,
    cancel: std::sync::Arc<std::sync::atomic::AtomicBool>,
}

impl RetryingGuard {
    /// Whether a `stop` has asked this retry sequence to end.
    fn is_cancelled(&self) -> bool {
        self.cancel.load(std::sync::atomic::Ordering::Acquire)
    }
}

impl Drop for RetryingGuard {
    fn drop(&mut self) {
        let mut retrying = SUPERVISOR
            .retrying
            .lock()
            .unwrap_or_else(|e| e.into_inner());
        // Drop this claim's flag only. Another sequence for the same daemon
        // may still be running, and it has to stay both protected from the
        // retry checker and reachable by a stop.
        if let Some(claims) = retrying.get_mut(&self.id) {
            claims.retain(|flag| !std::sync::Arc::ptr_eq(flag, &self.cancel));
            if claims.is_empty() {
                retrying.remove(&self.id);
            }
        }
    }
}

impl Supervisor {
    /// Whether a foreground `run` is already working through this daemon's
    /// retries.
    pub(crate) fn is_retrying(&self, id: &DaemonId) -> bool {
        self.retrying
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .get(id)
            .is_some_and(|claims| !claims.is_empty())
    }

    /// How many times this daemon has been stopped so far.
    pub(crate) fn stop_epoch(&self, id: &DaemonId) -> u64 {
        self.stop_epochs
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .get(id)
            .copied()
            .unwrap_or(0)
    }

    fn bump_stop_epoch(&self, id: &DaemonId) {
        *self
            .stop_epochs
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .entry(id.clone())
            .or_default() += 1;
    }

    fn mark_retrying(&self, id: &DaemonId) -> RetryingGuard {
        let cancel = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
        self.retrying
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .entry(id.clone())
            .or_default()
            .push(cancel.clone());
        RetryingGuard {
            id: id.clone(),
            cancel,
        }
    }

    /// Ask a foreground retry sequence for this daemon, if there is one, to
    /// end. A stop is a decision about the daemon, not about one of its
    /// attempts, so the attempts left must not go ahead behind it.
    pub(crate) fn cancel_retrying(&self, id: &DaemonId) {
        if let Some(claims) = self
            .retrying
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .get(id)
        {
            // Every claim, not just the newest: a start that is sleeping out a
            // backoff is as much a sequence the stop has to end as the one that
            // claimed the daemon last.
            for cancel in claims {
                cancel.store(true, std::sync::atomic::Ordering::Release);
            }
        }
    }

    /// Run a daemon, handling retries if configured
    pub async fn run(&self, opts: RunOptions) -> Result<IpcResponse> {
        self.run_inner(opts, None).await
    }

    /// Run an attempt the retry checker decided on while `stop_epoch` read
    /// `approved_at`. If the daemon has been stopped since, the attempt is
    /// abandoned instead of started.
    pub(crate) async fn run_retry(
        &self,
        opts: RunOptions,
        approved_at: u64,
    ) -> Result<IpcResponse> {
        self.run_inner(opts, Some(approved_at)).await
    }

    async fn run_inner(&self, opts: RunOptions, approved_at: Option<u64>) -> Result<IpcResponse> {
        let id = &opts.id;
        let cmd = opts.cmd.clone();

        // Clear any pending autostop for this daemon since it's being started
        {
            let mut pending = self.pending_autostops.lock().await;
            if pending.remove(id).is_some() {
                info!("cleared pending autostop for {id} (daemon starting)");
            }
        }

        // Serialize against any in-flight stop of this daemon: a stop now
        // waits for the whole process group to exit, so the Stopping window
        // can last seconds instead of milliseconds. Starting through that
        // window would collide with the dying instance (duplicate processes,
        // port conflicts). Acquiring the stop lock waits the stop out; the
        // state is re-read afterwards. The guard is owned and handed to
        // run_once, which holds it until the new daemon's Running state and
        // PID are persisted — releasing it before that point would let a
        // concurrent run pass this same check (duplicate processes) or let a
        // concurrent stop see no PID and return without stopping anything.
        let mut stop_guard = Some(self.stop_lock(id).await.lock_owned().await);
        // Checked here, under the daemon's lock, because that is what a stop
        // takes too: an approval from before the stop cannot slip past it.
        // Writing `stopped` over the record is not enough on its own, since an
        // attempt already approved would read that as a daemon free to start.
        if let Some(approved_at) = approved_at
            && self.stop_epoch(id) != approved_at
        {
            info!("daemon {id} was stopped after this retry was decided on; not starting it");
            return Ok(IpcResponse::DaemonNotRunning);
        }
        if let Some(response) = self.claim_or_defer(&opts, &mut stop_guard).await? {
            return Ok(response);
        }

        // If wait_ready is true and retry is configured, implement retry loop
        if opts.wait_ready && opts.retry.count() > 0 {
            // Claim this daemon's retries for the duration of the loop. The
            // backoff between attempts leaves the record errored with no PID,
            // which is what `check_retry` scans for, and an attempt started
            // there would leave this call reporting on a run it does not own.
            let retrying_claim = self.mark_retrying(id);
            // Use saturating_add to avoid overflow when retry = u32::MAX (infinite)
            let max_attempts = opts.retry.count().saturating_add(1);
            for attempt in 0..max_attempts {
                let mut retry_opts = opts.clone();
                retry_opts.retry_count = attempt;
                retry_opts.cmd = cmd.clone();

                // The first attempt starts under the guard held since the
                // running check above; later attempts re-acquire it so stops
                // are not locked out during the backoff sleeps.
                let mut guard = Some(match stop_guard.take() {
                    Some(guard) => guard,
                    None => self.stop_lock(id).await.lock_owned().await,
                });
                // Ownership has to be re-checked on every attempt, not just
                // the first. The backoff leaves the daemon errored with no PID,
                // which is exactly what `check_retry` looks for, so the
                // background checker can start the next attempt during the
                // sleep. Spawning another process here would replace that
                // attempt's monitor registration and leave its process running
                // unmonitored.
                if let Some(response) = self.claim_or_defer(&retry_opts, &mut guard).await? {
                    return Ok(response);
                }
                // The background retry checker may have run this attempt for
                // us and seen it succeed while we slept. Starting again would
                // repeat a task that has already done its work — for a
                // migration or a seed, repeating its side effects.
                //
                // Only after a backoff, though. A completed record on the first
                // attempt is the previous run's, and a start is defined to
                // re-run a completed oneshot; short-circuiting here would make
                // that true only for oneshots without `retry`.
                if attempt > 0
                    && let Some(daemon) = self.get_daemon(id).await
                    && daemon.status.is_completed()
                {
                    info!("daemon {id} completed while waiting to retry; not running it again");
                    return Ok(IpcResponse::DaemonReady { daemon });
                }
                // A stop that arrived during the backoff ends the sequence.
                // Without this the loop would start the next attempt on a
                // daemon the user has just stopped, and the stop would look
                // like it had done nothing.
                if retrying_claim.is_cancelled() {
                    info!("daemon {id} was stopped while waiting to retry; abandoning its retries");
                    return Ok(IpcResponse::DaemonFailed {
                        error: "stopped while retrying".to_string(),
                    });
                }
                let Some(guard) = guard else {
                    // Only the deferring paths take the guard, and each of
                    // those returned above.
                    return Ok(IpcResponse::DaemonAlreadyRunning);
                };
                let result = self.run_once(retry_opts, guard).await?;

                match result {
                    IpcResponse::DaemonReady { daemon } => {
                        return Ok(IpcResponse::DaemonReady { daemon });
                    }
                    IpcResponse::DaemonFailedWithCode {
                        exit_code,
                        resolved_ports,
                    } => {
                        if attempt < opts.retry.count() {
                            // `run_once` reports failure the moment the process
                            // exits, but its monitor finalizes the record only
                            // after draining the process's remaining output.
                            // Until then the record still names this attempt's
                            // PID, and the next attempt's ownership check would
                            // read its own dead predecessor as a competing run
                            // and abandon the retries that are left.
                            let attempt_pid = self.get_daemon(id).await.and_then(|d| d.pid);
                            self.wait_for_exit_finalized(id, attempt_pid).await;
                            let backoff_secs = 2u64.saturating_pow(attempt).min(3600);
                            info!(
                                "daemon {id} failed (attempt {}/{}), retrying in {}s",
                                attempt + 1,
                                max_attempts,
                                backoff_secs
                            );
                            fire_hook(
                                HookType::OnRetry,
                                id.clone(),
                                opts.dir.0.clone(),
                                attempt + 1,
                                opts.env.clone(),
                                resolved_ports,
                                vec![],
                            )
                            .await;
                            // Slept in slices so a stop arriving during a
                            // long backoff — they grow to an hour — is acted
                            // on when it arrives rather than when the sleep
                            // happens to end.
                            let backoff_deadline =
                                tokio::time::Instant::now() + Duration::from_secs(backoff_secs);
                            while tokio::time::Instant::now() < backoff_deadline
                                && !retrying_claim.is_cancelled()
                            {
                                let remaining = backoff_deadline - tokio::time::Instant::now();
                                time::sleep(remaining.min(Duration::from_millis(200))).await;
                            }
                            continue;
                        } else {
                            info!("daemon {id} failed after {max_attempts} attempts");
                            return Ok(IpcResponse::DaemonFailedWithCode {
                                exit_code,
                                resolved_ports,
                            });
                        }
                    }
                    other => return Ok(other),
                }
            }
        }

        // No retry or wait_ready is false
        let guard = match stop_guard.take() {
            Some(guard) => guard,
            None => self.stop_lock(id).await.lock_owned().await,
        };
        self.run_once(opts, guard).await
    }

    /// Wait for a just-failed attempt's monitor to write its terminal state,
    /// clearing the PID from the record.
    ///
    /// Bounded a little beyond the monitor's own five-second output drain, the
    /// longest it can hold the record after the process has gone. Giving up
    /// early is safe: the ownership check that follows simply sees a PID and
    /// defers, which is what it would have done anyway.
    ///
    /// `pid` names the run being waited for, so a record that has moved on to
    /// another run is not mistaken for this one still finishing.
    async fn wait_for_exit_finalized(&self, id: &DaemonId, pid: Option<u32>) {
        let deadline = tokio::time::Instant::now() + Duration::from_secs(8);
        loop {
            match self.get_daemon(id).await {
                Some(daemon) if pid.map_or(daemon.pid.is_some(), |pid| daemon.pid == Some(pid)) => {
                }
                _ => return,
            }
            if tokio::time::Instant::now() >= deadline {
                debug!("daemon {id}: previous attempt has not finalized yet; continuing anyway");
                return;
            }
            time::sleep(Duration::from_millis(50)).await;
        }
    }

    /// Decide whether this start may take the daemon's record, or must stand
    /// down because a live run already owns it.
    ///
    /// Returns `Some(response)` when the caller must report that run's outcome
    /// instead of spawning a second process, and `None` when the record is free
    /// (including after a forced stop of the previous instance).
    ///
    /// `stop_guard` is released before an in-flight oneshot is awaited: that
    /// wait lasts as long as the task does, and holding the lock would block a
    /// stop of the very run being waited on.
    async fn claim_or_defer(
        &self,
        opts: &RunOptions,
        stop_guard: &mut Option<tokio::sync::OwnedMutexGuard<()>>,
    ) -> Result<Option<IpcResponse>> {
        let id = &opts.id;
        let Some(daemon) = self.get_daemon(id).await else {
            return Ok(None);
        };
        // Entering a directory does not re-run a finished task, at any level of
        // the dependency graph — the layout this exists for reaches the task
        // through a service's `depends`, not by naming it. Decided here rather
        // than in the client because this is the authoritative state: the state
        // file lags it by up to the flush interval, which is exactly the window
        // a second entry lands in after the task completes.
        // `opts.oneshot` rather than the record's: the request carries what
        // config says now, while the stored flag is only refreshed by a run, so
        // a daemon that used to be a task would otherwise stay skipped forever
        // after being turned into a service.
        if opts.on_directory_enter && opts.oneshot && daemon.status.is_completed() {
            debug!("daemon {id} already completed; directory entry leaves it alone");
            return Ok(Some(IpcResponse::DaemonReady { daemon }));
        }
        // Stopping is treated as "not running": the monitoring task will clean
        // it up. Only a live PID under a non-terminal status blocks a start.
        if daemon.status.is_stopping() || daemon.status.is_stopped() || daemon.status.is_completed()
        {
            return Ok(None);
        }
        let Some(pid) = daemon.pid else {
            return Ok(None);
        };
        if opts.force {
            self.stop_locked(id).await?;
            info!("run: stop completed for daemon {id}");
            return Ok(None);
        }
        if daemon.oneshot && opts.wait_ready {
            // An in-flight oneshot has not done its work yet, so reporting
            // "already running" would let dependents start against the state
            // the task is still establishing. Wait for the run already under
            // way instead.
            info!("daemon {id} is an in-flight oneshot (pid {pid}); waiting for it to finish");
            drop(stop_guard.take());
            return Ok(Some(
                self.await_running_oneshot(id, opts.oneshot_wait, pid).await,
            ));
        }
        // A record can name a PID that has already exited: `stop` leaves the
        // terminal state to a monitor that still owns the daemon, and that
        // monitor writes it only after draining the process's output. Rejecting
        // a start against a dead PID would fail an ordinary stop-then-start for
        // the length of that drain, so confirm the process is really there
        // before refusing. The oneshot branch above deliberately comes first: a
        // task whose process has exited is about to be recorded as completed,
        // and starting a second copy of it is exactly what waiting prevents.
        PROCS.refresh_pids(&[pid]);
        if !PROCS.is_running(pid) {
            debug!(
                "daemon {id}: record still names pid {pid}, which has exited; its monitor has not finalized yet"
            );
            return Ok(None);
        }
        warn!("daemon {id} already running with pid {pid}");
        Ok(Some(IpcResponse::DaemonAlreadyRunning))
    }

    /// Wait for a oneshot that is already running to reach a terminal state,
    /// and report it as if this call had started the task itself.
    ///
    /// Polls the state file because the terminal state is written by the
    /// monitoring task of the *other* run; this call has no readiness channel
    /// of its own to await.
    async fn await_running_oneshot(
        &self,
        id: &DaemonId,
        wait: Option<OneshotWait>,
        watched_pid: u32,
    ) -> IpcResponse {
        let interval = settings().supervisor_ready_check_interval();
        // The caller resolved this from the project's settings and sent it, so
        // both processes wait exactly as long. Falling back to this process's
        // own settings would read the directory the supervisor happens to have
        // started in, where a project's `oneshot_timeout` is not visible — and
        // the shorter of the two deadlines would silently win, releasing
        // dependents while the task was still running.
        //
        // `None` here means the setting asked for no limit, so there is no
        // deadline to reach rather than a distant one.
        // One deadline for the whole wait, retries and backoffs included.
        // `oneshot_timeout` is documented as the longest `pitchfork start` will
        // wait, and the client bounds its own request by the same value without
        // restarting it, so a per-attempt budget here would both break that
        // promise — unboundedly, with infinite retries — and put the two sides
        // back to disagreeing about when one task has gone on too long.
        let deadline = wait
            .unwrap_or_else(|| settings().supervisor_oneshot_wait())
            .duration()
            .map(|d| tokio::time::Instant::now() + d);
        // Which run this wait is reporting on. A terminal state is only that
        // run's while the record still names its PID or names none at all; once
        // another PID appears, something else has started the task and the
        // outcome that follows belongs to that run, not this one. Following the
        // handoff keeps the answer useful to a dependent — it still learns
        // whether the task succeeded — without quietly attributing an unrelated
        // run's failure to the one it asked about.
        let mut watched_pid = watched_pid;
        loop {
            let Some(daemon) = self.get_daemon(id).await else {
                return IpcResponse::DaemonNotFound;
            };
            if let Some(current) = daemon.pid
                && current != watched_pid
            {
                info!(
                    "daemon {id}: the run being waited on (pid {watched_pid}) was replaced by pid {current}; following it"
                );
                watched_pid = current;
            }
            match &daemon.status {
                DaemonStatus::Completed => {
                    info!("daemon {id}: the in-flight oneshot completed");
                    return IpcResponse::DaemonReady { daemon };
                }
                DaemonStatus::Errored(code) => {
                    // A failed attempt is persisted before the in-flight `run`
                    // sleeps out its backoff, so an errored record with
                    // attempts left is a gap between tries rather than the
                    // result. Same condition `check_retry` uses to decide
                    // whether another attempt is still owed.
                    if daemon.retry.count() > 0 && daemon.retry_count < daemon.retry.count() {
                        debug!(
                            "daemon {id}: in-flight oneshot failed attempt {} of {}; still waiting",
                            daemon.retry_count + 1,
                            daemon.retry.count() + 1
                        );
                    } else {
                        // -1 records an unobservable exit code; the caller
                        // renders `None` as a plain failure rather than
                        // "exit code -1".
                        let exit_code = Some(*code).filter(|c| *c != -1);
                        return IpcResponse::DaemonFailedWithCode {
                            exit_code,
                            resolved_ports: daemon.resolved_port.clone(),
                        };
                    }
                }
                DaemonStatus::Failed(error) => {
                    return IpcResponse::DaemonFailed {
                        error: error.clone(),
                    };
                }
                DaemonStatus::Stopped => {
                    // Stopped, not completed: the task was interrupted, so it
                    // never established what its dependents are waiting for.
                    warn!("daemon {id}: the in-flight oneshot was stopped before completing");
                    return IpcResponse::DaemonFailedWithCode {
                        exit_code: None,
                        resolved_ports: daemon.resolved_port.clone(),
                    };
                }
                DaemonStatus::Running | DaemonStatus::Waiting | DaemonStatus::Stopping => {}
            }
            if deadline.is_some_and(|deadline| tokio::time::Instant::now() >= deadline) {
                warn!("daemon {id}: gave up waiting for the in-flight oneshot to finish");
                // Reported as a failure rather than as "already running": the
                // batch start path only counts a result carrying an exit code
                // as failed, so anything else would let dependents start
                // against a task that never finished. 124 is the code a
                // readiness timeout already uses.
                return IpcResponse::DaemonFailedWithCode {
                    exit_code: Some(124),
                    resolved_ports: Vec::new(),
                };
            }
            time::sleep(interval).await;
        }
    }

    /// Run a daemon once (single attempt).
    ///
    /// `stop_guard` is this daemon's stop lock, acquired by `run` before the
    /// already-running check. It is held through spawning until the Running
    /// state and PID are persisted (or an early failure returns), then dropped
    /// before the potentially unbounded readiness wait.
    pub(crate) async fn run_once(
        &self,
        opts: RunOptions,
        stop_guard: tokio::sync::OwnedMutexGuard<()>,
    ) -> Result<IpcResponse> {
        let id = &opts.id;
        let original_cmd = opts.cmd.clone(); // Save original command for persistence

        // Create channel for readiness notification if wait_ready is true
        let (ready_tx, ready_rx) = if opts.wait_ready {
            let (tx, rx) = oneshot::channel();
            (Some(tx), Some(rx))
        } else {
            (None, None)
        };

        // Check port availability and apply auto-bump if configured
        let expected_ports = opts
            .port
            .as_ref()
            .map(|p| p.expect.clone())
            .unwrap_or_default();
        let (resolved_ports, effective_ready_port) = if !expected_ports.is_empty() {
            let port_cfg = opts.port.as_ref().unwrap();
            match check_ports_available(
                &expected_ports,
                port_cfg.auto_bump(),
                port_cfg.max_bump_attempts(),
            )
            .await
            {
                Ok(resolved) => {
                    let ready_port = if let Some(configured_port) =
                        opts.ready_port.as_ref().and_then(|p| p.as_port())
                    {
                        Some(resolve_configured_ready_port(
                            configured_port,
                            &expected_ports,
                            &resolved,
                        ))
                    } else if opts.ready_output.is_none()
                        && opts.ready_http.is_none()
                        && opts.ready_cmd.is_none()
                        && opts.ready_delay.is_none()
                    {
                        // No other ready check configured — use the first expected port as a
                        // TCP port readiness check so the daemon is considered ready once it
                        // starts listening.  Skip port 0 (ephemeral port request).
                        resolved.first().copied().filter(|&p| p != 0)
                    } else {
                        // Another ready check is configured (output/http/cmd/delay).
                        // Don't add an implicit TCP port check — it could race and fire
                        // before the daemon has produced any output.
                        None
                    };
                    info!("daemon {id}: ports {expected_ports:?} resolved to {resolved:?}");
                    (resolved, ready_port)
                }
                Err(e) => {
                    error!("daemon {id}: port check failed: {e}");
                    // Convert PortError to structured IPC response
                    if let Some(port_error) = e.downcast_ref::<PortError>() {
                        match port_error {
                            PortError::InUse { port, process, pid } => {
                                return Ok(IpcResponse::PortConflict {
                                    port: *port,
                                    process: process.clone(),
                                    pid: *pid,
                                });
                            }
                            PortError::NoAvailablePort {
                                start_port,
                                attempts,
                            } => {
                                return Ok(IpcResponse::NoAvailablePort {
                                    start_port: *start_port,
                                    attempts: *attempts,
                                });
                            }
                        }
                    }
                    return Ok(IpcResponse::DaemonFailed {
                        error: e.to_string(),
                    });
                }
            }
        } else {
            // When ready_port is set without expected_port, check that the port
            // is not already occupied.  If another process is listening on it,
            // the TCP readiness probe would immediately succeed and pitchfork
            // would falsely consider the daemon ready — routing proxy traffic to
            // the wrong process.
            if let Some(port) = opts.ready_port.as_ref().and_then(|p| p.as_port())
                && port > 0
                && let Some((pid, process)) = detect_port_conflict(port).await
            {
                return Ok(IpcResponse::PortConflict { port, process, pid });
            }
            (
                Vec::new(),
                opts.ready_port.as_ref().and_then(|p| p.as_port()),
            )
        };

        // The program and arguments that start the daemon, before any mise
        // wrapping.
        // The program and arguments that start the daemon, before any mise
        // wrapping, and the script for the shell when `run` is a string.
        let (mut words, script) = if opts.no_shell {
            // The argv form of `run`: started as written, with no shell to
            // reinterpret quotes, `%`, `&` or anything else in the arguments.
            if let Some(error) = invalid_argv_program(id, &original_cmd) {
                return Ok(IpcResponse::DaemonFailed { error });
            }
            (original_cmd.clone(), None)
        } else {
            // Resolve the shell for this platform into program + args. The run
            // script is passed verbatim as the final argument, avoiding the lossy
            // split->join round-trip that previously mangled $VAR/glob expansion.
            let words = match resolve_shell() {
                Ok(parts) => parts,
                Err(error) => return Ok(IpcResponse::DaemonFailed { error }),
            };
            // Use the original run string verbatim; fall back to joining cmd for
            // ad-hoc commands (e.g. `pitchfork run -- cmd args`) that have no run string.
            // We don't prepend `exec` because it breaks compound commands (e.g. `exec a && b`
            // silently drops `b`). Users can add `exec` themselves in the run string.
            let script = opts
                .run
                .clone()
                .unwrap_or_else(|| shell_words::join(&original_cmd));
            (words, Some(script))
        };

        let mise_bin = if opts.mise.unwrap_or(settings().general.mise) {
            let mise_bin = settings().resolve_mise_bin();
            if mise_bin.is_none() {
                warn!("daemon {id}: mise=true but mise binary not found, running without mise");
            }
            mise_bin
        } else {
            None
        };
        // Started directly, the shell gets its script from `shell_script`.
        // Under mise it goes in as an ordinary argument: mise starts the shell
        // itself, re-quoting each argument, so the raw command line cmd.exe
        // needs for a script with `"` cannot reach it that way.
        let script = match &mise_bin {
            Some(mise_bin) => {
                info!(
                    "daemon {id}: wrapping command with mise ({})",
                    mise_bin.display()
                );
                words.extend(script);
                None
            }
            None => script,
        };
        let (program, args) = launch_command(words, mise_bin.as_deref());
        #[cfg(unix)]
        let run_identity = match resolve_effective_run_identity(opts.user.as_deref()) {
            Ok(identity) => identity,
            Err(e) => {
                return Ok(IpcResponse::DaemonFailed {
                    error: e.to_string(),
                });
            }
        };
        info!("run: spawning daemon {id} with {program} {args:?} {script:?}");

        // Allocate PTY if configured
        #[cfg(unix)]
        let pty_pair = if opts.pty.unwrap_or(false) {
            match super::pty::openpty() {
                Ok(pair) => {
                    info!("daemon {id}: allocated PTY (pty = true)");
                    Some(pair)
                }
                Err(e) => {
                    warn!("daemon {id}: failed to allocate PTY, falling back to pipes: {e}");
                    None
                }
            }
        } else {
            None
        };

        // Output reaches the monitoring task either from readers this process
        // owns or, when a sink owns the stream, relayed over IPC. The channel is
        // created here rather than in that task so it exists before the sink
        // starts: a daemon whose very first line matches its readiness pattern
        // would otherwise have the match reported with nowhere to deliver it.
        let (output_tx, output_rx) = tokio::sync::mpsc::channel::<super::OutputLine>(256);
        let mut output_relay = None;

        // Set up out-of-process capture before building the command, so the
        // daemon can be handed the pipe's write end directly.
        let mut sink_pipe = None;
        let mut sink_writer = None;
        let mut sink_child = None;
        if super::log_sink::is_supported(&opts) {
            let log_format = opts
                .log_format
                .clone()
                .unwrap_or_else(|| settings().logs.log_format.clone());
            let watch_for = super::log_sink::WatchFor::from_opts(id, &opts);
            // The token ties this attempt's sink to this attempt's channel, so
            // a sink still draining a previous attempt cannot report into it.
            let relay_token = if watch_for.is_empty() {
                0
            } else {
                let relay = super::log_sink::OutputRelay::register(id, output_tx.clone());
                let token = relay.token();
                output_relay = Some(relay);
                token
            };
            match super::log_sink::SinkPipe::new(log_format, watch_for, relay_token) {
                Ok((pipe, writer)) => match pipe.start(id) {
                    Ok(child) => {
                        sink_child = Some(super::log_sink::PendingSink::new(child));
                        sink_pipe = Some(pipe);
                        sink_writer = Some(writer);
                    }
                    Err(e) => {
                        warn!("could not start log sink for {id}, capturing in-process: {e}");
                    }
                },
                Err(e) => {
                    // Fall back to in-process capture rather than refusing to
                    // start the daemon.
                    warn!("could not create log pipe for {id}, capturing in-process: {e}");
                }
            }
        }

        let mut cmd = tokio::process::Command::new(&program);

        #[cfg(unix)]
        if let Some(ref pair) = pty_pair {
            // PTY mode: connect both stdout and stderr to the slave PTY.
            // The child uses the slave for stdin/stdout/stderr, and we read
            // output from the master.
            let slave_file = std::fs::File::from(
                pair.slave
                    .try_clone()
                    .map_err(|e| miette::miette!("failed to dup slave PTY fd: {e}"))?,
            );
            cmd.stdin(std::process::Stdio::from(slave_file.try_clone().map_err(
                |e| miette::miette!("failed to clone slave PTY fd for stdin: {e}"),
            )?));
            cmd.stdout(std::process::Stdio::from(slave_file.try_clone().map_err(
                |e| miette::miette!("failed to clone slave PTY fd for stdout: {e}"),
            )?));
            cmd.stderr(std::process::Stdio::from(slave_file));
        } else if let Some(writer) = sink_writer.take() {
            // Capture belongs to a sibling sink process, so the daemon writes
            // to a pipe this process does not read. See supervisor::log_sink.
            let dup = writer
                .try_clone()
                .map_err(|e| miette::miette!("failed to dup log pipe for stderr: {e}"))?;
            cmd.stdout(std::process::Stdio::from(writer))
                .stderr(std::process::Stdio::from(dup));
        } else {
            cmd.stdout(std::process::Stdio::piped())
                .stderr(std::process::Stdio::piped());
        }

        #[cfg(not(unix))]
        if let Some(writer) = sink_writer.take() {
            let dup = writer
                .try_clone()
                .map_err(|e| miette::miette!("failed to dup log pipe for stderr: {e}"))?;
            cmd.stdout(std::process::Stdio::from(writer))
                .stderr(std::process::Stdio::from(dup));
        } else {
            cmd.stdout(std::process::Stdio::piped())
                .stderr(std::process::Stdio::piped());
        }

        match &script {
            Some(script) => cmd.shell_script(&program, &args, script),
            None => cmd.args(&args),
        };
        cmd.current_dir(&opts.dir).hide_console_window();

        #[cfg(unix)]
        if pty_pair.is_none() {
            cmd.stdin(std::process::Stdio::null());
        }

        #[cfg(not(unix))]
        cmd.stdin(std::process::Stdio::null());

        // Before the runtime env, so a daemon's own `env` entries win.
        #[cfg(unix)]
        apply_identity_env(&mut cmd, &run_identity);
        apply_runtime_env(
            &mut cmd,
            id,
            opts.retry_count,
            opts.env.as_ref(),
            &resolved_ports,
        );

        // Inject proxy-related environment variables
        inject_proxy_env(&mut cmd, &daemon_proxy_host(&opts).await);

        #[cfg(unix)]
        {
            let run_identity = run_identity.clone();
            let use_pty = pty_pair.is_some();
            unsafe {
                cmd.pre_exec(move || {
                    nix::unistd::setsid().map_err(nix_to_io_error)?;

                    // When using a PTY, set the slave as the controlling terminal.
                    // The slave FD has already been dup'd onto stdin/stdout/stderr
                    // by tokio, so we can use stdin (fd 0) for TIOCSCTTY.
                    if use_pty {
                        let ret = libc::ioctl(0, libc::TIOCSCTTY as libc::c_ulong, 0);
                        if ret < 0 {
                            // Non-fatal: the process can still run without
                            // a controlling terminal.
                            #[cfg(target_os = "linux")]
                            eprintln!(
                                "pitchfork: TIOCSCTTY failed: {}",
                                std::io::Error::last_os_error()
                            );
                        }
                    }

                    apply_run_identity(&run_identity)?;
                    Ok(())
                });
            }
        }

        // Timestamp the run so a failed start can wait for this attempt's output
        // specifically, rather than seeing an earlier attempt's.
        let spawn_time = chrono::Local::now();
        // A sink is already running at this point. Both bail-outs below have to
        // reap it explicitly: dropping the handle only reaps on a best-effort
        // basis, and run_once runs once per retry attempt, so a daemon that
        // consistently fails to spawn would otherwise accumulate sinks.
        // A failed spawn returns here; the sink is terminated by PendingSink.
        let mut child = cmd.spawn().into_diagnostic()?;
        // A process now exists, which is exactly what `last_cron_run` records.
        // Written here rather than from the watcher's view of the response
        // because that view cannot tell a start that failed before spawning
        // from one that spawned and exited before its PID could be read: a
        // port conflict, an unresolvable shell and an instant exit all report
        // `DaemonFailed`. Only the last of those ran, and a cron job that
        // fails that fast is precisely the one whose timing a user needs.
        // Ordered before the `Running` upsert below, which inherits it.
        //
        // Written synchronously rather than left to the background flush, for
        // the same reason `last_cron_triggered` is: a supervisor that dies in
        // the window between the two would come back with no record that this
        // run happened, and for a short-lived job that window is as long as
        // the job itself. A scheduled spawn is rare enough -- at most one per
        // `cron_check_interval` -- for the extra write to cost nothing.
        if opts.cron_started {
            let mut state_file = self.state_file.lock().await;
            if state_file.set_last_cron_run(id, spawn_time)
                && let Err(e) = state_file.write()
            {
                error!("failed to persist last_cron_run for daemon {id}: {e}");
            }
        }
        let pid = match child.id() {
            Some(p) => p,
            None => {
                warn!("Daemon {id} exited before PID could be captured");
                // Unlike a daemon that never started, this one ran and may have
                // said why it gave up, and its output is the only diagnosis
                // available. Its write end is already closed, so the sink is on
                // its way to end of file: let it finish writing before reporting,
                // then reap whatever is left of it.
                if sink_child.is_some() {
                    super::log_sink::wait_for_output(id, spawn_time, SINK_OUTPUT_TIMEOUT).await;
                }
                return Ok(IpcResponse::DaemonFailed {
                    error: "Process exited immediately".to_string(),
                });
            }
        };
        info!("started daemon {id} with pid {pid}");
        PROCS.refresh_pids(&[pid]);
        // Register the daemon as monitored BEFORE persisting the Running
        // state. The orphan reconciler treats any running, unmonitored PID
        // as an orphan; if the state became visible first, a concurrent
        // reconciliation pass could adopt — or under the kill policy,
        // terminate — a daemon that was just legitimately started. The RAII
        // guard unregisters on any early-error path below and is otherwise
        // handed to the monitoring task.
        let monitored_guard = super::adopt::MonitoredGuard::register(id.clone(), pid);
        let monitor_token = monitored_guard.token();

        // Hand the retained read end to a sink and keep one running for as long
        // as this daemon is monitored.
        let using_sink = sink_pipe.is_some();
        // Take the sink out of the guard only once there is a pipe to supervise
        // it with, so it is never left running unsupervised.
        if let Some(pipe) = sink_pipe.take()
            && let Some(child) = sink_child.as_mut().and_then(|pending| pending.take())
        {
            pipe.supervise(id.clone(), monitor_token, child);
        }
        // The attempt's actual resolved ports, captured before the upsert
        // moves them into state. Hooks, readiness probes, and the failure
        // response must reflect this attempt: the state merge keeps the
        // existing resolved_port when an update is empty, so a no-port
        // attempt would otherwise inherit a previous run's stale ports
        // through the upserted record.
        let attempt_resolved_ports = resolved_ports.clone();
        let daemon = self
            .upsert_daemon(
                UpsertDaemonOpts::from_run_options(&opts, DaemonStatus::Running)
                    .set(|o| {
                        o.pid = Some(pid);
                        o.cmd = Some(original_cmd);
                        o.ready_port = effective_ready_port.map(|p| ReadyPort {
                            port: Some(p),
                            template: None,
                            timeout: opts.ready_port.as_ref().and_then(|rp| rp.timeout),
                        });
                        o.port = crate::config_types::PortConfig::from_parts(
                            expected_ports,
                            opts.port.as_ref().map(|p| p.bump).unwrap_or_default(),
                        );
                        o.resolved_port = Some(resolved_ports);
                    })
                    .build(),
            )
            .await?;

        // Running state and PID are now persisted: concurrent run/stop calls
        // observe a running daemon and behave correctly, so release the stop
        // lock rather than holding it through the readiness wait below, which
        // can take arbitrarily long.
        drop(stop_guard);

        let id_clone = id.clone();
        // A oneshot is ready only when its process exits 0, so no readiness
        // check may run alongside it — one that fired first would report the
        // task ready before it had done its work, and would suppress the
        // completion notification entirely. Config load rejects explicit
        // `ready_*` fields and the client clears CLI overrides, but the
        // implicit port check is derived here from `port.expect`, so the
        // suppression has to happen here rather than being trusted to callers.
        let ready_delay = (!opts.oneshot).then_some(opts.ready_delay).flatten();
        let ready_output = (!opts.oneshot).then(|| opts.ready_output.clone()).flatten();
        let ready_http = (!opts.oneshot).then(|| opts.ready_http.clone()).flatten();
        let ready_port = (!opts.oneshot).then_some(effective_ready_port).flatten();
        let implicit_ready_port = ready_port.map(|p| ReadyPort {
            port: Some(p),
            template: None,
            timeout: None,
        });
        let ready_port_config = (!opts.oneshot)
            .then(|| opts.ready_port.clone())
            .flatten()
            .or(implicit_ready_port);
        let ready_cmd = (!opts.oneshot).then(|| opts.ready_cmd.clone()).flatten();
        let daemon_dir = opts.dir.0.clone();
        let hook_retry_count = opts.retry_count;
        let hook_retry = opts.retry;
        let hook_daemon_env = opts.env.clone();
        // Ports of THIS attempt, snapshotted before the monitor starts: a retry
        // or restart may replace state.resolved_port before a hook task runs.
        // Sourced from the attempt's local value, not the upserted record —
        // the state merge inherits stale ports for a no-port attempt.
        let hook_resolved_ports = attempt_resolved_ports.clone();
        let readiness_daemon_env = opts.env.clone();
        let readiness_resolved_ports = attempt_resolved_ports.clone();
        let on_output_hook = opts.on_output_hook.clone();
        // Whether this daemon has any port-related config — used to skip the
        // active_port detection task for daemons that never bind a port (e.g. `sleep 60`).
        // When the proxy is enabled, only detect active_port for daemons that are
        // actually referenced by a registered slug, rather than blanket-polling every
        // daemon (which wastes ~7.5 s of listeners::get_all() calls per port-less daemon).
        let has_port_config = opts.port.as_ref().is_some_and(|p| !p.expect.is_empty())
            || (settings().proxy.enable && is_daemon_slug_target(id));
        // When the ready_port check succeeds on the first resolved port we can
        // set active_port directly instead
        // of spawning detect_and_store_active_port (which relies on
        // listeners::get_all() + process-tree traversal and is unreliable on
        // Windows where Git Bash PID mapping can break descendant lookups).
        let daemon_pid = pid;

        // Prepare output readers before spawning the monitoring task.
        // In PTY mode, we read from the PTY master FD.
        // In pipe mode, we read from separate stdout/stderr pipes.
        #[cfg(unix)]
        let pty_reader = pty_pair.map(|p| {
            tokio::io::BufReader::new(tokio::fs::File::from_std(std::fs::File::from(p.master)))
        });
        #[cfg(not(unix))]
        let pty_reader: Option<tokio::io::BufReader<tokio::fs::File>> = None;
        let stdout_reader = if pty_reader.is_none() {
            child.stdout.take().map(tokio::io::BufReader::new)
        } else {
            None
        };
        let stderr_reader = if pty_reader.is_none() {
            child.stderr.take().map(tokio::io::BufReader::new)
        } else {
            None
        };

        if !using_sink
            && pty_reader.is_none()
            && (stdout_reader.is_none() || stderr_reader.is_none())
        {
            error!("Failed to capture stdout/stderr for daemon {id}");
        }

        tokio::spawn(async move {
            let id = id_clone;
            // Registered before the Running upsert above; unregisters when
            // this monitoring task ends.
            let _monitored_guard = monitored_guard;
            // Likewise for sink-relayed output: dropping this stops the
            // supervisor delivering into a channel nobody is reading. Dropped
            // explicitly once the daemon exits, before the drain below.
            let output_relay = output_relay;

            // Merge all output sources (PTY master OR stdout+stderr, or a
            // sink's IPC reports) into a single channel.
            let mut output_rx = output_rx;

            if let Some(reader) = pty_reader {
                // PTY mode: single merged stream from the master.
                // output_tx is moved into the spawn; when the reader ends the
                // channel closes automatically.
                tokio::spawn(forward_output_lines(reader, output_tx));
            } else {
                // Pipe mode: stdout and stderr are merged into the same channel.
                // Both `ready_output` and `on_output_hook` patterns match against
                // lines from either stream, which is the expected behavior (a
                // "server ready" message may appear on stderr in some tools).
                if let Some(stdout) = stdout_reader {
                    tokio::spawn(forward_output_lines(stdout, output_tx.clone()));
                }
                if let Some(stderr) = stderr_reader {
                    tokio::spawn(forward_output_lines(stderr, output_tx.clone()));
                }
                // Drop the last sender so the channel closes when all readers
                // finish. The relay holds its own clone, so a sink's reports
                // still have somewhere to go after these end.
                drop(output_tx);
            }
            let log_store = Arc::clone(&LOG_STORE);
            let log_format = opts
                .log_format
                .clone()
                .unwrap_or_else(|| crate::settings::settings().logs.log_format.clone());
            let parse_line = move |line: &str| crate::log_parse::parse(line, &log_format);

            const LOG_BATCH_SIZE: usize = 100;
            const LOG_FLUSH_INTERVAL: Duration = Duration::from_millis(100);
            let mut log_buffer: Vec<crate::log_parse::ParsedLog> =
                Vec::with_capacity(LOG_BATCH_SIZE);
            let mut log_flush_interval = tokio::time::interval(LOG_FLUSH_INTERVAL);
            log_flush_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);

            let flush_logs =
                |buffer: &mut Vec<crate::log_parse::ParsedLog>| -> Option<tokio::task::JoinHandle<()>> {
                    if buffer.is_empty() {
                        return None;
                }
                let store = Arc::clone(&log_store);
                let id = id.clone();
                let batch = std::mem::take(buffer);
                Some(tokio::task::spawn_blocking(move || {
                    if let Err(e) = store.append_structured_batch(&id, &batch) {
                        error!("Failed to write batch to log for daemon {id}: {e}");
                    }
                }))
            };

            // SQLite WAL mode provides automatic durability; no explicit flush needed.

            // Setup readiness checking
            let mut ready_notified = false;
            // Set when a oneshot's process exits 0. Its readiness *is* its
            // completion, so the notification is held back until the
            // `completed` state has been persisted — a caller that returns
            // from `pitchfork start` must not still see the daemon running.
            let mut oneshot_completion_pending = false;
            let mut ready_tx = ready_tx;
            let ready_pattern = ready_output
                .as_ref()
                .and_then(|o| get_or_compile_regex(&o.pattern));
            // Track whether we've already spawned the active_port detection task
            let mut active_port_spawned = false;

            // Validate on_output config early; discard the hook on any error so
            // a bad regex does not silently fall through to the (None, None) => true
            // match arm and fire on every line.
            let on_output_hook = match on_output_hook {
                Some(ref hook) => match hook.validate(id.name()) {
                    Ok(()) => on_output_hook,
                    Err(e) => {
                        error!("{e}");
                        None
                    }
                },
                None => None,
            };

            // Compile the regex pattern after validation so we only attempt this
            // when the hook is known-good (validate() already checked the syntax).
            let on_output_pattern: Option<regex::Regex> = on_output_hook
                .as_ref()
                .and_then(|h| h.regex.as_deref().and_then(get_or_compile_regex));
            let on_output_debounce = on_output_hook
                .as_ref()
                .map(|h| h.debounce_duration())
                .unwrap_or(Duration::from_millis(1000));
            // Last time the on_output hook fired; None means it has never fired.
            let mut on_output_last_fired: Option<std::time::Instant> = None;

            let mut delay_timer =
                ready_delay.map(|secs| Box::pin(time::sleep(Duration::from_secs(secs))));

            // Track exhaustion of timed checks
            let mut http_exhausted = false;
            let mut cmd_exhausted = false;
            let mut port_exhausted = false;
            let mut output_exhausted = false;

            // Get settings for intervals
            let s = settings();
            let ready_check_interval = s.supervisor_ready_check_interval();
            let http_client_timeout = s.supervisor_http_client_timeout();

            // Setup output readiness check deadline
            let mut output_deadline = ready_output
                .as_ref()
                .and_then(|o| o.timeout)
                .map(|d| Box::pin(time::sleep(d)));

            // Setup HTTP readiness check interval and deadline
            let mut http_check_interval = ready_http
                .as_ref()
                .map(|_| tokio::time::interval(ready_check_interval));
            let mut http_deadline = ready_http
                .as_ref()
                .and_then(|h| h.timeout)
                .map(|d| Box::pin(time::sleep(d)));
            let http_client = ready_http.as_ref().map(|_| {
                reqwest::Client::builder()
                    .timeout(http_client_timeout)
                    .build()
                    .unwrap_or_default()
            });

            // Setup TCP port readiness check interval and deadline
            let mut port_check_interval =
                ready_port.map(|_| tokio::time::interval(ready_check_interval));
            let mut port_deadline = ready_port_config
                .as_ref()
                .and_then(|p| p.timeout)
                .map(|d| Box::pin(time::sleep(d)));

            // Setup command readiness check state. Probes are spawned one at a time;
            // a non-zero result triggers a respawn delay, and a timeout stops the probe.
            let mut cmd_probe: Option<CmdProbe> = None;
            let mut cmd_respawn_delay: Option<_> = None;
            let mut cmd_deadline = ready_cmd
                .as_ref()
                .and_then(|c| c.timeout)
                .map(|d| Box::pin(time::sleep(d)));
            if let Some(ref cmd) = ready_cmd {
                cmd_probe = Some(spawn_cmd_probe(
                    &id,
                    &cmd.run,
                    daemon_dir.as_path(),
                    hook_retry_count,
                    readiness_daemon_env.as_ref(),
                    &readiness_resolved_ports,
                ));
            }

            // Use a channel to communicate process exit status
            let (exit_tx, mut exit_rx) =
                tokio::sync::mpsc::channel::<std::io::Result<std::process::ExitStatus>>(1);

            // Spawn a task to wait for process exit
            let child_pid = child.id().unwrap_or(0);
            tokio::spawn(async move {
                let result = child.wait().await;
                // On non-Linux Unix (e.g. macOS) the zombie reaper may win the
                // race and consume the exit status via waitpid(None, WNOHANG)
                // before Tokio's child.wait() gets to it. When that happens,
                // Tokio returns an ECHILD io::Error. We recover by checking
                // REAPED_STATUSES for the stashed exit code.
                //
                // On Linux this is unnecessary because the reaper uses
                // waitid(WNOWAIT) to peek before reaping, which avoids the
                // race entirely.
                #[cfg(all(unix, not(target_os = "linux")))]
                let result = match &result {
                    Err(e) if e.raw_os_error() == Some(nix::libc::ECHILD) => {
                        if let Some(code) = super::REAPED_STATUSES.lock().await.remove(&child_pid) {
                            warn!(
                                "daemon pid {child_pid} wait() got ECHILD; \
                                 recovered exit code {code} from zombie reaper"
                            );
                            // Synthesize an ExitStatus from the stashed code.
                            // On Unix we can use `ExitStatus::from_raw()` with
                            // a wait-style status word (code << 8 for normal
                            // exit, or raw signal number for signal death).
                            use std::os::unix::process::ExitStatusExt;
                            if code >= 0 {
                                Ok(std::process::ExitStatus::from_raw(code << 8))
                            } else {
                                // Negative code means killed by signal (-sig)
                                Ok(std::process::ExitStatus::from_raw((-code) & 0x7f))
                            }
                        } else {
                            warn!(
                                "daemon pid {child_pid} wait() got ECHILD but no \
                                 stashed status found; reporting as error"
                            );
                            result
                        }
                    }
                    _ => result,
                };
                debug!("daemon pid {child_pid} wait() completed with result: {result:?}");
                let _ = exit_tx.send(result).await;
            });

            #[allow(unused_assignments)]
            // Initial None is a safety net; loop only exits via exit_rx.recv() which sets it
            let mut exit_status = None;

            // If there is no ready check of any kind and no delay, the daemon is
            // considered immediately ready and the active_port detection task would
            // never be triggered inside the select loop.  Kick it off right away so
            // that daemons without any readiness configuration still get their
            // active_port populated (needed for proxy routing).
            if has_port_config
                && ready_pattern.is_none()
                && ready_http.is_none()
                && ready_port.is_none()
                && ready_cmd.is_none()
                && delay_timer.is_none()
            {
                active_port_spawned = true;
                detect_and_store_active_port(id.clone(), daemon_pid);
            }

            // Set when readiness checks exhaust. The group kill runs as a
            // separate task so this loop can exit and the post-loop drain
            // keeps consuming output — children logging during SIGTERM
            // cleanup would otherwise block on a full pipe and never exit.
            // The ready failure is only sent once the kill task completes,
            // so the retry loop cannot respawn into the dying group.
            let mut ready_fail_kill: Option<tokio::task::JoinHandle<()>> = None;

            loop {
                // biased: evaluate in exit → output → delay order so that
                // process exit pre-empts both buffered output and the delay
                // timer, preventing a dead daemon from being marked ready.
                select! {
                    biased;
                    Some(result) = exit_rx.recv() => {
                        // Process exited - save exit status and notify if not ready yet
                        exit_status = Some(result);
                        debug!("daemon {id} process exited, exit_status: {exit_status:?}");
                        if !ready_notified {
                            // Check if process exited successfully
                            let is_success = exit_status.as_ref()
                                .and_then(|r| r.as_ref().ok())
                                .map(|s| s.success())
                                .unwrap_or(false);
                            if is_success && opts.oneshot {
                                debug!("daemon {id} completed, deferring success notification until the completed state is persisted");
                                oneshot_completion_pending = true;
                            } else if let Some(tx) = ready_tx.take() {
                                if is_success {
                                    debug!("daemon {id} exited successfully before ready check, sending success notification");
                                    let _ = tx.send(Ok(()));
                                } else {
                                    let exit_code = exit_status.as_ref()
                                        .and_then(|r| r.as_ref().ok())
                                        .and_then(|s| s.code());
                                    debug!("daemon {id} exited with failure before ready check, sending failure notification with exit_code: {exit_code:?}");
                                    let _ = tx.send(Err(exit_code));
                                }
                            }
                        } else {
                            debug!("daemon {id} was already marked ready, not sending notification");
                        }
                        break;
                    },
                    Some(super::OutputLine { text: line, source }) = output_rx.recv() => {
                        // A line relayed by a sink is already in the store —
                        // the sink wrote and flushed it before reporting it —
                        // so it arrives here only to be acted on.
                        if matches!(source, super::OutputSource::Local) {
                            let parsed = parse_line(&line);
                            log_buffer.push(parsed);
                            if log_buffer.len() >= LOG_BATCH_SIZE {
                                let _ = flush_logs(&mut log_buffer);
                            }
                        }
                        trace!("output: {id} {line}");

                        // Strip ANSI for pattern matching so user-written patterns
                        // work regardless of whether the process emits color codes.
                        let line_clean = console::strip_ansi_codes(&line).to_string();

                        // Check if output matches ready pattern
                        if !ready_notified
                            && !output_exhausted
                            && let Some(ref pattern) = ready_pattern
                            && pattern.is_match(&line_clean)
                        {
                            // Flush buffered logs synchronously before signalling
                            // readiness, so collect_startup_logs sees the line
                            // that triggered the match (and any co-buffered lines)
                            // in SQLite.
                            if let Some(handle) = flush_logs(&mut log_buffer) {
                                let _ = handle.await;
                            }
                            info!("daemon {id} ready: output matched pattern");
                            ready_notified = true;
                            if let Some(tx) = ready_tx.take() {
                                let _ = tx.send(Ok(()));
                            }
                            fire_hook(HookType::OnReady, id.clone(), daemon_dir.clone(), hook_retry_count, hook_daemon_env.clone(), hook_resolved_ports.clone(), vec![]).await;
                            stop_cmd_probe_state(&mut cmd_probe);
                            http_deadline = None;
                            cmd_deadline = None;
                            port_deadline = None;
                            output_deadline = None;
                            if !active_port_spawned && has_port_config {
                                active_port_spawned = true;
                                detect_and_store_active_port(id.clone(), daemon_pid);
                            }
                        }

                        // Check on_output hook. A sink has already applied the
                        // filter, and says so per line: a line reported only
                        // because it announced readiness must not fire a hook
                        // that filters for something else.
                        if let Some(ref hook) = on_output_hook {
                            let matched = match source {
                                super::OutputSource::Sink { fires_hook } => fires_hook,
                                super::OutputSource::Local => match (&hook.filter, &on_output_pattern) {
                                    (Some(substr), _) => line_clean.contains(substr.as_str()),
                                    (None, Some(re)) => re.is_match(&line_clean),
                                    (None, None) => true,
                                },
                            };
                            if matched {
                                // The debounce is applied here as well as in the
                                // sink. A replacement sink starts with a fresh
                                // clock, and would otherwise let the hook fire
                                // twice inside one configured window.
                                let now = std::time::Instant::now();
                                let elapsed = on_output_last_fired.map(|t| now.duration_since(t));
                                if elapsed.is_none_or(|e| e >= on_output_debounce) {
                                    on_output_last_fired = Some(now);
                                    hooks::fire_output_hook(id.clone(), daemon_dir.clone(), hook_retry_count, hook_daemon_env.clone(), hook_resolved_ports.clone(), hook.run.clone(), line_clean.clone()).await;
                                }
                            }
                        }
                        // Yield briefly so that the output readiness deadline can be
                        // evaluated even when output is produced continuously.
                        tokio::task::yield_now().await;
                    }
                    _ = async {
                        if let Some(ref mut deadline) = http_deadline {
                            deadline.await;
                        } else {
                            std::future::pending::<()>().await;
                        }
                    }, if !ready_notified && ready_http.is_some() => {
                        http_exhausted = true;
                        http_deadline = None;
                        http_check_interval = None;
                        warn!("daemon {id}: HTTP readiness check timed out");
                        let any_remaining = any_ready_check_remaining(
                            ready_output.as_ref(),
                            output_exhausted,
                            ready_port_config.as_ref(),
                            port_exhausted,
                            ready_http.as_ref(),
                            http_exhausted,
                            ready_cmd.as_ref(),
                            cmd_exhausted,
                        );
                        if !any_remaining {
                            error!("daemon {id}: all readiness checks exhausted, failing");
                            stop_cmd_probe_state(&mut cmd_probe);
                            ready_fail_kill = Some(spawn_ready_fail_kill(
                                id.clone(),
                                daemon_pid,
                                opts.stop_signal.unwrap_or_default(),
                            ));
                            break;
                        }
                    }
                    _ = async {
                        if let Some(ref mut deadline) = output_deadline {
                            deadline.await;
                        } else {
                            std::future::pending::<()>().await;
                        }
                    }, if !ready_notified && ready_output.is_some() => {
                        output_exhausted = true;
                        output_deadline = None;
                        warn!("daemon {id}: output readiness check timed out");
                        let any_remaining = any_ready_check_remaining(
                            ready_output.as_ref(),
                            output_exhausted,
                            ready_port_config.as_ref(),
                            port_exhausted,
                            ready_http.as_ref(),
                            http_exhausted,
                            ready_cmd.as_ref(),
                            cmd_exhausted,
                        );
                        if !any_remaining {
                            error!("daemon {id}: all readiness checks exhausted, failing");
                            stop_cmd_probe_state(&mut cmd_probe);
                            ready_fail_kill = Some(spawn_ready_fail_kill(
                                id.clone(),
                                daemon_pid,
                                opts.stop_signal.unwrap_or_default(),
                            ));
                            break;
                        }
                    }
                    _ = async {
                        if let Some(ref mut interval) = http_check_interval {
                            interval.tick().await;
                        } else {
                            std::future::pending::<()>().await;
                        }
                    }, if !ready_notified && ready_http.is_some() && !http_exhausted => {
                        if let (Some(http), Some(client)) = (&ready_http, &http_client) {
                            match client.get(&http.url).send().await {
                                Ok(response) if http.accepts_status(response.status().as_u16()) => {
                                    info!("daemon {id} ready: HTTP check passed (status {})", response.status());
                                    ready_notified = true;
                                    if let Some(tx) = ready_tx.take() {
                                        let _ = tx.send(Ok(()));
                                    }
                                    fire_hook(HookType::OnReady, id.clone(), daemon_dir.clone(), hook_retry_count, hook_daemon_env.clone(), hook_resolved_ports.clone(), vec![]).await;
                                    http_check_interval = None;
                                    http_deadline = None;
                                    stop_cmd_probe_state(&mut cmd_probe);
                                    cmd_deadline = None;
                                    port_deadline = None;
                                    output_deadline = None;
                                    if !active_port_spawned && has_port_config {
                                        active_port_spawned = true;
                                        detect_and_store_active_port(id.clone(), daemon_pid);
                                    }
                                }
                                Ok(response) => {
                                    trace!("daemon {id} HTTP check: status {} (not ready)", response.status());
                                }
                                Err(e) => {
                                    trace!("daemon {id} HTTP check failed: {e}");
                                }
                            }
                        }
                    }
                    _ = async {
                        if let Some(ref mut deadline) = port_deadline {
                            deadline.await;
                        } else {
                            std::future::pending::<()>().await;
                        }
                    }, if !ready_notified && ready_port.is_some() => {
                        port_exhausted = true;
                        port_deadline = None;
                        port_check_interval = None;
                        warn!("daemon {id}: TCP port readiness check timed out");
                        let any_remaining = any_ready_check_remaining(
                            ready_output.as_ref(),
                            output_exhausted,
                            ready_port_config.as_ref(),
                            port_exhausted,
                            ready_http.as_ref(),
                            http_exhausted,
                            ready_cmd.as_ref(),
                            cmd_exhausted,
                        );
                        if !any_remaining {
                            error!("daemon {id}: all readiness checks exhausted, failing");
                            stop_cmd_probe_state(&mut cmd_probe);
                            ready_fail_kill = Some(spawn_ready_fail_kill(
                                id.clone(),
                                daemon_pid,
                                opts.stop_signal.unwrap_or_default(),
                            ));
                            break;
                        }
                    }
                    _ = async {
                        if let Some(ref mut interval) = port_check_interval {
                            interval.tick().await;
                        } else {
                            std::future::pending::<()>().await;
                        }
                    }, if !ready_notified && ready_port.is_some() && !port_exhausted => {
                        if let Some(port) = ready_port {
                            match tokio::net::TcpStream::connect(("127.0.0.1", port)).await {
                                Ok(_) => {
                                    info!("daemon {id} ready: TCP port {port} is listening");
                                    ready_notified = true;
                                    if let Some(tx) = ready_tx.take() {
                                        let _ = tx.send(Ok(()));
                                    }
                                    fire_hook(HookType::OnReady, id.clone(), daemon_dir.clone(), hook_retry_count, hook_daemon_env.clone(), hook_resolved_ports.clone(), vec![]).await;
                                    // Stop checking once ready
                                    port_check_interval = None;
                                    port_deadline = None;
                                    stop_cmd_probe_state(&mut cmd_probe);
                                    http_deadline = None;
                                    cmd_deadline = None;
                                    output_deadline = None;
                                    if !active_port_spawned && has_port_config {
                                        active_port_spawned = true;
                                        // ready_port check just TCP-connected to this
                                        // port, so it is definitely listening. If it
                                        // matches the first resolved port, write
                                        // active_port directly instead of spawning
                                        // detect_and_store_active_port, which sleeps
                                        // 500 ms then relies on listeners::get_all()
                                        // + process-tree traversal — unreliable on
                                        // Windows where Git Bash PID mapping can
                                        // break descendant lookups.
                                        if let Some(active_port) = active_port_from_ready_port(
                                            port,
                                            &readiness_resolved_ports,
                                        ) {
                                            let mut state_file =
                                                SUPERVISOR.state_file.lock().await;
                                            if let Some(d) = state_file.daemons.get(&id)
                                                && d.pid == Some(daemon_pid)
                                            {
                                                state_file.set_active_port(&id, active_port);
                                            }
                                        } else {
                                            detect_and_store_active_port(
                                                id.clone(),
                                                daemon_pid,
                                            );
                                        }
                                    }
                                }
                                Err(_) => {
                                    trace!("daemon {id} port check: port {port} not listening yet");
                                }
                            }
                        }
                    }
                    _ = async {
                        if let Some(ref mut delay) = cmd_respawn_delay {
                            delay.await;
                        } else {
                            std::future::pending::<()>().await;
                        }
                    }, if !ready_notified && ready_cmd.is_some() && !cmd_exhausted && cmd_probe.is_none() => {
                        if let Some(ref cmd) = ready_cmd {
                            cmd_probe = Some(spawn_cmd_probe(
                                &id,
                                &cmd.run,
                                daemon_dir.as_path(),
                                hook_retry_count,
                                readiness_daemon_env.as_ref(),
                                &readiness_resolved_ports,
                            ));
                        }
                        cmd_respawn_delay = None;
                    }
                    result = async {
                        if let Some(probe) = cmd_probe.as_mut() {
                            std::pin::Pin::new(&mut probe.result_rx).await
                        } else {
                            std::future::pending::<Result<Result<std::process::ExitStatus, std::io::Error>, tokio::sync::oneshot::error::RecvError>>().await
                        }
                    }, if !ready_notified && ready_cmd.is_some() && !cmd_exhausted => {
                        // The probe task has finished; remove the handle so it is not
                        // cancelled or reused. This must happen only after this branch
                        // actually wins the select, not while constructing the future.
                        let _ = cmd_probe.take();
                        match result {
                            Ok(Ok(status)) if status.success() => {
                                info!("daemon {id} ready: readiness command succeeded");
                                ready_notified = true;
                                if let Some(tx) = ready_tx.take() {
                                    let _ = tx.send(Ok(()));
                                }
                                fire_hook(HookType::OnReady, id.clone(), daemon_dir.clone(), hook_retry_count, hook_daemon_env.clone(), hook_resolved_ports.clone(), vec![]).await;
                                cmd_respawn_delay = None;
                                cmd_deadline = None;
                                http_deadline = None;
                                port_deadline = None;
                                output_deadline = None;
                                if !active_port_spawned && has_port_config {
                                    active_port_spawned = true;
                                    detect_and_store_active_port(id.clone(), daemon_pid);
                                }
                            }
                            Ok(Ok(_)) | Ok(Err(_)) | Err(_) => {
                                trace!("daemon {id} cmd check: command not ready, will respawn");
                                cmd_respawn_delay = Some(Box::pin(time::sleep(ready_check_interval)));
                            }
                        }
                    }
                    _ = async {
                        if let Some(ref mut deadline) = cmd_deadline {
                            deadline.await;
                        } else {
                            std::future::pending::<()>().await;
                        }
                    }, if !ready_notified && ready_cmd.is_some() => {
                        cmd_exhausted = true;
                        cmd_deadline = None;
                        stop_cmd_probe_state(&mut cmd_probe);
                        cmd_respawn_delay = None;
                        warn!("daemon {id}: command readiness check timed out");
                        let any_remaining = any_ready_check_remaining(
                            ready_output.as_ref(),
                            output_exhausted,
                            ready_port_config.as_ref(),
                            port_exhausted,
                            ready_http.as_ref(),
                            http_exhausted,
                            ready_cmd.as_ref(),
                            cmd_exhausted,
                        );
                        if !any_remaining {
                            error!("daemon {id}: all readiness checks exhausted, failing");
                            ready_fail_kill = Some(spawn_ready_fail_kill(
                                id.clone(),
                                daemon_pid,
                                opts.stop_signal.unwrap_or_default(),
                            ));
                            break;
                        }
                    }
                    _ = async {
                        if let Some(ref mut timer) = delay_timer {
                            timer.await;
                        } else {
                            std::future::pending::<()>().await;
                        }
                    } => {
                        let has_other_ready_check = ready_pattern.is_some()
                            || ready_http.is_some()
                            || ready_port.is_some()
                            || ready_cmd.is_some();
                        let delay_is_only_readiness = !ready_notified && !has_other_ready_check;
                        let process_exited = exit_status.is_some();
                        let process_running = if delay_is_only_readiness && !process_exited {
                            // Force-refresh sysinfo for this PID before checking.
                            // On Windows, the cached process list may be stale.
                            PROCS.refresh_pids(&[daemon_pid]);
                            PROCS.is_running(daemon_pid)
                        } else {
                            false
                        };

                        if delay_readiness_succeeded(
                            ready_notified,
                            has_other_ready_check,
                            process_exited,
                            process_running,
                        ) {
                            info!("daemon {id} ready: delay elapsed");
                            ready_notified = true;
                            if let Some(tx) = ready_tx.take() {
                                let _ = tx.send(Ok(()));
                            }
                            fire_hook(HookType::OnReady, id.clone(), daemon_dir.clone(), hook_retry_count, hook_daemon_env.clone(), hook_resolved_ports.clone(), vec![]).await;
                            if !active_port_spawned && has_port_config {
                                active_port_spawned = true;
                                detect_and_store_active_port(id.clone(), daemon_pid);
                            }
                        } else if delay_is_only_readiness {
                            if process_exited {
                                debug!("daemon {id} exited during ready_delay, not marking as ready");
                            } else {
                                debug!("daemon {id} pid {daemon_pid} not running during ready_delay, deferring to exit handler");
                            }
                        }

                        if delay_is_only_readiness {
                            // Clear all deadlines — no other checks are configured
                            // when delay fires as readiness, but clear defensively.
                            output_deadline = None;
                            http_deadline = None;
                            cmd_deadline = None;
                            port_deadline = None;
                            stop_cmd_probe_state(&mut cmd_probe);
                        }
                        // Disable timer after it fires
                        delay_timer = None;
                    }
                    _ = log_flush_interval.tick() => {
                        let _ = flush_logs(&mut log_buffer);
                    }
                }
            }

            // Snapshot the daemon state BEFORE draining output.
            //
            // The drain can take up to 5s (e.g. when child processes keep the
            // stdout pipe open). During that time, a subsequent start() call
            // (e.g. from `pitchfork restart`) can upsert the daemon with a new
            // PID and Running status. If we only checked state AFTER the drain,
            // the monitoring task would see d.pid != Some(old_pid) && !is_stopped()
            // && !is_stopping() and return early without firing on_stop/on_exit
            // hooks.
            //
            // By snapshotting is_stopping before the drain, we preserve the
            // knowledge that stop() was called, so hooks fire correctly even
            // if start() has since changed the state.
            let pre_drain_daemon = SUPERVISOR.get_daemon(&id).await;
            let pre_drain_is_stopping = pre_drain_daemon
                .as_ref()
                .is_some_and(|d| d.status.is_stopped() || d.status.is_stopping());

            // Drain any in-flight output lines that were still in the mpsc
            // channel or the OS pipe buffer when the child exited. Without
            // this, trailing log lines from short-lived daemons get dropped.
            // The reader tasks drop their senders on EOF, so recv() returns
            // None when all data has been consumed. A total deadline of 5 s
            // guards against a stuck reader (e.g. PTY master FD not closing)
            // while ensuring drain doesn't block post-exit cleanup indefinitely.
            //
            // Stop accepting relayed output first: the relay holds a sender of
            // its own, so leaving it registered would keep the channel open and
            // make every drain wait out the whole deadline. Readiness is moot
            // now anyway — the process has exited.
            drop(output_relay);
            let drain_deadline = tokio::time::Instant::now() + Duration::from_secs(5);
            loop {
                let now = tokio::time::Instant::now();
                if now >= drain_deadline {
                    break;
                }
                let Ok(Some(line)) =
                    tokio::time::timeout(drain_deadline - now, output_rx.recv()).await
                else {
                    break;
                };
                // Sink-relayed lines are already stored; see the select loop.
                if matches!(line.source, super::OutputSource::Local) {
                    log_buffer.push(parse_line(&line.text));
                }
            }
            // Flush any remaining log lines (including drained) before the process exits.
            // Await the flush to guarantee all buffered logs are persisted before cleanup.
            if let Some(handle) = flush_logs(&mut log_buffer) {
                let _ = handle.await;
            }

            // Clear active_port since the process is no longer running
            {
                let mut state_file = SUPERVISOR.state_file.lock().await;
                state_file.clear_active_port(&id);
            }

            // Get the final exit status
            let exit_status = if let Some(status) = exit_status {
                status
            } else {
                // Streams closed but process hasn't exited yet, wait for it
                match exit_rx.recv().await {
                    Some(status) => status,
                    None => {
                        warn!("daemon {id} exit channel closed without receiving status");
                        Err(std::io::Error::other("exit channel closed"))
                    }
                }
            };

            // If the loop exited via readiness exhaustion, wait for the group
            // kill to finish before reporting the failure so the retry loop
            // (or a waiting client) cannot start a replacement while the old
            // process group is still terminating.
            if let Some(kill) = ready_fail_kill {
                let _ = kill.await;
                if let Some(tx) = ready_tx.take() {
                    let _ = tx.send(Err(Some(124)));
                }
            }

            let current_daemon = SUPERVISOR.get_daemon(&id).await;

            // Signal that this monitoring task is processing its exit path.
            // The RAII guard will decrement the counter and notify close()
            // when the task finishes (including all fire_hook registrations),
            // regardless of which return path is taken.
            SUPERVISOR
                .active_monitors
                .fetch_add(1, atomic::Ordering::Release);
            struct MonitorGuard;
            impl Drop for MonitorGuard {
                fn drop(&mut self) {
                    SUPERVISOR
                        .active_monitors
                        .fetch_sub(1, atomic::Ordering::Release);
                    SUPERVISOR.monitor_done.notify_waiters();
                }
            }
            let _monitor_guard = MonitorGuard;
            // Check if this monitoring task is for the current daemon process.
            // If the daemon was intentionally stopped (pre_drain_is_stopping),
            // skip this check — we must still fire on_stop/on_exit hooks even
            // if start() has since changed the PID and status.
            if !pre_drain_is_stopping
                && (current_daemon.is_none()
                    || current_daemon.as_ref().is_some_and(|d| {
                        d.pid != Some(pid) && !d.status.is_stopped() && !d.status.is_stopping()
                    }))
            {
                // Another process has taken over, don't update status. The
                // task itself did finish, so a caller waiting on it is still
                // told so rather than left to time out.
                if oneshot_completion_pending && let Some(tx) = ready_tx.take() {
                    let _ = tx.send(Ok(()));
                }
                return;
            }
            // Capture the intentional-stop flag. Combine pre-drain and
            // post-drain state to handle both race orders:
            //  - stop() set Stopping before drain → pre_drain_is_stopping
            //  - stop() set Stopped during drain → current_daemon.is_stopped()
            let already_stopped = current_daemon
                .as_ref()
                .is_some_and(|d| d.status.is_stopped());
            let is_stopping = already_stopped
                || pre_drain_is_stopping
                || current_daemon
                    .as_ref()
                    .is_some_and(|d| d.status.is_stopping());

            // --- Phase 1: Determine exit_code, exit_reason, and update daemon state ---
            let (exit_code, exit_reason) = match (&exit_status, is_stopping) {
                (Ok(status), true) => {
                    // Intentional stop (by pitchfork). status.code() returns None
                    // on Unix when killed by signal (e.g. SIGTERM); use -1 to
                    // distinguish from a clean exit code 0.
                    (status.code().unwrap_or(-1), "stop")
                }
                (Ok(status), false) if status.success() => (status.code().unwrap_or(-1), "exit"),
                (Ok(status), false) => (status.code().unwrap_or(-1), "fail"),
                (Err(_), true) => {
                    // child.wait() error while stopping (e.g. sysinfo reaped the process)
                    (-1, "stop")
                }
                (Err(_), false) => (-1, "fail"),
            };
            // A stop that arrived while this monitor was draining did not get
            // to write anything, so it is applied here. A run that had already
            // succeeded keeps that outcome — there was nothing left to
            // interrupt — but a failed one is recorded as stopped, so the
            // retry checker leaves it alone.

            // Update daemon state unless stop() already did it (won the race),
            // OR the daemon was intentionally stopped before the drain
            // (pre_drain_is_stopping). In the latter case, start() may have
            // upserted Running during the 5s drain, and we must NOT overwrite
            // it with Stopped — that would undo the restart.
            if !already_stopped && !pre_drain_is_stopping {
                if let Ok(status) = &exit_status {
                    info!("daemon {id} exited with status {status}");
                }
                let (new_status, last_exit_success) = terminal_exit_state(
                    exit_reason,
                    opts.oneshot,
                    exit_code,
                    exit_status.as_ref().map(|s| s.success()).unwrap_or(true),
                );
                // Revalidate ownership inside the same state-lock section that
                // performs the write. The snapshot above was taken without
                // holding the lock, so a restart running on another thread can
                // install a successor in between; overwriting its record would
                // clear a live daemon's PID and undo the restart.
                if !SUPERVISOR
                    .finalize_monitored_exit(
                        &id,
                        pid,
                        monitor_token,
                        new_status,
                        Some(last_exit_success),
                    )
                    .await
                {
                    debug!("daemon {id} exit state was not written; a successor owns the record");
                }
            }

            // The terminal state is now visible, so a caller waiting on this
            // oneshot can return and see it. A task that was stopped partway
            // never did its work, so it does not satisfy anything waiting on
            // it — even when the process caught the signal and exited 0.
            if oneshot_completion_pending && let Some(tx) = ready_tx.take() {
                if exit_reason == "exit" {
                    let _ = tx.send(Ok(()));
                } else {
                    warn!("daemon {id}: oneshot was stopped before completing");
                    let _ = tx.send(Err(None));
                }
            }

            // --- Phase 2: Fire hooks ---
            let hook_extra_env = vec![
                ("PITCHFORK_EXIT_CODE".to_string(), exit_code.to_string()),
                ("PITCHFORK_EXIT_REASON".to_string(), exit_reason.to_string()),
            ];

            // Determine which hooks to fire based on exit reason
            let hooks_to_fire: Vec<HookType> = match exit_reason {
                "stop" => vec![HookType::OnStop, HookType::OnExit],
                "exit" => vec![HookType::OnExit],
                // "fail": fire on_fail + on_exit only when retries are exhausted
                _ if hook_retry_count >= hook_retry.count() => {
                    vec![HookType::OnFail, HookType::OnExit]
                }
                _ => vec![],
            };

            for hook_type in hooks_to_fire {
                fire_hook(
                    hook_type,
                    id.clone(),
                    daemon_dir.clone(),
                    hook_retry_count,
                    hook_daemon_env.clone(),
                    hook_resolved_ports.clone(),
                    hook_extra_env.clone(),
                )
                .await;
            }
        });

        // If wait_ready is true, wait for readiness notification
        if let Some(ready_rx) = ready_rx {
            match ready_rx.await {
                Ok(Ok(())) => {
                    info!("daemon {id} is ready");
                    // Re-read rather than returning the snapshot taken at
                    // spawn: a completed oneshot has since been finalized, and
                    // the snapshot would tell the caller it is still running
                    // under a PID that has exited.
                    //
                    // Only when the record still describes this run, though. A
                    // successor that claimed it carries its own PID and start
                    // time, and reporting those as the outcome of the process
                    // this call spawned would misattribute them.
                    let daemon = match self.get_daemon(id).await {
                        Some(current) if current.pid.is_none() || current.pid == Some(pid) => {
                            current
                        }
                        // A successor owns the record, so neither it nor the
                        // spawn snapshot describes this run: one carries
                        // another process's identity, the other still says
                        // running under a PID that has exited. A oneshot that
                        // reported ready did finish, so report that outcome
                        // directly rather than either misleading record.
                        _ if opts.oneshot => crate::daemon::Daemon {
                            status: DaemonStatus::Completed,
                            pid: None,
                            start_time: None,
                            boot_time: None,
                            last_exit_success: Some(true),
                            ..daemon
                        },
                        _ => daemon,
                    };
                    Ok(IpcResponse::DaemonReady { daemon })
                }
                Ok(Err(exit_code)) => {
                    error!("daemon {id} failed before becoming ready");
                    // The caller reports this by querying the log store for
                    // what the daemon printed, so wait for the sink's final
                    // write first. The in-process path got this ordering by
                    // flushing synchronously before signalling.
                    //
                    // Only on the attempt that gives up: `run` retries inline,
                    // and waiting after every attempt would both delay the
                    // backoff and widen the window in which the daemon looks
                    // errored and idle — long enough for the background retry
                    // checker to start an attempt of its own alongside it.
                    let last_attempt = opts.retry_count >= opts.retry.count();
                    if using_sink && last_attempt {
                        super::log_sink::wait_for_output(id, spawn_time, SINK_OUTPUT_TIMEOUT).await;
                    }
                    Ok(IpcResponse::DaemonFailedWithCode {
                        exit_code,
                        resolved_ports: attempt_resolved_ports,
                    })
                }
                Err(_) => {
                    error!("readiness channel closed unexpectedly for daemon {id}");
                    Ok(IpcResponse::DaemonStart { daemon })
                }
            }
        } else {
            Ok(IpcResponse::DaemonStart { daemon })
        }
    }

    /// Stop a running daemon
    pub async fn stop(&self, id: &DaemonId) -> Result<IpcResponse> {
        // Hold the daemon's stop lock for the whole stop (including the
        // whole-group termination wait) so starts and concurrent stops of the
        // same daemon serialize against it instead of racing the Stopping window.
        let lock = self.stop_lock(id).await;
        let _guard = lock.lock().await;
        self.stop_locked(id).await
    }

    /// Stop implementation. Caller must hold the daemon's stop lock.
    pub(super) async fn stop_locked(&self, id: &DaemonId) -> Result<IpcResponse> {
        let pitchfork_id = DaemonId::pitchfork();
        if *id == pitchfork_id {
            return Ok(IpcResponse::Error(
                "Cannot stop supervisor via stop command".into(),
            ));
        }
        info!("stopping daemon: {id}");
        // A foreground `start` may be working through this daemon's retries,
        // sleeping out a backoff with no process of its own to kill. Tell it to
        // give up, or it would start the next attempt once the stop has
        // returned.
        self.cancel_retrying(id);
        // ...and the retry checker may already have decided on an attempt it
        // has not started yet. The count is raised when this stop is done
        // rather than now, and while its lock is still held, so a checker that
        // reads the count while the stop is still recording itself reads the
        // old value and stands down when it reaches the lock. Raising it up
        // front would hand that reader a value that still matches once the
        // stop has finished.
        let _stop_epoch_bump = StopEpochGuard(id.clone());
        if let Some(daemon) = self.get_daemon(id).await {
            trace!("daemon to stop: {daemon}");
            if let Some(pid) = daemon.pid {
                trace!("killing pid: {pid}");
                if PROCS.is_running(pid) {
                    // Something is alive on that PID, but the kill below signals
                    // the entire process group: if the PID was recycled while
                    // this record sat unsupervised, that group belongs to an
                    // unrelated process tree. The daemon itself is gone either
                    // way, so report it as not running and clear the record.
                    if !super::signalling_pid_is_authorized(
                        daemon.start_time,
                        PROCS.start_time(pid),
                    ) {
                        warn!(
                            "pid {pid} recorded for daemon {id} belongs to another process now; not signalling it"
                        );
                        self.upsert_daemon(
                            UpsertDaemonOpts::builder(id.clone())
                                .set(|o| {
                                    o.pid = None;
                                    o.status = DaemonStatus::Stopped;
                                })
                                .build(),
                        )
                        .await?;
                        return Ok(IpcResponse::DaemonWasNotRunning);
                    }

                    // First set status to Stopping (preserve PID for monitoring task)
                    self.upsert_daemon(
                        UpsertDaemonOpts::builder(id.clone())
                            .set(|o| {
                                o.pid = Some(pid);
                                o.status = DaemonStatus::Stopping;
                            })
                            .build(),
                    )
                    .await?;

                    // Kill the entire process group atomically (daemon PID == PGID
                    // because we called setsid() at spawn time)
                    let stop_cfg = daemon.stop_signal.unwrap_or_default();
                    let stop_signal: i32 = stop_cfg.signal.into();
                    if let Err(e) = PROCS
                        .kill_process_group_async(pid, stop_signal, stop_cfg.timeout)
                        .await
                    {
                        debug!("failed to kill pid {pid}: {e}");
                        // Check if the process group is actually gone despite the
                        // error. Checking only the leader here would mark the daemon
                        // Stopped while surviving group members (e.g. one stuck in
                        // uninterruptible sleep) are still alive — letting a restart
                        // collide with them.
                        if PROCS.process_group_alive(pid) {
                            // Group still has live members - set back to Running
                            debug!(
                                "failed to stop pid {pid}: process group still alive after kill"
                            );
                            self.upsert_daemon(
                                UpsertDaemonOpts::builder(id.clone())
                                    .set(|o| {
                                        o.pid = Some(pid); // Preserve PID to avoid orphaning the process
                                        o.status = DaemonStatus::Running;
                                    })
                                    .build(),
                            )
                            .await?;
                            return Ok(IpcResponse::DaemonStopFailed {
                                error: format!(
                                    "process group of {pid} still alive after kill attempt: {e}"
                                ),
                            });
                        }
                    }

                    // Process successfully stopped
                    // Note: kill_process_group_async waits for the ENTIRE process
                    // group to exit (stop signal -> stop_timeout -> SIGKILL, then a
                    // bounded verification), so a replacement daemon can be started
                    // without colliding with a still-terminating instance. The only
                    // exception is a member stuck in uninterruptible sleep, which is
                    // logged with a warning.
                    self.upsert_daemon(
                        UpsertDaemonOpts::builder(id.clone())
                            .set(|o| {
                                o.pid = None;
                                o.status = DaemonStatus::Stopped;
                                o.last_exit_success = Some(true);
                            })
                            .build(),
                    )
                    .await?;
                } else if daemon.oneshot && self.is_monitored(id, pid) {
                    // The task's process is gone but its monitor is still
                    // running, so the run's real outcome has not been written
                    // yet — and for a task that finished on its own that
                    // outcome is `completed`. Writing `stopped` straight over
                    // it would discard a success the task actually achieved
                    // and report failure to anyone waiting on it, purely
                    // because a stop arrived a moment late.
                    //
                    // So wait for whoever is monitoring this run — the native
                    // monitor or an adopted one — to finish, then decide from
                    // what it wrote. Waiting rather than leaving a note for
                    // the monitor to find means there is no window in which
                    // the note lands too late to be read, and nothing left
                    // behind if it is never read at all. The wait is bounded,
                    // as is the monitor's own five-second output drain.
                    //
                    // Only oneshots take this path. A service has no
                    // successful exit to preserve, so it falls through to the
                    // arm below, which records the stop immediately.
                    debug!(
                        "pid {pid} not running but daemon {id} is still monitored; waiting for its monitor to settle the outcome"
                    );
                    self.wait_for_exit_finalized(id, Some(pid)).await;
                    let finished = self.get_daemon(id).await;
                    if finished
                        .as_ref()
                        .is_some_and(|d| stop_keeps_finalized_status(&d.status))
                    {
                        return Ok(IpcResponse::DaemonWasNotRunning);
                    }
                    // The run did not finish its work, so record the stop. A
                    // failure left in place would be picked up by the retry
                    // checker, which would start a task the user just stopped.
                    self.upsert_daemon(
                        UpsertDaemonOpts::builder(id.clone())
                            .set(|o| {
                                o.pid = None;
                                o.status = DaemonStatus::Stopped;
                            })
                            .build(),
                    )
                    .await?;
                    return Ok(IpcResponse::DaemonWasNotRunning);
                } else {
                    debug!("pid {pid} not running, process may have exited unexpectedly");
                    // Process already dead and unmonitored, so nothing else
                    // will record an outcome — transition to Stopped so the
                    // retry checker sees a terminal state and stops
                    // scheduling new attempts. This is important for an
                    // explicit `pitchfork stop` on an Errored daemon: the
                    // user wants to abort retries.
                    self.upsert_daemon(
                        UpsertDaemonOpts::builder(id.clone())
                            .set(|o| {
                                o.pid = None;
                                o.status = DaemonStatus::Stopped;
                            })
                            .build(),
                    )
                    .await?;
                    return Ok(IpcResponse::DaemonWasNotRunning);
                }
                Ok(IpcResponse::Ok)
            } else {
                debug!("daemon {id} not running");
                // No process to signal, but a failed record with retries left
                // is not inert: `check_retry` starts the next attempt from it,
                // whether or not a foreground start is also working through
                // them. Record the stop so nothing picks the daemon back up.
                if daemon.status.is_errored() && daemon.retry_count < daemon.retry.count() {
                    self.upsert_daemon(
                        UpsertDaemonOpts::builder(id.clone())
                            .set(|o| {
                                o.pid = None;
                                o.status = DaemonStatus::Stopped;
                            })
                            .build(),
                    )
                    .await?;
                    return Ok(IpcResponse::DaemonWasNotRunning);
                }
                Ok(IpcResponse::DaemonNotRunning)
            }
        } else {
            debug!("daemon {id} not found");
            Ok(IpcResponse::DaemonNotFound)
        }
    }
}

#[cfg(unix)]
fn resolve_effective_run_identity(daemon_user: Option<&str>) -> Result<RunIdentity> {
    let s = settings();
    let settings_user = s.supervisor.user.trim();
    let daemon_user = daemon_user.map(str::trim).filter(|user| !user.is_empty());
    let settings_user = (!settings_user.is_empty()).then_some(settings_user);
    let configured = daemon_user.or(settings_user);
    let current_uid = nix::unistd::Uid::effective().as_raw();
    let current_gid = nix::unistd::Gid::effective().as_raw();
    // The recorded invoking user of a boot service stands in for the sudo
    // environment that launchd and systemd do not provide.
    let invoking = env::invoking_user_ids().map(|(uid, gid)| (uid.to_string(), gid.to_string()));
    resolve_run_identity(
        configured,
        current_uid,
        current_gid,
        invoking.as_ref().map(|(uid, _)| uid.as_str()),
        invoking.as_ref().map(|(_, gid)| gid.as_str()),
    )
}

#[cfg(unix)]
fn resolve_run_identity(
    configured: Option<&str>,
    current_uid: u32,
    current_gid: u32,
    sudo_uid: Option<&str>,
    sudo_gid: Option<&str>,
) -> Result<RunIdentity> {
    let current_uid = nix::unistd::Uid::from_raw(current_uid);
    let current_gid = nix::unistd::Gid::from_raw(current_gid);
    if let Some(user) = configured {
        let identity = resolve_configured_user(user)?;
        ensure_can_use_identity(user, &identity, current_uid, current_gid)?;
        if identity.matches(current_uid, current_gid) {
            return Ok(RunIdentity::Inherit);
        }
        return Ok(identity);
    }

    if current_uid.is_root()
        && let Some(identity) = resolve_sudo_identity(sudo_uid, sudo_gid)
    {
        if identity.matches(current_uid, current_gid) {
            return Ok(RunIdentity::Inherit);
        }
        return Ok(identity);
    }

    Ok(RunIdentity::Inherit)
}

#[cfg(unix)]
fn resolve_configured_user(user: &str) -> Result<RunIdentity> {
    if user.chars().all(|c| c.is_ascii_digit()) {
        let uid = user
            .parse::<u32>()
            .map_err(|e| miette::miette!("invalid run user UID '{}': {}", user, e))?;
        let user_record = nix::unistd::User::from_uid(nix::unistd::Uid::from_raw(uid))
            .into_diagnostic()?
            .ok_or_else(|| miette::miette!("run user UID '{}' does not exist", user))?;
        return run_identity_from_user_record(user_record);
    }

    let user_record = nix::unistd::User::from_name(user)
        .into_diagnostic()?
        .ok_or_else(|| miette::miette!("run user '{}' does not exist", user))?;
    run_identity_from_user_record(user_record)
}

#[cfg(unix)]
fn run_identity_from_user_record(user: nix::unistd::User) -> Result<RunIdentity> {
    let username = CString::new(user.name)
        .map_err(|e| miette::miette!("run user name contains an interior nul byte: {}", e))?;
    Ok(RunIdentity::Switch {
        uid: user.uid,
        gid: user.gid,
        username: Some(username),
        home: Some(user.dir),
    })
}

#[cfg(unix)]
fn resolve_sudo_identity(sudo_uid: Option<&str>, sudo_gid: Option<&str>) -> Option<RunIdentity> {
    let uid = sudo_uid?.parse::<u32>().ok()?;
    let gid = sudo_gid?.parse::<u32>().ok()?;
    let user = nix::unistd::User::from_uid(nix::unistd::Uid::from_raw(uid))
        .ok()
        .flatten();
    let (username, home) = match user {
        Some(user) => (CString::new(user.name).ok(), Some(user.dir)),
        None => (None, None),
    };
    Some(RunIdentity::Switch {
        uid: nix::unistd::Uid::from_raw(uid),
        gid: nix::unistd::Gid::from_raw(gid),
        username,
        home,
    })
}

#[cfg(unix)]
fn ensure_can_use_identity(
    configured_user: &str,
    identity: &RunIdentity,
    current_uid: nix::unistd::Uid,
    current_gid: nix::unistd::Gid,
) -> Result<()> {
    let RunIdentity::Switch { uid, gid, .. } = identity else {
        return Ok(());
    };
    if *uid == current_uid && *gid == current_gid {
        return Ok(());
    }
    if current_uid.is_root() {
        return Ok(());
    }
    Err(miette::miette!(
        "daemon is configured to run as '{}', but the supervisor is running as uid={} gid={}. Restart the supervisor with sudo to switch to uid={} gid={}, or choose a user matching the supervisor.",
        configured_user,
        current_uid.as_raw(),
        current_gid.as_raw(),
        uid.as_raw(),
        gid.as_raw()
    ))
}

/// Point HOME, USER and LOGNAME at the user a daemon switches to.
///
/// The supervisor's environment describes the supervisor's own user (usually
/// root), so without this a daemon run as another user would read and write
/// root's home. A value the passwd entry cannot supply is removed rather than
/// left describing root.
#[cfg(unix)]
fn apply_identity_env(command: &mut tokio::process::Command, identity: &RunIdentity) {
    let RunIdentity::Switch { username, home, .. } = identity else {
        return;
    };
    match home.as_ref().filter(|home| !home.as_os_str().is_empty()) {
        Some(home) => command.env("HOME", home),
        None => command.env_remove("HOME"),
    };
    match username.as_ref().and_then(|name| name.to_str().ok()) {
        Some(name) => command.env("USER", name).env("LOGNAME", name),
        None => command.env_remove("USER").env_remove("LOGNAME"),
    };
}

#[cfg(unix)]
fn apply_run_identity(identity: &RunIdentity) -> std::io::Result<()> {
    let RunIdentity::Switch {
        uid, gid, username, ..
    } = identity
    else {
        return Ok(());
    };
    if let Some(username) = username {
        initgroups_for_user(username, *gid)?;
    } else {
        setgroups_to_primary(*gid)?;
    }
    nix::unistd::setgid(*gid).map_err(nix_to_io_error)?;
    nix::unistd::setuid(*uid).map_err(nix_to_io_error)?;
    Ok(())
}

#[cfg(unix)]
impl RunIdentity {
    fn matches(&self, uid: nix::unistd::Uid, gid: nix::unistd::Gid) -> bool {
        matches!(self, RunIdentity::Switch { uid: u, gid: g, .. } if *u == uid && *g == gid)
    }
}

#[cfg(unix)]
fn setgroups_to_primary(gid: nix::unistd::Gid) -> std::io::Result<()> {
    let groups = [gid.as_raw() as libc::gid_t];
    #[cfg(any(target_os = "linux", target_os = "android"))]
    let group_count = groups.len();
    #[cfg(not(any(target_os = "linux", target_os = "android")))]
    let group_count = groups.len() as libc::c_int;
    let rc = unsafe { libc::setgroups(group_count, groups.as_ptr()) };
    if rc == -1 {
        Err(std::io::Error::last_os_error())
    } else {
        Ok(())
    }
}

#[cfg(unix)]
fn initgroups_for_user(username: &CString, gid: nix::unistd::Gid) -> std::io::Result<()> {
    let gid = gid.as_raw();
    #[cfg(any(
        target_os = "macos",
        target_os = "ios",
        target_os = "tvos",
        target_os = "watchos"
    ))]
    let base_gid = i32::try_from(gid)
        .map_err(|_| std::io::Error::other(format!("gid {gid} is out of range")))?;

    #[cfg(not(any(
        target_os = "macos",
        target_os = "ios",
        target_os = "tvos",
        target_os = "watchos"
    )))]
    let base_gid = gid as libc::gid_t;

    // SAFETY: `username` is a valid nul-terminated C string and `base_gid`
    // is derived from a resolved system account or sudo-provided gid.
    let rc = unsafe { libc::initgroups(username.as_ptr(), base_gid) };
    if rc == -1 {
        Err(std::io::Error::last_os_error())
    } else {
        Ok(())
    }
}

#[cfg(unix)]
fn nix_to_io_error(err: nix::errno::Errno) -> std::io::Error {
    std::io::Error::from_raw_os_error(err as i32)
}

/// Check if multiple ports are available and optionally auto-bump to find available ports.
///
/// All ports are bumped by the same offset to maintain relative port spacing.
/// Returns the resolved ports (either the original or bumped ones).
/// Returns an error if any port is in use and auto_bump is disabled,
/// or if no available ports can be found after max attempts.
async fn check_ports_available(
    expected_ports: &[u16],
    auto_bump: bool,
    max_attempts: u32,
) -> Result<Vec<u16>> {
    if expected_ports.is_empty() {
        return Ok(Vec::new());
    }

    for bump_offset in 0..=max_attempts {
        // Use wrapping_add to handle overflow correctly - ports wrap around at 65535
        let candidate_ports: Vec<u16> = expected_ports
            .iter()
            .map(|&p| p.wrapping_add(bump_offset as u16))
            .collect();

        // Check if all ports in this set are available
        let mut all_available = true;
        let mut conflicting_port = None;

        for &port in &candidate_ports {
            // Port 0 is a special case - it requests an ephemeral port from the OS.
            // Skip the availability check for port 0 since binding to it always succeeds.
            if port == 0 {
                continue;
            }

            // Use spawn_blocking to avoid blocking the async runtime during TCP bind checks.
            //
            // We check multiple addresses to avoid false-negatives caused by SO_REUSEADDR.
            // On macOS/BSD, Rust's TcpListener::bind sets SO_REUSEADDR by default, which
            // allows binding 0.0.0.0:port even when 127.0.0.1:port is already in use
            // (because 0.0.0.0 is technically a different address).  Most daemons bind
            // to localhost, so checking 127.0.0.1 is essential to detect real conflicts.
            // We also check [::1] to cover IPv6 loopback listeners.
            //
            // NOTE: This check has a time-of-check-to-time-of-use (TOCTOU) race condition.
            // Another process could grab the port between our check and the daemon actually
            // binding. This is inherent to the approach and acceptable for our use case
            // since we're primarily detecting conflicts with already-running daemons.
            if is_port_in_use(port).await {
                all_available = false;
                conflicting_port = Some(port);
                break;
            }
        }

        if all_available {
            // Check for overflow (port wrapped around to 0 due to wrapping_add)
            // If any candidate port is 0 but the original expected port wasn't 0,
            // it means we've wrapped around and should stop
            if candidate_ports.contains(&0) && !expected_ports.contains(&0) {
                return Err(PortError::NoAvailablePort {
                    start_port: expected_ports[0],
                    attempts: bump_offset + 1,
                }
                .into());
            }
            if bump_offset > 0 {
                info!("ports {expected_ports:?} bumped by {bump_offset} to {candidate_ports:?}");
            }
            return Ok(candidate_ports);
        }

        // Port is in use
        if bump_offset == 0
            && !auto_bump
            && let Some(port) = conflicting_port
        {
            let (pid, process) = identify_port_owner(port).await;
            return Err(PortError::InUse { port, process, pid }.into());
        }
    }

    // No available ports found after max attempts
    Err(PortError::NoAvailablePort {
        start_port: expected_ports[0],
        attempts: max_attempts + 1,
    }
    .into())
}

/// Check whether a port is currently in use by attempting to bind on multiple addresses.
///
/// Returns `true` when at least one bind attempt gets `AddrInUse`, meaning another
/// process is listening.  Other errors (e.g. `AddrNotAvailable` on an address family
/// the OS doesn't support) are ignored so they don't produce false positives.
async fn is_port_in_use(port: u16) -> bool {
    tokio::task::spawn_blocking(move || {
        for &addr in &["0.0.0.0", "127.0.0.1", "::1"] {
            match std::net::TcpListener::bind((addr, port)) {
                Ok(listener) => drop(listener),
                Err(e) if e.kind() == std::io::ErrorKind::AddrInUse => return true,
                Err(_) => continue,
            }
        }
        false
    })
    .await
    .unwrap_or(false)
}

/// Best-effort lookup of the process occupying a port via `listeners::get_all()`.
///
/// Returns `(pid, process_name)`.  Falls back to `(0, "unknown")` when the
/// system call fails (permission error, unsupported OS, etc.).
async fn identify_port_owner(port: u16) -> (u32, String) {
    tokio::task::spawn_blocking(move || {
        listeners::get_all()
            .ok()
            .and_then(|list| {
                list.into_iter()
                    .find(|l| l.socket.port() == port)
                    .map(|l| (l.process.pid, l.process.name))
            })
            .unwrap_or((0, "unknown".to_string()))
    })
    .await
    .unwrap_or((0, "unknown".to_string()))
}

/// Detect whether a port is in use, and if so, identify the owning process.
///
/// Combines `is_port_in_use` (reliable bind probe) with `identify_port_owner`
/// (best-effort process lookup).  Returns `None` when the port is free.
async fn detect_port_conflict(port: u16) -> Option<(u32, String)> {
    if !is_port_in_use(port).await {
        return None;
    }
    Some(identify_port_owner(port).await)
}

#[derive(Debug, PartialEq, Eq)]
enum ActivePortSelection {
    NoCandidates,
    Selected(u16),
    Ambiguous(Vec<u16>),
}

fn discovery_preferred_port(daemon: &crate::daemon::Daemon) -> Option<u16> {
    daemon
        .resolved_port
        .first()
        .copied()
        .or_else(|| {
            daemon
                .port
                .as_ref()
                .and_then(|port| port.expect.first().copied())
        })
        .filter(|&port| port > 0)
}

fn select_active_port(
    listeners: impl IntoIterator<Item = listeners::Listener>,
    descendant_pids: &std::collections::HashSet<u32>,
    preferred_port: Option<u16>,
) -> ActivePortSelection {
    let process_ports: std::collections::BTreeSet<u16> = listeners
        .into_iter()
        .filter(|listener| {
            listener.protocol == listeners::Protocol::TCP
                && listener.state == listeners::SocketState::Listen
                && descendant_pids.contains(&listener.process.pid)
        })
        .map(|listener| listener.socket.port())
        .filter(|&port| port > 0)
        .collect();

    if let Some(port) = preferred_port
        && process_ports.contains(&port)
    {
        return ActivePortSelection::Selected(port);
    }

    match process_ports.len() {
        0 => ActivePortSelection::NoCandidates,
        1 => ActivePortSelection::Selected(*process_ports.first().unwrap()),
        _ => ActivePortSelection::Ambiguous(process_ports.into_iter().collect()),
    }
}

/// Spawn a background task that detects the daemon process's active listening port
/// and stores it in the state file as `active_port`.
///
/// This is called once when the daemon becomes ready. The port is cleared when the daemon stops.
///
/// Port selection strategy:
/// 1. Consider only TCP listening sockets owned by the daemon or its descendants.
/// 2. Prefer the first resolved port, falling back to the first expected port for
///    legacy state without resolved ports.
/// 3. Select a sole distinct candidate; leave `active_port` unset when ambiguous.
fn detect_and_store_active_port(id: DaemonId, pid: u32) {
    tokio::spawn(async move {
        // Retry with exponential backoff so that slow-starting daemons (JVM,
        // Node.js, Python, etc.) that take more than 500 ms to bind their port
        // are still detected.  Total wait budget: 500+1000+2000+4000 = 7.5 s.
        for delay_ms in [500u64, 1000, 2000, 4000] {
            tokio::time::sleep(std::time::Duration::from_millis(delay_ms)).await;

            // Read daemon state atomically: check if still alive and get the preferred port
            // in a single lock acquisition to avoid TOCTOU and unnecessary lock overhead.
            let preferred_port: Option<u16> = {
                let state_file = SUPERVISOR.state_file.lock().await;
                match state_file.daemons.get(&id) {
                    Some(d) if d.pid.is_none() => {
                        debug!("daemon {id}: aborting active_port detection — process exited");
                        return;
                    }
                    Some(d) => discovery_preferred_port(d),
                    None => None,
                }
            };

            let selection = tokio::task::spawn_blocking(move || {
                let listeners = listeners::get_all().ok()?;

                // Refresh process tree so all_children sees current descendants.
                PROCS.refresh_processes();

                let descendant_pids: std::collections::HashSet<u32> = PROCS
                    .all_children(pid)
                    .into_iter()
                    .chain(std::iter::once(pid))
                    .collect();

                Some(select_active_port(
                    listeners,
                    &descendant_pids,
                    preferred_port,
                ))
            })
            .await
            .ok()
            .flatten()
            .unwrap_or(ActivePortSelection::NoCandidates);

            let port = match selection {
                ActivePortSelection::Selected(port) => port,
                ActivePortSelection::Ambiguous(ports) => {
                    debug!(
                        "daemon {id}: ambiguous active_port candidates {ports:?} for pid {pid} \
                         and its descendants; leaving active_port unset (will retry)"
                    );
                    continue;
                }
                ActivePortSelection::NoCandidates => {
                    debug!(
                        "daemon {id}: no active port detected for pid {pid} or its descendants \
                         (will retry)"
                    );
                    continue;
                }
            };

            debug!("daemon {id} active_port detected: {port}");
            let mut state_file = SUPERVISOR.state_file.lock().await;
            if let Some(d) = state_file.daemons.get(&id) {
                // Guard against PID reuse: if the original process exited and the OS
                // assigned the same PID to an unrelated process that happens to bind
                // a port, we must not route proxy traffic to that unrelated service.
                if d.pid == Some(pid) {
                    state_file.set_active_port(&id, port);
                } else {
                    debug!(
                        "daemon {id}: skipping active_port write — PID mismatch \
                         (expected {pid}, current {:?})",
                        d.pid
                    );
                }
            }
            return;
        }

        debug!(
            "daemon {id}: active port detection exhausted all retries for pid {pid} and its descendants"
        );
    });
}

#[cfg(test)]
mod active_port_tests {
    use super::*;
    use crate::config_types::PortConfig;
    use listeners::{Listener, Process, Protocol, SocketState};
    use std::net::{IpAddr, Ipv4Addr, SocketAddr};

    fn listener(pid: u32, port: u16, protocol: Protocol, state: SocketState) -> Listener {
        Listener {
            process: Process {
                pid,
                name: "test".to_string(),
                path: "/test".to_string(),
            },
            socket: SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), port),
            protocol,
            state,
        }
    }

    #[test]
    fn active_port_candidates_exclude_outbound_tcp_and_udp_sockets() {
        let daemon_pid = 100;
        let child_pid = 101;
        let descendant_pids = [daemon_pid, child_pid].into_iter().collect();
        let listeners = vec![
            listener(child_pid, 3004, Protocol::TCP, SocketState::Listen),
            listener(daemon_pid, 47082, Protocol::TCP, SocketState::Established),
            listener(daemon_pid, 5353, Protocol::UDP, SocketState::Unknown),
            listener(999, 9000, Protocol::TCP, SocketState::Listen),
        ];

        assert_eq!(
            select_active_port(listeners, &descendant_pids, None),
            ActivePortSelection::Selected(3004)
        );
    }

    #[test]
    fn bumped_cmd_readiness_prefers_resolved_primary_port() {
        let daemon = crate::daemon::Daemon {
            resolved_port: vec![3004],
            port: Some(PortConfig {
                expect: vec![3000],
                ..PortConfig::default()
            }),
            ..crate::daemon::Daemon::default()
        };
        let descendant_pids = [100].into_iter().collect();
        let listeners = vec![
            listener(100, 9000, Protocol::TCP, SocketState::Listen),
            listener(100, 3004, Protocol::TCP, SocketState::Listen),
        ];

        assert_eq!(discovery_preferred_port(&daemon), Some(3004));
        assert_eq!(
            select_active_port(
                listeners,
                &descendant_pids,
                discovery_preferred_port(&daemon),
            ),
            ActivePortSelection::Selected(3004)
        );
    }

    #[test]
    fn legacy_state_uses_expected_primary_port_for_discovery() {
        let daemon = crate::daemon::Daemon {
            port: Some(PortConfig {
                expect: vec![3000],
                ..PortConfig::default()
            }),
            ..crate::daemon::Daemon::default()
        };

        assert_eq!(discovery_preferred_port(&daemon), Some(3000));
    }

    #[test]
    fn ambiguous_candidates_leave_active_port_unset() {
        let descendant_pids = [100].into_iter().collect();
        let listeners = vec![
            listener(100, 9000, Protocol::TCP, SocketState::Listen),
            listener(100, 3000, Protocol::TCP, SocketState::Listen),
        ];

        assert_eq!(
            select_active_port(listeners, &descendant_pids, None),
            ActivePortSelection::Ambiguous(vec![3000, 9000])
        );
    }

    #[test]
    fn bumped_ready_port_sets_only_the_resolved_primary_without_scanning() {
        assert_eq!(active_port_from_ready_port(3004, &[3004]), Some(3004));
        assert_eq!(active_port_from_ready_port(4003, &[3003, 4003]), None);
    }

    #[test]
    fn delay_readiness_requires_delay_only_and_a_running_process() {
        assert!(delay_readiness_succeeded(false, false, false, true));
        assert!(!delay_readiness_succeeded(false, true, false, true));
        assert!(!delay_readiness_succeeded(false, false, true, false));
        assert!(!delay_readiness_succeeded(false, false, false, false));
        assert!(!delay_readiness_succeeded(true, false, false, true));
    }
}

/// Check whether a daemon (by its qualified ID) is the target of any registered
/// slug in the global config.  This is used to decide whether to run the
/// `detect_and_store_active_port` polling task — only slug-targeted daemons need
/// it, avoiding wasted `listeners::get_all()` calls for port-less daemons.
///
/// Delegates to `proxy::server::is_slug_target()` which uses the same in-memory
/// slug cache as the proxy hot path, so this check is cheap.
fn is_daemon_slug_target(id: &DaemonId) -> bool {
    // read_global_slugs is called once per daemon start — acceptable cost.
    // We intentionally avoid making this async to keep has_port_config evaluation
    // simple and synchronous in run_once().
    let slugs = crate::pitchfork_toml::PitchforkToml::read_global_slugs();
    slugs.iter().any(|(slug, entry)| {
        let daemon_name = entry.daemon.as_deref().unwrap_or(slug);
        id.name() == daemon_name
    })
}

#[cfg(test)]
mod oneshot_tests {
    use super::*;

    #[test]
    fn oneshot_clean_exit_is_completed() {
        let (status, success) = terminal_exit_state("exit", true, 0, true);
        assert!(matches!(status, DaemonStatus::Completed));
        assert!(success);
    }

    #[test]
    fn service_clean_exit_is_still_stopped() {
        let (status, success) = terminal_exit_state("exit", false, 0, true);
        assert!(matches!(status, DaemonStatus::Stopped));
        assert!(success);
    }

    #[test]
    fn oneshot_failure_is_errored_so_retry_applies() {
        // check_retry() only picks up errored daemons, so a non-zero exit must
        // not be recorded as completed.
        let (status, success) = terminal_exit_state("fail", true, 3, false);
        assert!(matches!(status, DaemonStatus::Errored(3)));
        assert!(!success);
    }

    #[test]
    fn a_stop_cancels_the_retry_sequence_it_finds() {
        let id = DaemonId::new("retry-cancel-test", "task");
        let claim = SUPERVISOR.mark_retrying(&id);
        assert!(!claim.is_cancelled());
        assert!(SUPERVISOR.is_retrying(&id));
        SUPERVISOR.cancel_retrying(&id);
        assert!(claim.is_cancelled());
        drop(claim);
        assert!(!SUPERVISOR.is_retrying(&id));
    }

    #[test]
    fn a_stop_cancels_every_sequence_for_the_daemon() {
        // Two starts can be working through the same daemon's retries: the
        // first releases the daemon's lock while it sleeps out a backoff. A
        // stop has to end both, not just whichever claimed it last.
        let id = DaemonId::new("retry-cancel-test", "concurrent");
        let first = SUPERVISOR.mark_retrying(&id);
        let second = SUPERVISOR.mark_retrying(&id);
        SUPERVISOR.cancel_retrying(&id);
        assert!(first.is_cancelled());
        assert!(second.is_cancelled());
        drop(second);
        // The first is still going, so the retry checker must still stand off.
        assert!(SUPERVISOR.is_retrying(&id));
        drop(first);
        assert!(!SUPERVISOR.is_retrying(&id));
    }

    #[test]
    fn a_stop_invalidates_an_attempt_decided_on_before_it() {
        // The retry checker reads the epoch when it decides on an attempt and
        // `run_retry` compares it under the daemon's lock, so an approval from
        // before a stop cannot slip past that stop.
        let id = DaemonId::new("stop-epoch-test", "task");
        let approved_at = SUPERVISOR.stop_epoch(&id);
        assert_eq!(SUPERVISOR.stop_epoch(&id), approved_at);
        SUPERVISOR.bump_stop_epoch(&id);
        assert_ne!(SUPERVISOR.stop_epoch(&id), approved_at);
        // An attempt decided on after the stop is still fine to start.
        let approved_after = SUPERVISOR.stop_epoch(&id);
        assert_eq!(SUPERVISOR.stop_epoch(&id), approved_after);
    }

    #[test]
    fn stop_epochs_are_tracked_per_daemon() {
        let stopped = DaemonId::new("stop-epoch-test", "stopped");
        let untouched = DaemonId::new("stop-epoch-test", "untouched");
        let approved_at = SUPERVISOR.stop_epoch(&untouched);
        SUPERVISOR.bump_stop_epoch(&stopped);
        assert_eq!(SUPERVISOR.stop_epoch(&untouched), approved_at);
    }

    #[test]
    fn a_stop_leaves_a_completed_task_alone() {
        // It had already done its work, so the stop had nothing to interrupt.
        assert!(stop_keeps_finalized_status(&DaemonStatus::Completed));
    }

    #[test]
    fn a_stop_replaces_a_failure_so_retries_do_not_resume() {
        // check_retry() picks up errored daemons, so a stop has to overwrite
        // one or it will start the task again.
        assert!(!stop_keeps_finalized_status(&DaemonStatus::Errored(1)));
        assert!(!stop_keeps_finalized_status(&DaemonStatus::Running));
        assert!(!stop_keeps_finalized_status(&DaemonStatus::Stopped));
    }

    #[test]
    fn stopped_oneshot_did_not_complete() {
        let (status, _) = terminal_exit_state("stop", true, 0, true);
        assert!(matches!(status, DaemonStatus::Stopped));
    }
}

#[cfg(all(test, unix))]
mod tests {
    use super::*;

    #[test]
    fn test_resolve_run_identity_empty_without_sudo() {
        let identity = resolve_run_identity(None, 501, 20, None, None).unwrap();
        assert_eq!(identity, RunIdentity::Inherit);
    }

    #[test]
    fn test_resolve_run_identity_sudo_fallback() {
        let identity = resolve_run_identity(None, 0, 0, Some("501"), Some("20")).unwrap();
        let RunIdentity::Switch { uid, gid, .. } = identity else {
            panic!("expected identity switch");
        };
        assert_eq!(uid.as_raw(), 501);
        assert_eq!(gid.as_raw(), 20);
    }

    #[test]
    fn test_resolve_run_identity_ignores_stale_sudo_when_not_root() {
        let identity = resolve_run_identity(None, 501, 20, Some("0"), Some("0")).unwrap();
        assert_eq!(identity, RunIdentity::Inherit);
    }

    #[test]
    fn test_resolve_configured_user_root_name() {
        let identity = resolve_configured_user("root").unwrap();
        let RunIdentity::Switch { uid, username, .. } = identity else {
            panic!("expected identity switch");
        };
        assert_eq!(uid.as_raw(), 0);
        assert_eq!(
            username.as_deref().and_then(|s| s.to_str().ok()),
            Some("root")
        );
    }

    #[test]
    fn test_resolve_configured_user_root_uid() {
        let identity = resolve_configured_user("0").unwrap();
        let RunIdentity::Switch { uid, username, .. } = identity else {
            panic!("expected identity switch");
        };
        assert_eq!(uid.as_raw(), 0);
        assert_eq!(
            username.as_deref().and_then(|s| s.to_str().ok()),
            Some("root")
        );
    }

    #[test]
    fn test_resolve_configured_user_missing_user_fails() {
        let err = resolve_configured_user("pitchfork-user-that-should-not-exist")
            .unwrap_err()
            .to_string();
        assert!(err.contains("does not exist"));
    }

    #[test]
    fn test_resolve_run_identity_requires_root_for_user_switch() {
        let err = resolve_run_identity(Some("root"), 501, 20, None, None)
            .unwrap_err()
            .to_string();
        assert!(err.contains("Restart the supervisor with sudo"));
    }

    #[test]
    fn test_resolve_run_identity_same_user_is_noop() {
        let identity = resolve_run_identity(Some("root"), 0, 0, Some("501"), Some("20")).unwrap();
        assert_eq!(identity, RunIdentity::Inherit);
    }

    #[test]
    fn test_resolve_configured_user_records_home() {
        let identity = resolve_configured_user("root").unwrap();
        let RunIdentity::Switch { home, .. } = identity else {
            panic!("expected identity switch");
        };
        let expected = nix::unistd::User::from_name("root").unwrap().unwrap().dir;
        assert_eq!(home, Some(expected));
    }

    fn switch_to(name: Option<&str>, home: Option<&str>) -> RunIdentity {
        RunIdentity::Switch {
            uid: nix::unistd::Uid::from_raw(501),
            gid: nix::unistd::Gid::from_raw(20),
            username: name.map(|n| CString::new(n).unwrap()),
            home: home.map(std::path::PathBuf::from),
        }
    }

    /// The env a command will run with, as set on the command itself:
    /// `Some(None)` is an explicit removal, `None` means inherited.
    fn command_env(
        command: &tokio::process::Command,
        key: &str,
    ) -> Option<Option<std::ffi::OsString>> {
        command
            .as_std()
            .get_envs()
            .find(|(k, _)| *k == key)
            .map(|(_, v)| v.map(ToOwned::to_owned))
    }

    /// Mirrors the order run_once applies them in.
    fn daemon_command(
        identity: &RunIdentity,
        daemon_env: Option<&IndexMap<String, String>>,
    ) -> tokio::process::Command {
        let mut command = tokio::process::Command::new("true");
        apply_identity_env(&mut command, identity);
        apply_runtime_env(
            &mut command,
            &DaemonId::new("identity-env-test", "api"),
            0,
            daemon_env,
            &[],
        );
        command
    }

    #[test]
    fn test_identity_env_describes_the_switched_user() {
        let command = daemon_command(&switch_to(Some("alice"), Some("/home/alice")), None);
        assert_eq!(
            command_env(&command, "HOME"),
            Some(Some("/home/alice".into()))
        );
        assert_eq!(command_env(&command, "USER"), Some(Some("alice".into())));
        assert_eq!(command_env(&command, "LOGNAME"), Some(Some("alice".into())));
    }

    #[test]
    fn test_identity_env_leaves_inherited_env_alone() {
        let command = daemon_command(&RunIdentity::Inherit, None);
        assert_eq!(command_env(&command, "HOME"), None);
        assert_eq!(command_env(&command, "USER"), None);
        assert_eq!(command_env(&command, "LOGNAME"), None);
    }

    #[test]
    fn test_root_sudo_to_root_preserves_environment() {
        let identity = resolve_run_identity(None, 0, 0, Some("0"), Some("0")).unwrap();
        assert_eq!(identity, RunIdentity::Inherit);
        let command = daemon_command(&identity, None);
        for key in ["HOME", "USER", "LOGNAME"] {
            assert_eq!(command_env(&command, key), None);
        }
    }

    #[test]
    fn test_daemon_env_overrides_identity_env() {
        let mut env = IndexMap::new();
        env.insert("HOME".to_string(), "/srv/app".to_string());
        env.insert("USER".to_string(), "app".to_string());
        let command = daemon_command(&switch_to(Some("alice"), Some("/home/alice")), Some(&env));
        assert_eq!(command_env(&command, "HOME"), Some(Some("/srv/app".into())));
        assert_eq!(command_env(&command, "USER"), Some(Some("app".into())));
        assert_eq!(command_env(&command, "LOGNAME"), Some(Some("alice".into())));
    }

    #[test]
    fn test_identity_env_drops_values_without_a_passwd_entry() {
        // A sudo uid with no passwd entry: the supervisor's values would still
        // describe root, so they are removed instead.
        let command = daemon_command(&switch_to(None, None), None);
        assert_eq!(command_env(&command, "HOME"), Some(None));
        assert_eq!(command_env(&command, "USER"), Some(None));
        assert_eq!(command_env(&command, "LOGNAME"), Some(None));
    }

    #[test]
    fn test_daemon_env_restores_values_without_a_passwd_entry() {
        let mut env = IndexMap::new();
        env.insert("HOME".to_string(), "/srv/app".to_string());
        let command = daemon_command(&switch_to(None, None), Some(&env));
        assert_eq!(command_env(&command, "HOME"), Some(Some("/srv/app".into())));
    }
}

/// Inject proxy-related environment variables into a daemon's command.
///
/// Adds:
/// - `HOST` — the address the daemon should bind to (`127.0.0.1`, omitted in LAN mode)
/// - `PITCHFORK_URL` — the public proxy URL for this daemon (if it has a slug)
/// - `PITCHFORK_CA_FILE` / `NODE_EXTRA_CA_CERTS` — path to the pitchfork CA cert (if HTTPS enabled)
/// - `__VITE_ADDITIONAL_SERVER_ALLOWED_HOSTS` — `.<tld>` for Vite host allowlisting
/// - `PITCHFORK_LAN` — set to `"1"` when LAN mode is active
fn inject_proxy_env(cmd: &mut tokio::process::Command, host: &Option<String>) {
    let s = crate::settings::settings();
    let lan_enabled = s.proxy.lan || !s.proxy.lan_ip.is_empty();

    if s.proxy.enable && host.is_some() && !lan_enabled {
        // Only force loopback binding for daemons the proxy actually routes to.
        // In LAN mode, daemons need to bind to 0.0.0.0 to be reachable from the network.
        cmd.env("HOST", "127.0.0.1");
    }

    // PITCHFORK_URL: the daemon's public proxy URL (only if it is routed and proxy is enabled)
    if let Some(url) = build_pitchfork_url(host, &s) {
        cmd.env("PITCHFORK_URL", &url);
    }

    // PITCHFORK_CA_FILE / NODE_EXTRA_CA_CERTS: let daemons verify TLS to each
    // other through the proxy.  `PITCHFORK_CA_FILE` is the runtime-agnostic
    // name; most TLS libraries take a CA bundle path from configuration, and
    // several read one straight out of the environment (for example
    // `SSL_CERT_FILE` for OpenSSL or `REQUESTS_CA_BUNDLE` for Python).
    if s.proxy.enable && s.proxy.https {
        let ca_path = if s.proxy.tls_cert.is_empty() {
            crate::env::PITCHFORK_STATE_DIR.join("proxy").join("ca.pem")
        } else {
            std::path::PathBuf::from(&s.proxy.tls_cert)
        };
        if ca_path.exists() {
            let ca_path = ca_path.to_string_lossy().to_string();
            cmd.env("PITCHFORK_CA_FILE", &ca_path);
            cmd.env("NODE_EXTRA_CA_CERTS", &ca_path);
        }
    }

    // __VITE_ADDITIONAL_SERVER_ALLOWED_HOSTS: Vite host allowlisting
    if s.proxy.enable {
        let tld = if lan_enabled { "local" } else { &s.proxy.tld };
        cmd.env("__VITE_ADDITIONAL_SERVER_ALLOWED_HOSTS", format!(".{tld}"));
    }

    // PITCHFORK_LAN: signal to daemons that LAN mode is active
    if lan_enabled {
        cmd.env("PITCHFORK_LAN", "1");
    }
}

/// The hostname the proxy routes to this daemon, without the TLD.
///
/// A daemon registered under a legacy `[slugs]` entry keeps that spelling,
/// because the proxy resolves slugs first. Otherwise the hostname is derived
/// from where the daemon's configuration lives.
async fn daemon_proxy_host(opts: &RunOptions) -> Option<String> {
    // Nothing consumes a hostname while the proxy is off, and deriving one
    // reads configuration and walks the project, so daemon starts skip it.
    if !crate::settings::settings().proxy.enable {
        return None;
    }
    // A slug carried on the run options skips the lookup below, so it needs the
    // same length check that lookup applies; otherwise the daemon is told a URL
    // the proxy refuses to route.
    if let Some(slug) = opts.slug.as_deref()
        && crate::proxy::hostname::hostname_fits(slug)
    {
        return opts.slug.clone();
    }
    // The daemon's own `dir` can point outside its project, so look the config
    // up from where it was defined.
    let config_dir = opts
        .watch_base_dir
        .clone()
        .unwrap_or_else(|| opts.dir.0.clone());
    let id = opts.id.clone();
    // Reading the config, the slug registry and the project's checkouts is all
    // file I/O, so it happens together on a blocking worker rather than on the
    // supervisor's executor. The lookup is the one the CLI and the proxy use,
    // so the daemon is told the address they advertise for it — a registered
    // slug when it has one, otherwise its automatic hostname.
    tokio::task::spawn_blocking(move || {
        let pt = crate::pitchfork_toml::PitchforkToml::all_merged_from(&config_dir).ok()?;
        let slugs = crate::pitchfork_toml::PitchforkToml::read_global_slugs();
        crate::proxy::hostname::host_for_daemon(&id, pt.daemons.get(&id), &slugs)
    })
    .await
    .unwrap_or_default()
}

/// Compute the public proxy URL for a daemon.
///
/// Returns `None` if the daemon has no hostname or the proxy is not enabled.
fn build_pitchfork_url(host: &Option<String>, s: &crate::settings::Settings) -> Option<String> {
    crate::proxy::build_proxy_url(host.as_deref(), s)
}

#[cfg(test)]
mod ready_check_tests {
    use super::*;
    use std::time::Duration;

    #[test]
    fn any_ready_check_remaining_prefers_unbounded_checks() {
        let http = ReadyHttp::new("http://localhost/health");
        let cmd = ReadyCmd::new("true");

        assert!(any_ready_check_remaining(
            None,
            false,
            None,
            false,
            Some(&http),
            false,
            None,
            false
        ));
        assert!(any_ready_check_remaining(
            None,
            false,
            None,
            false,
            None,
            false,
            Some(&cmd),
            false
        ));
        assert!(any_ready_check_remaining(
            None,
            false,
            Some(&ReadyPort::new(8080)),
            false,
            Some(&http),
            true,
            Some(&cmd),
            true
        ));
    }

    #[test]
    fn any_ready_check_remaining_exhausted_timed_checks() {
        let http = ReadyHttp {
            url: "http://localhost/health".to_string(),
            status: vec![],
            timeout: Some(Duration::from_secs(5)),
        };
        let cmd = ReadyCmd {
            run: "true".into(),
            timeout: Some(Duration::from_secs(5)),
        };

        assert!(any_ready_check_remaining(
            None,
            false,
            None,
            false,
            Some(&http),
            false,
            Some(&cmd),
            false
        ));
        assert!(!any_ready_check_remaining(
            None,
            false,
            None,
            false,
            Some(&http),
            true,
            Some(&cmd),
            true
        ));
    }

    #[tokio::test]
    async fn spawn_cmd_probe_reports_success() {
        let id = DaemonId::new("global", "probe-test");
        let probe = spawn_cmd_probe(&id, "true", &std::env::temp_dir(), 0, None, &[]);
        let status = probe.result_rx.await.unwrap().unwrap();
        assert!(status.success());
    }

    #[tokio::test]
    async fn spawn_cmd_probe_stops_on_request() {
        let id = DaemonId::new("global", "probe-test");
        let probe = spawn_cmd_probe(&id, "sleep 30", &std::env::temp_dir(), 0, None, &[]);
        let CmdProbe {
            cancel_tx,
            result_rx,
        } = probe;
        let _ = cancel_tx.send(());
        let status = result_rx.await.unwrap().unwrap();
        assert!(!status.success());
    }

    #[tokio::test]
    async fn spawn_cmd_probe_receives_daemon_and_resolved_port_environment() {
        let id = DaemonId::new("worktree", "api");
        let daemon_env = IndexMap::from([("CUSTOM_VALUE".to_string(), "yes".to_string())]);
        // Written for the platform's default shell: cmd.exe reads the probe
        // as written, so it needs cmd's own `%VAR%` syntax there.
        let check = if cfg!(windows) {
            r#"(if "%CUSTOM_VALUE%"=="yes" if "%PORT%"=="4100" if "%PORT0%"=="4100" if "%PORT1%"=="5100" if "%PITCHFORK_DAEMON_ID%"=="worktree/api" if "%PITCHFORK_RETRY_COUNT%"=="2" exit 0) & exit 1"#
        } else {
            r#"test "$CUSTOM_VALUE" = yes && test "$PORT" = 4100 && test "$PORT0" = 4100 && test "$PORT1" = 5100 && test "$PITCHFORK_DAEMON_ID" = worktree/api && test "$PITCHFORK_RETRY_COUNT" = 2"#
        };
        let probe = spawn_cmd_probe(
            &id,
            check,
            &std::env::temp_dir(),
            2,
            Some(&daemon_env),
            &[4100, 5100],
        );
        let status = probe.result_rx.await.unwrap().unwrap();
        assert!(status.success());
    }

    #[test]
    fn configured_ready_port_follows_expected_port_bump() {
        assert_eq!(resolve_configured_ready_port(3000, &[3000], &[3004]), 3004);
        assert_eq!(
            resolve_configured_ready_port(4000, &[3000, 4000], &[3003, 4003]),
            4003
        );
        assert_eq!(resolve_configured_ready_port(8080, &[3000], &[3004]), 8080);
    }
}

#[cfg(test)]
mod launch_command_tests {
    use super::{invalid_argv_program, launch_command};
    use crate::daemon_id::DaemonId;

    #[test]
    fn refuses_an_argv_whose_program_rendered_empty_or_to_exec() {
        let id = DaemonId::new("proj", "api");
        assert!(invalid_argv_program(&id, &[]).is_some());
        assert!(invalid_argv_program(&id, &words(&["", "server.js"])).is_some());
        assert!(invalid_argv_program(&id, &words(&["exec", "node"])).is_some());
        assert_eq!(invalid_argv_program(&id, &words(&["node", ""])), None);
    }

    fn words(words: &[&str]) -> Vec<String> {
        words.iter().map(|w| w.to_string()).collect()
    }

    #[test]
    fn starts_the_first_word_with_the_rest_as_arguments() {
        let argv = words(&["node", "my server.js", "--name=\"a b\"", "&", "%PATH%"]);
        assert_eq!(
            launch_command(argv, None),
            (
                "node".to_string(),
                words(&["my server.js", "--name=\"a b\"", "&", "%PATH%"])
            )
        );
    }

    #[test]
    fn mise_receives_every_word_after_the_separator() {
        let argv = words(&["node", "my server.js", "'single'"]);
        let mise = std::path::Path::new("/opt/mise/bin/mise");
        let (program, args) = launch_command(argv, Some(mise));
        assert_eq!(program, mise.to_string_lossy());
        assert_eq!(
            args,
            words(&["x", "--", "node", "my server.js", "'single'"])
        );
    }
}

#[cfg(test)]
mod output_reader_tests {
    use super::{forward_output_lines, output_line_text};

    #[test]
    fn line_endings_are_removed() {
        assert_eq!(output_line_text(b"ready\n"), "ready");
        // A PTY's ONLCR, and a program that already wrote `\r\n` through it.
        assert_eq!(output_line_text(b"ready\r\n"), "ready");
        assert_eq!(output_line_text(b"ready\r\r\n"), "ready");
        // The last line of the output may have no newline at all.
        assert_eq!(output_line_text(b"ready"), "ready");
    }

    #[tokio::test]
    async fn a_line_that_is_not_utf8_does_not_end_the_read() {
        // What cut the output off before: `next_line` fails on the second
        // line, and nothing after it was read.
        let output: &[u8] = b"before\nbad \xff\xfe bytes\nafter 1\nafter 2\nlast";
        let (tx, mut rx) = tokio::sync::mpsc::channel(16);
        forward_output_lines(tokio::io::BufReader::new(output), tx).await;

        let mut lines = Vec::new();
        while let Some(line) = rx.recv().await {
            lines.push(line.text);
        }
        assert_eq!(lines.len(), 5, "{lines:?}");
        assert_eq!(lines[0], "before");
        assert!(lines[1].starts_with("bad "), "{:?}", lines[1]);
        assert_eq!(&lines[2..], ["after 1", "after 2", "last"]);
    }
}

#[cfg(test)]
mod output_reader_eio_tests {
    use super::forward_output_lines;
    use std::pin::Pin;
    use std::task::{Context, Poll};

    /// Output that ends as a Linux PTY master does once the slave closes:
    /// the last bytes, with no newline, then `EIO` instead of end of file.
    struct EndsWithEio(Option<&'static [u8]>);

    impl tokio::io::AsyncRead for EndsWithEio {
        fn poll_read(
            mut self: Pin<&mut Self>,
            _cx: &mut Context<'_>,
            buf: &mut tokio::io::ReadBuf<'_>,
        ) -> Poll<std::io::Result<()>> {
            match self.0.take() {
                Some(bytes) => {
                    buf.put_slice(bytes);
                    Poll::Ready(Ok(()))
                }
                None => Poll::Ready(Err(std::io::Error::from_raw_os_error(5))),
            }
        }
    }

    #[tokio::test]
    async fn a_last_line_cut_off_by_eio_is_still_forwarded() {
        let reader = tokio::io::BufReader::new(EndsWithEio(Some(b"first\nlast without newline")));
        let (tx, mut rx) = tokio::sync::mpsc::channel(16);
        forward_output_lines(reader, tx).await;

        let mut lines = Vec::new();
        while let Some(line) = rx.recv().await {
            lines.push(line.text);
        }
        assert_eq!(lines, ["first", "last without newline"]);
    }
}