magi-code 0.63.4

#[cfg(unix)]
use std::os::raw::c_int;
use std::{
    collections::HashMap,
    error::Error,
    fmt, fs,
    io::{ErrorKind, Read, Write},
    path::{Path, PathBuf},
    sync::{
        Arc, Mutex, OnceLock, Weak,
        atomic::{AtomicBool, Ordering},
    },
    thread,
    time::{Duration, Instant, SystemTime, UNIX_EPOCH},
};

const LOCK_WAIT_TIMEOUT: Duration = Duration::from_secs(30);
const LOCK_POLL_INTERVAL: Duration = Duration::from_millis(10);
const STALE_LOCK_MAX_AGE: Duration = Duration::from_secs(5);
/// How often a live lock holder refreshes its lease by touching the lock file mtime. A holder that
/// stops renewing (crash/kill) lets the lease age past `STALE_LOCK_MAX_AGE`, after which a
/// contender may safely recover it. Kept well under the lease window so renewal survives jitter.
const LEASE_RENEWAL_INTERVAL: Duration = Duration::from_secs(1);

/// Best-effort cross-process file lock using atomic lock-file creation.
///
/// This keeps read-modify-write state updates and JSONL appends single-writer across magi-code
/// processes without adding dependencies. Lock files record the owner PID so a killed Unix process
/// can be detected and recovered before the bounded timeout backstop.
pub(crate) struct CrossProcessFileLock {
    path: PathBuf,
    owner: LockSnapshot,
    renewer: Option<LeaseRenewer>,
}

impl CrossProcessFileLock {
    pub(crate) fn acquire(target: &Path) -> anyhow::Result<Self> {
        let parent = target.parent().ok_or_else(|| {
            anyhow::anyhow!("lock target path has no parent: {}", target.display())
        })?;
        fs::create_dir_all(parent)?;
        let path = lock_path(target);
        let start = Instant::now();
        loop {
            match fs::OpenOptions::new()
                .write(true)
                .create_new(true)
                .open(&path)
            {
                Ok(mut file) => {
                    writeln!(file, "pid={}", std::process::id())?;
                    writeln!(file, "token={}", lock_owner_token())?;
                    file.flush()?;
                    let owner = LockSnapshot::read(&path)?.ok_or_else(|| {
                        anyhow::anyhow!(
                            "created file lock disappeared before ownership snapshot: {}",
                            path.display()
                        )
                    })?;
                    let renewer = LeaseRenewer::spawn(path.clone(), owner.clone());
                    return Ok(Self {
                        path,
                        owner,
                        renewer: Some(renewer),
                    });
                }
                Err(error) if error.kind() == ErrorKind::AlreadyExists => {
                    if recover_stale_lock(&path)? {
                        continue;
                    }
                    if start.elapsed() >= LOCK_WAIT_TIMEOUT {
                        anyhow::bail!(
                            "timed out waiting for file lock {} for {}",
                            path.display(),
                            target.display()
                        );
                    }
                    thread::sleep(LOCK_POLL_INTERVAL);
                }
                Err(error) => return Err(error.into()),
            }
        }
    }
}

#[cfg(unix)]
fn recover_stale_lock(path: &Path) -> anyhow::Result<bool> {
    let Some(observed) = LockSnapshot::read(path)? else {
        return Ok(true);
    };
    match lock_pid_from_contents(&observed.contents) {
        LockPid::Alive => Ok(false),
        LockPid::Dead => remove_lock_file_if_unchanged(path, &observed),
        LockPid::MissingOrCorrupt => steal_lock_if_old_enough(path, &observed),
    }
}

#[cfg(not(unix))]
fn recover_stale_lock(path: &Path) -> anyhow::Result<bool> {
    let Some(observed) = LockSnapshot::read(path)? else {
        return Ok(true);
    };
    recover_lock_if_lease_expired(path, &observed, STALE_LOCK_MAX_AGE)
}

/// Lease-based stale recovery: a lock is recoverable only if its holder stopped renewing the lease
/// (lock mtime older than `max_age`). On non-Unix there is no portable PID-liveness probe, so a
/// renewed lease is the liveness signal: a live holder refreshes the mtime via [`LeaseRenewer`],
/// and only a holder that stopped renewing (crash/kill) lets the lease expire. Gated so the
/// non-Unix semantics can be exercised by tests on Unix hosts.
#[cfg(any(not(unix), test))]
fn recover_lock_if_lease_expired(
    path: &Path,
    observed: &LockSnapshot,
    max_age: Duration,
) -> anyhow::Result<bool> {
    if !observed.is_older_than(max_age) {
        return Ok(false);
    }
    remove_lock_file_if_unchanged(path, observed)
}

#[cfg(unix)]
fn steal_lock_if_old_enough(path: &Path, observed: &LockSnapshot) -> anyhow::Result<bool> {
    if !observed.is_older_than(STALE_LOCK_MAX_AGE) {
        return Ok(false);
    }
    remove_lock_file_if_unchanged(path, observed)
}

fn remove_lock_file_if_unchanged(path: &Path, observed: &LockSnapshot) -> anyhow::Result<bool> {
    let Some(_guard) = StaleLockRecoveryGuard::try_acquire(path)? else {
        return Ok(false);
    };
    let Some(current) = LockSnapshot::read(path)? else {
        return Ok(true);
    };
    if !current.same_file_and_contents(observed) {
        return Ok(false);
    }
    match fs::remove_file(path) {
        Ok(()) => Ok(true),
        Err(error) if error.kind() == ErrorKind::NotFound => Ok(true),
        Err(error) => Err(error.into()),
    }
}

struct StaleLockRecoveryGuard {
    path: PathBuf,
    owner: LockSnapshot,
}

impl StaleLockRecoveryGuard {
    fn try_acquire(lock_path: &Path) -> anyhow::Result<Option<Self>> {
        let path = stale_lock_recovery_guard_path(lock_path);
        match fs::OpenOptions::new()
            .write(true)
            .create_new(true)
            .open(&path)
        {
            Ok(mut file) => {
                writeln!(file, "pid={}", std::process::id())?;
                file.flush()?;
                let owner = LockSnapshot::read(&path)?.ok_or_else(|| {
                    anyhow::anyhow!(
                        "created recovery guard disappeared before ownership snapshot: {}",
                        path.display()
                    )
                })?;
                Ok(Some(Self { path, owner }))
            }
            Err(error) if error.kind() == ErrorKind::AlreadyExists => {
                let Some(observed) = LockSnapshot::read(&path)? else {
                    return Ok(None);
                };
                let orphaned = {
                    #[cfg(unix)]
                    {
                        match lock_pid_from_contents(&observed.contents) {
                            LockPid::Dead => true,
                            LockPid::MissingOrCorrupt => observed.is_older_than(STALE_LOCK_MAX_AGE),
                            LockPid::Alive => false,
                        }
                    }
                    #[cfg(not(unix))]
                    {
                        observed.is_older_than(STALE_LOCK_MAX_AGE)
                    }
                };
                if orphaned {
                    let Some(current) = LockSnapshot::read(&path)? else {
                        return Ok(None);
                    };
                    if current.same_file_and_contents(&observed) {
                        match fs::remove_file(&path) {
                            Ok(()) => return Self::try_acquire(lock_path),
                            Err(error) if error.kind() == ErrorKind::NotFound => {}
                            Err(error) => return Err(error.into()),
                        }
                    }
                }
                Ok(None)
            }
            Err(error) => Err(error.into()),
        }
    }
}

impl Drop for StaleLockRecoveryGuard {
    fn drop(&mut self) {
        if let Err(error) = remove_owned_lock_file(&self.path, &self.owner) {
            eprintln!(
                "warning: failed to clean up stale-lock recovery guard {}: {error:#}",
                self.path.display()
            );
        }
    }
}

fn stale_lock_recovery_guard_path(path: &Path) -> PathBuf {
    let name = path
        .file_name()
        .and_then(|name| name.to_str())
        .unwrap_or("state.lock");
    // ponytail: recovery guard is best-effort stdlib coordination for magi-code contenders;
    // external unlink/replacement actors can still race outside the protocol. Use OS file locks
    // if cross-tool mutual exclusion is required.
    path.with_file_name(format!(".{name}.recovery"))
}

#[cfg(unix)]
#[derive(Debug, Clone, PartialEq, Eq)]
struct LockSnapshot {
    dev: u64,
    ino: u64,
    modified: Option<SystemTime>,
    contents: Vec<u8>,
}

#[cfg(not(unix))]
#[derive(Debug, Clone, PartialEq, Eq)]
struct LockSnapshot {
    modified: Option<SystemTime>,
    contents: Vec<u8>,
}

#[cfg(unix)]
impl LockSnapshot {
    fn read(path: &Path) -> anyhow::Result<Option<Self>> {
        use std::os::unix::fs::MetadataExt;

        let mut file = match fs::File::open(path) {
            Ok(file) => file,
            Err(error) if error.kind() == ErrorKind::NotFound => return Ok(None),
            Err(error) => return Err(error.into()),
        };
        let metadata = file.metadata()?;
        let mut contents = Vec::new();
        file.read_to_end(&mut contents)?;
        Ok(Some(Self {
            dev: metadata.dev(),
            ino: metadata.ino(),
            modified: metadata.modified().ok(),
            contents,
        }))
    }

    fn is_older_than(&self, max_age: Duration) -> bool {
        self.modified
            .and_then(|modified| modified.elapsed().ok())
            .is_some_and(|age| age >= max_age)
    }

    fn same_file_and_contents(&self, other: &Self) -> bool {
        self.dev == other.dev && self.ino == other.ino && self.contents == other.contents
    }
}

#[cfg(not(unix))]
impl LockSnapshot {
    fn read(path: &Path) -> anyhow::Result<Option<Self>> {
        let mut file = match fs::File::open(path) {
            Ok(file) => file,
            Err(error) if error.kind() == ErrorKind::NotFound => return Ok(None),
            Err(error) => return Err(error.into()),
        };
        let metadata = file.metadata()?;
        let mut contents = Vec::new();
        file.read_to_end(&mut contents)?;
        Ok(Some(Self {
            modified: metadata.modified().ok(),
            contents,
        }))
    }

    fn is_older_than(&self, max_age: Duration) -> bool {
        self.modified
            .and_then(|modified| modified.elapsed().ok())
            .is_some_and(|age| age >= max_age)
    }

    fn same_file_and_contents(&self, other: &Self) -> bool {
        self.contents == other.contents
    }
}

#[cfg(unix)]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum LockPid {
    Alive,
    Dead,
    MissingOrCorrupt,
}

#[cfg(unix)]
fn lock_pid_from_contents(contents: &[u8]) -> LockPid {
    let Ok(contents) = std::str::from_utf8(contents) else {
        return LockPid::MissingOrCorrupt;
    };
    let Some(pid) = contents.lines().find_map(parse_lock_pid) else {
        return LockPid::MissingOrCorrupt;
    };
    pid_liveness(pid)
}

#[cfg(unix)]
fn parse_lock_pid(line: &str) -> Option<u32> {
    line.strip_prefix("pid=")?.trim().parse().ok()
}

#[cfg(unix)]
fn pid_liveness(pid: u32) -> LockPid {
    let Ok(pid) = c_int::try_from(pid) else {
        return LockPid::MissingOrCorrupt;
    };
    if pid <= 0 {
        return LockPid::MissingOrCorrupt;
    }
    // SAFETY: `kill(pid, 0)` sends no signal; it only asks the OS whether `pid` exists and whether
    // this process may signal it. `pid` is validated positive and converted to platform `c_int`.
    let result = unsafe { kill(pid, 0) };
    if result == 0 {
        return LockPid::Alive;
    }
    match std::io::Error::last_os_error().raw_os_error() {
        Some(ESRCH) => LockPid::Dead,
        Some(EPERM) => LockPid::Alive,
        _ => LockPid::Alive,
    }
}

/// POSIX errno returned by `kill(2)` when `pid` does not exist. The numeric value `3` (ESRCH) is
/// stable across all supported Unix targets (Linux, macOS, FreeBSD, NetBSD, OpenBSD).
#[cfg(unix)]
const ESRCH: c_int = 3;
/// POSIX errno returned by `kill(2)` when `pid` exists but this process may not signal it; treated
/// as alive. The numeric value `1` (EPERM) is stable across supported Unix targets.
#[cfg(unix)]
const EPERM: c_int = 1;

// `kill(2)` per POSIX: `int kill(pid_t pid, int sig)`. `pid_t` is `c_int` on all supported Unix
// targets, so the FFI uses `c_int` rather than a hand-assumed fixed-width `i32`.
#[cfg(unix)]
unsafe extern "C" {
    fn kill(pid: c_int, sig: c_int) -> c_int;
}

struct LeaseRenewer {
    stop: Arc<AtomicBool>,
    thread: Option<thread::Thread>,
    handle: Option<thread::JoinHandle<()>>,
}

impl LeaseRenewer {
    fn spawn(path: PathBuf, owner: LockSnapshot) -> Self {
        let stop = Arc::new(AtomicBool::new(false));
        let stop_for_thread = stop.clone();
        let handle = thread::spawn(move || renew_loop(path, owner, stop_for_thread));
        let thread = handle.thread().clone();
        Self {
            stop,
            thread: Some(thread),
            handle: Some(handle),
        }
    }
}

/// Refreshes the held lock's mtime on [`LEASE_RENEWAL_INTERVAL`] so a contender's lease-expiry
/// check only recovers locks whose holder stopped renewing. Stops as soon as ownership changes or
/// the renewer is signaled to stop.
fn renew_loop(path: PathBuf, owner: LockSnapshot, stop: Arc<AtomicBool>) {
    loop {
        thread::park_timeout(LEASE_RENEWAL_INTERVAL);
        if stop.load(Ordering::Acquire) {
            break;
        }
        // Only renew while we still own this exact lock file; stop silently if replaced.
        let Ok(Some(current)) = LockSnapshot::read(&path) else {
            break;
        };
        if !current.same_file_and_contents(&owner) {
            break;
        }
        if let Ok(file) = fs::OpenOptions::new().write(true).open(&path) {
            let _ = file.set_modified(SystemTime::now());
        }
    }
}

impl Drop for LeaseRenewer {
    fn drop(&mut self) {
        self.stop.store(true, Ordering::Release);
        if let Some(thread) = self.thread.take() {
            thread.unpark();
        }
        if let Some(handle) = self.handle.take() {
            let _ = handle.join();
        }
    }
}

impl Drop for CrossProcessFileLock {
    fn drop(&mut self) {
        // Stop the lease renewer before removing the file so it never touches the file during or
        // after the ownership-checked removal.
        self.renewer.take();
        if let Err(error) = remove_owned_lock_file(&self.path, &self.owner) {
            eprintln!(
                "warning: failed to clean up file lock {}: {error:#}",
                self.path.display()
            );
        }
    }
}

fn remove_owned_lock_file(path: &Path, owner: &LockSnapshot) -> anyhow::Result<()> {
    let Some(current) = LockSnapshot::read(path)? else {
        return Ok(());
    };
    if !current.same_file_and_contents(owner) {
        anyhow::bail!(
            "lock file ownership changed; leaving replacement lock in place: {}",
            path.display()
        );
    }
    match fs::remove_file(path) {
        Ok(()) => Ok(()),
        Err(error) if error.kind() == ErrorKind::NotFound => Ok(()),
        Err(error) => Err(error.into()),
    }
}

fn lock_owner_token() -> String {
    let stamp = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_default()
        .as_nanos();
    format!("{}-{stamp}", std::process::id())
}

fn lock_path(path: &Path) -> PathBuf {
    let name = path
        .file_name()
        .and_then(|name| name.to_str())
        .unwrap_or("state");
    path.with_file_name(format!(".{name}.lock"))
}

static IN_PROCESS_FILE_LOCKS: OnceLock<Mutex<HashMap<PathBuf, Weak<Mutex<()>>>>> = OnceLock::new();

pub(crate) fn in_process_file_lock(
    path: &Path,
    registry_label: &'static str,
) -> anyhow::Result<Arc<Mutex<()>>> {
    let key = normalize_lock_path(path);
    let registry = IN_PROCESS_FILE_LOCKS.get_or_init(|| Mutex::new(HashMap::new()));
    let mut locks = registry
        .lock()
        .map_err(|_| anyhow::anyhow!("{registry_label} lock registry was poisoned"))?;
    if let Some(lock) = locks.get(&key).and_then(Weak::upgrade) {
        return Ok(lock);
    }
    locks.retain(|_, lock| lock.strong_count() > 0);
    let lock = Arc::new(Mutex::new(()));
    locks.insert(key, Arc::downgrade(&lock));
    Ok(lock)
}

fn normalize_lock_path(path: &Path) -> PathBuf {
    if let Ok(canonical) = path.canonicalize() {
        return canonical;
    }
    if let (Some(parent), Some(file_name)) = (path.parent(), path.file_name())
        && let Ok(parent) = parent.canonicalize()
    {
        return parent.join(file_name);
    }
    path.to_path_buf()
}

#[derive(Debug)]
pub(crate) struct AtomicWriteCommittedButUndurable {
    path: PathBuf,
    parent: PathBuf,
    source: anyhow::Error,
}

impl AtomicWriteCommittedButUndurable {
    #[cfg(test)]
    fn path(&self) -> &Path {
        &self.path
    }
}

impl fmt::Display for AtomicWriteCommittedButUndurable {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(
            formatter,
            "atomic write committed to {} but parent directory sync failed for {}; file contents changed but durability is uncertain",
            self.path.display(),
            self.parent.display()
        )
    }
}

impl Error for AtomicWriteCommittedButUndurable {
    fn source(&self) -> Option<&(dyn Error + 'static)> {
        Some(self.source.as_ref())
    }
}

pub(crate) fn atomic_write(path: &Path, bytes: &[u8]) -> anyhow::Result<()> {
    atomic_write_with_permissions(path, bytes, None)
}

pub(crate) fn atomic_write_with_permissions(
    path: &Path,
    bytes: &[u8],
    #[allow(unused_variables)] unix_mode: Option<u32>,
) -> anyhow::Result<()> {
    atomic_write_with_permissions_and_parent_sync(path, bytes, unix_mode, sync_parent_dir)
}

fn atomic_write_with_permissions_and_parent_sync(
    path: &Path,
    bytes: &[u8],
    #[allow(unused_variables)] unix_mode: Option<u32>,
    sync_parent: impl FnOnce(&Path) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
    let parent = path
        .parent()
        .ok_or_else(|| anyhow::anyhow!("target path has no parent: {}", path.display()))?;
    fs::create_dir_all(parent)?;
    let temp = temp_path(path);
    let mut temp_guard = TempFileCleanupGuard::new(temp.clone());
    let target_mode = unix_target_mode(path, unix_mode)?;
    let mut options = fs::OpenOptions::new();
    options.write(true).create_new(true).truncate(true);
    #[cfg(unix)]
    {
        use std::os::unix::fs::OpenOptionsExt;
        options.mode(target_mode.unwrap_or(0o666));
    }
    let mut file = options.open(&temp)?;
    file.write_all(bytes)?;
    file.flush()?;
    file.sync_all()?;
    drop(file);
    #[cfg(unix)]
    if let Some(mode) = target_mode {
        use std::os::unix::fs::PermissionsExt;
        fs::set_permissions(&temp, fs::Permissions::from_mode(mode))?;
    }
    fs::rename(&temp, path)?;
    temp_guard.disarm();
    sync_parent(parent).map_err(|error| AtomicWriteCommittedButUndurable {
        path: path.to_path_buf(),
        parent: parent.to_path_buf(),
        source: error,
    })?;
    Ok(())
}

#[cfg(unix)]
fn unix_target_mode(path: &Path, unix_mode: Option<u32>) -> anyhow::Result<Option<u32>> {
    use std::os::unix::fs::PermissionsExt;

    if unix_mode.is_some() {
        return Ok(unix_mode);
    }
    match fs::metadata(path) {
        Ok(metadata) => Ok(Some(metadata.permissions().mode() & 0o777)),
        Err(error) if error.kind() == ErrorKind::NotFound => Ok(None),
        Err(error) => Err(error.into()),
    }
}

#[cfg(not(unix))]
fn unix_target_mode(_path: &Path, unix_mode: Option<u32>) -> anyhow::Result<Option<u32>> {
    Ok(unix_mode)
}

struct TempFileCleanupGuard {
    path: Option<PathBuf>,
}

impl TempFileCleanupGuard {
    fn new(path: PathBuf) -> Self {
        Self { path: Some(path) }
    }

    fn disarm(&mut self) {
        self.path = None;
    }
}

impl Drop for TempFileCleanupGuard {
    fn drop(&mut self) {
        if let Some(path) = &self.path {
            let _ = fs::remove_file(path);
        }
    }
}

fn temp_path(path: &Path) -> PathBuf {
    let stamp = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_default()
        .as_nanos();
    let pid = std::process::id();
    let name = path
        .file_name()
        .and_then(|name| name.to_str())
        .unwrap_or("atomic");
    path.with_file_name(format!(".{name}.{pid}.{stamp}.tmp"))
}

pub(crate) fn sync_parent_dir(parent: &Path) -> anyhow::Result<()> {
    #[cfg(unix)]
    {
        fs::File::open(parent)?.sync_all()?;
    }
    #[cfg(not(unix))]
    {
        let _ = parent;
        // ponytail: stdlib exposes no portable directory fsync on non-Unix; keep atomic rename
        // behavior and rely on platform/filesystem durability semantics unless native support is added.
    }
    Ok(())
}

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

    #[cfg(unix)]
    #[test]
    fn parse_lock_pid_accepts_pid_line() {
        assert_eq!(parse_lock_pid("pid=1234"), Some(1234));
        assert_eq!(parse_lock_pid("pid= 1234"), Some(1234));
        assert_eq!(parse_lock_pid("owner=1234"), None);
        assert_eq!(parse_lock_pid("pid=not-a-pid"), None);
    }

    #[cfg(unix)]
    #[test]
    fn lock_pid_from_contents_treats_corrupt_content_as_missing_or_corrupt() {
        assert_eq!(
            lock_pid_from_contents(b"not a lock"),
            LockPid::MissingOrCorrupt
        );
    }

    #[cfg(unix)]
    #[test]
    fn recover_stale_lock_removes_dead_pid_lock() {
        let dir = tempfile::tempdir().expect("tempdir");
        let lock = dir.path().join("state.lock");
        let mut child = std::process::Command::new("sh")
            .arg("-c")
            .arg("exit 0")
            .spawn()
            .expect("spawn child");
        let pid = child.id();
        child.wait().expect("wait child");
        fs::write(&lock, format!("pid={pid}\n")).expect("write lock");

        assert!(recover_stale_lock(&lock).expect("recover lock"));
        assert!(!lock.exists());
    }

    #[cfg(unix)]
    #[test]
    fn recover_stale_lock_reclaims_orphaned_recovery_guard() {
        let dir = tempfile::tempdir().expect("tempdir");
        let lock = dir.path().join("state.lock");
        let guard_path = stale_lock_recovery_guard_path(&lock);
        let mut child = std::process::Command::new("sh")
            .arg("-c")
            .arg("exit 0")
            .spawn()
            .expect("spawn child");
        let pid = child.id();
        child.wait().expect("wait child");
        fs::write(&lock, format!("pid={pid}\n")).expect("write lock");
        fs::write(&guard_path, format!("pid={pid}\n")).expect("write orphaned guard");

        assert!(recover_stale_lock(&lock).expect("recover lock"));
        assert!(!lock.exists());
        assert!(!guard_path.exists());
    }

    #[cfg(unix)]
    #[test]
    fn recover_stale_lock_keeps_live_pid_lock() {
        let dir = tempfile::tempdir().expect("tempdir");
        let lock = dir.path().join("state.lock");
        fs::write(&lock, format!("pid={}\n", std::process::id())).expect("write lock");

        assert!(!recover_stale_lock(&lock).expect("recover lock"));
        assert!(lock.exists());
    }

    #[cfg(unix)]
    #[test]
    fn stale_lock_removal_keeps_replaced_lock() {
        let dir = tempfile::tempdir().expect("tempdir");
        let lock = dir.path().join("state.lock");
        fs::write(&lock, "pid=1\n").expect("write observed lock");
        let observed = LockSnapshot::read(&lock)
            .expect("read snapshot")
            .expect("snapshot exists");
        fs::remove_file(&lock).expect("remove observed lock");
        fs::write(&lock, format!("pid={}\n", std::process::id())).expect("write replacement lock");

        assert!(!remove_lock_file_if_unchanged(&lock, &observed).expect("guarded remove"));
        assert_eq!(
            fs::read_to_string(&lock).expect("replacement remains"),
            format!("pid={}\n", std::process::id())
        );
    }

    #[cfg(unix)]
    #[test]
    fn stale_lock_recovery_guard_blocks_parallel_removal() {
        let dir = tempfile::tempdir().expect("tempdir");
        let lock = dir.path().join("state.lock");
        fs::write(&lock, "pid=1\n").expect("write observed lock");
        let observed = LockSnapshot::read(&lock)
            .expect("read snapshot")
            .expect("snapshot exists");
        let _guard = StaleLockRecoveryGuard::try_acquire(&lock)
            .expect("acquire guard")
            .expect("guard acquired");

        assert!(!remove_lock_file_if_unchanged(&lock, &observed).expect("guarded remove"));
        assert_eq!(fs::read_to_string(&lock).expect("lock remains"), "pid=1\n");
    }

    #[test]
    fn stale_lock_recovery_guard_drop_keeps_replacement_guard() {
        let dir = tempfile::tempdir().expect("tempdir");
        let lock = dir.path().join("state.lock");
        let guard_path = stale_lock_recovery_guard_path(&lock);
        let guard = StaleLockRecoveryGuard::try_acquire(&lock)
            .expect("acquire guard")
            .expect("guard acquired");
        fs::remove_file(&guard_path).expect("remove original guard");
        fs::write(&guard_path, "pid=999999\n").expect("write replacement guard");

        drop(guard);

        assert_eq!(
            fs::read_to_string(&guard_path).expect("replacement remains"),
            "pid=999999\n"
        );
    }

    #[test]
    fn stale_lock_recovery_guard_is_removed_on_drop() {
        let dir = tempfile::tempdir().expect("tempdir");
        let lock = dir.path().join("state.lock");
        let guard_path = stale_lock_recovery_guard_path(&lock);
        {
            let _guard = StaleLockRecoveryGuard::try_acquire(&lock)
                .expect("acquire guard")
                .expect("guard acquired");
            assert!(guard_path.exists());
        }
        assert!(!guard_path.exists());
    }

    #[test]
    fn temp_file_cleanup_guard_removes_armed_temp_on_drop() {
        let dir = tempfile::tempdir().expect("tempdir");
        let temp = dir.path().join(".state.tmp");
        fs::write(&temp, "partial").expect("write temp");

        {
            let _guard = TempFileCleanupGuard::new(temp.clone());
            assert!(temp.exists());
        }

        assert!(!temp.exists());
    }

    #[test]
    fn temp_file_cleanup_guard_leaves_disarmed_temp_on_drop() {
        let dir = tempfile::tempdir().expect("tempdir");
        let temp = dir.path().join(".state.tmp");
        fs::write(&temp, "complete").expect("write temp");

        {
            let mut guard = TempFileCleanupGuard::new(temp.clone());
            guard.disarm();
        }

        assert_eq!(fs::read_to_string(&temp).expect("temp remains"), "complete");
    }

    #[cfg(unix)]
    #[test]
    fn sync_parent_dir_reports_missing_parent() {
        let dir = tempfile::tempdir().expect("tempdir");
        let missing = dir.path().join("missing");

        assert!(sync_parent_dir(&missing).is_err());
    }

    #[test]
    fn cross_process_lock_drop_keeps_replacement_lock_file() {
        let dir = tempfile::tempdir().expect("tempdir");
        let target = dir.path().join("state.json");
        let lock = lock_path(&target);
        let guard = CrossProcessFileLock::acquire(&target).expect("acquire lock");
        fs::remove_file(&lock).expect("remove owned lock");
        fs::write(&lock, "pid=999999\ntoken=replacement\n").expect("write replacement lock");

        drop(guard);

        assert_eq!(
            fs::read_to_string(&lock).expect("replacement remains"),
            "pid=999999\ntoken=replacement\n"
        );
    }

    #[test]
    fn atomic_write_reports_committed_but_undurable_after_parent_sync_failure() {
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("state.json");
        fs::write(&path, "old").expect("write original");

        let error = atomic_write_with_permissions_and_parent_sync(&path, b"new", None, |_| {
            Err(anyhow::anyhow!(std::io::Error::other("sync failed")))
        })
        .expect_err("parent sync failure should report error");
        let committed = error
            .downcast_ref::<AtomicWriteCommittedButUndurable>()
            .expect("committed-but-undurable error");

        assert_eq!(committed.path(), path.as_path());
        assert_eq!(fs::read_to_string(&path).expect("read committed"), "new");
    }

    #[cfg(unix)]
    #[test]
    fn atomic_write_preserves_existing_mode_without_explicit_mode() {
        use std::os::unix::fs::PermissionsExt;

        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("state.json");
        fs::write(&path, "old").expect("write original");
        fs::set_permissions(&path, fs::Permissions::from_mode(0o640)).expect("chmod original");

        atomic_write(&path, b"new").expect("atomic write");

        assert_eq!(fs::read_to_string(&path).expect("read updated"), "new");
        assert_eq!(
            fs::metadata(&path).expect("metadata").permissions().mode() & 0o777,
            0o640
        );
    }

    #[cfg(unix)]
    #[test]
    fn atomic_write_explicit_mode_still_wins() {
        use std::os::unix::fs::PermissionsExt;

        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("state.json");
        fs::write(&path, "old").expect("write original");
        fs::set_permissions(&path, fs::Permissions::from_mode(0o640)).expect("chmod original");

        atomic_write_with_permissions(&path, b"new", Some(0o600)).expect("atomic write");

        assert_eq!(fs::read_to_string(&path).expect("read updated"), "new");
        assert_eq!(
            fs::metadata(&path).expect("metadata").permissions().mode() & 0o777,
            0o600
        );
    }
    #[test]
    fn lease_renewer_refreshes_lock_mtime_while_held() {
        let dir = tempfile::tempdir().expect("tempdir");
        let target = dir.path().join("state.json");
        let lock_path = lock_path(&target);
        let guard = CrossProcessFileLock::acquire(&target).expect("acquire lock");
        let before = fs::metadata(&lock_path)
            .and_then(|m| m.modified())
            .expect("initial mtime");
        // Renewer runs every LEASE_RENEWAL_INTERVAL; wait past one renewal cycle.
        thread::sleep(LEASE_RENEWAL_INTERVAL * 2);
        let after = fs::metadata(&lock_path)
            .and_then(|m| m.modified())
            .expect("renewed mtime");
        assert!(
            after > before,
            "lease renewer did not refresh lock mtime while held"
        );
        drop(guard);
        assert!(!lock_path.exists(), "lock removed on drop");
    }

    #[test]
    fn recover_lock_if_lease_expired_keeps_recent_lock() {
        let dir = tempfile::tempdir().expect("tempdir");
        let lock = dir.path().join("state.lock");
        fs::write(&lock, "pid=1\ntoken=abc\n").expect("write lock");
        let observed = LockSnapshot::read(&lock)
            .expect("read snapshot")
            .expect("snapshot exists");

        assert!(
            !recover_lock_if_lease_expired(&lock, &observed, Duration::from_secs(5))
                .expect("recover")
        );
        assert!(lock.exists(), "recent lock must not be stolen");
    }

    #[test]
    fn recover_lock_if_lease_expired_removes_unrenewed_lock() {
        let dir = tempfile::tempdir().expect("tempdir");
        let lock = dir.path().join("state.lock");
        fs::write(&lock, "pid=1\ntoken=abc\n").expect("write lock");
        // Force the lock to look abandoned: mtime older than the lease window.
        let stale = SystemTime::now() - Duration::from_secs(60);
        fs::File::open(&lock)
            .expect("open lock")
            .set_modified(stale)
            .expect("age lock");
        let observed = LockSnapshot::read(&lock)
            .expect("read snapshot")
            .expect("snapshot exists");

        assert!(
            recover_lock_if_lease_expired(&lock, &observed, Duration::from_secs(5))
                .expect("recover")
        );
        assert!(!lock.exists(), "unrenewed lock must be recovered");
    }
}