truth-mirror 0.13.1

//! 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,
    sync::atomic::{AtomicU64, Ordering},
    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";
static LOCK_TOKEN_COUNTER: AtomicU64 = AtomicU64::new(0);

fn new_lock_token() -> String {
    let nanos = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map_or(0, |duration| duration.as_nanos());
    let counter = LOCK_TOKEN_COUNTER.fetch_add(1, Ordering::Relaxed);
    format!("{}-{nanos}-{counter}", std::process::id())
}

/// 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.
pub(crate) 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),
    #[error("review queue is owned by {owner}")]
    QueueOwned { owner: String },
}

/// 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,
        }
    }

    pub(crate) fn matches(&self, other: &Self) -> bool {
        self.pid == other.pid
            && (self.start_token.is_empty()
                || other.start_token.is_empty()
                || self.start_token == other.start_token)
    }
}

/// 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,
    #[serde(default)]
    pub acquisition_token: String,
    pub created_at_unix: u64,
}

impl WatcherLock {
    fn for_current() -> Self {
        Self {
            acquisition_token: new_lock_token(),
            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,
    /// Probe error or identity mismatch could not be resolved safely.
    /// A lock with unknown state is treated as held.
    Unknown,
    /// 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:
/// - worker proven dead or replaced → `Stale`
/// - worker probe inconclusive → `Unknown`
/// - worker proven alive → `Alive`
pub fn decide_liveness(_recorded: &ProcessIdentity, worker: WorkerLiveness) -> Liveness {
    match worker {
        WorkerLiveness::Alive => Liveness::Alive,
        WorkerLiveness::Dead => Liveness::Stale,
        WorkerLiveness::Unknown => Liveness::Unknown,
    }
}

/// 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, worker: WorkerLiveness, now: u64) -> bool {
    match decide_liveness(&lock.identity, worker) {
        Liveness::Alive | Liveness::Unknown => true,
        Liveness::Stale => 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 {
    lock_is_held_at(lock, probe_worker_identity(&lock.identity), unix_now())
}

/// 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_worker_identity(&lock.identity), now)
            }
            // A malformed lock is an Unknown state, not proof that its owner is
            // dead. It is held indefinitely after the settling window and must
            // be removed only by an operator who has inspected the state dir.
            // The same fail-closed policy applies to a valid lock whose identity
            // probe remains Unknown: callers keep observing HeldByLiveWatcher
            // until the owner is proven dead or an operator clears the lease.
            LockRead::Corrupt { .. } => true,
        }
    }

    fn is_held_with_unknown_identity(&self) -> bool {
        let LockRead::Valid(lock) = self else {
            return false;
        };
        matches!(
            probe_worker_identity(&lock.identity),
            WorkerLiveness::Unknown
        )
    }
}

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) {
        if state.is_held_with_unknown_identity() {
            tracing::warn!(
                state_dir = %state_dir.display(),
                "watcher lock probe returned unknown identity; leaving lock in place"
            );
        }
        return Ok(LockClaim::HeldByLiveWatcher);
    }

    let LockRead::Valid(stale_lock) = &state else {
        return Ok(LockClaim::HeldByLiveWatcher);
    };
    let stale_token = stale_lock.acquisition_token.clone();

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

    let state = read_lock_state(state_dir)?;
    let LockRead::Valid(current_lock) = &state else {
        return Ok(LockClaim::HeldByLiveWatcher);
    };
    if current_lock.acquisition_token != stale_token {
        return Ok(LockClaim::HeldByLiveWatcher);
    }
    if state.is_held_at(unix_now()) {
        if state.is_held_with_unknown_identity() {
            tracing::warn!(
                state_dir = %state_dir.display(),
                "watcher lock probe returned unknown identity during reclaim; leaving lock in place"
            );
        }
        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 when a single ownership snapshot matches.
///
/// 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 and token match the current process. The snapshot and unlink are not
/// one atomic filesystem operation; callers must not treat this helper as proof
/// against an uninstrumented replacement interleaving.
pub fn release_lock_if_owned(
    state_dir: &Path,
    expected_acquisition_token: &str,
) -> Result<(), WatcherError> {
    let Some(lock) = read_lock(state_dir)? else {
        return Ok(());
    };
    if expected_acquisition_token.is_empty()
        || lock.acquisition_token != expected_acquisition_token
        || !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(())
}

/// Return the token for the currently recorded watcher lease.
pub fn lock_acquisition_token(state_dir: &Path) -> Option<String> {
    read_lock(state_dir)
        .ok()
        .flatten()
        .map(|lock| lock.acquisition_token)
}

fn identity_matches_owner(recorded: &ProcessIdentity, current: &ProcessIdentity) -> bool {
    recorded.matches(current)
}

/// Whether the recorded lock owner is THIS process.
///
/// `ensure-watcher` re-points the lock at the detached child's identity right
/// after spawning it, so a fresh `watch --until-empty` finds its own pid
/// already holding the slot — that is an ownership handoff, not contention.
/// A looping `watch` treats "held by a live watcher" as fatal only when the
/// holder is a DIFFERENT process.
pub fn lock_owned_by_current_process(state_dir: &Path) -> bool {
    match read_lock(state_dir) {
        Ok(Some(lock)) => identity_matches_owner(&lock.identity, &ProcessIdentity::current()),
        _ => false,
    }
}

/// 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> {
    // Reap orphaned `running` review rows left by a killed watcher BEFORE the
    // spawn decision: a dead worker's rows must never masquerade as live
    // work, and 0.9.2 left them `running` forever with no self-healing path.
    // Best-effort — a reconciliation hiccup must not block spawning the
    // watcher that would drain the queue.
    match crate::reviewer::ReviewRunStore::new(state_dir).reconcile_stale_runs() {
        Ok(reaped) if reaped > 0 => {
            tracing::info!(
                reaped,
                "ensure-watcher: reaped stale review run(s) from a dead worker"
            );
        }
        Ok(_) => {}
        Err(error) => {
            tracing::warn!(%error, "ensure-watcher: failed to reconcile stale review runs");
        }
    }

    // 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 acquisition_token = match read_lock(state_dir)? {
                Some(lock) => lock.acquisition_token,
                None => {
                    return Err(WatcherError::Io(io::Error::other(
                        "watcher lock disappeared after acquisition",
                    )));
                }
            };
            let child_pid = match spawn(state_dir) {
                Ok(pid) => pid,
                Err(error) => {
                    if let Err(release_error) = release_lock_if_owned(state_dir, &acquisition_token)
                    {
                        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,
                },
                acquisition_token,
                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")
        // The child must absorb the lock handoff window: this parent re-points
        // the lock at the child right after spawn returns, and a child that
        // checked before the re-point landed would otherwise refuse its own
        // slot and strand the queue.
        .arg("--expect-lock-handoff")
        .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) {}

/// Tri-state answer to "is this pid alive?".
///
/// Reconciliation decisions MUST distinguish a proven-dead process from an
/// unanswerable probe: treating a probe failure as death fails live runs —
/// 0.9.2 marked healthy reviews failed whenever the probe hiccuped.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum PidLiveness {
    /// The process exists (or exists but is owned by someone else — EPERM).
    Alive,
    /// The OS confirmed no such process exists.
    Dead,
    /// The probe itself failed (spawn error, unparsable result) — NOT proof
    /// of death. Callers must not reap on this answer.
    Unknown,
}

/// Probe whether `pid` is alive, distinguishing a proven-dead process from a
/// failed probe. Prefer this over [`pid_is_alive`] wherever a wrong "dead"
/// answer would destroy state.
#[cfg(unix)]
pub fn probe_pid_liveness(pid: u32) -> PidLiveness {
    let Some(pid) = i32::try_from(pid)
        .ok()
        .and_then(rustix::process::Pid::from_raw)
    else {
        return PidLiveness::Unknown;
    };
    classify_pid_probe(rustix::process::test_kill_process(pid))
}

#[cfg(unix)]
fn classify_pid_probe(result: Result<(), rustix::io::Errno>) -> PidLiveness {
    match result {
        Ok(()) => PidLiveness::Alive,
        Err(error) if error == rustix::io::Errno::SRCH => PidLiveness::Dead,
        // EPERM means the process exists but is owned by someone else.
        Err(error) if error == rustix::io::Errno::PERM => PidLiveness::Alive,
        Err(_) => PidLiveness::Unknown,
    }
}

/// Probe whether `pid` is alive, distinguishing a proven-dead process from a
/// failed probe.
#[cfg(windows)]
pub fn probe_pid_liveness(pid: u32) -> PidLiveness {
    // `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);
            if text.contains(&pid.to_string()) {
                PidLiveness::Alive
            } else {
                PidLiveness::Dead
            }
        }
        _ => PidLiveness::Unknown,
    }
}

#[cfg(not(any(unix, windows)))]
pub fn probe_pid_liveness(_pid: u32) -> PidLiveness {
    PidLiveness::Unknown
}

/// Tri-state liveness for a recorded worker identity (pid + start token).
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum WorkerLiveness {
    /// pid alive AND start token matches (or no token was recorded to compare).
    Alive,
    /// pid dead, OR pid alive under a DIFFERENT start token (pid reuse).
    Dead,
    /// The probe could not answer; NOT proof of death — the row must be left
    /// alone and the inconclusive probe logged.
    Unknown,
}

/// Pure worker-liveness decision over injected probe results, unit-testable
/// without real processes.
///
/// - pid proven dead → `Dead`
/// - pid probe inconclusive → `Unknown`
/// - pid alive, no recorded token (legacy row) → `Alive` (bare-pid fallback)
/// - pid alive, tokens differ → `Dead` (the pid was reused by another process)
/// - pid alive, recorded token but token unprobeable → `Unknown` (never guess
///   death when the identity check itself failed)
pub fn decide_worker_liveness(
    pid: PidLiveness,
    recorded_token: &str,
    probed_token: Option<String>,
) -> WorkerLiveness {
    match pid {
        PidLiveness::Dead => WorkerLiveness::Dead,
        PidLiveness::Unknown => WorkerLiveness::Unknown,
        PidLiveness::Alive => {
            if recorded_token.is_empty() {
                return WorkerLiveness::Alive;
            }
            match probed_token {
                Some(token) if token == recorded_token => WorkerLiveness::Alive,
                Some(_) => WorkerLiveness::Dead,
                None => WorkerLiveness::Unknown,
            }
        }
    }
}

/// Probe a recorded worker identity against the live system: pid liveness
/// plus the pid-reuse-defeating start-token comparison.
pub fn probe_worker_identity(recorded: &ProcessIdentity) -> WorkerLiveness {
    let pid = probe_pid_liveness(recorded.pid);
    let probed_token = if pid == PidLiveness::Alive && !recorded.start_token.is_empty() {
        probe_start_token(recorded.pid)
    } else {
        None
    };
    decide_worker_liveness(pid, &recorded.start_token, probed_token)
}

/// Probe whether `pid` is alive.
#[cfg(unix)]
pub fn pid_is_alive(pid: u32) -> bool {
    probe_pid_liveness(pid) == PidLiveness::Alive
}

#[cfg(windows)]
pub fn pid_is_alive(pid: u32) -> bool {
    probe_pid_liveness(pid) == PidLiveness::Alive
}

#[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)]
pub(crate) 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. Pin the locale so the identity token is stable across the
    // caller's environment. Empty/failed output degrades to bare pid liveness.
    let output = std::process::Command::new("ps")
        .args(["-o", "lstart=", "-p", &pid.to_string()])
        .env("LC_ALL", "C")
        .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)]
pub(crate) 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)))]
pub(crate) 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(),
        }
    }

    /// A pid guaranteed to be dead: spawn a trivial process, reap it, return
    /// its pid. Copilot (PR #13, watcher.rs ~844): spawning the external
    /// `true` binary isn't portable (notably absent on Windows). Spawning
    /// this test binary itself with `--help` (which the standard libtest
    /// harness understands and exits 0 for) is a guaranteed-dead-pid source
    /// without any external command dependency.
    fn reaped_pid() -> u32 {
        let mut child = std::process::Command::new(std::env::current_exe().unwrap())
            .arg("--help")
            .stdout(std::process::Stdio::null())
            .stderr(std::process::Stdio::null())
            .spawn()
            .unwrap();
        let pid = child.id();
        child.wait().unwrap();
        pid
    }

    #[test]
    fn worker_liveness_dead_pid_is_dead() {
        assert_eq!(
            decide_worker_liveness(PidLiveness::Dead, "token", None),
            WorkerLiveness::Dead
        );
    }

    #[cfg(unix)]
    #[test]
    fn pid_probe_classifies_os_errno_without_localized_output() {
        assert_eq!(
            classify_pid_probe(Err(rustix::io::Errno::SRCH)),
            PidLiveness::Dead
        );
        assert_eq!(
            classify_pid_probe(Err(rustix::io::Errno::PERM)),
            PidLiveness::Alive
        );
        assert_eq!(
            classify_pid_probe(Err(rustix::io::Errno::INVAL)),
            PidLiveness::Unknown
        );
    }

    #[test]
    fn worker_liveness_unknown_pid_probe_is_unknown() {
        // A failed pid probe is never proof of death.
        assert_eq!(
            decide_worker_liveness(PidLiveness::Unknown, "token", None),
            WorkerLiveness::Unknown
        );
    }

    #[test]
    fn worker_liveness_alive_without_recorded_token_degrades_to_pid() {
        // Legacy row: no start identity recorded, so bare pid liveness is the
        // best available signal.
        assert_eq!(
            decide_worker_liveness(PidLiveness::Alive, "", None),
            WorkerLiveness::Alive
        );
    }

    #[test]
    fn worker_liveness_matching_token_is_alive() {
        assert_eq!(
            decide_worker_liveness(
                PidLiveness::Alive,
                "Mon Jan 1 00:00:00 2026",
                Some("Mon Jan 1 00:00:00 2026".to_owned())
            ),
            WorkerLiveness::Alive
        );
    }

    #[test]
    fn worker_liveness_mismatched_token_is_pid_reuse_death() {
        assert_eq!(
            decide_worker_liveness(
                PidLiveness::Alive,
                "Mon Jan 1 00:00:00 2026",
                Some("Tue Feb 2 11:11:11 2026".to_owned())
            ),
            WorkerLiveness::Dead
        );
    }

    #[test]
    fn worker_liveness_unprobeable_token_is_unknown_not_death() {
        // The pid is alive but its start marker could not be read: never
        // guess death when the identity check itself failed.
        assert_eq!(
            decide_worker_liveness(PidLiveness::Alive, "Mon Jan 1 00:00:00 2026", None),
            WorkerLiveness::Unknown
        );
    }

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

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

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

    #[test]
    fn empty_start_token_falls_back_to_pid_liveness() {
        let recorded = identity(4242, "");
        assert_eq!(
            decide_liveness(&recorded, WorkerLiveness::Alive),
            Liveness::Alive
        );
    }

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

    #[test]
    fn unknown_probe_is_not_stale_even_after_settling_window() {
        let recorded = identity(4242, "recorded-token");
        assert_eq!(
            decide_liveness(&recorded, WorkerLiveness::Unknown),
            Liveness::Unknown
        );
    }

    #[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(reaped_pid(), "definitely-not-a-live-token"),
            acquisition_token: "stale-token".to_owned(),
            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 unknown_pid_probe_is_held_after_settling_instead_of_reclaimed() {
        let temp = tempfile::tempdir().unwrap();
        let state = temp.path();

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

        assert_eq!(
            try_acquire_lock(state).unwrap(),
            LockClaim::HeldByLiveWatcher
        );
        assert_eq!(live_watcher_pid(state), Some(4_000_000_000));
    }

    #[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(reaped_pid(), "definitely-not-a-live-token"),
            acquisition_token: "stale-token".to_owned(),
            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"),
            acquisition_token: "other-token".to_owned(),
            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, "not-our-token").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);
        let token = lock_acquisition_token(state).unwrap();
        release_lock_if_owned(state, &token).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(), ""),
            acquisition_token: "current-token".to_owned(),
            created_at_unix: 1,
        };
        write_lock(state, &current).unwrap();

        release_lock_if_owned(state, "current-token").unwrap();
        assert!(read_lock(state).unwrap().is_none());
    }

    #[test]
    fn release_does_not_remove_same_process_replacement_with_new_token() {
        let temp = tempfile::tempdir().unwrap();
        let state = temp.path();
        let owner = WatcherLock::for_current();
        write_lock(state, &owner).unwrap();

        let replacement = WatcherLock {
            identity: owner.identity.clone(),
            acquisition_token: "replacement-token".to_owned(),
            created_at_unix: unix_now(),
        };
        write_lock(state, &replacement).unwrap();

        release_lock_if_owned(state, &owner.acquisition_token).unwrap();
        assert_eq!(read_lock(state).unwrap(), Some(replacement));
    }

    #[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 ensure_watcher_reconciles_orphaned_running_rows_before_deciding() {
        // 0.9.2 field failure: after a killed watcher left `running` rows
        // behind, ensure-watcher silently no-opped. It must liveness-check
        // recorded pids and reap dead rows as part of check-and-spawn.
        let temp = tempfile::tempdir().unwrap();
        let state = temp.path();

        // Plant an orphaned running row whose recorded worker pid is provably
        // dead (a real reaped child — a made-up pid number probes as
        // UNPROVABLE on some platforms and would correctly be left alone).
        let dead_pid = reaped_pid();
        let runs_dir = state.join(crate::reviewer::REVIEW_RUNS_DIR);
        fs::create_dir_all(&runs_dir).unwrap();
        fs::write(
            runs_dir.join("orphan.json"),
            serde_json::json!({
                "id": "orphan",
                "commit_sha": "abc123",
                "target": "commit",
                "status": "running",
                "phase": "reviewing",
                "ledger_entries": 0,
                "error": null,
                "worker_pid": dead_pid,
                "created_at_unix": 1,
                "updated_at_unix": 1,
                "started_at_unix": 1,
                "completed_at_unix": null
            })
            .to_string(),
        )
        .unwrap();

        let spawned = AtomicUsize::new(0);
        let claim = ensure_watcher_with_spawner(state, |_| {
            spawned.fetch_add(1, Ordering::SeqCst);
            Ok(4242)
        })
        .unwrap();

        // The orphan was reaped (marked failed, persisted) AND, with no live
        // watcher, a spawn happened.
        assert_eq!(claim, LockClaim::Acquired);
        assert_eq!(spawned.load(Ordering::SeqCst), 1);
        let reaped: serde_json::Value =
            serde_json::from_str(&fs::read_to_string(runs_dir.join("orphan.json")).unwrap())
                .unwrap();
        assert_eq!(reaped["status"], "failed");
        assert!(
            reaped["error"].as_str().unwrap().contains("stale run"),
            "stale reason should be recorded, got: {reaped}"
        );
    }

    #[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 lock_owned_by_current_process_true_for_own_identity() {
        let temp = tempfile::tempdir().unwrap();
        let state = temp.path();
        write_lock(state, &WatcherLock::for_current()).unwrap();

        assert!(lock_owned_by_current_process(state));
    }

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

        // No lock at all.
        assert!(!lock_owned_by_current_process(state));

        // A lock owned by a different (here: dead) identity.
        let foreign = WatcherLock {
            identity: identity(4_000_000_000, "not-us"),
            acquisition_token: "foreign-token".to_owned(),
            created_at_unix: 1,
        };
        write_lock(state, &foreign).unwrap();
        assert!(!lock_owned_by_current_process(state));
    }

    #[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"),
            acquisition_token: "dead-token".to_owned(),
            created_at_unix: 1_000,
        };
        // now == created (elapsed 0) → within settling → held.
        assert!(lock_is_held_at(&lock, WorkerLiveness::Unknown, 1_000));
        // Just before the settling boundary → still held.
        assert!(lock_is_held_at(
            &lock,
            WorkerLiveness::Unknown,
            1_000 + LOCK_SETTLING_SECS - 1
        ));
    }

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

    #[test]
    fn unknown_probe_identity_is_held_even_after_settling_window() {
        let own_pid = std::process::id();
        let lock = WatcherLock {
            identity: identity(own_pid, "unknown-token"),
            acquisition_token: "unknown-token".to_owned(),
            created_at_unix: 1_000,
        };
        // A probe that confirms liveness but cannot recover a start marker
        // (empty token) must be treated as uncertain/held, never reclaimed.
        assert!(lock_is_held_at(
            &lock,
            WorkerLiveness::Unknown,
            1_000 + LOCK_SETTLING_SECS
        ));
    }

    #[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"),
            acquisition_token: "live-token".to_owned(),
            created_at_unix: 1,
        };
        assert!(lock_is_held_at(&lock, WorkerLiveness::Alive, 9_999_999));
    }
}