bcode_session 0.0.1-alpha.0

//! Cross-process session ownership and catalog lock primitives.
//!
//! One daemon instance owns a session at a time and serves any number of clients through that
//! owner. Reentrant registrations from the same daemon instance remain compatible; another daemon
//! is refused even when it supports the same writer epoch. Maintenance takes the coordinator
//! exclusively and cannot run underneath a live owner.

use bcode_session_models::SessionId;
use serde::{Deserialize, Serialize};
use std::fs::{self, File, OpenOptions};
use std::io::{self, Write as _};
use std::path::{Path, PathBuf};
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
use thiserror::Error;

/// Durable session-storage writer contract supported by this Bcode build.
pub const CURRENT_SESSION_STORAGE_WRITER_EPOCH: u32 =
    bcode_session_models::CURRENT_SESSION_STORAGE_WRITER_EPOCH;

/// Serializable metadata describing one process currently accessing a session.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct SessionLeaseOwner {
    /// Metadata schema version.
    pub schema_version: u32,
    /// Session protected by this access record.
    pub session_id: SessionId,
    /// Unique token for this access registration.
    pub lease_token: String,
    /// Accessing process id.
    pub pid: u32,
    /// Durable session-storage writer epoch supported by this process.
    #[serde(default)]
    pub storage_writer_epoch: Option<u32>,
    /// Bcode daemon namespace, when known.
    pub daemon_namespace: Option<String>,
    /// Bcode build fingerprint, when known.
    pub build_fingerprint: Option<String>,
    /// Bcode IPC protocol version, when known.
    pub protocol_version: Option<u32>,
    /// Accessing daemon endpoint, when known.
    pub endpoint: Option<String>,
    /// Executable that registered access, when known.
    pub executable_path: Option<PathBuf>,
    /// Daemon instance id, when known.
    pub daemon_instance_id: Option<String>,
    /// Access registration time in Unix milliseconds.
    pub acquired_at_ms: u64,
    /// Latest owner heartbeat/update time in Unix milliseconds.
    pub last_seen_ms: u64,
}

impl SessionLeaseOwner {
    fn new(session_id: SessionId, context: &SessionLeaseOwnerContext) -> Self {
        let now = unix_time_millis();
        Self {
            schema_version: 3,
            session_id,
            lease_token: format!("{}-{now}-{session_id}", std::process::id()),
            pid: std::process::id(),
            storage_writer_epoch: context.storage_writer_epoch,
            daemon_namespace: context.daemon_namespace.clone(),
            build_fingerprint: context.build_fingerprint.clone(),
            protocol_version: context.protocol_version,
            endpoint: context.endpoint.clone(),
            executable_path: context
                .executable_path
                .clone()
                .or_else(|| std::env::current_exe().ok()),
            daemon_instance_id: context.daemon_instance_id.clone(),
            acquired_at_ms: now,
            last_seen_ms: now,
        }
    }
}

/// Optional daemon identity to write into session access metadata.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SessionLeaseOwnerContext {
    /// Durable session-storage writer epoch.
    pub storage_writer_epoch: Option<u32>,
    /// Bcode daemon namespace.
    pub daemon_namespace: Option<String>,
    /// Bcode build fingerprint.
    pub build_fingerprint: Option<String>,
    /// Bcode IPC protocol version.
    pub protocol_version: Option<u32>,
    /// Bcode IPC endpoint description.
    pub endpoint: Option<String>,
    /// Daemon executable path.
    pub executable_path: Option<PathBuf>,
    /// Daemon instance id.
    pub daemon_instance_id: Option<String>,
}

impl Default for SessionLeaseOwnerContext {
    fn default() -> Self {
        Self {
            storage_writer_epoch: Some(CURRENT_SESSION_STORAGE_WRITER_EPOCH),
            daemon_namespace: None,
            build_fingerprint: None,
            protocol_version: None,
            endpoint: None,
            executable_path: None,
            daemon_instance_id: Some(format!("process-{}", std::process::id())),
        }
    }
}

/// Errors returned while registering or releasing cross-process session access.
#[derive(Debug, Error)]
pub enum SessionLeaseError {
    /// Filesystem operation failed.
    #[error("session access I/O error at {}: {source}", path.display())]
    Io {
        /// Path involved in the failed operation.
        path: PathBuf,
        /// Original I/O error.
        source: io::Error,
    },
    /// Owner metadata could not be serialized.
    #[error("session access metadata serialization failed: {0}")]
    Serialize(#[from] serde_json::Error),
    /// A mutation-capable session lease omitted its required storage writer identity.
    #[error("session {session_id} access requires an explicit storage writer epoch")]
    MissingStorageWriterEpoch {
        /// Session whose lease identity was incomplete.
        session_id: SessionId,
    },
    /// Another daemon with an incompatible storage writer currently has the session open.
    #[error("session {session_id} is open in another incompatible Bcode writer{owner_summary}")]
    OwnedByOtherDaemon {
        /// Session whose access could not be registered.
        session_id: SessionId,
        /// Best-effort incompatible owner metadata.
        owner: Option<Box<SessionLeaseOwner>>,
        /// Human-readable owner summary.
        owner_summary: String,
    },
    /// Owner metadata exists but cannot be classified safely.
    #[error("session owner metadata at {path} is malformed or uses an unsupported schema")]
    UnverifiableOwnerMetadata {
        /// Owner metadata path that could not be classified.
        path: PathBuf,
    },
    /// A session lock could not be acquired within its bounded wait budget.
    #[error(
        "timed out after {waited_ms}ms acquiring the session lock at {}",
        path.display()
    )]
    LockWaitTimeout {
        /// Lock path that stayed contended for the whole wait budget.
        path: PathBuf,
        /// Milliseconds spent waiting before giving up.
        waited_ms: u64,
    },
    /// Current platform does not support Bcode's session access mechanism.
    #[error("session access guards are unsupported on this platform")]
    Unsupported,
}

/// Held compatible access registration for one session.
#[derive(Debug)]
pub struct SessionLeaseGuard {
    owner_path: PathBuf,
    owner_file: Option<File>,
    owner: SessionLeaseOwner,
}

impl SessionLeaseGuard {
    /// Return owner metadata for this held access registration.
    #[must_use]
    pub const fn owner(&self) -> &SessionLeaseOwner {
        &self.owner
    }
}

impl Drop for SessionLeaseGuard {
    fn drop(&mut self) {
        remove_owner_metadata_if_token_matches(&self.owner_path, &self.owner.lease_token);
        if let Some(file) = self.owner_file.take() {
            let _ = unlock_file(&file);
        }
    }
}

/// Held exclusive maintenance access to one session.
#[derive(Debug)]
pub struct SessionMaintenanceGuard {
    coordinator: File,
}

impl Drop for SessionMaintenanceGuard {
    fn drop(&mut self) {
        let _ = unlock_file(&self.coordinator);
    }
}

/// Held short-lived exclusive access to one session write critical section.
#[derive(Debug)]
pub struct SessionWriteGuard {
    file: File,
    coordinator: Option<File>,
}

impl Drop for SessionWriteGuard {
    fn drop(&mut self) {
        let _ = unlock_file(&self.file);
        if let Some(coordinator) = &self.coordinator {
            let _ = unlock_file(coordinator);
        }
    }
}

/// Held exclusive access to the global catalog database.
#[derive(Debug)]
pub struct CatalogLockGuard {
    file: File,
}

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

/// Read-only liveness observation for one decodable owner record.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct SessionOwnerObservation {
    /// Owner metadata.
    pub owner: SessionLeaseOwner,
    /// Lock/PID-backed liveness classification.
    pub liveness: SessionOwnerLiveness,
}

/// Inspect every decodable owner record without mutating lease metadata.
///
/// # Errors
///
/// Returns an error when the owner directory cannot be inspected.
pub fn session_owner_observations(
    root: &Path,
    session_id: SessionId,
) -> Result<Vec<SessionOwnerObservation>, SessionLeaseError> {
    let access_dir = session_owner_dir(root, session_id);
    let Ok(entries) = fs::read_dir(&access_dir) else {
        return Ok(Vec::new());
    };
    let mut observations = Vec::new();
    for entry in entries {
        let entry = entry.map_err(|source| SessionLeaseError::Io {
            path: access_dir.clone(),
            source,
        })?;
        let path = entry.path();
        if path.extension().and_then(|extension| extension.to_str()) != Some("json") {
            continue;
        }
        if let Some(owner) = read_owner_metadata(&path) {
            observations.push(SessionOwnerObservation {
                liveness: classify_owner_liveness(&path, &owner),
                owner,
            });
        }
    }
    observations.sort_by(|left, right| left.owner.lease_token.cmp(&right.owner.lease_token));
    Ok(observations)
}

/// Return live owner metadata for one session without mutating owner records.
///
/// # Errors
///
/// Returns an error when the owner directory cannot be inspected.
pub fn active_session_owners(
    root: &Path,
    session_id: SessionId,
) -> Result<Vec<SessionLeaseOwner>, SessionLeaseError> {
    let access_dir = session_owner_dir(root, session_id);
    let Ok(entries) = fs::read_dir(&access_dir) else {
        return Ok(Vec::new());
    };
    let mut owners = Vec::new();
    for entry in entries {
        let entry = entry.map_err(|source| SessionLeaseError::Io {
            path: access_dir.clone(),
            source,
        })?;
        let path = entry.path();
        if path.extension().and_then(|extension| extension.to_str()) != Some("json") {
            continue;
        }
        if let Some(owner) = read_owner_metadata(&path)
            && matches!(
                classify_owner_liveness(&path, &owner),
                SessionOwnerLiveness::Live | SessionOwnerLiveness::Unverifiable
            )
        {
            owners.push(owner);
        }
    }
    owners.sort_by(|left, right| left.lease_token.cmp(&right.lease_token));
    Ok(owners)
}

/// Register ownership for a session.
///
/// This briefly takes the coordinator lock, removes dead process registrations, rejects a live
/// owner from another daemon instance or writer epoch, and writes this process registration.
/// Multiple registrations are allowed only for the same explicit daemon instance so many clients
/// can continue routing through one owning daemon.
///
/// # Errors
///
/// Returns an error if the coordinator lock cannot be opened/locked, the lock stays contended for
/// the bounded wait budget, owner metadata cannot be
/// read/written, or an incompatible live build is already registered for the session.
pub fn acquire_session_lease(
    root: &Path,
    session_id: SessionId,
    context: &SessionLeaseOwnerContext,
) -> Result<SessionLeaseGuard, SessionLeaseError> {
    if context.storage_writer_epoch.is_none() {
        return Err(SessionLeaseError::MissingStorageWriterEpoch { session_id });
    }
    let lock_path = session_lock_path(root, session_id);
    let file = open_lock_file(&lock_path)?;
    lock_file_exclusive_bounded(&file, &lock_path)?;

    let access_dir = session_owner_dir(root, session_id);
    fs::create_dir_all(&access_dir).map_err(|source| SessionLeaseError::Io {
        path: access_dir.clone(),
        source,
    })?;
    prune_dead_owner_records(&access_dir)?;

    if let Some(owner) = find_incompatible_owner(&access_dir, context)? {
        let owner_summary = format!(": {}", format_owner(&owner));
        let _ = unlock_file(&file);
        return Err(SessionLeaseError::OwnedByOtherDaemon {
            session_id,
            owner: Some(Box::new(owner)),
            owner_summary,
        });
    }

    let owner = SessionLeaseOwner::new(session_id, context);
    let owner_path = access_dir.join(format!("{}.json", owner.lease_token));
    let owner_file = write_owner_metadata(&owner_path, &owner)?;
    let _ = unlock_file(&file);
    Ok(SessionLeaseGuard {
        owner_path,
        owner_file: Some(owner_file),
        owner,
    })
}

/// Report whether exclusive session maintenance currently holds the coordinator lock.
///
/// This is a non-mutating admission check. A `false` result is not a lease: callers that need
/// exclusive maintenance must still acquire and retain [`SessionMaintenanceGuard`].
///
/// # Errors
///
/// Returns an error when the coordinator lock file cannot be opened or inspected.
pub fn session_maintenance_is_active(
    root: &Path,
    session_id: SessionId,
) -> Result<bool, SessionLeaseError> {
    let lock_path = session_lock_path(root, session_id);
    let coordinator = open_lock_file(&lock_path)?;
    if try_lock_file_exclusive(&coordinator).map_err(|source| SessionLeaseError::Io {
        path: lock_path,
        source,
    })? {
        let _ = unlock_file(&coordinator);
        Ok(false)
    } else {
        Ok(true)
    }
}

/// Acquire exclusive maintenance access, refusing every live session owner.
///
/// The coordinator lock remains held for the guard lifetime, preventing new owners from
/// registering while an offline migration is active.
///
/// # Errors
///
/// Returns an error if locking or owner inspection fails, the coordinator lock stays contended for
/// the bounded wait budget, or if any live owner exists.
pub fn acquire_session_maintenance_guard(
    root: &Path,
    session_id: SessionId,
) -> Result<SessionMaintenanceGuard, SessionLeaseError> {
    let lock_path = session_lock_path(root, session_id);
    let coordinator = open_lock_file(&lock_path)?;
    lock_file_exclusive_bounded(&coordinator, &lock_path)?;
    let access_dir = session_owner_dir(root, session_id);
    fs::create_dir_all(&access_dir).map_err(|source| SessionLeaseError::Io {
        path: access_dir.clone(),
        source,
    })?;
    prune_dead_owner_records(&access_dir)?;
    if let Some(owner) = first_owner(&access_dir)? {
        return Err(SessionLeaseError::OwnedByOtherDaemon {
            session_id,
            owner_summary: format!(": {}", format_owner(&owner)),
            owner: Some(Box::new(owner)),
        });
    }
    Ok(SessionMaintenanceGuard { coordinator })
}

#[cfg(test)]
fn abort_at_lease_handoff_boundary() {
    if std::env::var_os("BCODE_MIGRATION_LEASE_HANDOFF_CRASH").is_some() {
        std::process::abort();
    }
}

/// Atomically replace exclusive maintenance ownership with a compatible session lease.
///
/// The maintenance coordinator remains locked while owner metadata is written, so another writer
/// cannot claim the session between migration completion and runtime ownership registration.
///
/// # Errors
///
/// Returns an error if writer identity is missing or owner metadata cannot be written.
pub fn transition_session_maintenance_to_lease(
    maintenance: SessionMaintenanceGuard,
    root: &Path,
    session_id: SessionId,
    context: &SessionLeaseOwnerContext,
) -> Result<SessionLeaseGuard, SessionLeaseError> {
    if context.storage_writer_epoch.is_none() {
        return Err(SessionLeaseError::MissingStorageWriterEpoch { session_id });
    }
    let access_dir = session_owner_dir(root, session_id);
    fs::create_dir_all(&access_dir).map_err(|source| SessionLeaseError::Io {
        path: access_dir.clone(),
        source,
    })?;
    prune_dead_owner_records(&access_dir)?;
    if let Some(owner) = first_owner(&access_dir)? {
        return Err(SessionLeaseError::OwnedByOtherDaemon {
            session_id,
            owner_summary: format!(": {}", format_owner(&owner)),
            owner: Some(Box::new(owner)),
        });
    }
    let owner = SessionLeaseOwner::new(session_id, context);
    let owner_path = access_dir.join(format!("{}.json", owner.lease_token));
    let owner_file = write_owner_metadata(&owner_path, &owner)?;
    #[cfg(test)]
    abort_at_lease_handoff_boundary();
    drop(maintenance);
    Ok(SessionLeaseGuard {
        owner_path,
        owner_file: Some(owner_file),
        owner,
    })
}

fn first_owner(access_dir: &Path) -> Result<Option<SessionLeaseOwner>, SessionLeaseError> {
    for entry in fs::read_dir(access_dir).map_err(|source| SessionLeaseError::Io {
        path: access_dir.to_path_buf(),
        source,
    })? {
        let entry = entry.map_err(|source| SessionLeaseError::Io {
            path: access_dir.to_path_buf(),
            source,
        })?;
        if entry.path().extension().and_then(|value| value.to_str()) == Some("json")
            && let Some(owner) = read_owner_metadata(&entry.path())
        {
            return Ok(Some(owner));
        }
    }
    Ok(None)
}

/// Acquire exclusive short-lived access to one session write critical section.
///
/// This does not represent session ownership. It only serializes app-assigned event sequence
/// allocation across same-version daemons until sequence allocation is fully database-owned.
///
/// # Errors
///
/// Returns an error if the write lock file cannot be opened or locked, or if either the
/// coordinator or write lock stays contended for the bounded wait budget.
pub fn acquire_session_write_lock(
    root: &Path,
    session_id: SessionId,
) -> Result<SessionWriteGuard, SessionLeaseError> {
    let coordinator_path = session_lock_path(root, session_id);
    let coordinator = open_lock_file(&coordinator_path)?;
    lock_file_shared_bounded(&coordinator, &coordinator_path)?;
    acquire_session_write_lock_inner(root, session_id, Some(coordinator))
}

/// Acquire the session write lock while exclusive maintenance coordination is already held.
///
/// # Errors
///
/// Returns an error if the write lock file cannot be opened or locked, or if it stays contended for
/// the bounded wait budget.
pub fn acquire_maintenance_session_write_lock(
    _maintenance: &SessionMaintenanceGuard,
    root: &Path,
    session_id: SessionId,
) -> Result<SessionWriteGuard, SessionLeaseError> {
    acquire_session_write_lock_inner(root, session_id, None)
}

fn acquire_session_write_lock_inner(
    root: &Path,
    session_id: SessionId,
    coordinator: Option<File>,
) -> Result<SessionWriteGuard, SessionLeaseError> {
    let lock_path = session_write_lock_path(root, session_id);
    let file = open_lock_file(&lock_path)?;
    lock_file_exclusive_bounded(&file, &lock_path)?;
    Ok(SessionWriteGuard { file, coordinator })
}

/// Acquire exclusive short-lived access to the global catalog database.
///
/// # Errors
///
/// Returns an error if the catalog lock file cannot be opened or locked, or if it stays contended
/// for the bounded wait budget.
pub fn acquire_catalog_lock(root: &Path) -> Result<CatalogLockGuard, SessionLeaseError> {
    let lock_path = root.join("catalog.lock");
    let file = open_lock_file(&lock_path)?;
    lock_file_exclusive_bounded(&file, &lock_path)?;
    Ok(CatalogLockGuard { file })
}

fn session_lock_path(root: &Path, session_id: SessionId) -> PathBuf {
    root.join("locks").join(format!("{session_id}.lock"))
}

fn session_write_lock_path(root: &Path, session_id: SessionId) -> PathBuf {
    root.join("locks").join(format!("{session_id}.write.lock"))
}

fn session_owner_dir(root: &Path, session_id: SessionId) -> PathBuf {
    root.join("leases").join(session_id.to_string())
}

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

#[cfg(test)]
fn abort_before_owner_metadata_publish() {
    if std::env::var_os("BCODE_LEASE_PRE_PUBLISH_CRASH").is_some() {
        std::process::abort();
    }
}

fn write_owner_metadata(path: &Path, owner: &SessionLeaseOwner) -> Result<File, SessionLeaseError> {
    if let Some(parent) = path.parent() {
        fs::create_dir_all(parent).map_err(|source| SessionLeaseError::Io {
            path: parent.to_path_buf(),
            source,
        })?;
    }
    let temp_path = path.with_extension(format!("json.tmp-{}", owner.lease_token));
    let contents = serde_json::to_vec_pretty(owner)?;
    let mut file = OpenOptions::new()
        .create_new(true)
        .read(true)
        .write(true)
        .open(&temp_path)
        .map_err(|source| SessionLeaseError::Io {
            path: temp_path.clone(),
            source,
        })?;
    file.write_all(&contents)
        .and_then(|()| file.sync_all())
        .map_err(|source| SessionLeaseError::Io {
            path: temp_path.clone(),
            source,
        })?;
    // Publish complete immutable metadata under a shared liveness lock. Windows
    // byte-range locks otherwise prevent observers from reading the owner record.
    file.lock_shared().map_err(|source| SessionLeaseError::Io {
        path: temp_path.clone(),
        source,
    })?;
    #[cfg(test)]
    abort_before_owner_metadata_publish();
    fs::rename(&temp_path, path).map_err(|source| SessionLeaseError::Io {
        path: path.to_path_buf(),
        source,
    })?;
    Ok(file)
}

/// Conservative liveness classification for one owner record.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum SessionOwnerLiveness {
    /// Positive evidence shows the owner is live.
    Live,
    /// Positive evidence shows the owner is stale.
    Stale,
    /// The owner cannot be classified safely.
    Unverifiable,
}

fn classify_owner_liveness(path: &Path, owner: &SessionLeaseOwner) -> SessionOwnerLiveness {
    if owner.schema_version == 3 {
        match OpenOptions::new().read(true).write(true).open(path) {
            Ok(file) => match try_lock_file_exclusive(&file) {
                Ok(true) => {
                    let _ = unlock_file(&file);
                    SessionOwnerLiveness::Stale
                }
                Ok(false) => SessionOwnerLiveness::Live,
                Err(_) => SessionOwnerLiveness::Unverifiable,
            },
            Err(error) if error.kind() == io::ErrorKind::NotFound => SessionOwnerLiveness::Stale,
            Err(_) => SessionOwnerLiveness::Unverifiable,
        }
    } else if owner.schema_version == 2 && process_is_alive(owner.pid) {
        SessionOwnerLiveness::Live
    } else if owner.schema_version == 2 {
        SessionOwnerLiveness::Stale
    } else {
        SessionOwnerLiveness::Unverifiable
    }
}

fn read_owner_metadata(path: &Path) -> Option<SessionLeaseOwner> {
    let contents = fs::read(path).ok()?;
    serde_json::from_slice(&contents).ok()
}

fn remove_owner_metadata_if_token_matches(path: &Path, token: &str) {
    let Some(owner) = read_owner_metadata(path) else {
        return;
    };
    if owner.lease_token == token {
        let _ = fs::remove_file(path);
    }
}

fn prune_dead_owner_records(access_dir: &Path) -> Result<(), SessionLeaseError> {
    for entry in fs::read_dir(access_dir).map_err(|source| SessionLeaseError::Io {
        path: access_dir.to_path_buf(),
        source,
    })? {
        let entry = entry.map_err(|source| SessionLeaseError::Io {
            path: access_dir.to_path_buf(),
            source,
        })?;
        let path = entry.path();
        if path.extension().and_then(|extension| extension.to_str()) != Some("json") {
            continue;
        }
        let Some(owner) = read_owner_metadata(&path) else {
            return Err(SessionLeaseError::UnverifiableOwnerMetadata { path });
        };
        if classify_owner_liveness(&path, &owner) == SessionOwnerLiveness::Stale {
            remove_file_best_effort(&path)?;
        }
    }
    Ok(())
}

fn find_incompatible_owner(
    access_dir: &Path,
    context: &SessionLeaseOwnerContext,
) -> Result<Option<SessionLeaseOwner>, SessionLeaseError> {
    for entry in fs::read_dir(access_dir).map_err(|source| SessionLeaseError::Io {
        path: access_dir.to_path_buf(),
        source,
    })? {
        let entry = entry.map_err(|source| SessionLeaseError::Io {
            path: access_dir.to_path_buf(),
            source,
        })?;
        let path = entry.path();
        if path.extension().and_then(|extension| extension.to_str()) != Some("json") {
            continue;
        }
        let Some(owner) = read_owner_metadata(&path) else {
            continue;
        };
        if !owner_is_compatible(&owner, context) {
            return Ok(Some(owner));
        }
    }
    Ok(None)
}

fn owner_is_compatible(owner: &SessionLeaseOwner, context: &SessionLeaseOwnerContext) -> bool {
    matches!(
        (owner.storage_writer_epoch, context.storage_writer_epoch),
        (Some(owner_epoch), Some(context_epoch)) if owner_epoch == context_epoch
    ) && owner.daemon_instance_id.is_some()
        && owner.daemon_instance_id == context.daemon_instance_id
}

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

fn format_owner(owner: &SessionLeaseOwner) -> String {
    let mut parts = vec![format!("pid {}", owner.pid)];
    if let Some(instance) = &owner.daemon_instance_id {
        parts.push(format!("daemon instance {instance}"));
    }
    if let Some(epoch) = owner.storage_writer_epoch {
        parts.push(format!("storage writer epoch {epoch}"));
    }
    if let Some(namespace) = &owner.daemon_namespace {
        parts.push(format!("namespace {namespace}"));
    }
    if let Some(fingerprint) = &owner.build_fingerprint {
        parts.push(format!("build {fingerprint}"));
    }
    if let Some(endpoint) = &owner.endpoint {
        parts.push(format!("endpoint {endpoint}"));
    }
    parts.join(", ")
}

fn unix_time_millis() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map_or(0, |duration| {
            u64::try_from(duration.as_millis()).unwrap_or(u64::MAX)
        })
}

#[cfg(unix)]
fn process_is_alive(pid: u32) -> bool {
    let Ok(pid) = i32::try_from(pid) else {
        return false;
    };
    // SAFETY: `kill(pid, 0)` does not send a signal; it only asks the kernel whether the process
    // exists or is inaccessible. The pid is converted to the platform `i32` representation above.
    let result = unsafe { kill(pid, 0) };
    if result == 0 {
        return true;
    }
    std::io::Error::last_os_error().raw_os_error() == Some(1)
}

#[cfg(not(unix))]
const fn process_is_alive(_pid: u32) -> bool {
    true
}

fn try_lock_file_shared(file: &File) -> io::Result<bool> {
    match file.try_lock_shared() {
        Ok(()) => Ok(true),
        Err(fs::TryLockError::WouldBlock) => Ok(false),
        Err(fs::TryLockError::Error(error)) => Err(error),
    }
}

fn try_lock_file_exclusive(file: &File) -> io::Result<bool> {
    match file.try_lock() {
        Ok(()) => Ok(true),
        Err(fs::TryLockError::WouldBlock) => Ok(false),
        Err(fs::TryLockError::Error(error)) => Err(error),
    }
}

#[cfg(test)]
fn lock_file_exclusive(file: &File) -> io::Result<()> {
    file.lock()
}

/// Bounded wait budget for acquiring a contended session lock.
///
/// An unbounded blocking `flock` on an async executor thread turns lock contention into a silent
/// hang with no error and no cancellation point, so every lock wait is bounded and reported.
const LOCK_WAIT_BUDGET: Duration = Duration::from_secs(10);

const LOCK_WAIT_POLL_INTERVAL: Duration = Duration::from_millis(25);

/// Acquire an exclusive file lock, failing with a bounded timeout instead of blocking forever.
fn lock_file_exclusive_bounded(file: &File, path: &Path) -> Result<(), SessionLeaseError> {
    lock_file_bounded(file, path, try_lock_file_exclusive)
}

/// Acquire a shared file lock, failing with a bounded timeout instead of blocking forever.
fn lock_file_shared_bounded(file: &File, path: &Path) -> Result<(), SessionLeaseError> {
    lock_file_bounded(file, path, try_lock_file_shared)
}

fn lock_file_bounded(
    file: &File,
    path: &Path,
    try_lock: fn(&File) -> io::Result<bool>,
) -> Result<(), SessionLeaseError> {
    let started_at = Instant::now();
    loop {
        if try_lock(file).map_err(|source| SessionLeaseError::Io {
            path: path.to_path_buf(),
            source,
        })? {
            return Ok(());
        }
        let waited = started_at.elapsed();
        if waited >= LOCK_WAIT_BUDGET {
            return Err(SessionLeaseError::LockWaitTimeout {
                path: path.to_path_buf(),
                waited_ms: u64::try_from(waited.as_millis()).unwrap_or(u64::MAX),
            });
        }
        std::thread::sleep(LOCK_WAIT_POLL_INTERVAL.min(LOCK_WAIT_BUDGET.saturating_sub(waited)));
    }
}

fn unlock_file(file: &File) -> io::Result<()> {
    file.unlock()
}

#[cfg(unix)]
unsafe extern "C" {
    fn kill(pid: i32, sig: i32) -> i32;
}

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

    #[test]
    fn contended_lock_wait_times_out_instead_of_blocking_forever() {
        let temp_dir = tempfile::tempdir().expect("temp dir");
        let lock_path = temp_dir.path().join("contended.lock");
        let holder = open_lock_file(&lock_path).expect("open holder");
        lock_file_exclusive(&holder).expect("hold lock");

        let waiter = open_lock_file(&lock_path).expect("open waiter");
        let started_at = Instant::now();
        let error = lock_file_exclusive_bounded(&waiter, &lock_path)
            .expect_err("contended lock must not block forever");

        assert!(
            matches!(error, SessionLeaseError::LockWaitTimeout { .. }),
            "expected a bounded wait timeout, got {error:?}"
        );
        assert!(started_at.elapsed() >= LOCK_WAIT_BUDGET);
        let _ = unlock_file(&holder);
    }

    #[test]
    fn uncontended_bounded_lock_wait_succeeds() {
        let temp_dir = tempfile::tempdir().expect("temp dir");
        let lock_path = temp_dir.path().join("free.lock");
        let file = open_lock_file(&lock_path).expect("open lock");

        lock_file_exclusive_bounded(&file, &lock_path).expect("uncontended lock must succeed");
        let _ = unlock_file(&file);
    }

    fn context(build: &str, storage_writer_epoch: u32) -> SessionLeaseOwnerContext {
        SessionLeaseOwnerContext {
            storage_writer_epoch: Some(storage_writer_epoch),
            build_fingerprint: Some(build.to_string()),
            protocol_version: Some(2),
            daemon_instance_id: Some(build.to_owned()),
            ..SessionLeaseOwnerContext::default()
        }
    }

    fn same_daemon_context(build: &str, instance: &str, epoch: u32) -> SessionLeaseOwnerContext {
        SessionLeaseOwnerContext {
            daemon_instance_id: Some(instance.to_owned()),
            ..context(build, epoch)
        }
    }

    #[test]
    fn two_current_daemons_racing_for_one_session_yield_one_owner() {
        let temp_dir = tempfile::tempdir().expect("temp dir");
        let session_id = SessionId::new();
        let barrier = std::sync::Arc::new(std::sync::Barrier::new(3));
        let root = temp_dir.path().to_path_buf();
        let mut joins = Vec::new();
        for daemon in ["daemon-one", "daemon-two"] {
            let root = root.clone();
            let barrier = std::sync::Arc::clone(&barrier);
            joins.push(std::thread::spawn(move || {
                barrier.wait();
                acquire_session_lease(&root, session_id, &context(daemon, 7))
            }));
        }
        barrier.wait();
        let results = joins
            .into_iter()
            .map(|join| join.join().expect("daemon thread"))
            .collect::<Vec<_>>();
        assert_eq!(results.iter().filter(|result| result.is_ok()).count(), 1);
        assert_eq!(
            results
                .iter()
                .filter(|result| matches!(
                    result,
                    Err(SessionLeaseError::OwnedByOtherDaemon { .. })
                ))
                .count(),
            1
        );
    }

    #[test]
    fn different_writer_versions_can_own_different_sessions() {
        let temp_dir = tempfile::tempdir().expect("temp dir");
        let older_session = SessionId::new();
        let current_session = SessionId::new();
        let older = acquire_session_lease(temp_dir.path(), older_session, &context("older", 4))
            .expect("older writer session");
        let current =
            acquire_session_lease(temp_dir.path(), current_session, &context("current", 5))
                .expect("current writer session");
        assert_eq!(older.owner().storage_writer_epoch, Some(4));
        assert_eq!(current.owner().storage_writer_epoch, Some(5));
    }

    #[test]
    fn schema_v3_owner_uses_lock_liveness_and_normal_drop_removes_record() {
        let temp_dir = tempfile::tempdir().expect("temp dir");
        let session_id = SessionId::new();
        let guard = acquire_session_lease(temp_dir.path(), session_id, &context("v3", 5))
            .expect("v3 owner");
        assert_eq!(guard.owner().schema_version, 3);
        assert_eq!(
            classify_owner_liveness(&guard.owner_path, guard.owner()),
            SessionOwnerLiveness::Live
        );
        assert_eq!(
            active_session_owners(temp_dir.path(), session_id)
                .expect("owners")
                .len(),
            1
        );
        let owner_path = guard.owner_path.clone();
        drop(guard);
        assert!(!owner_path.exists());
        assert!(
            active_session_owners(temp_dir.path(), session_id)
                .expect("owners")
                .is_empty()
        );
    }

    #[test]
    fn stale_schema_v3_record_is_pruned_without_pid_evidence() {
        let temp_dir = tempfile::tempdir().expect("temp dir");
        let session_id = SessionId::new();
        let access_dir = session_owner_dir(temp_dir.path(), session_id);
        let owner = SessionLeaseOwner::new(session_id, &context("stale-v3", 5));
        let owner_path = access_dir.join(format!("{}.json", owner.lease_token));
        let file = write_owner_metadata(&owner_path, &owner).expect("owner metadata");
        unlock_file(&file).expect("unlock abandoned owner");
        drop(file);
        assert_eq!(
            classify_owner_liveness(&owner_path, &owner),
            SessionOwnerLiveness::Stale
        );
        let next = acquire_session_lease(temp_dir.path(), session_id, &context("next", 5))
            .expect("stale v3 owner should be pruned");
        assert_eq!(next.owner().daemon_instance_id.as_deref(), Some("next"));
    }

    #[test]
    fn schema_v2_owner_preserves_conservative_pid_fallback() {
        let temp_dir = tempfile::tempdir().expect("temp dir");
        let session_id = SessionId::new();
        let access_dir = session_owner_dir(temp_dir.path(), session_id);
        fs::create_dir_all(&access_dir).expect("owner dir");
        let mut owner = SessionLeaseOwner::new(session_id, &context("v2", 5));
        owner.schema_version = 2;
        owner.pid = std::process::id();
        let owner_path = access_dir.join(format!("{}.json", owner.lease_token));
        fs::write(
            &owner_path,
            serde_json::to_vec_pretty(&owner).expect("serialize v2"),
        )
        .expect("write v2");
        assert_eq!(
            classify_owner_liveness(&owner_path, &owner),
            SessionOwnerLiveness::Live
        );
        assert_eq!(
            active_session_owners(temp_dir.path(), session_id)
                .expect("owners")
                .len(),
            1
        );
    }

    #[test]
    #[ignore = "subprocess helper for pre_publish_crash_does_not_publish_owner"]
    fn pre_publish_crash_helper() {
        let Some(root) = std::env::var_os("BCODE_LEASE_PRE_PUBLISH_ROOT") else {
            return;
        };
        let session_id = std::env::var("BCODE_LEASE_PRE_PUBLISH_SESSION")
            .expect("pre-publish session id")
            .parse::<SessionId>()
            .expect("valid pre-publish session id");
        let _ = acquire_session_lease(Path::new(&root), session_id, &context("crashing", 5));
    }

    #[test]
    fn pre_publish_crash_does_not_publish_owner() {
        use std::process::{Command, Stdio};

        let temp_dir = tempfile::tempdir().expect("temp dir");
        let session_id = SessionId::new();
        let status = Command::new(std::env::current_exe().expect("test executable"))
            .args([
                "--exact",
                "lease::tests::pre_publish_crash_helper",
                "--ignored",
                "--nocapture",
            ])
            .env("BCODE_LEASE_PRE_PUBLISH_CRASH", "1")
            .env("BCODE_LEASE_PRE_PUBLISH_ROOT", temp_dir.path())
            .env("BCODE_LEASE_PRE_PUBLISH_SESSION", session_id.to_string())
            .stdin(Stdio::null())
            .stdout(Stdio::null())
            .stderr(Stdio::null())
            .status()
            .expect("run crashing lease publisher");
        assert!(!status.success(), "publisher must terminate at crash hook");
        assert!(
            active_session_owners(temp_dir.path(), session_id)
                .expect("owner scan")
                .is_empty(),
            "pre-publish crash must not expose an owner record"
        );
        acquire_session_lease(temp_dir.path(), session_id, &context("next", 5))
            .expect("next owner must acquire after pre-publish crash");
    }

    #[test]
    fn abandoned_pre_publish_owner_temp_file_does_not_block_acquisition() {
        let temp_dir = tempfile::tempdir().expect("temp dir");
        let session_id = SessionId::new();
        let access_dir = session_owner_dir(temp_dir.path(), session_id);
        fs::create_dir_all(&access_dir).expect("owner dir");
        let abandoned = access_dir.join("abandoned.json.tmp-pre-publish");
        fs::write(&abandoned, b"partial metadata").expect("abandoned temp file");

        assert!(
            active_session_owners(temp_dir.path(), session_id)
                .expect("owner scan")
                .is_empty()
        );
        let owner = acquire_session_lease(temp_dir.path(), session_id, &context("current", 5))
            .expect("pre-publish temp file must not block acquisition");
        assert_eq!(owner.owner().daemon_instance_id.as_deref(), Some("current"));
        assert!(
            abandoned.exists(),
            "normal acquisition must not mutate temp debris"
        );
    }

    #[test]
    fn schema_v2_reused_live_pid_remains_conservatively_owned() {
        let temp_dir = tempfile::tempdir().expect("temp dir");
        let session_id = SessionId::new();
        let access_dir = session_owner_dir(temp_dir.path(), session_id);
        fs::create_dir_all(&access_dir).expect("owner dir");
        let mut owner = SessionLeaseOwner::new(session_id, &context("historical", 4));
        owner.schema_version = 2;
        owner.pid = std::process::id();
        let owner_path = access_dir.join(format!("{}.json", owner.lease_token));
        fs::write(
            &owner_path,
            serde_json::to_vec_pretty(&owner).expect("serialize v2"),
        )
        .expect("write v2");

        assert_eq!(
            classify_owner_liveness(&owner_path, &owner),
            SessionOwnerLiveness::Live
        );
        assert!(matches!(
            acquire_session_lease(temp_dir.path(), session_id, &context("current", 5))
                .expect_err("v2 PID reuse ambiguity must fail closed"),
            SessionLeaseError::OwnedByOtherDaemon { .. }
        ));
        assert!(owner_path.exists(), "ambiguous v2 owner must not be pruned");
    }

    #[test]
    fn voluntary_release_allows_next_daemon_owner() {
        let temp_dir = tempfile::tempdir().expect("temp dir");
        let session_id = SessionId::new();
        let owner = acquire_session_lease(temp_dir.path(), session_id, &context("first", 5))
            .expect("first owner");
        drop(owner);
        let next = acquire_session_lease(temp_dir.path(), session_id, &context("next", 5))
            .expect("next owner after release");
        assert_eq!(next.owner().daemon_instance_id.as_deref(), Some("next"));
    }

    #[cfg(unix)]
    #[test]
    fn release_stop_and_killed_owner_workflows_allow_next_owner() {
        use std::process::Command;

        let temp_dir = tempfile::tempdir().expect("temp dir");
        let release_session = SessionId::new();
        let released = acquire_session_lease(
            temp_dir.path(),
            release_session,
            &context("release-owner", 5),
        )
        .expect("release owner");
        drop(released);
        acquire_session_lease(
            temp_dir.path(),
            release_session,
            &context("after-release", 5),
        )
        .expect("owner after voluntary release");

        let stop_session = SessionId::new();
        let stopped = vec![
            acquire_session_lease(
                temp_dir.path(),
                stop_session,
                &same_daemon_context("stop-client-one", "stopping-daemon", 5),
            )
            .expect("first stopping-daemon registration"),
            acquire_session_lease(
                temp_dir.path(),
                stop_session,
                &same_daemon_context("stop-client-two", "stopping-daemon", 5),
            )
            .expect("second stopping-daemon registration"),
        ];
        drop(stopped);
        acquire_session_lease(temp_dir.path(), stop_session, &context("after-stop", 5))
            .expect("owner after graceful daemon stop");

        let killed_session = SessionId::new();
        let mut child = Command::new("sh")
            .args(["-c", "exit 0"])
            .spawn()
            .expect("spawn killed-owner process");
        let dead_pid = child.id();
        assert!(child.wait().expect("wait for killed owner").success());
        let access_dir = session_owner_dir(temp_dir.path(), killed_session);
        let dead_owner = SessionLeaseOwner {
            lease_token: format!("killed-owner-{dead_pid}"),
            pid: dead_pid,
            ..SessionLeaseOwner::new(killed_session, &context("killed-daemon", 5))
        };
        let dead_path = access_dir.join(format!("{}.json", dead_owner.lease_token));
        write_owner_metadata(&dead_path, &dead_owner).expect("write killed owner metadata");
        acquire_session_lease(temp_dir.path(), killed_session, &context("after-kill", 5))
            .expect("owner after dead process pruning");
        assert!(!dead_path.exists(), "killed owner metadata must be pruned");
    }

    #[test]
    fn maintenance_activity_probe_is_non_mutating_and_tracks_guard_lifetime() {
        let temp_dir = tempfile::tempdir().expect("temp dir");
        let session_id = SessionId::new();
        assert!(!session_maintenance_is_active(temp_dir.path(), session_id).expect("idle probe"));

        let maintenance = acquire_session_maintenance_guard(temp_dir.path(), session_id)
            .expect("maintenance guard");
        assert!(
            session_maintenance_is_active(temp_dir.path(), session_id)
                .expect("active maintenance probe")
        );
        assert!(
            active_session_owners(temp_dir.path(), session_id)
                .expect("owner scan")
                .is_empty(),
            "probing must not publish owner metadata"
        );

        drop(maintenance);
        assert!(
            !session_maintenance_is_active(temp_dir.path(), session_id)
                .expect("released maintenance probe")
        );
    }

    #[test]
    fn allows_reentrant_registrations_from_one_daemon_instance() {
        let temp_dir = tempfile::tempdir().expect("temp dir");
        let session_id = SessionId::new();
        let first = acquire_session_lease(
            temp_dir.path(),
            session_id,
            &same_daemon_context("first", "daemon-1", 7),
        )
        .expect("first guard");
        let second = acquire_session_lease(
            temp_dir.path(),
            session_id,
            &same_daemon_context("second", "daemon-1", 7),
        )
        .expect("same daemon reentrant guard");

        assert_eq!(first.owner().storage_writer_epoch, Some(7));
        assert_eq!(second.owner().storage_writer_epoch, Some(7));
        assert_ne!(
            first.owner().build_fingerprint,
            second.owner().build_fingerprint
        );
    }

    #[test]
    fn rejects_second_daemon_at_same_writer_epoch() {
        let temp_dir = tempfile::tempdir().expect("temp dir");
        let session_id = SessionId::new();
        let _owner = acquire_session_lease(temp_dir.path(), session_id, &context("daemon-one", 7))
            .expect("first daemon owner");
        let error = acquire_session_lease(temp_dir.path(), session_id, &context("daemon-two", 7))
            .expect_err("second daemon must route through existing owner");
        assert!(matches!(
            error,
            SessionLeaseError::OwnedByOtherDaemon {
                owner: Some(ref owner),
                ..
            } if owner.daemon_instance_id.as_deref() == Some("daemon-one")
                && owner.storage_writer_epoch == Some(7)
        ));
    }

    #[test]
    fn rejects_live_different_storage_writer_epoch() {
        let temp_dir = tempfile::tempdir().expect("temp dir");
        let session_id = SessionId::new();
        let _first = acquire_session_lease(temp_dir.path(), session_id, &context("old", 1))
            .expect("first guard");
        let error = acquire_session_lease(temp_dir.path(), session_id, &context("new", 2))
            .expect_err("different storage writer epoch should be rejected");

        assert!(matches!(
            error,
            SessionLeaseError::OwnedByOtherDaemon { .. }
        ));
    }

    #[test]
    fn rejects_missing_storage_writer_identity() {
        let temp_dir = tempfile::tempdir().expect("temp dir");
        let session_id = SessionId::new();
        let error = acquire_session_lease(
            temp_dir.path(),
            session_id,
            &SessionLeaseOwnerContext {
                storage_writer_epoch: None,
                ..SessionLeaseOwnerContext::default()
            },
        )
        .expect_err("anonymous storage writer must be rejected");
        assert!(matches!(
            error,
            SessionLeaseError::MissingStorageWriterEpoch { .. }
        ));
    }

    #[cfg(unix)]
    #[test]
    fn prunes_dead_owner_before_compatibility_check() {
        use std::process::Command;

        let temp_dir = tempfile::tempdir().expect("temp dir");
        let session_id = SessionId::new();
        let mut child = Command::new("sh")
            .args(["-c", "exit 0"])
            .spawn()
            .expect("spawn short-lived process");
        let dead_pid = child.id();
        assert!(child.wait().expect("wait for child").success());
        assert!(
            !process_is_alive(dead_pid),
            "child pid must be dead for test"
        );

        let access_dir = session_owner_dir(temp_dir.path(), session_id);
        let owner = SessionLeaseOwner {
            lease_token: format!("dead-owner-{dead_pid}"),
            pid: dead_pid,
            ..SessionLeaseOwner::new(session_id, &context("dead-incompatible", 1))
        };
        let owner_path = access_dir.join(format!("{}.json", owner.lease_token));
        write_owner_metadata(&owner_path, &owner).expect("write dead owner record");

        let live = acquire_session_lease(temp_dir.path(), session_id, &context("live-current", 2))
            .expect("dead incompatible owner must be pruned");
        assert!(!owner_path.exists(), "dead owner record must be removed");
        assert_eq!(live.owner().storage_writer_epoch, Some(2));
    }

    #[test]
    fn owner_error_exposes_actionable_identity_without_database_access() {
        let temp_dir = tempfile::tempdir().expect("temp dir");
        let session_id = SessionId::new();
        let mut owner_context = context("owner-build", 7);
        owner_context.daemon_namespace = Some("owner-namespace".to_owned());
        owner_context.endpoint = Some("unix:///tmp/owner.sock".to_owned());
        let _owner = acquire_session_lease(temp_dir.path(), session_id, &owner_context)
            .expect("owner lease");

        let error = acquire_session_maintenance_guard(temp_dir.path(), session_id)
            .expect_err("maintenance must report owner");
        let SessionLeaseError::OwnedByOtherDaemon {
            owner: Some(owner),
            owner_summary,
            ..
        } = error
        else {
            panic!("expected owner metadata");
        };
        assert_eq!(owner.daemon_instance_id.as_deref(), Some("owner-build"));
        assert_eq!(owner.storage_writer_epoch, Some(7));
        assert_eq!(owner.endpoint.as_deref(), Some("unix:///tmp/owner.sock"));
        assert!(owner_summary.contains("daemon instance owner-build"));
        assert!(owner_summary.contains("storage writer epoch 7"));
        assert!(owner_summary.contains("namespace owner-namespace"));
        assert!(owner_summary.contains("endpoint unix:///tmp/owner.sock"));
    }

    #[test]
    fn maintenance_refuses_any_live_session_owner() {
        let temp_dir = tempfile::tempdir().expect("temp dir");
        let session_id = SessionId::new();
        let _owner = acquire_session_lease(temp_dir.path(), session_id, &context("owner", 7))
            .expect("session owner");
        assert!(matches!(
            acquire_session_maintenance_guard(temp_dir.path(), session_id)
                .expect_err("maintenance must refuse live owner"),
            SessionLeaseError::OwnedByOtherDaemon { .. }
        ));
    }

    #[test]
    #[ignore = "subprocess helper for lease_handoff_process_crash_leaves_recoverable_owner"]
    fn lease_handoff_crash_helper() {
        let root =
            PathBuf::from(std::env::var_os("BCODE_MIGRATION_LEASE_HANDOFF_ROOT").expect("root"));
        let session_id = std::env::var("BCODE_MIGRATION_LEASE_HANDOFF_SESSION_ID")
            .expect("session ID")
            .parse::<SessionId>()
            .expect("valid session ID");
        let maintenance =
            acquire_session_maintenance_guard(&root, session_id).expect("maintenance guard");
        transition_session_maintenance_to_lease(
            maintenance,
            &root,
            session_id,
            &context("handoff-crash", 7),
        )
        .expect("handoff crash boundary must abort first");
    }

    #[test]
    fn lease_handoff_process_crash_leaves_recoverable_owner() {
        let temp_dir = tempfile::tempdir().expect("temp dir");
        let session_id = SessionId::new();
        let status =
            std::process::Command::new(std::env::current_exe().expect("current test binary"))
                .args([
                    "--exact",
                    "lease::tests::lease_handoff_crash_helper",
                    "--ignored",
                    "--nocapture",
                ])
                .env("BCODE_MIGRATION_LEASE_HANDOFF_ROOT", temp_dir.path())
                .env(
                    "BCODE_MIGRATION_LEASE_HANDOFF_SESSION_ID",
                    session_id.to_string(),
                )
                .env("BCODE_MIGRATION_LEASE_HANDOFF_CRASH", "1")
                .status()
                .expect("run handoff crash child");
        assert!(
            !status.success(),
            "handoff helper must terminate abnormally"
        );

        let owners = active_session_owners(temp_dir.path(), session_id).expect("owner scan");
        assert!(owners.is_empty(), "dead handoff owner must be pruned");
        let retry = acquire_session_lease(temp_dir.path(), session_id, &context("retry", 7))
            .expect("retry ownership after handoff crash");
        assert_eq!(retry.owner().storage_writer_epoch, Some(7));
    }

    #[test]
    fn failed_maintenance_to_lease_transition_releases_coordinator() {
        let temp_dir = tempfile::tempdir().expect("temp dir");
        let session_id = SessionId::new();
        let maintenance = acquire_session_maintenance_guard(temp_dir.path(), session_id)
            .expect("maintenance guard");
        let error = transition_session_maintenance_to_lease(
            maintenance,
            temp_dir.path(),
            session_id,
            &SessionLeaseOwnerContext {
                storage_writer_epoch: None,
                ..SessionLeaseOwnerContext::default()
            },
        )
        .expect_err("missing writer identity should fail transition");
        assert!(matches!(
            error,
            SessionLeaseError::MissingStorageWriterEpoch { .. }
        ));

        let next = acquire_session_maintenance_guard(temp_dir.path(), session_id)
            .expect("failed transition must release maintenance coordinator");
        drop(next);
        assert!(
            active_session_owners(temp_dir.path(), session_id)
                .expect("owner scan")
                .is_empty()
        );
    }

    #[test]
    fn maintenance_to_lease_transition_prevents_incompatible_handoff_race() {
        let temp_dir = tempfile::tempdir().expect("temp dir");
        let session_id = SessionId::new();
        let maintenance = acquire_session_maintenance_guard(temp_dir.path(), session_id)
            .expect("maintenance guard");
        let lease = transition_session_maintenance_to_lease(
            maintenance,
            temp_dir.path(),
            session_id,
            &context("current", 7),
        )
        .expect("transition to runtime lease");
        assert_eq!(lease.owner().storage_writer_epoch, Some(7));
        assert!(matches!(
            acquire_session_lease(temp_dir.path(), session_id, &context("incompatible", 8))
                .expect_err("incompatible writer must not claim transitioned session"),
            SessionLeaseError::OwnedByOtherDaemon { .. }
        ));
        drop(lease);
        acquire_session_lease(temp_dir.path(), session_id, &context("next", 8))
            .expect("new epoch may claim after runtime lease release");
    }

    #[test]
    fn live_schema_v3_owner_is_not_pruned_across_processes() {
        use std::process::{Command, Stdio};
        use std::time::{Duration, Instant};

        let temp_dir = tempfile::tempdir().expect("temp dir");
        let session_id = SessionId::new();
        let ready = temp_dir.path().join("hold-lease-ready");
        let acquired = temp_dir.path().join("hold-lease-acquired");
        let mut child = Command::new(std::env::current_exe().expect("test executable"))
            .args([
                "--exact",
                "lease::tests::subprocess_lock_helper",
                "--nocapture",
            ])
            .env("BCODE_SESSION_LOCK_TEST_ACTION", "hold-lease")
            .env("BCODE_SESSION_LOCK_TEST_ROOT", temp_dir.path())
            .env("BCODE_SESSION_LOCK_TEST_ID", session_id.to_string())
            .env("BCODE_SESSION_LOCK_TEST_READY", &ready)
            .env("BCODE_SESSION_LOCK_TEST_ACQUIRED", &acquired)
            .stdin(Stdio::null())
            .stdout(Stdio::null())
            .stderr(Stdio::null())
            .spawn()
            .expect("spawn owner child");
        let deadline = Instant::now() + Duration::from_secs(2);
        while !acquired.exists() {
            assert!(Instant::now() < deadline, "child did not acquire lease");
            std::thread::sleep(Duration::from_millis(5));
        }
        let owners = active_session_owners(temp_dir.path(), session_id).expect("owner scan");
        assert_eq!(owners.len(), 1);
        assert_eq!(owners[0].schema_version, 3);
        assert!(matches!(
            acquire_session_lease(temp_dir.path(), session_id, &context("other", 7)),
            Err(SessionLeaseError::OwnedByOtherDaemon { .. })
        ));
        fs::remove_file(&ready).expect("release child");
        assert!(child.wait().expect("wait child").success());
        acquire_session_lease(temp_dir.path(), session_id, &context("next", 7))
            .expect("next owner after child release");
    }

    #[cfg(unix)]
    #[test]
    fn owner_record_open_error_is_unverifiable_and_fails_closed() {
        let temp_dir = tempfile::tempdir().expect("temp dir");
        let session_id = SessionId::new();
        let access_dir = session_owner_dir(temp_dir.path(), session_id);
        fs::create_dir_all(&access_dir).expect("owner dir");
        let owner = SessionLeaseOwner::new(session_id, &context("unreadable", 4));
        let owner_path = access_dir.join(format!("{}.json", owner.lease_token));
        fs::create_dir(&owner_path).expect("directory at owner record path");

        assert_eq!(
            classify_owner_liveness(&owner_path, &owner),
            SessionOwnerLiveness::Unverifiable
        );
        assert!(matches!(
            acquire_session_lease(temp_dir.path(), session_id, &context("current", 5))
                .expect_err("unverifiable owner path must fail closed"),
            SessionLeaseError::UnverifiableOwnerMetadata { .. }
        ));
        assert!(
            owner_path.is_dir(),
            "failed classification must not mutate owner path"
        );
    }

    #[test]
    fn unknown_owner_schema_is_unverifiable_and_fails_closed() {
        let temp_dir = tempfile::tempdir().expect("temp dir");
        let session_id = SessionId::new();
        let access_dir = session_owner_dir(temp_dir.path(), session_id);
        fs::create_dir_all(&access_dir).expect("owner dir");
        let mut owner = SessionLeaseOwner::new(session_id, &context("future", 5));
        owner.schema_version = 99;
        let owner_path = access_dir.join(format!("{}.json", owner.lease_token));
        fs::write(
            &owner_path,
            serde_json::to_vec_pretty(&owner).expect("serialize owner"),
        )
        .expect("write owner");
        assert_eq!(
            classify_owner_liveness(&owner_path, &owner),
            SessionOwnerLiveness::Unverifiable
        );
        assert!(matches!(
            acquire_session_lease(temp_dir.path(), session_id, &context("next", 5)),
            Err(SessionLeaseError::OwnedByOtherDaemon { .. })
        ));
        assert!(owner_path.exists());
    }

    #[test]
    fn malformed_owner_metadata_fails_closed_during_mutating_prune() {
        let temp_dir = tempfile::tempdir().expect("temp dir");
        let session_id = SessionId::new();
        let access_dir = session_owner_dir(temp_dir.path(), session_id);
        fs::create_dir_all(&access_dir).expect("owner dir");
        let malformed = access_dir.join("malformed.json");
        fs::write(&malformed, b"not-json").expect("malformed metadata");
        assert!(matches!(
            acquire_session_lease(temp_dir.path(), session_id, &context("next", 7)),
            Err(SessionLeaseError::UnverifiableOwnerMetadata { path }) if path == malformed
        ));
        assert!(
            malformed.exists(),
            "unverifiable metadata must not be pruned"
        );
    }

    #[test]
    fn subprocess_lock_helper() {
        let Ok(action) = std::env::var("BCODE_SESSION_LOCK_TEST_ACTION") else {
            return;
        };
        let root = PathBuf::from(std::env::var_os("BCODE_SESSION_LOCK_TEST_ROOT").expect("root"));
        let session_id = std::env::var("BCODE_SESSION_LOCK_TEST_ID")
            .expect("session id")
            .parse::<SessionId>()
            .expect("valid session id");
        let ready =
            PathBuf::from(std::env::var_os("BCODE_SESSION_LOCK_TEST_READY").expect("ready"));
        let acquired =
            PathBuf::from(std::env::var_os("BCODE_SESSION_LOCK_TEST_ACQUIRED").expect("acquired"));
        fs::write(&ready, b"ready").expect("write ready marker");
        match action.as_str() {
            "lease" => {
                let _guard = acquire_session_lease(&root, session_id, &context("subprocess", 7))
                    .expect("acquire subprocess lease");
                fs::write(acquired, b"acquired").expect("write acquired marker");
            }
            "hold-lease" => {
                let _guard = acquire_session_lease(&root, session_id, &context("subprocess", 7))
                    .expect("acquire subprocess lease");
                fs::write(acquired, b"acquired").expect("write acquired marker");
                while ready.exists() {
                    std::thread::sleep(std::time::Duration::from_millis(5));
                }
            }
            "write" => {
                let _guard = acquire_session_write_lock(&root, session_id)
                    .expect("acquire subprocess write lock");
                fs::write(acquired, b"acquired").expect("write acquired marker");
            }
            other => panic!("unknown subprocess lock action {other}"),
        }
    }

    #[cfg(unix)]
    fn assert_subprocess_blocked_by_maintenance(action: &str) {
        use std::process::{Command, Stdio};
        use std::thread;
        use std::time::{Duration, Instant};

        let temp_dir = tempfile::tempdir().expect("temp dir");
        let session_id = SessionId::new();
        let maintenance = acquire_session_maintenance_guard(temp_dir.path(), session_id)
            .expect("maintenance guard");
        let ready = temp_dir.path().join(format!("{action}-ready"));
        let acquired = temp_dir.path().join(format!("{action}-acquired"));
        let mut child = Command::new(std::env::current_exe().expect("current test executable"))
            .args([
                "--exact",
                "lease::tests::subprocess_lock_helper",
                "--nocapture",
            ])
            .env("BCODE_SESSION_LOCK_TEST_ACTION", action)
            .env("BCODE_SESSION_LOCK_TEST_ROOT", temp_dir.path())
            .env("BCODE_SESSION_LOCK_TEST_ID", session_id.to_string())
            .env("BCODE_SESSION_LOCK_TEST_READY", &ready)
            .env("BCODE_SESSION_LOCK_TEST_ACQUIRED", &acquired)
            .stdin(Stdio::null())
            .stdout(Stdio::null())
            .stderr(Stdio::null())
            .spawn()
            .expect("spawn lock subprocess");
        let ready_deadline = Instant::now() + Duration::from_secs(5);
        while !ready.exists() {
            assert!(
                Instant::now() < ready_deadline,
                "subprocess did not become ready"
            );
            assert!(
                child.try_wait().expect("inspect subprocess").is_none(),
                "subprocess exited before attempting the lock"
            );
            thread::sleep(Duration::from_millis(10));
        }
        thread::sleep(Duration::from_millis(150));
        assert!(!acquired.exists(), "subprocess bypassed maintenance lock");
        assert!(
            child
                .try_wait()
                .expect("inspect blocked subprocess")
                .is_none(),
            "subprocess exited while maintenance still held the lock"
        );
        drop(maintenance);
        let exit_deadline = Instant::now() + Duration::from_secs(5);
        loop {
            if let Some(status) = child.try_wait().expect("wait for subprocess") {
                assert!(status.success(), "subprocess lock helper failed: {status}");
                break;
            }
            assert!(
                Instant::now() < exit_deadline,
                "subprocess remained blocked after release"
            );
            thread::sleep(Duration::from_millis(10));
        }
        assert!(
            acquired.exists(),
            "subprocess did not acquire released lock"
        );
    }

    #[cfg(unix)]
    #[test]
    fn maintenance_blocks_session_lease_in_another_process() {
        assert_subprocess_blocked_by_maintenance("lease");
    }

    #[cfg(unix)]
    #[test]
    fn maintenance_blocks_session_write_lock_in_another_process() {
        assert_subprocess_blocked_by_maintenance("write");
    }

    #[test]
    fn maintenance_coordinator_blocks_new_session_lease_until_release() {
        let temp_dir = tempfile::tempdir().expect("temp dir");
        let session_id = SessionId::new();
        let maintenance = acquire_session_maintenance_guard(temp_dir.path(), session_id)
            .expect("maintenance guard");
        let root = temp_dir.path().to_path_buf();
        let (started_tx, started_rx) = std::sync::mpsc::channel();
        let (finished_tx, finished_rx) = std::sync::mpsc::channel();
        let join = std::thread::spawn(move || {
            started_tx.send(()).expect("started signal");
            let result = acquire_session_lease(&root, session_id, &context("writer", 7));
            finished_tx.send(result.is_ok()).expect("finished signal");
        });
        started_rx.recv().expect("lease attempt started");
        assert!(
            finished_rx
                .recv_timeout(std::time::Duration::from_millis(100))
                .is_err(),
            "lease acquisition must remain blocked while maintenance owns the coordinator"
        );
        drop(maintenance);
        assert!(
            finished_rx
                .recv_timeout(std::time::Duration::from_secs(2))
                .expect("lease attempt should finish after maintenance release")
        );
        join.join().expect("lease thread");
    }

    #[test]
    fn allows_different_storage_writer_epoch_after_guard_drop() {
        let temp_dir = tempfile::tempdir().expect("temp dir");
        let session_id = SessionId::new();
        {
            let _first = acquire_session_lease(temp_dir.path(), session_id, &context("old", 1))
                .expect("first guard");
        }

        acquire_session_lease(temp_dir.path(), session_id, &context("new", 2))
            .expect("different writer after drop");
    }
}