subc_daemon/

bootstrap.rs

use std::{
    collections::BTreeMap,
    env,
    error::Error,
    ffi::OsString,
    fmt, fs, io,
    net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr},
    path::{Path, PathBuf},
    process,
    time::Duration,
};

use fs4::{FileExt, TryLockError};
use subc_protocol::PROTOCOL_VERSION;
pub use subc_transport::user_connection_token;
use subc_transport::{
    authenticate_client, connection_file, generate_daemon_id, generate_key, write_atomic,
    AuthError, ConnectionFileError, ConnectionInfo, Endpoint, SCHEMA_VERSION,
};
use tokio::{
    net::{TcpListener, TcpStream},
    task::{JoinError, JoinHandle},
    time::{sleep, timeout},
};
use tracing::{error, info, warn};

use crate::{
    daemon_config::{self, ConfiguredModule, DaemonConfigError},
    server::{serve_listeners, ServerAuth, ServerError},
    supervise::HealthConfig,
    ConnectedClients, ControlHandler, DaemonSelfWatchdog, DaemonSelfWatchdogConfig,
    ForwardingTable, Registry, RestartPolicy, Router, Supervisor, SupervisorHandle,
    SupervisorProcessLiveness,
};
use std::sync::Arc;

pub const DEFAULT_SUBC_PORT: u16 = 8757;
pub const SUBC_PORT_ENV: &str = "SUBC_PORT";
use subc_transport::CONNECTION_FILE_NAME;
const DAEMON_VERSION: &str = env!("CARGO_PKG_VERSION");
const CONNECT_TIMEOUT: Duration = Duration::from_secs(2);
const PROBE_AUTH_DEADLINE: Duration = Duration::from_secs(2);
const START_LOCK_RETRIES: usize = 40;
const START_LOCK_RETRY_DELAY: Duration = Duration::from_millis(25);

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ConnectionFileSource {
    XdgRuntimeDir,
    TempDirFallback,
    Explicit,
}

impl ConnectionFileSource {
    fn reason(self) -> &'static str {
        match self {
            Self::XdgRuntimeDir => "XDG_RUNTIME_DIR set and non-empty",
            Self::TempDirFallback => "XDG_RUNTIME_DIR unset or empty",
            Self::Explicit => "configured path",
        }
    }
}

impl fmt::Display for ConnectionFileSource {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter.write_str(match self {
            Self::XdgRuntimeDir => "xdg_runtime_dir",
            Self::TempDirFallback => "temp_dir_fallback",
            Self::Explicit => "explicit",
        })
    }
}

/// Runtime bootstrap configuration. Production uses the default fixed port and
/// optional daemon-config override; tests pass port 0 to let the OS assign a free
/// loopback port and discover it from the connection file.
#[derive(Debug, Clone, Default)]
struct AdmissionFactsConfig {
    carrier_module_id: Option<String>,
    targets: Option<Vec<String>>,
}

/// Controls where module cgroups are prepared.
///
/// In-process daemons default to [`Self::Disabled`] so they never derive a
/// production location from the host process. The shipped daemon explicitly uses
/// [`Self::Current`]. The `ck-subc` binary accepts
/// `SUBC_CGROUP_PLACEMENT=disabled` for isolated test processes; an unset
/// variable retains `Current`, and every other value is rejected at startup.
/// Tests that exercise placement can own a [`Self::Root`].
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub enum CgroupPlacementConfig {
    #[default]
    Disabled,
    Current,
    Root(PathBuf),
}

#[derive(Debug, Clone)]
pub struct BootstrapConfig {
    pub operator_queue_wait: Duration,
    pub operator_timeout: Duration,
    pub operator_stuck_grace: Duration,
    #[cfg(feature = "test-support")]
    pub operator_script: Option<PathBuf>,
    #[cfg(feature = "test-support")]
    pub operator_events: Option<PathBuf>,
    pub connection_file_path: PathBuf,
    pub port: u16,
    pub daemon_ver: String,
    configured_modules: Vec<ConfiguredModule>,
    storage_config: Option<daemon_config::StorageConfig>,
    admission_facts: AdmissionFactsConfig,
    /// Module ids whose scopes may carry the attributes that grant authority.
    scope_authority_owners: Vec<String>,
    activity_config: daemon_config::ActivityConfig,
    daemon_config_path: Option<PathBuf>,
    configured_port: Option<u16>,
    /// Daemon-wide route.bind relay budget in milliseconds (the fallback for
    /// any module without a per-module override). `None` = built-in default
    /// (12s — see `control::DEFAULT_ROUTE_BIND_RELAY_TIMEOUT`).
    route_bind_relay_default_ms: Option<u64>,
    reserved_capabilities: BTreeMap<String, String>,
    watchdog_config: DaemonSelfWatchdogConfig,
    connection_file_source: ConnectionFileSource,
    /// Where module cgroups are prepared. Disabled unless a caller explicitly
    /// opts in, because `Current` derives a host location from `/proc/self/cgroup`.
    cgroup_placement: CgroupPlacementConfig,
    /// Directory the supervisor writes per-module stdout/stderr capture files
    /// into. `None` disables capture; the shipped binary supplies its real run
    /// directory explicitly.
    capture_logs_dir: Option<PathBuf>,
    /// File the supervisor appends every module's terminal exits to, so exit
    /// history survives a daemon restart. `None` keeps terminal history in
    /// memory only (each module's ring). Absent by default for the same reason
    /// as `capture_logs_dir`: an in-process daemon booted by a test must not
    /// append its exits to the operator's real `terminals.jsonl`, where they
    /// would show up in `ck module terminals`. The shipped binary supplies
    /// `<run dir>/terminals.jsonl` explicitly.
    terminal_journal_path: Option<PathBuf>,
    /// Where the machine id is read from, or minted into when absent. `None`
    /// serves no machine id. Absent by default for the same reason as
    /// `capture_logs_dir`: an in-process daemon booted by a test must never
    /// derive the operator's real data home and mint into it. The shipped binary
    /// supplies `<data home>/cortexkit/machine-id` explicitly.
    machine_id_path: Option<PathBuf>,
    /// The live-children record: every supervised process this daemon has
    /// running, kept so the next daemon can end the ones a crash left behind.
    /// At startup, before any module is spawned, the previous daemon's record
    /// here is swept. `None` keeps no record and sweeps nothing, for the same
    /// reason as `capture_logs_dir`: an in-process daemon booted by a test
    /// must never signal processes listed in the operator's real run
    /// directory. The shipped binary supplies `<run dir>/live-children.json`.
    live_children_path: Option<PathBuf>,
    privacy_trampoline: Option<PathBuf>,
}

impl BootstrapConfig {
    pub fn new(connection_file_path: impl Into<PathBuf>, port: u16) -> Self {
        Self {
            connection_file_path: connection_file_path.into(),
            operator_queue_wait: Duration::from_secs(150),
            operator_timeout: Duration::from_secs(120),
            operator_stuck_grace: Duration::from_secs(10),
            #[cfg(feature = "test-support")]
            operator_script: None,
            #[cfg(feature = "test-support")]
            operator_events: None,
            port,
            daemon_ver: DAEMON_VERSION.to_owned(),
            configured_modules: Vec::new(),
            storage_config: None,
            admission_facts: AdmissionFactsConfig::default(),
            scope_authority_owners: daemon_config::default_scope_authority_owners(),
            activity_config: daemon_config::ActivityConfig::default(),
            daemon_config_path: None,
            configured_port: None,
            route_bind_relay_default_ms: None,
            reserved_capabilities: BTreeMap::new(),
            watchdog_config: DaemonSelfWatchdogConfig::default(),
            connection_file_source: ConnectionFileSource::Explicit,
            cgroup_placement: CgroupPlacementConfig::default(),
            capture_logs_dir: None,
            terminal_journal_path: None,
            machine_id_path: None,
            live_children_path: None,
            privacy_trampoline: None,
        }
    }

    /// Keep the live-children record at `path`, and at startup end the
    /// processes a previous daemon recorded there that are still running.
    /// Embedding daemons and tests pass a path inside their own fixture tree.
    pub fn with_live_children_record(mut self, path: impl Into<PathBuf>) -> Self {
        self.live_children_path = Some(path.into());
        self
    }

    /// macOS module launches require an explicit executable implementing the
    /// subc-os trampoline entry point. Other systems ignore this setting.
    pub fn with_privacy_trampoline(mut self, path: impl Into<PathBuf>) -> Self {
        self.privacy_trampoline = Some(path.into());
        self
    }

    /// Serve the machine id stored at `path`, minting it there at startup when
    /// the file is absent. Embedding daemons and tests pass a path inside their
    /// own fixture tree.
    pub fn with_machine_id_path(mut self, path: impl Into<PathBuf>) -> Self {
        self.machine_id_path = Some(path.into());
        self
    }

    /// Selects module cgroup placement. The default is disabled.
    pub fn with_cgroup_placement(mut self, placement: CgroupPlacementConfig) -> Self {
        self.cgroup_placement = placement;
        self
    }

    /// Redirects per-module stdout/stderr capture files out of the real run
    /// directory. Tests that start an in-process daemon must call this with a
    /// path inside their fixture tree.
    pub fn with_capture_logs_dir(mut self, dir: impl Into<PathBuf>) -> Self {
        self.capture_logs_dir = Some(dir.into());
        self
    }

    /// Redirects the daemon-private journal, allowing embedded daemons and tests
    /// to keep their observations out of the operator's live run directory.
    pub fn with_terminal_journal_path(mut self, path: impl Into<PathBuf>) -> Self {
        self.terminal_journal_path = Some(path.into());
        self
    }

    pub fn from_env() -> Result<Self, BootstrapError> {
        Self::from_env_with_daemon_config_path(daemon_config::default_config_path())
    }

    /// The SHIPPED BINARY's config, which is the only caller that should capture
    /// child output into the operator's real run directory.
    ///
    /// Kept separate from `from_env` deliberately: see `capture_logs_dir` and
    /// `terminal_journal_path` on this struct for why an absent value must mean
    /// NO CAPTURE and NO JOURNAL rather than the operator's real run directory.
    /// A run directory that cannot be resolved (a relative data home) refuses
    /// startup instead of landing under the working directory.
    pub fn from_env_for_daemon_binary() -> Result<Self, BootstrapError> {
        let run_dir = daemon_config::daemon_run_dir().map_err(BootstrapError::RunDir)?;
        let machine_id_path =
            crate::machine_id::default_machine_id_path().map_err(BootstrapError::MachineId)?;
        let config = Self::from_env()?
            .with_capture_logs_dir(run_dir.join("logs"))
            .with_terminal_journal_path(run_dir.join("terminals.jsonl"))
            .with_live_children_record(crate::live_children::record_path(&run_dir))
            .with_machine_id_path(machine_id_path);
        #[cfg(feature = "test-support")]
        let config = {
            let mut config = config;
            config.operator_script =
                std::env::var_os("SUBC_TEST_OPERATOR_SCRIPT").map(PathBuf::from);
            config.operator_events =
                std::env::var_os("SUBC_TEST_OPERATOR_EVENTS").map(PathBuf::from);
            for (name, value) in [
                (
                    "SUBC_TEST_OPERATOR_QUEUE_WAIT_MS",
                    &mut config.operator_queue_wait,
                ),
                (
                    "SUBC_TEST_OPERATOR_TIMEOUT_MS",
                    &mut config.operator_timeout,
                ),
                (
                    "SUBC_TEST_OPERATOR_STUCK_MS",
                    &mut config.operator_stuck_grace,
                ),
            ] {
                if let Ok(ms) = std::env::var(name) {
                    if let Ok(ms) = ms.parse::<u64>() {
                        *value = Duration::from_millis(ms);
                    }
                }
            }
            config
        };
        Ok(config)
    }

    pub fn from_env_with_daemon_config_path(
        daemon_config_path: impl AsRef<Path>,
    ) -> Result<Self, BootstrapError> {
        let daemon_config_path = daemon_config_path.as_ref().to_path_buf();
        let (daemon_config, activity_config) =
            daemon_config::load_with_activity(&daemon_config_path)
                .map_err(BootstrapError::DaemonConfig)?
                .map(|(config, activity)| (Some(config), activity))
                .unwrap_or_default();
        let config_port = daemon_config.as_ref().and_then(|config| config.port);
        let storage_config = daemon_config
            .as_ref()
            .and_then(|config| config.storage.clone());
        let admission_facts_carrier_module_id = daemon_config
            .as_ref()
            .and_then(|config| config.admission_facts_carrier_module_id.clone());
        let admission_facts_targets = daemon_config
            .as_ref()
            .and_then(|config| config.admission_facts_targets.clone());
        let scope_authority_owners = daemon_config
            .as_ref()
            .map(|config| config.scope_authority_owners.clone())
            .unwrap_or_else(daemon_config::default_scope_authority_owners);
        let route_bind_relay_default_ms = daemon_config
            .as_ref()
            .and_then(|config| config.route_bind_relay_timeout_ms);
        let reserved_capabilities = daemon_config
            .as_ref()
            .map(|config| config.reserved_capabilities.clone())
            .unwrap_or_default();
        let configured_modules = daemon_config
            .map(|config| config.modules)
            .unwrap_or_default();

        let port = match env::var(SUBC_PORT_ENV) {
            Ok(raw) if !raw.trim().is_empty() => {
                let port = raw
                    .parse::<u16>()
                    .map_err(|source| BootstrapError::InvalidPort { raw, source })?;
                if let Some(config_port) = config_port {
                    info!(
                        env = SUBC_PORT_ENV,
                        env_port = port,
                        config_port,
                        "SUBC_PORT overrides daemon config port"
                    );
                }
                port
            }
            Ok(_) | Err(_) => config_port.unwrap_or(DEFAULT_SUBC_PORT),
        };

        let (connection_file_path, connection_file_source) =
            connection_file_path_with_source(non_empty_os_var("XDG_RUNTIME_DIR"));
        Ok(Self::new(connection_file_path, port)
            .with_configured_modules(configured_modules)
            .with_storage_config(storage_config)
            .with_admission_facts_config(admission_facts_carrier_module_id, admission_facts_targets)
            .with_scope_authority_owners(scope_authority_owners)
            .with_activity_config(activity_config)
            .with_route_bind_relay_default_ms(route_bind_relay_default_ms)
            .with_reserved_capabilities(reserved_capabilities)
            .with_daemon_config_source(daemon_config_path, config_port)
            .with_connection_file_source(connection_file_source))
    }

    pub fn with_daemon_config_path(
        self,
        daemon_config_path: impl AsRef<Path>,
    ) -> Result<Self, BootstrapError> {
        let daemon_config_path = daemon_config_path.as_ref().to_path_buf();
        let (daemon_config, activity_config) =
            daemon_config::load_with_activity(&daemon_config_path)
                .map_err(BootstrapError::DaemonConfig)?
                .map(|(config, activity)| (Some(config), activity))
                .unwrap_or_default();
        let configured_port = daemon_config.as_ref().and_then(|config| config.port);
        let storage_config = daemon_config
            .as_ref()
            .and_then(|config| config.storage.clone());
        let admission_facts_carrier_module_id = daemon_config
            .as_ref()
            .and_then(|config| config.admission_facts_carrier_module_id.clone());
        let admission_facts_targets = daemon_config
            .as_ref()
            .and_then(|config| config.admission_facts_targets.clone());
        let scope_authority_owners = daemon_config
            .as_ref()
            .map(|config| config.scope_authority_owners.clone())
            .unwrap_or_else(daemon_config::default_scope_authority_owners);
        let route_bind_relay_default_ms = daemon_config
            .as_ref()
            .and_then(|config| config.route_bind_relay_timeout_ms);
        let reserved_capabilities = daemon_config
            .as_ref()
            .map(|config| config.reserved_capabilities.clone())
            .unwrap_or_default();
        let configured_modules = daemon_config
            .map(|config| config.modules)
            .unwrap_or_default();
        Ok(self
            .with_configured_modules(configured_modules)
            .with_storage_config(storage_config)
            .with_admission_facts_config(admission_facts_carrier_module_id, admission_facts_targets)
            .with_scope_authority_owners(scope_authority_owners)
            .with_activity_config(activity_config)
            .with_route_bind_relay_default_ms(route_bind_relay_default_ms)
            .with_reserved_capabilities(reserved_capabilities)
            .with_daemon_config_source(daemon_config_path, configured_port))
    }

    pub fn with_configured_modules(
        mut self,
        modules: impl IntoIterator<Item = ConfiguredModule>,
    ) -> Self {
        self.configured_modules = modules.into_iter().collect();
        self.configured_modules
            .sort_by(|left, right| left.module_id.cmp(&right.module_id));
        self
    }

    pub fn with_storage_config(
        mut self,
        storage_config: Option<daemon_config::StorageConfig>,
    ) -> Self {
        self.storage_config = storage_config;
        self
    }

    pub fn with_admission_facts_config(
        mut self,
        carrier_module_id: Option<String>,
        targets: Option<Vec<String>>,
    ) -> Self {
        self.admission_facts = AdmissionFactsConfig {
            carrier_module_id,
            targets,
        };
        self
    }

    /// Set the module ids whose scopes may carry `agent_id` and `delegates`.
    /// Absent from a config, this is `daemon_config::default_scope_authority_owners`.
    pub fn with_scope_authority_owners(mut self, owners: Vec<String>) -> Self {
        self.scope_authority_owners = owners;
        self
    }

    pub fn with_activity_config(mut self, config: daemon_config::ActivityConfig) -> Self {
        self.activity_config = config;
        self
    }

    /// Set the daemon-wide route.bind relay default (the fallback for any
    /// module without a per-module override). `None` preserves the built-in
    /// default (12s). `serve_bound_daemon` reads this at startup and threads
    /// it into the control handler's daemon-wide field.
    pub fn with_route_bind_relay_default_ms(mut self, ms: Option<u64>) -> Self {
        self.route_bind_relay_default_ms = ms;
        self
    }

    pub fn with_reserved_capabilities(
        mut self,
        reserved_capabilities: BTreeMap<String, String>,
    ) -> Self {
        self.reserved_capabilities = reserved_capabilities;
        self
    }

    fn with_daemon_config_source(
        mut self,
        daemon_config_path: PathBuf,
        configured_port: Option<u16>,
    ) -> Self {
        self.daemon_config_path = Some(daemon_config_path);
        self.configured_port = configured_port;
        self
    }

    fn with_connection_file_source(mut self, source: ConnectionFileSource) -> Self {
        self.connection_file_source = source;
        self
    }

    pub fn with_watchdog_config(mut self, watchdog_config: DaemonSelfWatchdogConfig) -> Self {
        self.watchdog_config = watchdog_config;
        self
    }
}

/// Result of singleton discovery.
// Built once per daemon start and immediately matched, so the size gap between
// the variants costs nothing worth a box on the public shape.
#[allow(clippy::large_enum_variant)]
#[derive(Debug)]
pub enum Outcome {
    /// A live daemon authenticated from the connection file; this invocation should exit 0.
    AlreadyRunning,
    /// This process won the singleton race, owns bound loopback listener(s), and
    /// has published a fresh connection file.
    Bound(BoundDaemon),
}

#[derive(Debug)]
pub struct BoundDaemon {
    pub listeners: Vec<TcpListener>,
    pub connection_info: ConnectionInfo,
    pub connection_file_path: PathBuf,
    pub connection_file_source: ConnectionFileSource,
    /// The machine id this daemon serves, established before any connection is
    /// accepted. `None` when the config named no machine id path.
    pub machine_id: Option<crate::machine_id::MachineId>,
    /// Ownership of the run directory, held until the daemon exits so no other
    /// daemon sweeps or writes this one's run state. `None` when the config
    /// keeps no live-children record, and so has no run directory to own.
    run_dir_lock: Option<crate::run_dir_lock::RunDirLock>,
    /// The singleton start lock, kept when binding deferred publishing the
    /// connection file. It stays held until the daemon has verified it is serving
    /// and written the connection file, so a second daemon starting meanwhile
    /// cannot claim the singleton too. `None` when the connection
    /// file was already written at bind time.
    publication_lock: Option<StartLock>,
}

/// Resolve subc's per-user TCP connection-file path.
///
/// `$XDG_RUNTIME_DIR/subc-connection.json` is preferred because the runtime
/// directory is already per-user on Unix desktops. Without it, subc falls back
/// to the system temp dir with a per-user token in the filename so different OS
/// users do not collide on shared temp directories.
pub fn connection_file_path() -> PathBuf {
    connection_file_path_with_source(non_empty_os_var("XDG_RUNTIME_DIR")).0
}

fn connection_file_path_with_source(
    runtime_dir: Option<OsString>,
) -> (PathBuf, ConnectionFileSource) {
    if let Some(runtime_dir) = runtime_dir.filter(|value| !value.is_empty()) {
        return (
            PathBuf::from(runtime_dir).join(CONNECTION_FILE_NAME),
            ConnectionFileSource::XdgRuntimeDir,
        );
    }

    (
        env::temp_dir().join(format!("subc-{}.connection.json", user_connection_token())),
        ConnectionFileSource::TempDirFallback,
    )
}

/// Resolve, claim, and serve the per-user daemon singleton.
///
/// A second invocation is successful: if a live daemon authenticates from the
/// existing connection file, this returns `Ok(())` after logging and the caller
/// exits with status 0.
pub async fn run() -> Result<(), BootstrapError> {
    // `run` is a binary entry point, so it opts into the production cgroup and
    // child-log locations. In-process callers keep both features disabled by default.
    run_with_config(
        BootstrapConfig::from_env_for_daemon_binary()?
            .with_cgroup_placement(CgroupPlacementConfig::Current),
    )
    .await
}

/// Serve a daemon from an explicit config. This is the entry point the twelve
/// sibling repos use to boot an in-process daemon in their integration tests.
///
/// # THIS INSTALLS NO TRACING SUBSCRIBER, SO THE DAEMON IS SILENT BY DEFAULT
///
/// The daemon's own diagnostics go through `tracing`, and `tracing` DISCARDS
/// every event when no subscriber is installed. The shipped binary installs one
/// in `main` (`init_tracing`); this function deliberately does not, because a
/// library that installs a global subscriber fights with whatever the host
/// process already set up.
///
/// The consequence for a test harness is not "less verbose": it is that the
/// daemon has NOTHING TO SAY about any failure, and a missing instrument reads
/// exactly like a clean one. PLEX found this on 2026-09-18 while trying to
/// capture daemon logs beside an intermittent bind failure, and discovered the
/// daemon had been silent in every conformance run that repo had ever done --
/// so the one client-side error string was all the evidence that could exist,
/// and they had spent a real investigation on a failure whose second source was
/// never being recorded.
///
/// Install one in the harness before calling this, and assert it did something
/// (they measured 236 daemon lines with the subscriber installed, 0 with the
/// call commented out) -- otherwise the fix is itself unverified.
pub async fn run_with_config(config: BootstrapConfig) -> Result<(), BootstrapError> {
    let forwarding = Arc::new(ForwardingTable::default());
    forwarding.operator_confirms().configure_bootstrap(&config);
    let configured_modules = config.configured_modules.clone();
    let storage_config = config.storage_config.clone();
    let admission_facts = config.admission_facts.clone();
    let scope_authority_owners = config.scope_authority_owners.clone();
    let activity_config = config.activity_config;
    let daemon_config_path = config.daemon_config_path.clone();
    let configured_port = config.configured_port;
    let route_bind_relay_default_ms = config.route_bind_relay_default_ms;
    let reserved_capabilities = config.reserved_capabilities.clone();
    let watchdog_config = config.watchdog_config.clone();
    let cgroup_placement_config = config.cgroup_placement.clone();
    let capture_logs_dir = config.capture_logs_dir.clone();
    let terminal_journal_path = config.terminal_journal_path.clone();
    let live_children_path = config.live_children_path.clone();
    let privacy_trampoline = config.privacy_trampoline.clone();
    match ensure_singleton_inner(config, false).await? {
        Outcome::AlreadyRunning => {
            info!("subc daemon already running");
            Ok(())
        }
        Outcome::Bound(bound) => {
            #[cfg(target_os = "linux")]
            let cgroup_placement = prepare_cgroup_placement(&cgroup_placement_config);
            #[cfg(not(target_os = "linux"))]
            let _ = cgroup_placement_config;
            serve_bound_daemon(
                bound,
                forwarding,
                configured_modules,
                storage_config,
                admission_facts,
                scope_authority_owners,
                activity_config,
                daemon_config_path,
                configured_port,
                route_bind_relay_default_ms,
                reserved_capabilities,
                watchdog_config,
                capture_logs_dir,
                terminal_journal_path,
                live_children_path,
                privacy_trampoline,
                #[cfg(target_os = "linux")]
                cgroup_placement,
            )
            .await
        }
    }
}

#[cfg(target_os = "linux")]
fn prepare_cgroup_placement(config: &CgroupPlacementConfig) -> Option<subc_cgroup::Placement> {
    let result = match config {
        CgroupPlacementConfig::Disabled => return None,
        CgroupPlacementConfig::Current => subc_cgroup::prepare_current(),
        CgroupPlacementConfig::Root(root) => subc_cgroup::prepare_at(root),
    };

    match result {
        Ok(Some(placement)) => Some(placement),
        Ok(None) => {
            warn!(
                placement = ?config,
                "module cgroup placement is disabled: configured cgroup root is not delegated"
            );
            None
        }
        Err(error) => {
            warn!(
                placement = ?config,
                error = %error,
                "module cgroup placement is disabled by an unexpected cgroup probe error"
            );
            None
        }
    }
}

/// Target soft limit for open file descriptors, applied to the daemon before any
/// module is spawned so children inherit it. Multi-root modules (one process
/// aggregating every project root's sqlite stores, index caches, watchers, and
/// LSP pipes) trivially exceed the macOS default soft limit of 256; a launchd
/// user agent does not pass login-shell ulimits through, so the raise must
/// happen in-process.
#[cfg(unix)]
const NOFILE_TARGET: u64 = 65536;

/// Raise RLIMIT_NOFILE to `NOFILE_TARGET` (clamped to the hard limit).
/// Best-effort: failure is logged and never fatal, since the daemon can run
/// under the inherited limit — modules with few roots just have less headroom.
#[cfg(unix)]
fn raise_nofile_limit() {
    #[cfg(target_os = "macos")]
    let ceiling = std::process::Command::new("/usr/sbin/sysctl")
        .args(["-n", "kern.maxfilesperproc"])
        .output()
        .ok()
        .filter(|output| output.status.success())
        .and_then(|output| String::from_utf8(output.stdout).ok())
        .and_then(|value| value.trim().parse::<u64>().ok())
        .filter(|value| *value > 0);
    #[cfg(not(target_os = "macos"))]
    let ceiling = None;
    raise_nofile_limit_with_ceiling(ceiling);
}

#[cfg(unix)]
fn raise_nofile_limit_with_ceiling(kernel_ceiling: Option<u64>) {
    match rlimit::Resource::NOFILE.get() {
        Ok((soft, hard)) => {
            // On XNU an infinite hard limit does not remove the separate
            // per-process kernel ceiling. Exceeding it makes setrlimit fail
            // without raising the inherited (often 256) soft limit at all.
            let target = NOFILE_TARGET
                .min(hard)
                .min(kernel_ceiling.unwrap_or(u64::MAX));
            if soft >= target {
                return;
            }
            match rlimit::Resource::NOFILE.set(target, hard) {
                Ok(()) => info!(
                    previous_soft = soft,
                    new_soft = target,
                    hard,
                    "raised open-file soft limit for daemon and module children"
                ),
                Err(err) => warn!(
                    soft,
                    hard,
                    error = %err,
                    "could not raise open-file soft limit; multi-root modules may exhaust descriptors"
                ),
            }
        }
        Err(err) => warn!(error = %err, "could not read open-file limit"),
    }
}

/// CRT stdio-stream target on Windows (the `_setmaxstdio` maximum). Win32
/// HANDLEs — what Rust `File`, tokio sockets, and SQLite's Win32 VFS actually
/// consume — have a per-process quota in the millions and need no raise; the
/// C-runtime stream table (default 512) is the only low ceiling, and it is
/// per-process rather than inherited, so supervised modules linking the CRT
/// must raise their own. Raising it here covers the daemon itself.
#[cfg(windows)]
fn raise_nofile_limit() {
    const MAXSTDIO_TARGET: u32 = 8192;
    let current = rlimit::getmaxstdio();
    if current >= MAXSTDIO_TARGET {
        return;
    }
    match rlimit::setmaxstdio(MAXSTDIO_TARGET) {
        Ok(new_max) => info!(
            previous = current,
            new_max, "raised CRT stdio-stream limit for daemon"
        ),
        Err(err) => warn!(
            current,
            error = %err,
            "could not raise CRT stdio-stream limit"
        ),
    }
}

#[cfg(not(any(unix, windows)))]
fn raise_nofile_limit() {}

pub async fn run_with_daemon_config_path(
    config: BootstrapConfig,
    daemon_config_path: impl AsRef<Path>,
) -> Result<(), BootstrapError> {
    run_with_config(config.with_daemon_config_path(daemon_config_path)?).await
}

#[allow(clippy::too_many_arguments)]
async fn serve_bound_daemon(
    bound: BoundDaemon,
    forwarding: Arc<ForwardingTable>,
    configured_modules: Vec<ConfiguredModule>,
    storage_config: Option<daemon_config::StorageConfig>,
    admission_facts: AdmissionFactsConfig,
    scope_authority_owners: Vec<String>,
    activity_config: daemon_config::ActivityConfig,
    daemon_config_path: Option<PathBuf>,
    configured_port: Option<u16>,
    route_bind_relay_default_ms: Option<u64>,
    reserved_capabilities: BTreeMap<String, String>,
    watchdog_config: DaemonSelfWatchdogConfig,
    capture_logs_dir: Option<PathBuf>,
    terminal_journal_path: Option<PathBuf>,
    live_children_path: Option<PathBuf>,
    privacy_trampoline: Option<PathBuf>,
    #[cfg(target_os = "linux")] cgroup_placement: Option<subc_cgroup::Placement>,
) -> Result<(), BootstrapError> {
    #[cfg(unix)]
    let mut terminate = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())
        .map_err(BootstrapError::Signal)?;
    // Windows has no SIGTERM. Ctrl-C is a different event, not an equivalent
    // service-stop contract, so this shutdown handler is intentionally Unix-only.
    raise_nofile_limit();

    info!(
        connection_file = %bound.connection_file_path.display(),
        connection_file_source = %bound.connection_file_source,
        connection_file_source_reason = bound.connection_file_source.reason(),
        endpoints = ?bound.connection_info.endpoints,
        configured_modules = configured_modules.len(),
        machine_id = bound.machine_id.as_ref().map(|id| id.as_str()).unwrap_or("none"),
        "subc daemon starting"
    );

    // Before anything can spawn a module (the configured modules below, or a
    // client's start request once the listeners are served): a previous daemon
    // that died without its shutdown stop may have left children running, and
    // a fresh copy beside one would fight it for its port and stores. The
    // record is not a running daemon's because this daemon holds the lock on
    // the run directory the record lives in, and a running daemon holds that
    // lock for its whole life. (Claiming the singleton is not enough: it is
    // keyed on the connection file, which can live in a different runtime
    // directory from the run directory.) The lock stays held until this
    // function returns, which is when the daemon exits.
    let run_dir_lock = bound.run_dir_lock;
    let publication_lock = bound.publication_lock;
    if let Some(owner) = &run_dir_lock {
        crate::live_children::sweep_orphans(
            owner,
            &crate::live_children::AdoptedPids::none(),
            crate::live_children::SweepBounds::default(),
        )
        .await;
    }

    let registry = Arc::new(Registry::default());
    let process_liveness = Arc::new(SupervisorProcessLiveness::new());
    let supervisor_handle = SupervisorHandle::new();
    let connected_clients = ConnectedClients::new();
    let daemon_incarnation = format!(
        "{:032x}",
        u128::from_be_bytes(bound.connection_info.daemon_id)
    );
    let supervisor = Supervisor::new(Arc::clone(&registry), RestartPolicy::default())
        .with_process_liveness(process_liveness.clone())
        .with_forwarding(Arc::clone(&forwarding))
        .with_handle(supervisor_handle.clone())
        .with_connection_file_path(bound.connection_file_path.clone())
        .with_daemon_incarnation(daemon_incarnation.clone());
    let supervisor = match privacy_trampoline {
        Some(path) => supervisor.with_privacy_trampoline(path),
        None => supervisor,
    };
    // ABSENT MEANS NO CAPTURE AND NO JOURNAL, NOT "THE REAL RUN DIRECTORY", and
    // the difference is a production-corruption hazard rather than a preference.
    // Both fields below follow the same rule: `None` for `capture_logs_dir`
    // means supervised output is not captured, and `None` for
    // `terminal_journal_path` means terminal history lives only in each
    // module's in-memory ring (`supervisor.terminals` still answers, with the
    // journal counters at zero).
    //
    // The terminal journal used to fall back to
    // `daemon_run_dir().join("terminals.jsonl")`, so an in-process test daemon
    // appended its fixture exits to the operator's journal, where they then
    // appeared in `ck module terminals`.
    //
    // The capture line used to be `unwrap_or_else(|| daemon_run_dir().join("logs"))`,
    // so ANY caller that did not set the field captured supervised children into
    // the operator's live `~/.local/share/cortexkit/run/logs/`. That is twelve
    // sibling repos whose integration tests boot an in-process daemon through
    // `run_with_config` -- none of which asked for it, and none of which can see
    // it from their side.
    //
    // Harmless while fixture module ids are fixture-shaped: this host carries 20
    // zero-byte files from subc's own tests (good-aft, missing-aft,
    // preview-consumer...). THE HAZARD IS A COLLISION. A fixture named "broca"
    // or "aft" appends to a PRODUCTION capture file that operators read
    // forensically and that placement gates count lines in -- with no residue to
    // notice, because the file legitimately exists and legitimately grows.
    //
    // Found by BROCA (2026-09-19) from the other side: their rigs spawn the
    // SHIPPED ck-subc and set XDG_CONFIG_HOME + XDG_RUNTIME_DIR but not
    // XDG_DATA_HOME, so every local rig run supervised a module named "broca"
    // and captured it into production's broca.stderr.log -- the same file I
    // count seal lines in before and after placing their binaries.
    //
    // The supervisor already treats `None` as no-capture and no-journal, so
    // this only removes invented defaults. The binary keeps capturing and
    // journaling via `BootstrapConfig::from_env_for_daemon_binary`.

    // Keep both ledgers in the explicitly selected run directory. An embedded
    // daemon without a journal path must not write to the operator's live ledger.
    let activity_run_dir = terminal_journal_path
        .as_ref()
        .and_then(|path| path.parent())
        .map(Path::to_path_buf);
    let supervisor = match terminal_journal_path {
        Some(path) => supervisor.with_terminal_journal(path, daemon_incarnation.clone()),
        None => supervisor,
    };
    let supervisor = match capture_logs_dir {
        Some(dir) => supervisor.with_capture_logs_dir(dir),
        None => supervisor,
    };
    let supervisor = match live_children_path {
        Some(path) => supervisor.with_live_children_record(path),
        None => supervisor,
    };
    #[cfg(target_os = "linux")]
    let supervisor = supervisor.with_cgroup_placement(cgroup_placement);
    // Collect per-module route.bind relay overrides BEFORE handing the
    // `configured_modules` vector to the supervisor (which only needs each
    // module's `drain_timeout_ms`). Each entry was filled in by parse-time
    // resolution (per-module > daemon-wide > absent), so modules with no
    // override are absent from this map and the daemon-wide default applies.
    let route_bind_relay_timeouts = configured_modules
        .iter()
        .filter_map(|module| {
            module
                .route_bind_relay_timeout_ms
                .map(|ms| (module.module_id.clone(), Duration::from_millis(ms)))
        })
        .collect::<std::collections::BTreeMap<_, _>>();
    let control_start_clock = crate::clock::StartClock::capture();
    let mut control = ControlHandler::with_forwarding(Arc::clone(&registry), forwarding)
        .with_process_liveness(process_liveness)
        .with_supervisor(supervisor_handle)
        .with_connected_clients(connected_clients.clone())
        .with_storage_config(storage_config)
        .with_machine_id(bound.machine_id.clone())
        .with_admission_facts_config(admission_facts.carrier_module_id, admission_facts.targets)
        .with_scope_authority_owners(scope_authority_owners)
        .with_route_bind_relay_timeouts(route_bind_relay_timeouts)
        .with_daemon_provenance(
            bound.connection_info.pid,
            control_start_clock.started_at_ms(),
            std::env::current_exe().ok(),
            normalized_build_provenance(env!("SUBC_BUILD_GIT_SHA")),
            normalized_build_provenance(env!("SUBC_BUILD_LOCK_DIGEST")),
        )
        .with_daemon_start_clock(control_start_clock)
        .with_capability_config(
            configured_modules
                .iter()
                .map(|module| (module.module_id.clone(), module.enabled)),
            reserved_capabilities,
        );
    if let Some(run_dir) = activity_run_dir {
        control =
            control.with_activity_ledger_config(&run_dir, daemon_incarnation, activity_config);
    }
    if let Some(ms) = route_bind_relay_default_ms {
        // A daemon-wide config value overrides the built-in default; a
        // `None` here leaves the ControlHandler's 12s default in place.
        control = control.with_route_bind_relay_timeout(Duration::from_millis(ms));
    }
    if let Some(config_path) = daemon_config_path {
        control = control.with_supervisor_rescan(supervisor.clone(), config_path, configured_port);
    }
    let control = Arc::new(control);
    let router = Arc::new(Router::with_control_handler(Arc::clone(&control)));
    let auth = ServerAuth::new(
        bound.connection_info.key.clone(),
        bound.connection_info.daemon_id,
        bound.connection_info.daemon_ver.clone(),
    )
    .with_connected_clients(connected_clients);

    let mut serve_task =
        AbortOnDrop::new(tokio::spawn(serve_listeners(bound.listeners, router, auth)));
    verify_serving(&bound.connection_info).await?;
    // A published endpoint promises that authentication can complete now.
    // Orphan cleanup and control-plane construction can take longer than an
    // existing-daemon probe's deadline, so finish both before discovery.
    // The write fsyncs, so it runs on tokio's blocking pool (see
    // `publish_connection_file`). The singleton lock travels with it and comes
    // back only once the file is in place.
    let (published, publication_lock) = publish_connection_file(
        &bound.connection_file_path,
        &bound.connection_info,
        publication_lock,
    )
    .await;
    published?;
    drop(publication_lock);
    let _clock_step_task = AbortOnDrop::new(crate::watchdog::spawn_clock_step_monitor());
    // Off the startup path: it runs `systemctl`, and only ever logs.
    #[cfg(target_os = "linux")]
    let _kill_mode_check = AbortOnDrop::new(tokio::spawn(
        crate::systemd_kill_mode::warn_if_kill_mode_defeats_ordered_shutdown(),
    ));
    let _watchdog_task = AbortOnDrop::new(
        DaemonSelfWatchdog::new(
            bound.connection_info.clone(),
            bound.connection_file_path.clone(),
        )
        .with_config(watchdog_config)
        .spawn(),
    );

    for configured in configured_modules {
        let enabled = configured.enabled;
        let health = configured.health.clone();
        let module_id = configured.module_id.clone();
        match supervisor.supervise_configured_with_health(
            configured.module_spec(),
            enabled,
            health.clone(),
            configured.drain_timeout_ms,
            configured.restart,
        ) {
            Ok(_) => {
                // A raised failure threshold is normally a temporary allowance for a
                // drive that deliberately stops a module, and it widens the window in
                // which a genuinely wedged module looks fine. It is only ever noticed
                // when someone thinks to re-read the config, so a relaxation outlives
                // its reason silently: a rig ran five days at 240s of tolerance against
                // a 90s default because a comment promising a revert was mistaken for
                // the revert. Saying so on every boot costs one line and removes the
                // need for anyone to remember.
                let default_threshold = HealthConfig::default().failure_threshold;
                if enabled && health.failure_threshold > default_threshold {
                    warn!(
                        module_id = %module_id,
                        failure_threshold = health.failure_threshold,
                        default_threshold,
                        tolerance_secs = health.cadence.as_secs() * u64::from(health.failure_threshold),
                        "health failure threshold is relaxed above the default; a wedged module stays unflagged for longer"
                    );
                }
                info!(module_id = %module_id, enabled, "configured module supervised");
            }
            Err(err) => {
                error!(module_id = %module_id, error = %err, "failed to supervise configured module; continuing daemon startup");
            }
        }
    }

    control.refresh_capability_requirements();
    Arc::clone(&control).spawn_capability_deadline_loop();
    Arc::clone(&control).spawn_scope_expiry_loop();

    #[cfg(unix)]
    {
        tokio::select! {
            result = serve_task.join() => {
                return result.map_err(BootstrapError::ServeJoin)?.map_err(BootstrapError::Serve);
            }
            _ = terminate.recv() => {}
        }
        // First, before the notice, the drain, or any connection close: from
        // here on a module exit is recorded as `daemon_shutdown` and never
        // respawned. Also before allowing a second signal to cut the bounded
        // wait short, so the journal marker is always written.
        supervisor.begin_daemon_shutdown();
        // Stop the self-watchdog before closing the listener. Its next tick would
        // connect to that listener, fail, and log an ERROR indistinguishable from
        // a wedged daemon, once for every interval a planned stop lasts.
        drop(_watchdog_task);
        // Dropping the listener stops new accepts, not established connections:
        // their detached tasks must remain live throughout notice and drain.
        drop(serve_task);
        let escalated = tokio::select! {
            biased;
            _ = terminate.recv() => {
                info!("second SIGTERM: abandoning daemon shutdown wait");
                true
            }
            result = supervisor.drain_for_daemon_shutdown() => {
                if let Err(error) = result {
                    warn!(%error, "daemon shutdown drain failed; exiting anyway");
                }
                false
            }
        };
        // Supervised modules lead their own process groups, so a service
        // manager's kill of this process's group does not reach them. (A
        // systemd unit with KillMode=control-group kills by cgroup instead and
        // does reach them; see `systemd_kill_mode`.) The daemon ends them
        // itself: EOF (or SIGTERM for a protocol none child) first, then
        // signals at each child's own drain deadline.
        supervisor
            .end_children_for_daemon_shutdown(escalated, async {
                terminate.recv().await;
            })
            .await;
        Ok(())
    }
    #[cfg(not(unix))]
    serve_task
        .join()
        .await
        .map_err(BootstrapError::ServeJoin)?
        .map_err(BootstrapError::Serve)
}

fn normalized_build_provenance(value: &str) -> Option<String> {
    match value.trim() {
        "" | "unavailable" => None,
        value => Some(value.to_string()),
    }
}

/// Find an existing daemon or atomically bind loopback TCP for this daemon.
///
/// The algorithm is intentionally connect-first: an endpoint from the connection
/// file is treated as live only after the TCP+key server-proof authenticates for
/// that file's key and daemon_id. Stale or foreign connection files are reclaimed
/// only while holding the per-user start lock; the TCP port is never the
/// singleton primitive.
pub async fn ensure_singleton(
    connection_file_path: impl AsRef<Path>,
    port: u16,
) -> Result<Outcome, BootstrapError> {
    ensure_singleton_with_config(BootstrapConfig::new(connection_file_path.as_ref(), port)).await
}

pub async fn ensure_singleton_with_config(
    config: BootstrapConfig,
) -> Result<Outcome, BootstrapError> {
    ensure_singleton_inner(config, true).await
}

async fn ensure_singleton_inner(
    config: BootstrapConfig,
    publish: bool,
) -> Result<Outcome, BootstrapError> {
    let path = config.connection_file_path;

    if matches!(probe_existing(&path).await?, Probe::Live) {
        return Ok(Outcome::AlreadyRunning);
    }

    let lock = StartLock::acquire(&path).await?;

    // Re-probe after acquiring the start lock so a peer that won the race between
    // our first failed probe and the lock acquisition is observed instead of
    // overwritten.
    if matches!(probe_existing(&path).await?, Probe::Live) {
        return Ok(Outcome::AlreadyRunning);
    }

    // The start lock and the probe above only rule out a daemon using this
    // connection file. A daemon started with another runtime directory but
    // the same data home shares the run directory while passing both, so the
    // run directory needs its own owner: refuse at once if a live daemon holds
    // it, before anything below reads or writes run or data-home state.
    let run_dir_lock = config
        .live_children_path
        .as_deref()
        .map(crate::run_dir_lock::RunDirLock::acquire)
        .transpose()?;

    remove_stale_connection_file_if_present(&path)?;

    // Established under the start lock, and under the run-directory lock when
    // there is one, and before binding: a corrupt file stops boot before
    // anything is published, and two daemons racing to start cannot both mint.
    //
    // Minting a new machine id fsyncs its file, so loading or minting runs on
    // tokio's blocking pool. Both locks travel with that work, so neither is
    // released while a mint is in flight.
    let (machine_id, lock, run_dir_lock) = match config.machine_id_path {
        None => (None, lock, run_dir_lock),
        Some(machine_id_path) => {
            let task_path = machine_id_path.clone();
            match run_blocking_holding((lock, run_dir_lock), move || {
                crate::machine_id::load_or_mint(&task_path)
            })
            .await
            {
                Ok((Ok(id), (lock, run_dir_lock))) => (Some(id), lock, run_dir_lock),
                Ok((Err(error), _locks)) => return Err(BootstrapError::MachineId(error)),
                Err(join_error) => {
                    return Err(BootstrapError::MachineId(
                        crate::machine_id::MachineIdFileError::Write {
                            path: machine_id_path,
                            source: io::Error::other(describe_blocking_task_failure(
                                "loading or minting the machine id",
                                &join_error,
                            )),
                        },
                    ))
                }
            }
        }
    };

    let (listeners, endpoints) = bind_loopback(config.port).await?;
    let connection_info = ConnectionInfo {
        schema: SCHEMA_VERSION,
        wire_version: Some(PROTOCOL_VERSION),
        endpoints,
        key: generate_key().map_err(BootstrapError::GenerateConnectionFile)?,
        daemon_id: generate_daemon_id().map_err(BootstrapError::GenerateConnectionFile)?,
        pid: process::id(),
        daemon_ver: config.daemon_ver,
    };

    // The low-level binding API historically publishes for callers that serve
    // these listeners themselves. The daemon entry point defers publication
    // until its own authentication server is running.
    let publication_lock = if publish {
        let (published, lock) = publish_connection_file(&path, &connection_info, lock).await;
        if let Err(error) = published {
            drop(listeners);
            return Err(error);
        }
        drop(lock);
        None
    } else {
        Some(lock)
    };

    Ok(Outcome::Bound(BoundDaemon {
        listeners,
        connection_info,
        connection_file_path: path,
        connection_file_source: config.connection_file_source,
        machine_id,
        run_dir_lock,
        publication_lock,
    }))
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Probe {
    Live,
    StaleOrAbsent,
}

async fn probe_existing(path: &Path) -> Result<Probe, BootstrapError> {
    let info = match connection_file::read(path) {
        Ok(info) => info,
        Err(source) if is_absent_or_stale_connection_file(&source) => {
            return Ok(Probe::StaleOrAbsent)
        }
        Err(source) => {
            return Err(BootstrapError::ConnectionFileRead {
                path: path.to_path_buf(),
                source,
            })
        }
    };

    for endpoint in &info.endpoints {
        if matches!(probe_endpoint(&info, endpoint).await, Probe::Live) {
            return Ok(Probe::Live);
        }
    }

    Ok(Probe::StaleOrAbsent)
}

async fn probe_endpoint(info: &ConnectionInfo, endpoint: &Endpoint) -> Probe {
    let Ok(ip) = endpoint.host.parse::<IpAddr>() else {
        return Probe::StaleOrAbsent;
    };
    if !ip.is_loopback() {
        return Probe::StaleOrAbsent;
    }
    let addr = SocketAddr::new(ip, endpoint.port);

    let mut stream = match timeout(CONNECT_TIMEOUT, TcpStream::connect(addr)).await {
        Ok(Ok(stream)) => stream,
        Ok(Err(_)) | Err(_) => return Probe::StaleOrAbsent,
    };

    match authenticate_client(&mut stream, info, PROBE_AUTH_DEADLINE).await {
        Ok(()) => Probe::Live,
        Err(AuthError::DaemonIdMismatch)
        | Err(AuthError::InvalidServerProof)
        | Err(AuthError::UnexpectedEof { .. })
        | Err(AuthError::Timeout { .. })
        | Err(AuthError::JsonEncode { .. })
        | Err(AuthError::JsonDecode { .. })
        | Err(AuthError::Io { .. })
        | Err(AuthError::MessageTooLarge { .. })
        | Err(AuthError::KeyTooShort { .. })
        | Err(AuthError::Random(_))
        | Err(AuthError::InvalidClientAuth) => Probe::StaleOrAbsent,
    }
}

async fn verify_serving(info: &ConnectionInfo) -> Result<(), BootstrapError> {
    // Scheduling a server task is not a readiness barrier. Authenticate once
    // against the mandatory IPv4 listener before exposing discovery to peers.
    // This uses the normal probe deadline, with no retries or widened budget.
    if let Some(endpoint) = info.endpoints.first() {
        if matches!(probe_endpoint(info, endpoint).await, Probe::Live) {
            return Ok(());
        }
    }
    Err(BootstrapError::StartupNotServing)
}

/// Run blocking filesystem work (an fsync, a rename) on tokio's blocking-thread
/// pool rather than on the async worker thread that awaits it. On a slow or
/// busy disk an fsync can take seconds, and a runtime has few worker threads
/// (an in-process test daemon may share two with everything else), so doing it
/// inline stalls unrelated tasks for that long.
///
/// `guards` (lock handles) move into the blocking task and are handed back
/// with the work's result. If the awaiting future is dropped first, the work
/// still runs to completion on its thread and the guards are released only
/// after it ends, so a lock is never given up while the work it protects is
/// still writing.
///
/// `Err` means the task panicked, or was cancelled before it started because
/// the runtime is shutting down; the guards have then been released already.
async fn run_blocking_holding<G, T>(
    guards: G,
    work: impl FnOnce() -> T + Send + 'static,
) -> Result<(T, G), JoinError>
where
    G: Send + 'static,
    T: Send + 'static,
{
    tokio::task::spawn_blocking(move || {
        let output = work();
        (output, guards)
    })
    .await
}

/// Explain, for a startup error, why blocking work produced no result.
fn describe_blocking_task_failure(work: &str, join_error: &JoinError) -> String {
    if join_error.is_panic() {
        format!("the background thread {work} panicked before finishing: {join_error}")
    } else {
        format!(
            "the background thread {work} was cancelled before finishing, because the async runtime is shutting down: {join_error}"
        )
    }
}

/// Publish the connection file: owner-only permissions, written to a temporary
/// file, fsynced and renamed over `path` (all done by `write_atomic`). The
/// write runs on tokio's blocking pool through [`run_blocking_holding`], so the
/// fsync never holds an async worker thread.
///
/// `guard` is the singleton start lock (or whatever the caller must keep held
/// across publication); it is returned with the result. It comes back `None`
/// only when the write task panicked or was cancelled, in which case the
/// result is an error and the file must be treated as not published.
async fn publish_connection_file<G: Send + 'static>(
    path: &Path,
    info: &ConnectionInfo,
    guard: G,
) -> (Result<(), BootstrapError>, Option<G>) {
    let task_path = path.to_path_buf();
    let task_info = info.clone();
    let outcome =
        run_blocking_holding(guard, move || write_connection_file(&task_path, &task_info)).await;
    match outcome {
        Ok((Ok(()), guard)) => (Ok(()), Some(guard)),
        Ok((Err(source), guard)) => (
            Err(BootstrapError::ConnectionFileWrite {
                path: path.to_path_buf(),
                source,
            }),
            Some(guard),
        ),
        Err(join_error) => (
            Err(BootstrapError::ConnectionFileWrite {
                path: path.to_path_buf(),
                source: ConnectionFileError::Io {
                    op: "write",
                    path: path.to_path_buf(),
                    source: io::Error::other(describe_blocking_task_failure(
                        "writing the connection file",
                        &join_error,
                    )),
                },
            }),
            None,
        ),
    }
}

/// The synchronous connection-file write, including its fsync. Call it only
/// through [`publish_connection_file`], which keeps it off the async workers.
fn write_connection_file(path: &Path, info: &ConnectionInfo) -> Result<(), ConnectionFileError> {
    #[cfg(test)]
    connection_file_write_hook::run(path);
    write_atomic(path, info)
}

/// Lets a test run code inside the connection-file write, on whichever thread
/// performs it, for example to block there and check what else can still run.
#[cfg(test)]
mod connection_file_write_hook {
    use std::{
        path::{Path, PathBuf},
        sync::{Mutex, PoisonError},
    };

    type Hook = Box<dyn FnOnce() + Send>;

    /// Keyed by connection-file path, so a hook fires only for the test that
    /// installed it while other tests publish their own files in parallel.
    static HOOKS: Mutex<Vec<(PathBuf, Hook)>> = Mutex::new(Vec::new());

    /// Run `hook` once, at the start of the next write of `path`.
    pub(super) fn install(path: &Path, hook: impl FnOnce() + Send + 'static) {
        HOOKS
            .lock()
            .unwrap_or_else(PoisonError::into_inner)
            .push((path.to_path_buf(), Box::new(hook)));
    }

    pub(super) fn run(path: &Path) {
        let hook = {
            let mut hooks = HOOKS.lock().unwrap_or_else(PoisonError::into_inner);
            hooks
                .iter()
                .position(|(hooked, _)| hooked == path)
                .map(|index| hooks.swap_remove(index).1)
        };
        if let Some(hook) = hook {
            hook();
        }
    }
}

fn is_absent_or_stale_connection_file(err: &ConnectionFileError) -> bool {
    match err {
        ConnectionFileError::Io { source, .. } if source.kind() == io::ErrorKind::NotFound => true,
        ConnectionFileError::JsonRead { .. }
        | ConnectionFileError::UnsupportedSchema { .. }
        | ConnectionFileError::Invalid { .. }
        | ConnectionFileError::KeyTooShort { .. }
        // A live daemon always publishes the file owner-only (0600), so a file
        // with insecure permissions is never a daemon we should defer to: treat it
        // as stale and take over (which republishes a correct 0600 file).
        | ConnectionFileError::InsecurePermissions { .. } => true,
        ConnectionFileError::MissingParent { .. }
        | ConnectionFileError::MissingFileName { .. }
        // A writable ancestor is an operator misconfiguration, never evidence
        // about whether a daemon is live. Reclaiming the file would republish key
        // material into the same directory the refusal is about.
        | ConnectionFileError::InsecureParentDirectory { .. }
        | ConnectionFileError::Io { .. }
        | ConnectionFileError::JsonWrite { .. }
        | ConnectionFileError::Random(_)
        // A wire mismatch may identify a newer live daemon, so never reclaim its
        // connection file merely because this binary cannot speak its envelope.
        | ConnectionFileError::WireVersionMismatch { .. } => false,
    }
}

fn remove_stale_connection_file_if_present(path: &Path) -> Result<(), BootstrapError> {
    match fs::remove_file(path) {
        Ok(()) => Ok(()),
        Err(err) if err.kind() == io::ErrorKind::NotFound => Ok(()),
        Err(source) => Err(BootstrapError::RemoveStale {
            path: path.to_path_buf(),
            source,
        }),
    }
}

async fn bind_loopback(port: u16) -> Result<(Vec<TcpListener>, Vec<Endpoint>), BootstrapError> {
    bind_loopback_with(port, |port| TcpListener::bind((Ipv6Addr::LOCALHOST, port))).await
}

async fn bind_loopback_with<F, Fut>(
    port: u16,
    mut bind_v6: F,
) -> Result<(Vec<TcpListener>, Vec<Endpoint>), BootstrapError>
where
    F: FnMut(u16) -> Fut,
    Fut: std::future::Future<Output = io::Result<TcpListener>>,
{
    let v4_host = Ipv4Addr::LOCALHOST;
    let v4 = TcpListener::bind((v4_host, port))
        .await
        .map_err(|source| BootstrapError::Bind {
            host: v4_host.to_string(),
            port,
            source,
        })?;
    let actual_port = v4
        .local_addr()
        .map_err(|source| BootstrapError::LocalAddr {
            host: v4_host.to_string(),
            source,
        })?
        .port();

    let mut listeners = vec![v4];
    let mut endpoints = vec![Endpoint {
        host: v4_host.to_string(),
        port: actual_port,
    }];

    let v6_host = Ipv6Addr::LOCALHOST;
    match bind_v6(actual_port).await {
        Ok(v6) => {
            listeners.push(v6);
            endpoints.push(Endpoint {
                host: v6_host.to_string(),
                port: actual_port,
            });
        }
        Err(err)
            if ipv6_loopback_unavailable(&err)
                || (port == 0 && err.kind() == io::ErrorKind::AddrInUse) =>
        {
            warn!(
                port = actual_port,
                error = %err,
                "IPv6 loopback unavailable; serving only IPv4 loopback"
            );
        }
        Err(source) => {
            drop(listeners);
            return Err(BootstrapError::Bind {
                host: v6_host.to_string(),
                port: actual_port,
                source,
            });
        }
    }

    Ok((listeners, endpoints))
}

fn ipv6_loopback_unavailable(err: &io::Error) -> bool {
    matches!(
        err.kind(),
        io::ErrorKind::AddrNotAvailable | io::ErrorKind::Unsupported
    ) || matches!(err.raw_os_error(), Some(47) | Some(49) | Some(97))
}

struct AbortOnDrop<T> {
    handle: JoinHandle<T>,
}

impl<T> AbortOnDrop<T> {
    fn new(handle: JoinHandle<T>) -> Self {
        Self { handle }
    }

    async fn join(&mut self) -> Result<T, JoinError> {
        (&mut self.handle).await
    }
}

impl<T> Drop for AbortOnDrop<T> {
    fn drop(&mut self) {
        if !self.handle.is_finished() {
            self.handle.abort();
        }
    }
}

#[derive(Debug)]
struct StartLock {
    // Keep the locked file handle alive for the duration of bootstrap; closing
    // it releases the advisory lock while leaving the stable path in place.
    _file: fs::File,
}

impl StartLock {
    async fn acquire(connection_file_path: &Path) -> Result<Self, BootstrapError> {
        let path = start_lock_path(connection_file_path);
        for _ in 0..START_LOCK_RETRIES {
            let file = match open_owner_only_lock(&path) {
                Ok(file) => file,
                Err(source) => return Err(BootstrapError::StartLockCreate { path, source }),
            };
            match FileExt::try_lock(&file) {
                Ok(()) => return Ok(Self { _file: file }),
                Err(TryLockError::WouldBlock) => sleep(START_LOCK_RETRY_DELAY).await,
                Err(TryLockError::Error(source)) => {
                    return Err(BootstrapError::StartLockCreate { path, source });
                }
            }
        }

        Err(BootstrapError::StartLockBusy {
            path,
            attempts: START_LOCK_RETRIES,
        })
    }
}

pub(crate) fn open_owner_only_lock(path: &Path) -> io::Result<fs::File> {
    let mut options = fs::OpenOptions::new();
    options.read(true).write(true).create(true);
    #[cfg(unix)]
    {
        use std::os::unix::fs::OpenOptionsExt;
        options.mode(0o600);
    }
    options.open(path)
}

fn start_lock_path(connection_file_path: &Path) -> PathBuf {
    let file_name = connection_file_path
        .file_name()
        .map(|name| name.to_string_lossy())
        .unwrap_or_else(|| CONNECTION_FILE_NAME.into());
    let lock_name = format!("{file_name}.start-lock");
    connection_file_path
        .parent()
        .filter(|parent| !parent.as_os_str().is_empty())
        .unwrap_or_else(|| Path::new("."))
        .join(lock_name)
}

fn non_empty_os_var(key: &str) -> Option<OsString> {
    let value = env::var_os(key)?;
    if value.is_empty() {
        None
    } else {
        Some(value)
    }
}

/// Bootstrap-layer errors are deliberately typed so startup never panics for
/// ordinary daemon-discovery races or stale filesystem state.
#[derive(Debug)]
pub enum BootstrapError {
    /// The bound server could not authenticate its own readiness probe.
    StartupNotServing,
    #[cfg(unix)]
    Signal(io::Error),
    InvalidPort {
        raw: String,
        source: std::num::ParseIntError,
    },
    ConnectionFileRead {
        path: PathBuf,
        source: ConnectionFileError,
    },
    ConnectionFileWrite {
        path: PathBuf,
        source: ConnectionFileError,
    },
    GenerateConnectionFile(ConnectionFileError),
    StartLockCreate {
        path: PathBuf,
        source: io::Error,
    },
    StartLockBusy {
        path: PathBuf,
        attempts: usize,
    },
    /// The run-directory lock file could not be created, opened or locked.
    RunDirLockCreate {
        path: PathBuf,
        source: io::Error,
    },
    /// Another live process, almost certainly a daemon started with a
    /// different runtime directory over the same data home, owns the run
    /// directory. The daemon does not start. `holder_pid` is read from the
    /// lock file when possible and is informational only.
    RunDirBusy {
        path: PathBuf,
        holder_pid: Option<u32>,
    },
    RemoveStale {
        path: PathBuf,
        source: io::Error,
    },
    Bind {
        host: String,
        port: u16,
        source: io::Error,
    },
    LocalAddr {
        host: String,
        source: io::Error,
    },
    DaemonConfig(DaemonConfigError),
    /// The machine id could not be established: its file is corrupt, unreadable
    /// or unwritable, or the data home is relative. The daemon does not start.
    MachineId(crate::machine_id::MachineIdFileError),
    /// The daemon run directory could not be resolved because the data home is
    /// relative. The daemon does not start.
    RunDir(daemon_config::DaemonRunDirError),
    Serve(ServerError),
    ServeJoin(tokio::task::JoinError),
}

impl fmt::Display for BootstrapError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::StartupNotServing => write!(f, "bound daemon did not authenticate its readiness probe; connection file was not published"),
            #[cfg(unix)]
            Self::Signal(error) => write!(f, "failed to register SIGTERM handler: {error}"),
            Self::InvalidPort { raw, source } => {
                write!(f, "invalid {SUBC_PORT_ENV} value '{raw}': {source}")
            }
            Self::ConnectionFileRead { path, source } => write!(
                f,
                "failed to read connection file {}: {source}",
                path.display()
            ),
            Self::ConnectionFileWrite { path, source } => write!(
                f,
                "failed to publish connection file {}: {source}",
                path.display()
            ),
            Self::GenerateConnectionFile(err) => {
                write!(f, "failed to generate connection-file auth material: {err}")
            }
            Self::StartLockCreate { path, source } => {
                write!(
                    f,
                    "failed to create start lock {}: {source}",
                    path.display()
                )
            }
            Self::StartLockBusy { path, attempts } => write!(
                f,
                "start lock {} remained busy after {attempts} attempts",
                path.display()
            ),
            Self::RunDirLockCreate { path, source } => write!(
                f,
                "refusing to start: failed to lock run directory via {}: {source}",
                path.display()
            ),
            Self::RunDirBusy { path, holder_pid } => {
                write!(
                    f,
                    "refusing to start: run directory lock {} is held by another daemon",
                    path.display()
                )?;
                if let Some(pid) = holder_pid {
                    write!(f, " (pid {pid})")?;
                }
                write!(
                    f,
                    "; a live daemon owns this data home's run state (typically one started with a different XDG_RUNTIME_DIR)"
                )
            }
            Self::RemoveStale { path, source } => write!(
                f,
                "failed to remove stale connection file {}: {source}",
                path.display()
            ),
            Self::Bind { host, port, source } if source.kind() == io::ErrorKind::AddrInUse => {
                write!(
                    f,
                    "port {port} in use on loopback {host}: {source}; set the port in config"
                )
            }
            Self::Bind { host, port, source } => {
                write!(f, "failed to bind loopback TCP {host}:{port}: {source}")
            }
            Self::LocalAddr { host, source } => {
                write!(f, "failed to read local address for {host}: {source}")
            }
            Self::DaemonConfig(err) => write!(f, "failed to load daemon config: {err}"),
            Self::MachineId(err) => write!(f, "refusing to start: {err}"),
            Self::RunDir(err) => write!(f, "refusing to start: {err}"),
            Self::Serve(err) => write!(f, "daemon server failed: {err}"),
            Self::ServeJoin(err) => write!(f, "daemon server task failed: {err}"),
        }
    }
}

impl Error for BootstrapError {
    fn source(&self) -> Option<&(dyn Error + 'static)> {
        match self {
            Self::StartupNotServing => None,
            #[cfg(unix)]
            Self::Signal(source) => Some(source),
            Self::InvalidPort { source, .. } => Some(source),
            Self::ConnectionFileRead { source, .. }
            | Self::ConnectionFileWrite { source, .. }
            | Self::GenerateConnectionFile(source) => Some(source),
            Self::StartLockCreate { source, .. }
            | Self::RunDirLockCreate { source, .. }
            | Self::RemoveStale { source, .. }
            | Self::Bind { source, .. }
            | Self::LocalAddr { source, .. } => Some(source),
            Self::DaemonConfig(err) => Some(err),
            Self::MachineId(err) => Some(err),
            Self::RunDir(err) => Some(err),
            Self::Serve(err) => Some(err),
            Self::ServeJoin(err) => Some(err),
            Self::StartLockBusy { .. } | Self::RunDirBusy { .. } => None,
        }
    }
}

#[cfg(test)]
mod tests {
    mod activity {
        include!("activity_bootstrap_tests.rs");
    }
    use super::*;
    use crate::server::ServerAuth;
    use cortexkit_test_support::ScratchDir;
    #[cfg(target_os = "linux")]
    use std::collections::BTreeSet;
    use std::sync::Mutex;
    #[cfg(target_os = "linux")]
    use subc_control::ModuleProtocol;
    use subc_transport::MIN_KEY_LEN;
    use tokio::io::AsyncReadExt;
    use tokio::task::JoinHandle;

    #[cfg(unix)]
    use std::os::unix::fs::PermissionsExt;

    static ENV_LOCK: Mutex<()> = Mutex::new(());

    #[test]
    fn normalized_build_provenance_preserves_real_values() {
        assert_eq!(normalized_build_provenance("abc"), Some("abc".to_string()));
    }

    #[test]
    fn normalized_build_provenance_omits_unavailable_and_empty_values() {
        assert_eq!(normalized_build_provenance("unavailable"), None);
        assert_eq!(normalized_build_provenance(""), None);
    }

    #[cfg(target_os = "linux")]
    fn current_cgroup_path_for_test() -> io::Result<PathBuf> {
        let cgroups = fs::read_to_string("/proc/self/cgroup")?;
        let relative = cgroups
            .lines()
            .find_map(|line| line.strip_prefix("0::"))
            .ok_or_else(|| {
                io::Error::new(io::ErrorKind::Unsupported, "cgroup v2 is unavailable")
            })?;
        Ok(Path::new("/sys/fs/cgroup").join(relative.trim_start_matches('/')))
    }

    #[cfg(target_os = "linux")]
    fn module_cgroup_directories() -> io::Result<BTreeSet<OsString>> {
        let modules = current_cgroup_path_for_test()?.join("subc-modules");
        let entries = match fs::read_dir(modules) {
            Ok(entries) => entries,
            Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(BTreeSet::new()),
            Err(error) => return Err(error),
        };
        let mut directories = BTreeSet::new();
        for entry in entries {
            let entry = entry?;
            if entry.file_type()?.is_dir() {
                directories.insert(entry.file_name());
            }
        }
        Ok(directories)
    }

    #[cfg(target_os = "linux")]
    async fn wait_for_path(path: &Path, task: &JoinHandle<Result<(), BootstrapError>>) {
        let deadline = tokio::time::Instant::now() + Duration::from_secs(5);
        while !path.exists() && tokio::time::Instant::now() < deadline {
            assert!(
                !task.is_finished(),
                "daemon exited before creating {}",
                path.display()
            );
            sleep(Duration::from_millis(10)).await;
        }
        assert!(path.exists(), "daemon did not create {}", path.display());
    }

    /// An in-process daemon with the default (disabled) cgroup placement must
    /// leave the host's live module cgroup tree exactly as it found it.
    /// Red-checking this test (making it fail) creates a
    /// `cgroup-isolation-probe-*` directory in the live cgroup tree, which must be removed by hand.
    #[cfg(target_os = "linux")]
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn run_with_config_does_not_reconcile_the_ambient_cgroup_by_default() {
        let temp = unique_temp_dir("bootstrap-cgroup-default-disabled");
        let module_id = format!("cgroup-isolation-probe-{}", process::id());
        let before = module_cgroup_directories().expect("read ambient module cgroups before boot");
        assert!(
            !before.contains(&OsString::from(&module_id)),
            "isolation probe cgroup already exists before this daemon starts"
        );
        let capture = temp.join("logs").join(format!("{module_id}.stderr.log"));
        let module = ConfiguredModule {
            module_id,
            program: PathBuf::from("sh"),
            args: vec!["-c".to_string(), "sleep 30".to_string()],
            env: Vec::new(),
            log: None,
            enabled: true,
            reserved: false,
            reserved_prefixes: Vec::new(),
            protocol: ModuleProtocol::None,
            overlap: Default::default(),
            health: HealthConfig::default(),
            drain_timeout_ms: None,
            route_bind_relay_timeout_ms: None,
            restart: RestartPolicy::default(),
        };
        let config = BootstrapConfig::new(temp.join("connection.json"), 0)
            .with_configured_modules([module])
            .with_capture_logs_dir(temp.join("logs"))
            .with_terminal_journal_path(temp.join("terminals.jsonl"));
        let task = tokio::spawn(run_with_config(config));

        wait_for_path(&capture, &task).await;
        let after = module_cgroup_directories().expect("read ambient module cgroups after boot");

        task.abort();
        assert!(task
            .await
            .expect_err("aborted daemon task must cancel")
            .is_cancelled());
        assert_eq!(
            before, after,
            "an in-process daemon must not create or reconcile ambient module cgroups"
        );
    }

    #[cfg(target_os = "linux")]
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn explicit_cgroup_root_is_prepared_inside_the_fixture_tree() {
        let temp = unique_temp_dir("bootstrap-cgroup-explicit-root");
        let cgroup_root = temp.join("cgroup");
        fs::create_dir(&cgroup_root).expect("create scratch cgroup root");
        fs::write(cgroup_root.join("cgroup.procs"), b"").expect("write scratch cgroup marker");
        let modules = cgroup_root.join("subc-modules");
        let config = BootstrapConfig::new(temp.join("connection.json"), 0)
            .with_cgroup_placement(CgroupPlacementConfig::Root(cgroup_root))
            .with_terminal_journal_path(temp.join("terminals.jsonl"));
        let task = tokio::spawn(run_with_config(config));

        wait_for_path(&modules, &task).await;

        task.abort();
        assert!(task
            .await
            .expect_err("aborted daemon task must cancel")
            .is_cancelled());
        assert!(
            fs::read_dir(&modules)
                .expect("read prepared modules directory")
                .next()
                .is_none(),
            "the delegation probe must clean up after itself"
        );
    }

    struct EnvGuard {
        key: &'static str,
        previous: Option<OsString>,
    }

    impl EnvGuard {
        fn set(key: &'static str, value: &Path) -> Self {
            let previous = env::var_os(key);
            env::set_var(key, value);
            Self { key, previous }
        }

        fn set_str(key: &'static str, value: &str) -> Self {
            let previous = env::var_os(key);
            env::set_var(key, value);
            Self { key, previous }
        }

        fn unset(key: &'static str) -> Self {
            let previous = env::var_os(key);
            env::remove_var(key);
            Self { key, previous }
        }
    }

    impl Drop for EnvGuard {
        fn drop(&mut self) {
            match &self.previous {
                Some(value) => env::set_var(self.key, value),
                None => env::remove_var(self.key),
            }
        }
    }

    fn unique_temp_dir(name: &str) -> ScratchDir {
        ScratchDir::new(name)
    }

    fn temp_connection_file_path(name: &str) -> (ScratchDir, PathBuf) {
        let dir = unique_temp_dir(name);
        let path = dir.join("conn.json");
        (dir, path)
    }

    fn auth_for(info: &ConnectionInfo) -> ServerAuth {
        ServerAuth::new(info.key.clone(), info.daemon_id, info.daemon_ver.clone())
    }

    fn start_server(bound: BoundDaemon) -> JoinHandle<Result<(), ServerError>> {
        let auth = auth_for(&bound.connection_info);
        tokio::spawn(serve_listeners(
            bound.listeners,
            Arc::new(Router::with_default_self_handler()),
            auth,
        ))
    }

    fn expect_bound(outcome: Outcome) -> BoundDaemon {
        match outcome {
            Outcome::Bound(bound) => bound,
            Outcome::AlreadyRunning => panic!("fresh connection file unexpectedly had a daemon"),
        }
    }

    async fn connect_from_info(conn: &ConnectionInfo) -> io::Result<TcpStream> {
        let endpoint = conn
            .endpoints
            .first()
            .expect("test connection file should have an endpoint");
        let ip: IpAddr = endpoint.host.parse().unwrap();
        TcpStream::connect(SocketAddr::new(ip, endpoint.port)).await
    }

    fn make_connection_info(port: u16) -> ConnectionInfo {
        ConnectionInfo {
            schema: SCHEMA_VERSION,
            wire_version: Some(PROTOCOL_VERSION),
            endpoints: vec![Endpoint {
                host: "127.0.0.1".to_owned(),
                port,
            }],
            key: generate_key().unwrap(),
            daemon_id: generate_daemon_id().unwrap(),
            pid: process::id(),
            daemon_ver: "test-subc".to_owned(),
        }
    }

    fn write_raw_owner_only_connection_file(path: &Path, contents: &[u8]) {
        fs::write(path, contents).unwrap();
        #[cfg(unix)]
        fs::set_permissions(path, fs::Permissions::from_mode(0o600)).unwrap();
    }

    fn assert_owner_only_connection_file(path: &Path) {
        // `path` is only inspected on Unix (mode bits); on Windows the owner-only
        // guarantee comes from the inherited %TEMP% ACL, nothing to assert here.
        #[cfg(unix)]
        {
            let mode = fs::metadata(path).unwrap().permissions().mode() & 0o777;
            assert_eq!(mode, 0o600);
        }
        #[cfg(not(unix))]
        let _ = path;
    }

    #[test]
    fn connection_file_path_uses_xdg_runtime_dir_when_set() {
        let _env_lock = ENV_LOCK.lock().unwrap();
        let runtime_dir = unique_temp_dir("xdg-runtime");
        let _xdg = EnvGuard::set("XDG_RUNTIME_DIR", runtime_dir.path());

        assert_eq!(
            connection_file_path(),
            runtime_dir.join(CONNECTION_FILE_NAME)
        );
    }

    #[test]
    fn connection_file_path_source_is_xdg_runtime_dir_when_set() {
        let runtime_dir = OsString::from("/run/user/1000");

        let (path, source) = connection_file_path_with_source(Some(runtime_dir));

        assert_eq!(
            path,
            PathBuf::from("/run/user/1000").join(CONNECTION_FILE_NAME)
        );
        assert_eq!(source, ConnectionFileSource::XdgRuntimeDir);
    }

    #[test]
    fn connection_file_path_falls_back_to_temp_dir_with_user_token_when_xdg_unset() {
        let _env_lock = ENV_LOCK.lock().unwrap();
        let _xdg = EnvGuard::unset("XDG_RUNTIME_DIR");

        assert_eq!(
            connection_file_path(),
            env::temp_dir().join(format!("subc-{}.connection.json", user_connection_token()))
        );
    }

    #[test]
    fn connection_file_path_source_is_temp_dir_when_xdg_unset() {
        let (path, source) = connection_file_path_with_source(None);

        assert_eq!(
            path,
            env::temp_dir().join(format!("subc-{}.connection.json", user_connection_token()))
        );
        assert_eq!(source, ConnectionFileSource::TempDirFallback);
    }

    /// Concurrent callers must derive one token. The former temp-file uid probe
    /// could fail transiently (same-tick name collision, fd exhaustion) and send
    /// the loser down the env-derived fallback with a different identity for the
    /// same user; this fence keeps identity independent of filesystem luck.
    #[test]
    fn user_connection_token_is_stable_under_concurrent_callers() {
        let expected = user_connection_token();
        let workers: Vec<_> = (0..32)
            .map(|_| {
                std::thread::spawn(|| (0..40).map(|_| user_connection_token()).collect::<Vec<_>>())
            })
            .collect();
        for worker in workers {
            for token in worker.join().expect("probe thread") {
                assert_eq!(token, expected, "token diverged under concurrent probes");
            }
        }
    }

    #[test]
    fn connection_file_path_source_is_temp_dir_when_xdg_empty() {
        let (path, source) = connection_file_path_with_source(Some(OsString::new()));

        assert_eq!(
            path,
            env::temp_dir().join(format!("subc-{}.connection.json", user_connection_token()))
        );
        assert_eq!(source, ConnectionFileSource::TempDirFallback);
    }

    /// The shipped binary is the only caller that journals terminal exits into
    /// the real run directory, so its constructor must supply that path itself;
    /// the in-process default leaves it unset.
    #[test]
    fn daemon_binary_config_journals_and_captures_into_the_run_dir() {
        let _env_lock = ENV_LOCK.lock().unwrap();
        let root = unique_temp_dir("daemon-binary-config");
        let data_home = root.join("data");
        let _data = EnvGuard::set("XDG_DATA_HOME", &data_home);
        let _config = EnvGuard::set("XDG_CONFIG_HOME", &root.join("config"));
        let _port = EnvGuard::unset(SUBC_PORT_ENV);

        let config = BootstrapConfig::from_env_for_daemon_binary().unwrap();

        let run_dir = data_home.join("cortexkit").join("run");
        assert_eq!(
            config.terminal_journal_path,
            Some(run_dir.join("terminals.jsonl"))
        );
        assert_eq!(config.capture_logs_dir, Some(run_dir.join("logs")));
        assert_eq!(
            BootstrapConfig::new(root.join("connection.json"), 0).terminal_journal_path,
            None,
            "an in-process config must not journal anywhere unless asked to"
        );
    }

    #[test]
    fn daemon_binary_config_refuses_a_relative_data_home() {
        let _env_lock = ENV_LOCK.lock().unwrap();
        let root = unique_temp_dir("daemon-binary-relative-data");
        let _data = EnvGuard::set_str("XDG_DATA_HOME", "relative-data-home");
        let _config = EnvGuard::set("XDG_CONFIG_HOME", &root.join("config"));
        let _port = EnvGuard::unset(SUBC_PORT_ENV);

        let error = BootstrapConfig::from_env_for_daemon_binary()
            .expect_err("a relative data home must refuse the daemon binary's config");
        assert!(
            matches!(error, BootstrapError::RunDir(_)),
            "expected a run-directory refusal, got {error}"
        );
        assert!(error.to_string().contains("XDG_DATA_HOME"), "{error}");
    }

    #[test]
    fn configured_port_uses_default_config_and_env_override() {
        let _env_lock = ENV_LOCK.lock().unwrap();
        let (_dir, conn_path) = temp_connection_file_path("daemon-config-port");
        let config_path = conn_path.with_file_name("subc.jsonc");

        let _port = EnvGuard::unset(SUBC_PORT_ENV);
        assert_eq!(
            BootstrapConfig::from_env_with_daemon_config_path(&config_path)
                .unwrap()
                .port,
            DEFAULT_SUBC_PORT
        );

        fs::write(&config_path, r#"{ "version": 1, "port": 8123 }"#).unwrap();
        assert_eq!(
            BootstrapConfig::from_env_with_daemon_config_path(&config_path)
                .unwrap()
                .port,
            8123
        );

        let _port = EnvGuard::set_str(SUBC_PORT_ENV, "9012");
        assert_eq!(
            BootstrapConfig::from_env_with_daemon_config_path(&config_path)
                .unwrap()
                .port,
            9012
        );
    }

    #[tokio::test]
    async fn second_singleton_probe_against_served_tcp_daemon_reports_already_running() {
        let (_dir, path) = temp_connection_file_path("already-running");

        let bound = expect_bound(ensure_singleton(&path, 0).await.unwrap());
        let server = start_server(bound);

        let second = ensure_singleton(&path, 0).await.unwrap();
        assert!(matches!(second, Outcome::AlreadyRunning));

        server.abort();
        let _ = server.await;
    }

    #[tokio::test]
    async fn daemon_connection_file_publishes_protocol_wire_version() {
        let (_dir, path) = temp_connection_file_path("wire-version");
        let bound = expect_bound(ensure_singleton(&path, 0).await.unwrap());
        assert_eq!(bound.connection_info.wire_version, Some(PROTOCOL_VERSION));
        assert_eq!(
            connection_file::read(&path).unwrap().wire_version,
            Some(PROTOCOL_VERSION)
        );

        drop(bound.listeners);
    }

    #[tokio::test]
    async fn stale_unbound_connection_file_is_reclaimed() {
        let (_dir, path) = temp_connection_file_path("stale-reclaim");
        let stale = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).await.unwrap();
        let stale_port = stale.local_addr().unwrap().port();
        drop(stale);
        let stale_info = make_connection_info(stale_port);
        write_atomic(&path, &stale_info).unwrap();

        let bound = expect_bound(ensure_singleton(&path, 0).await.unwrap());
        assert_ne!(bound.connection_info.key, stale_info.key);
        drop(bound.listeners);
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn ensure_singleton_reclaims_insecure_connection_file() {
        let (_dir, path) = temp_connection_file_path("insecure-reclaim");
        let stale = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).await.unwrap();
        let stale_port = stale.local_addr().unwrap().port();
        drop(stale);
        let stale_info = make_connection_info(stale_port);
        write_atomic(&path, &stale_info).unwrap();
        fs::set_permissions(&path, fs::Permissions::from_mode(0o644)).unwrap();

        let bound = expect_bound(ensure_singleton(&path, 0).await.unwrap());
        assert_ne!(bound.connection_info.key, stale_info.key);
        assert_ne!(bound.connection_info.daemon_id, stale_info.daemon_id);
        assert_owner_only_connection_file(&path);

        drop(bound.listeners);
    }

    #[tokio::test]
    async fn ensure_singleton_reclaims_non_loopback_connection_file() {
        let (_dir, path) = temp_connection_file_path("non-loopback-reclaim");
        let mut stale_info = make_connection_info(8757);
        stale_info.endpoints = vec![Endpoint {
            host: "192.0.2.10".to_owned(),
            port: 8757,
        }];
        write_atomic(&path, &stale_info).unwrap();

        let bound = expect_bound(ensure_singleton(&path, 0).await.unwrap());
        assert_ne!(bound.connection_info.key, stale_info.key);
        assert_ne!(bound.connection_info.daemon_id, stale_info.daemon_id);
        assert!(bound
            .connection_info
            .endpoints
            .iter()
            .all(|endpoint| endpoint.host.parse::<IpAddr>().unwrap().is_loopback()));
        assert_owner_only_connection_file(&path);

        drop(bound.listeners);
    }

    #[tokio::test]
    async fn ensure_singleton_reclaims_invalid_connection_file_shapes() {
        let mut unsupported_schema = make_connection_info(8757);
        unsupported_schema.schema = SCHEMA_VERSION + 1;

        let mut empty_endpoints = make_connection_info(8757);
        empty_endpoints.endpoints.clear();

        let mut short_key = make_connection_info(8757);
        short_key.key = vec![0x5A; MIN_KEY_LEN - 1];

        let cases = vec![
            (
                "unsupported-schema",
                serde_json::to_vec(&unsupported_schema).unwrap(),
                Some(unsupported_schema),
            ),
            (
                "empty-endpoints",
                serde_json::to_vec(&empty_endpoints).unwrap(),
                Some(empty_endpoints),
            ),
            (
                "short-key",
                serde_json::to_vec(&short_key).unwrap(),
                Some(short_key),
            ),
            ("invalid-json", b"{not valid connection json".to_vec(), None),
        ];

        for (label, contents, old_info) in cases {
            let (_dir, path) = temp_connection_file_path(label);
            write_raw_owner_only_connection_file(&path, &contents);

            let bound = expect_bound(ensure_singleton(&path, 0).await.unwrap());
            if let Some(old_info) = old_info {
                assert_ne!(bound.connection_info.key, old_info.key, "{label}");
                assert_ne!(
                    bound.connection_info.daemon_id, old_info.daemon_id,
                    "{label}"
                );
            }
            assert!(bound.connection_info.key.len() >= MIN_KEY_LEN, "{label}");
            assert_ne!(bound.connection_info.daemon_id, [0u8; 16], "{label}");
            assert_owner_only_connection_file(&path);

            drop(bound.listeners);
        }
    }

    #[tokio::test]
    async fn foreign_reused_port_connection_file_is_reclaimed_after_auth_probe_fails() {
        let (_dir, path) = temp_connection_file_path("foreign-reclaim");
        let foreign = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).await.unwrap();
        let foreign_port = foreign.local_addr().unwrap().port();
        write_atomic(&path, &make_connection_info(foreign_port)).unwrap();
        let foreign_task = tokio::spawn(async move {
            if let Ok((mut stream, _)) = foreign.accept().await {
                let mut buf = [0u8; 64];
                let _ = stream.read(&mut buf).await;
            }
        });

        let bound = expect_bound(ensure_singleton(&path, 0).await.unwrap());
        assert!(bound
            .connection_info
            .endpoints
            .iter()
            .all(|endpoint| endpoint.port != foreign_port));

        drop(bound.listeners);
        let _ = foreign_task.await;
    }

    #[tokio::test]
    async fn stale_start_lock_file_is_reclaimable() {
        let (_dir, path) = temp_connection_file_path("start-lock-stale-file");
        let lock_path = start_lock_path(&path);
        drop(open_owner_only_lock(&lock_path).unwrap());
        assert!(lock_path.is_file());

        let lock = StartLock::acquire(&path).await.unwrap();
        assert!(lock_path.is_file());

        drop(lock);
        assert!(lock_path.is_file());
    }

    #[tokio::test]
    async fn held_start_lock_blocks_second_acquire_until_release() {
        let (_dir, path) = temp_connection_file_path("start-lock-held");
        let lock_path = start_lock_path(&path);
        let first = StartLock::acquire(&path).await.unwrap();

        let err = match StartLock::acquire(&path).await {
            Ok(_) => panic!("second acquire while held must stay busy"),
            Err(err) => err,
        };
        assert!(matches!(
            err,
            BootstrapError::StartLockBusy {
                ref path,
                attempts: START_LOCK_RETRIES,
            } if path == &lock_path
        ));

        drop(first);

        let second = StartLock::acquire(&path)
            .await
            .expect("released advisory lock should be reclaimable");
        drop(second);
    }

    #[tokio::test]
    async fn bind_conflict_on_fixed_port_fails_loud_without_reselecting() {
        let (_dir, path) = temp_connection_file_path("bind-conflict");
        let occupied = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).await.unwrap();
        let occupied_port = occupied.local_addr().unwrap().port();

        let err = ensure_singleton(&path, occupied_port).await.unwrap_err();
        assert!(matches!(
            err,
            BootstrapError::Bind { ref source, .. } if source.kind() == io::ErrorKind::AddrInUse
        ));
        assert!(err.to_string().contains("set the port in config"));

        drop(occupied);
    }

    #[tokio::test]
    async fn ephemeral_ipv6_collision_still_serves_ipv4() {
        let (listeners, endpoints) = bind_loopback_with(0, |port| async move {
            let occupied = TcpListener::bind((Ipv6Addr::LOCALHOST, port)).await?;
            let result = TcpListener::bind((Ipv6Addr::LOCALHOST, port)).await;
            drop(occupied);
            result
        })
        .await
        .expect("an incidental IPv6 collision must not prevent an ephemeral daemon from starting");
        assert_eq!(endpoints.len(), 1);
        assert_eq!(endpoints[0].host, "127.0.0.1");
        TcpStream::connect((Ipv4Addr::LOCALHOST, endpoints[0].port))
            .await
            .unwrap();
        drop(listeners);
    }

    #[tokio::test]
    async fn unpublished_boot_holds_singleton_lock_until_publication() {
        let dir = ScratchDir::new("unpublished-singleton-lock");
        let path = dir.join("connection.json");
        let bound = expect_bound(
            ensure_singleton_inner(BootstrapConfig::new(&path, 0), false)
                .await
                .unwrap(),
        );
        assert!(!path.exists());
        let contender = open_owner_only_lock(&start_lock_path(&path)).unwrap();
        assert!(
            matches!(FileExt::try_lock(&contender), Err(TryLockError::WouldBlock)),
            "an unpublished boot must keep its singleton claim"
        );
        drop(bound);
        FileExt::try_lock(&contender).unwrap();
    }

    #[tokio::test]
    async fn readiness_probe_requires_an_authenticating_server() {
        let dir = ScratchDir::new("unserved-readiness");
        let bound = expect_bound(
            ensure_singleton_inner(BootstrapConfig::new(dir.join("connection.json"), 0), false)
                .await
                .unwrap(),
        );
        assert!(matches!(
            verify_serving(&bound.connection_info).await,
            Err(BootstrapError::StartupNotServing)
        ));
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn orphan_sweep_finishes_before_connection_file_publication() {
        use crate::live_children::{ExecutableIdentity, LiveChild};
        let dir = ScratchDir::new("publish-after-sweep");
        let path = dir.join("connection.json");
        let record = dir.join("live-children.json");
        let ready = dir.join("ready");
        let mut child = std::process::Command::new("/bin/sh")
            .args([
                "-c",
                "trap '' TERM; : > \"$0\"; read -r _",
                ready.to_str().unwrap(),
            ])
            .stdin(std::process::Stdio::piped())
            .env("XDG_DATA_HOME", dir.path())
            .env("XDG_RUNTIME_DIR", dir.path())
            .env("XDG_CONFIG_HOME", dir.path())
            .spawn()
            .unwrap();
        timeout(Duration::from_secs(5), async {
            while !ready.exists() {
                sleep(Duration::from_millis(5)).await;
            }
        })
        .await
        .unwrap();
        let observed = subc_os::Process::open(child.id())
            .unwrap()
            .unwrap()
            .observe()
            .unwrap();
        crate::live_children::write_record(
            &record,
            &[LiveChild {
                module_id: "old-sleep".into(),
                pid: child.id(),
                protocol: subc_control::ModuleProtocol::None,
                start_time: Some(observed.start_time),
                executable: observed.executable.map(ExecutableIdentity::from),
                cgroup_name: None,
            }],
        )
        .unwrap();
        let config = BootstrapConfig::new(&path, 0).with_live_children_record(&record);
        let daemon = tokio::spawn(run_with_config(config));
        timeout(Duration::from_secs(5), async {
            while !path.exists() {
                sleep(Duration::from_millis(5)).await;
            }
        })
        .await
        .unwrap();
        assert!(path.exists());
        let ended = child.try_wait().unwrap().is_some();
        if !ended {
            child.kill().unwrap();
        }
        child.wait().unwrap();
        if ended {
            assert!(matches!(probe_existing(&path).await.unwrap(), Probe::Live));
        }
        daemon.abort();
        let _ = daemon.await;
        assert!(
            ended,
            "the published daemon must not still owe its orphan sweep"
        );
    }

    /// Publishing the connection file fsyncs, which can take seconds on a busy
    /// disk, so the daemon must do it off the async worker thread. The hook
    /// blocks inside the write until another task on the same single-threaded
    /// runtime has run. Writing on the worker thread would starve that task: the
    /// hook then gives up after its deadline and the test fails, as it does if
    /// startup publishes without going through the hooked write at all.
    #[tokio::test(flavor = "current_thread")]
    async fn connection_file_publication_leaves_the_async_runtime_free() {
        use std::sync::atomic::{AtomicBool, Ordering};

        let (_dir, path) = temp_connection_file_path("publish-off-runtime");
        let write_started = Arc::new(AtomicBool::new(false));
        let (progress_tx, progress_rx) = std::sync::mpsc::channel::<()>();
        let (verdict_tx, verdict_rx) = std::sync::mpsc::channel::<bool>();
        let hook_started = Arc::clone(&write_started);
        connection_file_write_hook::install(&path, move || {
            hook_started.store(true, Ordering::SeqCst);
            let progressed = progress_rx.recv_timeout(Duration::from_secs(5)).is_ok();
            let _ = verdict_tx.send(progressed);
        });
        // Signals only after the write has started, so a message can only come
        // from this task running while the write is still blocked.
        let ticker = tokio::spawn(async move {
            loop {
                if write_started.load(Ordering::SeqCst) {
                    let _ = progress_tx.send(());
                    return;
                }
                sleep(Duration::from_millis(5)).await;
            }
        });

        let daemon = tokio::spawn(run_with_config(BootstrapConfig::new(&path, 0)));
        let published = timeout(Duration::from_secs(15), async {
            while !path.exists() && !daemon.is_finished() {
                sleep(Duration::from_millis(5)).await;
            }
        })
        .await;
        let daemon_exited = daemon.is_finished();
        daemon.abort();
        let daemon_result = daemon.await;
        ticker.abort();

        assert!(
            published.is_ok() && !daemon_exited,
            "the daemon must start and publish its connection file: {daemon_result:?}"
        );
        match verdict_rx.try_recv() {
            Ok(true) => {}
            Ok(false) => panic!(
                "no other task on the runtime ran while the connection file was being written: the write is blocking the async worker thread"
            ),
            Err(_) => panic!("startup published the connection file without the hooked write"),
        }
    }

    /// A write that never reports back must stop startup with an error naming
    /// the connection file, never be taken as published.
    #[tokio::test]
    async fn a_panicking_connection_file_write_fails_startup() {
        let (_dir, path) = temp_connection_file_path("publish-panics");
        connection_file_write_hook::install(&path, || {
            panic!("deliberate panic inside the connection-file write")
        });

        let error = timeout(
            Duration::from_secs(15),
            run_with_config(BootstrapConfig::new(&path, 0)),
        )
        .await
        .expect("startup must end once the write task fails")
        .expect_err("startup must fail when the connection-file write panics");

        assert!(
            matches!(&error, BootstrapError::ConnectionFileWrite { path: failed, .. } if failed == &path),
            "expected a connection-file write error, got {error:?}"
        );
        let message = error.to_string();
        assert!(message.contains(&path.display().to_string()), "{message}");
        assert!(message.contains("panicked"), "{message}");
        assert!(!path.exists(), "no connection file may be published");
    }

    #[cfg(target_os = "macos")]
    #[test]
    fn nofile_raise_worker() {
        let Some(path) = std::env::var_os("SUBC_NOFILE_TEST_FILE") else {
            return;
        };
        let (_, hard) = rlimit::Resource::NOFILE.get().unwrap();
        rlimit::Resource::NOFILE.set(256, hard).unwrap();
        raise_nofile_limit_with_ceiling(Some(10240));
        std::fs::write(path, rlimit::Resource::NOFILE.get().unwrap().0.to_string()).unwrap();
    }

    #[cfg(target_os = "macos")]
    #[test]
    fn macos_nofile_raise_clamps_to_kernel_ceiling() {
        let dir = ScratchDir::new("nofile-ceiling");
        let result = dir.join("soft");
        let status = std::process::Command::new(std::env::current_exe().unwrap())
            .args(["--exact", "bootstrap::tests::nofile_raise_worker"])
            .env("SUBC_NOFILE_TEST_FILE", &result)
            .env("XDG_DATA_HOME", dir.path())
            .env("XDG_RUNTIME_DIR", dir.path())
            .env("XDG_CONFIG_HOME", dir.path())
            .status()
            .unwrap();
        assert!(status.success());
        let soft: u64 = std::fs::read_to_string(result).unwrap().parse().unwrap();
        assert!(
            soft > 256 && soft <= 10240,
            "the inherited limit must be raised within the kernel ceiling; got {soft}"
        );
    }

    #[tokio::test]
    async fn key_rotation_republishes_new_material_and_old_file_fails_auth() {
        let (_dir, path) = temp_connection_file_path("key-rotation");
        let first = expect_bound(ensure_singleton(&path, 0).await.unwrap());
        let old_info = first.connection_info.clone();
        let fixed_port = old_info.endpoints[0].port;
        drop(first.listeners);

        let second = expect_bound(ensure_singleton(&path, fixed_port).await.unwrap());
        let new_info = second.connection_info.clone();
        assert_ne!(old_info.key, new_info.key);
        assert_ne!(old_info.daemon_id, new_info.daemon_id);
        let server = start_server(second);

        let mut old_stream = connect_from_info(&old_info).await.unwrap();
        let old_auth = authenticate_client(&mut old_stream, &old_info, PROBE_AUTH_DEADLINE).await;
        assert!(
            old_auth.is_err(),
            "old key must not authenticate after restart"
        );

        let reread = connection_file::read(&path).unwrap();
        let mut new_stream = connect_from_info(&reread).await.unwrap();
        authenticate_client(&mut new_stream, &reread, PROBE_AUTH_DEADLINE)
            .await
            .unwrap();

        server.abort();
        let _ = server.await;
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn published_connection_file_permissions_are_owner_only() {
        let (_dir, path) = temp_connection_file_path("permissions");
        let bound = expect_bound(ensure_singleton(&path, 0).await.unwrap());

        let mode = fs::metadata(&path).unwrap().permissions().mode() & 0o777;
        assert_eq!(mode, 0o600);

        drop(bound.listeners);
    }
}