pitchfork_cli/supervisor/

mod.rs

//! Supervisor module - daemon process supervisor
//!
//! This module is split into focused submodules:
//! - `state`: State access layer (get/set operations)
//! - `lifecycle`: Daemon start/stop operations
//! - `adopt`: Re-adoption of orphaned daemons after a supervisor crash
//! - `log_sink`: Out-of-process capture of daemon output
//! - `autostop`: Autostop logic and boot daemon startup
//! - `retry`: Retry logic with backoff
//! - `watchers`: Background tasks (interval, cron, file watching)
//! - `ipc_handlers`: IPC request dispatch

mod adopt;
mod autostop;
mod health;
mod hooks;
mod idle;
mod ipc_handlers;
mod lifecycle;
mod log_sink;
#[cfg(unix)]
mod pty;
mod retry;
mod state;
mod watchers;

use crate::daemon_id::DaemonId;
use crate::daemon_status::DaemonStatus;
use crate::deps::compute_reverse_stop_order;
use crate::ipc::server::{IpcServer, IpcServerHandle, StartupLock};

use crate::procs::PROCS;
use crate::settings::settings;
use crate::state_file::StateFile;
use crate::{Result, env};
#[cfg(unix)]
use duct::cmd;
#[cfg(unix)]
use miette::IntoDiagnostic;
use once_cell::sync::Lazy;
use std::collections::HashMap;
#[cfg(unix)]
use std::collections::HashSet;
use std::fs;
#[cfg(unix)]
use std::os::unix::fs::PermissionsExt;
use std::path::PathBuf;
use std::process::exit;
use std::sync::atomic;
use std::sync::atomic::{AtomicBool, AtomicU32};
use std::time::Duration;
#[cfg(unix)]
use tokio::signal::unix::SignalKind;
use tokio::sync::{Mutex, Notify};
use tokio::task::JoinHandle;
use tokio::{signal, time};

/// Exit statuses reaped by the container-mode zombie reaper for managed daemon
/// PIDs. On non-Linux Unix platforms where `waitid(WNOWAIT)` is unavailable,
/// `waitpid(None, WNOHANG)` may race with Tokio's `child.wait()`. When the
/// zombie reaper wins, the exit status is stashed here so the monitoring task
/// in lifecycle.rs can recover it instead of treating the ECHILD as a failure.
///
/// On Linux this map is unused because the reaper uses `waitid` with `WNOWAIT`
/// to peek before reaping, which avoids the race entirely.
#[cfg(all(unix, not(target_os = "linux")))]
pub(crate) static REAPED_STATUSES: Lazy<Mutex<HashMap<u32, i32>>> =
    Lazy::new(|| Mutex::new(HashMap::new()));

// Re-export types needed by other modules
pub(crate) use state::UpsertDaemonOpts;

pub struct Supervisor {
    pub(crate) state_file: Mutex<StateFile>,
    pub(crate) pending_notifications: Mutex<Vec<(log::LevelFilter, String)>>,
    pub(crate) last_refreshed_at: Mutex<time::Instant>,
    /// Daemons whose retry sequence a foreground `run` is already working
    /// through, each with the flag that asks it to stop. The backoff between
    /// its attempts leaves the record errored with no PID, which is exactly
    /// what `check_retry` looks for, so without this the background checker
    /// would start the next attempt itself and the foreground call would be
    /// left reporting on a run it does not own. `stop` raises the flag, so the
    /// sequence ends rather than starting another attempt behind the user's
    /// back.
    /// One flag per claim: two starts can be working through the same
    /// daemon's retries at once, and a stop has to reach all of them.
    pub(crate) retrying:
        std::sync::Mutex<HashMap<DaemonId, Vec<std::sync::Arc<std::sync::atomic::AtomicBool>>>>,
    /// How many times each daemon has been stopped. The retry checker reads
    /// this when it decides to run an attempt and again when it is about to
    /// start one, holding the daemon's lock: a stop in between means the
    /// attempt it approved is one the user has since called off.
    pub(crate) stop_epochs: std::sync::Mutex<HashMap<DaemonId, u64>>,
    /// Map of daemon ID to scheduled autostop time
    pub(crate) pending_autostops: Mutex<HashMap<DaemonId, time::Instant>>,
    /// Autostop stops that have been spawned as detached tasks but have not
    /// yet begun stopping. `cancel_pending_autostops_for_dir` flips the flag
    /// to call off the stop when a shell re-enters the directory while the
    /// stop task is still in flight.
    pub(crate) in_flight_autostops:
        Mutex<HashMap<DaemonId, std::sync::Arc<std::sync::atomic::AtomicBool>>>,
    /// Handle for graceful IPC server shutdown
    pub(crate) ipc_shutdown: Mutex<Option<IpcServerHandle>>,
    /// Tracks in-flight hook tasks so shutdown can wait for them to complete
    pub(crate) hook_tasks: Mutex<Vec<JoinHandle<()>>>,
    /// Number of monitoring tasks that are still running (between process exit
    /// and hook registration completion). Used by `close()` to know when it is
    /// safe to drain `hook_tasks`.
    pub(crate) active_monitors: AtomicU32,
    /// Signalled by each monitoring task after it finishes registering hooks
    /// (or decides it has nothing to register). `close()` waits on this.
    pub(crate) monitor_done: Notify,
    /// Cancellation token for the proxy server — cancelled on shutdown to
    /// stop accepting new connections and drain in-flight ones.
    pub(crate) proxy_cancel: Mutex<Option<tokio_util::sync::CancellationToken>>,
    /// Join handle for the proxy task so shutdown can wait for cleanup.
    pub(crate) proxy_task: Mutex<Option<JoinHandle<()>>>,
    /// Join handle for the loopback DNS resolver, cancelled by `proxy_cancel`.
    pub(crate) dns_task: Mutex<Option<JoinHandle<()>>>,
    /// mDNS publisher for LAN mode (None if LAN mode is disabled).
    /// Shared with the LAN IP monitor task so it can re-publish on IP change.
    pub(crate) mdns_publisher:
        Mutex<Option<std::sync::Arc<tokio::sync::Mutex<crate::proxy::mdns::MdnsPublisher>>>>,
    /// Join handle for the LAN IP monitor task.
    pub(crate) lan_monitor_task: Mutex<Option<JoinHandle<()>>>,
    /// Cancellation token for the background state flush task.
    pub(crate) flush_cancel: std::sync::Mutex<Option<tokio_util::sync::CancellationToken>>,
    /// Daemons that currently have a live monitoring task (child `wait()`
    /// monitor or adopted-orphan poll monitor), keyed to the PID being
    /// monitored plus a unique registration token. Lets orphan
    /// reconciliation tell a supervised daemon from one whose monitor died
    /// with a previous supervisor process.
    pub(crate) monitored: std::sync::Mutex<HashMap<DaemonId, adopt::MonitorEntry>>,
    /// Where to deliver output a log sink reports over IPC.
    ///
    /// A daemon whose output is captured by a sink writes nothing this process
    /// reads, so the sink evaluates the daemon's readiness pattern itself and
    /// sends back the line that matched. The monitoring task registers here
    /// before the sink starts and unregisters when it ends, so a line arriving
    /// from a sink that outlived its daemon has nowhere to go and is dropped.
    pub(crate) sink_output: std::sync::Mutex<HashMap<DaemonId, log_sink::Relay>>,
    /// Per-daemon stop locks. A stop holds the daemon's lock for its whole
    /// duration (which now includes waiting for the entire process group to
    /// exit), and starts/orphan-cleanup acquire it first — serializing them
    /// against in-flight stops instead of racing the Stopping window.
    pub(crate) stop_locks: Mutex<HashMap<DaemonId, std::sync::Arc<tokio::sync::Mutex<()>>>>,
    /// Set when `close()` begins. From then on this supervisor's own
    /// state-file record is on its way out and must not be restored.
    pub(crate) shutting_down: AtomicBool,
}

/// A line of daemon output on its way to the monitoring task.
#[derive(Debug, Clone)]
pub(crate) struct OutputLine {
    pub(crate) text: String,
    pub(crate) source: OutputSource,
}

/// Where a line of daemon output came from, which decides what is left to do
/// with it.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum OutputSource {
    /// Read by this process from the daemon's stdout, stderr or PTY master.
    /// Nothing has been done with it yet.
    Local,
    /// Reported by the daemon's log sink, which has already written the line to
    /// the store — storing it again here would duplicate it.
    ///
    /// `fires_hook` is the sink's answer to whether this line passed the
    /// `on_output` hook's filter and debounce. It is carried rather than
    /// re-derived because a line can be reported for readiness alone, and
    /// firing a hook that filters for something else would be wrong.
    Sink { fires_hook: bool },
}

pub(crate) fn interval_duration() -> Duration {
    settings().general_interval()
}

pub static SUPERVISOR: Lazy<Supervisor> =
    Lazy::new(|| Supervisor::new().expect("Error creating supervisor"));

pub async fn start_if_not_running() -> Result<()> {
    let sf = StateFile::get();
    if let Some(d) = sf.daemons.get(&DaemonId::pitchfork())
        && supervisor_record_is_live(d)
    {
        return Ok(());
    }
    // The record can be missing or stale while a supervisor keeps serving
    // IPC, e.g. when state.toml was replaced or rewritten by another tool.
    // Starting another one would take over the socket and leave the running
    // supervisor, and every daemon it manages, unreachable.
    if crate::ipc::supervisor_listening().await {
        debug!("supervisor is listening on the IPC socket but not recorded in the state file");
        return Ok(());
    }
    start_in_background()
}

/// How long `supervisor start/run --force` waits for the supervisor it
/// replaces to stop serving IPC. A supervisor being stopped keeps its socket until its daemons
/// have stopped, which can take a while.
pub(crate) const IPC_SOCKET_RELEASE_TIMEOUT: Duration = Duration::from_secs(30);

/// Wait until no supervisor is listening on the IPC socket, failing if one
/// still is after [`IPC_SOCKET_RELEASE_TIMEOUT`].
///
/// Used when replacing a supervisor: the old one keeps serving IPC while it
/// shuts down, and binding the socket before it lets go would leave clients
/// talking to whichever supervisor they happen to reach.
pub(crate) async fn wait_for_ipc_socket_release() -> Result<()> {
    let deadline = time::Instant::now() + IPC_SOCKET_RELEASE_TIMEOUT;
    let mut waiting = false;
    while crate::ipc::supervisor_listening().await {
        if time::Instant::now() >= deadline {
            return Err(miette::miette!(
                "another pitchfork supervisor is still listening on {} after {}s; \
                 stop it with `pitchfork supervisor stop` before starting a new one",
                crate::ipc::socket_display(),
                IPC_SOCKET_RELEASE_TIMEOUT.as_secs()
            ));
        }
        if !waiting {
            info!(
                "waiting for the supervisor listening on {} to shut down",
                crate::ipc::socket_display()
            );
            waiting = true;
        }
        time::sleep(Duration::from_millis(100)).await;
    }
    Ok(())
}

/// Whether the supervisor's own state-file record still describes a live
/// pitchfork supervisor, rather than a stale entry whose PID the OS has since
/// handed to an unrelated process.
///
/// The record survives crashes and reboots, so a bare liveness probe on its
/// PID is not enough: after a reboot low PIDs go to early system daemons, and
/// a `kill(pid, 0)` on one of those says "alive". The check therefore also
/// requires the identity the supervisor recorded about itself at startup to
/// match the live process — see [`supervisor_identity_matches`]. When the
/// PID is alive but the identity does not match, the record is stale; callers
/// are free to overwrite it and must never signal that PID.
pub(crate) fn supervisor_record_is_live(record: &crate::daemon::Daemon) -> bool {
    let Some(pid) = record.pid else {
        return false;
    };
    if !PROCS.is_running(pid) {
        return false;
    }
    if record.start_time.is_none() && record.boot_time.is_none() {
        // A record from a supervisor older than v2.18.0 carries no identity
        // at all, so nothing can contradict it. Rather than trust any live
        // PID, require the process to at least be a pitchfork binary: that
        // rejects the reboot case (an unrelated system daemon on the PID)
        // while a still-running old supervisor stays recognisable.
        PROCS.refresh_pids(&[pid]);
        let title = PROCS.title(pid);
        if legacy_supervisor_title_matches(title.as_deref()) {
            return true;
        }
        warn!(
            "pid {pid} recorded for the supervisor by an older pitchfork is now {title:?}, not a pitchfork process; treating the record as stale"
        );
        return false;
    }
    if supervisor_identity_matches(
        record.start_time,
        PROCS.start_time(pid),
        record.boot_time,
        PROCS.boot_time(),
    ) {
        return true;
    }
    warn!(
        "pid {pid} recorded for the supervisor belongs to another process now (recorded start_time {:?} boot_time {:?}, live start_time {:?} boot_time {}); treating the record as stale",
        record.start_time,
        record.boot_time,
        PROCS.start_time(pid),
        PROCS.boot_time()
    );
    false
}

/// Whether a live process name can be a pitchfork supervisor. Only used for
/// legacy records that carry no start or boot time (see
/// [`supervisor_record_is_live`]); an unreadable name is not accepted, since
/// the record has nothing else vouching for it.
pub(crate) fn legacy_supervisor_title_matches(title: Option<&str>) -> bool {
    title.is_some_and(|t| t.to_ascii_lowercase().starts_with("pitchfork"))
}

/// Whether a live process can be the supervisor a state-file record describes.
///
/// The kernel start token is the identity, as in [`process_identity_matches`]:
/// when both the recorded and the live token can be read, they alone decide.
/// Equal tokens mean the same process generation; different tokens mean the
/// PID was recycled, within this boot or across a reboot.
///
/// The recorded boot time is only consulted when a token is missing on either
/// side. It must not veto matching tokens: on Linux and macOS the reported
/// boot time is derived from the realtime clock, so an NTP step or a
/// sleep/resume moves it while the supervisor keeps running, and treating that
/// as a reboot would spawn a second supervisor and orphan the first. Without
/// tokens, though, a boot time from a previous boot is the one thing that can
/// still prove the record stale, and a record predating both fields is not
/// contradicted by anything (callers apply a weaker check to those).
pub(crate) fn supervisor_identity_matches(
    recorded_start_time: Option<u64>,
    current_start_time: Option<u64>,
    recorded_boot_time: Option<u64>,
    current_boot_time: u64,
) -> bool {
    match (recorded_start_time, current_start_time) {
        (Some(recorded), Some(current)) => recorded == current,
        _ => recorded_boot_time.is_none_or(|recorded| {
            recorded.abs_diff(current_boot_time) <= BOOT_TIME_TOLERANCE_SECS
        }),
    }
}

pub fn start_in_background() -> Result<()> {
    debug!("starting supervisor in background");
    // Ensure the log directory exists so we can redirect stderr there.
    // Panics and other fatal errors from the background supervisor process
    // would otherwise be silently swallowed.
    let log_file = &*env::PITCHFORK_LOG_FILE;
    if let Some(parent) = log_file.parent() {
        let _ = fs::create_dir_all(parent);
    }
    #[cfg(unix)]
    fix_state_dir_permissions();

    // On Unix, use duct with stderr redirected to the log file.
    #[cfg(unix)]
    {
        let stderr_file = fs::OpenOptions::new()
            .create(true)
            .append(true)
            .open(log_file)
            .into_diagnostic()?;
        // Computed here rather than in the child: sysconf is not on the
        // async-signal-safe list.
        let max_fd = max_inherited_fd();
        cmd!(&*env::PITCHFORK_BIN, "supervisor", "run")
            .env_remove("PITCHFORK_CONFIG")
            .stdin_null()
            .stdout_null()
            .stderr_file(stderr_file)
            .before_spawn(move |cmd| {
                use std::os::unix::process::CommandExt;
                // SAFETY: the hook only issues close_range/fcntl syscalls,
                // which are async-signal-safe, and does not allocate.
                unsafe {
                    cmd.pre_exec(move || {
                        cloexec_inherited_fds(max_fd);
                        Ok(())
                    });
                }
                Ok(())
            })
            .start()
            .into_diagnostic()?;
    }

    // On Windows, use CreateProcessW directly with bInheritHandles=FALSE.
    // std::process::Command always sets bInheritHandles=TRUE when any stdio
    // handle is configured (even Stdio::null()), which causes the background
    // supervisor to inherit ALL inheritable handles from the parent —
    // including bats' stdout capture pipe. The supervisor keeps the pipe
    // open after the CLI exits, and bats hangs forever waiting for EOF.
    //
    // CreateProcessW with bInheritHandles=FALSE prevents any handle
    // inheritance. We pass NUL device handles for stdin/stdout/stderr
    // via STARTUPINFO without inheriting any parent handles.
    #[cfg(windows)]
    {
        use windows_sys::Win32::Foundation::{CloseHandle, FALSE};
        use windows_sys::Win32::System::Threading::{
            CREATE_NO_WINDOW, CREATE_UNICODE_ENVIRONMENT, CreateProcessW, DETACHED_PROCESS,
            PROCESS_INFORMATION, STARTUPINFOW,
        };

        // With bInheritHandles=FALSE, the child inherits NO parent handles.
        // The supervisor uses its own internal file-based logger
        // (PITCHFORK_LOG_FILE), so it doesn't need stdio from the parent.
        // We don't set STARTF_USESTDHANDLES because that flag requires
        // bInheritHandles=TRUE to function correctly per Microsoft docs.
        // Without stdio handles, the detached process gets null stdio by
        // default, which is exactly what we want.
        let mut si: STARTUPINFOW = unsafe { std::mem::zeroed() };
        si.cb = std::mem::size_of::<STARTUPINFOW>() as u32;

        let bin_path = &*env::PITCHFORK_BIN;
        let mut cmd_line: Vec<u16> = format!("\"{}\" supervisor run\0", bin_path.to_string_lossy())
            .encode_utf16()
            .collect();

        use std::os::windows::ffi::OsStrExt;
        let mut vars: Vec<_> = std::env::vars_os()
            .filter(|(k, _)| !k.to_string_lossy().eq_ignore_ascii_case("PITCHFORK_CONFIG"))
            .collect();
        vars.sort_by_key(|(k, _)| k.to_string_lossy().to_uppercase());
        let mut environment = Vec::<u16>::new();
        for (key, value) in vars {
            environment.extend(key.encode_wide());
            environment.push(b'=' as u16);
            environment.extend(value.encode_wide());
            environment.push(0);
        }
        environment.extend([0, 0]);
        let mut pi: PROCESS_INFORMATION = unsafe { std::mem::zeroed() };
        let ok = unsafe {
            CreateProcessW(
                std::ptr::null(),
                cmd_line.as_mut_ptr(),
                std::ptr::null(),
                std::ptr::null(),
                FALSE, // bInheritHandles = FALSE — the whole point
                DETACHED_PROCESS | CREATE_NO_WINDOW | CREATE_UNICODE_ENVIRONMENT,
                environment.as_ptr().cast(),
                std::ptr::null(),
                &si,
                &mut pi,
            )
        };

        if ok == 0 {
            return Err(miette::miette!(
                "CreateProcessW failed for supervisor: {}",
                std::io::Error::last_os_error()
            ));
        }

        // Close process/thread handles — we don't need them (detached process).
        unsafe {
            CloseHandle(pi.hProcess);
            CloseHandle(pi.hThread);
        }
    }

    Ok(())
}

/// Upper bound (exclusive) for the fd scan in [`cloexec_fd_scan`], used when
/// `close_range` is unavailable (macOS, Linux < 5.11).
#[cfg(unix)]
fn max_inherited_fd() -> libc::c_int {
    // SAFETY: sysconf has no preconditions.
    let open_max = unsafe { libc::sysconf(libc::_SC_OPEN_MAX) };
    // RLIMIT_NOFILE alone is not a safe bound: a caller can open a
    // high-numbered descriptor and then lower the limit before running us.
    // Also list the fds that are actually open (both macOS and Linux provide
    // /dev/fd) so the scan reaches every one of them.
    let open_fds = std::fs::read_dir("/dev/fd")
        .into_iter()
        .flatten()
        .filter_map(|entry| entry.ok()?.file_name().to_str()?.parse().ok());
    fd_scan_end(open_max, open_fds)
}

#[cfg(unix)]
fn fd_scan_end(
    open_max: libc::c_long,
    open_fds: impl IntoIterator<Item = libc::c_int>,
) -> libc::c_int {
    // -1 means "no limit"; cap the scan so an unlimited or huge RLIMIT_NOFILE
    // doesn't make spawning the supervisor noticeably slow.
    let limit = if open_max <= 0 {
        1 << 16
    } else {
        open_max.min(1 << 20) as libc::c_int
    };
    let past_highest_open = open_fds
        .into_iter()
        .max()
        .map_or(0, |fd| fd.saturating_add(1));
    limit.max(past_highest_open)
}

/// Keep descriptors the CLI inherited without O_CLOEXEC from leaking into the
/// background supervisor (and from there into every daemon it spawns).
///
/// Callers routinely hand us such descriptors: bats' fd 3, or pipes from a
/// wrapping script. The long-lived supervisor would otherwise hold them open
/// forever, so whoever waits for EOF on the pipe (bats, `$(...)`, CI log
/// capture) hangs even after the CLI has exited. This is the Unix counterpart
/// of `bInheritHandles=FALSE` in the Windows branch of `start_in_background`.
///
/// Runs in the forked child after stdio has been set up on fds 0-2. It marks
/// fds >= 3 close-on-exec instead of closing them outright so that std's
/// internal exec-status pipe (already CLOEXEC) keeps working and exec failures
/// are still reported to the parent. Must stay async-signal-safe: no
/// allocation, no locks, syscalls only.
#[cfg(unix)]
fn cloexec_inherited_fds(max_fd: libc::c_int) {
    #[cfg(any(target_os = "linux", target_os = "android"))]
    {
        // SAFETY: close_range takes plain integer arguments. CLOSE_RANGE_CLOEXEC
        // needs Linux 5.11; older kernels return ENOSYS/EINVAL and we fall
        // through to the scan below.
        let ret = unsafe {
            libc::syscall(
                libc::SYS_close_range,
                3 as libc::c_uint,
                libc::c_uint::MAX,
                libc::CLOSE_RANGE_CLOEXEC as libc::c_uint,
            )
        };
        if ret == 0 {
            return;
        }
    }
    cloexec_fd_scan(max_fd);
}

/// Fallback for [`cloexec_inherited_fds`]: mark fds `3..max_fd` close-on-exec
/// one at a time. Async-signal-safe.
#[cfg(unix)]
fn cloexec_fd_scan(max_fd: libc::c_int) {
    for fd in 3..max_fd {
        // SAFETY: fcntl on an arbitrary fd number is safe; unused numbers
        // just fail with EBADF.
        unsafe {
            let flags = libc::fcntl(fd, libc::F_GETFD);
            if flags >= 0 && flags & libc::FD_CLOEXEC == 0 {
                libc::fcntl(fd, libc::F_SETFD, flags | libc::FD_CLOEXEC);
            }
        }
    }
}

/// Decide whether a project session should be removed during refresh.
///
/// `recorded_title` is the title snapshot taken at the start of refresh.
/// `session` is the current state entry re-read under the lock. If the state
/// has been updated since the snapshot (e.g., re-entered with the same
/// PID/dir but a new title), we must skip removal to avoid deleting the new
/// session. The host PID lives in the session key now, so there is no
/// `liveness_pid` field to compare against.
#[cfg(any(unix, test))]
fn should_remove_liveness_session(
    session: &crate::state_file::ProjectSession,
    recorded_title: &Option<String>,
    current_title: Option<&str>,
    is_running: bool,
) -> bool {
    // If the state was updated since the snapshot (e.g., re-entered with the
    // same PID/dir but a new title), skip removal to avoid deleting the new
    // session.
    if session.liveness_title.as_ref() != recorded_title.as_ref() {
        return false;
    }
    // Dead host process — evict.
    if !is_running {
        return true;
    }
    // Host is alive. Evict only on a real title mismatch (PID reuse). If no
    // title was recorded, or the current title is unavailable, we cannot
    // reliably detect PID reuse — keep the session rather than risk evicting
    // a live process.
    match (recorded_title.as_deref(), current_title) {
        (Some(recorded), Some(current)) => recorded != current,
        _ => false,
    }
}

impl Supervisor {
    pub fn new() -> Result<Self> {
        Ok(Self {
            state_file: Mutex::new(StateFile::read(&*env::PITCHFORK_STATE_FILE).unwrap_or_else(
                |e| {
                    warn!("failed to read state file, starting with empty state: {e}");
                    StateFile::new(env::PITCHFORK_STATE_FILE.clone())
                },
            )),
            last_refreshed_at: Mutex::new(time::Instant::now()),
            pending_notifications: Mutex::new(vec![]),
            retrying: std::sync::Mutex::new(HashMap::new()),
            stop_epochs: std::sync::Mutex::new(HashMap::new()),
            pending_autostops: Mutex::new(HashMap::new()),
            in_flight_autostops: Mutex::new(HashMap::new()),
            ipc_shutdown: Mutex::new(None),
            hook_tasks: Mutex::new(Vec::new()),
            active_monitors: AtomicU32::new(0),
            monitor_done: Notify::new(),
            proxy_cancel: Mutex::new(None),
            proxy_task: Mutex::new(None),
            dns_task: Mutex::new(None),
            mdns_publisher: Mutex::new(None),
            lan_monitor_task: Mutex::new(None),
            flush_cancel: std::sync::Mutex::new(None),
            monitored: std::sync::Mutex::new(HashMap::new()),
            sink_output: std::sync::Mutex::new(HashMap::new()),
            stop_locks: Mutex::new(HashMap::new()),
            shutting_down: AtomicBool::new(false),
        })
    }

    /// Get (or create) the per-daemon stop lock for `id`.
    pub(crate) async fn stop_lock(&self, id: &DaemonId) -> std::sync::Arc<tokio::sync::Mutex<()>> {
        self.stop_locks
            .lock()
            .await
            .entry(id.clone())
            .or_default()
            .clone()
    }

    pub async fn start(
        &self,
        is_boot: bool,
        container: bool,
        web_port: Option<u16>,
        web_path: Option<String>,
    ) -> Result<()> {
        // Ensure the state directory and its contents are accessible by non-root
        // users. This is needed when the supervisor is started with `sudo` — all
        // files it creates are owned by root, which prevents normal CLI clients
        // from reading/writing state or connecting to the IPC socket.
        #[cfg(unix)]
        fix_state_dir_permissions();

        // Refuse to run beside a supervisor that is already listening, and
        // do so before recording ourselves in the state file or starting any
        // daemons: taking over its socket would leave it running but
        // unreachable. (`--force` has already waited for the one it replaced
        // to let go of the socket.) The lock is held until our own listener
        // is bound, so a supervisor starting at the same time waits here and
        // then finds this one listening.
        let startup_lock = StartupLock::acquire().await?;
        if crate::ipc::supervisor_listening().await {
            return Err(miette::miette!(
                "another pitchfork supervisor is already listening on {}",
                crate::ipc::socket_display()
            ));
        }

        let pid = std::process::id();
        // Ensure PROCS has data for the supervisor PID before upsert_daemon reads title()
        PROCS.refresh_pids(&[pid]);
        // Determine container mode: CLI flag takes priority, then settings.
        // Running as PID 1 always enables it: orphaned descendants of daemons
        // re-parent to us, and without the zombie reaper they would accumulate
        // as unreaped zombies — which also keep their process group alive,
        // stalling whole-group stop waits indefinitely.
        let container_mode =
            container || settings().supervisor.container || std::process::id() == 1;
        if container_mode {
            info!("Starting supervisor in container/PID1 mode with pid {pid}");
        } else {
            info!("Starting supervisor with pid {pid}");
        }

        // Whether the previous supervisor exited uncleanly must be read before
        // we record ourselves in the state file just below (see
        // `supervisor_exited_uncleanly`); the background cleanup task runs
        // after that record exists, so it receives the answer instead of
        // reading it too late.
        let unclean = supervisor_exited_uncleanly(self).await;

        self.upsert_daemon(
            UpsertDaemonOpts::builder(DaemonId::pitchfork())
                .set(|o| {
                    o.pid = Some(pid);
                    o.status = DaemonStatus::Running;
                })
                .build(),
        )
        .await?;
        #[cfg(unix)]
        fix_state_dir_permissions();

        // Self-heal: if the boot registration points to a stale binary path
        // (e.g. after a brew/mise upgrade), re-register with the current path.
        // Runs in the background — must not block or fail supervisor startup.
        tokio::task::spawn_blocking(|| {
            if let Ok(boot_manager) = crate::boot_manager::BootManager::new() {
                boot_manager.check_and_reregister_if_stale();
            }
        });

        // If the previous supervisor died uncleanly, its daemon child processes
        // may still be alive (orphaned, re-parented to init).  Terminate them
        // before starting replacements so we don't end up with duplicate
        // processes holding the same ports.
        //
        // This runs in the background: each orphan kill now waits for its
        // whole process group to exit (seconds per orphan), and doing that
        // inline would delay IPC socket creation past the CLI's short connect
        // budget on autostart. Per-daemon stop locks serialize the cleanup
        // against any Run/Stop requests that arrive for the same daemon in the
        // meantime, and boot daemons start after cleanup completes so they
        // cannot observe an orphan as "already running".
        let boot_after_cleanup = is_boot;
        tokio::spawn(async move {
            cleanup_orphaned_daemons(&SUPERVISOR, unclean).await;
            if boot_after_cleanup {
                info!("Boot start mode enabled, starting boot_start daemons");
                if let Err(e) = SUPERVISOR.start_boot_daemons().await {
                    error!("failed to start boot daemons: {e}");
                }
            }
        });

        self.interval_watch()?;

        // Run the first cron check synchronously before starting the cron
        // watcher and IPC server. This registers config-only cron daemons and
        // fires any `immediate=true` triggers in the foreground, so they cannot
        // race with a concurrent `pitchfork start` IPC. By the time the cron
        // watcher's first tick runs, `last_cron_triggered` is already anchored
        // and the immediate daemons are already running.
        if let Err(e) = self.check_cron_schedules().await {
            error!("failed to check cron schedules on startup: {e}");
        }

        self.cron_watch()?;
        self.signals()?;
        self.daemon_file_watch()?;

        // In container mode, install SIGCHLD handler to reap orphaned/zombie processes
        #[cfg(unix)]
        if container_mode {
            self.reap_zombies()?;
        }

        // Start web server: CLI --web-port takes priority, then settings.web.auto_start + bind_port
        let s = settings();
        let effective_port = web_port.or_else(|| {
            if s.web.auto_start {
                match u16::try_from(s.web.bind_port).ok().filter(|&p| p > 0) {
                    Some(p) => Some(p),
                    None => {
                        error!(
                            "web.bind_port {} is out of valid port range (1-65535), web UI disabled",
                            s.web.bind_port
                        );
                        None
                    }
                }
            } else {
                None
            }
        });
        // CLI --web-path takes priority, then settings.web.base_path
        let effective_path = web_path.or_else(|| {
            let bp = s.web.base_path.clone();
            if bp.is_empty() { None } else { Some(bp) }
        });
        if let Some(port) = effective_port {
            tokio::spawn(async move {
                if let Err(e) = crate::web::serve(port, effective_path).await {
                    error!("Web server error: {e}");
                }
            });
        }

        // Start standalone API server if configured
        let api_port = if s.api.auto_start {
            match u16::try_from(s.api.bind_port).ok().filter(|&p| p > 0) {
                Some(p) => Some(p),
                None => {
                    error!(
                        "api.bind_port {} is out of valid port range (1-65535), API server disabled",
                        s.api.bind_port
                    );
                    None
                }
            }
        } else {
            None
        };
        if let Some(port) = api_port {
            tokio::spawn(async move {
                if let Err(e) = crate::web::serve_api(port, None).await {
                    error!("API server error: {e}");
                }
            });
        }

        // Start reverse proxy server if enabled
        if s.proxy.enable {
            // Pre-generate the TLS certificate synchronously before spawning the proxy
            // task. This ensures the cert exists immediately after `sup start` returns,
            // so `proxy trust` can be run right away without waiting for the async task.
            #[cfg(feature = "proxy-tls")]
            if s.proxy.https {
                let proxy_dir = crate::env::PITCHFORK_STATE_DIR.join("proxy");
                let ca_cert_path = proxy_dir.join("ca.pem");
                let ca_key_path = proxy_dir.join("ca-key.pem");
                // Checked and written under the CA lock: `proxy setup` may be
                // generating the same pair right now.
                match crate::proxy::server::ensure_ca(&ca_cert_path, &ca_key_path, || {
                    ca_cert_path.exists() && ca_key_path.exists()
                }) {
                    Ok(true) => {
                        info!(
                            "Generated local CA certificate at {}",
                            ca_cert_path.display()
                        );
                    }
                    Ok(false) => {}
                    Err(e) => {
                        error!("Failed to generate CA certificate: {e}");
                    }
                }

                // Auto-trust: attempt to install the CA certificate into the
                // system trust store. May fail silently due to permissions;
                // user can run `pitchfork proxy trust` manually.
                if s.proxy.auto_trust && ca_cert_path.exists() {
                    use crate::proxy::trust::{AutoTrustResult, auto_trust};
                    match auto_trust(&ca_cert_path) {
                        AutoTrustResult::AlreadyTrusted => {}
                        AutoTrustResult::Trusted => {
                            info!("CA certificate auto-trusted in system store");
                        }
                        AutoTrustResult::NotTrusted { reason } => {
                            warn!("Auto-trust skipped: {reason}");
                            warn!("Run `pitchfork proxy trust` to install manually");
                        }
                    }
                }
            }
            // Spawn the proxy server and wait for its bind result via a oneshot
            // channel.  This avoids the TOCTOU race of a pre-flight bind check
            // while still surfacing binding failures immediately.
            let (bind_tx, bind_rx) = tokio::sync::oneshot::channel();
            let proxy_cancel = tokio_util::sync::CancellationToken::new();
            let proxy_cancel_clone = proxy_cancel.clone();
            *self.proxy_cancel.lock().await = Some(proxy_cancel);
            let proxy_task = tokio::spawn(async move {
                if let Err(e) = crate::proxy::server::serve(bind_tx, proxy_cancel_clone).await {
                    error!("Proxy server error: {e}");
                }
            });
            *self.proxy_task.lock().await = Some(proxy_task);
            match bind_rx.await {
                Ok(Ok(())) => {
                    info!("Proxy server bound successfully");
                    // Resolved once and shared: mDNS and the DNS resolver
                    // must advertise the same address.
                    let lan_ip = self.resolve_lan_ip().await;
                    self.start_mdns(lan_ip).await;
                    self.start_dns_resolver(lan_ip).await;
                }
                Ok(Err(msg)) => {
                    error!("{msg}");
                    self.add_notification(log::LevelFilter::Error, msg).await;
                }
                Err(_) => {
                    // Sender dropped without sending — serve() panicked or
                    // returned before signalling.  Already logged by the
                    // spawn error handler above.
                }
            }
        }

        // Pre-warm slug cache so the first /api/proxies request is fast.
        // Spawned as a background task so it does not block startup.
        tokio::spawn(async {
            crate::proxy::server::get_cached_slugs().await;
        });

        let (ipc, ipc_handle) = IpcServer::new(startup_lock).await?;
        *self.ipc_shutdown.lock().await = Some(ipc_handle);
        self.start_state_flush_task();
        self.conn_watch(ipc).await
    }

    /// Start the loopback DNS resolver for the proxy TLD.
    ///
    /// The resolver shares the proxy's cancellation token, so it stops with the
    /// proxy. A bind failure is a notification rather than a fatal error: the
    /// proxy still works for anyone who reaches it some other way.
    async fn start_dns_resolver(&self, lan_ip: Option<std::net::Ipv4Addr>) {
        let s = crate::settings::settings();
        if !s.proxy.dns {
            return;
        }
        let cfg = crate::proxy::dns::config_from_settings(&s, lan_ip);
        let port = crate::proxy::dns::dns_port(&s);
        // Loopback only: the resolver is for this machine's stub resolver, and
        // LAN peers are served by mDNS instead.
        let addr = std::net::SocketAddr::from((std::net::Ipv4Addr::LOCALHOST, port));

        // The token's guard is held until the handle is stored. `close` takes
        // the token under this lock before it collects `dns_task`, so it either
        // runs first — and there is no token to spawn with — or waits until
        // the handle is in place to be drained. Releasing it earlier left a
        // window where `close` found no handle and the task outlived shutdown
        // holding the resolver's sockets.
        let cancel_guard = self.proxy_cancel.lock().await;
        let Some(cancel) = cancel_guard.clone() else {
            return;
        };
        let (bind_tx, bind_rx) = tokio::sync::oneshot::channel();
        let task = tokio::spawn(async move {
            if let Err(e) = crate::proxy::dns::serve(cfg, addr, bind_tx, cancel).await {
                error!("DNS resolver error: {e}");
            }
        });
        *self.dns_task.lock().await = Some(task);
        drop(cancel_guard);
        match bind_rx.await {
            Ok(Ok(())) => info!("DNS resolver bound successfully"),
            Ok(Err(msg)) => {
                let msg = format!(
                    "{msg}\nProxy host names will not resolve through pitchfork. \
                     Choose another port with proxy.dns_port, or set proxy.dns = false."
                );
                error!("{msg}");
                self.add_notification(log::LevelFilter::Error, msg).await;
            }
            Err(_) => {}
        }
    }

    /// Watch the LAN address and keep the DNS responder, and mDNS when it is
    /// running, pointed at the current one.
    ///
    /// Started even when the mDNS publisher could not be created: the DNS
    /// responder serves this machine regardless, and an address it keeps
    /// answering with after the interface has moved is worse than useless.
    ///
    /// Does nothing when `proxy.lan_ip` pins an address. That is a choice to
    /// respect, not a starting point to drift from — the check lives here so
    /// neither caller can forget it.
    async fn start_lan_ip_monitor(
        &self,
        initial_ip: std::net::Ipv4Addr,
        port: u16,
        publisher: Option<std::sync::Arc<tokio::sync::Mutex<crate::proxy::mdns::MdnsPublisher>>>,
    ) {
        if !crate::settings::settings().proxy.lan_ip.is_empty() {
            return;
        }
        // No cancellation token means `close` has already taken it, so shutdown
        // is under way. Spawning here would leave a task polling with no way to
        // stop it, and past the point where `close` collects the handle.
        //
        // Held until the handle is stored, for the reason `start_dns_resolver`
        // gives: otherwise `close` can collect `lan_monitor_task` in between.
        let cancel_guard = self.proxy_cancel.lock().await;
        let Some(cancel) = cancel_guard.clone() else {
            debug!("Not starting the LAN IP monitor: the supervisor is shutting down");
            return;
        };
        let monitor_cancel = Some(cancel);
        let task = tokio::spawn(async move {
            let mut last_ip = initial_ip;
            let mut ticker = tokio::time::interval(std::time::Duration::from_secs(5));
            ticker.tick().await; // first tick is immediate
            loop {
                // Cancellation is raced against the tick, not checked after
                // it. Checking afterwards means the task only notices once the
                // full interval has elapsed, so a shutdown that waits a second
                // for it always gives up and aborts instead — the graceful
                // path would never once be taken.
                match monitor_cancel.as_ref() {
                    Some(cancel) => {
                        tokio::select! {
                            _ = ticker.tick() => {}
                            _ = cancel.cancelled() => break,
                        }
                    }
                    None => {
                        ticker.tick().await;
                    }
                }
                if let Some(new_ip) = crate::proxy::lan_ip::detect_lan_ip_if_changed(last_ip).await
                {
                    log::info!("LAN IP changed: {last_ip} → {new_ip}");
                    last_ip = new_ip;
                    crate::proxy::dns::update_lan_ip(new_ip);
                    if let Some(publisher) = publisher.as_ref() {
                        publisher.lock().await.republish_all(new_ip, port);
                    }
                }
            }
        });
        *self.lan_monitor_task.lock().await = Some(task);
        drop(cancel_guard);
    }

    /// Start mDNS publishing for LAN mode (called after the proxy binds successfully).
    /// The LAN address mDNS publishes and the DNS resolver answers with.
    ///
    /// Resolved once and handed to both. Detecting separately in each let them
    /// disagree when the interface address changed in between, which would
    /// advertise one address over mDNS and serve another over DNS, and probed
    /// the network twice at startup for one answer.
    ///
    /// `None` means LAN mode is off, or is on and the address could not be
    /// determined; either way the caller has nothing to publish. The reason is
    /// reported here so it is said once rather than by each caller.
    async fn resolve_lan_ip(&self) -> Option<std::net::Ipv4Addr> {
        let s = crate::settings::settings();
        let lan_enabled = s.proxy.lan || !s.proxy.lan_ip.is_empty();
        if !s.proxy.enable || !lan_enabled {
            return None;
        }
        if s.proxy.lan_ip.is_empty() {
            let detected = crate::proxy::lan_ip::detect_lan_ip().await;
            if detected.is_none() {
                error!(
                    "LAN mode is enabled but no LAN IP address could be detected. \
                     Set proxy.lan_ip to a specific address, or ensure you are connected to a network."
                );
            }
            return detected;
        }
        match s.proxy.lan_ip.parse::<std::net::Ipv4Addr>() {
            Ok(ip) => Some(ip),
            Err(e) => {
                let msg = format!(
                    concat!(
                        "proxy.lan_ip {:?} is not a valid IPv4 address: {}. ",
                        "LAN mode will not start; fix the setting or clear it ",
                        "to auto-detect."
                    ),
                    s.proxy.lan_ip, e
                );
                error!("{msg}");
                self.add_notification(log::LevelFilter::Error, msg).await;
                None
            }
        }
    }

    async fn start_mdns(&self, lan_ip: Option<std::net::Ipv4Addr>) {
        let s = crate::settings::settings();
        let lan_enabled = s.proxy.lan || !s.proxy.lan_ip.is_empty();
        if !s.proxy.enable || !lan_enabled {
            return;
        }

        let Some(lan_ip) = lan_ip else { return };
        let port = u16::try_from(s.proxy.port).unwrap_or(443);

        let Some(mut publisher) = crate::proxy::mdns::MdnsPublisher::new(lan_ip) else {
            error!("Failed to start mDNS publisher. Is Avahi (Linux) or Bonjour (macOS) running?");
            // The DNS responder hands out this address too, and it is useful on
            // this machine whether or not mDNS came up. Keep watching the
            // interface so its answers do not go stale.
            self.start_lan_ip_monitor(lan_ip, port, None).await;
            return;
        };

        // Publish all registered slugs.
        let slugs = crate::pitchfork_toml::PitchforkToml::read_global_slugs();
        for slug in slugs.keys() {
            let hostname = format!("{slug}.local");
            publisher.publish(&hostname, port);
        }

        log::info!(
            "LAN mode: mDNS publishing on {lan_ip}, {} slug(s) registered",
            slugs.len()
        );

        let publisher = std::sync::Arc::new(tokio::sync::Mutex::new(publisher));

        // Start the IP monitor. It declines on its own when the address is
        // pinned rather than auto-detected.
        self.start_lan_ip_monitor(lan_ip, port, Some(publisher.clone()))
            .await;

        *self.mdns_publisher.lock().await = Some(publisher);
    }

    /// Re-read slugs from config and update mDNS records.
    ///
    /// Publishes new slugs and unpublishes removed ones. Called via IPC when
    /// `proxy add` or `proxy remove` modifies the slug registry.
    async fn sync_mdns(&self) {
        // Clone the Arc and release the outer lock immediately so we don't
        // block close() from taking the publisher during shutdown.
        let publisher = {
            let guard = self.mdns_publisher.lock().await;
            match guard.as_ref() {
                Some(p) => p.clone(),
                None => {
                    debug!("sync_mdns: mDNS publisher not active, skipping");
                    return;
                }
            }
        };

        let s = crate::settings::settings();
        let port = u16::try_from(s.proxy.port).unwrap_or(443);

        let slugs = crate::pitchfork_toml::PitchforkToml::read_global_slugs();
        let mut pub_guard = publisher.lock().await;

        // Unpublish slugs that no longer exist in config.
        let current_keys: Vec<&String> = slugs.keys().collect();
        let registered: Vec<String> = pub_guard.registered_hostnames();
        for hostname in &registered {
            // hostname is "slug.local" — extract slug part.
            let slug = hostname.strip_suffix(".local").unwrap_or(hostname);
            if !current_keys.iter().any(|k| k.as_str() == slug) {
                log::info!("mDNS: unpublishing removed slug {slug}");
                pub_guard.unpublish(hostname);
            }
        }

        // Publish new slugs that aren't yet registered.
        for slug in slugs.keys() {
            let hostname = format!("{slug}.local");
            if !pub_guard.is_published(&hostname) {
                log::info!("mDNS: publishing new slug {slug}");
                pub_guard.publish(&hostname, port);
            }
        }
    }

    /// Spawn a background task that periodically flushes the state file to
    /// disk if it has been marked dirty.  Uses debouncing (1s interval) to
    /// batch rapid state changes.
    fn start_state_flush_task(&self) {
        let cancel = tokio_util::sync::CancellationToken::new();
        *self.flush_cancel.lock().unwrap() = Some(cancel.clone());
        tokio::spawn(async move {
            let mut interval = time::interval(Duration::from_secs(1));
            interval.set_missed_tick_behavior(time::MissedTickBehavior::Skip);
            loop {
                tokio::select! {
                    _ = interval.tick() => {}
                    _ = cancel.cancelled() => {
                        debug!("state flush task received shutdown signal");
                        break;
                    }
                }
                let state = SUPERVISOR.state_file.lock().await;
                if state.is_dirty()
                    && let Err(e) = state.write()
                {
                    warn!("failed to flush state file: {e}");
                }
            }
            debug!("state flush task exiting");
        });
    }

    pub(crate) async fn flush_state(&self) {
        let state = self.state_file.lock().await;
        if state.is_dirty()
            && let Err(e) = state.write()
        {
            warn!("failed to flush state file: {e}");
        }
    }

    pub(crate) async fn refresh(&self) -> Result<()> {
        trace!("refreshing");

        // Collect PIDs we need to check (shell PIDs and liveness PIDs)
        // This is more efficient than refreshing all processes on the system
        let dirs_with_pids = self.get_dirs_with_shell_pids().await;
        let liveness_sessions = self.get_liveness_sessions().await;
        let pids_to_check: Vec<u32> = dirs_with_pids
            .values()
            .flatten()
            .copied()
            .chain(liveness_sessions.iter().map(|(pid, _, _)| *pid))
            .collect::<std::collections::HashSet<_>>()
            .into_iter()
            .collect();

        if pids_to_check.is_empty() {
            // No PIDs to check, skip the expensive refresh
            trace!("no tracked PIDs to check, skipping process refresh");
        } else {
            debug!("refreshing PIDs: {pids_to_check:?}");
            PROCS.refresh_pids(&pids_to_check);
        }

        let mut last_refreshed_at = self.last_refreshed_at.lock().await;
        *last_refreshed_at = time::Instant::now();

        self.restore_own_record().await;

        #[cfg_attr(not(unix), allow(unused_mut))]
        let mut dirs_to_leave: Vec<PathBuf> = Vec::new();

        // Prune shell PIDs that are no longer running. This is essential on
        // Unix so that exited shells don't keep daemons alive forever.
        //
        // On Windows, skip this check: Git Bash (MSYS2) PIDs from `$$` are
        // Cygwin-internal PIDs that are invisible to sysinfo (which sees
        // Windows PIDs). The is_running check would always return false,
        // immediately removing every registered shell and breaking autostop.
        // Shell registration/deregistration relies on UpdateShellDir IPC
        // messages instead.
        #[cfg(unix)]
        for (dir, pids) in dirs_with_pids {
            let to_remove = pids
                .iter()
                .filter(|pid| !PROCS.is_running(**pid))
                .collect::<Vec<_>>();
            for pid in &to_remove {
                self.remove_shell_pid(**pid).await?
            }
            if to_remove.len() == pids.len() {
                dirs_to_leave.push(dir);
            }
        }

        // Atomically remove project sessions whose host PID has died or whose
        // recorded title no longer matches the current process title. Every
        // project session carries a host PID in its key, so we iterate all of
        // them. Re-reading the sessions under the lock prevents enter/leave
        // interleaving from deleting a session that was just replaced with a
        // new title snapshot.
        //
        // Gated to Unix to mirror the shell-PID pruning above: on Windows,
        // Git Bash (MSYS2) `$$` PIDs are Cygwin-internal and invisible to
        // sysinfo, so the liveness check would immediately revoke every
        // freshly-entered session. Windows relies on explicit `project leave`
        // (or shell UpdateShellDir) for deregistration instead.
        #[cfg(unix)]
        {
            let mut state = self.state_file.lock().await;
            for (pid, dir, recorded_title) in liveness_sessions {
                let Some(session) = state.get_project_session(pid, &dir) else {
                    continue;
                };
                let current_title = PROCS.title(pid);
                let is_running = PROCS.is_running(pid);
                debug!(
                    "refresh liveness session pid {pid} dir {} recorded_title={recorded_title:?} current_title={current_title:?} is_running={is_running}",
                    dir.display()
                );
                if should_remove_liveness_session(
                    session,
                    &recorded_title,
                    current_title.as_deref(),
                    is_running,
                ) {
                    warn!(
                        "removing project session pid {pid} dir {} (liveness pid title mismatch or dead)",
                        dir.display()
                    );
                    if state.remove_project_session(pid, &dir).is_some() {
                        dirs_to_leave.push(dir);
                    }
                }
            }
        }

        for dir in dirs_to_leave {
            self.leave_dir(&dir).await?;
        }

        // Catch state-`running` daemons that lost their monitor (e.g. the
        // monitor died with a previous supervisor): mark dead ones errored
        // and re-adopt live ones. Runs before check_retry so a daemon marked
        // errored here is retried on this same tick.
        self.reconcile_unmonitored_daemons().await;

        self.check_retry().await?;
        self.process_pending_autostops().await?;

        Ok(())
    }

    /// Install a SIGCHLD handler that reaps orphaned zombie child processes.
    ///
    /// When running as PID 1 inside a container, orphaned processes are
    /// re-parented to PID 1. Without explicit reaping, they accumulate
    /// as zombies in the process table indefinitely.
    ///
    /// Only reaps processes that are NOT managed by the supervisor (i.e.
    /// not tracked in the state file). Managed daemon processes are reaped
    /// by their monitoring tasks via `child.wait()`.
    ///
    /// ## Strategy
    ///
    /// **Linux**: Uses `waitid(Id::All, WNOHANG | WNOWAIT | WEXITED)` to
    /// *peek* at the next zombie without consuming its status. If the PID
    /// belongs to a managed daemon, the reaper skips it so Tokio's
    /// `child.wait()` can collect the status normally. Only unmanaged
    /// orphans are actually reaped (via `waitpid(Pid, WNOHANG)`). This
    /// eliminates the race entirely.
    ///
    /// **Non-Linux Unix** (e.g. macOS — mainly for local development;
    /// container mode targets Linux): `waitid` is unavailable, so we fall
    /// back to `waitpid(None, WNOHANG)`. If the reaper accidentally
    /// consumes a managed PID's status, it stashes the exit code in
    /// [`REAPED_STATUSES`] for the monitoring task to recover.
    #[cfg(unix)]
    fn reap_zombies(&self) -> Result<()> {
        let mut stream = signal::unix::signal(SignalKind::child())
            .map_err(|e| miette::miette!("Failed to register SIGCHLD handler: {e}"))?;
        tokio::spawn(async move {
            loop {
                stream.recv().await;
                // Collect PIDs of managed daemons so we don't steal their exit status
                let managed_pids: HashSet<u32> = SUPERVISOR
                    .state_file
                    .lock()
                    .await
                    .daemons
                    .values()
                    .filter_map(|d| d.pid)
                    .collect();
                // Reap all available zombie children that are NOT managed
                Self::reap_unmanaged_zombies(&managed_pids).await;
            }
        });
        info!("container mode: SIGCHLD zombie reaper installed");
        Ok(())
    }

    /// Linux implementation: peek with `waitid(WNOWAIT)` then selectively reap.
    ///
    /// `WNOWAIT` leaves the zombie in the table so we can inspect its PID
    /// without consuming the exit status. Only if the PID is *not* managed
    /// do we call `waitpid(Pid, WNOHANG)` to actually reap it.
    #[cfg(target_os = "linux")]
    async fn reap_unmanaged_zombies(managed_pids: &HashSet<u32>) {
        use nix::sys::wait::{Id, WaitPidFlag, WaitStatus, waitid, waitpid};
        use nix::unistd::Pid;

        loop {
            // Peek at the next zombie without consuming it
            let peek_flags = WaitPidFlag::WNOHANG | WaitPidFlag::WNOWAIT | WaitPidFlag::WEXITED;
            match waitid(Id::All, peek_flags) {
                Ok(WaitStatus::StillAlive) => break,
                Ok(status) => {
                    let Some(pid_raw) = status.pid().map(|p| p.as_raw() as u32) else {
                        break;
                    };
                    if managed_pids.contains(&pid_raw) {
                        // This is a managed daemon — leave it for Tokio's child.wait().
                        // We must break out of the loop because waitid(Id::All) would
                        // keep returning the same zombie if we don't consume it.
                        trace!(
                            "zombie reaper: skipping managed daemon pid {pid_raw}, \
                             leaving for Tokio to reap"
                        );
                        break;
                    }
                    // Not managed — actually reap it
                    match waitpid(Pid::from_raw(pid_raw as i32), Some(WaitPidFlag::WNOHANG)) {
                        Ok(s) => trace!("reaped orphaned zombie child: {s:?}"),
                        Err(nix::errno::Errno::ECHILD) => break,
                        Err(e) => {
                            trace!("waitpid error reaping pid {pid_raw}: {e}");
                            break;
                        }
                    }
                }
                Err(nix::errno::Errno::ECHILD) => break, // no children at all
                Err(e) => {
                    trace!("waitid error in zombie reaper: {e}");
                    break;
                }
            }
        }
    }

    /// Non-Linux fallback: blind `waitpid(None, WNOHANG)` with stash recovery.
    ///
    /// Since `waitid(WNOWAIT)` is not available, we cannot peek. If we
    /// accidentally reap a managed PID, we stash the exit code in
    /// [`REAPED_STATUSES`] so the monitoring task can recover it.
    #[cfg(all(unix, not(target_os = "linux")))]
    async fn reap_unmanaged_zombies(managed_pids: &HashSet<u32>) {
        use nix::sys::wait::{WaitPidFlag, WaitStatus, waitpid};

        loop {
            match waitpid(None, Some(WaitPidFlag::WNOHANG)) {
                Ok(WaitStatus::StillAlive) => break,
                Ok(status) => {
                    let Some(pid) = status.pid().map(|p| p.as_raw() as u32) else {
                        continue;
                    };
                    if managed_pids.contains(&pid) {
                        // Race lost — stash the exit code for lifecycle recovery
                        let exit_code = match status {
                            WaitStatus::Exited(_, code) => code,
                            WaitStatus::Signaled(_, sig, _) => -(sig as i32),
                            _ => -1,
                        };
                        warn!(
                            "zombie reaper reaped managed daemon pid {pid} \
                             (exit_code={exit_code}); stashing status for recovery"
                        );
                        REAPED_STATUSES.lock().await.insert(pid, exit_code);
                    } else {
                        trace!("reaped orphaned zombie child: {status:?}");
                    }
                }
                Err(nix::errno::Errno::ECHILD) => break, // no more children
                Err(e) => {
                    trace!("waitpid error in zombie reaper: {e}");
                    break;
                }
            }
        }
    }

    #[cfg(unix)]
    fn signals(&self) -> Result<()> {
        let signals = [
            SignalKind::terminate(),
            SignalKind::alarm(),
            SignalKind::interrupt(),
            SignalKind::quit(),
            SignalKind::hangup(),
            SignalKind::user_defined1(),
            SignalKind::user_defined2(),
        ];
        static RECEIVED_SIGNAL: AtomicBool = AtomicBool::new(false);
        for signal in signals {
            let stream = match signal::unix::signal(signal) {
                Ok(s) => s,
                Err(e) => {
                    warn!("Failed to register signal handler for {signal:?}: {e}");
                    continue;
                }
            };
            tokio::spawn(async move {
                let mut stream = stream;
                loop {
                    stream.recv().await;
                    if RECEIVED_SIGNAL.swap(true, atomic::Ordering::SeqCst) {
                        exit(1);
                    } else {
                        SUPERVISOR.handle_signal().await;
                    }
                }
            });
        }
        Ok(())
    }

    #[cfg(windows)]
    fn signals(&self) -> Result<()> {
        tokio::spawn(async move {
            static RECEIVED_SIGNAL: AtomicBool = AtomicBool::new(false);
            loop {
                if let Err(e) = signal::ctrl_c().await {
                    error!("Failed to wait for ctrl-c: {}", e);
                    return;
                }
                if RECEIVED_SIGNAL.swap(true, atomic::Ordering::SeqCst) {
                    exit(1);
                } else {
                    SUPERVISOR.handle_signal().await;
                }
            }
        });
        Ok(())
    }

    async fn handle_signal(&self) {
        info!("received signal, stopping");
        self.close().await;
        exit(0)
    }

    pub(crate) async fn close(&self) {
        self.shutting_down.store(true, atomic::Ordering::Release);
        // Signal the proxy server to stop accepting new connections
        // and drain in-flight ones, *before* stopping daemons so the
        // proxy has time to finish forwarding active requests.
        if let Some(cancel) = self.proxy_cancel.lock().await.take() {
            cancel.cancel();
        }

        // Stop the LAN IP monitor task. It watches the token that was just
        // cancelled, so give it a moment to come back on its own rather than
        // cutting it off mid-iteration the instant after asking it to stop.
        if let Some(mut monitor_task) = self.lan_monitor_task.lock().await.take()
            && tokio::time::timeout(Duration::from_secs(1), &mut monitor_task)
                .await
                .is_err()
        {
            monitor_task.abort();
        }

        // Shutdown the mDNS publisher (sends goodbye packets).
        if let Some(publisher) = self.mdns_publisher.lock().await.take() {
            publisher.lock().await.shutdown();
        }

        if let Some(mut dns_task) = self.dns_task.lock().await.take()
            && tokio::time::timeout(Duration::from_secs(5), &mut dns_task)
                .await
                .is_err()
        {
            // Cancelled but still running: aborted, as the LAN monitor is,
            // so its sockets are not left bound to the resolver port.
            dns_task.abort();
        }

        if let Some(proxy_task) = self.proxy_task.lock().await.take() {
            // Longer than the proxy's own drain budget, so the task finishes on
            // its own terms rather than being cut off mid-drain.
            let _ = tokio::time::timeout(
                crate::proxy::server::SHUTDOWN_DRAIN_BUDGET + Duration::from_secs(2),
                proxy_task,
            )
            .await;
        }

        // Clean up /etc/hosts entries managed by pitchfork
        let s = settings();
        if s.proxy.enable && s.proxy.sync_hosts {
            crate::proxy::hosts::clean_hosts_file();
        }

        let pitchfork_id = DaemonId::pitchfork();
        let active = self.active_daemons().await;
        let active_ids: Vec<DaemonId> = active
            .iter()
            .filter(|d| d.id != pitchfork_id)
            .map(|d| d.id.clone())
            .collect();

        // Stop daemons in reverse dependency order.
        // If dependency resolution fails (e.g. config changed), fall back to
        // stopping in arbitrary order so we still shut down cleanly.
        // Daemons within the same level are stopped concurrently.
        //
        // Each stop waits for the daemon's whole process group (bounded by its
        // stop budget) and levels are sequential, so total shutdown time is the
        // sum of the slowest stop per level. If an external manager (docker,
        // systemd) kills us before this completes, cleanup_orphaned_daemons()
        // recovers the leftover processes and stale state on the next start.
        let stop_levels = compute_reverse_stop_order(&active_ids);
        for level in &stop_levels {
            let mut tasks = Vec::new();
            for id in level {
                let id = id.clone();
                tasks.push(tokio::spawn(async move {
                    if let Err(err) = SUPERVISOR.stop(&id).await {
                        error!("failed to stop daemon {id}: {err}");
                    }
                }));
            }
            for task in tasks {
                let _ = task.await;
            }
        }
        let _ = self.remove_daemon(&pitchfork_id).await;

        // Signal the background state flush task to exit so it doesn't
        // keep waking up and acquiring the state mutex after shutdown.
        if let Some(cancel) = self.flush_cancel.lock().unwrap().take() {
            cancel.cancel();
        }

        // Force-flush state to disk before shutting down IPC so no
        // in-memory-only changes are lost.
        {
            let state = self.state_file.lock().await;
            if state.is_dirty()
                && let Err(e) = state.write()
            {
                warn!("failed to flush state file during shutdown: {e}");
            }
        }

        // Signal IPC server to shut down gracefully
        if let Some(mut handle) = self.ipc_shutdown.lock().await.take() {
            handle.shutdown().await;
        }

        // Wait for all in-flight monitoring tasks to finish registering their
        // hook handles. Each monitoring task increments `active_monitors` when
        // its process exits, and decrements it (+ notifies `monitor_done`)
        // after all fire_hook() calls complete. This replaces the old
        // yield_now() approach which had a race window.
        let drain_timeout = time::sleep(Duration::from_secs(5));
        tokio::pin!(drain_timeout);
        loop {
            if self.active_monitors.load(atomic::Ordering::Acquire) == 0 {
                break;
            }
            tokio::select! {
                _ = self.monitor_done.notified() => {}
                _ = &mut drain_timeout => {
                    warn!("timed out waiting for monitoring tasks to register hooks, proceeding with shutdown");
                    break;
                }
            }
        }
        let handles: Vec<JoinHandle<()>> = std::mem::take(&mut *self.hook_tasks.lock().await);
        let hook_timeout = Duration::from_secs(30);
        for handle in handles {
            match time::timeout(hook_timeout, handle).await {
                Ok(_) => {} // Hook completed (success or error, doesn't matter)
                Err(_) => {
                    warn!(
                        "hook task did not complete within {hook_timeout:?} during shutdown, skipping"
                    );
                }
            }
        }

        // Unix: remove the socket directory if it is empty. The IPC server
        // already removed our socket; anything left belongs to a supervisor
        // that replaced this one (e.g. `supervisor run --force`) while we
        // were stopping daemons, and must not be deleted.
        // Windows: named pipes have no filesystem component.
        #[cfg(unix)]
        let _ = fs::remove_dir(&*env::IPC_SOCK_DIR);
    }

    pub(crate) async fn add_notification(&self, level: log::LevelFilter, message: String) {
        self.pending_notifications
            .lock()
            .await
            .push((level, message));
    }
}

/// Fix ownership on the state directory so non-root users can access files
/// created by a `sudo`-started supervisor.
///
/// When `[settings.supervisor] user`, a recorded invoking user (see
/// [`env::InvokingUser`]), or `SUDO_UID`/`SUDO_GID` are set, we
/// `chown` the state directory and safe subdirectories back to that non-root
/// runtime user. This is strictly better than `chmod 0o666` because it does not
/// widen the permission bits — the files stay owner-only (0o600/0o700) but the
/// *owner* is the user that daemon processes and CLI clients need to share.
///
/// **Security**: The `proxy/` subtree is intentionally skipped. It contains
/// `ca-key.pem` which must remain `0o600` and owned by the process that
/// generated it. Changing its ownership or permissions would expose the CA
/// private key to other local users.
///
/// If none of these are available (e.g. a direct root login, or a boot service
/// installed from a root shell), we fall back to relaxing permissions on only
/// the `sock/` and `logs/` subdirectories (plus `state.toml`) so CLI clients
/// can still function.
#[cfg(unix)]
fn fix_state_dir_permissions() {
    let state_dir = &*env::PITCHFORK_STATE_DIR;
    if let Some((uid, gid)) = state_owner_ids() {
        if !state_dir.exists()
            && let Err(err) = fs::create_dir_all(state_dir)
        {
            warn!(
                "failed to create state directory for ownership fix at {}: {err}",
                state_dir.display()
            );
            return;
        }

        // Best path: chown back to the runtime user. Permissions stay tight.
        chown_recursive(state_dir, uid, gid, true);
        debug!(
            "chowned state directory to uid={uid} gid={gid} at {}",
            state_dir.display()
        );
    } else {
        if !state_dir.exists() {
            return;
        }

        // Fallback: relax permissions on safe subdirectories only.
        // proxy/ is never touched.
        chmod_safe_subtrees(state_dir);
        debug!(
            "relaxed permissions on safe subtrees at {}",
            state_dir.display()
        );
    }
}

#[cfg(unix)]
pub(crate) fn state_owner_ids() -> Option<(u32, u32)> {
    if !nix::unistd::Uid::effective().is_root() {
        return None;
    }

    let s = settings();
    let user = s.supervisor.user.trim();
    if !user.is_empty() {
        return resolve_supervisor_user_ids(user).or_else(|| {
            warn!(
                "failed to resolve supervisor.user '{user}' for state ownership; falling back to the invoking user"
            );
            env::invoking_user_ids()
        });
    }

    env::invoking_user_ids()
}

#[cfg(unix)]
fn resolve_supervisor_user_ids(user: &str) -> Option<(u32, u32)> {
    let user_record = if user.chars().all(|c| c.is_ascii_digit()) {
        let uid = user.parse::<u32>().ok()?;
        nix::unistd::User::from_uid(nix::unistd::Uid::from_raw(uid))
            .ok()
            .flatten()
    } else {
        nix::unistd::User::from_name(user).ok().flatten()
    }?;

    Some((user_record.uid.as_raw(), user_record.gid.as_raw()))
}

/// Recursively `chown` a directory tree. If `skip_proxy` is true, the `proxy/`
/// subdirectory is skipped entirely to protect the CA private key.
#[cfg(unix)]
fn chown_recursive(dir: &std::path::Path, uid: u32, gid: u32, skip_proxy: bool) {
    // chown the directory itself
    let _ = chown_path(dir, uid, gid);

    let entries = match std::fs::read_dir(dir) {
        Ok(e) => e,
        Err(_) => return,
    };
    for entry in entries.flatten() {
        let path = entry.path();
        if path.is_dir() {
            // Skip proxy/ at the top level of the state directory
            if skip_proxy
                && let Some(name) = path.file_name().and_then(|n| n.to_str())
                && name == "proxy"
            {
                continue;
            }
            chown_recursive(&path, uid, gid, false);
        } else {
            let _ = chown_path(&path, uid, gid);
        }
    }
}

/// `chown` a single path using libc. Returns Ok(()) on success.
#[cfg(unix)]
fn chown_path(path: &std::path::Path, uid: u32, gid: u32) -> std::io::Result<()> {
    use std::ffi::CString;
    use std::os::unix::ffi::OsStrExt;
    let c_path = CString::new(path.as_os_str().as_bytes())
        .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidInput, e))?;
    let ret = unsafe { libc::chown(c_path.as_ptr(), uid, gid) };
    if ret == 0 {
        Ok(())
    } else {
        Err(std::io::Error::last_os_error())
    }
}

/// Fallback: relax permissions on safe subdirectories only (sock/, logs/, and
/// state.toml). The proxy/ subtree is never touched.
#[cfg(unix)]
fn chmod_safe_subtrees(state_dir: &std::path::Path) {
    // The state directory itself needs to be traversable
    let _ = fs::set_permissions(state_dir, fs::Permissions::from_mode(0o755));

    // state.toml — needs to be readable by CLI clients
    let state_file = state_dir.join("state.toml");
    if state_file.exists() {
        let _ = fs::set_permissions(&state_file, fs::Permissions::from_mode(0o644));
    }

    // Safe subdirectories: sock/ and logs/
    for subdir_name in &["sock", "logs"] {
        let subdir = state_dir.join(subdir_name);
        if subdir.is_dir() {
            chmod_recursive(&subdir);
        }
    }
}

/// On startup, reconcile daemon processes left behind by a previous supervisor
/// that was terminated unexpectedly (e.g. `kill -9`).
///
/// This iterates the state file for daemon entries with a recorded PID. If the
/// PID is still alive and its current identity matches the recorded start time
/// (or the recorded title for older state files), it is assumed to be an orphan
/// from the previous supervisor session and `supervisor.orphan_policy` decides
/// its fate: `adopt` (default) resumes supervision via a poll monitor and keeps
/// the daemon's state intact; `kill` terminates it and resets its state to
/// `Stopped` with no PID. If a matching live process cannot be terminated
/// securely, its running state is retained to prevent a duplicate instance
/// from being started.
///
/// Missing or mismatched identity data fails closed so a PID recycled by an
/// unrelated process is never adopted or killed. On Unix platforms without
/// durable process handles, orphan termination also fails closed because the
/// PID/PGID cannot be pinned between identity validation and signaling.
///
/// This is gated by the `supervisor.cleanup_orphans` setting (default: true).
///
/// `unclean` is whether the previous supervisor exited uncleanly (see
/// [`supervisor_exited_uncleanly`]). It is read in `start()` before the
/// starting supervisor records itself in the state file — this function runs
/// in the background after that record exists, so reading it here would
/// always report unclean.
async fn cleanup_orphaned_daemons(supervisor: &Supervisor, unclean: bool) {
    if !settings().supervisor.cleanup_orphans {
        return;
    }

    let candidates: Vec<_> = {
        let state = supervisor.state_file.lock().await;
        state
            .daemons
            .values()
            .filter(|d| d.id != DaemonId::pitchfork() && d.pid.is_some())
            .cloned()
            .collect()
    };

    if candidates.is_empty() {
        return;
    }

    info!(
        "checking {} daemon(s) for orphaned processes",
        candidates.len()
    );

    let policy = orphan_policy();
    let boot_time = PROCS.boot_time();

    // Reconcile orphans in parallel — a kill waits for the daemon's whole
    // process group to exit, bounded by that daemon's stop budget, so
    // sequential processing would make total cleanup time the sum of the
    // budgets.
    let tasks: Vec<_> = candidates
        .into_iter()
        .map(|daemon| {
            let policy = policy.clone();
            tokio::spawn(cleanup_orphaned_daemon(daemon, policy, boot_time, unclean))
        })
        .collect();
    for task in tasks {
        let _ = task.await;
    }
}

/// Reconcile a single orphan candidate: adopt it, kill it, or reset its state,
/// per the policy and identity checks described on [`cleanup_orphaned_daemons`].
///
/// Holds the daemon's stop lock so a concurrent Run/Stop request for the same
/// daemon (cleanup runs in the background) serializes with the orphan kill,
/// and re-checks the recorded PID under the lock: if it changed, another path
/// already replaced or cleaned up this record and the snapshot is stale.
async fn cleanup_orphaned_daemon(
    daemon: crate::daemon::Daemon,
    policy: String,
    boot_time: u64,
    unclean: bool,
) {
    let supervisor: &Supervisor = &SUPERVISOR;
    let Some(pid) = daemon.pid else { return };

    let lock = supervisor.stop_lock(&daemon.id).await;
    let _guard = lock.lock().await;
    let current_pid = {
        let state = supervisor.state_file.lock().await;
        state.daemons.get(&daemon.id).and_then(|d| d.pid)
    };
    if current_pid != Some(pid) {
        debug!(
            "orphan cleanup: daemon {} pid changed (recorded {pid}, now {current_pid:?}), skipping",
            daemon.id
        );
        return;
    }

    // Refresh the candidate immediately before checking it: waiting for the
    // stop lock can await another path's stop timeout, during which this PID
    // may exit and be recycled.
    PROCS.refresh_pids(&[pid]);

    if !PROCS.is_running(pid) {
        // PID already dead — the daemon exited while unsupervised, so
        // record a terminal status that reflects whether it died under a
        // crashed supervisor (retryable) or with the machine.
        let status = unobserved_exit_status(
            &daemon.status,
            daemon.boot_time,
            boot_time,
            unclean,
            daemon.oneshot,
        );
        reset_daemon_state(supervisor, &daemon.id, status, ExitObservation::Unobserved).await;
        return;
    }

    // Safety check: verify the live process really is the daemon we
    // recorded, not an unrelated process that received a recycled PID.
    // The kernel start time is a stable identity for the lifetime of a
    // process, and is the only thing accepted as one.
    let current_start_time = PROCS.start_time(pid);
    let matches = process_identity_matches(daemon.start_time, current_start_time);

    if !matches {
        // Either side missing means the identity cannot be checked at all,
        // which is different from checking it and finding a stranger: retain
        // the running state rather than resetting a record whose process may
        // well still be the daemon.
        if daemon.start_time.is_none() || current_start_time.is_none() {
            warn!(
                "could not verify the identity of live pid {pid} recorded for daemon {}; retaining running state",
                daemon.id,
            );
            return;
        }
        warn!(
            "pid {pid} recorded for daemon {} belongs to a different process now (PID recycled); resetting state without killing",
            daemon.id,
        );
        // The daemon died at some unknown point and the OS handed its PID
        // to something else — same unobserved exit as a dead PID.
        let status = unobserved_exit_status(
            &daemon.status,
            daemon.boot_time,
            boot_time,
            unclean,
            daemon.oneshot,
        );
        reset_daemon_state(supervisor, &daemon.id, status, ExitObservation::Unobserved).await;
        return;
    }

    // Both policies need a verified start time: killing revalidates it
    // while pinned to the process, and adoption anchors its poll monitor
    // to it so a later PID recycle is never mistaken for the daemon.
    let Some(expected_start_time) = current_start_time else {
        warn!(
            "could not read start time for live pid {pid} recorded for daemon {}; retaining running state",
            daemon.id,
        );
        return;
    };

    // Identity verified — the process really is our orphaned daemon.
    // The policy decides whether supervision resumes or the slate is
    // wiped clean.
    if policy == "adopt" {
        supervisor
            .adopt_daemon(&daemon, pid, expected_start_time)
            .await;
        return;
    }

    info!("terminating orphaned daemon {} (pid {pid})", daemon.id);

    let stop_cfg = daemon.stop_signal.unwrap_or_default();
    let termination_result = PROCS
        .kill_process_group_if_start_time_matches_async(
            pid,
            Some(expected_start_time),
            stop_cfg.signal.into(),
            stop_cfg.timeout,
        )
        .await;

    match termination_result {
        Ok(true) => {}
        Ok(false) => {
            warn!(
                "could not securely terminate orphaned daemon {} (pid {pid}); retaining running state",
                daemon.id
            );
            return;
        }
        Err(err) => {
            warn!(
                "failed to terminate orphaned daemon {} (pid {pid}): {err}; retaining running state",
                daemon.id
            );
            return;
        }
    }

    // We terminated the orphan ourselves, so this is an observed,
    // intentional stop rather than an unobserved exit.
    reset_daemon_state(
        supervisor,
        &daemon.id,
        DaemonStatus::Stopped,
        ExitObservation::Terminated,
    )
    .await;
}

/// Effective `supervisor.orphan_policy`, warning on an unrecognized value
/// (which falls back to the default of adopting).
pub(crate) fn orphan_policy() -> String {
    let policy = settings().supervisor.orphan_policy.clone();
    match policy.as_str() {
        "adopt" | "kill" => policy,
        other => {
            warn!("unknown supervisor.orphan_policy '{other}', defaulting to 'adopt'");
            "adopt".to_string()
        }
    }
}

/// Verify that live process identity matches the persisted daemon identity.
///
/// Both start times are required. A process name was once accepted in place of
/// a recorded start time, for state written before start times existed, but a
/// name is not an identity: a recycled PID belonging to another copy of the same
/// program matches it, and adopting or killing on that basis acts on the wrong
/// process. Missing identity, on either side, means unverifiable — and
/// unverifiable must never authorize acting on a process.
fn process_identity_matches(
    recorded_start_time: Option<u64>,
    current_start_time: Option<u64>,
) -> bool {
    match (recorded_start_time, current_start_time) {
        (Some(recorded), Some(current)) => recorded == current,
        _ => false,
    }
}

/// Whether a PID read from persisted state may be signalled.
///
/// Stopping a daemon signals its whole process *group*, so acting on a PID that
/// has been recycled since it was recorded takes down an unrelated process tree.
/// Records are refused only when their identity is positively contradicted: if
/// either start time is unknown the PID stays as signallable as it was before
/// identities were recorded, so a daemon whose record predates the field can
/// still be stopped rather than becoming permanently unstoppable.
///
/// This is deliberately weaker than [`process_identity_matches`], which decides
/// whether to adopt or kill a process nobody asked about. Here the user has
/// named the daemon and asked for it to stop; the check exists to catch the
/// case where the answer is provably the wrong process.
pub(crate) fn signalling_pid_is_authorized(
    recorded_start_time: Option<u64>,
    current_start_time: Option<u64>,
) -> bool {
    !matches!(
        (recorded_start_time, current_start_time),
        (Some(recorded), Some(current)) if recorded != current
    )
}

/// How a daemon's run ended, which decides what happens to the recorded
/// `last_exit_success` that cron `retrigger = "success" | "fail"` reads.
#[derive(Clone, Copy, PartialEq, Eq)]
pub(crate) enum ExitObservation {
    /// Nobody saw how the run ended, because the monitor that would have
    /// observed it died with a previous supervisor. The recorded outcome is
    /// cleared to `None`.
    ///
    /// Every option here is imperfect, so this picks the one that asserts
    /// nothing false. `Some(false)` would fabricate a failure, silently
    /// breaking a `retrigger = "success"` chain whose run may well have
    /// succeeded; `Some(true)` fabricates the opposite; keeping the previous
    /// value attributes an earlier run's outcome to this one. `None` says
    /// "unknown", reusing the reading the cron watcher already applies to a
    /// daemon that has never run.
    ///
    /// The tradeoff is that `None` satisfies both `retrigger = "success"`
    /// (`unwrap_or(true)`) and `retrigger = "fail"` (`!unwrap_or(false)`), so
    /// such a daemon fires once at its next scheduled time regardless of which
    /// it configured. That is schedule-gated rather than a loop, and it biases
    /// toward running the daemon over leaving it permanently untriggered.
    /// Distinguishing "unknown" from "never ran" would require a third cron
    /// state and is deliberately left out of scope here.
    Unobserved,
    /// We terminated the process ourselves, so the outcome is not a mystery:
    /// it stopped because we asked it to. Recorded as a success, matching the
    /// convention `Supervisor::stop` already uses for a deliberate stop.
    Terminated,
}

impl ExitObservation {
    /// The `last_exit_success` value this observation implies.
    pub(crate) fn last_exit_success(self) -> Option<bool> {
        match self {
            ExitObservation::Unobserved => None,
            ExitObservation::Terminated => Some(true),
        }
    }
}

/// Clear a daemon's runtime state (pid, process identity, active port) after
/// its process is gone or is no longer ours to manage.
///
/// Config fields are preserved by cloning the existing record, so a reset can
/// never drop a daemon's command, retry policy, or schedule.
async fn reset_daemon_state(
    supervisor: &Supervisor,
    id: &DaemonId,
    status: DaemonStatus,
    observation: ExitObservation,
) {
    let mut state_file = supervisor.state_file.lock().await;
    let Some(existing) = state_file.daemons.get(id) else {
        return;
    };
    let mut daemon = existing.clone();
    daemon.pid = None;
    daemon.title = None;
    daemon.start_time = None;
    daemon.boot_time = None;
    daemon.status = status;
    daemon.last_exit_success = observation.last_exit_success();
    daemon.active_port = None;
    state_file.clear_active_port(id);
    state_file.insert_daemon(id, daemon);
}

/// Boot times this far apart are treated as different boots.
///
/// Sized to the only platform that reports a jittery value: Windows derives
/// boot time as `now - GetTickCount64()`, sampling two clocks independently,
/// so consecutive calls within one boot can differ by about a second. Linux
/// (`/proc/stat` btime) and macOS (`kern.boottime`) report stable values.
///
/// Deliberately kept this tight so a genuine reboot can never fall inside it:
/// a prior session would have to boot, start the supervisor, spawn a daemon,
/// have that daemon die, and complete a reboot inside two seconds, which no
/// real boot cycle reaches. A larger window would misread a short-lived
/// previous boot (e.g. a device in a reboot loop) as the current one and
/// resurrect daemons a reboot should have left stopped.
const BOOT_TIME_TOLERANCE_SECS: u64 = 2;

/// Terminal status for a daemon whose process is gone and whose exit was
/// never observed, because the monitor that would have seen it died with a
/// previous supervisor.
///
/// A daemon recorded `Running` was expected to still be alive, so it died
/// under the crashed supervisor: `Errored(-1)` ("unknown exit code") makes it
/// eligible for its configured retries. Two cases stay `Stopped` instead:
///
/// - records from an earlier boot, whose processes died with the machine —
///   auto-restarting those is what `boot_start` is for, and reviving every
///   retry-configured daemon after a reboot would be a surprise
/// - any other status (in practice `Stopping`), i.e. an intentional stop that
///   completed while the supervisor was gone
pub(crate) fn unobserved_exit_status(
    status: &DaemonStatus,
    recorded_boot_time: Option<u64>,
    current_boot_time: u64,
    supervisor_exited_uncleanly: bool,
    oneshot: bool,
) -> DaemonStatus {
    let same_boot = recorded_boot_time
        .is_some_and(|recorded| recorded.abs_diff(current_boot_time) <= BOOT_TIME_TOLERANCE_SECS);
    // A task gets `stopped` rather than `errored` for the same reason the
    // adopted path does: `errored` is what `check_retry` looks for, and
    // nobody saw how this run ended, so retrying it would re-run a migration
    // or a seed that may well have succeeded. Leave re-running to an explicit
    // start.
    if status.is_running() && same_boot && supervisor_exited_uncleanly && !oneshot {
        DaemonStatus::Errored(-1)
    } else {
        DaemonStatus::Stopped
    }
}

/// Whether the supervisor that owned this state file failed to shut down
/// cleanly, meaning any daemon it left behind stopped for reasons nobody
/// recorded.
///
/// A clean shutdown removes the supervisor's own entry: `close()` does it on
/// Unix, where the stop signal is delivered and handled, and the
/// `supervisor stop` command does it on Windows, which has no POSIX signals
/// and force-terminates the process instead. A crash, an external `kill -9`,
/// or a `--force` replacement all leave the entry behind.
///
/// This must be read before the starting supervisor records itself, which is
/// why `cleanup_orphaned_daemons` runs first in `start()`.
async fn supervisor_exited_uncleanly(supervisor: &Supervisor) -> bool {
    supervisor
        .state_file
        .lock()
        .await
        .daemons
        .contains_key(&DaemonId::pitchfork())
}

/// Recursively chmod: directories → 0o755, files → 0o644.
#[cfg(unix)]
fn chmod_recursive(dir: &std::path::Path) {
    let _ = fs::set_permissions(dir, fs::Permissions::from_mode(0o755));
    let entries = match fs::read_dir(dir) {
        Ok(e) => e,
        Err(_) => return,
    };
    for entry in entries.flatten() {
        let path = entry.path();
        if path.is_dir() {
            chmod_recursive(&path);
        } else {
            let _ = fs::set_permissions(&path, fs::Permissions::from_mode(0o644));
        }
    }
}

#[cfg(test)]
mod tests {
    use super::{
        BOOT_TIME_TOLERANCE_SECS, legacy_supervisor_title_matches, process_identity_matches,
        should_remove_liveness_session, signalling_pid_is_authorized, supervisor_identity_matches,
        unobserved_exit_status,
    };
    use crate::daemon_status::DaemonStatus;
    use crate::state_file::ProjectSession;

    const BOOT: u64 = 1_700_000_000;

    #[test]
    fn unobserved_running_death_in_current_boot_is_retryable() {
        // Died under a crashed supervisor during this boot: Errored(-1) makes
        // the daemon eligible for its configured retries.
        assert!(matches!(
            unobserved_exit_status(&DaemonStatus::Running, Some(BOOT), BOOT, true, false),
            DaemonStatus::Errored(-1)
        ));
    }

    #[test]
    fn unobserved_task_death_is_stopped_not_retried() {
        // Nobody saw how the run ended, so `errored` would hand a migration or
        // a seed to check_retry on a guess. Matches what the adopted path
        // records, and what the guide promises.
        assert!(matches!(
            unobserved_exit_status(&DaemonStatus::Running, Some(BOOT), BOOT, true, true),
            DaemonStatus::Stopped
        ));
    }

    #[test]
    fn unobserved_running_death_from_previous_boot_is_stopped() {
        // The process died with the machine; reviving every retry-configured
        // daemon after a reboot is what boot_start is for.
        assert!(matches!(
            unobserved_exit_status(
                &DaemonStatus::Running,
                Some(BOOT - 86_400),
                BOOT,
                true,
                false
            ),
            DaemonStatus::Stopped
        ));
    }

    #[test]
    fn unobserved_exit_tolerates_boot_time_jitter() {
        // Windows recomputes boot time as now - GetTickCount64(), which can
        // drift about a second between samples within one boot.
        let within = BOOT + BOOT_TIME_TOLERANCE_SECS;
        assert!(matches!(
            unobserved_exit_status(&DaemonStatus::Running, Some(within), BOOT, true, false),
            DaemonStatus::Errored(-1)
        ));
        let beyond = BOOT + BOOT_TIME_TOLERANCE_SECS + 1;
        assert!(matches!(
            unobserved_exit_status(&DaemonStatus::Running, Some(beyond), BOOT, true, false),
            DaemonStatus::Stopped
        ));
    }

    #[test]
    fn unobserved_exit_after_clean_shutdown_is_stopped() {
        // A deliberate `supervisor stop` can leave running records behind on
        // platforms where the supervisor cannot handle the stop signal. Those
        // daemons were stopped on purpose, so they must not be reported as
        // failures or resurrected by the retry checker.
        assert!(matches!(
            unobserved_exit_status(&DaemonStatus::Running, Some(BOOT), BOOT, false, false),
            DaemonStatus::Stopped
        ));
    }

    #[test]
    fn unobserved_exit_treats_short_previous_boot_as_previous() {
        // A device in a reboot loop can produce consecutive boots seconds
        // apart. The jitter window must stay far below that so those records
        // are still recognised as belonging to an earlier boot.
        for gap in [5, 30, 59, 60] {
            assert!(
                matches!(
                    unobserved_exit_status(
                        &DaemonStatus::Running,
                        Some(BOOT - gap),
                        BOOT,
                        true,
                        false
                    ),
                    DaemonStatus::Stopped
                ),
                "boot {gap}s earlier should be treated as a previous boot"
            );
        }
    }

    #[test]
    fn unobserved_exit_without_recorded_boot_time_is_stopped() {
        // Legacy state files predating the field fail closed to today's
        // behavior rather than triggering surprise retries.
        assert!(matches!(
            unobserved_exit_status(&DaemonStatus::Running, None, BOOT, true, false),
            DaemonStatus::Stopped
        ));
    }

    #[test]
    fn supervisor_identity_matches_same_generation() {
        assert!(supervisor_identity_matches(
            Some(100),
            Some(100),
            Some(BOOT),
            BOOT
        ));
    }

    #[test]
    fn supervisor_identity_survives_clock_steps_when_tokens_match() {
        // An NTP step or sleep/resume moves the realtime-derived boot time by
        // far more than the tolerance while the supervisor keeps running.
        // Matching start tokens prove it is the same process; declaring it
        // stale here would start a second supervisor.
        assert!(supervisor_identity_matches(
            Some(100),
            Some(100),
            Some(BOOT),
            BOOT + 3600
        ));
        assert!(supervisor_identity_matches(
            Some(100),
            Some(100),
            Some(BOOT + 3600),
            BOOT
        ));
    }

    #[test]
    fn supervisor_identity_rejects_recycled_pid() {
        // The supervisor died and something else got its PID, within this
        // boot or across a reboot: the token differs either way.
        assert!(!supervisor_identity_matches(
            Some(100),
            Some(200),
            Some(BOOT),
            BOOT
        ));
        assert!(!supervisor_identity_matches(
            Some(100),
            Some(200),
            Some(BOOT),
            BOOT + 3600
        ));
    }

    #[test]
    fn supervisor_identity_uses_boot_time_when_a_token_is_missing() {
        // Discussion #877: the record survived a reboot and an early system
        // daemon now owns the PID. With no live token to compare, the boot
        // time is what proves the record stale.
        assert!(!supervisor_identity_matches(
            Some(100),
            None,
            Some(BOOT),
            BOOT + 3600
        ));
        // Same boot (within Windows boot-time jitter) and no contradiction.
        assert!(supervisor_identity_matches(
            Some(100),
            None,
            Some(BOOT),
            BOOT + BOOT_TIME_TOLERANCE_SECS
        ));
    }

    #[test]
    fn supervisor_identity_tolerates_legacy_records() {
        // A record written before either field existed is not contradicted by
        // anything here; `supervisor_record_is_live` applies the process-name
        // check to those instead.
        assert!(supervisor_identity_matches(None, Some(100), None, BOOT));
        // A record with only a boot time is still rejected across a reboot.
        assert!(!supervisor_identity_matches(
            None,
            Some(100),
            Some(BOOT),
            BOOT + 3600
        ));
        assert!(supervisor_identity_matches(
            None,
            Some(100),
            Some(BOOT),
            BOOT
        ));
    }

    #[test]
    fn legacy_supervisor_title_requires_a_pitchfork_process() {
        assert!(legacy_supervisor_title_matches(Some("pitchfork")));
        assert!(legacy_supervisor_title_matches(Some("pitchfork.exe")));
        assert!(legacy_supervisor_title_matches(Some("Pitchfork")));
        // Discussion #877: an Apple LaunchAgent inherited the PID after a reboot.
        assert!(!legacy_supervisor_title_matches(Some(
            "AMPDeviceDiscoveryAgent"
        )));
        assert!(!legacy_supervisor_title_matches(Some("sleep")));
        assert!(!legacy_supervisor_title_matches(None));
    }

    #[test]
    fn unobserved_exit_of_stopping_daemon_is_stopped() {
        // An intentional stop that completed while the supervisor was gone is
        // not a failure, even within the same boot.
        assert!(matches!(
            unobserved_exit_status(&DaemonStatus::Stopping, Some(BOOT), BOOT, true, false),
            DaemonStatus::Stopped
        ));
    }

    #[test]
    fn orphan_identity_requires_both_start_times() {
        assert!(process_identity_matches(Some(123), Some(123)));
        assert!(!process_identity_matches(Some(123), Some(456)));
        // Unreadable current identity: unverifiable, so not a match.
        assert!(!process_identity_matches(Some(123), None));
    }

    #[test]
    fn signalling_is_refused_only_for_a_contradicted_identity() {
        // Provably someone else's process group: refuse.
        assert!(!signalling_pid_is_authorized(Some(123), Some(456)));
        // Verified as the daemon's own.
        assert!(signalling_pid_is_authorized(Some(123), Some(123)));
        // Unknown on either side. Stopping stays possible, because the user has
        // named this daemon and a record that cannot be verified must not become
        // one that can never be stopped.
        assert!(signalling_pid_is_authorized(None, Some(123)));
        assert!(signalling_pid_is_authorized(Some(123), None));
        assert!(signalling_pid_is_authorized(None, None));
    }

    #[test]
    fn orphan_identity_rejects_records_without_a_start_time() {
        // State written before start times were recorded. A process name used
        // to stand in here, but another copy of the same program on a recycled
        // PID matches a name, so such records are no longer verifiable and must
        // not authorize adopting or killing anything.
        assert!(!process_identity_matches(None, Some(123)));
        assert!(!process_identity_matches(None, None));
    }

    #[test]
    fn should_not_remove_when_state_title_differs_from_snapshot() {
        // The session was re-entered after the snapshot was taken, producing a
        // new title in state. The snapshot title is stale; skip removal.
        let session = ProjectSession {
            liveness_title: Some("new_title".to_string()),
        };
        let recorded_title = Some("old_title".to_string());

        assert!(!should_remove_liveness_session(
            &session,
            &recorded_title,
            Some("new_title"),
            true,
        ));
    }

    #[test]
    fn should_remove_when_running_title_mismatches() {
        let session = ProjectSession {
            liveness_title: Some("recorded_title".to_string()),
        };
        let recorded_title = Some("recorded_title".to_string());

        assert!(should_remove_liveness_session(
            &session,
            &recorded_title,
            Some("different_title"),
            true,
        ));
    }

    #[test]
    fn should_remove_when_dead() {
        let session = ProjectSession {
            liveness_title: Some("recorded_title".to_string()),
        };
        let recorded_title = Some("recorded_title".to_string());

        assert!(should_remove_liveness_session(
            &session,
            &recorded_title,
            Some("recorded_title"),
            false,
        ));
    }

    #[test]
    fn should_not_remove_when_alive_and_title_matches() {
        let session = ProjectSession {
            liveness_title: Some("recorded_title".to_string()),
        };
        let recorded_title = Some("recorded_title".to_string());

        assert!(!should_remove_liveness_session(
            &session,
            &recorded_title,
            Some("recorded_title"),
            true,
        ));
    }

    #[cfg(unix)]
    #[test]
    fn fd_scan_end_reaches_open_fds_above_the_fd_limit() {
        // A caller opened fd 900, then lowered RLIMIT_NOFILE to 256.
        assert_eq!(super::fd_scan_end(256, [0, 1, 2, 900]), 901);
        assert_eq!(super::fd_scan_end(256, [0, 1, 2, 10]), 256);
        assert_eq!(super::fd_scan_end(-1, []), 1 << 16);
        assert_eq!(super::fd_scan_end(libc::c_long::MAX, []), 1 << 20);
    }

    /// Exercises the fcntl fallback directly: on Linux >= 5.11 the spawn path
    /// never reaches it, but it is the only implementation on macOS.
    #[cfg(unix)]
    #[test]
    fn cloexec_fd_scan_hides_inherited_fds_from_children() {
        use std::os::unix::process::CommandExt;
        use std::process::{Command, Stdio};

        // A pipe without O_CLOEXEC, like bats' fd 3 or a wrapper script's pipe.
        let mut fds = [0; 2];
        assert_eq!(unsafe { libc::pipe(fds.as_mut_ptr()) }, 0);
        let fd = fds[1];
        let child_sees_fd = |scan: bool| {
            let mut cmd = Command::new("sh");
            cmd.arg("-c")
                .arg(format!("[ -e /dev/fd/{fd} ]"))
                .stdin(Stdio::null())
                .stdout(Stdio::null())
                .stderr(Stdio::null());
            if scan {
                let max_fd = super::max_inherited_fd();
                unsafe {
                    cmd.pre_exec(move || {
                        super::cloexec_fd_scan(max_fd);
                        Ok(())
                    });
                }
            }
            cmd.status().unwrap().success()
        };
        let without_scan = child_sees_fd(false);
        let with_scan = child_sees_fd(true);
        unsafe {
            libc::close(fds[0]);
            libc::close(fds[1]);
        }
        assert!(without_scan, "control: child should inherit fd {fd}");
        assert!(!with_scan, "fd {fd} leaked past cloexec_fd_scan");
    }
}