truth_mirror/

watcher.rs

//! Queue-tied watcher lifecycle: single-flight lock, crash-safe staleness, and
//! detached spawn.
//!
//! The watcher's lifetime is bound to the review queue, not to any agent session
//! or terminal. [`ensure_watcher`] is an idempotent check-and-spawn: if a live
//! watcher already owns the state dir it does nothing, otherwise it spawns
//! `truth-mirror watch --until-empty` as a detached background process and returns
//! immediately. It is safe to call from a git hook — fast, quiet on success, and
//! never blocking.
//!
//! Single-flight is enforced with a lock file ([`WATCHER_LOCK_FILE`]) recorded in
//! the state dir. The lock carries a [`ProcessIdentity`] — a pid plus a start-time
//! token — so a dead watcher never wedges the next spawn and a reused pid cannot
//! fool the liveness probe: the recorded start token must match the live process's
//! start token for the lock to count as held.

use std::{
    fs,
    io::{self, Write},
    path::{Path, PathBuf},
    process::Stdio,
    time::{SystemTime, UNIX_EPOCH},
};

use serde::{Deserialize, Serialize};
use thiserror::Error;

use crate::{reviewer::ReviewQueue, time::unix_now};

/// File name of the single-flight watcher lock inside the state dir.
pub const WATCHER_LOCK_FILE: &str = "watcher.lock";
const WATCHER_RECLAIM_DIR: &str = "watcher.lock.reclaim";

/// A just-written lock is treated as held even if its recorded owner cannot yet
/// be confirmed alive, to cover the window where a parent `ensure-watcher` has
/// claimed the lock and is about to hand it to its detached child (the parent may
/// exit before a racing caller probes it). Without this settling grace, two
/// `ensure-watcher` calls racing within milliseconds could both judge the other's
/// fresh lock "stale" and each spawn a watcher. Kept small so a genuinely crashed
/// spawn is still reclaimed promptly.
const LOCK_SETTLING_SECS: u64 = 10;

#[derive(Debug, Error)]
pub enum WatcherError {
    #[error("watcher lock io error: {0}")]
    Io(io::Error),
    #[error("watcher lock json error: {0}")]
    Json(serde_json::Error),
    #[error("failed to resolve the truth-mirror executable path: {0}")]
    ResolveExe(io::Error),
    #[error("failed to spawn detached watcher: {0}")]
    Spawn(io::Error),
    #[error("failed to read review queue: {0}")]
    Queue(crate::reviewer::ReviewerError),
}

/// A process's identity strong enough to survive pid reuse.
///
/// `pid` alone is not enough: the OS recycles pids, so a *different* process could
/// later hold the pid recorded in a stale lock. `start_token` is an opaque,
/// process-specific string (the OS-reported start time) that changes when the pid
/// is reused, so a lock is only "held" when both the pid is alive AND its live
/// start token matches the recorded one.
#[derive(Clone, Debug, Eq, PartialEq, Deserialize, Serialize)]
pub struct ProcessIdentity {
    pub pid: u32,
    /// OS-reported process start marker. Empty when the platform could not report
    /// one; liveness then degrades to a bare pid-alive check (documented tradeoff).
    #[serde(default)]
    pub start_token: String,
}

impl ProcessIdentity {
    /// The identity of the current process.
    pub fn current() -> Self {
        let pid = std::process::id();
        Self {
            start_token: probe_start_token(pid).unwrap_or_default(),
            pid,
        }
    }
}

/// On-disk single-flight lock. Records who owns the watcher slot and when it was
/// claimed. Serialized as pretty JSON so it is human-inspectable in the state dir.
#[derive(Clone, Debug, Eq, PartialEq, Deserialize, Serialize)]
pub struct WatcherLock {
    pub identity: ProcessIdentity,
    pub created_at_unix: u64,
}

impl WatcherLock {
    fn for_current() -> Self {
        Self {
            identity: ProcessIdentity::current(),
            created_at_unix: unix_now(),
        }
    }
}

/// Outcome of probing whether a recorded lock owner is still alive.
///
/// Kept as a distinct type (rather than a bare `bool`) so callers can treat
/// `Unknown` conservatively — an unprobeable owner is not silently reclaimed.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum Liveness {
    /// The recorded owner is alive and its start token matches: the lock is held.
    Alive,
    /// The recorded owner is gone (pid dead) or was replaced (pid reused): the
    /// lock is stale and may be reclaimed.
    Stale,
}

/// Decide whether a recorded lock owner is still the live watcher.
///
/// Pure over its inputs — the probed identity is injected — so the staleness
/// decision is unit-testable without any real processes:
/// - pid not alive        → `Stale`
/// - pid alive, tokens differ (pid reuse) → `Stale`
/// - pid alive, tokens match, or either token empty → `Alive`
///
/// When either token is empty (platform could not report a start marker) the
/// decision falls back to bare pid liveness, which is the best signal available.
pub fn decide_liveness(recorded: &ProcessIdentity, probed: Option<ProcessIdentity>) -> Liveness {
    let Some(probed) = probed else {
        return Liveness::Stale;
    };
    if probed.pid != recorded.pid {
        return Liveness::Stale;
    }
    if recorded.start_token.is_empty() || probed.start_token.is_empty() {
        // No start marker to compare on at least one side; pid is alive, so treat
        // as held rather than risk reclaiming a live watcher.
        return Liveness::Alive;
    }
    if probed.start_token == recorded.start_token {
        Liveness::Alive
    } else {
        Liveness::Stale
    }
}

/// Whether a lock counts as held right now: either its recorded owner probes
/// [`Liveness::Alive`], or the lock is still within its settling window (a spawn is
/// in flight). `now` is injected so the settling decision is unit-testable.
///
/// The settling clause is one-directional safety: it can only make a fresh lock
/// look *more* held, never reclaim a live one, so it cannot spawn a duplicate.
fn lock_is_held_at(lock: &WatcherLock, probed: Option<ProcessIdentity>, now: u64) -> bool {
    if decide_liveness(&lock.identity, probed) == Liveness::Alive {
        return true;
    }
    now.saturating_sub(lock.created_at_unix) < LOCK_SETTLING_SECS
}

/// Whether a lock counts as held now, probing the live process and clock.
fn lock_is_held(lock: &WatcherLock) -> bool {
    let probed = probe_identity(lock.identity.pid);
    lock_is_held_at(lock, probed, unix_now())
}

/// Probe the live identity for `pid`, or `None` if no such process exists.
fn probe_identity(pid: u32) -> Option<ProcessIdentity> {
    if !pid_is_alive(pid) {
        return None;
    }
    Some(ProcessIdentity {
        pid,
        start_token: probe_start_token(pid).unwrap_or_default(),
    })
}

/// Whether a live watcher currently owns the lock in `state_dir`.
///
/// Reads the lock (if any) and probes its recorded owner. A missing, unreadable,
/// or stale lock all mean "no live owner". Never mutates the lock.
pub fn watcher_is_alive(state_dir: &Path) -> bool {
    match read_lock(state_dir) {
        Ok(Some(lock)) => lock_is_held(&lock),
        _ => false,
    }
}

fn lock_path(state_dir: &Path) -> PathBuf {
    state_dir.join(WATCHER_LOCK_FILE)
}

#[derive(Clone, Debug, Eq, PartialEq)]
enum LockRead {
    Missing,
    Valid(WatcherLock),
    Corrupt { modified_at_unix: u64 },
}

impl LockRead {
    fn is_held_at(&self, now: u64) -> bool {
        match self {
            LockRead::Missing => false,
            LockRead::Valid(lock) => lock_is_held_at(lock, probe_identity(lock.identity.pid), now),
            LockRead::Corrupt { modified_at_unix } => {
                corrupt_lock_is_held_at(*modified_at_unix, now)
            }
        }
    }
}

fn corrupt_lock_is_held_at(modified_at_unix: u64, now: u64) -> bool {
    now.saturating_sub(modified_at_unix) < LOCK_SETTLING_SECS
}

fn read_lock(state_dir: &Path) -> Result<Option<WatcherLock>, WatcherError> {
    match read_lock_state(state_dir)? {
        LockRead::Valid(lock) => Ok(Some(lock)),
        LockRead::Missing | LockRead::Corrupt { .. } => Ok(None),
    }
}

fn read_lock_state(state_dir: &Path) -> Result<LockRead, WatcherError> {
    let path = lock_path(state_dir);
    match fs::read_to_string(&path) {
        Ok(contents) => match serde_json::from_str(contents.trim()) {
            Ok(lock) => Ok(LockRead::Valid(lock)),
            Err(_) => Ok(LockRead::Corrupt {
                modified_at_unix: file_modified_at_unix(&path),
            }),
        },
        Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(LockRead::Missing),
        Err(error) => Err(WatcherError::Io(error)),
    }
}

fn file_modified_at_unix(path: &Path) -> u64 {
    fs::metadata(path)
        .and_then(|metadata| metadata.modified())
        .ok()
        .and_then(|modified| modified.duration_since(UNIX_EPOCH).ok())
        .map_or_else(unix_now, |duration| duration.as_secs())
}

fn write_lock(state_dir: &Path, lock: &WatcherLock) -> Result<(), WatcherError> {
    fs::create_dir_all(state_dir).map_err(WatcherError::Io)?;
    let bytes = serde_json::to_vec_pretty(lock).map_err(WatcherError::Json)?;
    let temp_path = temp_lock_path(state_dir);
    {
        let mut file = fs::File::create(&temp_path).map_err(WatcherError::Io)?;
        file.write_all(&bytes).map_err(WatcherError::Io)?;
        file.write_all(b"\n").map_err(WatcherError::Io)?;
        file.sync_all().map_err(WatcherError::Io)?;
    }
    fs::rename(&temp_path, lock_path(state_dir)).map_err(WatcherError::Io)
}

fn temp_lock_path(state_dir: &Path) -> PathBuf {
    let nanos = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map_or(0, |duration| duration.as_nanos());
    state_dir.join(format!(
        "{WATCHER_LOCK_FILE}.{}.{}.tmp",
        std::process::id(),
        nanos
    ))
}

/// Result of attempting to claim the single-flight watcher lock.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum LockClaim {
    /// This process now owns the lock (it was free or the prior owner was stale).
    Acquired,
    /// A live watcher already owns the lock; the caller should stand down.
    HeldByLiveWatcher,
}

/// Try to claim the watcher lock for the current process, atomically and
/// crash-safe.
///
/// The claim uses an O_EXCL create as the racing gate: at most one of two
/// concurrent `create_new` calls succeeds, so two `ensure-watcher` invocations
/// racing (e.g. two pushes) resolve to exactly one [`LockClaim::Acquired`]. If the
/// exclusive create loses the race because a lock file already exists, the loser
/// inspects the existing owner:
/// - a live owner ⇒ [`LockClaim::HeldByLiveWatcher`] (stand down),
/// - a stale owner ⇒ overwrite in place and take it ([`LockClaim::Acquired`]).
///
/// Reclaiming a stale lock never blocks on the dead owner, so a crashed watcher
/// can never wedge the next spawn.
pub fn try_acquire_lock(state_dir: &Path) -> Result<LockClaim, WatcherError> {
    fs::create_dir_all(state_dir).map_err(WatcherError::Io)?;
    let lock = WatcherLock::for_current();
    let bytes = serde_json::to_vec_pretty(&lock).map_err(WatcherError::Json)?;

    match fs::OpenOptions::new()
        .write(true)
        .create_new(true)
        .open(lock_path(state_dir))
    {
        Ok(mut file) => {
            file.write_all(&bytes).map_err(WatcherError::Io)?;
            file.write_all(b"\n").map_err(WatcherError::Io)?;
            file.sync_all().map_err(WatcherError::Io)?;
            Ok(LockClaim::Acquired)
        }
        Err(error) if error.kind() == io::ErrorKind::AlreadyExists => {
            // Lost the create race — inspect the incumbent. A held lock (live owner
            // OR still settling after a fresh claim) means stand down; only a truly
            // stale lock is reclaimed. An unparseable lock means the winner is
            // mid-write (it created the file with O_EXCL but has not written its
            // bytes yet): stand down rather than clobber an in-flight claim.
            reclaim_existing_lock(state_dir, &lock)
        }
        Err(error) => Err(WatcherError::Io(error)),
    }
}

fn reclaim_existing_lock(state_dir: &Path, lock: &WatcherLock) -> Result<LockClaim, WatcherError> {
    let now = unix_now();
    let state = read_lock_state(state_dir)?;
    if state.is_held_at(now) {
        return Ok(LockClaim::HeldByLiveWatcher);
    }

    let Some(_guard) = try_reclaim_guard(state_dir)? else {
        return Ok(LockClaim::HeldByLiveWatcher);
    };

    let state = read_lock_state(state_dir)?;
    if state.is_held_at(unix_now()) {
        return Ok(LockClaim::HeldByLiveWatcher);
    }

    match fs::remove_file(lock_path(state_dir)) {
        Ok(()) => {}
        Err(error) if error.kind() == io::ErrorKind::NotFound => {}
        Err(error) => return Err(WatcherError::Io(error)),
    }

    match fs::OpenOptions::new()
        .write(true)
        .create_new(true)
        .open(lock_path(state_dir))
    {
        Ok(mut file) => {
            let bytes = serde_json::to_vec_pretty(lock).map_err(WatcherError::Json)?;
            file.write_all(&bytes).map_err(WatcherError::Io)?;
            file.write_all(b"\n").map_err(WatcherError::Io)?;
            file.sync_all().map_err(WatcherError::Io)?;
            Ok(LockClaim::Acquired)
        }
        Err(error) if error.kind() == io::ErrorKind::AlreadyExists => {
            Ok(LockClaim::HeldByLiveWatcher)
        }
        Err(error) => Err(WatcherError::Io(error)),
    }
}

struct ReclaimGuard {
    path: PathBuf,
}

impl Drop for ReclaimGuard {
    fn drop(&mut self) {
        let _ = fs::remove_dir(&self.path);
    }
}

fn try_reclaim_guard(state_dir: &Path) -> Result<Option<ReclaimGuard>, WatcherError> {
    let path = state_dir.join(WATCHER_RECLAIM_DIR);
    match fs::create_dir(&path) {
        Ok(()) => Ok(Some(ReclaimGuard { path })),
        Err(error) if error.kind() == io::ErrorKind::AlreadyExists => {
            if corrupt_lock_is_held_at(file_modified_at_unix(&path), unix_now()) {
                return Ok(None);
            }
            match fs::remove_dir(&path) {
                Ok(()) => {}
                Err(error) if error.kind() == io::ErrorKind::NotFound => {}
                Err(error) => return Err(WatcherError::Io(error)),
            }
            match fs::create_dir(&path) {
                Ok(()) => Ok(Some(ReclaimGuard { path })),
                Err(error) if error.kind() == io::ErrorKind::AlreadyExists => Ok(None),
                Err(error) => Err(WatcherError::Io(error)),
            }
        }
        Err(error) => Err(WatcherError::Io(error)),
    }
}

/// Release the watcher lock iff the current process still owns it.
///
/// A watcher that reclaimed a stale slot or lost a race must not delete a lock now
/// held by a *different* live watcher, so this is a no-op unless the recorded
/// identity matches the current process.
pub fn release_lock_if_owned(state_dir: &Path) -> Result<(), WatcherError> {
    let Some(lock) = read_lock(state_dir)? else {
        return Ok(());
    };
    if !identity_matches_owner(&lock.identity, &ProcessIdentity::current()) {
        return Ok(());
    }
    match fs::remove_file(lock_path(state_dir)) {
        Ok(()) => {}
        Err(error) if error.kind() == io::ErrorKind::NotFound => {}
        Err(error) => return Err(WatcherError::Io(error)),
    }
    Ok(())
}

fn identity_matches_owner(recorded: &ProcessIdentity, current: &ProcessIdentity) -> bool {
    recorded.pid == current.pid
        && (recorded.start_token.is_empty()
            || current.start_token.is_empty()
            || recorded.start_token == current.start_token)
}

/// Idempotent check-and-spawn for the queue-tied watcher.
///
/// If a live watcher already owns `state_dir`, returns without spawning. Otherwise
/// claims the single-flight lock and spawns a detached `truth-mirror watch
/// --until-empty` that outlives this process (and its terminal / git hook), then
/// returns immediately. Quiet on success; safe to call after every enqueue.
///
/// The spawned watcher — not this caller — is responsible for releasing the lock
/// when it exits, so this returning `Acquired` means "a watcher is now starting",
/// not "a watcher is done".
pub fn ensure_watcher(state_dir: &Path) -> Result<LockClaim, WatcherError> {
    ensure_watcher_with_spawner(state_dir, spawn_detached_watcher)
}

fn ensure_watcher_with_spawner(
    state_dir: &Path,
    spawn: impl FnOnce(&Path) -> Result<u32, WatcherError>,
) -> Result<LockClaim, WatcherError> {
    // Fast path: a live watcher already owns the slot, so do nothing without even
    // touching the lock file. Keeps hook latency negligible in the common case.
    if watcher_is_alive(state_dir) {
        return Ok(LockClaim::HeldByLiveWatcher);
    }

    match try_acquire_lock(state_dir)? {
        LockClaim::HeldByLiveWatcher => Ok(LockClaim::HeldByLiveWatcher),
        LockClaim::Acquired => {
            // We hold the lock, but this parent is about to exit — so immediately
            // re-point the lock at the detached child's identity. Otherwise a
            // second `ensure-watcher` racing right after this one could see a lock
            // owned by our (now-dead) pid, judge it stale, and spawn a *second*
            // watcher. Recording the child's identity closes that window without
            // depending on the child having started up yet.
            let child_pid = match spawn(state_dir) {
                Ok(pid) => pid,
                Err(error) => {
                    if let Err(release_error) = release_lock_if_owned(state_dir) {
                        eprintln!(
                            "truth-mirror ensure-watcher: failed to release lock after spawn failure: {release_error}"
                        );
                    }
                    return Err(error);
                }
            };
            let child_lock = WatcherLock {
                identity: ProcessIdentity {
                    start_token: probe_start_token(child_pid).unwrap_or_default(),
                    pid: child_pid,
                },
                created_at_unix: unix_now(),
            };
            write_lock(state_dir, &child_lock)?;
            Ok(LockClaim::Acquired)
        }
    }
}

/// Spawn `truth-mirror watch --until-empty` detached from this process, returning
/// the child's pid.
///
/// Detachment is platform-specific but kept thin:
/// - unix: a new process group ([`process_group(0)`]) so the child is not killed
///   when the parent's terminal / git hook exits, with stdio redirected to null.
/// - windows: `DETACHED_PROCESS | CREATE_NEW_PROCESS_GROUP` creation flags.
///
/// The state dir is passed explicitly so the child watches the same queue.
fn spawn_detached_watcher(state_dir: &Path) -> Result<u32, WatcherError> {
    let exe = std::env::current_exe().map_err(WatcherError::ResolveExe)?;
    let mut command = std::process::Command::new(exe);
    command
        .arg("--state-dir")
        .arg(state_dir)
        .arg("watch")
        .arg("--until-empty")
        .stdin(Stdio::null())
        .stdout(Stdio::null())
        .stderr(Stdio::null());

    detach(&mut command);

    let child = command.spawn().map_err(WatcherError::Spawn)?;
    Ok(child.id())
}

#[cfg(unix)]
fn detach(command: &mut std::process::Command) {
    use std::os::unix::process::CommandExt as _;
    // Own process group: the child survives the parent terminal / hook exiting.
    command.process_group(0);
}

#[cfg(windows)]
fn detach(command: &mut std::process::Command) {
    use std::os::windows::process::CommandExt as _;
    // DETACHED_PROCESS | CREATE_NEW_PROCESS_GROUP | CREATE_NO_WINDOW.
    const DETACHED_PROCESS: u32 = 0x0000_0008;
    const CREATE_NEW_PROCESS_GROUP: u32 = 0x0000_0200;
    const CREATE_NO_WINDOW: u32 = 0x0800_0000;
    command.creation_flags(DETACHED_PROCESS | CREATE_NEW_PROCESS_GROUP | CREATE_NO_WINDOW);
}

#[cfg(not(any(unix, windows)))]
fn detach(_command: &mut std::process::Command) {}

/// Probe whether `pid` is alive.
#[cfg(unix)]
pub fn pid_is_alive(pid: u32) -> bool {
    std::process::Command::new("kill")
        .arg("-0")
        .arg(pid.to_string())
        .stdout(Stdio::null())
        .stderr(Stdio::null())
        .status()
        .map(|status| status.success())
        .unwrap_or(false)
}

#[cfg(windows)]
pub fn pid_is_alive(pid: u32) -> bool {
    // `tasklist` filtered to the pid prints a row iff the process exists.
    let output = std::process::Command::new("tasklist")
        .args(["/FI", &format!("PID eq {pid}"), "/NH", "/FO", "CSV"])
        .stderr(Stdio::null())
        .output();
    match output {
        Ok(output) if output.status.success() => {
            let text = String::from_utf8_lossy(&output.stdout);
            text.contains(&pid.to_string())
        }
        _ => false,
    }
}

#[cfg(not(any(unix, windows)))]
pub fn pid_is_alive(_pid: u32) -> bool {
    false
}

/// OS-reported start marker for `pid`, used as the pid-reuse-defeating token.
#[cfg(unix)]
fn probe_start_token(pid: u32) -> Option<String> {
    // `ps -o lstart=` prints the absolute process start timestamp on both Linux
    // and macOS. It changes when the pid is reused, which is exactly the identity
    // signal we need. Empty/failed output degrades to bare pid liveness.
    let output = std::process::Command::new("ps")
        .args(["-o", "lstart=", "-p", &pid.to_string()])
        .stderr(Stdio::null())
        .output()
        .ok()?;
    if !output.status.success() {
        return None;
    }
    let token = String::from_utf8_lossy(&output.stdout).trim().to_owned();
    (!token.is_empty()).then_some(token)
}

#[cfg(windows)]
fn probe_start_token(pid: u32) -> Option<String> {
    // Query the process creation date; it changes on pid reuse. `wmic` is gone
    // from current Windows 11 installs, so use CIM through PowerShell instead.
    let script = format!(
        "$p = Get-CimInstance Win32_Process -Filter 'ProcessId = {pid}' -ErrorAction Stop; \
         if ($null -ne $p) {{ $p.CreationDate.ToUniversalTime().ToString('o') }}"
    );
    let output = std::process::Command::new("powershell")
        .args(["-NoProfile", "-Command", &script])
        .stderr(Stdio::null())
        .output()
        .ok()?;
    if !output.status.success() {
        return None;
    }
    parse_powershell_start_token_output(&output.stdout)
}

#[cfg(any(windows, test))]
fn parse_powershell_start_token_output(stdout: &[u8]) -> Option<String> {
    let text = String::from_utf8_lossy(stdout);
    let token = text
        .lines()
        .find_map(|line| {
            let trimmed = line.trim();
            (!trimmed.is_empty()).then_some(trimmed)
        })
        .map(str::trim)
        .map(str::to_owned)?;
    (!token.is_empty()).then_some(token)
}

#[cfg(not(any(unix, windows)))]
fn probe_start_token(_pid: u32) -> Option<String> {
    None
}

/// Summarize the watcher lock for `status` output: `None` when no live watcher,
/// otherwise the owning pid.
pub fn live_watcher_pid(state_dir: &Path) -> Option<u32> {
    match read_lock(state_dir) {
        Ok(Some(lock)) => lock_is_held(&lock).then_some(lock.identity.pid),
        _ => None,
    }
}

/// Whether the review queue in `state_dir` currently has pending items.
pub fn queue_has_pending(state_dir: &Path) -> Result<bool, WatcherError> {
    let summary = ReviewQueue::new(state_dir)
        .summary()
        .map_err(WatcherError::Queue)?;
    Ok(summary.pending_count > 0)
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::{
        Arc, Barrier,
        atomic::{AtomicUsize, Ordering},
    };

    fn identity(pid: u32, token: &str) -> ProcessIdentity {
        ProcessIdentity {
            pid,
            start_token: token.to_owned(),
        }
    }

    #[test]
    fn dead_pid_is_stale() {
        let recorded = identity(4242, "Mon Jan 1 00:00:00 2026");
        assert_eq!(decide_liveness(&recorded, None), Liveness::Stale);
    }

    #[test]
    fn reused_pid_with_different_start_token_is_stale() {
        let recorded = identity(4242, "Mon Jan 1 00:00:00 2026");
        let probed = identity(4242, "Tue Feb 2 11:11:11 2026");
        assert_eq!(decide_liveness(&recorded, Some(probed)), Liveness::Stale);
    }

    #[test]
    fn same_pid_and_start_token_is_alive() {
        let recorded = identity(4242, "Mon Jan 1 00:00:00 2026");
        let probed = recorded.clone();
        assert_eq!(decide_liveness(&recorded, Some(probed)), Liveness::Alive);
    }

    #[test]
    fn empty_start_token_falls_back_to_pid_liveness() {
        let recorded = identity(4242, "");
        let probed = identity(4242, "some-token");
        assert_eq!(decide_liveness(&recorded, Some(probed)), Liveness::Alive);

        let recorded = identity(4242, "recorded-token");
        let probed = identity(4242, "");
        assert_eq!(decide_liveness(&recorded, Some(probed)), Liveness::Alive);
    }

    #[test]
    fn different_pid_is_stale_even_with_matching_token() {
        let recorded = identity(4242, "same-token");
        let probed = identity(9999, "same-token");
        assert_eq!(decide_liveness(&recorded, Some(probed)), Liveness::Stale);
    }

    #[test]
    fn acquire_then_reacquire_by_live_current_process_stands_down() {
        let temp = tempfile::tempdir().unwrap();
        let state = temp.path();

        // First claim succeeds via the exclusive create.
        assert_eq!(try_acquire_lock(state).unwrap(), LockClaim::Acquired);

        // The current process IS the recorded owner and is alive, so a second
        // claim from this same process observes a live owner and stands down.
        assert_eq!(
            try_acquire_lock(state).unwrap(),
            LockClaim::HeldByLiveWatcher
        );

        // And a liveness check agrees.
        assert!(watcher_is_alive(state));
        assert_eq!(live_watcher_pid(state), Some(std::process::id()));
    }

    #[test]
    fn empty_or_corrupt_lock_is_held_only_during_settling_window() {
        assert!(corrupt_lock_is_held_at(1_000, 1_000));
        assert!(corrupt_lock_is_held_at(
            1_000,
            1_000 + LOCK_SETTLING_SECS - 1
        ));
        assert!(!corrupt_lock_is_held_at(1_000, 1_000 + LOCK_SETTLING_SECS));
    }

    #[test]
    fn empty_lock_file_does_not_error_or_get_reclaimed_while_settling() {
        let temp = tempfile::tempdir().unwrap();
        let state = temp.path();
        fs::create_dir_all(state).unwrap();
        fs::write(lock_path(state), "").unwrap();

        assert!(matches!(
            read_lock_state(state).unwrap(),
            LockRead::Corrupt { .. }
        ));
        assert_eq!(
            try_acquire_lock(state).unwrap(),
            LockClaim::HeldByLiveWatcher
        );
    }

    #[test]
    fn stale_lock_from_dead_pid_is_reclaimed() {
        let temp = tempfile::tempdir().unwrap();
        let state = temp.path();

        // Plant a lock owned by a pid that is (almost certainly) dead, with a
        // start token so liveness cannot silently pass on the fallback path.
        let stale = WatcherLock {
            identity: identity(4_000_000_000, "definitely-not-a-live-token"),
            created_at_unix: 1,
        };
        write_lock(state, &stale).unwrap();
        assert!(!watcher_is_alive(state));

        // The next claim reclaims the stale slot rather than wedging.
        assert_eq!(try_acquire_lock(state).unwrap(), LockClaim::Acquired);
        assert_eq!(live_watcher_pid(state), Some(std::process::id()));
    }

    #[test]
    fn stale_lock_with_unknown_start_token_from_dead_pid_is_reclaimed() {
        let temp = tempfile::tempdir().unwrap();
        let state = temp.path();

        let stale = WatcherLock {
            identity: identity(4_000_000_000, ""),
            created_at_unix: 1,
        };
        write_lock(state, &stale).unwrap();

        assert_eq!(try_acquire_lock(state).unwrap(), LockClaim::Acquired);
        assert_eq!(live_watcher_pid(state), Some(std::process::id()));
    }

    #[test]
    fn threaded_stale_lock_reclaim_race_yields_exactly_one_winner() {
        const THREADS: usize = 8;

        let temp = tempfile::tempdir().unwrap();
        let state = Arc::new(temp.path().to_path_buf());
        let stale = WatcherLock {
            identity: identity(4_000_000_000, "definitely-not-a-live-token"),
            created_at_unix: 1,
        };
        write_lock(state.as_ref(), &stale).unwrap();

        let barrier = Arc::new(Barrier::new(THREADS));
        let acquired = Arc::new(AtomicUsize::new(0));
        let held = Arc::new(AtomicUsize::new(0));
        let mut threads = Vec::with_capacity(THREADS);

        for _ in 0..THREADS {
            let state = Arc::clone(&state);
            let barrier = Arc::clone(&barrier);
            let acquired = Arc::clone(&acquired);
            let held = Arc::clone(&held);
            threads.push(std::thread::spawn(move || {
                barrier.wait();
                match try_acquire_lock(state.as_ref()).unwrap() {
                    LockClaim::Acquired => {
                        acquired.fetch_add(1, Ordering::SeqCst);
                    }
                    LockClaim::HeldByLiveWatcher => {
                        held.fetch_add(1, Ordering::SeqCst);
                    }
                }
            }));
        }

        for thread in threads {
            thread.join().unwrap();
        }

        assert_eq!(acquired.load(Ordering::SeqCst), 1);
        assert_eq!(held.load(Ordering::SeqCst), THREADS - 1);
    }

    #[test]
    fn release_is_a_noop_when_not_owner() {
        let temp = tempfile::tempdir().unwrap();
        let state = temp.path();

        let other = WatcherLock {
            identity: identity(4_000_000_000, "other-token"),
            created_at_unix: 1,
        };
        write_lock(state, &other).unwrap();

        // We are not the recorded owner, so release leaves the lock intact.
        release_lock_if_owned(state).unwrap();
        assert!(read_lock(state).unwrap().is_some());
    }

    #[test]
    fn release_removes_lock_when_owner() {
        let temp = tempfile::tempdir().unwrap();
        let state = temp.path();

        assert_eq!(try_acquire_lock(state).unwrap(), LockClaim::Acquired);
        release_lock_if_owned(state).unwrap();
        assert!(read_lock(state).unwrap().is_none());
    }

    #[test]
    fn release_removes_lock_when_owner_start_token_is_unknown() {
        let temp = tempfile::tempdir().unwrap();
        let state = temp.path();

        let current = WatcherLock {
            identity: identity(std::process::id(), ""),
            created_at_unix: 1,
        };
        write_lock(state, &current).unwrap();

        release_lock_if_owned(state).unwrap();
        assert!(read_lock(state).unwrap().is_none());
    }

    #[test]
    fn ensure_watcher_releases_claimed_lock_when_spawn_fails() {
        let temp = tempfile::tempdir().unwrap();
        let state = temp.path();

        let error = ensure_watcher_with_spawner(state, |_| {
            Err(WatcherError::Spawn(io::Error::other("spawn failed")))
        })
        .unwrap_err();

        assert!(matches!(error, WatcherError::Spawn(_)));
        assert!(read_lock(state).unwrap().is_none());
        assert_eq!(try_acquire_lock(state).unwrap(), LockClaim::Acquired);
    }

    #[test]
    fn powershell_start_token_parser_uses_first_nonempty_line() {
        let output = b"\r\n  2026-07-09T01:02:03.0000000Z  \r\n";

        assert_eq!(
            parse_powershell_start_token_output(output).as_deref(),
            Some("2026-07-09T01:02:03.0000000Z")
        );
    }

    #[test]
    fn powershell_start_token_parser_rejects_empty_output() {
        assert_eq!(parse_powershell_start_token_output(b"\r\n  \n"), None);
    }

    #[test]
    fn current_identity_has_our_pid() {
        assert_eq!(ProcessIdentity::current().pid, std::process::id());
    }

    #[test]
    fn fresh_lock_with_dead_owner_is_held_during_settling_window() {
        // A just-claimed lock whose recorded pid can't be confirmed alive (the
        // parent that claimed it may have already exited) still counts as held
        // while settling, so a racing caller stands down instead of double-spawning.
        let lock = WatcherLock {
            identity: identity(4_000_000_000, "dead-token"),
            created_at_unix: 1_000,
        };
        // now == created (elapsed 0) → within settling → held.
        assert!(lock_is_held_at(&lock, None, 1_000));
        // Just before the settling boundary → still held.
        assert!(lock_is_held_at(&lock, None, 1_000 + LOCK_SETTLING_SECS - 1));
    }

    #[test]
    fn stale_lock_past_settling_window_is_not_held() {
        let lock = WatcherLock {
            identity: identity(4_000_000_000, "dead-token"),
            created_at_unix: 1_000,
        };
        // At/after the settling boundary with a dead owner → reclaimable.
        assert!(!lock_is_held_at(&lock, None, 1_000 + LOCK_SETTLING_SECS));
        assert!(!lock_is_held_at(&lock, None, 5_000));
    }

    #[test]
    fn live_owner_is_held_regardless_of_lock_age() {
        // Even an old lock is held if its recorded owner is alive with a matching
        // start token — settling only ever *adds* held-ness, never removes it.
        let lock = WatcherLock {
            identity: identity(4242, "live-token"),
            created_at_unix: 1,
        };
        let probed = identity(4242, "live-token");
        assert!(lock_is_held_at(&lock, Some(probed), 9_999_999));
    }
}