bcode_daemon_lifecycle 0.0.1-alpha.0

#![cfg_attr(feature = "fail-on-warnings", deny(warnings))]
#![warn(clippy::all, clippy::pedantic, clippy::nursery, clippy::cargo)]
#![allow(clippy::multiple_crate_versions)]

//! Daemon lifecycle registry and cleanup models.

mod execution_lifetime;
pub use execution_lifetime::{
    ExecutionLifetime, ExecutionLifetimeStatus, ExecutionMaintenance, execution_lifetime_status,
};

use bcode_ipc::{BUILD_FINGERPRINT, CURRENT_PROTOCOL_VERSION, IpcEndpoint, daemon_namespace};
use bcode_plugin_sdk::path::display_from_current_dir;
use serde::{Deserialize, Serialize};
use sha2::{Digest as _, Sha256};
use std::fs;
use std::io::{Read, Seek as _, Write as _};
#[cfg(unix)]
use std::os::fd::AsRawFd as _;
#[cfg(target_os = "macos")]
use std::os::unix::ffi::OsStrExt as _;
#[cfg(all(unix, test))]
use std::os::unix::fs::MetadataExt as _;
#[cfg(unix)]
use std::os::unix::fs::PermissionsExt;
use std::path::{Path, PathBuf};
use std::sync::{Mutex, OnceLock};
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use sysinfo::{Pid, ProcessesToUpdate, System};
use thiserror::Error;

/// Environment variable carrying parent-verified executable digest evidence to a spawned daemon.
pub const BCODE_EXECUTABLE_DIGEST_ENV: &str = "BCODE_EXECUTABLE_DIGEST";

/// Current daemon registry record schema version.
///
/// Version 4 adds `state_location_id`. The field is optional so version 3 records
/// still decode; a record without it is unverifiable rather than assumed local.
/// Version 5 widens that identity to the runtime scope pair
/// `(state root, config directory)`, so a version 4 record's narrower identity no
/// longer matches a current daemon and is treated as foreign rather than local.
pub const DAEMON_RECORD_SCHEMA_VERSION: u32 = 5;

/// Errors returned by daemon lifecycle registry operations.
#[derive(Debug, Error)]
pub enum DaemonLifecycleError {
    /// Registry I/O failed.
    #[error("daemon registry I/O error at {}: {source}", path.display())]
    Io {
        /// Path associated with the failed operation.
        path: PathBuf,
        /// Original I/O error.
        source: std::io::Error,
    },
    /// Registry serialization failed.
    #[error("daemon registry serialization error: {0}")]
    Serialize(#[from] serde_json::Error),
    /// System time was before the Unix epoch.
    #[error("system clock is before Unix epoch")]
    Clock,
}

/// Serializable local IPC endpoint metadata.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum DaemonEndpointRecord {
    /// Unix domain socket endpoint.
    UnixSocket {
        /// Socket path.
        path: PathBuf,
    },
    /// Windows named pipe endpoint.
    WindowsNamedPipe {
        /// Named pipe path.
        name: String,
    },
    /// Endpoint shape not known by this build.
    Unknown {
        /// Debug representation captured for diagnostics.
        debug: String,
    },
}

impl DaemonEndpointRecord {
    /// Convert this record into an IPC endpoint when supported by the current platform.
    #[must_use]
    pub fn to_ipc_endpoint(&self) -> Option<IpcEndpoint> {
        match self {
            Self::UnixSocket { path } => {
                #[cfg(unix)]
                {
                    Some(IpcEndpoint::unix_socket(path.clone()))
                }
                #[cfg(not(unix))]
                {
                    let _ = path;
                    None
                }
            }
            Self::WindowsNamedPipe { name } => {
                #[cfg(windows)]
                {
                    Some(IpcEndpoint::windows_named_pipe(name.clone()))
                }
                #[cfg(not(windows))]
                {
                    let _ = name;
                    None
                }
            }
            Self::Unknown { .. } => None,
        }
    }
}

/// Persistent metadata for one daemon instance.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct DaemonRecord {
    /// Record schema version.
    pub schema_version: u32,
    /// Daemon namespace.
    pub namespace: String,
    /// IPC protocol version.
    pub protocol_version: u32,
    /// Exact produced-artifact identity, when advertised by this daemon generation.
    #[serde(default)]
    pub artifact_id: Option<bcode_ipc::ArtifactId>,
    /// Build fingerprint retained as source/build diagnostic evidence.
    pub build_fingerprint: String,
    /// Durable session-storage writer epoch supported by this daemon, when known.
    #[serde(default)]
    pub storage_writer_epoch: Option<u32>,
    /// Identity of the durable state location this daemon serves.
    ///
    /// Records already live under the state location they describe, so this is
    /// corroborating evidence rather than the routing key. It lets classification
    /// surface a record copied or inherited from another location instead of
    /// silently treating it as local.
    #[serde(default)]
    pub state_location_id: Option<String>,
    /// Process identifier, when available.
    pub pid: Option<u32>,
    /// IPC endpoint for this daemon.
    pub endpoint: DaemonEndpointRecord,
    /// Daemon log path.
    pub log_path: PathBuf,
    /// Executable path used to start the daemon.
    pub executable_path: Option<PathBuf>,
    /// SHA-256 digest of the executable bytes running this daemon.
    #[serde(default)]
    pub executable_digest: Option<String>,
    /// Daemon start time in Unix milliseconds.
    pub started_at_unix_ms: u64,
    /// Last registry write/update time in Unix milliseconds.
    pub last_seen_unix_ms: u64,
    /// Random per-process identity token.
    pub instance_id: String,
}

impl DaemonRecord {
    /// Build a daemon record for the current process and build.
    ///
    /// # Errors
    ///
    /// Returns an error when the system clock is before the Unix epoch.
    pub fn current(
        endpoint: &IpcEndpoint,
        log_path: PathBuf,
        executable_path: Option<PathBuf>,
        instance_id: String,
    ) -> Result<Self, DaemonLifecycleError> {
        let executable_digest = executable_path
            .as_deref()
            .map(executable_sha256)
            .transpose()?;
        Self::current_with_digest(
            endpoint,
            log_path,
            executable_path,
            executable_digest,
            instance_id,
        )
    }

    /// Build a daemon record from already-verified executable digest evidence.
    ///
    /// # Errors
    ///
    /// Returns an error when the system clock is before the Unix epoch.
    pub fn current_with_digest(
        endpoint: &IpcEndpoint,
        log_path: PathBuf,
        executable_path: Option<PathBuf>,
        executable_digest: Option<String>,
        instance_id: String,
    ) -> Result<Self, DaemonLifecycleError> {
        let now = unix_time_millis()?;
        Ok(Self {
            schema_version: DAEMON_RECORD_SCHEMA_VERSION,
            namespace: daemon_namespace(),
            protocol_version: u32::from(CURRENT_PROTOCOL_VERSION),
            artifact_id: Some(bcode_ipc::ArtifactId::current()),
            build_fingerprint: BUILD_FINGERPRINT.to_string(),
            storage_writer_epoch: None,
            state_location_id: Some(bcode_ipc::state_location_id()),
            pid: Some(std::process::id()),
            endpoint: endpoint_record(endpoint),
            log_path,
            executable_path,
            executable_digest,
            started_at_unix_ms: now,
            last_seen_unix_ms: now,
            instance_id,
        })
    }

    /// Return true when this record describes the current daemon namespace.
    #[must_use]
    pub fn is_current_namespace(&self) -> bool {
        self.namespace == daemon_namespace()
    }
}

/// Return the daemon registry directory under a Bcode state directory.
#[must_use]
pub fn registry_dir(state_dir: &Path) -> PathBuf {
    state_dir.join("daemons")
}

/// Return the registry path for a daemon namespace.
#[must_use]
pub fn record_path(state_dir: &Path, namespace: &str) -> PathBuf {
    registry_dir(state_dir).join(format!("{namespace}.json"))
}

/// Return a registry path isolated by artifact and runtime scope.
#[must_use]
pub fn scoped_record_path(state_dir: &Path, record: &DaemonRecord) -> PathBuf {
    record.state_location_id.as_ref().map_or_else(
        || record_path(state_dir, &record.namespace),
        |scope| record_path(state_dir, &format!("{}-{scope}", record.namespace)),
    )
}

fn startup_lock_path(state_dir: &Path, namespace: &str, scope: &str) -> PathBuf {
    registry_dir(state_dir).join(format!("{namespace}-{scope}.lock"))
}

/// Write a daemon registry record atomically.
///
/// # Errors
///
/// Returns an error when creating directories, serializing, writing, or renaming fails.
pub fn write_record(
    state_dir: &Path,
    record: &DaemonRecord,
) -> Result<PathBuf, DaemonLifecycleError> {
    let dir = registry_dir(state_dir);
    fs::create_dir_all(&dir).map_err(|source| DaemonLifecycleError::Io {
        path: dir.clone(),
        source,
    })?;
    let path = scoped_record_path(state_dir, record);
    let temp_path = path.with_extension(format!("json.tmp-{}", record.instance_id));
    let contents = serde_json::to_vec_pretty(record)?;
    fs::write(&temp_path, contents).map_err(|source| DaemonLifecycleError::Io {
        path: temp_path.clone(),
        source,
    })?;
    fs::rename(&temp_path, &path).map_err(|source| DaemonLifecycleError::Io {
        path: path.clone(),
        source,
    })?;
    Ok(path)
}

/// Read daemon registry records from a state directory.
///
/// Invalid records are ignored so one bad file does not block cleanup.
#[must_use]
pub fn read_records(state_dir: &Path) -> Vec<(PathBuf, DaemonRecord)> {
    let dir = registry_dir(state_dir);
    let Ok(entries) = fs::read_dir(dir) else {
        return Vec::new();
    };
    entries
        .flatten()
        .filter_map(|entry| {
            let path = entry.path();
            if path.extension().and_then(|extension| extension.to_str()) != Some("json") {
                return None;
            }
            let contents = fs::read(&path).ok()?;
            let record = serde_json::from_slice(&contents).ok()?;
            Some((path, record))
        })
        .collect()
}

/// Read a complete bounded daemon inventory for safety-sensitive coordination.
///
/// Unlike [`read_records`], this never treats unreadable, malformed, oversized, unsupported or
/// duplicate records as absent. This is an observation, not a lock against concurrent registration;
/// callers must separately coordinate registration before relying on absence as authority.
///
/// # Errors
/// Returns a secret-safe IO error for missing/unsafe registry roots, unknown entry types, invalid
/// records, duplicate namespaces/instances, or exhausted entry/record-byte budgets.
pub fn read_records_strict(
    state_dir: &Path,
    entry_budget: usize,
) -> std::io::Result<Vec<DaemonRecord>> {
    const MAX_RECORD_BYTES: u64 = 64 * 1024;
    fn invalid() -> std::io::Error {
        std::io::Error::new(
            std::io::ErrorKind::InvalidData,
            "daemon inventory is incomplete or unverifiable",
        )
    }
    if entry_budget == 0 || entry_budget > 4096 {
        return Err(invalid());
    }
    let root = state_dir.canonicalize()?;
    let directory = root.join("daemons");
    if !fs::symlink_metadata(&directory)?.is_dir() || !directory.canonicalize()?.starts_with(&root)
    {
        return Err(invalid());
    }
    let mut records = Vec::new();
    let mut namespaces = std::collections::BTreeSet::new();
    let mut instances = std::collections::BTreeSet::new();
    for (index, entry) in fs::read_dir(&directory)?.enumerate() {
        if index == entry_budget {
            return Err(invalid());
        }
        let entry = entry?;
        let path = entry.path();
        let metadata = fs::symlink_metadata(&path)?;
        // Launch coordination and image publication locks share this directory with records.
        // They carry no registry identity, but must still be ordinary bounded files.
        if metadata.is_file()
            && metadata.len() <= MAX_RECORD_BYTES
            && (entry.file_name() == "daemon-images.lock"
                || (path.extension().and_then(|value| value.to_str()) == Some("lock")
                    && path
                        .file_stem()
                        .and_then(|value| value.to_str())
                        .is_some_and(|name| name.starts_with("ipc-v"))))
        {
            continue;
        }
        if !metadata.is_file()
            || metadata.len() > MAX_RECORD_BYTES
            || path.extension().and_then(|value| value.to_str()) != Some("json")
        {
            return Err(invalid());
        }
        let mut bytes = Vec::new();
        fs::File::open(&path)?
            .take(MAX_RECORD_BYTES + 1)
            .read_to_end(&mut bytes)?;
        if bytes.len() as u64 > MAX_RECORD_BYTES {
            return Err(invalid());
        }
        let record: DaemonRecord = serde_json::from_slice(&bytes).map_err(|_| invalid())?;
        if record.schema_version != DAEMON_RECORD_SCHEMA_VERSION
            || record.namespace.is_empty()
            || record.instance_id.is_empty()
            || (path != scoped_record_path(state_dir, &record)
                && path != record_path(state_dir, &record.namespace))
            || !namespaces.insert((record.namespace.clone(), record.state_location_id.clone()))
            || !instances.insert(record.instance_id.clone())
        {
            return Err(invalid());
        }
        records.push(record);
    }
    records.sort_by(|left, right| left.namespace.cmp(&right.namespace));
    Ok(records)
}

/// Remove a daemon registry record.
///
/// # Errors
///
/// Returns an error when removing the registry file fails for reasons other than not found.
pub fn remove_record_path(path: &Path) -> Result<(), DaemonLifecycleError> {
    match fs::remove_file(path) {
        Ok(()) => Ok(()),
        Err(source) if source.kind() == std::io::ErrorKind::NotFound => Ok(()),
        Err(source) => Err(DaemonLifecycleError::Io {
            path: path.to_path_buf(),
            source,
        }),
    }
}

/// Remove a daemon registry record when it still belongs to the provided instance.
///
/// # Errors
///
/// Returns an error when reading or removing the registry file fails for reasons
/// other than not found.
pub fn remove_record_if_instance(
    path: &Path,
    instance_id: &str,
) -> Result<(), DaemonLifecycleError> {
    let contents = match fs::read(path) {
        Ok(contents) => contents,
        Err(source) if source.kind() == std::io::ErrorKind::NotFound => return Ok(()),
        Err(source) => {
            return Err(DaemonLifecycleError::Io {
                path: path.to_path_buf(),
                source,
            });
        }
    };
    let record = serde_json::from_slice::<DaemonRecord>(&contents)?;
    if record.instance_id == instance_id {
        remove_record_path(path)?;
    }
    Ok(())
}

async fn probe_daemon_status(endpoint: &IpcEndpoint) -> Option<bcode_ipc::DaemonStatus> {
    let mut stream = bcode_ipc::LocalIpcStream::connect(endpoint).await.ok()?;
    let envelope = bcode_ipc::request_envelope(
        1,
        &bcode_ipc::Request::Hello {
            client_name: "bcode-daemon-readiness".to_owned(),
            runtime_context: None,
            daemon_namespace: daemon_namespace(),
            artifact_id: Some(bcode_ipc::ArtifactId::current()),
            build_fingerprint: BUILD_FINGERPRINT.to_owned(),
            state_location_id: Some(bcode_ipc::state_location_id()),
        },
    )
    .ok()?;
    bcode_ipc::send_envelope(&mut stream, &envelope)
        .await
        .ok()?;
    loop {
        let envelope = bcode_ipc::recv_envelope(&mut stream).await.ok()?;
        if envelope.kind != bcode_ipc::EnvelopeKind::Response || envelope.request_id != 1 {
            continue;
        }
        return match bcode_ipc::decode_response(&envelope.payload).ok()? {
            bcode_ipc::Response::Ok(bcode_ipc::ResponsePayload::Hello { daemon, .. }) => {
                Some(daemon)
            }
            _ => None,
        };
    }
}

/// Conservative classification of one daemon registry record.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum DaemonRecordClassification {
    /// Current-build daemon responded with the exact persisted identity.
    CurrentHealthy,
    /// Historical daemon responded with the exact persisted identity.
    HistoricalExactResponsive,
    /// Endpoint could not be decoded with the current protocol, but independent process evidence
    /// exactly identifies the historical daemon.
    HistoricalProcessVerifiedProtocolUnsupported,
    /// An endpoint responded, but its identity does not match this record.
    ResponsiveIdentityMismatch,
    /// Endpoint is unreachable and positive process evidence proves the recorded process is gone
    /// or replaced.
    UnreachableStale,
    /// Evidence is incomplete or ambiguous; callers must preserve the record.
    Unverifiable,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum ProcessIdentityEvidence {
    Exact,
    MissingOrReused,
    Unverifiable,
}

fn classify_daemon_record_evidence(
    record: &DaemonRecord,
    status: Option<&bcode_ipc::DaemonStatus>,
    endpoint_reachable: bool,
    process: ProcessIdentityEvidence,
) -> DaemonRecordClassification {
    status.map_or_else(
        || {
            if endpoint_reachable && process == ProcessIdentityEvidence::Exact {
                DaemonRecordClassification::HistoricalProcessVerifiedProtocolUnsupported
            } else if !endpoint_reachable && process == ProcessIdentityEvidence::MissingOrReused {
                DaemonRecordClassification::UnreachableStale
            } else {
                DaemonRecordClassification::Unverifiable
            }
        },
        |status| {
            if daemon_status_matches_record(record, status) {
                if record.is_current_namespace() {
                    DaemonRecordClassification::CurrentHealthy
                } else {
                    DaemonRecordClassification::HistoricalExactResponsive
                }
            } else {
                DaemonRecordClassification::ResponsiveIdentityMismatch
            }
        },
    )
}

fn daemon_status_matches_record(record: &DaemonRecord, status: &bcode_ipc::DaemonStatus) -> bool {
    status.instance_id == record.instance_id
        && status.namespace == record.namespace
        && status.protocol_version == record.protocol_version
        && status.artifact_id == record.artifact_id
        && status.build_fingerprint == record.build_fingerprint
        && status.executable_digest == record.executable_digest
        && status.storage_writer_epoch == record.storage_writer_epoch
        && status.pid == record.pid
        && status.started_at_unix_ms == record.started_at_unix_ms
}

fn process_identity_evidence(record: &DaemonRecord) -> ProcessIdentityEvidence {
    let (Some(pid), Some(expected_digest)) = (record.pid, record.executable_digest.as_deref())
    else {
        return ProcessIdentityEvidence::Unverifiable;
    };
    let pid = Pid::from_u32(pid);
    let mut system = System::new();
    system.refresh_processes(ProcessesToUpdate::Some(&[pid]), true);
    let Some(process) = system.process(pid) else {
        return ProcessIdentityEvidence::MissingOrReused;
    };
    let process_started_at_seconds = process.start_time();
    if process_started_at_seconds != record.started_at_unix_ms / 1_000 {
        return ProcessIdentityEvidence::MissingOrReused;
    }
    let Some(executable) = process.exe() else {
        return ProcessIdentityEvidence::Unverifiable;
    };
    // A digest mismatch only proves PID reuse when the pathname is content-addressed, because an
    // immutable image path can never be rewritten in place. A daemon spawned from a mutable
    // pathname (such as a `cargo build` output) keeps running its original bytes even after that
    // path is rebuilt, so a mismatch there is ambiguous rather than positive reuse evidence.
    let path_is_immutable = executable_path_matches_digest(executable, expected_digest);
    match executable_sha256(executable) {
        Ok(actual_digest) if actual_digest == expected_digest => ProcessIdentityEvidence::Exact,
        Ok(_) if path_is_immutable => ProcessIdentityEvidence::MissingOrReused,
        Ok(_) | Err(_) => ProcessIdentityEvidence::Unverifiable,
    }
}

/// Resolve a registry target only when it belongs to this process's pinned identity.
///
/// # Errors
/// Rejects foreign artifacts, executable bytes, state locations, and unsupported endpoints
/// before opening any transport connection.
pub fn pinned_daemon_endpoint(record: &DaemonRecord) -> Result<IpcEndpoint, DaemonLifecycleError> {
    let digest = initialize_artifact_bootstrap()?.digest()?;
    if record.artifact_id.as_ref() != Some(&bcode_ipc::ArtifactId::current())
        || !record.is_current_namespace()
        || record.executable_digest.as_deref() != Some(digest.as_str())
        || record.state_location_id.as_deref() != Some(bcode_ipc::state_location_id().as_str())
    {
        return Err(DaemonLifecycleError::Io {
            path: PathBuf::from("<daemon-target>"),
            source: std::io::Error::new(
                std::io::ErrorKind::PermissionDenied,
                "daemon target differs from the process-pinned executable or state location; use a separately started matching process",
            ),
        });
    }
    record
        .endpoint
        .to_ipc_endpoint()
        .ok_or_else(|| DaemonLifecycleError::Io {
            path: PathBuf::from("<daemon-target>"),
            source: std::io::Error::new(
                std::io::ErrorKind::Unsupported,
                "unsupported daemon endpoint",
            ),
        })
}

/// Classify one daemon record without mutating registry or endpoint state.
pub async fn classify_daemon_record(record: &DaemonRecord) -> DaemonRecordClassification {
    if pinned_daemon_endpoint(record).is_err() {
        // Foreign endpoints must never be contacted, even for liveness probes.
        // Process evidence can prove death; everything else remains conservative.
        return match process_identity_evidence(record) {
            ProcessIdentityEvidence::MissingOrReused => {
                DaemonRecordClassification::UnreachableStale
            }
            ProcessIdentityEvidence::Exact | ProcessIdentityEvidence::Unverifiable => {
                DaemonRecordClassification::Unverifiable
            }
        };
    }
    let endpoint = record.endpoint.to_ipc_endpoint();
    let status = if let Some(endpoint) = endpoint.as_ref() {
        tokio::time::timeout(Duration::from_millis(500), probe_daemon_status(endpoint))
            .await
            .ok()
            .flatten()
    } else {
        None
    };
    let endpoint_reachable = if status.is_some() {
        true
    } else if let Some(endpoint) = endpoint.as_ref() {
        tokio::time::timeout(
            Duration::from_millis(250),
            bcode_ipc::LocalIpcStream::connect(endpoint),
        )
        .await
        .is_ok_and(|result| result.is_ok())
    } else {
        false
    };
    let classification = classify_daemon_record_evidence(
        record,
        status.as_ref(),
        endpoint_reachable,
        process_identity_evidence(record),
    );
    tracing::debug!(
        namespace = record.namespace,
        instance_id = record.instance_id,
        pid = record.pid,
        classification = ?classification,
        "classified daemon registry record"
    );
    classification
}

async fn live_record_matches_instance(record: &DaemonRecord) -> bool {
    matches!(
        classify_daemon_record(record).await,
        DaemonRecordClassification::CurrentHealthy
            | DaemonRecordClassification::HistoricalExactResponsive
    )
}

/// Classify every decodable daemon record without mutating registry state.
pub async fn classified_records(
    state_dir: &Path,
) -> Vec<(PathBuf, DaemonRecord, DaemonRecordClassification)> {
    let mut classified = Vec::new();
    for (path, record) in read_records(state_dir) {
        let classification = classify_daemon_record(&record).await;
        classified.push((path, record, classification));
    }
    classified
}

/// Return whether any live or ambiguous daemon evidence exists for an artifact identity.
///
/// `identity` may be a daemon namespace, build fingerprint, or exact artifact ID. Cleanup
/// callers must refuse mutation when this returns true. Malformed registry records are treated
/// as ambiguous evidence and therefore fail closed.
///
/// # Errors
///
/// Returns an error if the registry directory cannot be read.
pub async fn namespace_has_live_or_ambiguous_evidence(
    state_dir: &Path,
    identity: &str,
) -> Result<bool, DaemonLifecycleError> {
    let dir = registry_dir(state_dir);
    let entries = match fs::read_dir(&dir) {
        Ok(entries) => entries,
        Err(source) if source.kind() == std::io::ErrorKind::NotFound => return Ok(false),
        Err(source) => return Err(DaemonLifecycleError::Io { path: dir, source }),
    };
    for entry in entries {
        let entry = entry.map_err(|source| DaemonLifecycleError::Io {
            path: dir.clone(),
            source,
        })?;
        let path = entry.path();
        if path.extension().and_then(|extension| extension.to_str()) != Some("json") {
            continue;
        }
        let contents = fs::read(&path).map_err(|source| DaemonLifecycleError::Io {
            path: path.clone(),
            source,
        })?;
        let Ok(record) = serde_json::from_slice::<DaemonRecord>(&contents) else {
            return Ok(true);
        };
        let identity_matches = record.namespace == identity
            || record.build_fingerprint == identity
            || record
                .artifact_id
                .as_ref()
                .is_some_and(|artifact_id| artifact_id.as_str() == identity);
        if !identity_matches {
            continue;
        }
        if !matches!(
            classify_daemon_record(&record).await,
            DaemonRecordClassification::UnreachableStale
        ) {
            return Ok(true);
        }
    }
    Ok(false)
}

/// Return daemon registry records whose IPC endpoints currently respond and whose server identity
/// matches the persisted instance record.
///
/// Stale, malformed, or endpoint-reused records are excluded. This function does not mutate
/// registry files.
pub async fn live_records(state_dir: &Path) -> Vec<(PathBuf, DaemonRecord)> {
    let mut live = Vec::new();
    for (path, record) in read_records(state_dir) {
        if live_record_matches_instance(&record).await {
            live.push((path, record));
        }
    }
    live
}

fn storage_writer_is_incompatible(record: &DaemonRecord, storage_writer_epoch: u32) -> bool {
    record.storage_writer_epoch != Some(storage_writer_epoch)
}

/// Return live daemons that do not advertise `storage_writer_epoch`.
///
/// Legacy records with no writer epoch are incompatible by default.
pub async fn incompatible_storage_writer_records(
    state_dir: &Path,
    storage_writer_epoch: u32,
) -> Vec<(PathBuf, DaemonRecord)> {
    classified_records(state_dir)
        .await
        .into_iter()
        .filter(|(_, record, classification)| {
            matches!(
                classification,
                DaemonRecordClassification::CurrentHealthy
                    | DaemonRecordClassification::HistoricalExactResponsive
            ) && storage_writer_is_incompatible(record, storage_writer_epoch)
        })
        .map(|(path, record, _)| (path, record))
        .collect()
}

/// Metadata published beside one immutable daemon image.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct DaemonImageMetadata {
    /// Metadata schema version.
    pub schema_version: u32,
    /// Exact produced-artifact identity embedded in the image.
    pub artifact_id: bcode_ipc::ArtifactId,
    /// SHA-256 digest of the image bytes.
    pub executable_digest: String,
}

const DAEMON_IMAGE_METADATA_SCHEMA_VERSION: u32 = 1;
const DAEMON_IMAGE_METADATA_FILE: &str = "image.json";
const DAEMON_IMAGE_CLEANUP_LOCK_FILE: &str = "daemon-images.lock";

/// Return the artifact-scoped directory that stores cached daemon executables.
#[must_use]
pub fn daemon_image_dir(state_dir: &Path) -> PathBuf {
    state_dir
        .join("daemon-images")
        .join(bcode_ipc::ArtifactId::current().as_str())
}

/// Bootstrap snapshot of the exact running artifact retained independently of its pathname.
#[derive(Debug)]
pub struct ArtifactBootstrap {
    artifact_id: bcode_ipc::ArtifactId,
    source_path: PathBuf,
    source: Mutex<fs::File>,
    executable_digest: OnceLock<String>,
}

static ARTIFACT_BOOTSTRAP: OnceLock<ArtifactBootstrap> = OnceLock::new();

/// Capture and retain the running artifact before its original pathname can be replaced.
///
/// Repeated calls return the same process bootstrap. The retained file handle remains readable
/// after Unix pathname replacement and prevents later lifecycle work from reopening a different
/// file at the original path.
///
/// # Panics
///
/// Panics only if the process-global bootstrap cannot be observed immediately after this function
/// successfully initializes it.
///
/// # Errors
///
/// Returns an error when the current executable cannot be located or opened.
pub fn initialize_artifact_bootstrap() -> Result<&'static ArtifactBootstrap, DaemonLifecycleError> {
    if let Some(bootstrap) = ARTIFACT_BOOTSTRAP.get() {
        return Ok(bootstrap);
    }
    let source_path = std::env::current_exe().map_err(|source| DaemonLifecycleError::Io {
        path: PathBuf::from("<current_exe>"),
        source,
    })?;
    let artifact_id = bcode_ipc::ArtifactId::current();
    let bootstrap = ArtifactBootstrap::open_verified(source_path, artifact_id)?;
    let _ = ARTIFACT_BOOTSTRAP.set(bootstrap);
    Ok(ARTIFACT_BOOTSTRAP
        .get()
        .expect("artifact bootstrap initialized"))
}

impl ArtifactBootstrap {
    fn open_verified(
        source_path: PathBuf,
        artifact_id: bcode_ipc::ArtifactId,
    ) -> Result<Self, DaemonLifecycleError> {
        let bootstrap = Self::open(source_path, artifact_id)?;
        let mut source = bootstrap
            .source
            .lock()
            .map_err(|_| DaemonLifecycleError::Io {
                path: bootstrap.source_path.clone(),
                source: std::io::Error::other("artifact bootstrap lock poisoned"),
            })?;
        let observed = bcode_ipc::ArtifactId::for_executable(&mut *source).map_err(|source| {
            DaemonLifecycleError::Io {
                path: bootstrap.source_path.clone(),
                source,
            }
        })?;
        if observed != bootstrap.artifact_id {
            return Err(DaemonLifecycleError::Io {
                path: bootstrap.source_path.clone(),
                source: std::io::Error::other("executable changed during artifact bootstrap"),
            });
        }
        drop(source);
        Ok(bootstrap)
    }

    fn open(
        source_path: PathBuf,
        artifact_id: bcode_ipc::ArtifactId,
    ) -> Result<Self, DaemonLifecycleError> {
        let source = fs::File::open(&source_path).map_err(|source| DaemonLifecycleError::Io {
            path: source_path.clone(),
            source,
        })?;
        Ok(Self {
            artifact_id,
            source_path,
            source: Mutex::new(source),
            executable_digest: OnceLock::new(),
        })
    }

    /// Return the exact embedded artifact identity.
    #[must_use]
    pub const fn artifact_id(&self) -> &bcode_ipc::ArtifactId {
        &self.artifact_id
    }

    /// Return the pathname observed at bootstrap for diagnostics only.
    #[must_use]
    pub fn source_path(&self) -> &Path {
        &self.source_path
    }

    /// Report whether the bootstrap source path still resolves to the retained artifact bytes.
    ///
    /// Daemons are never spawned from this mutable pathname, so this is diagnostic evidence only.
    #[cfg(all(unix, test))]
    fn source_path_still_current(&self) -> bool {
        let Ok(path_metadata) = fs::metadata(&self.source_path) else {
            return false;
        };
        let Ok(source) = self.source.lock() else {
            return false;
        };
        let Ok(source_metadata) = source.metadata() else {
            return false;
        };
        drop(source);
        path_metadata.dev() == source_metadata.dev() && path_metadata.ino() == source_metadata.ino()
    }

    fn digest(&self) -> Result<String, DaemonLifecycleError> {
        if let Some(digest) = self.executable_digest.get() {
            return Ok(digest.clone());
        }
        let mut source = self.source.lock().map_err(|_| DaemonLifecycleError::Io {
            path: self.source_path.clone(),
            source: std::io::Error::other("artifact bootstrap lock poisoned"),
        })?;
        source.rewind().map_err(|error| DaemonLifecycleError::Io {
            path: self.source_path.clone(),
            source: error,
        })?;
        let digest = sha256_reader(&mut *source, &self.source_path)?;
        drop(source);
        let _ = self.executable_digest.set(digest.clone());
        Ok(digest)
    }

    #[cfg(any(target_os = "linux", target_os = "macos"))]
    fn clone_to(&self, target: &Path) -> Result<bool, DaemonLifecycleError> {
        let source = self.source.lock().map_err(|_| DaemonLifecycleError::Io {
            path: self.source_path.clone(),
            source: std::io::Error::other("artifact bootstrap lock poisoned"),
        })?;
        try_clone_file_from_handle(&source, target).map_err(|source| DaemonLifecycleError::Io {
            path: target.to_path_buf(),
            source,
        })
    }

    fn copy_to(&self, target: &Path) -> Result<String, DaemonLifecycleError> {
        let mut source = self.source.lock().map_err(|_| DaemonLifecycleError::Io {
            path: self.source_path.clone(),
            source: std::io::Error::other("artifact bootstrap lock poisoned"),
        })?;
        source.rewind().map_err(|error| DaemonLifecycleError::Io {
            path: self.source_path.clone(),
            source: error,
        })?;
        let mut output = fs::File::create(target).map_err(|source| DaemonLifecycleError::Io {
            path: target.to_path_buf(),
            source,
        })?;
        let mut hasher = Sha256::new();
        let mut buffer = vec![0_u8; 1024 * 1024].into_boxed_slice();
        loop {
            let read = source
                .read(&mut buffer)
                .map_err(|source| DaemonLifecycleError::Io {
                    path: self.source_path.clone(),
                    source,
                })?;
            if read == 0 {
                break;
            }
            output
                .write_all(&buffer[..read])
                .map_err(|source| DaemonLifecycleError::Io {
                    path: target.to_path_buf(),
                    source,
                })?;
            hasher.update(&buffer[..read]);
        }
        drop(source);
        output
            .sync_all()
            .map_err(|source| DaemonLifecycleError::Io {
                path: target.to_path_buf(),
                source,
            })?;
        let digest = hex::encode(hasher.finalize());
        let _ = self.executable_digest.set(digest.clone());
        Ok(digest)
    }
}

#[cfg(target_os = "macos")]
fn try_clone_file_from_handle(source: &fs::File, target: &Path) -> std::io::Result<bool> {
    let target = std::ffi::CString::new(target.as_os_str().as_bytes())
        .map_err(|_| std::io::Error::new(std::io::ErrorKind::InvalidInput, "NUL in image path"))?;
    // SAFETY: `source` remains open for the call and `target` is a valid NUL-terminated path.
    let result =
        unsafe { libc::fclonefileat(source.as_raw_fd(), libc::AT_FDCWD, target.as_ptr(), 0) };
    if result == 0 {
        return Ok(true);
    }
    let error = std::io::Error::last_os_error();
    if error
        .raw_os_error()
        .is_some_and(|code| [libc::ENOTSUP, libc::EXDEV, libc::EINVAL].contains(&code))
    {
        return Ok(false);
    }
    Err(error)
}

#[cfg(target_os = "linux")]
fn try_clone_file_from_handle(source: &fs::File, target: &Path) -> std::io::Result<bool> {
    let output = fs::OpenOptions::new()
        .write(true)
        .create_new(true)
        .open(target)?;
    // SAFETY: both file descriptors remain valid for the duration of the ioctl call.
    let result = unsafe { libc::ioctl(output.as_raw_fd(), libc::FICLONE, source.as_raw_fd()) };
    if result == 0 {
        return Ok(true);
    }
    let error = std::io::Error::last_os_error();
    drop(output);
    let _ = fs::remove_file(target);
    if error.raw_os_error().is_some_and(|code| {
        [libc::ENOTTY, libc::EOPNOTSUPP, libc::EXDEV, libc::EINVAL].contains(&code)
    }) {
        return Ok(false);
    }
    Err(error)
}

fn sha256_reader(mut reader: impl Read, path: &Path) -> Result<String, DaemonLifecycleError> {
    let mut hasher = Sha256::new();
    let mut buffer = vec![0_u8; 1024 * 1024].into_boxed_slice();
    loop {
        let read = reader
            .read(&mut buffer)
            .map_err(|source| DaemonLifecycleError::Io {
                path: path.to_path_buf(),
                source,
            })?;
        if read == 0 {
            break;
        }
        hasher.update(&buffer[..read]);
    }
    Ok(hex::encode(hasher.finalize()))
}

/// Return the immutable cached executable path for one binary digest.
#[must_use]
pub fn cached_daemon_executable_path_for_digest(
    state_dir: &Path,
    executable_digest: &str,
) -> PathBuf {
    daemon_image_dir(state_dir)
        .join(executable_digest)
        .join(if cfg!(windows) { "bcode.exe" } else { "bcode" })
}

/// Return the cached executable path for the current process image.
///
/// # Errors
///
/// Returns an error when the current executable cannot be located or read.
pub fn current_cached_daemon_executable_path(
    state_dir: &Path,
) -> Result<PathBuf, DaemonLifecycleError> {
    let bootstrap = initialize_artifact_bootstrap()?;
    let digest = bootstrap.digest()?;
    Ok(cached_daemon_executable_path_for_digest(state_dir, &digest))
}

/// Return the lowercase SHA-256 digest of a file.
///
/// # Errors
///
/// Returns an error when the file cannot be opened or read.
pub fn executable_sha256(path: &Path) -> Result<String, DaemonLifecycleError> {
    let mut file = fs::File::open(path).map_err(|source| DaemonLifecycleError::Io {
        path: path.to_path_buf(),
        source,
    })?;
    sha256_reader(&mut file, path)
}

/// Return the current process executable and its byte digest.
///
/// # Errors
///
/// Returns an error when the executable cannot be located or read.
pub fn current_executable_identity() -> Result<(PathBuf, String), DaemonLifecycleError> {
    let bootstrap = initialize_artifact_bootstrap()?;
    Ok((bootstrap.source_path().to_path_buf(), bootstrap.digest()?))
}

fn cached_executable_digest(path: &Path) -> Option<&str> {
    path.parent()?.file_name()?.to_str()
}

/// Verify that an executable digest agrees with its content-addressed cache path.
#[must_use]
pub fn executable_path_matches_digest(path: &Path, executable_digest: &str) -> bool {
    path.parent()
        .and_then(Path::file_name)
        .and_then(|component| component.to_str())
        == Some(executable_digest)
}

fn daemon_image_metadata_path(executable: &Path) -> PathBuf {
    executable
        .parent()
        .expect("content-addressed daemon image has a parent")
        .join(DAEMON_IMAGE_METADATA_FILE)
}

fn read_daemon_image_metadata(
    executable: &Path,
) -> Result<Option<DaemonImageMetadata>, DaemonLifecycleError> {
    let metadata_path = daemon_image_metadata_path(executable);
    let contents = match fs::read(&metadata_path) {
        Ok(contents) => contents,
        Err(source) if source.kind() == std::io::ErrorKind::NotFound => return Ok(None),
        Err(source) => {
            return Err(DaemonLifecycleError::Io {
                path: metadata_path,
                source,
            });
        }
    };
    Ok(serde_json::from_slice(&contents).ok())
}

fn daemon_image_is_valid(
    executable: &Path,
    expected_digest: &str,
) -> Result<bool, DaemonLifecycleError> {
    if !executable.is_file() || executable_sha256(executable)? != expected_digest {
        return Ok(false);
    }
    let Some(metadata) = read_daemon_image_metadata(executable)? else {
        return Ok(false);
    };
    Ok(
        metadata.schema_version == DAEMON_IMAGE_METADATA_SCHEMA_VERSION
            && metadata.artifact_id == bcode_ipc::ArtifactId::current()
            && metadata.executable_digest == expected_digest,
    )
}

fn write_daemon_image_metadata(
    executable: &Path,
    digest: &str,
) -> Result<(), DaemonLifecycleError> {
    let metadata_path = daemon_image_metadata_path(executable);
    let metadata = DaemonImageMetadata {
        schema_version: DAEMON_IMAGE_METADATA_SCHEMA_VERSION,
        artifact_id: bcode_ipc::ArtifactId::current(),
        executable_digest: digest.to_owned(),
    };
    let contents = serde_json::to_vec_pretty(&metadata)?;
    fs::write(&metadata_path, contents).map_err(|source| DaemonLifecycleError::Io {
        path: metadata_path,
        source,
    })
}

fn remove_interrupted_image_publications(state_dir: &Path) -> Result<(), DaemonLifecycleError> {
    let artifact_dir = daemon_image_dir(state_dir);
    let entries = match fs::read_dir(&artifact_dir) {
        Ok(entries) => entries,
        Err(source) if source.kind() == std::io::ErrorKind::NotFound => return Ok(()),
        Err(source) => {
            return Err(DaemonLifecycleError::Io {
                path: artifact_dir,
                source,
            });
        }
    };
    for entry in entries {
        let entry = entry.map_err(|source| DaemonLifecycleError::Io {
            path: artifact_dir.clone(),
            source,
        })?;
        let path = entry.path();
        if entry
            .file_name()
            .to_string_lossy()
            .starts_with(".bcode.tmp-")
        {
            fs::remove_file(&path).map_err(|source| DaemonLifecycleError::Io { path, source })?;
            continue;
        }
        if !entry.file_type().is_ok_and(|file_type| file_type.is_dir()) {
            continue;
        }
        let child_entries = fs::read_dir(&path).map_err(|source| DaemonLifecycleError::Io {
            path: path.clone(),
            source,
        })?;
        for child in child_entries {
            let child = child.map_err(|source| DaemonLifecycleError::Io {
                path: path.clone(),
                source,
            })?;
            if child
                .file_name()
                .to_string_lossy()
                .starts_with(".bcode.tmp-")
            {
                let child_path = child.path();
                fs::remove_file(&child_path).map_err(|source| DaemonLifecycleError::Io {
                    path: child_path,
                    source,
                })?;
            }
        }
    }
    Ok(())
}

fn current_cached_daemon_image(
    state_dir: &Path,
) -> Result<Option<(PathBuf, String)>, DaemonLifecycleError> {
    let bootstrap = initialize_artifact_bootstrap()?;
    let digest = bootstrap.digest()?;
    let executable = cached_daemon_executable_path_for_digest(state_dir, &digest);
    if daemon_image_is_valid(&executable, &digest)? {
        Ok(Some((executable, digest)))
    } else {
        Ok(None)
    }
}

/// Verified immutable executable selected from the process bootstrap.
#[derive(Debug)]
pub struct PinnedBcodeExecutable {
    path: PathBuf,
    digest: String,
}

impl PinnedBcodeExecutable {
    /// Return the verified immutable launch path.
    #[must_use]
    pub fn path(&self) -> &Path {
        &self.path
    }

    /// Return the startup executable's SHA-256 digest.
    #[must_use]
    pub fn digest(&self) -> &str {
        &self.digest
    }

    /// Return the embedded process artifact identity.
    #[must_use]
    pub fn artifact_id(&self) -> bcode_ipc::ArtifactId {
        bcode_ipc::ArtifactId::current()
    }
}

static PINNED_EXECUTABLE: OnceLock<PinnedBcodeExecutable> = OnceLock::new();

/// Access the process-pinned Bcode executable, never a PATH-resolved replacement.
///
/// # Panics
/// Panics only if the process-global pin disappears after initialization.
///
/// # Errors
/// Returns an error if materialization or verification fails. A missing or modified
/// published image fails closed; this accessor never selects another executable.
pub fn pinned_bcode_executable() -> Result<&'static PinnedBcodeExecutable, DaemonLifecycleError> {
    if PINNED_EXECUTABLE.get().is_none() {
        let path = ensure_current_executable_cached_in_state(&bcode_config::default_state_dir())?;
        let digest = initialize_artifact_bootstrap()?.digest()?;
        let _ = PINNED_EXECUTABLE.set(PinnedBcodeExecutable { path, digest });
    }
    let executable = PINNED_EXECUTABLE
        .get()
        .expect("pinned executable initialized");
    verify_pinned_bcode_executable(executable)?;
    Ok(executable)
}

fn verify_pinned_bcode_executable(
    executable: &PinnedBcodeExecutable,
) -> Result<(), DaemonLifecycleError> {
    if !daemon_image_is_valid(executable.path(), executable.digest())? {
        return Err(DaemonLifecycleError::Io {
            path: executable.path.clone(),
            source: std::io::Error::new(
                std::io::ErrorKind::InvalidData,
                "pinned Bcode executable is missing or modified",
            ),
        });
    }
    Ok(())
}

/// Return the immutable process-pinned executable path.
///
/// # Errors
/// Returns an error if the pinned executable cannot be materialized or verified.
pub fn ensure_current_executable_cached() -> Result<PathBuf, DaemonLifecycleError> {
    Ok(pinned_bcode_executable()?.path().to_path_buf())
}

fn materialize_verified_daemon_image(
    bootstrap: &ArtifactBootstrap,
    source: &Path,
    temp: &Path,
) -> Result<(String, bool), DaemonLifecycleError> {
    #[cfg(not(unix))]
    let _ = source;
    #[cfg(any(target_os = "linux", target_os = "macos"))]
    let cloned = bootstrap.clone_to(temp)?;
    #[cfg(not(any(target_os = "linux", target_os = "macos")))]
    let cloned = false;
    tracing::debug!(
        target: "bcode_daemon_lifecycle::startup",
        cloned,
        "daemon image clone attempted"
    );
    let verify_started_at = std::time::Instant::now();
    let digest = if cloned {
        let digest = executable_sha256(temp)?;
        fs::OpenOptions::new()
            .read(true)
            .open(temp)
            .and_then(|file| file.sync_all())
            .map_err(|source| DaemonLifecycleError::Io {
                path: temp.to_path_buf(),
                source,
            })?;
        let _ = bootstrap.executable_digest.set(digest.clone());
        digest
    } else {
        bootstrap.copy_to(temp)?
    };
    tracing::debug!(
        target: "bcode_daemon_lifecycle::startup",
        elapsed_ms = verify_started_at.elapsed().as_millis(),
        cloned,
        "daemon image bytes verified and synchronized"
    );
    #[cfg(unix)]
    preserve_executable_permissions(source, temp)?;
    Ok((digest, cloned))
}

fn ensure_current_executable_cached_in_state(
    state_dir: &Path,
) -> Result<PathBuf, DaemonLifecycleError> {
    let bootstrap = initialize_artifact_bootstrap()?;
    let source = bootstrap.source_path().to_path_buf();
    remove_interrupted_image_publications(state_dir)?;
    if let Some((target, _digest)) = current_cached_daemon_image(state_dir)?
        && target != source
    {
        return Ok(target);
    }
    let temp_parent = daemon_image_dir(state_dir);
    fs::create_dir_all(&temp_parent).map_err(|source| DaemonLifecycleError::Io {
        path: temp_parent.clone(),
        source,
    })?;
    let temp = temp_parent.join(format!(".bcode.tmp-{}", std::process::id()));
    let materialize_started_at = std::time::Instant::now();
    let (digest, cloned) = materialize_verified_daemon_image(bootstrap, &source, &temp)?;
    let target = cached_daemon_executable_path_for_digest(state_dir, &digest);
    if target == source {
        let _ = fs::remove_file(&temp);
        return Ok(target);
    }
    let parent = target
        .parent()
        .expect("content-addressed daemon image has a parent");
    fs::create_dir_all(parent).map_err(|source| DaemonLifecycleError::Io {
        path: parent.to_path_buf(),
        source,
    })?;
    if target.exists() {
        if daemon_image_is_valid(&target, &digest)? {
            let _ = fs::remove_file(&temp);
            return Ok(target);
        }
        fs::remove_file(&target).map_err(|source| DaemonLifecycleError::Io {
            path: target.clone(),
            source,
        })?;
        let metadata = daemon_image_metadata_path(&target);
        if let Err(source) = fs::remove_file(&metadata)
            && source.kind() != std::io::ErrorKind::NotFound
        {
            return Err(DaemonLifecycleError::Io {
                path: metadata,
                source,
            });
        }
    }
    match fs::rename(&temp, &target) {
        Ok(()) => {
            write_daemon_image_metadata(&target, &digest)?;
            tracing::debug!(
                target: "bcode_daemon_lifecycle::startup",
                elapsed_ms = materialize_started_at.elapsed().as_millis(),
                cloned,
                "daemon image published"
            );
            Ok(target)
        }
        Err(_source_error) if daemon_image_is_valid(&target, &digest)? => {
            let _ = fs::remove_file(&temp);
            Ok(target)
        }
        Err(source_error) => {
            let _ = fs::remove_file(&temp);
            Err(DaemonLifecycleError::Io {
                path: target,
                source: source_error,
            })
        }
    }
}

#[cfg(unix)]
fn preserve_executable_permissions(
    source: &Path,
    target: &Path,
) -> Result<(), DaemonLifecycleError> {
    let mode = fs::metadata(source)
        .map_err(|source_error| DaemonLifecycleError::Io {
            path: source.to_path_buf(),
            source: source_error,
        })?
        .permissions()
        .mode();
    let mut permissions = fs::metadata(target)
        .map_err(|source_error| DaemonLifecycleError::Io {
            path: target.to_path_buf(),
            source: source_error,
        })?
        .permissions();
    permissions.set_mode(mode | 0o700);
    fs::set_permissions(target, permissions).map_err(|source_error| DaemonLifecycleError::Io {
        path: target.to_path_buf(),
        source: source_error,
    })
}

fn daemon_image_cleanup_lock_path(state_dir: &Path) -> PathBuf {
    state_dir
        .join("daemons")
        .join(DAEMON_IMAGE_CLEANUP_LOCK_FILE)
}

fn open_daemon_image_cleanup_lock(state_dir: &Path) -> Result<fs::File, DaemonLifecycleError> {
    let path = daemon_image_cleanup_lock_path(state_dir);
    if let Some(parent) = path.parent() {
        fs::create_dir_all(parent).map_err(|source| DaemonLifecycleError::Io {
            path: parent.to_path_buf(),
            source,
        })?;
    }
    fs::OpenOptions::new()
        .read(true)
        .write(true)
        .create(true)
        .truncate(false)
        .open(&path)
        .map_err(|source| DaemonLifecycleError::Io { path, source })
}

struct DaemonImageUseGuard {
    file: fs::File,
}

impl DaemonImageUseGuard {
    fn acquire(state_dir: &Path) -> Result<Self, DaemonLifecycleError> {
        let file = open_daemon_image_cleanup_lock(state_dir)?;
        file.lock_shared()
            .map_err(|source| DaemonLifecycleError::Io {
                path: daemon_image_cleanup_lock_path(state_dir),
                source,
            })?;
        Ok(Self { file })
    }
}

impl Drop for DaemonImageUseGuard {
    fn drop(&mut self) {
        let _ = self.file.unlock();
    }
}

struct DaemonImageCleanupGuard {
    file: fs::File,
}

impl Drop for DaemonImageCleanupGuard {
    fn drop(&mut self) {
        // Closing alone can leave the lock held by a duplicated/inherited handle.
        let _ = self.file.unlock();
    }
}

fn try_acquire_daemon_image_cleanup_lock(
    state_dir: &Path,
) -> Result<Option<DaemonImageCleanupGuard>, DaemonLifecycleError> {
    let file = open_daemon_image_cleanup_lock(state_dir)?;
    match file.try_lock() {
        Ok(()) => Ok(Some(DaemonImageCleanupGuard { file })),
        Err(std::fs::TryLockError::WouldBlock) => Ok(None),
        Err(std::fs::TryLockError::Error(source)) => Err(DaemonLifecycleError::Io {
            path: daemon_image_cleanup_lock_path(state_dir),
            source,
        }),
    }
}

fn retained_daemon_image_paths(
    state_dir: &Path,
) -> Result<(std::collections::BTreeSet<PathBuf>, bool), DaemonLifecycleError> {
    let registry = registry_dir(state_dir);
    let entries = match fs::read_dir(&registry) {
        Ok(entries) => entries,
        Err(source) if source.kind() == std::io::ErrorKind::NotFound => {
            return Ok((std::collections::BTreeSet::new(), false));
        }
        Err(source) => {
            return Err(DaemonLifecycleError::Io {
                path: registry,
                source,
            });
        }
    };
    let mut retained = std::collections::BTreeSet::new();
    let mut ambiguous_record_evidence = false;
    for entry in entries {
        let entry = entry.map_err(|source| DaemonLifecycleError::Io {
            path: registry.clone(),
            source,
        })?;
        let path = entry.path();
        if path.extension().and_then(|extension| extension.to_str()) != Some("json") {
            continue;
        }
        let contents = fs::read(&path).map_err(|source| DaemonLifecycleError::Io {
            path: path.clone(),
            source,
        })?;
        let Ok(record) = serde_json::from_slice::<DaemonRecord>(&contents) else {
            ambiguous_record_evidence = true;
            continue;
        };
        if !(1..=DAEMON_RECORD_SCHEMA_VERSION).contains(&record.schema_version) {
            ambiguous_record_evidence = true;
            continue;
        }
        if let Some(executable_path) = record.executable_path {
            retained.insert(executable_path);
        }
    }
    Ok((retained, ambiguous_record_evidence))
}

/// Remove cached daemon images that are not referenced by daemon records or the current build.
///
/// Cleanup fails closed when any registry record is malformed or has an unknown schema because
/// that record may contain image-retention evidence this build cannot safely interpret.
///
/// # Errors
///
/// Returns an error when reading registry evidence or removing image directories fails.
pub fn cleanup_stale_daemon_images(state_dir: &Path) -> Result<usize, DaemonLifecycleError> {
    cleanup_stale_daemon_images_retaining_artifacts(state_dir, &std::collections::BTreeSet::new())
}

/// Remove cached daemon images, preserving images for artifacts callers still require.
///
/// `retained_artifact_ids` names artifact identities that must keep their cached images even
/// without a live daemon record. Execution authority for durable work is fenced to an exact
/// artifact identity, so deleting that artifact's image would permanently strand the work: no
/// other artifact may legitimately claim ownership, and the owning artifact can no longer be
/// launched. Retaining the image keeps a matching daemon startable so recovery stays possible.
///
/// Cleanup fails closed when any registry record is malformed or has an unknown schema because
/// that record may contain image-retention evidence this build cannot safely interpret.
///
/// # Errors
///
/// Returns an error when reading registry evidence or removing image directories fails.
pub fn cleanup_stale_daemon_images_retaining_artifacts(
    state_dir: &Path,
    retained_artifact_ids: &std::collections::BTreeSet<String>,
) -> Result<usize, DaemonLifecycleError> {
    let Some(_cleanup_lock) = try_acquire_daemon_image_cleanup_lock(state_dir)? else {
        return Ok(0);
    };
    let root = state_dir.join("daemon-images");
    let namespace_entries = match fs::read_dir(&root) {
        Ok(entries) => entries,
        Err(source) if source.kind() == std::io::ErrorKind::NotFound => return Ok(0),
        Err(source) => return Err(DaemonLifecycleError::Io { path: root, source }),
    };
    let (mut retained, ambiguous_record_evidence) = retained_daemon_image_paths(state_dir)?;
    if ambiguous_record_evidence {
        return Ok(0);
    }
    retained.insert(current_cached_daemon_executable_path(state_dir)?);
    let mut removed = 0;
    for namespace_entry in namespace_entries.flatten() {
        if !namespace_entry
            .file_type()
            .is_ok_and(|file_type| file_type.is_dir())
        {
            continue;
        }
        // The namespace directory name is the artifact identity that owns these images. Durable
        // work fenced to this artifact keeps every image under it, because recovery requires
        // launching a daemon of exactly this identity.
        if namespace_entry
            .file_name()
            .to_str()
            .is_some_and(|artifact_id| retained_artifact_ids.contains(artifact_id))
        {
            continue;
        }
        let Ok(image_entries) = fs::read_dir(namespace_entry.path()) else {
            continue;
        };
        for image_entry in image_entries.flatten() {
            let path = image_entry.path();
            let candidate = if image_entry
                .file_type()
                .is_ok_and(|file_type| file_type.is_dir())
            {
                path.join(if cfg!(windows) { "bcode.exe" } else { "bcode" })
            } else {
                path.clone()
            };
            if retained.contains(&candidate) {
                continue;
            }
            if image_entry
                .file_type()
                .is_ok_and(|file_type| file_type.is_dir())
            {
                fs::remove_dir_all(&path)
            } else {
                fs::remove_file(&path)
            }
            .map_err(|source| DaemonLifecycleError::Io {
                path: path.clone(),
                source,
            })?;
            removed += 1;
        }
        if fs::read_dir(namespace_entry.path()).is_ok_and(|mut entries| entries.next().is_none()) {
            let _ = fs::remove_dir(namespace_entry.path());
        }
    }
    Ok(removed)
}

/// Return Unix time in milliseconds.
///
/// # Errors
///
/// Returns an error when the system clock is before the Unix epoch.
pub fn unix_time_millis() -> Result<u64, DaemonLifecycleError> {
    let duration = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map_err(|_| DaemonLifecycleError::Clock)?;
    Ok(duration.as_millis().try_into().unwrap_or(u64::MAX))
}

fn endpoint_record(endpoint: &IpcEndpoint) -> DaemonEndpointRecord {
    if let Some(path) = endpoint.as_unix_socket() {
        return DaemonEndpointRecord::UnixSocket {
            path: path.to_path_buf(),
        };
    }
    if let Some(name) = endpoint.as_windows_named_pipe() {
        return DaemonEndpointRecord::WindowsNamedPipe {
            name: name.to_owned(),
        };
    }
    DaemonEndpointRecord::Unknown {
        debug: format!("{endpoint:?}"),
    }
}

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

    #[test]
    fn strict_inventory_accepts_coordination_locks_but_rejects_unknown_files() {
        let root = tempfile::tempdir().unwrap();
        let directory = registry_dir(root.path());
        fs::create_dir(&directory).unwrap();
        let record = record_with_writer_epoch(Some(10));
        fs::write(
            directory.join("test.json"),
            serde_json::to_vec(&record).unwrap(),
        )
        .unwrap();
        fs::write(directory.join("daemon-images.lock"), b"").unwrap();
        fs::write(directory.join("ipc-v49-fixture.lock"), b"1234").unwrap();
        assert_eq!(read_records_strict(root.path(), 16).unwrap(), vec![record]);
        assert!(read_records_strict(root.path(), 2).is_err());
        fs::write(directory.join("unexpected"), b"").unwrap();
        assert!(read_records_strict(root.path(), 16).is_err());
    }

    #[test]
    fn registry_preserves_distinct_config_scopes_in_one_state_root() {
        let directory = tempfile::tempdir().unwrap();
        let mut first = record_with_writer_epoch(Some(10));
        first.state_location_id = Some("scope-a".into());
        first.instance_id = "instance-a".into();
        let mut second = first.clone();
        second.state_location_id = Some("scope-b".into());
        second.instance_id = "instance-b".into();
        let first_path = write_record(directory.path(), &first).unwrap();
        let second_path = write_record(directory.path(), &second).unwrap();
        assert_ne!(first_path, second_path);
        assert_eq!(read_records_strict(directory.path(), 16).unwrap().len(), 2);
        remove_record_path(&first_path).unwrap();
        assert_eq!(
            read_records_strict(directory.path(), 16).unwrap(),
            vec![second]
        );
    }

    #[test]
    fn strict_inventory_never_converts_damage_or_partial_scans_to_absence() {
        let root = tempfile::tempdir().expect("root");
        assert!(read_records_strict(root.path(), 16).is_err());
        let directory = registry_dir(root.path());
        fs::create_dir(&directory).expect("registry");
        assert!(
            read_records_strict(root.path(), 16)
                .expect("empty")
                .is_empty()
        );
        let record = record_with_writer_epoch(Some(10));
        fs::write(
            directory.join("test.json"),
            serde_json::to_vec(&record).expect("json"),
        )
        .expect("record");
        assert_eq!(
            read_records_strict(root.path(), 1).expect("record"),
            vec![record]
        );
        fs::write(directory.join("broken.json"), b"broken").expect("broken");
        assert!(read_records_strict(root.path(), 16).is_err());
        assert!(read_records_strict(root.path(), 1).is_err());
        assert_eq!(
            read_records(root.path()).len(),
            1,
            "best-effort cleanup remains separate"
        );
        fs::remove_file(directory.join("broken.json")).expect("fixture cleanup");
        let mut future = record_with_writer_epoch(Some(10));
        future.schema_version += 1;
        fs::write(
            directory.join("test.json"),
            serde_json::to_vec(&future).expect("json"),
        )
        .expect("future");
        assert!(read_records_strict(root.path(), 16).is_err());
    }

    #[test]
    fn strict_inventory_rejects_duplicate_instances_and_oversized_records() {
        let root = tempfile::tempdir().expect("root");
        let directory = registry_dir(root.path());
        fs::create_dir(&directory).expect("registry");
        let first = record_with_writer_epoch(Some(10));
        let mut second = first.clone();
        second.namespace = "other".into();
        fs::write(
            directory.join("test.json"),
            serde_json::to_vec(&first).expect("json"),
        )
        .expect("first");
        fs::write(
            directory.join("other.json"),
            serde_json::to_vec(&second).expect("json"),
        )
        .expect("duplicate");
        assert!(read_records_strict(root.path(), 16).is_err());
        fs::remove_file(directory.join("other.json")).expect("fixture cleanup");
        fs::write(directory.join("test.json"), vec![b' '; 65537]).expect("oversized");
        assert!(read_records_strict(root.path(), 16).is_err());
    }

    fn record_with_writer_epoch(storage_writer_epoch: Option<u32>) -> DaemonRecord {
        DaemonRecord {
            schema_version: DAEMON_RECORD_SCHEMA_VERSION,
            namespace: "test".to_string(),
            protocol_version: u32::from(CURRENT_PROTOCOL_VERSION),
            artifact_id: Some(bcode_ipc::ArtifactId::current()),
            build_fingerprint: "test-build".to_string(),
            storage_writer_epoch,
            state_location_id: Some(bcode_ipc::state_location_id()),
            pid: Some(std::process::id()),
            endpoint: DaemonEndpointRecord::Unknown {
                debug: "test".to_string(),
            },
            log_path: PathBuf::from("test.log"),
            executable_path: None,
            executable_digest: None,
            started_at_unix_ms: 0,
            last_seen_unix_ms: 0,
            instance_id: "test-instance".to_string(),
        }
    }

    fn stale_image_path(state_dir: &Path, namespace: &str, digest: &str) -> PathBuf {
        state_dir
            .join("daemon-images")
            .join(namespace)
            .join(digest)
            .join(if cfg!(windows) { "bcode.exe" } else { "bcode" })
    }

    fn write_test_image(path: &Path) {
        fs::create_dir_all(path.parent().expect("image parent")).expect("image directory");
        fs::write(path, b"stale daemon image").expect("image");
    }

    #[test]
    fn executable_digest_preserves_standard_sha256_encoding() {
        let temp = tempfile::tempdir().unwrap();
        let path = temp.path().join("fixture");
        for (bytes, expected) in [
            (
                b"".as_slice(),
                "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855",
            ),
            (
                b"abc".as_slice(),
                "ba7816bf8f01cfea414140de5dae2223b00361a396177a9cb410ff61f20015ad",
            ),
        ] {
            fs::write(&path, bytes).unwrap();
            assert_eq!(executable_sha256(&path).unwrap(), expected);
        }
    }

    #[test]
    fn readiness_identity_requires_pinned_executable_digest() {
        let matching = bcode_ipc::DaemonStatus {
            namespace: daemon_namespace(),
            protocol_version: u32::from(CURRENT_PROTOCOL_VERSION),
            artifact_id: Some(bcode_ipc::ArtifactId::current()),
            build_fingerprint: BUILD_FINGERPRINT.to_owned(),
            executable_digest: Some(initialize_artifact_bootstrap().unwrap().digest().unwrap()),
            storage_writer_epoch: Some(bcode_ipc::CURRENT_SESSION_STORAGE_WRITER_EPOCH),
            session_event_schema_version: Some(
                bcode_session_models::CURRENT_SESSION_EVENT_SCHEMA_VERSION,
            ),
            ..bcode_ipc::DaemonStatus::default()
        };
        assert!(daemon_status_matches_current_executable(&matching));
        assert!(!daemon_status_matches_current_executable(
            &bcode_ipc::DaemonStatus {
                executable_digest: Some("different-bytes".into()),
                ..matching.clone()
            }
        ));
        assert!(!daemon_status_matches_current_executable(
            &bcode_ipc::DaemonStatus {
                storage_writer_epoch: matching.storage_writer_epoch.map(|epoch| epoch + 1),
                ..matching.clone()
            }
        ));
        assert!(!daemon_status_matches_current_executable(
            &bcode_ipc::DaemonStatus {
                session_event_schema_version: matching
                    .session_event_schema_version
                    .map(|schema| schema + 1),
                ..matching
            }
        ));
    }

    #[test]
    fn daemon_image_cleanup_releases_lock_with_a_duplicated_handle() {
        let state_dir = std::env::temp_dir().join(format!(
            "bcode-daemon-image-cleanup-release-test-{}-{}",
            std::process::id(),
            unix_time_millis().expect("time")
        ));
        let guard = try_acquire_daemon_image_cleanup_lock(&state_dir)
            .expect("lock")
            .expect("acquired");
        let duplicate = guard.file.try_clone().expect("duplicate handle");
        assert!(
            try_acquire_daemon_image_cleanup_lock(&state_dir)
                .expect("contended lock")
                .is_none()
        );
        drop(guard);
        let next = try_acquire_daemon_image_cleanup_lock(&state_dir)
            .expect("released lock")
            .expect("reacquired despite duplicate");
        drop(next);
        drop(duplicate);
        fs::remove_dir_all(state_dir).expect("state cleanup");
    }

    #[test]
    fn daemon_image_cleanup_skips_while_any_artifact_start_uses_images() {
        let state_dir = std::env::temp_dir().join(format!(
            "bcode-daemon-image-use-lock-test-{}-{}",
            std::process::id(),
            unix_time_millis().expect("time")
        ));
        let stale_image = stale_image_path(&state_dir, "historical", "in-use");
        write_test_image(&stale_image);
        let use_guard = DaemonImageUseGuard::acquire(&state_dir).expect("image use guard");

        assert_eq!(cleanup_stale_daemon_images(&state_dir).expect("cleanup"), 0);
        assert!(stale_image.exists());

        drop(use_guard);
        assert_eq!(cleanup_stale_daemon_images(&state_dir).expect("cleanup"), 1);
        assert!(!stale_image.exists());
        fs::remove_dir_all(state_dir).expect("state cleanup");
    }

    #[test]
    fn daemon_image_cleanup_retains_images_for_artifacts_owning_durable_work() {
        let state_dir = std::env::temp_dir().join(format!(
            "bcode-daemon-image-retention-test-{}-{}",
            std::process::id(),
            unix_time_millis().expect("time")
        ));
        let fenced_image = stale_image_path(&state_dir, "artifact-with-work", "digest");
        let unreferenced_image = stale_image_path(&state_dir, "artifact-without-work", "digest");
        write_test_image(&fenced_image);
        write_test_image(&unreferenced_image);

        let retained = std::collections::BTreeSet::from(["artifact-with-work".to_owned()]);
        assert_eq!(
            cleanup_stale_daemon_images_retaining_artifacts(&state_dir, &retained)
                .expect("cleanup"),
            1
        );
        // Deleting this image would strand the fenced work permanently.
        assert!(fenced_image.exists());
        assert!(!unreferenced_image.exists());
        assert!(artifact_image_is_available(
            &state_dir,
            "artifact-with-work"
        ));
        assert!(!artifact_image_is_available(
            &state_dir,
            "artifact-without-work"
        ));

        // Once the work no longer requires the artifact, its image becomes collectable.
        assert_eq!(
            cleanup_stale_daemon_images_retaining_artifacts(
                &state_dir,
                &std::collections::BTreeSet::new()
            )
            .expect("cleanup"),
            1
        );
        assert!(!fenced_image.exists());
        fs::remove_dir_all(state_dir).expect("state cleanup");
    }

    #[test]
    fn registered_retained_artifact_providers_contribute_retention_evidence() {
        fn provider() -> std::collections::BTreeSet<String> {
            std::collections::BTreeSet::from(["provider-artifact".to_owned()])
        }

        register_retained_artifact_provider(provider);

        assert!(registered_retained_artifact_ids().contains("provider-artifact"));
    }

    #[test]
    fn daemon_image_cleanup_retains_record_references_and_removes_only_unreferenced_images() {
        let state_dir = std::env::temp_dir().join(format!(
            "bcode-daemon-image-cleanup-test-{}-{}",
            std::process::id(),
            unix_time_millis().expect("time")
        ));
        let retained_image = stale_image_path(&state_dir, "historical", "retained");
        let stale_image = stale_image_path(&state_dir, "historical", "stale");
        write_test_image(&retained_image);
        write_test_image(&stale_image);
        let record = DaemonRecord {
            namespace: "historical-cleanup-record".to_owned(),
            executable_path: Some(retained_image.clone()),
            ..record_with_writer_epoch(Some(2))
        };
        let record_path = write_record(&state_dir, &record).expect("record");

        assert_eq!(cleanup_stale_daemon_images(&state_dir).expect("cleanup"), 1);
        assert!(retained_image.exists());
        assert!(!stale_image.exists());

        remove_record_path(&record_path).expect("remove record");
        assert_eq!(cleanup_stale_daemon_images(&state_dir).expect("cleanup"), 1);
        assert!(!retained_image.exists());
        fs::remove_dir_all(state_dir).expect("state cleanup");
    }

    #[test]
    fn daemon_image_cleanup_fails_closed_for_ambiguous_registry_evidence() {
        for (case, contents) in [
            ("malformed", b"not valid json".to_vec()),
            (
                "future-schema",
                serde_json::to_vec(&DaemonRecord {
                    schema_version: DAEMON_RECORD_SCHEMA_VERSION + 1,
                    namespace: "future-record".to_owned(),
                    ..record_with_writer_epoch(Some(2))
                })
                .expect("future record"),
            ),
        ] {
            let state_dir = std::env::temp_dir().join(format!(
                "bcode-daemon-image-ambiguous-{case}-{}-{}",
                std::process::id(),
                unix_time_millis().expect("time")
            ));
            let stale_image = stale_image_path(&state_dir, "historical", "ambiguous");
            write_test_image(&stale_image);
            let registry = registry_dir(&state_dir);
            fs::create_dir_all(&registry).expect("registry");
            fs::write(registry.join("ambiguous.json"), contents).expect("ambiguous record");

            assert_eq!(cleanup_stale_daemon_images(&state_dir).expect("cleanup"), 0);
            assert!(stale_image.exists());
            fs::remove_dir_all(state_dir).expect("state cleanup");
        }
    }

    #[test]
    fn current_process_identity_evidence_rejects_pid_reuse_and_accepts_exact_record() {
        let (executable_path, executable_digest) = current_executable_identity().expect("identity");
        let mut system = System::new();
        let pid = Pid::from_u32(std::process::id());
        system.refresh_processes(ProcessesToUpdate::Some(&[pid]), true);
        let process = system.process(pid).expect("current process");
        let record = DaemonRecord {
            pid: Some(std::process::id()),
            executable_path: Some(executable_path),
            executable_digest: Some(executable_digest),
            started_at_unix_ms: process.start_time().saturating_mul(1_000),
            ..record_with_writer_epoch(Some(2))
        };
        assert_eq!(
            process_identity_evidence(&record),
            ProcessIdentityEvidence::Exact
        );
        assert_eq!(
            process_identity_evidence(&DaemonRecord {
                started_at_unix_ms: record.started_at_unix_ms.saturating_add(1_000),
                ..record.clone()
            }),
            ProcessIdentityEvidence::MissingOrReused
        );
        // The test binary runs from a mutable pathname, so a digest mismatch is ambiguous and
        // must not be reported as positive PID-reuse evidence.
        assert_eq!(
            process_identity_evidence(&DaemonRecord {
                executable_digest: Some("reused-image".to_owned()),
                ..record
            }),
            ProcessIdentityEvidence::Unverifiable
        );
    }

    #[test]
    fn rebuilt_mutable_executable_path_is_ambiguous_and_preserves_the_record() {
        // A rebuilt source path must never be classified as stale while the daemon still answers,
        // so a live daemon spawned from `target/` keeps serving its own artifact after a rebuild.
        let current = DaemonRecord {
            namespace: bcode_ipc::daemon_namespace(),
            ..record_with_writer_epoch(Some(2))
        };
        assert_eq!(
            classify_daemon_record_evidence(
                &current,
                None,
                true,
                ProcessIdentityEvidence::Unverifiable,
            ),
            DaemonRecordClassification::Unverifiable
        );
        assert_eq!(
            classify_daemon_record_evidence(
                &current,
                None,
                false,
                ProcessIdentityEvidence::Unverifiable,
            ),
            DaemonRecordClassification::Unverifiable
        );
    }

    #[test]
    fn stale_plugin_state_cleanup_is_namespace_confined_and_symlink_safe() {
        let state = tempfile::tempdir().expect("state");
        let plugins = state.path().join("derived/plugins");
        let stale_root = plugins.join("ipc-v1-stale");
        let retained = plugins.join("ipc-v1-retained");
        fs::create_dir_all(&stale_root).expect("stale state");
        fs::create_dir_all(&retained).expect("retained state");
        fs::write(stale_root.join("provider.json"), b"derived").expect("provider state");

        remove_stale_plugin_state(state.path(), "ipc-v1-stale").expect("cleanup stale state");
        assert!(!stale_root.exists());
        assert!(retained.exists());
        remove_stale_plugin_state(state.path(), "../ipc-v1-retained")
            .expect("reject hostile namespace");
        assert!(retained.exists());

        #[cfg(unix)]
        {
            use std::os::unix::fs::symlink;
            let external = tempfile::tempdir().expect("external");
            let link = plugins.join("ipc-v1-link");
            symlink(external.path(), &link).expect("namespace symlink");
            remove_stale_plugin_state(state.path(), "ipc-v1-link")
                .expect("symlink cleanup fails closed");
            assert!(link.exists());
            assert!(external.path().exists());
        }
    }

    #[test]
    fn daemon_launch_preserves_exclusive_config_as_one_argument() {
        let mut command = tokio::process::Command::new("bcode");
        let path = Path::new("trusted directory/config ; $(ignored).toml");
        super::configure_daemon_config(&mut command, Some(path));
        command.args(["server", "run"]);
        let args: Vec<_> = command.as_std().get_args().collect();
        assert_eq!(
            args,
            vec![
                std::ffi::OsStr::new("--config-only"),
                path.as_os_str(),
                std::ffi::OsStr::new("server"),
                std::ffi::OsStr::new("run"),
            ]
        );
        let mut ordinary = tokio::process::Command::new("bcode");
        super::configure_daemon_config(&mut ordinary, None);
        assert_eq!(ordinary.as_std().get_args().count(), 0);
    }

    #[test]
    fn daemon_config_home_reproduces_the_resolved_config_directory() {
        // A spawned daemon resolves `<XDG_CONFIG_HOME>/bcode`, so the parent must be
        // passed for the child to land on the same directory the client resolved.
        assert_eq!(
            super::config_home_for_daemon(Path::new("/home/user/.config/bcode")),
            PathBuf::from("/home/user/.config")
        );
        assert_eq!(
            super::config_home_for_daemon(Path::new("/opt/custom/bcode")),
            PathBuf::from("/opt/custom")
        );
        // A directory that is not the expected `bcode` leaf is passed through rather
        // than guessed, so a mismatch surfaces at the handshake instead of silently
        // redirecting the daemon's config.
        assert_eq!(
            super::config_home_for_daemon(Path::new("/tmp/unexpected")),
            PathBuf::from("/tmp/unexpected")
        );
    }

    #[test]
    fn daemon_record_identity_fields_are_independently_validated() {
        let record = record_with_writer_epoch(Some(2));
        let exact = bcode_ipc::DaemonStatus {
            namespace: record.namespace.clone(),
            protocol_version: record.protocol_version,
            artifact_id: record.artifact_id.clone(),
            build_fingerprint: record.build_fingerprint.clone(),
            executable_digest: record.executable_digest.clone(),
            storage_writer_epoch: record.storage_writer_epoch,
            state_location_id: record.state_location_id.clone(),
            session_event_schema_version: None,
            pid: record.pid,
            instance_id: record.instance_id.clone(),
            started_at_unix_ms: record.started_at_unix_ms,
        };
        assert!(daemon_status_matches_record(&record, &exact));

        let cases = [
            bcode_ipc::DaemonStatus {
                instance_id: "other-instance".to_owned(),
                ..exact.clone()
            },
            bcode_ipc::DaemonStatus {
                namespace: "other-namespace".to_owned(),
                ..exact.clone()
            },
            bcode_ipc::DaemonStatus {
                protocol_version: exact.protocol_version.saturating_add(1),
                ..exact.clone()
            },
            bcode_ipc::DaemonStatus {
                artifact_id: Some(
                    bcode_ipc::ArtifactId::parse("other-artifact")
                        .expect("other artifact identity"),
                ),
                ..exact.clone()
            },
            bcode_ipc::DaemonStatus {
                build_fingerprint: "other-build".to_owned(),
                ..exact.clone()
            },
            bcode_ipc::DaemonStatus {
                executable_digest: Some("other-digest".to_owned()),
                ..exact.clone()
            },
            bcode_ipc::DaemonStatus {
                storage_writer_epoch: exact.storage_writer_epoch.map(|epoch| epoch + 1),
                ..exact.clone()
            },
            bcode_ipc::DaemonStatus {
                pid: exact.pid.map(|pid| pid.saturating_add(1)),
                ..exact.clone()
            },
            bcode_ipc::DaemonStatus {
                started_at_unix_ms: exact.started_at_unix_ms.saturating_add(1),
                ..exact
            },
        ];
        for status in cases {
            assert!(!daemon_status_matches_record(&record, &status));
        }
    }

    #[test]
    fn daemon_record_classification_preserves_historical_and_ambiguous_evidence() {
        let current = record_with_writer_epoch(Some(2));
        let exact = bcode_ipc::DaemonStatus {
            namespace: current.namespace.clone(),
            protocol_version: current.protocol_version,
            artifact_id: current.artifact_id.clone(),
            build_fingerprint: current.build_fingerprint.clone(),
            executable_digest: current.executable_digest.clone(),
            storage_writer_epoch: current.storage_writer_epoch,
            state_location_id: current.state_location_id.clone(),
            session_event_schema_version: None,
            pid: current.pid,
            instance_id: current.instance_id.clone(),
            started_at_unix_ms: current.started_at_unix_ms,
        };
        assert_eq!(
            classify_daemon_record_evidence(
                &current,
                Some(&exact),
                true,
                ProcessIdentityEvidence::Exact,
            ),
            DaemonRecordClassification::HistoricalExactResponsive
        );
        assert_eq!(
            classify_daemon_record_evidence(&current, None, true, ProcessIdentityEvidence::Exact,),
            DaemonRecordClassification::HistoricalProcessVerifiedProtocolUnsupported
        );
        assert_eq!(
            classify_daemon_record_evidence(
                &current,
                Some(&bcode_ipc::DaemonStatus {
                    instance_id: "replacement".to_owned(),
                    ..exact
                }),
                true,
                ProcessIdentityEvidence::MissingOrReused,
            ),
            DaemonRecordClassification::ResponsiveIdentityMismatch
        );
        assert_eq!(
            classify_daemon_record_evidence(
                &current,
                None,
                false,
                ProcessIdentityEvidence::MissingOrReused,
            ),
            DaemonRecordClassification::UnreachableStale
        );
        assert_eq!(
            classify_daemon_record_evidence(
                &current,
                None,
                false,
                ProcessIdentityEvidence::Unverifiable,
            ),
            DaemonRecordClassification::Unverifiable
        );
    }

    #[test]
    fn storage_writer_compatibility_fails_closed_for_legacy_records() {
        assert!(storage_writer_is_incompatible(
            &record_with_writer_epoch(None),
            2
        ));
        assert!(storage_writer_is_incompatible(
            &record_with_writer_epoch(Some(1)),
            2
        ));
        assert!(!storage_writer_is_incompatible(
            &record_with_writer_epoch(Some(2)),
            2
        ));
    }

    #[test]
    fn daemon_record_accepts_verified_executable_outside_cache() {
        let (executable_path, digest) = current_executable_identity().unwrap();
        let record = DaemonRecord::current(
            &bcode_ipc::default_endpoint(),
            PathBuf::from("test.log"),
            Some(executable_path.clone()),
            "foreground-test".to_string(),
        )
        .unwrap();

        assert_eq!(
            record.executable_path.as_deref(),
            Some(executable_path.as_path())
        );
        assert_eq!(record.executable_digest.as_deref(), Some(digest.as_str()));
    }

    #[test]
    fn cached_daemon_executable_path_is_content_addressed() {
        assert_eq!(
            cached_daemon_executable_path_for_digest(Path::new("/state"), "abc123"),
            Path::new("/state")
                .join("daemon-images")
                .join(bcode_ipc::ArtifactId::current().as_str())
                .join("abc123")
                .join(if cfg!(windows) { "bcode.exe" } else { "bcode" })
        );
    }

    #[test]
    fn bootstrap_rejects_identity_from_different_executable_bytes() {
        let root = std::env::temp_dir().join(format!(
            "bcode-bootstrap-identity-{}-{}",
            std::process::id(),
            unix_time_millis().unwrap()
        ));
        fs::create_dir_all(&root).unwrap();
        let path = root.join("bcode");
        fs::write(&path, b"retained bytes").unwrap();
        let identity =
            bcode_ipc::ArtifactId::for_executable(&mut b"retained bytes".as_slice()).unwrap();
        let bootstrap = ArtifactBootstrap::open_verified(path.clone(), identity.clone()).unwrap();
        assert_eq!(bootstrap.artifact_id(), &identity);
        assert_eq!(
            bootstrap.digest().unwrap(),
            executable_sha256(&path).unwrap()
        );
        let foreign = bcode_ipc::ArtifactId::parse("foreign-bootstrap-identity").unwrap();
        assert!(ArtifactBootstrap::open_verified(path, foreign).is_err());
        fs::remove_dir_all(root).unwrap();
    }

    #[cfg(unix)]
    #[test]
    fn retained_artifact_handle_survives_source_path_replacement() {
        let root = std::env::temp_dir().join(format!(
            "bcode-retained-artifact-test-{}-{}",
            std::process::id(),
            unix_time_millis().unwrap()
        ));
        fs::create_dir_all(&root).unwrap();
        let source = root.join("bcode");
        fs::write(&source, b"original artifact bytes").unwrap();
        let bootstrap = ArtifactBootstrap::open(
            source.clone(),
            bcode_ipc::ArtifactId::parse("retained-test").unwrap(),
        )
        .unwrap();
        let original_digest = bootstrap.digest().unwrap();
        assert!(bootstrap.source_path_still_current());
        let replacement = root.join("replacement");
        fs::write(&replacement, b"replacement bytes").unwrap();
        fs::rename(&replacement, &source).unwrap();
        assert!(!bootstrap.source_path_still_current());
        let copied = root.join("copied");
        let copied_digest = bootstrap.copy_to(&copied).unwrap();

        assert_eq!(copied_digest, original_digest);
        assert_eq!(executable_sha256(&copied).unwrap(), original_digest);
        assert_ne!(executable_sha256(&source).unwrap(), original_digest);
        fs::remove_dir_all(root).unwrap();
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn foreign_record_classification_never_connects_to_its_socket() {
        let directory = tempfile::tempdir().unwrap();
        let socket = directory.path().join("foreign.sock");
        let listener = tokio::net::UnixListener::bind(&socket).unwrap();
        let endpoint = IpcEndpoint::unix_socket(socket);
        let mut record = DaemonRecord::current(
            &endpoint,
            directory.path().join("log"),
            None,
            "foreign".into(),
        )
        .unwrap();
        record.artifact_id = Some(bcode_ipc::ArtifactId::parse("foreign-artifact").unwrap());
        assert!(pinned_daemon_endpoint(&record).is_err());
        let _ = classify_daemon_record(&record).await;
        assert!(
            tokio::time::timeout(Duration::from_millis(25), listener.accept())
                .await
                .is_err()
        );
        record.artifact_id = Some(bcode_ipc::ArtifactId::current());
        record.executable_digest = Some("foreign-bytes".into());
        assert!(pinned_daemon_endpoint(&record).is_err());
        let _ = classify_daemon_record(&record).await;
        assert!(
            tokio::time::timeout(Duration::from_millis(25), listener.accept())
                .await
                .is_err()
        );
    }

    #[test]
    fn cached_image_with_same_artifact_but_different_bytes_is_not_selected() {
        let state_dir = std::env::temp_dir().join(format!(
            "bcode-wrong-image-{}-{}",
            std::process::id(),
            unix_time_millis().unwrap()
        ));
        let other = b"another executable with the same artifact identity";
        let digest = hex::encode(Sha256::digest(other));
        let path = cached_daemon_executable_path_for_digest(&state_dir, &digest);
        fs::create_dir_all(path.parent().unwrap()).unwrap();
        fs::write(&path, other).unwrap();
        write_daemon_image_metadata(&path, &digest).unwrap();
        assert!(daemon_image_is_valid(&path, &digest).unwrap());
        assert!(current_cached_daemon_image(&state_dir).unwrap().is_none());
        let selected = ensure_current_executable_cached_in_state(&state_dir).unwrap();
        assert_ne!(selected, path);
        assert_eq!(
            executable_sha256(&selected).unwrap(),
            initialize_artifact_bootstrap().unwrap().digest().unwrap()
        );
        fs::remove_dir_all(state_dir).unwrap();
    }

    #[test]
    fn corrupt_cached_image_is_repaired_with_exact_metadata() {
        let state_dir = std::env::temp_dir().join(format!(
            "bcode-daemon-image-repair-test-{}-{}",
            std::process::id(),
            unix_time_millis().unwrap()
        ));
        let cached = ensure_current_executable_cached_in_state(&state_dir).unwrap();
        fs::write(&cached, b"corrupt").unwrap();

        let repaired = ensure_current_executable_cached_in_state(&state_dir).unwrap();
        let (_source, digest) = current_executable_identity().unwrap();
        assert_eq!(repaired, cached);
        assert!(daemon_image_is_valid(&repaired, &digest).unwrap());
        assert_eq!(
            ensure_current_executable_cached_in_state(&state_dir).unwrap(),
            repaired
        );
        assert!(
            fs::read_dir(repaired.parent().expect("cached image parent"))
                .unwrap()
                .all(|entry| !entry
                    .unwrap()
                    .file_name()
                    .to_string_lossy()
                    .starts_with(".bcode.tmp-"))
        );
        fs::remove_dir_all(state_dir).unwrap();
    }

    #[test]
    fn interrupted_cache_copy_is_removed_before_verified_publication() {
        let state_dir = std::env::temp_dir().join(format!(
            "bcode-daemon-image-interrupted-test-{}-{}",
            std::process::id(),
            unix_time_millis().unwrap()
        ));
        let (_source, digest) = current_executable_identity().unwrap();
        let target = cached_daemon_executable_path_for_digest(&state_dir, &digest);
        let parent = target.parent().expect("cached image parent");
        fs::create_dir_all(parent).unwrap();
        let interrupted = parent.join(format!(".bcode.tmp-{}", std::process::id()));
        fs::write(&interrupted, b"partial executable bytes").unwrap();

        let cached = ensure_current_executable_cached_in_state(&state_dir).unwrap();

        assert_eq!(cached, target);
        assert!(!interrupted.exists());
        assert!(daemon_image_is_valid(&cached, &digest).unwrap());
        fs::remove_dir_all(state_dir).unwrap();
    }

    #[test]
    fn pinned_image_verification_never_repairs_or_substitutes() {
        let root = tempfile::tempdir().unwrap();
        let path = ensure_current_executable_cached_in_state(root.path()).unwrap();
        let digest = executable_sha256(&path).unwrap();
        let pinned = PinnedBcodeExecutable { path, digest };
        assert!(verify_pinned_bcode_executable(&pinned).is_ok());
        fs::write(pinned.path(), b"replacement").unwrap();
        assert!(verify_pinned_bcode_executable(&pinned).is_err());
        assert_eq!(fs::read(pinned.path()).unwrap(), b"replacement");
        fs::remove_file(pinned.path()).unwrap();
        assert!(verify_pinned_bcode_executable(&pinned).is_err());
        assert!(!pinned.path().exists());
    }

    #[test]
    fn content_addressed_paths_are_verified() {
        let path = cached_daemon_executable_path_for_digest(Path::new("/state"), "abc123");
        assert!(executable_path_matches_digest(&path, "abc123"));
        assert!(!executable_path_matches_digest(&path, "other"));
        assert!(!executable_path_matches_digest(
            Path::new("/state/daemon-images/legacy/bcode"),
            "abc123"
        ));
    }

    #[test]
    fn caches_current_executable_into_immutable_digest_directory() {
        let state_dir = std::env::temp_dir().join(format!(
            "bcode-daemon-image-test-{}-{}",
            std::process::id(),
            unix_time_millis().unwrap()
        ));

        let cached = ensure_current_executable_cached_in_state(&state_dir).unwrap();
        let (_source, digest) = current_executable_identity().unwrap();

        assert!(cached.exists());
        assert_eq!(
            cached,
            cached_daemon_executable_path_for_digest(&state_dir, &digest)
        );
        assert_eq!(executable_sha256(&cached).unwrap(), digest);
        assert!(daemon_image_is_valid(&cached, &digest).unwrap());
        assert_eq!(
            serde_json::from_slice::<DaemonImageMetadata>(
                &fs::read(daemon_image_metadata_path(&cached)).unwrap()
            )
            .unwrap()
            .artifact_id,
            bcode_ipc::ArtifactId::current()
        );
        assert_eq!(
            ensure_current_executable_cached_in_state(&state_dir).unwrap(),
            cached
        );
        let _ = fs::remove_dir_all(state_dir);
    }
}

/// Options controlling daemon startup orchestration.
#[derive(Debug, Clone)]
pub struct EnsureDaemonOptions {
    /// Endpoint the daemon should serve.
    pub endpoint: IpcEndpoint,
    /// Suppress user-facing status output.
    pub quiet: bool,
    /// Path used for daemon stdout/stderr logs.
    pub log_path: PathBuf,
}

impl EnsureDaemonOptions {
    /// Build default daemon startup options for the current namespace.
    #[must_use]
    pub fn default_for_current_namespace() -> Self {
        Self {
            endpoint: bcode_ipc::default_endpoint(),
            quiet: true,
            log_path: default_daemon_log_path(),
        }
    }
}

/// Error returned when daemon process startup fails.
#[derive(Debug, Error)]
pub enum DaemonStartError {
    /// Daemon lifecycle registry cleanup failed.
    #[error("daemon lifecycle error: {0}")]
    Lifecycle(#[from] DaemonLifecycleError),
    /// Daemon process I/O failed.
    #[error("daemon process I/O error: {0}")]
    Io(#[from] std::io::Error),
    /// The detached child's readiness notification was not observed before the deadline.
    /// This records the failed wait, not the child's current readiness or execution lifetime.
    #[error(
        "daemon readiness notification deadline elapsed; child readiness and execution lifetime are unknown; log: {log_path}\ntry `bcode server run` to see startup failures in the foreground\n\n{recent_log}"
    )]
    NotificationTimeout {
        /// Daemon log path.
        log_path: String,
        /// Recent daemon log excerpt.
        recent_log: String,
    },
    /// Daemon did not become ready before the startup timeout.
    #[error(
        "daemon did not become ready after auto-start; log: {log_path}\ntry `bcode server run` to see startup failures in the foreground\n\n{recent_log}"
    )]
    StartTimeout {
        /// Daemon log path.
        log_path: String,
        /// Recent daemon log excerpt.
        recent_log: String,
    },
    /// Daemon process exited before readiness.
    #[error(
        "daemon exited before becoming ready ({status}); log: {log_path}\ntry `bcode server run` to see startup failures in the foreground\n\n{recent_log}"
    )]
    Exited {
        /// Child process exit status.
        status: String,
        /// Daemon log path.
        log_path: String,
        /// Recent daemon log excerpt.
        recent_log: String,
    },
    /// Daemon readiness was transient and failed a follow-up health check.
    #[error(
        "daemon became ready but failed a follow-up health check; log: {log_path}\ntry `bcode server run` to see startup failures in the foreground\n\n{recent_log}"
    )]
    HealthCheckFailed {
        /// Daemon log path.
        log_path: String,
        /// Recent daemon log excerpt.
        recent_log: String,
    },
    /// A live process occupies the target endpoint but did not complete a compatible handshake.
    #[error(
        "daemon endpoint {endpoint} is occupied by a live process that did not become ready; refusing to start a competing daemon"
    )]
    EndpointOccupied {
        /// Display form of the occupied endpoint.
        endpoint: String,
    },
    /// Another process held the namespace startup lock beyond the bounded startup window.
    #[error("timed out waiting for daemon startup coordination lock at {}", lock_path.display())]
    StartupCoordinationTimeout {
        /// Namespace-scoped startup lock path.
        lock_path: PathBuf,
    },
}

impl DaemonStartError {
    /// Return true when startup likely lost a race to an already-running daemon.
    #[must_use]
    pub fn is_existing_daemon_race(&self) -> bool {
        match self {
            Self::Exited { recent_log, .. }
            | Self::NotificationTimeout { recent_log, .. }
            | Self::StartTimeout { recent_log, .. }
            | Self::HealthCheckFailed { recent_log, .. } => {
                recent_log.contains("refusing to replace live IPC socket")
                    || recent_log.contains("another bcode daemon is listening")
                    || recent_log.contains("Address already in use")
            }
            Self::Io(error) => error.kind() == std::io::ErrorKind::AddrInUse,
            Self::Lifecycle(_)
            | Self::EndpointOccupied { .. }
            | Self::StartupCoordinationTimeout { .. } => false,
        }
    }
}

/// Return the `XDG_CONFIG_HOME` value that reproduces `config_dir` in a child process.
///
/// `bcode_config::default_config_dir()` resolves to `<config-home>/bcode`, so the
/// daemon must inherit the parent of that directory for its own resolution to land on
/// the same path. When the resolved directory has no parent — or is not the expected
/// `bcode` leaf, which can happen for a temp-dir fallback — the directory is passed
/// through unchanged rather than guessing, so a mismatch surfaces at the handshake
/// instead of silently redirecting the daemon's config.
fn config_home_for_daemon(config_dir: &Path) -> PathBuf {
    if config_dir.file_name().and_then(|name| name.to_str()) == Some("bcode")
        && let Some(parent) = config_dir.parent()
    {
        return parent.to_path_buf();
    }
    config_dir.to_path_buf()
}

fn daemon_command(exe: &Path) -> tokio::process::Command {
    let mut command = tokio::process::Command::new(exe);
    detach_daemon_process(&mut command);
    configure_daemon_config(
        &mut command,
        bcode_config::process_exclusive_config_path().as_deref(),
    );
    command
}

fn configure_daemon_config(command: &mut tokio::process::Command, path: Option<&Path>) {
    if let Some(path) = path {
        command.arg("--config-only").arg(path);
    }
}

fn diagnostic_startup_id() -> u128 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map_or(0, |duration| duration.as_nanos())
}

/// Prepare and execute the immutable current-artifact image without starting a daemon.
///
/// Uses the normal startup fence and image retention guard. Preparation is explicit
/// so installer/build time can absorb first-execution OS validation without weakening it.
///
/// # Errors
/// Returns normal lifecycle errors or a failed cached-image identity probe.
pub async fn prepare_daemon_image(endpoint: &IpcEndpoint) -> Result<(), DaemonStartError> {
    let Some(_startup) = StartupLock::acquire(endpoint).await? else {
        return Ok(());
    };
    let _retention = DaemonImageUseGuard::acquire(&bcode_config::default_state_dir())?;
    let image = ensure_current_executable_cached()?;
    let output = tokio::time::timeout(
        Duration::from_secs(30),
        tokio::process::Command::new(image)
            .arg("artifact-id")
            .stdin(std::process::Stdio::null())
            .kill_on_drop(true)
            .output(),
    )
    .await
    .map_err(|_| {
        std::io::Error::new(
            std::io::ErrorKind::TimedOut,
            "prepared daemon image probe timed out",
        )
    })??;
    if !output.status.success()
        || output.stdout != format!("{}\n", bcode_ipc::ArtifactId::current()).as_bytes()
    {
        return Err(std::io::Error::other("prepared daemon image identity probe failed").into());
    }
    Ok(())
}

/// Ensure the current namespace daemon is running, starting it when needed.
///
/// # Errors
///
/// Returns an error when stale-record cleanup fails, spawning the daemon fails,
/// or the daemon does not pass bounded readiness checks.
pub async fn ensure_daemon_running(options: &EnsureDaemonOptions) -> Result<(), DaemonStartError> {
    use tracing::Instrument as _;

    let startup_id = diagnostic_startup_id();
    let launcher_pid = std::process::id();
    let span = tracing::debug_span!(
        target: "bcode_daemon_lifecycle::startup",
        "daemon_acquisition",
        startup_id,
        launcher_pid
    );
    let acquisition_phase = bcode_metrics::startup::phase_for(
        "launcher.acquire",
        &format!("{launcher_pid}-{startup_id}"),
    );
    let result = ensure_daemon_running_with_start(options, |options, startup_lock| {
        #[cfg(unix)]
        let handoff = format!("{}-{startup_id}", std::process::id());
        let inherited_startup_lock = startup_lock.file.try_clone();
        #[cfg(unix)]
        let descriptor_file = startup_lock.file.try_clone();
        let endpoint = options.endpoint.clone();
        let log_path = options.log_path.clone();
        async move {
            let log_file = open_startup_log(&log_path)?;
            let stderr_log = log_file.try_clone()?;

            let spawn_started_at = std::time::Instant::now();
            let phase = bcode_metrics::startup::phase("launcher.artifact_prepare");
            let cached_exe = ensure_current_executable_cached()?;
            phase.finish();
            let executable_digest = require_cached_executable_digest(&cached_exe)?;
            // Always spawn from the immutable content-addressed image. Spawning from the
            // bootstrap source path would pin the daemon to a mutable pathname (for example
            // `target/<triple>/release/bcode`), so rebuilding that path would later make the
            // running daemon's own identity unverifiable even though it keeps serving its
            // original artifact correctly.
            let exe = cached_exe;
            tracing::debug!(
                target: "bcode_daemon_lifecycle::startup",
                elapsed_ms = spawn_started_at.elapsed().as_millis(),
                "daemon image ready"
            );
            let (endpoint_env_name, endpoint_env_value) =
                bcode_ipc::endpoint_env_pair(&endpoint)
                    .map_err(|error| std::io::Error::other(error.to_string()))?;
            #[cfg(unix)]
            write_startup_owner(
                &descriptor_file?,
                &StartupOwnerDescriptor::current("handoff_pending", handoff.clone())?,
            )?;
            let mut command = daemon_command(&exe);
            #[cfg(unix)]
            command.env("BCODE_STARTUP_HANDOFF_V1", handoff);
            let spawn_phase = bcode_metrics::startup::phase("launcher.process_spawn");
            let mut child = command
                .args(["server", "run"])
                .env(endpoint_env_name, endpoint_env_value)
                .env(
                    bcode_ipc::BCODE_IPC_ENDPOINT_NAMESPACE_ENV,
                    bcode_ipc::daemon_namespace(),
                )
                // The daemon must serve the runtime scope this client resolved — both the
                // state location and the config directory — not whatever it would infer
                // from inherited environment. An explicit command-line `--state-root`
                // selection is process-scoped and would otherwise be lost across the
                // process boundary, silently starting a daemon on a different canonical
                // session root or under different plugin/permission policy.
                .env(
                    bcode_config::BCODE_STATE_DIR_ENV,
                    bcode_config::default_state_dir(),
                )
                .env(
                    bcode_config::BCODE_SESSION_STORE_DIR_ENV,
                    bcode_config::default_session_store_dir(),
                )
                .env(
                    "XDG_CONFIG_HOME",
                    config_home_for_daemon(&bcode_config::default_config_dir()),
                )
                .env(BCODE_EXECUTABLE_DIGEST_ENV, executable_digest)
                .env("BCODE_DAEMON_LOG", &log_path)
                .env("BCODE_DAEMON_READY_STDOUT", "v1")
                .env(
                    "BCODE_STARTUP_CORRELATION",
                    format!("{launcher_pid}-{startup_id}"),
                )
                // The daemon remains detached on cancellation. On Unix its inherited
                // startup lock fences later launchers even before endpoint publication.
                .kill_on_drop(false)
                // On Unix the inherited open file description retains startup ownership
                // even if the launcher exits before the child binds its endpoint.
                .stdin(startup_lock_stdio(inherited_startup_lock?))
                .stdout(std::process::Stdio::piped())
                .stderr(std::process::Stdio::from(stderr_log))
                .spawn()?;
            // Release Command's configured stdin duplicate; the child and scoped
            // StartupLock are now the only intentional owners of this OFD.
            drop(command);
            spawn_phase.finish();
            tracing::debug!(
                target: "bcode_daemon_lifecycle::startup",
                elapsed_ms = spawn_started_at.elapsed().as_millis(),
                "daemon child spawned"
            );

            measured_child_readiness(&endpoint, &mut child, &log_path).await
        }
    })
    .instrument(span)
    .await;
    acquisition_phase.finish_result(&result);
    result
}

fn require_cached_executable_digest(path: &Path) -> Result<String, DaemonLifecycleError> {
    cached_executable_digest(path)
        .map(ToOwned::to_owned)
        .ok_or_else(|| DaemonLifecycleError::Io {
            path: path.to_owned(),
            source: std::io::Error::new(
                std::io::ErrorKind::InvalidData,
                "cached daemon executable path has no digest component",
            ),
        })
}

fn open_startup_log(path: &Path) -> std::io::Result<fs::File> {
    if let Some(parent) = path.parent() {
        fs::create_dir_all(parent)?;
    }
    // Truncate before opening append-only handles: launcher and child stderr
    // share this path, so positional writes would overwrite trace records.
    fs::File::create(path)?;
    let mut file = fs::OpenOptions::new()
        .create(true)
        .append(true)
        .open(path)?;
    writeln!(file, "--- bcode daemon start ---")?;
    Ok(file)
}

/// Detach the child from the launching terminal, not merely its stdio handles.
fn detach_daemon_process(command: &mut tokio::process::Command) {
    #[cfg(unix)]
    // SAFETY: setsid is async-signal-safe and touches no Rust state in the
    // post-fork child. The child starts a new session before exec, so terminal
    // closure cannot deliver SIGHUP to the daemon's process group.
    unsafe {
        command.pre_exec(|| {
            if libc::setsid() == -1 {
                return Err(std::io::Error::last_os_error());
            }
            Ok(())
        });
    }
    #[cfg(windows)]
    command.creation_flags(0x0000_0008 | 0x0000_0200); // DETACHED_PROCESS | CREATE_NEW_PROCESS_GROUP
    #[cfg(not(any(unix, windows)))]
    let _ = command;
}

#[cfg(all(test, unix))]
#[tokio::test]
async fn daemon_child_has_a_new_session_and_process_group() {
    let mut command = tokio::process::Command::new("sleep");
    command.arg("10").kill_on_drop(true);
    detach_daemon_process(&mut command);
    let mut child = command.spawn().expect("spawn detached child");
    let pid = i32::try_from(child.id().expect("child pid")).expect("pid fits");
    // SAFETY: read-only process identity queries for the child we just spawned.
    let (session, group) = unsafe { (libc::getsid(pid), libc::getpgid(pid)) };
    child.kill().await.expect("stop test child");
    assert_eq!(
        session, pid,
        "daemon must not inherit the launching terminal session"
    );
    assert_eq!(
        group, pid,
        "daemon must not inherit the launching process group"
    );
}

/// Ensure the current namespace daemon is running using an in-process start callback.
///
/// # Errors
///
/// Returns an error when stale endpoint cleanup fails, the callback fails, or the daemon does not
/// pass bounded readiness checks.
pub async fn ensure_daemon_running_in_process(
    options: &EnsureDaemonOptions,
    mut start: impl FnMut() -> Result<(), DaemonStartError>,
) -> Result<(), DaemonStartError> {
    let log_path = options.log_path.clone();
    ensure_daemon_running_with_start(options, move |options, _startup_lock| {
        let result = start();
        let endpoint = options.endpoint.clone();
        let log_path = log_path.clone();
        async move {
            result?;
            wait_for_daemon_ready(&endpoint, &log_path).await
        }
    })
    .await
}

async fn ensure_daemon_running_with_start<F, Fut>(
    options: &EnsureDaemonOptions,
    mut start: F,
) -> Result<(), DaemonStartError>
where
    F: FnMut(&EnsureDaemonOptions, &StartupLock) -> Fut,
    Fut: std::future::Future<Output = Result<(), DaemonStartError>>,
{
    if ping_ready(&options.endpoint).await {
        print_daemon_status(options, "server already running");
        return Ok(());
    }

    let phase = bcode_metrics::startup::phase("launcher.startup_lock");
    let acquired = StartupLock::acquire(&options.endpoint).await;
    phase.finish_result(&acquired);
    let Some(lock) = acquired? else {
        print_daemon_status(options, "server already running");
        return Ok(());
    };
    #[cfg(unix)]
    if endpoint_has_listener(&options.endpoint).await {
        if wait_for_existing_daemon(&options.endpoint).await {
            drop(lock);
            print_daemon_status(options, "server already running");
            return Ok(());
        }
        return Err(DaemonStartError::EndpointOccupied {
            endpoint: format!("{:?}", options.endpoint),
        });
    }
    let image_use_guard = DaemonImageUseGuard::acquire(&bcode_config::default_state_dir())?;
    #[cfg(unix)]
    cleanup_stale_endpoint(&options.endpoint)?;
    if ping_ready(&options.endpoint).await {
        drop(image_use_guard);
        drop(lock);
        print_daemon_status(options, "server already running");
        return Ok(());
    }

    start(options, &lock).await?;
    drop(image_use_guard);
    let _cleanup_task = tokio::spawn(async {
        let _ = cleanup_stale_daemon_records().await;
        let _ = cleanup_stale_daemon_images_retaining_artifacts(
            &bcode_config::default_state_dir(),
            &registered_retained_artifact_ids(),
        );
    });
    drop(lock);
    print_daemon_status(options, "server started");
    Ok(())
}

/// Report whether any cached daemon image exists for one artifact identity.
///
/// Durable work fenced to an artifact can only be recovered by launching a daemon of exactly that
/// identity. When no cached image remains, that work is unrecoverable rather than merely deferred.
#[must_use]
pub fn artifact_image_is_available(state_dir: &Path, artifact_id: &str) -> bool {
    let namespace = state_dir.join("daemon-images").join(artifact_id);
    fs::read_dir(namespace).is_ok_and(|mut entries| entries.any(|entry| entry.is_ok()))
}

/// Supplies artifact identities whose cached daemon images must be preserved.
///
/// Domains that fence durable work to an exact daemon artifact register a provider so background
/// image cleanup cannot delete the only launchable daemon for still-recoverable work. The hook
/// keeps this crate free of any dependency on those domains.
type RetainedArtifactProvider = fn() -> std::collections::BTreeSet<String>;

static RETAINED_ARTIFACT_PROVIDERS: std::sync::Mutex<Vec<RetainedArtifactProvider>> =
    std::sync::Mutex::new(Vec::new());

/// Register a provider of artifact identities that must keep their cached daemon images.
///
/// Providers are consulted by background image cleanup. Registration is additive and idempotent
/// only by caller discipline; register each provider once during startup.
pub fn register_retained_artifact_provider(provider: RetainedArtifactProvider) {
    RETAINED_ARTIFACT_PROVIDERS
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner)
        .push(provider);
}

/// Collect retained artifact identities from every registered provider.
#[must_use]
pub fn registered_retained_artifact_ids() -> std::collections::BTreeSet<String> {
    let providers = RETAINED_ARTIFACT_PROVIDERS
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner)
        .clone();
    providers
        .into_iter()
        .flat_map(|provider| provider())
        .collect()
}

fn print_daemon_status(options: &EnsureDaemonOptions, status: &str) {
    if !options.quiet {
        println!("{status}");
        println!("namespace: {}", daemon_namespace());
        println!("log: {}", display_from_current_dir(&options.log_path));
    }
}

/// Return the default daemon log path for the current namespace.
#[must_use]
pub fn default_daemon_log_path() -> PathBuf {
    std::env::var_os("BCODE_DAEMON_LOG").map_or_else(
        || {
            bcode_config::default_state_dir()
                .join("logs")
                .join(format!("daemon-{}.log", daemon_namespace()))
        },
        PathBuf::from,
    )
}

#[cfg(unix)]
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct StartupOwnerDescriptor {
    version: u32,
    artifact: String,
    state: String,
    executable_digest: String,
    pid: u32,
    process_start_seconds: u64,
    phase: String,
    handoff: String,
}

#[cfg(unix)]
impl StartupOwnerDescriptor {
    fn current(phase: &str, handoff: String) -> std::io::Result<Self> {
        let pid = std::process::id();
        let mut system = System::new();
        system.refresh_processes(ProcessesToUpdate::Some(&[Pid::from_u32(pid)]), true);
        let process_start_seconds = system
            .process(Pid::from_u32(pid))
            .ok_or_else(|| std::io::Error::other("startup process identity unavailable"))?
            .start_time();
        Ok(Self {
            version: 1,
            artifact: bcode_ipc::ArtifactId::current().to_string(),
            state: bcode_ipc::state_location_id(),
            executable_digest: initialize_artifact_bootstrap()
                .and_then(ArtifactBootstrap::digest)
                .map_err(std::io::Error::other)?,
            pid,
            process_start_seconds,
            phase: phase.to_owned(),
            handoff,
        })
    }

    fn matches_scope(&self) -> bool {
        self.version == 1
            && matches!(
                self.phase.as_str(),
                "preparing" | "handoff_pending" | "daemon_ready"
            )
            && !self.handoff.is_empty()
            && self.handoff.len() <= 128
            && self.artifact == bcode_ipc::ArtifactId::current().to_string()
            && self.state == bcode_ipc::state_location_id()
            && initialize_artifact_bootstrap()
                .and_then(ArtifactBootstrap::digest)
                .is_ok_and(|digest| digest == self.executable_digest)
    }

    fn process_verified(&self) -> bool {
        let pid = Pid::from_u32(self.pid);
        let mut system = System::new();
        system.refresh_processes(ProcessesToUpdate::Some(&[pid]), true);
        system.process(pid).is_some_and(|process| {
            process.start_time() == self.process_start_seconds
                && process.exe().is_some_and(|exe| {
                    executable_path_matches_digest(exe, &self.executable_digest)
                        && executable_sha256(exe)
                            .is_ok_and(|digest| digest == self.executable_digest)
                })
        })
    }
}

// Positional access avoids the shared OFD cursor. Partial concurrent snapshots are
// explicitly unknown; they never authorize takeover or a second spawn.
#[cfg(unix)]
fn read_startup_owner(file: &fs::File) -> std::io::Result<Option<StartupOwnerDescriptor>> {
    use std::os::unix::fs::FileExt as _;
    let mut bytes = [0_u8; 8193];
    let size = file.read_at(&mut bytes, 0)?;
    if size == 0 || bytes[..size].starts_with(b"pid=") {
        return Ok(None);
    }
    if size > 8192 {
        return Err(std::io::Error::other("startup descriptor exceeds bound"));
    }
    serde_json::from_slice(&bytes[..size])
        .map(Some)
        .map_err(std::io::Error::other)
}

#[cfg(unix)]
fn write_startup_owner(file: &fs::File, owner: &StartupOwnerDescriptor) -> std::io::Result<()> {
    use std::os::unix::fs::FileExt as _;
    let bytes = serde_json::to_vec(owner).map_err(std::io::Error::other)?;
    if bytes.len() > 8192 {
        return Err(std::io::Error::other("startup descriptor exceeds bound"));
    }
    file.write_all_at(&bytes, 0)?;
    file.set_len(u64::try_from(bytes.len()).map_err(std::io::Error::other)?)?;
    file.sync_data()
}

#[cfg(unix)]
fn publish_daemon_startup_owner() -> std::io::Result<()> {
    use std::os::fd::BorrowedFd;
    let handoff = std::env::var("BCODE_STARTUP_HANDOFF_V1").map_err(std::io::Error::other)?;
    // SAFETY: duplicate process-owned stdin without taking ownership of it.
    let file = fs::File::from(unsafe { BorrowedFd::borrow_raw(0) }.try_clone_to_owned()?);
    let path = startup_lock_path(
        &bcode_config::default_state_dir(),
        &daemon_namespace(),
        &bcode_ipc::state_location_id(),
    );
    publish_startup_owner_at(&file, &path, &handoff)?;
    close_startup_stdin_on_exec()
}

#[cfg(unix)]
fn close_startup_stdin_on_exec() -> std::io::Result<()> {
    // Keep stdin open for this daemon's lifetime, but never transfer its fence
    // through a subsequent exec. F_GETFD/F_SETFD affect this process's fd only.
    let flags = unsafe { libc::fcntl(0, libc::F_GETFD) };
    if flags == -1 || unsafe { libc::fcntl(0, libc::F_SETFD, flags | libc::FD_CLOEXEC) } == -1 {
        return Err(std::io::Error::last_os_error());
    }
    Ok(())
}

#[cfg(unix)]
fn publish_startup_owner_at(file: &fs::File, path: &Path, handoff: &str) -> std::io::Result<()> {
    use std::os::unix::fs::MetadataExt as _;
    let expected = fs::symlink_metadata(path)?;
    let actual = file.metadata()?;
    if !expected.is_file() || expected.dev() != actual.dev() || expected.ino() != actual.ino() {
        return Err(std::io::Error::other(
            "startup descriptor path identity mismatch",
        ));
    }
    let owner = read_startup_owner(file)?
        .ok_or_else(|| std::io::Error::other("startup descriptor unavailable"))?;
    if !owner.matches_scope() || owner.phase != "handoff_pending" || owner.handoff != handoff {
        return Err(std::io::Error::other("startup descriptor handoff mismatch"));
    }
    write_startup_owner(
        file,
        &StartupOwnerDescriptor::current("daemon_ready", handoff.to_owned())?,
    )
}

#[cfg(unix)]
fn observe_startup_owner(file: &fs::File) {
    let owner = read_startup_owner(file)
        .ok()
        .flatten()
        .filter(StartupOwnerDescriptor::matches_scope);
    tracing::debug!(target: "bcode_daemon_lifecycle::startup",
        claimed_pid = ?owner.as_ref().map(|value| value.pid),
        claimed_process_start_seconds = ?owner.as_ref().map(|value| value.process_start_seconds),
        artifact = ?owner.as_ref().map(|value| value.artifact.as_str()),
        state = ?owner.as_ref().map(|value| value.state.as_str()),
        executable_digest = ?owner.as_ref().map(|value| value.executable_digest.as_str()),
        phase = ?owner.as_ref().map(|value| value.phase.as_str()),
        verified_process = owner.as_ref().is_some_and(StartupOwnerDescriptor::process_verified),
        kernel_lock_owner = "unknown",
        "fresh startup descriptor observation");
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum StartupLockObservation {
    Waiting,
    Acquired,
    ConcurrentReady,
    Timeout,
}

impl StartupLockObservation {
    const fn phase_name(self) -> &'static str {
        match self {
            Self::Waiting => "launcher.lock_wait",
            Self::Acquired => "launcher.lock_acquired",
            Self::ConcurrentReady => "launcher.lock_concurrent_ready",
            Self::Timeout => "launcher.lock_timeout",
        }
    }
}

#[derive(Debug)]
struct StartupLock {
    file: fs::File,
}

impl StartupLock {
    // Covers the bounded 30s explicit image probe plus scheduling/verification.
    const ACQUIRE_TIMEOUT: Duration = Duration::from_secs(35);
    const POLL_INTERVAL: Duration = Duration::from_millis(25);

    async fn acquire(endpoint: &IpcEndpoint) -> Result<Option<Self>, DaemonStartError> {
        let path = startup_lock_path(
            &bcode_config::default_state_dir(),
            &daemon_namespace(),
            &bcode_ipc::state_location_id(),
        );
        let artifact = bcode_ipc::ArtifactId::current().to_string();
        let state = bcode_ipc::state_location_id();
        let scope = format!("{artifact}.{state}");
        let acquired = Self::acquire_observed(
            path,
            Self::ACQUIRE_TIMEOUT,
            &scope,
            || ping_ready(endpoint),
            |observation, elapsed| {
                // Terminal outcomes are point observations in tracing below.
                // The scoped lock_wait phase owns the actual elapsed interval.
                tracing::debug!(
                    target: "bcode_daemon_lifecycle::startup",
                    artifact_id = %artifact,
                    state_location_id = %state,
                    phase = observation.phase_name(),
                    elapsed_us = elapsed.as_micros(),
                    holder_identity = "unknown",
                    lock_pid_evidence = "unverified_hint_not_read",
                    "startup coordination observation"
                );
            },
        )
        .await?;
        #[cfg(unix)]
        if let Some(lock) = &acquired {
            if read_startup_owner(&lock.file)?.is_some_and(|owner| !owner.matches_scope()) {
                return Err(
                    std::io::Error::other("unsupported or foreign startup descriptor").into(),
                );
            }
            write_startup_owner(
                &lock.file,
                &StartupOwnerDescriptor::current(
                    "preparing",
                    format!("{}-{}", std::process::id(), diagnostic_startup_id()),
                )?,
            )?;
        }
        Ok(acquired)
    }

    async fn acquire_observed<F, Fut>(
        path: PathBuf,
        timeout: Duration,
        scope: &str,
        mut ready: F,
        mut observe: impl FnMut(StartupLockObservation, Duration),
    ) -> Result<Option<Self>, DaemonStartError>
    where
        F: FnMut() -> Fut,
        Fut: std::future::Future<Output = bool>,
    {
        if let Some(parent) = path.parent() {
            fs::create_dir_all(parent)?;
        }
        let started = std::time::Instant::now();
        let wait_phase = bcode_metrics::startup::phase_for("launcher.lock_wait", scope);
        let mut waiting = false;
        loop {
            let mut open = fs::OpenOptions::new();
            open.read(true).write(true).create(true).truncate(false);
            #[cfg(unix)]
            {
                use std::os::unix::fs::OpenOptionsExt as _;
                open.custom_flags(libc::O_NOFOLLOW | libc::O_NONBLOCK)
                    .mode(0o600);
            }
            let file = open.open(&path)?;
            if !file.metadata()?.is_file() {
                return Err(
                    std::io::Error::other("startup descriptor is not a regular file").into(),
                );
            }
            match file.try_lock() {
                Ok(()) => {
                    #[cfg(not(unix))]
                    {
                        file.set_len(0)?;
                        writeln!(&file, "pid={}", std::process::id())?;
                        file.sync_data()?;
                    }
                    observe(StartupLockObservation::Acquired, started.elapsed());
                    wait_phase.finish();
                    return Ok(Some(Self { file }));
                }
                Err(std::fs::TryLockError::WouldBlock) => {
                    if !waiting {
                        observe(StartupLockObservation::Waiting, started.elapsed());
                        waiting = true;
                    }
                    let remaining = timeout.saturating_sub(started.elapsed());
                    if tokio::time::timeout(remaining, ready())
                        .await
                        .unwrap_or(false)
                    {
                        observe(StartupLockObservation::ConcurrentReady, started.elapsed());
                        wait_phase.finish();
                        return Ok(None);
                    }
                    if started.elapsed() >= timeout {
                        #[cfg(unix)]
                        observe_startup_owner(&file);
                        observe(StartupLockObservation::Timeout, started.elapsed());
                        wait_phase.fail();
                        return Err(DaemonStartError::StartupCoordinationTimeout {
                            lock_path: path,
                        });
                    }
                    tokio::time::sleep(
                        Self::POLL_INTERVAL.min(timeout.saturating_sub(started.elapsed())),
                    )
                    .await;
                }
                Err(std::fs::TryLockError::Error(error)) => return Err(error.into()),
            }
        }
    }

    #[cfg(test)]
    async fn acquire_at(path: PathBuf, timeout: Duration) -> Result<Self, DaemonStartError> {
        if let Some(parent) = path.parent() {
            fs::create_dir_all(parent)?;
        }
        let started = std::time::Instant::now();
        loop {
            let file = fs::OpenOptions::new()
                .read(true)
                .write(true)
                .create(true)
                .truncate(false)
                .open(&path)?;
            match file.try_lock() {
                Ok(()) => {
                    file.set_len(0)?;
                    writeln!(&file, "pid={}", std::process::id())?;
                    file.sync_data()?;
                    return Ok(Self { file });
                }
                Err(std::fs::TryLockError::WouldBlock) => {
                    if started.elapsed() >= timeout {
                        return Err(DaemonStartError::StartupCoordinationTimeout {
                            lock_path: path,
                        });
                    }
                    tokio::time::sleep(Self::POLL_INTERVAL).await;
                }
                Err(std::fs::TryLockError::Error(error)) => return Err(error.into()),
            }
        }
    }
}

// Unix file locks are attached to the open file description. Passing a duplicate
// through stdin transfers ownership atomically with spawn, before child Rust code
// runs. Do not explicitly unlock the parent's description: that would also unlock
// the child's duplicate. The daemon retains stdin for its lifetime.
fn startup_lock_stdio(file: fs::File) -> std::process::Stdio {
    #[cfg(unix)]
    {
        std::process::Stdio::from(file)
    }
    #[cfg(not(unix))]
    {
        drop(file);
        std::process::Stdio::null()
    }
}

#[cfg(not(unix))]
impl Drop for StartupLock {
    fn drop(&mut self) {
        let _ = self.file.unlock();
    }
}

const READINESS_TIMEOUT: Duration = Duration::from_secs(20);
const READINESS_RETRY_DELAY: Duration = Duration::from_millis(25);

/// Notify the launching client that the daemon is ready to accept connections.
///
/// This private, versioned launch-channel signal is only a wakeup: callers must
/// still verify the normal IPC identity handshake. Foreground servers do nothing.
/// A departed launcher must not prevent the daemon from serving other clients.
pub fn notify_launcher_ready() {
    if std::env::var("BCODE_DAEMON_READY_STDOUT").as_deref() == Ok("v1") {
        #[cfg(unix)]
        if publish_daemon_startup_owner().is_err() {
            tracing::error!(target: "bcode_daemon_lifecycle::startup",
                "startup descriptor publication failed; readiness notification withheld");
            return;
        }
        let mut stdout = std::io::stdout().lock();
        let _ = stdout.write_all(b"BCODE_READY_V1\n");
        let _ = stdout.flush();
    }
}

async fn measured_child_readiness(
    endpoint: &IpcEndpoint,
    child: &mut tokio::process::Child,
    log_path: &Path,
) -> Result<(), DaemonStartError> {
    let phase = bcode_metrics::startup::phase("launcher.spawn_to_notification");
    let result = wait_for_child_notification(endpoint, child, log_path).await;
    phase.finish_result(&result);
    result
}

async fn wait_for_child_notification(
    endpoint: &IpcEndpoint,
    child: &mut tokio::process::Child,
    log_path: &Path,
) -> Result<(), DaemonStartError> {
    wait_for_child_notification_with_timeout(endpoint, child, log_path, READINESS_TIMEOUT).await
}

async fn wait_for_child_notification_with_timeout(
    endpoint: &IpcEndpoint,
    child: &mut tokio::process::Child,
    log_path: &Path,
    timeout: Duration,
) -> Result<(), DaemonStartError> {
    use tokio::io::AsyncReadExt as _;

    let started = std::time::Instant::now();
    let mut stdout = child
        .stdout
        .take()
        .ok_or_else(|| std::io::Error::other("daemon startup notification channel missing"))?;
    let mut token = [0_u8; 15];
    let notification = tokio::time::timeout(timeout, stdout.read_exact(&mut token));
    tokio::pin!(notification);
    tokio::select! {
        status = child.wait() => {
            return Err(DaemonStartError::Exited {
                status: status?.to_string(),
                log_path: display_from_current_dir(log_path).to_string(),
                recent_log: recent_log_excerpt(log_path),
            });
        }
        result = &mut notification => {
            match result {
                Ok(Ok(_)) if &token == b"BCODE_READY_V1\n" => {}
                Ok(_) => return Err(std::io::Error::other("invalid or closed daemon startup notification channel").into()),
                Err(_) => return Err(DaemonStartError::NotificationTimeout {
                    log_path: display_from_current_dir(log_path).to_string(),
                    recent_log: recent_log_excerpt(log_path),
                }),
            }
        }
    }
    tracing::debug!(
        target: "bcode_daemon_lifecycle::startup",
        elapsed_us = started.elapsed().as_micros(),
        "daemon readiness notification received"
    );
    // A notification is only a wakeup, not verified readiness.
    let remaining = timeout.saturating_sub(started.elapsed());
    let phase = bcode_metrics::startup::phase("launcher.verified_handshake");
    let result = tokio::time::timeout(remaining, wait_for_daemon_ready(endpoint, log_path))
        .await
        .unwrap_or_else(|_| {
            Err(DaemonStartError::HealthCheckFailed {
                log_path: display_from_current_dir(log_path).to_string(),
                recent_log: recent_log_excerpt(log_path),
            })
        });
    phase.finish_result(&result);
    result
}

async fn wait_for_daemon_ready(
    endpoint: &IpcEndpoint,
    log_path: &Path,
) -> Result<(), DaemonStartError> {
    let deadline = tokio::time::Instant::now() + READINESS_TIMEOUT;
    loop {
        if ping_ready(endpoint).await {
            return Ok(());
        }
        if tokio::time::Instant::now() >= deadline {
            return Err(DaemonStartError::StartTimeout {
                log_path: display_from_current_dir(log_path).to_string(),
                recent_log: recent_log_excerpt(log_path),
            });
        }
        tokio::time::sleep(READINESS_RETRY_DELAY).await;
    }
}

async fn ping_ready(endpoint: &IpcEndpoint) -> bool {
    matches!(
        tokio::time::timeout(Duration::from_millis(500), async {
            let Some(status) = probe_daemon_status(endpoint).await else {
                return false;
            };
            daemon_status_matches_current_executable(&status)
        })
        .await,
        Ok(true)
    )
}

fn daemon_status_matches_current_executable(status: &bcode_ipc::DaemonStatus) -> bool {
    status.namespace == daemon_namespace()
        && initialize_artifact_bootstrap()
            .and_then(ArtifactBootstrap::digest)
            .is_ok_and(|digest| status.executable_digest.as_deref() == Some(digest.as_str()))
        && status.protocol_version == u32::from(CURRENT_PROTOCOL_VERSION)
        && status.artifact_id.as_ref() == Some(&bcode_ipc::ArtifactId::current())
        && status.build_fingerprint == BUILD_FINGERPRINT
        && status.storage_writer_epoch == Some(bcode_ipc::CURRENT_SESSION_STORAGE_WRITER_EPOCH)
        && status.session_event_schema_version
            == Some(bcode_session_models::CURRENT_SESSION_EVENT_SCHEMA_VERSION)
}

#[cfg(unix)]
async fn wait_for_existing_daemon(endpoint: &IpcEndpoint) -> bool {
    for delay in [50, 100, 200, 400, 800, 1_000] {
        if ping_ready(endpoint).await {
            return true;
        }
        tokio::time::sleep(Duration::from_millis(delay)).await;
    }
    false
}

#[cfg(unix)]
async fn endpoint_has_listener(endpoint: &IpcEndpoint) -> bool {
    let Some(path) = endpoint.as_unix_socket() else {
        return false;
    };
    let path = path.to_path_buf();
    tokio::task::spawn_blocking(move || unix_socket_has_listener(&path))
        .await
        .unwrap_or(true)
}

fn remove_stale_plugin_state(
    state_dir: &Path,
    namespace: &str,
) -> Result<(), DaemonLifecycleError> {
    if namespace.is_empty()
        || namespace.len() > 256
        || !namespace
            .bytes()
            .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_'))
    {
        return Ok(());
    }
    let plugins_root = state_dir.join("derived").join("plugins");
    let namespace_root = plugins_root.join(namespace);
    let metadata = match fs::symlink_metadata(&namespace_root) {
        Ok(metadata) => metadata,
        Err(source) if source.kind() == std::io::ErrorKind::NotFound => return Ok(()),
        Err(source) => {
            return Err(DaemonLifecycleError::Io {
                path: namespace_root,
                source,
            });
        }
    };
    if !metadata.file_type().is_dir() || metadata.file_type().is_symlink() {
        return Ok(());
    }
    fs::remove_dir_all(&namespace_root).map_err(|source| DaemonLifecycleError::Io {
        path: namespace_root.clone(),
        source,
    })?;
    if fs::read_dir(&plugins_root).is_ok_and(|mut entries| entries.next().is_none()) {
        let _ = fs::remove_dir(&plugins_root);
    }
    Ok(())
}

async fn cleanup_stale_daemon_records() -> Result<(), DaemonLifecycleError> {
    let state_dir = bcode_config::default_state_dir();
    for (path, record) in read_records(&state_dir) {
        if record.is_current_namespace() {
            continue;
        }
        if !matches!(
            classify_daemon_record(&record).await,
            DaemonRecordClassification::UnreachableStale
        ) {
            continue;
        }
        remove_stale_plugin_state(&state_dir, &record.namespace)?;
        remove_record_path(&path)?;
        remove_stale_socket(&record);
    }
    Ok(())
}

#[cfg(unix)]
fn remove_stale_socket(record: &DaemonRecord) {
    if let DaemonEndpointRecord::UnixSocket { path } = &record.endpoint {
        let _ = remove_stale_unix_socket_path(path);
    }
}

#[cfg(not(unix))]
const fn remove_stale_socket(_record: &DaemonRecord) {}

#[cfg(unix)]
fn cleanup_stale_endpoint(endpoint: &IpcEndpoint) -> Result<(), DaemonLifecycleError> {
    if let Some(path) = endpoint.as_unix_socket() {
        remove_stale_unix_socket_path(path)?;
    }
    Ok(())
}

#[cfg(unix)]
fn remove_stale_unix_socket_path(path: &Path) -> Result<(), DaemonLifecycleError> {
    if !path.exists() || !is_bcode_socket_path(path) || unix_socket_has_listener(path) {
        return Ok(());
    }
    std::thread::sleep(Duration::from_millis(100));
    if unix_socket_has_listener(path) {
        return Ok(());
    }
    match fs::remove_file(path) {
        Ok(()) => Ok(()),
        Err(source) if source.kind() == std::io::ErrorKind::NotFound => Ok(()),
        Err(source) => Err(DaemonLifecycleError::Io {
            path: path.to_path_buf(),
            source,
        }),
    }
}

#[cfg(unix)]
fn is_bcode_socket_path(path: &Path) -> bool {
    path.file_name()
        .and_then(|name| name.to_str())
        .is_some_and(|name| {
            name.starts_with("bcode-")
                && Path::new(name)
                    .extension()
                    .is_some_and(|extension| extension.eq_ignore_ascii_case("sock"))
        })
}

#[cfg(unix)]
fn unix_socket_has_listener(path: &Path) -> bool {
    std::os::unix::net::UnixStream::connect(path).is_ok()
}

#[cfg(test)]
fn run_isolated_fork_fixture(test_name: &str) -> bool {
    const ENV: &str = "BCODE_ISOLATED_FORK_FIXTURE";
    if std::env::var(ENV).as_deref() == Ok(test_name) {
        return false;
    }
    // A gated pre-exec/fork child copies every live OFD in its process, not
    // merely this fixture's handles. Execute the owning test alone so that its
    // intentional gate cannot retain unrelated parallel-test fences.
    let mut child = std::process::Command::new(std::env::current_exe().unwrap())
        .args([test_name, "--exact", "--nocapture"])
        .env(ENV, test_name)
        .stdin(std::process::Stdio::null())
        .spawn()
        .expect("isolated fork fixture spawn");
    let deadline = std::time::Instant::now() + Duration::from_secs(15);
    loop {
        if let Some(status) = child.try_wait().expect("isolated fixture status") {
            assert!(
                status.success(),
                "isolated fixture failed: {test_name}: {status}"
            );
            return true;
        }
        if std::time::Instant::now() >= deadline {
            child.kill().expect("terminate timed-out fixture");
            child.wait().expect("reap timed-out fixture");
            panic!("isolated fixture deadline exceeded: {test_name}");
        }
        std::thread::sleep(Duration::from_millis(10));
    }
}

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

    #[tokio::test]
    async fn startup_lock_observations_timeout_preserve_held_file() {
        let directory = tempfile::tempdir().expect("lock directory");
        let path = directory.path().join("startup.lock");
        let owner = StartupLock::acquire_at(path.clone(), Duration::from_secs(1))
            .await
            .expect("owner");
        let original = fs::read(&path).expect("owner hint");
        let mut observations = Vec::new();
        let result = StartupLock::acquire_observed(
            path.clone(),
            Duration::from_millis(50),
            "test-artifact.test-state",
            || std::future::ready(false),
            |event, elapsed| observations.push((event, elapsed)),
        )
        .await;
        assert!(matches!(
            result,
            Err(DaemonStartError::StartupCoordinationTimeout { .. })
        ));
        assert_eq!(observations.len(), 2);
        assert_eq!(observations[0].0, StartupLockObservation::Waiting);
        assert_eq!(observations[1].0, StartupLockObservation::Timeout);
        assert!(observations[1].1 >= Duration::from_millis(50));
        assert_eq!(fs::read(&path).expect("preserved hint"), original);
        drop(owner);
    }

    #[tokio::test]
    async fn startup_lock_observations_concurrent_ready_does_not_acquire() {
        let directory = tempfile::tempdir().expect("lock directory");
        let path = directory.path().join("startup.lock");
        let owner = StartupLock::acquire_at(path.clone(), Duration::from_secs(1))
            .await
            .expect("owner");
        let mut observations = Vec::new();
        let result = StartupLock::acquire_observed(
            path,
            Duration::from_millis(50),
            "test-artifact.test-state",
            || std::future::ready(true),
            |event, _| observations.push(event),
        )
        .await
        .expect("ready observation");
        assert!(result.is_none());
        assert_eq!(
            observations,
            [
                StartupLockObservation::Waiting,
                StartupLockObservation::ConcurrentReady
            ]
        );
        drop(owner);
    }

    #[tokio::test]
    async fn startup_lock_observations_acquired_after_contention() {
        let directory = tempfile::tempdir().expect("lock directory");
        let path = directory.path().join("startup.lock");
        let mut owner = Some(
            StartupLock::acquire_at(path.clone(), Duration::from_secs(1))
                .await
                .expect("owner"),
        );
        let mut observations = Vec::new();
        let replacement = StartupLock::acquire_observed(
            path,
            Duration::from_secs(1),
            "test-artifact.test-state",
            || {
                drop(owner.take());
                std::future::ready(false)
            },
            |event, _| observations.push(event),
        )
        .await
        .expect("replacement")
        .expect("acquired lock");
        assert_eq!(
            observations,
            [
                StartupLockObservation::Waiting,
                StartupLockObservation::Acquired
            ]
        );
        drop(replacement);
    }

    fn lock_path(name: &str) -> PathBuf {
        std::env::temp_dir().join(format!(
            "bcode-daemon-startup-lock-{name}-{}-{}",
            std::process::id(),
            unix_time_millis().expect("time")
        ))
    }

    #[cfg(unix)]
    #[test]
    fn stale_socket_cleanup_preserves_live_listener_then_removes_stale_path() {
        let root = PathBuf::from(format!(
            "/tmp/bcode-ds-{}-{}",
            std::process::id(),
            unix_time_millis().expect("time")
        ));
        fs::create_dir_all(&root).expect("socket directory");
        let live_path = root.join("bcode-foreign-artifact.sock");
        let listener =
            std::os::unix::net::UnixListener::bind(&live_path).expect("bind live listener");

        remove_stale_unix_socket_path(&live_path).expect("preserve live endpoint");
        assert!(live_path.exists());
        let connection = std::os::unix::net::UnixStream::connect(&live_path)
            .expect("live foreign endpoint remains reachable");
        drop(connection);
        drop(listener);

        let stale_path = root.join("bcode-stale-artifact.sock");
        drop(std::os::unix::net::UnixDatagram::bind(&stale_path).expect("bind stale socket"));
        remove_stale_unix_socket_path(&stale_path).expect("remove stale endpoint");
        assert!(!stale_path.exists());
        fs::remove_file(live_path).expect("live socket cleanup");
        fs::remove_dir_all(root).expect("socket directory cleanup");
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn occupied_unresponsive_endpoint_does_not_start_competing_daemon() {
        let root = PathBuf::from(format!(
            "/tmp/bcode-do-{}-{}",
            std::process::id(),
            unix_time_millis().expect("time")
        ));
        fs::create_dir_all(&root).expect("socket directory");
        let socket_path = root.join("bcode-occupied.sock");
        let listener =
            std::os::unix::net::UnixListener::bind(&socket_path).expect("bind occupied endpoint");
        let options = EnsureDaemonOptions {
            endpoint: IpcEndpoint::unix_socket(socket_path.clone()),
            quiet: true,
            log_path: root.join("daemon.log"),
        };
        let starts = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let observed_starts = std::sync::Arc::clone(&starts);
        let error = ensure_daemon_running_with_start(&options, move |_, _| {
            observed_starts.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
            async { Ok(()) }
        })
        .await
        .expect_err("occupied endpoint must block startup");

        assert!(matches!(error, DaemonStartError::EndpointOccupied { .. }));
        assert_eq!(starts.load(std::sync::atomic::Ordering::SeqCst), 0);
        drop(listener);
        fs::remove_file(socket_path).expect("socket cleanup");
        fs::remove_dir_all(root).expect("directory cleanup");
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn notification_timeout_preserves_failure_and_does_not_terminate_child() {
        let directory = tempfile::tempdir().expect("notification directory");
        let endpoint = IpcEndpoint::unix_socket(directory.path().join("absent.sock"));
        let release = directory.path().join("release");
        let late = directory.path().join("late");
        let mut child = tokio::process::Command::new("sh")
            .args([
                "-c",
                "trap '' PIPE; while [ ! -f \"$1\" ]; do sleep 0.01; done; printf late > \"$2\"; printf 'BCODE_READY_V1\\n' 2>/dev/null || :; exec sleep 30",
                "notification-test",
            ])
            .arg(&release)
            .arg(&late)
            .stdout(std::process::Stdio::piped())
            .kill_on_drop(true)
            .spawn()
            .expect("notification child");
        let result = wait_for_child_notification_with_timeout(
            &endpoint,
            &mut child,
            &directory.path().join("daemon.log"),
            Duration::ZERO,
        )
        .await;
        assert!(matches!(
            result,
            Err(DaemonStartError::NotificationTimeout { .. })
        ));
        assert!(child.try_wait().expect("observe child").is_none());
        fs::write(&release, b"release").expect("release late notification");
        tokio::time::timeout(Duration::from_secs(5), async {
            while !late.exists() {
                tokio::time::sleep(Duration::from_millis(10)).await;
            }
        })
        .await
        .expect("child continued after deadline");
        child.kill().await.expect("stop child");
        // A later notification cannot change the already returned failure into readiness.
        assert!(matches!(
            result,
            Err(DaemonStartError::NotificationTimeout { .. })
        ));
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn readiness_notification_does_not_substitute_for_identity_verification() {
        let directory = tempfile::tempdir().expect("notification directory");
        let endpoint = IpcEndpoint::unix_socket(directory.path().join("absent.sock"));
        let published = directory.path().join("notification-published");
        let mut child = tokio::process::Command::new("sh")
            .args([
                "-c",
                "printf 'BCODE_READY_V1\\n'; printf ready > \"$1\"; exec sleep 30",
                "notification",
            ])
            .arg(&published)
            .stdout(std::process::Stdio::piped())
            .kill_on_drop(true)
            .spawn()
            .expect("notification child");
        // This test targets identity verification after notification, not shell
        // scheduling within the 100ms handshake budget. Gate that boundary.
        tokio::time::timeout(Duration::from_secs(3), async {
            while !published.exists() {
                assert!(child.try_wait().expect("fixture status").is_none());
                tokio::time::sleep(Duration::from_millis(10)).await;
            }
        })
        .await
        .expect("fixture did not publish readiness notification");
        let result = super::wait_for_child_notification_with_timeout(
            &endpoint,
            &mut child,
            &directory.path().join("daemon.log"),
            Duration::from_millis(100),
        )
        .await;
        child.kill().await.expect("stop child");
        assert!(
            matches!(&result, Err(DaemonStartError::HealthCheckFailed { .. })),
            "expected verified-handshake failure after published notification; actual={result:?}"
        );
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn unsupported_readiness_notification_fails_closed() {
        let directory = tempfile::tempdir().expect("notification directory");
        let endpoint = IpcEndpoint::unix_socket(directory.path().join("absent.sock"));
        let mut child = tokio::process::Command::new("sh")
            .args(["-c", "printf 'BCODE_READY_V2\\n'; exec sleep 30"])
            .stdout(std::process::Stdio::piped())
            .kill_on_drop(true)
            .spawn()
            .expect("notification child");
        let result = super::wait_for_child_notification(
            &endpoint,
            &mut child,
            &directory.path().join("daemon.log"),
        )
        .await;
        child.kill().await.expect("stop child");
        assert!(matches!(result, Err(DaemonStartError::Io(_))));
    }

    #[tokio::test]
    async fn startup_locks_isolate_scopes_and_fence_same_scope() {
        let directory = tempfile::tempdir().unwrap();
        let first_path = super::startup_lock_path(directory.path(), "artifact", "scope-a");
        std::fs::create_dir_all(first_path.parent().unwrap()).unwrap();
        let first = StartupLock::acquire_at(first_path.clone(), Duration::from_secs(1))
            .await
            .unwrap();
        let second = StartupLock::acquire_at(
            super::startup_lock_path(directory.path(), "artifact", "scope-b"),
            Duration::from_secs(1),
        )
        .await
        .unwrap();
        assert!(matches!(
            StartupLock::acquire_at(first_path, Duration::from_millis(50)).await,
            Err(DaemonStartError::StartupCoordinationTimeout { .. })
        ));
        drop((first, second));
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn startup_lock_probe_timeout_is_bounded_and_preserves_descriptor() {
        let directory = tempfile::tempdir().unwrap();
        let path = directory.path().join("startup.lock");
        let owner = StartupLock::acquire_at(path.clone(), Duration::from_secs(1))
            .await
            .unwrap();
        let descriptor =
            super::StartupOwnerDescriptor::current("handoff_pending", "test".to_owned()).unwrap();
        super::write_startup_owner(&owner.file, &descriptor).unwrap();
        let original = fs::read(&path).unwrap();
        let result = tokio::time::timeout(
            Duration::from_secs(1),
            StartupLock::acquire_observed(
                path.clone(),
                Duration::from_millis(40),
                "test",
                std::future::pending::<bool>,
                |_, _| {},
            ),
        )
        .await
        .unwrap();
        assert!(matches!(
            result,
            Err(DaemonStartError::StartupCoordinationTimeout { .. })
        ));
        assert_eq!(fs::read(&path).unwrap(), original);
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn startup_descriptor_symlink_is_rejected_without_modifying_target() {
        let directory = tempfile::tempdir().unwrap();
        let target = directory.path().join("target");
        let path = directory.path().join("startup.lock");
        fs::write(&target, b"preserve").unwrap();
        std::os::unix::fs::symlink(&target, &path).unwrap();
        assert!(
            StartupLock::acquire_observed(
                path,
                Duration::from_millis(40),
                "test",
                || std::future::ready(false),
                |_, _| {}
            )
            .await
            .is_err()
        );
        assert_eq!(fs::read(target).unwrap(), b"preserve");
    }

    #[cfg(unix)]
    #[test]
    fn startup_descriptor_handoff_child() {
        use std::os::fd::BorrowedFd;

        let Ok(path) = std::env::var("BCODE_TEST_STARTUP_DESCRIPTOR") else {
            return;
        };
        let file = fs::File::from(
            unsafe { BorrowedFd::borrow_raw(0) }
                .try_clone_to_owned()
                .unwrap(),
        );
        super::publish_startup_owner_at(&file, Path::new(&path), "test-handoff").unwrap();
        super::close_startup_stdin_on_exec().unwrap();
        // A surviving exec descendant must not retain the daemon's lifetime fence.
        let mut grandchild_command = std::process::Command::new("sh");
        let grandchild_release = format!("{path}.grandchild-release");
        let mut grandchild = grandchild_command
            .args([
                "-c",
                "n=0; while [ ! -f \"$1\" ] && [ \"$n\" -lt 500 ]; do sleep 0.01; n=$((n+1)); done",
                "grandchild",
            ])
            .arg(&grandchild_release)
            .stdout(std::process::Stdio::null())
            .stderr(std::process::Stdio::null())
            .spawn()
            .unwrap();
        drop(grandchild_command);
        fs::write(
            format!("{path}.grandchild-pid"),
            grandchild.id().to_string(),
        )
        .unwrap();
        // Own the wait while this fixture process lives. On deliberate daemon
        // exit its surviving descendant is reparented to the OS reaper.
        std::thread::spawn(move || grandchild.wait().expect("grandchild wait"));
        println!("descriptor-published");
        std::io::stdout().flush().unwrap();
        // Separate release channel: never read the descriptor through shared stdin.
        let release = format!("{path}.release");
        let deadline = std::time::Instant::now() + Duration::from_secs(5);
        while !Path::new(&release).exists() {
            assert!(std::time::Instant::now() < deadline);
            std::thread::sleep(Duration::from_millis(10));
        }
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn startup_descriptor_detached_child_replaces_launcher_claim_and_retains_fence() {
        use tokio::io::{AsyncBufReadExt as _, BufReader};

        const FIXTURE: &str = "BCODE_TEST_DESCRIPTOR_ISOLATED";
        if std::env::var_os(FIXTURE).is_none() {
            let mut child = tokio::process::Command::new(std::env::current_exe().unwrap())
                .args(["startup_lock_tests::startup_descriptor_detached_child_replaces_launcher_claim_and_retains_fence", "--exact", "--nocapture"])
                .env(FIXTURE, "1")
                .stdin(std::process::Stdio::null())
                .kill_on_drop(true)
                .spawn().unwrap();
            assert!(
                tokio::time::timeout(Duration::from_secs(10), child.wait())
                    .await
                    .unwrap()
                    .unwrap()
                    .success()
            );
            return;
        }
        let directory = tempfile::tempdir().unwrap();
        let path = directory.path().join("startup.lock");
        let owner = StartupLock::acquire_at(path.clone(), Duration::from_secs(1))
            .await
            .unwrap();
        let pending =
            super::StartupOwnerDescriptor::current("handoff_pending", "test-handoff".to_owned())
                .unwrap();
        super::write_startup_owner(&owner.file, &pending).unwrap();
        let mut command = tokio::process::Command::new(std::env::current_exe().unwrap());
        super::detach_daemon_process(&mut command);
        let mut child = command
            .args([
                "startup_lock_tests::startup_descriptor_handoff_child",
                "--exact",
                "--nocapture",
            ])
            .env("BCODE_TEST_STARTUP_DESCRIPTOR", &path)
            .stdin(super::startup_lock_stdio(owner.file.try_clone().unwrap()))
            .stdout(std::process::Stdio::piped())
            .kill_on_drop(true)
            .spawn()
            .unwrap();
        // Command retains its configured Stdio even after spawn; it must not
        // remain a second launcher-side owner during the release assertion.
        drop(command);
        let child_pid = child.id().unwrap();
        let mut lines = BufReader::new(child.stdout.take().unwrap()).lines();
        let mut published = false;
        tokio::time::timeout(Duration::from_secs(3), async {
            while let Some(line) = lines.next_line().await.unwrap() {
                if line.contains("descriptor-published") {
                    published = true;
                    break;
                }
            }
        })
        .await
        .unwrap();
        assert!(published);
        drop(owner);
        let file = fs::OpenOptions::new()
            .read(true)
            .write(true)
            .open(&path)
            .unwrap();
        let actual = super::read_startup_owner(&file).unwrap().unwrap();
        assert_eq!(actual.pid, child_pid);
        assert_ne!(actual.pid, pending.pid);
        assert_eq!(actual.phase, "daemon_ready");
        assert!(matches!(
            file.try_lock(),
            Err(std::fs::TryLockError::WouldBlock)
        ));
        assert!(super::publish_startup_owner_at(&file, &path, "test-handoff").is_err());
        assert_eq!(
            super::read_startup_owner(&file).unwrap().unwrap().pid,
            child_pid
        );
        fs::write(path.with_extension("lock.release"), b"release").unwrap();
        assert!(
            tokio::time::timeout(Duration::from_secs(3), child.wait())
                .await
                .unwrap()
                .unwrap()
                .success()
        );
        let grandchild_pid: u32 = fs::read_to_string(format!("{}.grandchild-pid", path.display()))
            .unwrap()
            .parse()
            .unwrap();
        // SAFETY: signal zero only observes the isolated fixture descendant.
        assert_eq!(
            unsafe { libc::kill(i32::try_from(grandchild_pid).unwrap(), 0) },
            0
        );
        let release_result = file.try_lock();
        // Release the fixture descendant even if the lock assertion fails.
        fs::write(format!("{}.grandchild-release", path.display()), b"release").unwrap();
        release_result.unwrap();
    }

    #[cfg(unix)]
    #[test]
    fn startup_descriptor_rejects_foreign_future_legacy_and_reused_process() {
        let directory = tempfile::tempdir().unwrap();
        let path = directory.path().join("startup.lock");
        let file = fs::OpenOptions::new()
            .create(true)
            .truncate(true)
            .read(true)
            .write(true)
            .open(&path)
            .unwrap();
        file.lock().unwrap();
        let mut owner =
            super::StartupOwnerDescriptor::current("handoff_pending", "test".to_owned()).unwrap();
        owner.process_start_seconds = owner.process_start_seconds.saturating_add(1);
        assert!(!owner.process_verified());
        for (version, state) in [(2, owner.state.clone()), (1, "foreign".to_owned())] {
            owner.version = version;
            owner.state = state;
            super::write_startup_owner(&file, &owner).unwrap();
            let before = fs::read(&path).unwrap();
            assert!(super::publish_startup_owner_at(&file, &path, "test").is_err());
            assert_eq!(fs::read(&path).unwrap(), before);
        }
        fs::write(&path, b"pid=123\n").unwrap();
        assert!(super::read_startup_owner(&file).unwrap().is_none());
        assert!(super::publish_startup_owner_at(&file, &path, "test").is_err());
        let other = directory.path().join("other.lock");
        fs::write(&other, b"").unwrap();
        assert!(super::publish_startup_owner_at(&file, &other, "test").is_err());
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn inherited_startup_lock_fences_launchers_until_child_exit() {
        let directory = tempfile::tempdir().expect("lock directory");
        let path = directory.path().join("startup.lock");
        let owner = StartupLock::acquire_at(path.clone(), Duration::from_secs(1))
            .await
            .expect("launcher lock");
        let inherited = owner.file.try_clone().expect("inherited lock");
        let mut child = tokio::process::Command::new("sleep")
            .arg("30")
            .stdin(super::startup_lock_stdio(inherited))
            .kill_on_drop(true)
            .spawn()
            .expect("spawn child before endpoint publication");
        drop(owner);
        let result = StartupLock::acquire_at(path.clone(), Duration::from_millis(75)).await;
        assert!(matches!(
            result,
            Err(DaemonStartError::StartupCoordinationTimeout { .. })
        ));
        child.kill().await.expect("stop child");
        let replacement = StartupLock::acquire_at(path, Duration::from_secs(1))
            .await
            .expect("ownership released only after child exit");
        drop(replacement);
    }

    #[cfg(unix)]
    struct HandshakeChild {
        release: Option<std::io::PipeWriter>,
        spawning: Option<std::thread::JoinHandle<std::io::Result<std::process::Child>>>,
        child: Option<std::process::Child>,
        readers: Vec<std::thread::JoinHandle<()>>,
    }

    #[cfg(unix)]
    static INCOMPLETE_HANDSHAKE_CLEANUP: std::sync::Mutex<Vec<HandshakeChild>> =
        std::sync::Mutex::new(Vec::new());

    #[cfg(unix)]
    impl HandshakeChild {
        fn release_preexec(&mut self) {
            // At most two one-byte releases are written to a fresh pipe, so
            // this cannot fill its buffer. EOF alone cannot release the child:
            // it inherited a copy of the parent's write end too.
            if let Some(mut release) = self.release.take() {
                use std::io::Write as _;
                let _ = release.write_all(b"G");
            }
        }

        fn collect_spawn_if_finished(&mut self) {
            if self
                .spawning
                .as_ref()
                .is_some_and(std::thread::JoinHandle::is_finished)
                && let Some(spawning) = self.spawning.take()
                && let Ok(Ok(child)) = spawning.join()
            {
                self.child = Some(child);
            }
        }

        fn cleanup_until(&mut self, deadline: std::time::Instant) -> bool {
            self.release_preexec();
            loop {
                self.collect_spawn_if_finished();
                if let Some(child) = self.child.as_mut() {
                    let _ = child.kill();
                    // try_wait reaps an exited child without blocking. Keep
                    // ownership on errors as well as a still-running result.
                    if matches!(child.try_wait(), Ok(Some(_))) {
                        self.child = None;
                    }
                }
                let mut index = 0;
                while index < self.readers.len() {
                    if self.readers[index].is_finished() {
                        let _ = self.readers.swap_remove(index).join();
                    } else {
                        index += 1;
                    }
                }
                if self.spawning.is_none() && self.child.is_none() && self.readers.is_empty() {
                    return true;
                }
                if std::time::Instant::now() >= deadline {
                    return false;
                }
                std::thread::yield_now();
            }
        }
    }

    #[cfg(unix)]
    impl Drop for HandshakeChild {
        fn drop(&mut self) {
            if !self.cleanup_until(std::time::Instant::now() + Duration::from_secs(1)) {
                eprintln!(
                    "handshake_cleanup_timeout: budget_ms=1000, spawn_pending={}, child_pid={:?}, readers_pending={}, ownership=retained",
                    self.spawning.is_some(),
                    self.child.as_ref().map(std::process::Child::id),
                    self.readers.len()
                );
                // Static storage deliberately retains all unfinished ownership
                // until test-process exit. Never detach join handles, discard
                // an unreaped child, or falsely report cleanup completion.
                let retained = Self {
                    release: self.release.take(),
                    spawning: self.spawning.take(),
                    child: self.child.take(),
                    readers: std::mem::take(&mut self.readers),
                };
                match INCOMPLETE_HANDSHAKE_CLEANUP.try_lock() {
                    Ok(mut retained_cleanup) => retained_cleanup.push(retained),
                    Err(std::sync::TryLockError::Poisoned(error)) => {
                        error.into_inner().push(retained);
                    }
                    Err(std::sync::TryLockError::WouldBlock) => {
                        // Even the retention registry must not block Drop.
                        // Preserve an addressable owner rather than dropping
                        // its child or detaching its join handles.
                        let retained = Box::leak(Box::new(retained));
                        eprintln!("handshake_cleanup_retained_owner: address={retained:p}");
                    }
                }
            }
        }
    }

    #[cfg(unix)]
    fn handshake_marker(
        reader: impl std::io::Read + Send + 'static,
        owner: &mut HandshakeChild,
    ) -> std::sync::mpsc::Receiver<std::io::Result<u8>> {
        let (sender, receiver) = std::sync::mpsc::channel();
        owner.readers.push(std::thread::spawn(move || {
            let mut reader = reader;
            let mut marker = [0_u8];
            let result = reader.read_exact(&mut marker).map(|()| marker[0]);
            let _ = sender.send(result);
        }));
        receiver
    }

    #[cfg(unix)]
    fn spawn_handshake_child(
        ready_write: std::io::PipeWriter,
        release_read: std::io::PipeReader,
        emit_ready: bool,
    ) -> std::thread::JoinHandle<std::io::Result<std::process::Child>> {
        use std::os::fd::AsRawFd as _;
        use std::os::unix::process::CommandExt as _;

        let ready_fd = ready_write.as_raw_fd();
        let release_fd = release_read.as_raw_fd();
        std::thread::spawn(move || {
            let mut command = std::process::Command::new("sh");
            command
                .args(["-c", "printf E; exec sleep 30"])
                .stdin(std::process::Stdio::null())
                .stdout(std::process::Stdio::piped())
                .stderr(std::process::Stdio::null());
            // SAFETY: the child closure only uses async-signal-safe pipe write,
            // read and _exit, stack bytes, and returns Ok after release. No
            // allocation, locks, logging, or runtime callbacks occur after fork.
            // The spawning thread owns both descriptors until spawn returns;
            // the parent guard sends release, observes finished spawn, then
            // kills and polls reaping within its deadline. On timeout it retains
            // every unfinished owner, including during assertion unwinding.
            unsafe {
                command.pre_exec(move || {
                    let ready = b'R';
                    if emit_ready
                        && libc::write(ready_fd, std::ptr::from_ref(&ready).cast(), 1) != 1
                    {
                        libc::_exit(125);
                    }
                    let mut release = 0_u8;
                    if libc::read(release_fd, std::ptr::from_mut(&mut release).cast(), 1) != 1
                        || release != b'G'
                    {
                        libc::_exit(126);
                    }
                    Ok(())
                });
            }
            let result = command.spawn();
            drop(command);
            drop((ready_write, release_read));
            result
        })
    }

    #[cfg(unix)]
    fn release_child_and_observe_exec(owner: &mut HandshakeChild) {
        use std::io::Write as _;

        owner
            .release
            .as_mut()
            .expect("release writer")
            .write_all(b"G")
            .expect("release child to exec");
        let deadline = std::time::Instant::now() + Duration::from_secs(1);
        while owner.spawning.is_some() {
            owner.collect_spawn_if_finished();
            assert!(
                owner.spawning.is_none() || std::time::Instant::now() < deadline,
                "exec_spawn_timeout: budget_ms=1000, ownership=retained"
            );
            std::thread::yield_now();
        }
        let stdout = owner
            .child
            .as_mut()
            .expect("owned child")
            .stdout
            .take()
            .expect("post-exec readiness pipe");
        let postexec_ready = handshake_marker(stdout, owner);
        assert_eq!(
            postexec_ready
                .recv_timeout(Duration::from_secs(1))
                .expect("post-exec readiness deadline")
                .expect("post-exec readiness byte"),
            b'E'
        );
    }

    #[cfg(unix)]
    async fn observe_lock_release(
        contender: &fs::File,
        path: &Path,
        deadline: tokio::time::Instant,
    ) {
        // Local close need not release a description copied by another test's
        // pre-exec child. Reuse the existing 75 ms observation budget; never
        // delete the file, unlock another description, or wait for child exit.
        loop {
            match contender.try_lock() {
                Ok(()) => break,
                Err(std::fs::TryLockError::WouldBlock) => {
                    assert!(
                        tokio::time::Instant::now() < deadline,
                        "lock_release_timeout: fixture={}, budget_ms=75, holder=unknown",
                        path.display()
                    );
                    tokio::task::yield_now().await;
                }
                Err(error) => panic!(
                    "unexpected lock release error: fixture={}, error={error}",
                    path.display()
                ),
            }
        }
    }

    #[cfg(unix)]
    #[test]
    fn startup_lock_missing_preexec_readiness_has_bounded_owned_cleanup() {
        if super::run_isolated_fork_fixture(
            "startup_lock_tests::startup_lock_missing_preexec_readiness_has_bounded_owned_cleanup",
        ) {
            return;
        }
        let (ready_read, ready_write) = std::io::pipe().expect("ready pipe");
        let (release_read, release_write) = std::io::pipe().expect("release pipe");
        let mut owner = HandshakeChild {
            release: Some(release_write),
            spawning: None,
            child: None,
            readers: Vec::new(),
        };
        let readiness = handshake_marker(ready_read, &mut owner);
        owner.spawning = Some(spawn_handshake_child(ready_write, release_read, false));
        assert!(matches!(
            readiness.recv_timeout(Duration::from_millis(75)),
            Err(std::sync::mpsc::RecvTimeoutError::Timeout)
        ));
        assert!(owner.cleanup_until(std::time::Instant::now() + Duration::from_secs(1)));
        assert!(owner.spawning.is_none() && owner.child.is_none() && owner.readers.is_empty());
    }

    #[cfg(unix)]
    #[test]
    fn startup_lock_cleanup_expiry_preserves_unfinished_reader_ownership() {
        let (reader, writer) = std::io::pipe().expect("held readiness pipe");
        let mut owner = HandshakeChild {
            release: None,
            spawning: None,
            child: None,
            readers: Vec::new(),
        };
        let readiness = handshake_marker(reader, &mut owner);
        assert!(matches!(
            readiness.recv_timeout(Duration::from_millis(75)),
            Err(std::sync::mpsc::RecvTimeoutError::Timeout)
        ));
        assert!(!owner.cleanup_until(std::time::Instant::now()));
        assert_eq!(owner.readers.len(), 1);
        drop(writer);
        assert!(owner.cleanup_until(std::time::Instant::now() + Duration::from_secs(1)));
        assert!(owner.readers.is_empty());
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn startup_lock_preexec_listener_retention_ends_at_exec() {
        use std::os::fd::AsRawFd as _;

        if super::run_isolated_fork_fixture(
            "startup_lock_tests::startup_lock_preexec_listener_retention_ends_at_exec",
        ) {
            return;
        }
        let directory = tempfile::tempdir().expect("handshake directory");
        let lock_path = directory.path().join("startup.lock");
        let lock = StartupLock::acquire_at(lock_path.clone(), Duration::from_secs(1))
            .await
            .expect("fixture startup lock");
        let contender = fs::OpenOptions::new()
            .read(true)
            .write(true)
            .open(&lock_path)
            .expect("independent fixture contender");
        // SAFETY: F_GETFD only inspects this live, owned fixture description.
        let lock_flags = unsafe { libc::fcntl(lock.file.as_raw_fd(), libc::F_GETFD) };
        assert!(lock_flags >= 0 && lock_flags & libc::FD_CLOEXEC != 0);
        let socket = directory.path().join("daemon.sock");
        let listener = std::os::unix::net::UnixListener::bind(&socket).expect("listener");
        // SAFETY: F_GETFD only inspects this live, owned listener descriptor.
        let flags = unsafe { libc::fcntl(listener.as_raw_fd(), libc::F_GETFD) };
        assert!(flags >= 0 && flags & libc::FD_CLOEXEC != 0);
        let (ready_read, ready_write) = std::io::pipe().expect("pre-exec ready pipe");
        let (release_read, release_write) = std::io::pipe().expect("pre-exec release pipe");
        let mut owner = HandshakeChild {
            release: Some(release_write),
            spawning: None,
            child: None,
            readers: Vec::new(),
        };
        let preexec_ready = handshake_marker(ready_read, &mut owner);
        owner.spawning = Some(spawn_handshake_child(ready_write, release_read, true));
        assert_eq!(
            preexec_ready
                .recv_timeout(Duration::from_secs(1))
                .expect("pre-exec readiness deadline")
                .expect("pre-exec readiness byte"),
            b'R'
        );
        drop(listener);
        drop(lock);
        assert!(matches!(
            contender.try_lock(),
            Err(std::fs::TryLockError::WouldBlock)
        ));
        // This unique fixture's owned descriptions were closed in the parent;
        // the acknowledged child retains both socket and lock until exec.
        // The acknowledged child is still before exec and genuinely retains
        // the parent's CLOEXEC socket, despite the parent's local drop.
        let retained = tokio::time::timeout(
            Duration::from_millis(75),
            tokio::net::UnixStream::connect(&socket),
        )
        .await
        .expect("pre-exec connection deadline")
        .expect("pre-exec child retains listener");
        drop(retained);
        release_child_and_observe_exec(&mut owner);
        let deadline = tokio::time::Instant::now() + Duration::from_millis(75);
        loop {
            let connection =
                tokio::time::timeout_at(deadline, tokio::net::UnixStream::connect(&socket))
                    .await
                    .expect("post-exec connection deadline");
            match connection {
                Err(error) if error.kind() == std::io::ErrorKind::ConnectionRefused => break,
                Err(error) => panic!("unexpected post-exec connection error: {error}"),
                Ok(stream) => {
                    drop(stream);
                    assert!(
                        tokio::time::Instant::now() < deadline,
                        "listener retained after acknowledged exec"
                    );
                    tokio::task::yield_now().await;
                }
            }
        }
        observe_lock_release(&contender, &lock_path, deadline).await;
        assert!(
            owner
                .child
                .as_mut()
                .expect("owned child")
                .try_wait()
                .expect("child status")
                .is_none(),
            "listener refusal and lock release must precede child exit"
        );
        assert!(
            owner.cleanup_until(std::time::Instant::now() + Duration::from_secs(1)),
            "handshake_cleanup_timeout: budget_ms=1000, ownership=retained"
        );
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn listener_release_during_gated_cleanup_retains_inherited_startup_fence() {
        // Parallel pre-exec fixtures deliberately retain every open description
        // until their gate opens. Isolate this lifetime test instead of guessing
        // that another fixture has exec'd within the 75ms observation budget.
        const FIXTURE: &str = "BCODE_TEST_GATED_FENCE_ISOLATED";
        if std::env::var_os(FIXTURE).is_none() {
            let mut child = tokio::process::Command::new(std::env::current_exe().unwrap())
                .args(["startup_lock_tests::listener_release_during_gated_cleanup_retains_inherited_startup_fence", "--exact", "--nocapture"])
                .env(FIXTURE, "1")
                .stdin(std::process::Stdio::null())
                .kill_on_drop(true)
                .spawn().unwrap();
            assert!(
                tokio::time::timeout(Duration::from_secs(3), child.wait())
                    .await
                    .unwrap()
                    .unwrap()
                    .success()
            );
            return;
        }
        let directory = tempfile::tempdir().expect("temporary daemon directory");
        let path = directory.path().join("startup.lock");
        let owner = StartupLock::acquire_at(path.clone(), Duration::from_secs(1))
            .await
            .expect("launcher lock");
        let inherited = startup_lock_stdio(owner.file.try_clone().expect("inherited lock"));
        drop(owner);
        let socket = directory.path().join("daemon.sock");
        let listener = std::os::unix::net::UnixListener::bind(&socket).expect("listener");
        let contender = fs::OpenOptions::new()
            .read(true)
            .write(true)
            .open(&path)
            .expect("contender description");
        let (release, gate) = tokio::sync::oneshot::channel();
        let cleanup = async move {
            drop(listener);
            gate.await.expect("cleanup release");
            // Model the daemon's lifetime stdin: release only after cleanup settles.
            drop(inherited);
        };
        tokio::pin!(cleanup);
        std::future::poll_fn(|cx| {
            assert!(std::future::Future::poll(cleanup.as_mut(), cx).is_pending());
            std::task::Poll::Ready(())
        })
        .await;
        // A concurrent spawn can briefly retain the listener until exec. Local
        // drop is not proof that every process has released its description.
        // Keep cleanup gated and require the startup fence on every observation;
        // a persistent inherited listener must fail, not be hidden by cleanup.
        let refusal_deadline = tokio::time::Instant::now() + Duration::from_millis(75);
        loop {
            assert!(matches!(
                contender.try_lock(),
                Err(std::fs::TryLockError::WouldBlock)
            ));
            let connection =
                tokio::time::timeout_at(refusal_deadline, tokio::net::UnixStream::connect(&socket))
                    .await
                    .expect("listener connection did not settle within the refusal budget");
            match connection {
                Err(error) if error.kind() == std::io::ErrorKind::ConnectionRefused => break,
                Err(error) => panic!("unexpected stale-listener connection error: {error}"),
                Ok(stream) => {
                    drop(stream);
                    assert!(
                        tokio::time::Instant::now() < refusal_deadline,
                        "listener still accepts connections while cleanup is gated"
                    );
                    tokio::task::yield_now().await;
                }
            }
        }
        assert!(matches!(
            contender.try_lock(),
            Err(std::fs::TryLockError::WouldBlock)
        ));
        release.send(()).expect("cleanup still owns lifetime fence");
        cleanup.await;
        observe_lock_release(&contender, &path, refusal_deadline).await;
    }

    #[tokio::test]
    async fn concurrent_startup_lock_holders_are_exclusive() {
        let directory = tempfile::tempdir().expect("lock directory");
        let path = directory.path().join("startup.lock");
        let active = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let barrier = std::sync::Arc::new(tokio::sync::Barrier::new(16));
        let mut tasks = tokio::task::JoinSet::new();
        for _ in 0..16 {
            let path = path.clone();
            let active = active.clone();
            let barrier = barrier.clone();
            tasks.spawn(async move {
                barrier.wait().await;
                let lock = StartupLock::acquire_at(path, Duration::from_secs(5))
                    .await
                    .expect("acquire startup lock");
                assert_eq!(active.fetch_add(1, std::sync::atomic::Ordering::SeqCst), 0);
                tokio::task::yield_now().await;
                assert_eq!(active.fetch_sub(1, std::sync::atomic::Ordering::SeqCst), 1);
                drop(lock);
            });
        }
        while let Some(result) = tasks.join_next().await {
            result.expect("exclusive holder task");
        }
    }

    #[tokio::test]
    async fn startup_lock_waits_for_owner_instead_of_stealing_lock_file() {
        let path = lock_path("wait");
        let first = StartupLock::acquire_at(path.clone(), Duration::from_secs(1))
            .await
            .expect("first lock");
        let waiter_path = path.clone();
        let waiter = tokio::spawn(async move {
            StartupLock::acquire_at(waiter_path, Duration::from_secs(1)).await
        });
        tokio::time::sleep(Duration::from_millis(100)).await;
        assert!(!waiter.is_finished());
        drop(first);
        let second = waiter.await.expect("waiter task").expect("second lock");
        drop(second);
        let _ = fs::remove_file(path);
    }

    #[tokio::test]
    async fn startup_lock_crash_helper() {
        let Some(path) = std::env::var_os("BCODE_STARTUP_LOCK_CRASH_PATH") else {
            return;
        };
        let marker =
            std::env::var_os("BCODE_STARTUP_LOCK_CRASH_MARKER").expect("crash helper marker path");
        let _lock = StartupLock::acquire_at(PathBuf::from(path), Duration::from_secs(2))
            .await
            .expect("crash helper lock");
        fs::write(marker, b"locked").expect("write crash helper marker");
        tokio::time::sleep(Duration::from_millis(250)).await;
        std::process::exit(70);
    }

    #[tokio::test]
    async fn startup_lock_owner_crash_permits_bounded_takeover() {
        let path = lock_path("crash");
        let marker = path.with_extension("locked");
        let mut child = tokio::process::Command::new(std::env::current_exe().expect("test binary"))
            .args(["startup_lock_crash_helper", "--nocapture"])
            .env("BCODE_STARTUP_LOCK_CRASH_PATH", &path)
            .env("BCODE_STARTUP_LOCK_CRASH_MARKER", &marker)
            .stdin(std::process::Stdio::null())
            .stdout(std::process::Stdio::null())
            .stderr(std::process::Stdio::null())
            .spawn()
            .expect("spawn crash helper");

        let marker_deadline = tokio::time::Instant::now() + Duration::from_secs(2);
        while !marker.exists() {
            assert!(
                tokio::time::Instant::now() < marker_deadline,
                "crash helper did not acquire startup lock"
            );
            assert!(
                child.try_wait().expect("query crash helper").is_none(),
                "crash helper exited before acquiring startup lock"
            );
            tokio::time::sleep(Duration::from_millis(10)).await;
        }

        let replacement = StartupLock::acquire_at(path.clone(), Duration::from_secs(2))
            .await
            .expect("take over lock after owner crash");
        let status = child.wait().await.expect("wait for crash helper");
        assert!(!status.success());
        drop(replacement);
        let _ = fs::remove_file(path);
        let _ = fs::remove_file(marker);
    }

    #[test]
    fn startup_errors_preserve_actionable_log_context() {
        let log_path = PathBuf::from("/tmp/bcode-daemon-test.log");
        let recent_log = "fatal startup detail\n".to_owned();
        let exited = DaemonStartError::Exited {
            status: "exit status: 70".to_owned(),
            log_path: log_path.display().to_string(),
            recent_log: recent_log.clone(),
        };
        let timeout = DaemonStartError::StartTimeout {
            log_path: log_path.display().to_string(),
            recent_log,
        };

        for error in [exited, timeout] {
            let message = error.to_string();
            assert!(message.contains("/tmp/bcode-daemon-test.log"));
            assert!(message.contains("fatal startup detail"));
            assert!(message.contains("bcode server run"));
        }
    }

    #[test]
    fn recent_log_excerpt_is_bounded_to_latest_context() {
        let path = lock_path("recent-log");
        let contents = (0..40)
            .map(|line| format!("line-{line}"))
            .collect::<Vec<_>>()
            .join("\n");
        fs::write(&path, contents).expect("write test log");

        let excerpt = recent_log_excerpt(&path);

        assert!(!excerpt.contains("line-0\n"));
        assert!(excerpt.starts_with("line-10\n"));
        assert!(excerpt.ends_with("line-39\n"));
        fs::remove_file(path).expect("remove test log");
    }

    #[tokio::test]
    async fn startup_lock_times_out_without_removing_active_owner() {
        let path = lock_path("timeout");
        let first = StartupLock::acquire_at(path.clone(), Duration::from_secs(1))
            .await
            .expect("first lock");
        let error = StartupLock::acquire_at(path.clone(), Duration::from_millis(50))
            .await
            .expect_err("active owner must not be replaced");
        assert!(matches!(
            error,
            DaemonStartError::StartupCoordinationTimeout { lock_path } if lock_path == path
        ));
        assert!(path.exists());
        drop(first);
        let _ = fs::remove_file(path);
    }
}

fn recent_log_excerpt(log_path: &Path) -> String {
    let Ok(contents) = fs::read_to_string(log_path) else {
        return "daemon log could not be read".to_string();
    };
    let lines = contents.lines().rev().take(30).collect::<Vec<_>>();
    if lines.is_empty() {
        return "daemon log is empty".to_string();
    }
    let mut excerpt = lines.into_iter().rev().collect::<Vec<_>>().join("\n");
    if !excerpt.ends_with('\n') {
        excerpt.push('\n');
    }
    excerpt
}