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::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::settings;
use crate::shell::Shell;
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::io::AsyncBufReadExt;
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
    }
}

#[cfg(unix)]
#[derive(Clone, Debug, PartialEq, Eq)]
enum RunIdentity {
    Inherit,
    Switch {
        uid: nix::unistd::Uid,
        gid: nix::unistd::Gid,
        username: Option<CString>,
    },
}

/// 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.
struct CmdProbe {
    cancel_tx: tokio::sync::oneshot::Sender<()>,
    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());
        }
    }
}

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 configured general.shell setting (same as daemon run and hooks)
    // instead of default_for_platform(). On Windows, default_for_platform()
    // returns Shell::Cmd which cannot parse Unix-style commands like
    // "sleep 1; true". Falls back to default_for_platform() if the setting
    // is empty or unparseable.
    let shell_setting = settings().general.shell.clone();
    let mut command = match shell_words::split(&shell_setting) {
        Ok(parts) if !parts.is_empty() => {
            let (program, args) = parts.split_first().unwrap();
            let mut c = tokio::process::Command::new(program);
            c.args(args);
            c.arg(cmd);
            c
        }
        _ => Shell::default_for_platform().command(cmd),
    };
    command
        .current_dir(dir)
        .stdout(std::process::Stdio::null())
        .stderr(std::process::Stdio::null())
        .kill_on_drop(true);
    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 readiness 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)
}

/// 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);

impl Supervisor {
    /// Run a daemon, handling retries if configured
    pub async fn run(&self, opts: RunOptions) -> 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);
        let daemon = self.get_daemon(id).await;
        if let Some(daemon) = daemon {
            // Stopping state is treated as "not running" - the monitoring task will clean it up
            // Only check for Running state with a valid PID
            if !daemon.status.is_stopping()
                && !daemon.status.is_stopped()
                && let Some(pid) = daemon.pid
            {
                if opts.force {
                    self.stop_locked(id).await?;
                    info!("run: stop completed for daemon {id}");
                } else {
                    warn!("daemon {id} already running with pid {pid}");
                    return Ok(IpcResponse::DaemonAlreadyRunning);
                }
            }
        }

        // If wait_ready is true and retry is configured, implement retry loop
        if opts.wait_ready && opts.retry.count() > 0 {
            // 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 guard = match stop_guard.take() {
                    Some(guard) => guard,
                    None => self.stop_lock(id).await.lock_owned().await,
                };
                let result = self.run_once(retry_opts, guard).await?;

                match result {
                    IpcResponse::DaemonReady { daemon } => {
                        return Ok(IpcResponse::DaemonReady { daemon });
                    }
                    IpcResponse::DaemonFailedWithCode { exit_code } => {
                        if attempt < opts.retry.count() {
                            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(),
                                vec![],
                            )
                            .await;
                            time::sleep(Duration::from_secs(backoff_secs)).await;
                            continue;
                        } else {
                            info!("daemon {id} failed after {max_attempts} attempts");
                            return Ok(IpcResponse::DaemonFailedWithCode { exit_code });
                        }
                    }
                    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
    }

    /// 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()),
            )
        };

        // Parse the configured shell (default "sh -c") 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 shell_setting = settings().general.shell.clone();
        let shell_parts = match shell_words::split(&shell_setting) {
            Ok(parts) if !parts.is_empty() => parts,
            Ok(_) => {
                return Ok(IpcResponse::DaemonFailed {
                    error: "general.shell setting is empty".to_string(),
                });
            }
            Err(e) => {
                return Ok(IpcResponse::DaemonFailed {
                    error: format!("failed to parse general.shell setting {shell_setting:?}: {e}"),
                });
            }
        };
        let (shell_program, shell_args) = shell_parts.split_first().unwrap();

        // 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 run_script = opts
            .run
            .clone()
            .unwrap_or_else(|| shell_words::join(&original_cmd));

        let (program, args) = if opts.mise.unwrap_or(settings().general.mise) {
            match settings().resolve_mise_bin() {
                Some(mise_bin) => {
                    let mise_bin_str = mise_bin.to_string_lossy().to_string();
                    info!("daemon {id}: wrapping command with mise ({mise_bin_str})");
                    let mut args = vec!["x".to_string(), "--".to_string()];
                    args.push(shell_program.clone());
                    args.extend(shell_args.iter().cloned());
                    args.push(run_script);
                    (mise_bin_str, args)
                }
                None => {
                    warn!("daemon {id}: mise=true but mise binary not found, running without mise");
                    let mut args: Vec<String> = shell_args.to_vec();
                    args.push(run_script);
                    (shell_program.clone(), args)
                }
            }
        } else {
            let mut args: Vec<String> = shell_args.to_vec();
            args.push(run_script);
            (shell_program.clone(), args)
        };
        #[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:?}");

        // 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());
        }

        cmd.args(&args).current_dir(&opts.dir);

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

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

        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, &opts.slug);

        #[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()?;
        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);
        }
        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 = 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();
        let ready_delay = opts.ready_delay;
        let ready_output = opts.ready_output.clone();
        let ready_http = opts.ready_http.clone();
        let ready_port = effective_ready_port;
        let implicit_ready_port = ready_port.map(|p| ReadyPort {
            port: Some(p),
            template: None,
            timeout: None,
        });
        let ready_port_config = opts.ready_port.clone().or(implicit_ready_port);
        let ready_cmd = opts.ready_cmd.clone();
        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();
        let readiness_daemon_env = opts.env.clone();
        let readiness_resolved_ports = daemon.resolved_port.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));
        // The first expected port, if configured. When the ready_port check
        // succeeds on this exact 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 expected_port: Option<u16> = opts.port.as_ref().and_then(|p| p.expect.first().copied());
        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)))
                .lines()
        });
        #[cfg(not(unix))]
        let pty_reader: Option<tokio::io::Lines<tokio::io::BufReader<tokio::fs::File>>> = None;
        let stdout_reader = if pty_reader.is_none() {
            child
                .stdout
                .take()
                .map(|s| tokio::io::BufReader::new(s).lines())
        } else {
            None
        };
        let stderr_reader = if pty_reader.is_none() {
            child
                .stderr
                .take()
                .map(|s| tokio::io::BufReader::new(s).lines())
        } 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(mut 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(async move {
                    while let Ok(Some(mut line)) = reader.next_line().await {
                        // PTY slave uses ONLCR: \n → \r\n; strip the trailing \r.
                        if line.ends_with('\r') {
                            line.pop();
                        }
                        if output_tx
                            .send(super::OutputLine {
                                text: line,
                                source: super::OutputSource::Local,
                            })
                            .await
                            .is_err()
                        {
                            break;
                        }
                    }
                });
            } 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(mut stdout) = stdout_reader {
                    let tx = output_tx.clone();
                    tokio::spawn(async move {
                        while let Ok(Some(line)) = stdout.next_line().await {
                            if tx
                                .send(super::OutputLine {
                                    text: line,
                                    source: super::OutputSource::Local,
                                })
                                .await
                                .is_err()
                            {
                                break;
                            }
                        }
                    });
                }
                if let Some(mut stderr) = stderr_reader {
                    let tx = output_tx.clone();
                    tokio::spawn(async move {
                        while let Ok(Some(line)) = stderr.next_line().await {
                            if tx
                                .send(super::OutputLine {
                                    text: line,
                                    source: super::OutputSource::Local,
                                })
                                .await
                                .is_err()
                            {
                                break;
                            }
                        }
                    });
                }
                // 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;
            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 {
                            if let Some(tx) = ready_tx.take() {
                                // 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 {
                                    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(), 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.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(), 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(), 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 configured expected 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 expected_port == Some(port) {
                                            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, 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(), 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;
                        }
                    } => {
                        if !ready_notified && ready_pattern.is_none() && ready_http.is_none() && ready_port.is_none() && ready_cmd.is_none() {
                            // Check if the process already exited or is exiting before
                            // declaring it ready. On Windows, sleep(0) fires before
                            // child.wait() detects the exit, causing pitchfork start to
                            // return success for a daemon that already failed.
                            if exit_status.is_some() {
                                debug!("daemon {id} exited during ready_delay, not marking as ready");
                            } else {
                                // Force-refresh sysinfo for this PID before checking.
                                // On Windows, the cached process list may be stale.
                                PROCS.refresh_pids(&[daemon_pid]);
                                if !PROCS.is_running(daemon_pid) {
                                    debug!("daemon {id} pid {daemon_pid} not running during ready_delay, deferring to exit handler");
                                } else {
                                    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(), vec![]).await;
                                }
                            }
                            // 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;
                        if !active_port_spawned && has_port_config {
                            active_port_spawned = true;
                            detect_and_store_active_port(id.clone(), daemon_pid);
                        }
                    }
                    _ = 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
                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"),
            };

            // 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) = match exit_reason {
                    "stop" | "exit" => (
                        DaemonStatus::Stopped,
                        exit_status.as_ref().map(|s| s.success()).unwrap_or(true),
                    ),
                    _ => (DaemonStatus::Errored(exit_code), false),
                };
                // 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");
                }
            }

            // --- 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_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");
                    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 })
                }
                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.
    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}");
        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 {
                    debug!("pid {pid} not running, process may have exited unexpectedly");
                    // Process already dead — 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");
                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();
    resolve_run_identity(
        configured,
        current_uid,
        current_gid,
        std::env::var("SUDO_UID").ok().as_deref(),
        std::env::var("SUDO_GID").ok().as_deref(),
    )
}

#[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)
    {
        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),
    })
}

#[cfg(unix)]
fn run_identity_from_raw_ids(uid: u32, gid: u32, username: Option<CString>) -> RunIdentity {
    RunIdentity::Switch {
        uid: nix::unistd::Uid::from_raw(uid),
        gid: nix::unistd::Gid::from_raw(gid),
        username,
    }
}

#[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 username = nix::unistd::User::from_uid(nix::unistd::Uid::from_raw(uid))
        .ok()
        .flatten()
        .and_then(|u| CString::new(u.name).ok());
    Some(run_identity_from_raw_ids(uid, gid, username))
}

#[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()
    ))
}

#[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)
}

/// Spawn a background task that detects the first port the daemon process is listening on
/// 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. If the daemon has `expected_port` configured, prefer the first port from that list
///    (it is the port the operator explicitly designated as the primary service port).
/// 2. Otherwise, take the first port the process is actually listening on (in the order
///    returned by the OS), which is typically the port bound earliest.
///
/// Using `min()` (lowest port number) was previously used here but is incorrect: many
/// applications listen on multiple ports (e.g. HTTP + metrics) and the lowest-numbered
/// port is not necessarily the primary service port.
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 expected_port
            // in a single lock acquisition to avoid TOCTOU and unnecessary lock overhead.
            let expected_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) => d
                        .port
                        .as_ref()
                        .and_then(|p| p.expect.first().copied())
                        .filter(|&p| p > 0),
                    None => None,
                }
            };

            let active_port = 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();

                let process_ports: Vec<u16> = listeners
                    .into_iter()
                    .filter(|listener| descendant_pids.contains(&listener.process.pid))
                    .map(|listener| listener.socket.port())
                    .filter(|&port| port > 0)
                    .collect();

                if process_ports.is_empty() {
                    return None;
                }

                // Prefer the configured expected_port if the process is actually
                // listening on it; otherwise fall back to the first port found.
                if let Some(ep) = expected_port
                    && process_ports.contains(&ep)
                {
                    return Some(ep);
                }

                // No expected_port match — return the first port in the list.
                // The list order reflects the order the OS reports listeners,
                // which is generally the order they were bound (earliest first).
                // Do NOT sort: the lowest-numbered port is not necessarily the
                // primary service port (e.g. HTTP vs metrics).
                process_ports.into_iter().next()
            })
            .await
            .ok()
            .flatten();

            if let Some(port) = active_port {
                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;
                    }
                }
                return;
            }

            debug!(
                "daemon {id}: no active port detected for pid {pid} or its descendants (will retry)"
            );
        }

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

/// 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(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);
    }
}

/// 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)
/// - `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, slug: &Option<String>) {
    let s = crate::settings::settings();
    let lan_enabled = s.proxy.lan || !s.proxy.lan_ip.is_empty();

    if should_force_loopback_host(slug) && !lan_enabled {
        // Only force loopback binding for daemons that are actually routed via a slug.
        // 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 has a slug and proxy is enabled)
    if let Some(url) = build_pitchfork_url(slug, &s) {
        cmd.env("PITCHFORK_URL", &url);
    }

    // NODE_EXTRA_CA_CERTS: let Node.js backends trust the pitchfork CA
    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() {
            cmd.env("NODE_EXTRA_CA_CERTS", ca_path.to_string_lossy().to_string());
        }
    }

    // __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");
    }
}

fn should_force_loopback_host(slug: &Option<String>) -> bool {
    let Some(slug) = slug.as_deref() else {
        return false;
    };

    let s = crate::settings::settings();
    if !s.proxy.enable {
        return false;
    }

    let slugs = crate::pitchfork_toml::PitchforkToml::read_global_slugs();
    slugs.contains_key(slug)
}

/// Compute the public proxy URL for a daemon.
///
/// Returns `None` if the daemon has no slug or the proxy is not enabled.
fn build_pitchfork_url(slug: &Option<String>, s: &crate::settings::Settings) -> Option<String> {
    let slug = slug.as_ref()?;
    if !s.proxy.enable {
        return None;
    }
    let scheme = if s.proxy.https { "https" } else { "http" };
    let port = u16::try_from(s.proxy.port).ok().filter(|&p| p > 0)?;
    let port_suffix = if (scheme == "https" && port == 443) || (scheme == "http" && port == 80) {
        String::new()
    } else {
        format!(":{port}")
    };
    let lan_enabled = s.proxy.lan || !s.proxy.lan_ip.is_empty();
    let tld = if lan_enabled { "local" } else { &s.proxy.tld };
    Some(format!("{scheme}://{slug}.{tld}{port_suffix}",))
}

#[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".to_string(),
            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())]);
        let probe = spawn_cmd_probe(
            &id,
            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"#,
            &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);
    }
}