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 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};

use crate::procs::PROCS;
use crate::settings::settings;
use crate::state_file::StateFile;
use crate::{Result, env};
use duct::cmd;
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>,
    /// 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<()>>>,
    /// 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<()>>>>,
}

/// 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 fn start_if_not_running() -> Result<()> {
    let sf = StateFile::get();
    if let Some(d) = sf.daemons.get(&DaemonId::pitchfork())
        && let Some(pid) = d.pid
        && PROCS.is_running(pid)
    {
        return Ok(());
    }
    start_in_background()
}

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()?;
        cmd!(&*env::PITCHFORK_BIN, "supervisor", "run")
            .stdin_null()
            .stdout_null()
            .stderr_file(stderr_file)
            .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, 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();

        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,
                std::ptr::null(),
                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(())
}

/// 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.
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![]),
            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),
            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()),
        })
    }

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

        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");
                if !ca_cert_path.exists() || !ca_key_path.exists() {
                    match crate::proxy::server::generate_ca(&ca_cert_path, &ca_key_path) {
                        Ok(()) => {
                            info!(
                                "Generated local CA certificate at {}",
                                ca_cert_path.display()
                            );
                        }
                        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");
                    self.start_mdns().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()?;
        *self.ipc_shutdown.lock().await = Some(ipc_handle);
        self.start_state_flush_task();
        self.conn_watch(ipc).await
    }

    /// Start mDNS publishing for LAN mode (called after the proxy binds successfully).
    async fn start_mdns(&self) {
        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 lan_ip = if !s.proxy.lan_ip.is_empty() {
            match s.proxy.lan_ip.parse::<std::net::Ipv4Addr>() {
                Ok(ip) => Some(ip),
                Err(e) => {
                    error!(
                        "proxy.lan_ip {:?} is not a valid IPv4 address: {e}",
                        s.proxy.lan_ip
                    );
                    return;
                }
            }
        } else {
            match crate::proxy::lan_ip::detect_lan_ip().await {
                Some(ip) => Some(ip),
                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;
                }
            }
        };

        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?");
            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 (only when IP is auto-detected, not pinned).
        let ip_pinned = !s.proxy.lan_ip.is_empty();
        if !ip_pinned {
            let monitor_cancel = self.proxy_cancel.lock().await.clone();
            let publisher_clone = publisher.clone();
            let task = tokio::spawn(async move {
                let mut last_ip = lan_ip;
                let interval = std::time::Duration::from_secs(5);
                let mut ticker = tokio::time::interval(interval);
                ticker.tick().await; // first tick is immediate
                loop {
                    ticker.tick().await;
                    if let Some(cancel) = monitor_cancel.as_ref()
                        && cancel.is_cancelled()
                    {
                        break;
                    }
                    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;
                        let mut pub_guard = publisher_clone.lock().await;
                        pub_guard.republish_all(new_ip, port);
                    }
                }
            });
            *self.lan_monitor_task.lock().await = Some(task);
        }

        *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();

        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) {
        // 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.
        if let Some(monitor_task) = self.lan_monitor_task.lock().await.take() {
            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(proxy_task) = self.proxy_task.lock().await.take() {
            let _ = tokio::time::timeout(Duration::from_secs(12), 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();
        }

        // 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. Windows: named pipes have no filesystem component.
        #[cfg(unix)]
        let _ = fs::remove_dir_all(&*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` 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 neither `user` nor `SUDO_UID`/`SUDO_GID` are available (e.g. direct
/// root login), 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 SUDO_UID/SUDO_GID"
            );
            parse_sudo_ids()
        });
    }

    parse_sudo_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()))
}

/// Parse `SUDO_UID` and `SUDO_GID` environment variables into numeric IDs.
///
/// Returns `None` unless the effective UID is 0 (root). This prevents stale
/// `SUDO_UID`/`SUDO_GID` values inherited into non-sudo environments from
/// triggering incorrect `chown` operations.
#[cfg(unix)]
fn parse_sudo_ids() -> Option<(u32, u32)> {
    if !nix::unistd::Uid::effective().is_root() {
        return None;
    }
    let uid: u32 = std::env::var("SUDO_UID").ok()?.parse().ok()?;
    let gid: u32 = std::env::var("SUDO_GID").ok()?.parse().ok()?;
    Some((uid, gid))
}

/// 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);
        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);
        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,
) -> DaemonStatus {
    let same_boot = recorded_boot_time
        .is_some_and(|recorded| recorded.abs_diff(current_boot_time) <= BOOT_TIME_TOLERANCE_SECS);
    if status.is_running() && same_boot && supervisor_exited_uncleanly {
        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, process_identity_matches, should_remove_liveness_session,
        signalling_pid_is_authorized, 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),
            DaemonStatus::Errored(-1)
        ));
    }

    #[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),
            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),
            DaemonStatus::Errored(-1)
        ));
        let beyond = BOOT + BOOT_TIME_TOLERANCE_SECS + 1;
        assert!(matches!(
            unobserved_exit_status(&DaemonStatus::Running, Some(beyond), BOOT, true),
            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),
            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),
                    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),
            DaemonStatus::Stopped
        ));
    }

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