pitchfork_cli/supervisor/

state.rs

//! State access layer for the supervisor
//!
//! All state getter/setter operations for daemons, shell directories, and notifications.

use super::Supervisor;
use crate::Result;
use crate::daemon::Daemon;
use crate::daemon::RunOptions;
use crate::daemon_id::DaemonId;
use crate::daemon_status::DaemonStatus;
use crate::pitchfork_toml::CpuLimit;
use crate::pitchfork_toml::CronRetrigger;
use crate::pitchfork_toml::HealthCmd;
use crate::pitchfork_toml::HealthHttp;
use crate::pitchfork_toml::HealthPort;
use crate::pitchfork_toml::MemoryLimit;
use crate::pitchfork_toml::PitchforkToml;
use crate::pitchfork_toml::PortConfig;
use crate::pitchfork_toml::ReadyCmd;
use crate::pitchfork_toml::ReadyHttp;
use crate::pitchfork_toml::ReadyOutput;
use crate::pitchfork_toml::ReadyPort;
use crate::pitchfork_toml::Retry;
use crate::pitchfork_toml::StopConfig;
use crate::pitchfork_toml::WatchMode;
use crate::procs::PROCS;
use crate::state_file::DiskRecord;
use indexmap::IndexMap;
use std::collections::HashMap;
use std::path::PathBuf;

fn should_clean_daemon(
    id: &DaemonId,
    daemon: &Daemon,
    namespaces: &[String],
    daemons: &[DaemonId],
) -> bool {
    let namespace_matches =
        namespaces.is_empty() || namespaces.iter().any(|ns| ns == id.namespace());
    let daemon_matches = daemons.is_empty() || daemons.contains(id);
    daemon.pid.is_none() && namespace_matches && daemon_matches
}

fn prune_candidate(
    id: &DaemonId,
    daemon: &Daemon,
    namespaces: &[String],
    daemons: &[DaemonId],
) -> Option<(DaemonId, PathBuf)> {
    should_clean_daemon(id, daemon, namespaces, daemons)
        .then(|| daemon.dir.clone())
        .flatten()
        .map(|dir| (id.clone(), dir))
}

/// Options for upserting a daemon's state.
///
/// Use `UpsertDaemonOpts::builder(id)` to create, then set fields directly and call `.build()`.
#[derive(Debug, Default)]
pub(crate) struct UpsertDaemonOpts {
    pub id: DaemonId,
    pub pid: Option<u32>,
    pub status: DaemonStatus,
    pub shell_pid: Option<u32>,
    pub dir: Option<PathBuf>,
    pub cmd: Option<Vec<String>>,
    pub run: Option<String>,
    /// Start `cmd` directly, without a shell. `None` inherits the existing
    /// record's value, like `oneshot`.
    pub no_shell: Option<bool>,
    /// `None` inherits the existing record's value; see
    /// `Daemon::scheduled_from_config`.
    pub scheduled_from_config: Option<bool>,
    /// Stop the daemon when its directory is left. A start sets it from its
    /// `RunOptions`, so removing `auto = ["stop"]` takes effect; `None`
    /// inherits the existing record's value, like `oneshot`.
    pub autostop: Option<bool>,
    /// Run-to-completion task rather than a long-running service.
    /// `None` inherits the existing record's value, so a status-only upsert
    /// (stop, exit finalization) does not reclassify the daemon.
    pub oneshot: Option<bool>,
    /// A start sets it, `None` included, as it does `watch_base_dir`: an
    /// ad-hoc `pitchfork run` of a former config cron daemon's id must not keep
    /// that daemon's schedule, which also marks the record as made from
    /// config. Every other upsert inherits it.
    pub cron_schedule: Option<Option<String>>,
    pub cron_retrigger: Option<CronRetrigger>,
    pub cron_immediate: Option<bool>,
    pub last_exit_success: Option<bool>,
    pub retry: Option<Retry>,
    pub retry_count: Option<u32>,
    pub ready_delay: Option<u64>,
    pub ready_output: Option<ReadyOutput>,
    pub ready_http: Option<ReadyHttp>,
    pub ready_port: Option<ReadyPort>,
    /// See `Daemon::configured_ready_port`.
    pub configured_ready_port: Option<ReadyPort>,
    pub ready_cmd: Option<ReadyCmd>,
    /// The ready checks above replace the record's: one left `None` is
    /// cleared rather than inherited. See `RunOptions::replaces_ready_checks`.
    pub replaces_ready_checks: bool,
    pub health_cmd: Option<HealthCmd>,
    pub health_http: Option<HealthHttp>,
    pub health_port: Option<HealthPort>,
    /// Port configuration
    pub port: Option<PortConfig>,
    /// Resolved ports actually used after auto-bump (may differ from expected).
    /// `None` inherits the existing record's value; `Some` sets it explicitly,
    /// including an empty vec, which clears ports left by a previous run.
    pub resolved_port: Option<Vec<u16>>,
    /// The first port the process is actually listening on (detected at runtime).
    pub active_port: Option<u16>,
    /// Optional stable slug alias for this daemon.
    pub slug: Option<String>,
    /// Whether to proxy this daemon (None = use global proxy.enable setting).
    pub proxy: Option<bool>,
    pub depends: Option<Vec<DaemonId>>,
    pub env: Option<IndexMap<String, String>>,
    pub watch: Option<Vec<String>>,
    pub watch_mode: Option<WatchMode>,
    /// The directory of the project whose config defined the daemon. A
    /// start sets it, `None` included: an ad-hoc `pitchfork run` of a former
    /// config daemon's id must not keep that daemon's directory, which marks
    /// the record as made from config. Every other upsert inherits it.
    pub watch_base_dir: Option<Option<PathBuf>>,
    pub mise: Option<bool>,
    /// `Some` replaces the saved template context for a deferred `run`, including
    /// with `None` when a start no longer defers; `None` keeps what is saved.
    pub deferred_template_context: Option<Option<String>>,
    /// The project's mise binary. `None` keeps the record's, so a status-only
    /// upsert (stop, exit finalization) does not lose it; a start sets it from
    /// its `RunOptions`, so one removed from the project's settings is dropped.
    pub mise_bin: Option<Option<PathBuf>>,
    /// Unix user to run this daemon as
    pub user: Option<String>,
    /// Memory limit for the daemon process
    pub memory_limit: Option<MemoryLimit>,
    /// CPU usage limit as a percentage
    pub cpu_limit: Option<CpuLimit>,
    /// Unix signal to send for graceful shutdown
    pub stop_signal: Option<StopConfig>,
    /// Archive hook command invoked before retention prunes this daemon's logs.
    pub archive_hook: Option<String>,
    /// Log format for this daemon.
    pub log_format: Option<String>,
    /// Allocate a pseudo-terminal for the daemon process.
    pub pty: Option<bool>,
    /// True for config-only cron daemons auto-registered into state.
    pub config_registered: bool,
    /// A start that describes the daemon in full, so an unset configuration
    /// field clears the existing record's value instead of inheriting it. An
    /// ad-hoc run under a config daemon's id must not keep that daemon's
    /// command, schedule or environment. See `RunOptions::replaces_saved_record`.
    pub replaces_config: bool,
    /// Idle-shutdown ownership. `None` inherits the existing record's, so a
    /// status-only upsert (stop, exit finalization) keeps it; a start sets it
    /// from its `RunOptions`, which is what makes an explicit start explicit.
    pub proxy_idle_timeout_ms: Option<Option<u64>>,
}

/// Builder for UpsertDaemonOpts - ensures daemon ID is always provided.
///
/// # Example
/// ```ignore
/// let opts = UpsertDaemonOpts::builder(daemon_id)
///     .set(|o| {
///         o.pid = Some(pid);
///         o.status = DaemonStatus::Running;
///     })
///     .build();
/// ```
#[derive(Debug)]
pub(crate) struct UpsertDaemonOptsBuilder {
    pub opts: UpsertDaemonOpts,
}

impl UpsertDaemonOpts {
    /// Create a builder with the required daemon ID.
    pub fn builder(id: DaemonId) -> UpsertDaemonOptsBuilder {
        UpsertDaemonOptsBuilder {
            opts: UpsertDaemonOpts {
                id,
                ..Default::default()
            },
        }
    }

    /// Build an `UpsertDaemonOptsBuilder` from `RunOptions`, mapping all
    /// config-carried fields. Callers chain `.set()` to add runtime-specific
    /// fields (pid, resolved_port, etc.) and then call `.build()`.
    pub(crate) fn from_run_options(
        opts: &RunOptions,
        status: DaemonStatus,
    ) -> UpsertDaemonOptsBuilder {
        UpsertDaemonOpts::builder(opts.id.clone()).set(|o| {
            o.replaces_config = opts.replaces_saved_record;
            o.status = status;
            o.shell_pid = opts.shell_pid;
            o.dir = Some(opts.dir.0.clone());
            o.cmd = Some(opts.cmd.clone());
            o.run = opts.run.clone();
            o.no_shell = Some(opts.no_shell);
            // Only a client clears it; the supervisor's own starts (the
            // schedule, retries, file watching) leave it as it was.
            o.scheduled_from_config = opts.requested_by_client.then_some(false);
            o.autostop = Some(opts.autostop);
            o.oneshot = Some(opts.oneshot);
            o.cron_schedule = Some(opts.cron_schedule.clone());
            o.cron_retrigger = opts.cron_retrigger;
            o.cron_immediate = opts.cron_immediate;
            o.retry = Some(opts.retry);
            o.retry_count = Some(opts.retry_count);
            o.ready_delay = opts.ready_delay;
            o.ready_output = opts.ready_output.clone();
            o.ready_http = opts.ready_http.clone();
            o.ready_port = opts.ready_port.clone();
            o.configured_ready_port = opts.ready_port.clone();
            o.ready_cmd = opts.ready_cmd.clone();
            o.replaces_ready_checks = opts.replaces_ready_checks;
            o.health_cmd = opts.health_cmd.clone();
            o.health_http = opts.health_http.clone();
            o.health_port = opts.health_port.clone();
            o.port = opts.port.clone();
            o.depends = Some(opts.depends.clone());
            o.env = opts.env.clone();
            o.watch = Some(opts.watch.clone());
            o.watch_mode = Some(opts.watch_mode);
            o.watch_base_dir = Some(opts.watch_base_dir.clone());
            o.mise = opts.mise;
            o.deferred_template_context = Some(opts.deferred_template_context.clone());
            o.mise_bin = Some(opts.mise_bin.clone());
            o.user = opts.user.clone();
            o.memory_limit = opts.memory_limit;
            o.cpu_limit = opts.cpu_limit;
            o.stop_signal = opts.stop_signal;
            o.pty = opts.pty;
            o.archive_hook = opts.archive_hook.clone();
            o.log_format = opts.log_format.clone();
            o.proxy_idle_timeout_ms = Some(opts.proxy_idle_timeout_ms);
        })
    }
}

impl UpsertDaemonOptsBuilder {
    /// Modify opts fields with a closure.
    pub fn set<F: FnOnce(&mut UpsertDaemonOpts)>(mut self, f: F) -> Self {
        f(&mut self.opts);
        self
    }

    /// Build the UpsertDaemonOpts.
    pub fn build(self) -> UpsertDaemonOpts {
        self.opts
    }
}

impl Supervisor {
    /// Put this supervisor's record back into the state file if the file no
    /// longer has it.
    ///
    /// The CLI decides whether a supervisor is running, and which process
    /// `supervisor stop` signals, from this record. The supervisor only
    /// writes the file when its own state changes, so if the file is replaced
    /// or rewritten by something else the record could stay missing
    /// indefinitely, leaving this supervisor running but unaccounted for.
    /// Only the record is restored; the rest of the file is left as it is.
    pub(crate) async fn restore_own_record(&self) {
        if self
            .shutting_down
            .load(std::sync::atomic::Ordering::Acquire)
        {
            return;
        }
        let pitchfork_id = DaemonId::pitchfork();
        let state = self.state_file.lock().await;
        // Checked again under the lock: `close()` may have begun while this
        // waited for it. (Its removal of the record happens under this lock
        // too, so a restore can never outlive it; this just skips the work.)
        if self
            .shutting_down
            .load(std::sync::atomic::Ordering::Acquire)
        {
            return;
        }
        let Some(own) = state
            .daemons
            .get(&pitchfork_id)
            .filter(|d| d.pid.is_some())
            .cloned()
        else {
            return;
        };
        let own_pid = own.pid.unwrap_or_default();
        let path = state.path.clone();
        // The state lock stays held across the file work, as it is for a
        // flush, so the two cannot interleave.
        let result = tokio::task::spawn_blocking(move || {
            crate::state_file::StateFile::restore_daemon_in_file(&path, &own)
        })
        .await;
        match result {
            Ok(Ok(DiskRecord::Present)) => {}
            Ok(Ok(DiskRecord::Restored)) => {
                // Not flushed yet (e.g. a client connecting right after
                // startup): the record was simply written early.
                if state.is_dirty() {
                    debug!("recorded this supervisor in the state file ahead of the next flush");
                } else {
                    warn!(
                        "state file {} no longer recorded this supervisor (pid {own_pid}); restored it",
                        state.path.display()
                    );
                }
                // The file now differs from what this supervisor last wrote,
                // so its next flush must not be skipped as unchanged.
                state.forget_written_snapshot();
            }
            Ok(Ok(DiskRecord::Unparseable)) => {
                warn!(
                    "state file {} cannot be parsed; rewriting it from the supervisor's state",
                    state.path.display()
                );
                state.force_next_write();
            }
            Ok(Err(e)) => warn!("failed to restore the supervisor record in the state file: {e}"),
            Err(e) => warn!("failed to restore the supervisor record in the state file: {e}"),
        }
    }

    /// Run options for starting `id` from its config, as `pitchfork start`
    /// would build them: templates rendered and the top-level `[env]` merged.
    ///
    /// Other daemons' ports come from the state held here rather than the file
    /// on disk, which can trail it — a daemon started a moment ago may not be
    /// written out yet, and a template referring to its port would fail.
    pub(crate) async fn run_options_from_config(
        &self,
        id: &DaemonId,
        config: &crate::pitchfork_toml::PitchforkTomlDaemon,
        pt: &PitchforkToml,
    ) -> Result<RunOptions> {
        let resolved_daemons: std::collections::HashMap<DaemonId, Vec<u16>> = {
            let state = self.state_file.lock().await;
            state
                .daemons
                .iter()
                .filter(|(_, d)| !d.resolved_port.is_empty())
                .map(|(id, d)| (id.clone(), d.resolved_port.clone()))
                .collect()
        };
        // On a blocking worker, as for a client's start: building the context
        // reads configuration and derives hostnames by walking the project's
        // checkouts, which must not hold up the supervisor's other tasks.
        let id = id.clone();
        let mut config = config.clone();
        let pt = pt.clone();
        tokio::task::spawn_blocking(move || {
            crate::ipc::batch::render_daemon_config_with(&id, &mut config, &pt, &resolved_daemons)?;
            let cmd = config
                .run
                .argv()
                .map_err(|e| miette::miette!("failed to parse command for daemon {id}: {e}"))?;
            let mut opts = config.to_run_options(&id, cmd);
            // A client's start resolves these from the daemon's project before
            // it reaches the supervisor, which would otherwise use its own
            // settings. Rendering assumed the project's, so launch has to as well.
            let (mise, mise_bin) = crate::ipc::batch::project_mise_options(&config);
            opts.mise = Some(mise);
            opts.mise_bin = mise_bin;
            Ok(opts)
        })
        .await
        .map_err(|e| miette::miette!("rendering daemon config panicked: {e}"))?
    }

    /// Upsert a daemon's state, merging with existing values
    pub(crate) async fn upsert_daemon(&self, opts: UpsertDaemonOpts) -> Result<Daemon> {
        info!(
            "upserting daemon: {} pid: {} status: {}",
            opts.id,
            opts.pid.unwrap_or(0),
            opts.status
        );
        let mut state_file = self.state_file.lock().await;
        let existing = state_file.daemons.get(&opts.id);
        // What configuration fields fall back to: nothing for a start, which
        // sets them all; see `UpsertDaemonOpts::replaces_config`.
        let configured = existing.filter(|_| !opts.replaces_config);
        // What unset ready checks fall back to: also nothing when this start's
        // ready checks replace the daemon's; see `replaces_ready_checks`.
        let ready_base = configured.filter(|_| !opts.replaces_ready_checks);
        let daemon = Daemon {
            id: opts.id.clone(),
            // title/start_time identify the process for orphan cleanup after a
            // supervisor crash. They are looked up from the process cache; if
            // the cache has no entry (e.g. an upsert between refreshes) fall
            // back to the existing values as long as the PID is unchanged, so
            // the recorded identity is not wiped mid-lifetime.
            title: opts.pid.and_then(|pid| {
                PROCS.title(pid).or_else(|| {
                    existing
                        .filter(|d| d.pid == Some(pid))
                        .and_then(|d| d.title.clone())
                })
            }),
            start_time: opts.pid.and_then(|pid| {
                PROCS.start_time(pid).or_else(|| {
                    existing
                        .filter(|d| d.pid == Some(pid))
                        .and_then(|d| d.start_time)
                })
            }),
            // Boot time is a system-wide constant, so it needs no cache
            // fallback — it is recorded for any daemon with a live PID and
            // dropped when the PID is cleared.
            boot_time: opts.pid.map(|_| PROCS.boot_time()),
            pid: opts.pid,
            status: opts.status,
            shell_pid: opts.shell_pid,
            autostop: opts
                .autostop
                .unwrap_or_else(|| existing.is_some_and(|d| d.autostop)),
            // A start carries the current config value (including a removed
            // `oneshot = true`, which must reclassify the daemon); every other
            // upsert leaves it None and inherits, so finalizing a completed
            // oneshot's exit does not forget what it is.
            oneshot: opts
                .oneshot
                .unwrap_or_else(|| existing.is_some_and(|d| d.oneshot)),
            dir: opts.dir.or(configured.and_then(|d| d.dir.clone())),
            cmd: opts.cmd.or(configured.and_then(|d| d.cmd.clone())),
            // A start in the argv form has no command line, and must not
            // inherit the one a shell-form run left behind.
            run: if opts.no_shell == Some(true) {
                None
            } else {
                opts.run.or(configured.and_then(|d| d.run.clone()))
            },
            no_shell: opts
                .no_shell
                .unwrap_or_else(|| existing.is_some_and(|d| d.no_shell)),
            scheduled_from_config: opts
                .scheduled_from_config
                .unwrap_or_else(|| existing.is_some_and(|d| d.scheduled_from_config)),
            cron_schedule: opts
                .cron_schedule
                .unwrap_or_else(|| configured.and_then(|d| d.cron_schedule.clone())),
            cron_retrigger: opts
                .cron_retrigger
                .or(configured.and_then(|d| d.cron_retrigger)),
            cron_immediate: opts
                .cron_immediate
                .or(configured.and_then(|d| d.cron_immediate)),
            last_cron_triggered: existing.and_then(|d| d.last_cron_triggered),
            last_cron_run: existing.and_then(|d| d.last_cron_run),
            last_exit_success: opts
                .last_exit_success
                .or(existing.and_then(|d| d.last_exit_success)),
            retry: opts
                .retry
                .unwrap_or_else(|| existing.map(|d| d.retry).unwrap_or_default()),
            retry_count: opts
                .retry_count
                .unwrap_or(existing.map(|d| d.retry_count).unwrap_or(0)),
            ready_delay: opts.ready_delay.or(ready_base.and_then(|d| d.ready_delay)),
            ready_output: opts
                .ready_output
                .or(ready_base.and_then(|d| d.ready_output.clone())),
            ready_http: opts
                .ready_http
                .or(ready_base.and_then(|d| d.ready_http.clone())),
            ready_port: opts
                .ready_port
                .or(ready_base.and_then(|d| d.ready_port.clone())),
            configured_ready_port: opts
                .configured_ready_port
                .or(ready_base.and_then(|d| d.configured_ready_port.clone())),
            ready_cmd: opts
                .ready_cmd
                .or(ready_base.and_then(|d| d.ready_cmd.clone())),
            health_cmd: opts
                .health_cmd
                .or(configured.and_then(|d| d.health_cmd.clone())),
            health_http: opts
                .health_http
                .or(configured.and_then(|d| d.health_http.clone())),
            health_port: opts
                .health_port
                .or(configured.and_then(|d| d.health_port.clone())),
            port: opts
                .port
                .or_else(|| configured.and_then(|d| d.port.clone())),
            resolved_port: match opts.resolved_port {
                Some(ports) => ports,
                None => existing
                    .map(|d| d.resolved_port.clone())
                    .unwrap_or_default(),
            },
            depends: opts
                .depends
                .unwrap_or_else(|| existing.map(|d| d.depends.clone()).unwrap_or_default()),
            env: opts.env.or(configured.and_then(|d| d.env.clone())),
            watch: opts
                .watch
                .unwrap_or_else(|| existing.map(|d| d.watch.clone()).unwrap_or_default()),
            watch_mode: opts
                .watch_mode
                .unwrap_or_else(|| existing.map(|d| d.watch_mode).unwrap_or_default()),
            watch_base_dir: opts
                .watch_base_dir
                .unwrap_or_else(|| configured.and_then(|d| d.watch_base_dir.clone())),
            mise: opts.mise.or(configured.and_then(|d| d.mise)),
            deferred_template_context: opts
                .deferred_template_context
                .unwrap_or_else(|| existing.and_then(|d| d.deferred_template_context.clone())),
            mise_bin: opts
                .mise_bin
                .unwrap_or_else(|| configured.and_then(|d| d.mise_bin.clone())),
            user: opts.user.or(configured.and_then(|d| d.user.clone())),
            proxy: opts.proxy.or(existing.and_then(|d| d.proxy)),
            // active_port is intentionally NOT inherited from the existing daemon.
            // When a daemon restarts, the new process has not yet bound a port, so
            // carrying over the old process's active_port would cause the proxy to
            // route to a port that is no longer listening.  The port will be
            // re-detected by detect_and_store_active_port once the new process is ready.
            active_port: opts.active_port,
            slug: opts.slug.or(existing.and_then(|d| d.slug.clone())),
            memory_limit: opts
                .memory_limit
                .or(configured.and_then(|d| d.memory_limit)),
            cpu_limit: opts.cpu_limit.or(configured.and_then(|d| d.cpu_limit)),
            stop_signal: opts.stop_signal.or(configured.and_then(|d| d.stop_signal)),
            archive_hook: opts
                .archive_hook
                .or(configured.and_then(|d| d.archive_hook.clone())),
            log_format: opts
                .log_format
                .or(configured.and_then(|d| d.log_format.clone())),
            pty: opts.pty.or(configured.and_then(|d| d.pty)),
            config_registered: opts.config_registered,
            proxy_idle_timeout_ms: opts
                .proxy_idle_timeout_ms
                .unwrap_or_else(|| existing.and_then(|d| d.proxy_idle_timeout_ms)),
        };
        state_file.insert_daemon(&opts.id, daemon.clone());
        Ok(daemon)
    }

    /// Enable a daemon (remove from disabled set)
    pub async fn enable(&self, id: &DaemonId) -> Result<bool> {
        info!("enabling daemon: {id}");
        let config = PitchforkToml::all_merged_all_namespaces().await?;
        let mut state_file = self.state_file.lock().await;
        let exists = state_file.daemons.contains_key(id) || config.daemons.contains_key(id);
        if !exists {
            return Err(miette::miette!("daemon '{}' not found", id));
        }
        let result = state_file.enable_daemon(id);
        Ok(result)
    }

    /// Disable a daemon (add to disabled set)
    pub async fn disable(&self, id: &DaemonId) -> Result<bool> {
        info!("disabling daemon: {id}");
        let config = PitchforkToml::all_merged_all_namespaces().await?;
        let not_found = || miette::miette!("daemon '{}' not found", id);
        // Checked before `stop_lock`, whose entries are never removed, so an
        // unknown ID does not leave one behind.
        if !config.daemons.contains_key(id)
            && !self.state_file.lock().await.daemons.contains_key(id)
        {
            return Err(not_found());
        }
        // Taken before the state file, in `run_inner`'s order, so a retry
        // that has already checked the disabled set finishes its start first.
        let lock = self.stop_lock(id).await;
        let _guard = lock.lock().await;
        let mut state_file = self.state_file.lock().await;
        let exists = state_file.daemons.contains_key(id) || config.daemons.contains_key(id);
        if !exists {
            return Err(not_found());
        }
        let result = state_file.disable_daemon(id);
        // A start sleeping out a retry backoff makes no further attempt, so
        // it is woken now rather than when the backoff ends.
        self.cancel_retrying(id);
        Ok(result)
    }

    /// Get a daemon by ID
    pub(crate) async fn get_daemon(&self, id: &DaemonId) -> Option<Daemon> {
        self.state_file.lock().await.daemons.get(id).cloned()
    }

    /// Get all active daemons (those with PIDs, excluding pitchfork itself)
    pub(crate) async fn active_daemons(&self) -> Vec<Daemon> {
        let pitchfork_id = DaemonId::pitchfork();
        self.state_file
            .lock()
            .await
            .daemons
            .values()
            .filter(|d| d.pid.is_some() && d.id != pitchfork_id)
            .cloned()
            .collect()
    }

    /// Remove a daemon from state
    pub(crate) async fn remove_daemon(&self, id: &DaemonId) -> Result<()> {
        let mut state_file = self.state_file.lock().await;
        state_file.remove_daemon(id);
        Ok(())
    }

    /// Set the shell's working directory
    pub(crate) async fn set_shell_dir(&self, shell_pid: u32, dir: PathBuf) -> Result<()> {
        let mut state_file = self.state_file.lock().await;
        state_file.set_shell_dir(shell_pid, dir);
        Ok(())
    }

    /// Get the shell's working directory
    pub(crate) async fn get_shell_dir(&self, shell_pid: u32) -> Option<PathBuf> {
        self.state_file
            .lock()
            .await
            .shell_dirs
            .get(&shell_pid.to_string())
            .cloned()
    }

    /// Remove a shell PID from tracking
    #[cfg(unix)]
    pub(crate) async fn remove_shell_pid(&self, shell_pid: u32) -> Result<()> {
        let mut state_file = self.state_file.lock().await;
        state_file.remove_shell_dir(shell_pid);
        Ok(())
    }

    /// Get all directories with their associated shell PIDs
    pub(crate) async fn get_dirs_with_shell_pids(&self) -> HashMap<PathBuf, Vec<u32>> {
        self.state_file.lock().await.shell_dirs.iter().fold(
            HashMap::new(),
            |mut acc, (pid, dir)| {
                if let Ok(pid) = pid.parse() {
                    acc.entry(dir.clone()).or_default().push(pid);
                }
                acc
            },
        )
    }

    /// Get pending notifications and clear the queue
    pub(crate) async fn get_notifications(&self) -> Vec<(log::LevelFilter, String)> {
        self.pending_notifications.lock().await.drain(..).collect()
    }

    /// Clean up daemons that have no PID
    pub(crate) async fn clean(&self) -> Result<()> {
        self.clean_filtered(&[], &[], false).await?;
        Ok(())
    }

    /// Clean up PID-less daemon registrations matching all supplied filters.
    pub(crate) async fn clean_filtered(
        &self,
        namespaces: &[String],
        daemons: &[DaemonId],
        prune: bool,
    ) -> Result<u64> {
        if prune {
            let candidates: Vec<(DaemonId, PathBuf)> = {
                let state_file = self.state_file.lock().await;
                state_file
                    .daemons
                    .iter()
                    .filter_map(|(id, daemon)| prune_candidate(id, daemon, namespaces, daemons))
                    .collect()
            };

            let mut missing = HashMap::new();
            for (id, dir) in candidates {
                match tokio::fs::try_exists(&dir).await {
                    Ok(false) => {
                        missing.insert(id, dir);
                    }
                    Ok(true) => {}
                    // Unknown is not gone: keep the record, prune the rest.
                    Err(e) => warn!("cannot check {} to prune {id}: {e}", dir.display()),
                }
            }

            let mut removed = 0;
            let mut state_file = self.state_file.lock().await;
            state_file.retain_daemons(|id, daemon| {
                let remove = should_clean_daemon(id, daemon, namespaces, daemons)
                    && missing
                        .get(id)
                        .is_some_and(|missing_dir| daemon.dir.as_ref() == Some(missing_dir));
                removed += u64::from(remove);
                !remove
            });
            return Ok(removed);
        }

        let mut removed = 0;
        let mut state_file = self.state_file.lock().await;
        state_file.retain_daemons(|id, daemon| {
            let remove = should_clean_daemon(id, daemon, namespaces, daemons);
            removed += u64::from(remove);
            !remove
        });
        Ok(removed)
    }

    /// Return the union of active directories from shell tracking and project
    /// sessions. These are the directories that should keep auto-stop daemons
    /// alive.
    pub(crate) async fn get_active_directories(&self) -> Vec<PathBuf> {
        self.state_file.lock().await.active_directories()
    }

    /// Collect all project sessions as `(pid, dir, liveness_title)`. Every
    /// project session carries a host PID in its key, so every session is a
    /// liveness session. Used by the refresh loop to clean up stale sessions.
    pub(crate) async fn get_liveness_sessions(&self) -> Vec<(u32, PathBuf, Option<String>)> {
        self.state_file
            .lock()
            .await
            .iter_project_sessions()
            .into_iter()
            .filter_map(|(pid_str, dir, session)| {
                pid_str
                    .parse()
                    .ok()
                    .map(|pid| (pid, dir.clone(), session.liveness_title.clone()))
            })
            .collect()
    }

    /// Build a snapshot of all project sessions with live liveness status
    /// filled in from `PROCS`. Used to answer `GetProjectSessions` IPC
    /// requests.
    pub(crate) async fn get_project_sessions_info(&self) -> Vec<crate::ipc::ProjectSessionInfo> {
        let sessions: Vec<(u32, PathBuf, Option<String>)> = self
            .state_file
            .lock()
            .await
            .iter_project_sessions()
            .into_iter()
            .filter_map(|(pid_str, dir, session)| {
                pid_str
                    .parse()
                    .ok()
                    .map(|pid| (pid, dir.clone(), session.liveness_title.clone()))
            })
            .collect();
        let pids: Vec<u32> = sessions.iter().map(|(pid, _, _)| *pid).collect();
        if !pids.is_empty() {
            PROCS.refresh_pids(&pids);
        }
        sessions
            .into_iter()
            .map(
                |(pid, directory, liveness_title)| crate::ipc::ProjectSessionInfo {
                    pid,
                    directory,
                    liveness_title,
                    alive: PROCS.is_running(pid),
                    current_title: PROCS.title(pid),
                },
            )
            .collect()
    }

    /// Atomically enter (or replace) a project session for `(pid, dir)`.
    /// Returns the previous session, if any, so the caller can evaluate the
    /// previous entry for autostop.
    pub(crate) async fn enter_project_session(
        &self,
        pid: u32,
        dir: PathBuf,
    ) -> Result<Option<crate::state_file::ProjectSession>> {
        if pid == 0 {
            return Err(miette::miette!("invalid host PID 0"));
        }
        if pid > i32::MAX as u32 {
            return Err(miette::miette!("host PID {pid} exceeds i32::MAX"));
        }
        PROCS.refresh_pids(&[pid]);
        let liveness_title = PROCS.title(pid);
        // On Unix, reject dead host PIDs up front so we don't register a
        // session the refresh loop would immediately evict. On Windows, Git
        // Bash `$$` is a Cygwin-internal PID invisible to sysinfo, so the
        // liveness check is skipped and sessions rely on explicit
        // `project leave`.
        #[cfg(unix)]
        if !PROCS.is_running(pid) {
            return Err(miette::miette!("host PID {pid} is not running"));
        }
        let mut state_file = self.state_file.lock().await;
        let previous = state_file.set_project_session(
            pid,
            dir,
            crate::state_file::ProjectSession { liveness_title },
        );
        Ok(previous)
    }

    /// Atomically remove a project session for `(pid, dir)`. Returns the
    /// directory of the removed session so the caller can evaluate it for
    /// autostop.
    pub(crate) async fn leave_project_session(
        &self,
        pid: u32,
        dir: &std::path::Path,
    ) -> Result<Option<PathBuf>> {
        let mut state_file = self.state_file.lock().await;
        if state_file.remove_project_session(pid, dir).is_some() {
            Ok(Some(dir.to_path_buf()))
        } else {
            Ok(None)
        }
    }
}

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

    fn daemon(id: &DaemonId, pid: Option<u32>, dir: Option<PathBuf>) -> Daemon {
        Daemon {
            id: id.clone(),
            pid,
            dir,
            ..Daemon::default()
        }
    }

    #[test]
    fn clean_filters_intersect_and_preserve_running_daemons() {
        let api = DaemonId::new("project-a", "api");
        let worker = DaemonId::new("project-a", "worker");
        let namespaces = vec!["project-a".to_string()];
        let daemons = vec![api.clone()];

        assert!(should_clean_daemon(
            &api,
            &daemon(&api, None, None),
            &namespaces,
            &daemons,
        ));
        assert!(!should_clean_daemon(
            &worker,
            &daemon(&worker, None, None),
            &namespaces,
            &daemons,
        ));
        assert!(!should_clean_daemon(
            &api,
            &daemon(&api, Some(42), None),
            &namespaces,
            &daemons,
        ));
    }

    #[test]
    fn prune_candidates_require_a_recorded_directory() {
        let id = DaemonId::new("project-a", "api");
        assert!(prune_candidate(&id, &daemon(&id, None, None), &[], &[]).is_none());
        assert_eq!(
            prune_candidate(
                &id,
                &daemon(&id, None, Some(PathBuf::from("missing"))),
                &[],
                &[]
            ),
            Some((id, PathBuf::from("missing")))
        );
    }
}