rightkit_daemon/

discovery.rs

//! Installed app discovery and single-winner spawn coordination.
//!
//! Discovery is deliberately installation-bound: it scans known application
//! bundles (macOS) or uninstall registry entries (Windows), never `PATH` or a
//! checkout. Spawn claims use `create_new`; discovery is daemon-owned and is
//! never removed by a losing launcher.

use crate::record::{
    read_live_record, try_read_live_record, try_read_record, write_takeover, ServiceRecord,
    TakeoverRequest, RANK_UNKNOWN,
};
use crate::suite::{make_private_dir, Suite};
use crate::transport::{pid_alive, probe_endpoint};
use serde::Serialize;
use std::collections::BTreeMap;
use std::io;
use std::path::{Path, PathBuf};
use std::process::{Command, Stdio};
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};

/// Installed Right Suite applications and their engine executable.
#[derive(Debug, Clone, Serialize, PartialEq, Eq)]
pub struct InstalledApp {
    pub app: String,
    pub app_path: String,
    pub engine_path: String,
    pub version: Option<String>,
    pub source: String,
    /// Whether the engine executable was found. Always [`Completeness::Complete`] unless
    /// [`DiscoverOptions::include_incomplete`] was set.
    pub completeness: Completeness,
}

/// Whether an installed app is usable as a service host.
#[derive(Debug, Clone, Copy, Serialize, PartialEq, Eq, Default)]
#[serde(rename_all = "snake_case")]
pub enum Completeness {
    /// The app bundle and its engine executable are both present.
    #[default]
    Complete,
    /// The app is installed but its engine executable is missing (a broken or partial
    /// install). `engine_path` names where the engine was expected.
    EngineMissing,
}

/// Options for discovery. `DiscoverOptions::default()` reproduces the original behaviour.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct DiscoverOptions {
    /// Also return an app whose bundle is installed but whose engine is missing, marked
    /// [`Completeness::EngineMissing`]. A complete install in any root always wins over an
    /// incomplete one. Default `false`. Honoured on macOS ([`discover_macos_with`]) and on
    /// Windows ([`discover_windows_with`], where an entry whose install directory exists
    /// but whose engine file is missing is reported).
    pub include_incomplete: bool,
}

/// The original Right Suite table. Deprecated: the crate hard-coded one product
/// suite's names. Products pass their own `(app name, engine executable)` registry to
/// [`discover_apps_in`]. Kept with identical contents so existing callers behave the same
/// in this release.
#[deprecated(
    since = "0.1.2",
    note = "pass your own (app, engine) registry to discover_apps_in"
)]
pub const KNOWN_APPS: &[(&str, &str)] = SUITE_APPS;

/// Built-in table behind the deprecated [`KNOWN_APPS`] and [`discover_apps`].
const SUITE_APPS: &[(&str, &str)] = &[
    ("HeardRight", "heardright-engine"),
    ("ScrapeRight", "scraperight-engine"),
    ("GenRight", "genright-engine"),
    ("CutRight", "cutright-engine"),
];

/// Discover installed apps from the caller's `(app bundle or display name, engine executable)`
/// registry. The first complete match per app wins.
pub fn discover_apps_in(registry: &[(&str, &str)]) -> Vec<InstalledApp> {
    discover_apps_in_with(registry, &DiscoverOptions::default())
}

/// [`discover_apps_in`] with explicit [`DiscoverOptions`].
pub fn discover_apps_in_with(
    registry: &[(&str, &str)],
    options: &DiscoverOptions,
) -> Vec<InstalledApp> {
    #[cfg(target_os = "macos")]
    {
        discover_macos_with(&default_macos_roots(), registry, options)
    }
    #[cfg(windows)]
    {
        discover_windows_with(
            &read_windows_uninstall_entries(),
            registry,
            |p| Path::new(p).is_file(),
            |p| Path::new(p).is_dir(),
            options,
        )
    }
    #[cfg(not(any(target_os = "macos", windows)))]
    {
        let _ = (registry, options);
        Vec::new()
    }
}

#[cfg(target_os = "macos")]
fn default_macos_roots() -> Vec<PathBuf> {
    let mut roots = vec![PathBuf::from("/Applications")];
    // The user's own Applications folder comes from the account database (`getpwuid_r`), never
    // from `HOME`: whoever launches the process controls `HOME`, and this root decides which
    // installed apps are trusted as siblings.
    if let Ok(home) = crate::suite::passwd_home_dir() {
        roots.push(home.join("Applications"));
    }
    roots
}

/// Discover the built-in Right Suite apps. Deprecated: use [`discover_apps_in`].
#[deprecated(since = "0.1.2", note = "use discover_apps_in with your own registry")]
pub fn discover_apps() -> Vec<InstalledApp> {
    #[allow(deprecated)]
    discover_apps_in(KNOWN_APPS)
}

/// Scan roots in order; first complete bundle wins per app.
pub fn discover_macos(roots: &[PathBuf], apps: &[(&str, &str)]) -> Vec<InstalledApp> {
    discover_macos_with(roots, apps, &DiscoverOptions::default())
}

/// [`discover_macos`] with explicit [`DiscoverOptions`]. With `include_incomplete`, an app
/// that has a bundle directory but no engine in any root is returned once, from the first
/// root that has its bundle, as [`Completeness::EngineMissing`].
pub fn discover_macos_with(
    roots: &[PathBuf],
    apps: &[(&str, &str)],
    options: &DiscoverOptions,
) -> Vec<InstalledApp> {
    let mut found = Vec::new();
    for (name, engine) in apps {
        let mut incomplete: Option<InstalledApp> = None;
        let mut complete: Option<InstalledApp> = None;
        for root in roots {
            let bundle = root.join(format!("{name}.app"));
            let binary = bundle.join("Contents/MacOS").join(engine);
            let present = binary.is_file();
            if !present && (!options.include_incomplete || incomplete.is_some() || !bundle.is_dir())
            {
                continue;
            }
            let app = InstalledApp {
                app: (*name).to_string(),
                app_path: bundle.to_string_lossy().into_owned(),
                engine_path: binary.to_string_lossy().into_owned(),
                version: read_plist_version(&bundle.join("Contents/Info.plist")),
                source: "macos-bundle".into(),
                completeness: if present {
                    Completeness::Complete
                } else {
                    Completeness::EngineMissing
                },
            };
            if present {
                complete = Some(app);
                break;
            }
            incomplete = Some(app);
        }
        found.extend(complete.or(incomplete));
    }
    found
}

fn read_plist_version(path: &Path) -> Option<String> {
    parse_plist_string(
        &std::fs::read_to_string(path).ok()?,
        "CFBundleShortVersionString",
    )
}

/// Parse a plain XML plist without requiring a plist dependency. Binary plist
/// files return `None`, while the app remains discoverable.
pub fn parse_plist_string(xml: &str, key: &str) -> Option<String> {
    let marker = format!("<key>{key}</key>");
    let after = &xml[xml.find(&marker)? + marker.len()..];
    let string_start = after.find("<string>")? + "<string>".len();
    if after[..string_start].contains("<key>") {
        return None;
    }
    let end = after[string_start..].find("</string>")? + string_start;
    let value = after[string_start..end].trim();
    (!value.is_empty()).then(|| value.to_string())
}

/// One uninstall registry row, independent of the Windows registry API so it
/// can be tested from captured `reg query` output.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct UninstallEntry {
    pub display_name: String,
    pub install_location: Option<String>,
    pub display_icon: Option<String>,
    pub uninstall_string: Option<String>,
    pub display_version: Option<String>,
}

/// Parse `reg query ... /s` output. Unknown values and malformed rows are
/// ignored; keys are grouped by their registry-key heading.
pub fn parse_reg_query(text: &str) -> Vec<UninstallEntry> {
    let mut out = Vec::new();
    let mut current = UninstallEntry::default();
    let mut active = false;
    for raw in text.lines() {
        let line = raw.trim_end();
        let trimmed = line.trim();
        if trimmed.starts_with("HKEY_") {
            if active && !current.display_name.is_empty() {
                out.push(current);
            }
            current = UninstallEntry::default();
            active = true;
            continue;
        }
        let mut fields = trimmed.split_whitespace();
        let Some(name) = fields.next() else { continue };
        let Some(_kind) = fields.next() else { continue };
        let value = fields.collect::<Vec<_>>().join(" ");
        if value.is_empty() {
            continue;
        }
        match name {
            "DisplayName" => current.display_name = value,
            "InstallLocation" => current.install_location = Some(value),
            "DisplayIcon" => current.display_icon = Some(value),
            "UninstallString" => current.uninstall_string = Some(value),
            "DisplayVersion" => current.display_version = Some(value),
            _ => {}
        }
    }
    if active && !current.display_name.is_empty() {
        out.push(current);
    }
    out
}

fn strip_quotes(value: &str) -> &str {
    value.trim().trim_matches('"').trim()
}

fn parent(path: &str) -> Option<String> {
    let i = path.rfind(['\\', '/'])?;
    (!path[..i].is_empty()).then(|| path[..i].to_string())
}

fn executable_from_value(value: &str) -> Option<String> {
    let value = value.trim();
    let path = if let Some(rest) = value.strip_prefix('"') {
        rest.split('"').next()?.to_string()
    } else {
        let lower = value.to_ascii_lowercase();
        let end = lower.find(".exe").map(|i| i + 4).unwrap_or(value.len());
        value[..end].to_string()
    };
    (!path.is_empty()).then_some(path)
}

pub fn resolve_install_dir(entry: &UninstallEntry) -> Option<String> {
    if let Some(loc) = entry.install_location.as_deref() {
        let loc = strip_quotes(loc).trim_end_matches(['\\', '/']);
        if !loc.is_empty() {
            return Some(loc.to_string());
        }
    }
    entry
        .display_icon
        .as_deref()
        .and_then(executable_from_value)
        .and_then(|p| parent(&p))
        .or_else(|| {
            entry
                .uninstall_string
                .as_deref()
                .and_then(executable_from_value)
                .and_then(|p| parent(&p))
        })
}

/// Discover complete Windows installs only. Equivalent to [`discover_windows_with`] with
/// [`DiscoverOptions::default`].
pub fn discover_windows(
    entries: &[UninstallEntry],
    apps: &[(&str, &str)],
    exists: impl Fn(&str) -> bool,
) -> Vec<InstalledApp> {
    discover_windows_with(
        entries,
        apps,
        exists,
        |_| false,
        &DiscoverOptions::default(),
    )
}

/// Windows discovery with explicit options. `exists` checks the engine file and `dir_exists`
/// checks the install directory. With `include_incomplete`, an entry whose directory exists but
/// whose engine is missing is returned as [`Completeness::EngineMissing`]. A complete install
/// from any entry of the same app always wins.
pub fn discover_windows_with(
    entries: &[UninstallEntry],
    apps: &[(&str, &str)],
    exists: impl Fn(&str) -> bool,
    dir_exists: impl Fn(&str) -> bool,
    options: &DiscoverOptions,
) -> Vec<InstalledApp> {
    let mut found = Vec::new();
    for (name, engine) in apps {
        let mut incomplete: Option<InstalledApp> = None;
        let mut complete: Option<InstalledApp> = None;
        for entry in entries.iter().filter(|entry| entry.display_name == *name) {
            let Some(dir) = resolve_install_dir(entry) else {
                continue;
            };
            let engine_path = format!(r"{dir}\{engine}.exe");
            let present = exists(&engine_path);
            if !present
                && (!options.include_incomplete || incomplete.is_some() || !dir_exists(&dir))
            {
                continue;
            }
            let app = InstalledApp {
                app: (*name).to_string(),
                app_path: dir,
                engine_path,
                version: entry
                    .display_version
                    .clone()
                    .filter(|v| !v.trim().is_empty()),
                source: "windows-uninstall".into(),
                completeness: if present {
                    Completeness::Complete
                } else {
                    Completeness::EngineMissing
                },
            };
            if present {
                complete = Some(app);
                break;
            }
            incomplete = Some(app);
        }
        if let Some(app) = complete.or(incomplete) {
            found.push(app);
        }
    }
    found
}

#[derive(Debug)]
pub enum SpawnClaimError {
    AlreadyHeld(PathBuf),
    Io(io::Error),
}
impl std::fmt::Display for SpawnClaimError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::AlreadyHeld(path) => write!(f, "spawn claim is held: {}", path.display()),
            Self::Io(e) => write!(f, "spawn claim: {e}"),
        }
    }
}
impl std::error::Error for SpawnClaimError {}
impl From<io::Error> for SpawnClaimError {
    fn from(e: io::Error) -> Self {
        Self::Io(e)
    }
}

/// The claim around one process spawn attempt, backed by an OS exclusive lock
/// on `spawn.claim`. The lock lives as long as the open file: the kernel
/// releases it when the owner dies (any signal), and there is no time-based
/// eviction of a live owner. The file is never unlinked, so every contender
/// always locks the same inode; Drop closes only this claim's own handle.
pub struct SpawnClaim {
    _file: std::fs::File,
    path: PathBuf,
}

impl SpawnClaim {
    pub fn path(suite: &Suite, service: &str) -> PathBuf {
        suite.service_dir(service).join("spawn.claim")
    }

    pub fn acquire(suite: &Suite, service: &str) -> Result<Self, SpawnClaimError> {
        make_private_dir(&suite.service_dir(service))
            .map_err(|e| SpawnClaimError::Io(io::Error::other(e.to_string())))?;
        let path = Self::path(suite, service);
        match try_lock_exclusive(&path) {
            Ok(Some(file)) => Ok(Self { _file: file, path }),
            Ok(None) => Err(SpawnClaimError::AlreadyHeld(path)),
            Err(e) => Err(SpawnClaimError::Io(e)),
        }
    }

    /// Path of the claim file (for diagnostics).
    pub fn claim_path(&self) -> &Path {
        &self.path
    }
}

/// Open `path` and take a non-blocking exclusive lock. `Ok(None)` = held elsewhere.
#[cfg(unix)]
pub(crate) fn try_lock_exclusive(path: &Path) -> io::Result<Option<std::fs::File>> {
    use std::fs::TryLockError;
    use std::io::Write;
    let mut f = std::fs::OpenOptions::new()
        .read(true)
        .write(true)
        .create(true)
        .truncate(false)
        .open(path)?;
    // std's `File::try_lock` is a non-blocking `flock(LOCK_EX | LOCK_NB)` on Unix.
    match f.try_lock() {
        Ok(()) => {}
        Err(TryLockError::WouldBlock) => return Ok(None),
        Err(TryLockError::Error(e)) => return Err(e),
    }
    // Informational only (we hold the lock, so the write is ours).
    let _ = f.set_len(0);
    let _ = f.write_all(std::process::id().to_string().as_bytes());
    Ok(Some(f))
}

/// Windows: open with share mode 0 (no sharing). A second open fails with
/// ERROR_SHARING_VIOLATION (32) while the first handle lives; the kernel closes
/// the handle on process death. Not compiled on this host; std-only API
/// (`OpenOptionsExt::share_mode`), checked by reasoning against its docs.
#[cfg(windows)]
pub(crate) fn try_lock_exclusive(path: &Path) -> io::Result<Option<std::fs::File>> {
    use std::io::Write;
    use std::os::windows::fs::OpenOptionsExt;
    match std::fs::OpenOptions::new()
        .write(true)
        .create(true)
        .truncate(false)
        .share_mode(0)
        .open(path)
    {
        Ok(mut f) => {
            let _ = f.write_all(std::process::id().to_string().as_bytes());
            Ok(Some(f))
        }
        Err(e) if e.raw_os_error() == Some(32) => Ok(None),
        Err(e) => Err(e),
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub enum AdoptOrSpawn {
    Adopt(ServiceRecord),
    Spawn,
}

/// Adopt only a live, reachable record whose PID generation still matches.
///
/// A `service.json` that cannot be read is `Spawn` here, as an absent one is; the enum cannot say
/// "unknown". Use [`try_adopt_or_spawn`] where that must not start a second host.
pub fn adopt_or_spawn(suite: &Suite, service: &str) -> AdoptOrSpawn {
    try_adopt_or_spawn(suite, service).unwrap_or(AdoptOrSpawn::Spawn)
}

/// [`adopt_or_spawn`] with `Err` when `service.json` exists but could not be read: whether a
/// live host exists is then unknown, so look again rather than spawn.
pub fn try_adopt_or_spawn(suite: &Suite, service: &str) -> io::Result<AdoptOrSpawn> {
    Ok(match try_read_record(suite, service)? {
        Some(record) if record_process_matches(&record) && probe_endpoint(&record.endpoint) => {
            AdoptOrSpawn::Adopt(record)
        }
        _ => AdoptOrSpawn::Spawn,
    })
}

/// Wait for the winner of a competing claim to publish a matching live record.
pub fn wait_for_claim_winner(
    suite: &Suite,
    service: &str,
    timeout: Duration,
) -> Option<ServiceRecord> {
    let deadline = Instant::now() + timeout;
    while Instant::now() < deadline {
        // An unreadable record is "not yet": keep waiting, never stop early on it.
        if let Ok(AdoptOrSpawn::Adopt(record)) = try_adopt_or_spawn(suite, service) {
            return Some(record);
        }
        std::thread::sleep(
            Duration::from_millis(25).min(deadline.saturating_duration_since(Instant::now())),
        );
    }
    None
}

/// Check both liveness and process generation, protecting discovery from PID reuse.
pub fn record_process_matches(record: &ServiceRecord) -> bool {
    record_process_matches_inner(record)
}

fn record_process_matches_inner(record: &ServiceRecord) -> bool {
    if !pid_alive(record.pid) {
        return false;
    }
    // Legacy / wave-1 records carry no process birth time: PID liveness is all
    // we can know. `started_ms` is publication time and never validates a PID.
    if record.process_started_ms == 0 {
        return true;
    }
    match process_start_ms(record.pid) {
        Some(started) => started.abs_diff(record.process_started_ms as u128) <= 2_000,
        // Cannot read the birth time right now: the pid is alive, keep the record.
        None => true,
    }
}

/// Start time (ms since epoch) of this process, as `process_start_ms` reports it.
pub(crate) fn own_start_ms() -> Option<u64> {
    process_start_ms(std::process::id()).map(|v| v as u64)
}

fn process_start_ms(pid: u32) -> Option<u128> {
    #[cfg(target_os = "linux")]
    {
        let stat = std::fs::read_to_string(if pid == std::process::id() {
            "/proc/self/stat".into()
        } else {
            format!("/proc/{pid}/stat")
        })
        .ok()?;
        let ticks = stat
            .rsplit(')')
            .next()?
            .split_whitespace()
            .nth(19)?
            .parse::<u128>()
            .ok()?;
        let boot = std::fs::read_to_string("/proc/stat")
            .ok()?
            .lines()
            .find_map(|l| l.strip_prefix("btime "))?
            .parse::<u128>()
            .ok()?;
        return Some(boot * 1000 + ticks * 1000 / 100);
    }
    #[cfg(target_os = "macos")]
    {
        let out = std::process::Command::new("ps")
            .args(["-o", "lstart=", "-p", &pid.to_string()])
            .output()
            .ok()?;
        let text = String::from_utf8(out.stdout).ok()?;
        let value = text.trim();
        let parsed = std::process::Command::new("date")
            .args(["-j", "-f", "%a %b %e %H:%M:%S %Y", value, "+%s"])
            .output()
            .ok()?;
        let seconds = String::from_utf8(parsed.stdout)
            .ok()?
            .trim()
            .parse::<u128>()
            .ok()?;
        return Some(seconds * 1000);
    }
    #[cfg(windows)]
    {
        use std::ffi::c_void;
        #[repr(C)]
        struct FileTime {
            low: u32,
            high: u32,
        }
        #[link(name = "kernel32")]
        unsafe extern "system" {
            fn OpenProcess(access: u32, inherit: i32, pid: u32) -> *mut c_void;
            fn GetProcessTimes(
                handle: *mut c_void,
                creation: *mut FileTime,
                exit: *mut FileTime,
                kernel: *mut FileTime,
                user: *mut FileTime,
            ) -> i32;
            fn CloseHandle(handle: *mut c_void) -> i32;
        }
        const PROCESS_QUERY_LIMITED_INFORMATION: u32 = 0x1000;
        let handle = unsafe { OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION, 0, pid) };
        if handle.is_null() {
            return None;
        }
        let mut creation = FileTime { low: 0, high: 0 };
        let mut exit = FileTime { low: 0, high: 0 };
        let mut kernel = FileTime { low: 0, high: 0 };
        let mut user = FileTime { low: 0, high: 0 };
        let ok =
            unsafe { GetProcessTimes(handle, &mut creation, &mut exit, &mut kernel, &mut user) }
                != 0;
        unsafe {
            CloseHandle(handle);
        }
        if !ok {
            return None;
        }
        let ticks = ((creation.high as u128) << 32) | creation.low as u128;
        return Some(ticks / 10_000 - 11_644_473_600_000);
    }
    #[allow(unreachable_code)]
    None
}

#[cfg(windows)]
fn read_windows_uninstall_entries() -> Vec<UninstallEntry> {
    let mut all = String::new();
    for key in [
        r"HKCU\Software\Microsoft\Windows\CurrentVersion\Uninstall",
        r"HKLM\Software\Microsoft\Windows\CurrentVersion\Uninstall",
        r"HKLM\Software\WOW6432Node\Microsoft\Windows\CurrentVersion\Uninstall",
    ] {
        let mut reg = std::process::Command::new("reg");
        reg.args(["query", key, "/s"]);
        #[cfg(windows)]
        {
            use std::os::windows::process::CommandExt;
            reg.creation_flags(0x0800_0000); // CREATE_NO_WINDOW: never flash a console from a GUI host
        }
        if let Ok(output) = reg.output() {
            all.push_str(&String::from_utf8_lossy(&output.stdout));
            all.push('\n');
        }
    }
    parse_reg_query(&all)
}

// ---------------------------------------------------------------------------
// Host planning, takeover and detached start. Every service client plans through
// these; product crates map their own role names onto ranks (`known_roles`) and
// keep their own wire file names.
// ---------------------------------------------------------------------------

/// Rank of a live record or takeover request: its own `rank` when it writes one,
/// else `known_roles` by its `role`. `None` when neither resolves; such a peer is
/// never taken over from and never obliges a handoff.
pub fn resolve_rank(role: &str, rank: u8, known_roles: &BTreeMap<String, u8>) -> Option<u8> {
    if rank != RANK_UNKNOWN {
        Some(rank)
    } else {
        known_roles.get(role).copied()
    }
}

/// Outcome of [`plan_host`] for a process of a given rank.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum HostPlan {
    /// A live host of equal or better rank serves the service.
    UseRunning,
    /// A live host of lower rank serves it and this process outranks it: request takeover.
    TakeOver { from_role: String },
    /// Nothing live and nothing to start: host in this process.
    HostMyself,
    /// Nothing live: start this installed engine, then re-discover.
    StartInstalled(PathBuf),
}

/// Decide how a process of rank `my_rank` gets a host for `service`. A live record whose
/// rank cannot be resolved through `known_roles` counts as absent. `installed_engine` is
/// the executable the caller may start when nothing is live; it is used only when it exists.
pub fn plan_host(
    suite: &Suite,
    service: &str,
    my_rank: u8,
    known_roles: &BTreeMap<String, u8>,
    installed_engine: Option<&Path>,
) -> HostPlan {
    plan_from(
        read_live_record(suite, service),
        my_rank,
        known_roles,
        installed_engine,
    )
}

/// [`plan_host`] with `Err` when `service.json` exists but could not be read. Whether a live host
/// exists is then unknown, so no plan is made: `plan_host` would answer `HostMyself` or
/// `StartInstalled`, which starts a second host beside a live one.
pub fn try_plan_host(
    suite: &Suite,
    service: &str,
    my_rank: u8,
    known_roles: &BTreeMap<String, u8>,
    installed_engine: Option<&Path>,
) -> io::Result<HostPlan> {
    Ok(plan_from(
        try_read_live_record(suite, service)?,
        my_rank,
        known_roles,
        installed_engine,
    ))
}

fn plan_from(
    live: Option<ServiceRecord>,
    my_rank: u8,
    known_roles: &BTreeMap<String, u8>,
    installed_engine: Option<&Path>,
) -> HostPlan {
    if let Some(record) = live {
        if let Some(their) = resolve_rank(&record.role, record.rank, known_roles) {
            return if my_rank < their {
                HostPlan::TakeOver {
                    from_role: record.role,
                }
            } else {
                HostPlan::UseRunning
            };
        }
    }
    match installed_engine {
        Some(engine) if engine.exists() => HostPlan::StartInstalled(engine.to_path_buf()),
        _ => HostPlan::HostMyself,
    }
}

/// Takeover is requested only against a live, lower-ranked host held by another process.
/// `incumbent` is the live record's `(resolved rank, pid)`.
pub fn should_request_takeover(my_rank: u8, incumbent: Option<(u8, u32)>, own_pid: u32) -> bool {
    incumbent.is_some_and(|(their, pid)| my_rank < their && pid != own_pid)
}

/// Does a pending takeover request `(resolved rank, pid)` oblige a host of rank `my_rank`
/// to yield? Only a live, other-process, outranking requester does.
pub fn takeover_obliges_yield(
    my_rank: u8,
    request: Option<(u8, u32)>,
    own_pid: u32,
    alive: &dyn Fn(u32) -> bool,
) -> bool {
    request.is_some_and(|(their, pid)| pid != own_pid && their < my_rank && alive(pid))
}

/// Publish `takeover.json` for this process as `role` (with `rank`), atomically.
pub fn publish_takeover(suite: &Suite, service: &str, role: &str, rank: u8) -> io::Result<()> {
    write_takeover(
        suite,
        service,
        &TakeoverRequest {
            role: role.into(),
            rank,
            pid: std::process::id(),
            requested_at_ms: now_ms(),
        },
    )
}

fn now_ms() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_millis() as u64)
        .unwrap_or(0)
}

/// Start `program` detached: no inherited stdio, its own process group on Unix, and on
/// Windows no console window (`CREATE_NO_WINDOW | DETACHED_PROCESS`) so a GUI host never
/// flashes one. The child is not waited for.
pub fn spawn_detached(program: &Path, args: &[String], env: &[(String, String)]) -> io::Result<()> {
    let mut cmd = Command::new(program);
    cmd.args(args)
        .envs(env.iter().map(|(k, v)| (k.as_str(), v.as_str())))
        .stdin(Stdio::null())
        .stdout(Stdio::null())
        .stderr(Stdio::null());
    #[cfg(windows)]
    {
        use std::os::windows::process::CommandExt;
        const CREATE_NO_WINDOW: u32 = 0x0800_0000;
        const DETACHED_PROCESS: u32 = 0x0000_0008;
        cmd.creation_flags(CREATE_NO_WINDOW | DETACHED_PROCESS);
    }
    #[cfg(unix)]
    {
        use std::os::unix::process::CommandExt;
        cmd.process_group(0);
    }
    drop(cmd.spawn()?);
    Ok(())
}

/// Poll `ready` every 100 ms until it holds (`true`) or `wait` elapses (`false`).
/// Checks before the first sleep.
pub fn poll_until(wait: Duration, mut ready: impl FnMut() -> bool) -> bool {
    let deadline = Instant::now() + wait;
    loop {
        if ready() {
            return true;
        }
        if Instant::now() >= deadline {
            return false;
        }
        std::thread::sleep(Duration::from_millis(100));
    }
}

#[cfg(test)]
mod tests {
    /// flock (Unix `File::try_lock`) belongs to the open file description, which a child forked by another test in
    /// this binary shares until it execs (child-process spawns with pre_exec take the fork path on macOS). A lock
    /// released by drop can therefore read as held for that instant: re-acquire with a short bound. Not a release bug:
    /// the guard's drop closes our descriptor; the lock ends when the last shared copy closes (#788).
    fn eventually<T, E>(mut f: impl FnMut() -> Result<T, E>) -> Result<T, E> {
        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
        loop {
            match f() {
                Ok(v) => return Ok(v),
                Err(e) if std::time::Instant::now() >= deadline => return Err(e),
                Err(_) => std::thread::sleep(std::time::Duration::from_millis(20)),
            }
        }
    }

    use super::*;
    use std::fs;

    #[test]
    fn takeover_predicates_compare_resolved_ranks_and_skip_self() {
        let always = |_: u32| true;
        // Lower rank wins; equal rank never takes over.
        assert!(should_request_takeover(0, Some((1, 99)), 7));
        assert!(!should_request_takeover(1, Some((1, 99)), 7));
        // Own pid is never a takeover target.
        assert!(!should_request_takeover(0, Some((1, 7)), 7));
        assert!(!should_request_takeover(0, None, 7));
        assert!(takeover_obliges_yield(1, Some((0, 99)), 7, &always));
        assert!(!takeover_obliges_yield(0, Some((1, 99)), 7, &always));
        assert!(!takeover_obliges_yield(1, Some((0, 7)), 7, &always));
        assert!(!takeover_obliges_yield(1, Some((0, 99)), 7, &|_| false));
    }

    #[test]
    fn unresolvable_ranks_never_win_and_explicit_rank_beats_role_table() {
        let roles = BTreeMap::from([("shell".to_string(), 0u8)]);
        assert_eq!(resolve_rank("shell", RANK_UNKNOWN, &roles), Some(0));
        assert_eq!(resolve_rank("unknown", RANK_UNKNOWN, &roles), None);
        assert_eq!(resolve_rank("shell", 3, &roles), Some(3));
    }

    #[test]
    fn spawn_poll_and_takeover_publication_round_trip() {
        let d = tempfile::tempdir().unwrap();
        let s = Suite::at(d.path()).unwrap();
        publish_takeover(&s, "demo", "shell", 0).unwrap();
        let t = crate::record::read_takeover(&s, "demo").unwrap();
        assert_eq!(
            (t.role.as_str(), t.rank, t.pid),
            ("shell", 0, std::process::id())
        );
        assert!(poll_until(Duration::from_millis(50), || true));
        assert!(!poll_until(Duration::from_millis(50), || false));
    }

    #[test]
    fn mac_discovery_uses_known_bundle_and_first_root() {
        let d = tempfile::tempdir().unwrap();
        let bundle = d.path().join("DemoApp.app/Contents/MacOS");
        fs::create_dir_all(&bundle).unwrap();
        fs::write(bundle.join("demo-engine"), b"engine").unwrap();
        fs::write(
            d.path().join("DemoApp.app/Contents/Info.plist"),
            b"<key>CFBundleShortVersionString</key><string>1.2</string>",
        )
        .unwrap();
        let apps = discover_macos(&[d.path().to_path_buf()], &[("DemoApp", "demo-engine")]);
        assert_eq!(apps[0].version.as_deref(), Some("1.2"));
    }

    #[test]
    fn mac_discovery_reports_incomplete_installs_only_on_request() {
        let d = tempfile::tempdir().unwrap();
        // Broken install in the first root (bundle, no engine); nothing in the second.
        let first = d.path().join("first");
        let second = d.path().join("second");
        fs::create_dir_all(first.join("DemoApp.app/Contents/MacOS")).unwrap();
        fs::create_dir_all(&second).unwrap();
        let roots = [first.clone(), second.clone()];
        let reg = [("DemoApp", "demo-engine"), ("Absent", "absent-engine")];

        // Default behaviour is unchanged: incomplete installs are not reported.
        assert!(discover_macos(&roots, &reg).is_empty());
        assert!(discover_macos_with(&roots, &reg, &DiscoverOptions::default()).is_empty());

        let opts = DiscoverOptions {
            include_incomplete: true,
        };
        let apps = discover_macos_with(&roots, &reg, &opts);
        assert_eq!(
            apps.len(),
            1,
            "an app with no bundle at all is never reported"
        );
        assert_eq!(apps[0].app, "DemoApp");
        assert_eq!(apps[0].completeness, Completeness::EngineMissing);
        assert!(std::path::Path::new(&apps[0].engine_path).ends_with(
            std::path::Path::new("Contents")
                .join("MacOS")
                .join("demo-engine")
        ));
        assert!(apps[0].app_path.starts_with(first.to_str().unwrap()));

        // A complete install in a later root beats the incomplete one in an earlier root.
        let good = second.join("DemoApp.app/Contents/MacOS");
        fs::create_dir_all(&good).unwrap();
        fs::write(good.join("demo-engine"), b"engine").unwrap();
        let apps = discover_macos_with(&roots, &reg, &opts);
        assert_eq!(apps.len(), 1);
        assert_eq!(apps[0].completeness, Completeness::Complete);
        assert!(apps[0].app_path.starts_with(second.to_str().unwrap()));
        assert_eq!(discover_macos(&roots, &reg).len(), 1);
    }

    #[test]
    fn windows_discovery_reports_incomplete_installs_only_on_request() {
        let entry = |location: &str| UninstallEntry {
            display_name: "DemoApp".into(),
            install_location: Some(location.into()),
            display_icon: None,
            uninstall_string: None,
            display_version: Some("3.0".into()),
        };
        let reg = [("DemoApp", "demo-engine")];
        let broken = r"C:\Apps\Broken";
        let entries = [entry(broken), entry(r"C:\Apps\Gone")];
        // `Broken` has its directory but no engine; `Gone` has neither.
        let engine = |p: &str| p.starts_with("C:\\Apps\\Good\\");
        let dirs = |p: &str| p == broken;

        assert!(discover_windows(&entries, &reg, engine).is_empty());
        let opts = DiscoverOptions {
            include_incomplete: true,
        };
        let apps = discover_windows_with(&entries, &reg, engine, dirs, &opts);
        assert_eq!(apps.len(), 1);
        assert_eq!(apps[0].app_path, broken);
        assert_eq!(apps[0].completeness, Completeness::EngineMissing);
        assert_eq!(apps[0].version.as_deref(), Some("3.0"));

        // A complete install in a later entry beats the incomplete one.
        let entries = [entry(broken), entry(r"C:\Apps\Good")];
        let apps = discover_windows_with(&entries, &reg, engine, dirs, &opts);
        assert_eq!(apps.len(), 1);
        assert_eq!(apps[0].completeness, Completeness::Complete);
        assert_eq!(apps[0].app_path, r"C:\Apps\Good");
    }

    #[test]
    fn registry_parser_and_windows_projection_are_deterministic() {
        let text = "HKEY_CURRENT_USER\\x\n    DisplayName    REG_SZ    DemoApp\n    InstallLocation    REG_SZ    C:\\Apps\\DemoApp\n    DisplayVersion    REG_SZ    2.0\n";
        let rows = parse_reg_query(text);
        assert_eq!(rows.len(), 1);
        let apps = discover_windows(&rows, &[("DemoApp", "demo-engine")], |p| {
            p.ends_with("demo-engine.exe")
        });
        assert_eq!(apps[0].version.as_deref(), Some("2.0"));
    }

    #[test]
    fn spawn_claim_is_single_winner_and_released() {
        let d = tempfile::tempdir().unwrap();
        let s = Suite::at(d.path()).unwrap();
        let c = SpawnClaim::acquire(&s, "demo").unwrap();
        assert!(matches!(
            SpawnClaim::acquire(&s, "demo"),
            Err(SpawnClaimError::AlreadyHeld(_))
        ));
        drop(c);
        assert!(eventually(|| SpawnClaim::acquire(&s, "demo")).is_ok());
    }
}