code_moniker_query/

discovery.rs

use std::fs;
use std::fs::OpenOptions;
use std::hash::{Hash, Hasher};
use std::io::{Read, Write};
#[cfg(unix)]
use std::os::fd::AsRawFd;
#[cfg(windows)]
use std::os::windows::io::AsRawHandle;
#[cfg(windows)]
use std::os::windows::prelude::OsStrExt;
use std::path::{Path, PathBuf};
use std::sync::OnceLock;
use std::time::{SystemTime, UNIX_EPOCH};

use serde::{Deserialize, Serialize};

use crate::DaemonWorkspaceConfig;

#[derive(Clone, Debug, Default, Eq, PartialEq, Serialize, Deserialize)]
#[cfg_attr(feature = "schema", derive(schemars::JsonSchema))]
pub struct BuildIdentity {
	pub version: String,
	pub fingerprint: String,
}

pub fn current_build_identity(version: impl Into<String>) -> anyhow::Result<BuildIdentity> {
	static FINGERPRINT: OnceLock<Result<String, String>> = OnceLock::new();
	let fingerprint = FINGERPRINT.get_or_init(|| {
		std::env::current_exe()
			.map_err(|error| error.to_string())
			.and_then(|path| binary_fingerprint(path).map_err(|error| error.to_string()))
	});
	Ok(BuildIdentity {
		version: version.into(),
		fingerprint: fingerprint
			.clone()
			.map_err(|error| anyhow::anyhow!("cannot fingerprint current binary: {error}"))?,
	})
}

pub fn binary_fingerprint(path: impl AsRef<Path>) -> anyhow::Result<String> {
	let path = path.as_ref();
	let mut file = fs::File::open(path)
		.map_err(|error| anyhow::anyhow!("cannot read binary {}: {error}", path.display()))?;
	let mut buffer = [0_u8; 64 * 1024];
	let mut hash = 0xcbf29ce484222325_u64;
	loop {
		let read = file.read(&mut buffer)?;
		if read == 0 {
			break;
		}
		for byte in &buffer[..read] {
			hash ^= u64::from(*byte);
			hash = hash.wrapping_mul(0x100000001b3);
		}
	}
	Ok(format!("fnv1a64:{hash:016x}"))
}

#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[cfg_attr(feature = "schema", derive(schemars::JsonSchema))]
pub struct DaemonRegistryEntry {
	pub workspace_root: String,
	pub workspace_roots: Vec<String>,
	pub project: Option<String>,
	pub cache_dir: Option<String>,
	pub live_refresh: Option<String>,
	pub endpoint: String,
	pub token: String,
	pub pid: u32,
	#[serde(default)]
	pub build: BuildIdentity,
	#[serde(default)]
	pub heartbeat_unix_ms: u64,
}

pub const DAEMON_REGISTRY_HEARTBEAT_TIMEOUT_MS: u64 = 15_000;

pub fn registry_heartbeat_unix_ms() -> u64 {
	SystemTime::now()
		.duration_since(UNIX_EPOCH)
		.unwrap_or_default()
		.as_millis()
		.try_into()
		.unwrap_or(u64::MAX)
}

pub fn daemon_registry_heartbeat_expired(entry: &DaemonRegistryEntry) -> bool {
	entry.heartbeat_unix_ms == 0
		|| registry_heartbeat_unix_ms().saturating_sub(entry.heartbeat_unix_ms)
			> DAEMON_REGISTRY_HEARTBEAT_TIMEOUT_MS
}

pub fn registry_dir() -> PathBuf {
	std::env::var_os("CODE_MONIKER_REGISTRY_DIR")
		.map(PathBuf::from)
		.unwrap_or_else(|| std::env::temp_dir().join("code-moniker-daemons"))
}

pub fn canonical_workspace_root(root: impl AsRef<Path>) -> anyhow::Result<PathBuf> {
	let root = root.as_ref();
	root.canonicalize()
		.map_err(|err| anyhow::anyhow!("cannot canonicalize {}: {err}", root.display()))
}

pub fn canonical_workspace_roots<I, P>(roots: I) -> anyhow::Result<Vec<PathBuf>>
where
	I: IntoIterator<Item = P>,
	P: AsRef<Path>,
{
	let mut canonical = Vec::new();
	for root in roots {
		let root = canonical_workspace_root(root)?;
		if !canonical.contains(&root) {
			canonical.push(root);
		}
	}
	if canonical.is_empty() {
		canonical.push(canonical_workspace_root(".")?);
	}
	Ok(canonical)
}

pub fn daemon_workspace_config<I, P>(
	roots: I,
	project: Option<String>,
	cache_dir: Option<PathBuf>,
	live_refresh: Option<String>,
) -> anyhow::Result<DaemonWorkspaceConfig>
where
	I: IntoIterator<Item = P>,
	P: AsRef<Path>,
{
	let roots = canonical_workspace_roots(roots)?;
	Ok(DaemonWorkspaceConfig {
		roots: roots
			.into_iter()
			.map(|root| root.display().to_string())
			.collect(),
		project,
		cache_dir: cache_dir
			.map(normalize_path)
			.transpose()?
			.map(|path| path.display().to_string()),
		live_refresh,
	})
}

pub fn validate_daemon_start_config(config: &DaemonWorkspaceConfig) -> anyhow::Result<()> {
	let roots = config_roots(config);
	if let Some(root) = roots.iter().find(|root| root.parent().is_none()) {
		anyhow::bail!(
			"refusing to start a code-moniker daemon at filesystem root `{}`; pass an explicit project directory (MCP configurations should use an absolute project path)",
			root.display()
		);
	}
	Ok(())
}

pub fn canonical_workspace_config(
	config: DaemonWorkspaceConfig,
) -> anyhow::Result<DaemonWorkspaceConfig> {
	daemon_workspace_config(
		config.roots.iter().map(PathBuf::from),
		config.project,
		config.cache_dir.map(PathBuf::from),
		config.live_refresh,
	)
}

pub fn config_from_roots<I, P>(roots: I) -> anyhow::Result<DaemonWorkspaceConfig>
where
	I: IntoIterator<Item = P>,
	P: AsRef<Path>,
{
	daemon_workspace_config(roots, None, None, Some("on-demand".to_string()))
}

pub fn registry_path_for_root(root: impl AsRef<Path>) -> anyhow::Result<PathBuf> {
	registry_path_for_roots([root])
}

pub fn registry_path_for_roots<I, P>(roots: I) -> anyhow::Result<PathBuf>
where
	I: IntoIterator<Item = P>,
	P: AsRef<Path>,
{
	registry_path_for_config(&config_from_roots(roots)?)
}

pub fn registry_path_for_config(config: &DaemonWorkspaceConfig) -> anyhow::Result<PathBuf> {
	let config = canonical_workspace_config(config.clone())?;
	Ok(registry_dir().join(format!("{}.json", stable_config_hash(&config))))
}

pub fn daemon_log_path_for_config(config: &DaemonWorkspaceConfig) -> anyhow::Result<PathBuf> {
	Ok(registry_path_for_config(config)?.with_extension("log"))
}

pub fn read_registry_entry(
	config: &DaemonWorkspaceConfig,
) -> anyhow::Result<Option<DaemonRegistryEntry>> {
	let path = registry_path_for_config(config)?;
	match fs::read_to_string(&path) {
		Ok(text) => parse_registry_entry(&path, &text).map(Some),
		Err(err) if err.kind() == std::io::ErrorKind::NotFound => Ok(None),
		Err(err) => Err(anyhow::anyhow!(
			"cannot read daemon registry entry {}: {err}",
			path.display()
		)),
	}
}

pub fn write_registry_entry(
	config: &DaemonWorkspaceConfig,
	entry: &DaemonRegistryEntry,
) -> anyhow::Result<()> {
	fs::create_dir_all(registry_dir())?;
	atomic_write_registry_entry(&registry_path_for_config(config)?, entry)?;
	Ok(())
}

pub fn claim_registry_entry(
	config: &DaemonWorkspaceConfig,
	entry: &DaemonRegistryEntry,
) -> anyhow::Result<bool> {
	fs::create_dir_all(registry_dir())?;
	claim_registry_file(&registry_path_for_config(config)?, entry)
}

fn claim_registry_file(path: &Path, entry: &DaemonRegistryEntry) -> anyhow::Result<bool> {
	claim_registry_file_before_publish(path, entry, || {})
}

fn claim_registry_file_before_publish(
	path: &Path,
	entry: &DaemonRegistryEntry,
	before_publish: impl FnOnce(),
) -> anyhow::Result<bool> {
	let temp = path.with_extension(format!("{}.claim.tmp", entry.token));
	let text = serde_json::to_vec_pretty(entry)?;
	let mut file = OpenOptions::new()
		.write(true)
		.create_new(true)
		.open(&temp)?;
	let prepared = (|| {
		file.write_all(&text)?;
		file.sync_all()?;
		Ok::<_, anyhow::Error>(())
	})();
	drop(file);
	if let Err(error) = prepared {
		let _ = fs::remove_file(temp);
		return Err(error);
	}
	before_publish();
	let result = publish_registry_claim(&temp, path);
	let _ = fs::remove_file(temp);
	result
}

#[cfg(not(windows))]
fn publish_registry_claim(source: &Path, destination: &Path) -> anyhow::Result<bool> {
	match fs::hard_link(source, destination) {
		Ok(()) => Ok(true),
		Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => Ok(false),
		Err(error) => Err(error.into()),
	}
}

#[cfg(windows)]
fn publish_registry_claim(source: &Path, destination: &Path) -> anyhow::Result<bool> {
	use windows_sys::Win32::Storage::FileSystem::{MOVEFILE_WRITE_THROUGH, MoveFileExW};

	let source = source
		.as_os_str()
		.encode_wide()
		.chain(std::iter::once(0))
		.collect::<Vec<_>>();
	let destination = destination
		.as_os_str()
		.encode_wide()
		.chain(std::iter::once(0))
		.collect::<Vec<_>>();
	if unsafe {
		MoveFileExW(
			source.as_ptr(),
			destination.as_ptr(),
			MOVEFILE_WRITE_THROUGH,
		)
	} != 0
	{
		return Ok(true);
	}
	let error = std::io::Error::last_os_error();
	if error.kind() == std::io::ErrorKind::AlreadyExists {
		Ok(false)
	} else {
		Err(error.into())
	}
}

pub fn update_registry_entry_if_own(
	config: &DaemonWorkspaceConfig,
	entry: &DaemonRegistryEntry,
) -> anyhow::Result<bool> {
	let path = registry_path_for_config(config)?;
	with_registry_lock(&path, || {
		let current = match fs::read_to_string(&path) {
			Ok(text) => Some(parse_registry_entry(&path, &text)?),
			Err(error) if error.kind() == std::io::ErrorKind::NotFound => None,
			Err(error) => {
				return Err(anyhow::anyhow!(
					"cannot read daemon registry entry {}: {error}",
					path.display()
				));
			}
		};
		let owned = current
			.map(|current| current.token == entry.token && current.pid == entry.pid)
			.unwrap_or(false);
		if owned {
			atomic_write_registry_entry(&path, entry)?;
		}
		Ok(owned)
	})
}

fn parse_registry_entry(path: &Path, text: &str) -> anyhow::Result<DaemonRegistryEntry> {
	serde_json::from_str(text).map_err(|error| {
		anyhow::anyhow!(
			"cannot decode daemon registry entry {}: {error}",
			path.display()
		)
	})
}

fn atomic_write_registry_entry(path: &Path, entry: &DaemonRegistryEntry) -> anyhow::Result<()> {
	let temp = path.with_extension(format!("{}.tmp", entry.token));
	let text = serde_json::to_vec_pretty(entry)?;
	{
		let mut file = OpenOptions::new()
			.write(true)
			.create_new(true)
			.open(&temp)?;
		file.write_all(&text)?;
		file.sync_all()?;
	}
	replace_registry_file(&temp, path)?;
	Ok(())
}

#[cfg(windows)]
fn replace_registry_file(source: &Path, destination: &Path) -> anyhow::Result<()> {
	use windows_sys::Win32::Storage::FileSystem::{
		MOVEFILE_REPLACE_EXISTING, MOVEFILE_WRITE_THROUGH, MoveFileExW,
	};

	let source = source
		.as_os_str()
		.encode_wide()
		.chain(std::iter::once(0))
		.collect::<Vec<_>>();
	let destination = destination
		.as_os_str()
		.encode_wide()
		.chain(std::iter::once(0))
		.collect::<Vec<_>>();
	if unsafe {
		MoveFileExW(
			source.as_ptr(),
			destination.as_ptr(),
			MOVEFILE_REPLACE_EXISTING | MOVEFILE_WRITE_THROUGH,
		)
	} == 0
	{
		return Err(std::io::Error::last_os_error().into());
	}
	Ok(())
}

#[cfg(not(windows))]
fn replace_registry_file(source: &Path, destination: &Path) -> anyhow::Result<()> {
	fs::rename(source, destination)?;
	Ok(())
}

// Shutdown-time removal: a successor daemon may have overwritten this path
// with its own entry while we were stopping. Only unlink what is still ours,
// or the new daemon stays alive but invisible to the registry.
pub fn remove_registry_entry_if_own(path: &Path, own: &DaemonRegistryEntry) {
	let _ = with_registry_lock(path, || {
		let current = fs::read_to_string(path)
			.ok()
			.and_then(|text| serde_json::from_str::<DaemonRegistryEntry>(&text).ok());
		let owned = current
			.map(|entry| entry.token == own.token && entry.pid == own.pid)
			.unwrap_or(false);
		if owned {
			let _ = fs::remove_file(path);
		}
		Ok(())
	});
}

#[cfg(unix)]
fn with_registry_lock<T>(
	path: &Path,
	action: impl FnOnce() -> anyhow::Result<T>,
) -> anyhow::Result<T> {
	let lock_path = path.with_extension("lock");
	let lock = OpenOptions::new()
		.read(true)
		.write(true)
		.create(true)
		.truncate(false)
		.open(lock_path)?;
	// SAFETY: flock operates on this process-owned open descriptor. The lock is
	// released automatically when `lock` is dropped on every return path.
	if unsafe { libc::flock(lock.as_raw_fd(), libc::LOCK_EX) } == -1 {
		return Err(std::io::Error::last_os_error().into());
	}
	action()
}

#[cfg(not(unix))]
fn with_registry_lock<T>(
	path: &Path,
	action: impl FnOnce() -> anyhow::Result<T>,
) -> anyhow::Result<T> {
	#[cfg(windows)]
	{
		use windows_sys::Win32::Storage::FileSystem::{
			LOCKFILE_EXCLUSIVE_LOCK, LockFileEx, UnlockFileEx,
		};
		use windows_sys::Win32::System::IO::OVERLAPPED;

		let lock_path = path.with_extension("lock");
		let lock = OpenOptions::new()
			.read(true)
			.write(true)
			.create(true)
			.truncate(false)
			.open(lock_path)?;
		let mut overlapped = unsafe { std::mem::zeroed::<OVERLAPPED>() };
		let handle = lock.as_raw_handle();
		if unsafe {
			LockFileEx(
				handle,
				LOCKFILE_EXCLUSIVE_LOCK,
				0,
				u32::MAX,
				u32::MAX,
				&mut overlapped,
			)
		} == 0
		{
			return Err(std::io::Error::last_os_error().into());
		}
		let result = action();
		let unlock = unsafe { UnlockFileEx(handle, 0, u32::MAX, u32::MAX, &mut overlapped) };
		if unlock == 0 && result.is_ok() {
			return Err(std::io::Error::last_os_error().into());
		}
		result
	}
	#[cfg(not(windows))]
	{
		let _ = path;
		action()
	}
}

pub fn list_registry_files() -> anyhow::Result<Vec<(PathBuf, DaemonRegistryEntry)>> {
	let dir = registry_dir();
	if !dir.exists() {
		return Ok(Vec::new());
	}
	let mut entries = Vec::new();
	for entry in fs::read_dir(&dir)? {
		let entry = entry?;
		if entry.path().extension().and_then(|ext| ext.to_str()) != Some("json") {
			continue;
		}
		let text = fs::read_to_string(entry.path())?;
		if let Ok(registry) = serde_json::from_str::<DaemonRegistryEntry>(&text) {
			entries.push((entry.path(), registry));
		}
	}
	entries.sort_by(|(_, a), (_, b)| a.workspace_root.cmp(&b.workspace_root));
	Ok(entries)
}

pub fn pid_is_alive(pid: u32) -> bool {
	#[cfg(unix)]
	{
		// SAFETY: signal 0 never delivers a signal; it only asks the kernel
		// whether the PID exists and whether this process may signal it.
		let result = unsafe { libc::kill(pid as libc::pid_t, 0) };
		let errno = (result != 0)
			.then(|| std::io::Error::last_os_error().raw_os_error())
			.flatten();
		kill_result_means_alive(result, errno)
	}
	#[cfg(not(unix))]
	{
		#[cfg(windows)]
		{
			windows_pid_is_alive(pid)
		}
		#[cfg(not(windows))]
		{
			let _ = pid;
			false
		}
	}
}

#[cfg(unix)]
fn kill_result_means_alive(result: i32, errno: Option<i32>) -> bool {
	result == 0 || errno != Some(libc::ESRCH)
}

#[cfg(windows)]
fn windows_pid_is_alive(pid: u32) -> bool {
	use windows_sys::Win32::Foundation::{
		CloseHandle, ERROR_ACCESS_DENIED, GetLastError, WAIT_TIMEOUT,
	};
	use windows_sys::Win32::System::Threading::{
		OpenProcess, PROCESS_QUERY_LIMITED_INFORMATION, PROCESS_SYNCHRONIZE, WaitForSingleObject,
	};

	let handle = unsafe {
		OpenProcess(
			PROCESS_QUERY_LIMITED_INFORMATION | PROCESS_SYNCHRONIZE,
			0,
			pid,
		)
	};
	if handle.is_null() {
		return unsafe { GetLastError() } == ERROR_ACCESS_DENIED;
	}
	let state = unsafe { WaitForSingleObject(handle, 0) };
	unsafe {
		CloseHandle(handle);
	}
	state == WAIT_TIMEOUT
}

pub fn list_registry_entries() -> anyhow::Result<Vec<DaemonRegistryEntry>> {
	let mut entries = Vec::new();
	for (path, entry) in list_registry_files()? {
		if pid_is_alive(entry.pid) {
			entries.push(entry);
		} else {
			remove_registry_entry_if_own(&path, &entry);
		}
	}
	entries.sort_by(|a, b| a.workspace_root.cmp(&b.workspace_root));
	Ok(entries)
}

pub fn config_roots(config: &DaemonWorkspaceConfig) -> Vec<PathBuf> {
	config.roots.iter().map(PathBuf::from).collect()
}

pub fn workspace_label(roots: &[PathBuf]) -> String {
	if roots.len() == 1 {
		roots[0].display().to_string()
	} else {
		roots
			.iter()
			.map(|root| root.display().to_string())
			.collect::<Vec<_>>()
			.join(";")
	}
}

fn normalize_path(path: PathBuf) -> anyhow::Result<PathBuf> {
	if path.is_absolute() {
		Ok(path)
	} else {
		Ok(std::env::current_dir()?.join(path))
	}
}

// The registry key is the workspace identity: what gets indexed (roots,
// project, cache), never how it refreshes. Hashing live_refresh here once
// split one workspace across two registry slots — a daemon started with
// `--live-refresh auto` was invisible to a default-mode `daemon status`.
fn stable_config_hash(config: &DaemonWorkspaceConfig) -> String {
	let mut hasher = StableHasher::default();
	for root in &config.roots {
		root.hash(&mut hasher);
		0xff_u8.hash(&mut hasher);
	}
	config.project.hash(&mut hasher);
	0xfe_u8.hash(&mut hasher);
	config.cache_dir.hash(&mut hasher);
	format!("{:016x}", hasher.finish())
}

#[derive(Default)]
struct StableHasher(u64);

impl Hasher for StableHasher {
	fn finish(&self) -> u64 {
		self.0
	}

	fn write(&mut self, bytes: &[u8]) {
		let mut hash = if self.0 == 0 {
			0xcbf29ce484222325
		} else {
			self.0
		};
		for byte in bytes {
			hash ^= u64::from(*byte);
			hash = hash.wrapping_mul(0x100000001b3);
		}
		self.0 = hash;
	}
}

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

	fn entry(token: &str, pid: u32) -> DaemonRegistryEntry {
		DaemonRegistryEntry {
			workspace_root: "/tmp/ws".to_string(),
			workspace_roots: vec!["/tmp/ws".to_string()],
			project: None,
			cache_dir: None,
			live_refresh: None,
			endpoint: "127.0.0.1:1".to_string(),
			token: token.to_string(),
			pid,
			build: BuildIdentity::default(),
			heartbeat_unix_ms: registry_heartbeat_unix_ms(),
		}
	}

	#[test]
	fn registry_identity_ignores_the_refresh_mode() {
		let base = DaemonWorkspaceConfig {
			roots: vec!["/tmp/ws".to_string()],
			project: None,
			cache_dir: None,
			live_refresh: Some("auto".to_string()),
		};
		let mut on_demand = base.clone();
		on_demand.live_refresh = Some("on-demand".to_string());
		assert_eq!(
			stable_config_hash(&base),
			stable_config_hash(&on_demand),
			"one workspace must map to one registry slot, whatever the refresh mode"
		);

		let mut other_project = base.clone();
		other_project.project = Some("api".to_string());
		assert_ne!(
			stable_config_hash(&base),
			stable_config_hash(&other_project),
			"what gets indexed still separates registry slots"
		);
	}

	#[test]
	fn malformed_registry_entry_is_not_treated_as_missing_or_unowned() {
		let workspace = tempfile::tempdir().expect("workspace");
		let config = config_from_roots([workspace.path()]).expect("workspace config");
		let path = registry_path_for_config(&config).expect("registry path");
		fs::create_dir_all(path.parent().expect("registry directory"))
			.expect("create registry directory");
		fs::write(&path, "{not-json").expect("write malformed registry entry");

		let read_error = read_registry_entry(&config)
			.expect_err("malformed registry entry must be a typed read failure");
		assert!(
			read_error
				.to_string()
				.contains("cannot decode daemon registry entry"),
			"{read_error:#}"
		);

		let update_error = update_registry_entry_if_own(&config, &entry("owner", 111))
			.expect_err("malformed registry entry must not look like an unowned claim");
		assert!(
			update_error
				.to_string()
				.contains("cannot decode daemon registry entry"),
			"{update_error:#}"
		);

		let _ = fs::remove_file(&path);
		let _ = fs::remove_file(path.with_extension("lock"));
	}

	#[cfg(unix)]
	#[test]
	fn daemon_workspace_rejects_the_filesystem_root() {
		let config =
			daemon_workspace_config([Path::new("/")], None, None, Some("auto".to_string()))
				.expect("filesystem root identity remains available for status and cleanup");
		let error = validate_daemon_start_config(&config)
			.expect_err("filesystem root must fail before daemon startup");
		let message = error.to_string();
		assert!(message.contains("refusing to start"), "{message}");
		assert!(message.contains("absolute project path"), "{message}");
	}

	#[test]
	fn shutdown_removal_spares_a_successor_entry() {
		let dir = tempfile::tempdir().expect("tempdir");
		let path = dir.path().join("ws.json");
		let old = entry("old-token", 111);
		let new = entry("new-token", 222);

		fs::write(&path, serde_json::to_string(&new).expect("json")).expect("write");
		remove_registry_entry_if_own(&path, &old);
		assert!(path.exists(), "the successor's entry must survive");

		remove_registry_entry_if_own(&path, &new);
		assert!(!path.exists(), "the owner removes its own entry");

		remove_registry_entry_if_own(&path, &new);
	}

	#[test]
	fn legacy_registry_entries_default_missing_build_identity() {
		let mut value = serde_json::to_value(entry("legacy", 111)).expect("json");
		value.as_object_mut().expect("object").remove("build");
		let decoded: DaemonRegistryEntry = serde_json::from_value(value).expect("legacy entry");
		assert_eq!(decoded.build, BuildIdentity::default());
	}

	#[test]
	fn binary_fingerprint_changes_with_binary_content() {
		let temp = tempfile::tempdir().expect("tempdir");
		let binary = temp.path().join("code-moniker");
		fs::write(&binary, b"old build").expect("old build");
		let old = binary_fingerprint(&binary).expect("old fingerprint");
		fs::write(&binary, b"new build").expect("new build");
		let new = binary_fingerprint(&binary).expect("new fingerprint");

		assert_ne!(old, new);
		assert!(old.starts_with("fnv1a64:"));
		assert!(new.starts_with("fnv1a64:"));
	}

	#[test]
	fn missing_or_old_heartbeat_expires_but_a_fresh_claim_does_not() {
		let mut registry = entry("heartbeat", 111);
		registry.heartbeat_unix_ms = 0;
		assert!(daemon_registry_heartbeat_expired(&registry));
		registry.heartbeat_unix_ms =
			registry_heartbeat_unix_ms() - DAEMON_REGISTRY_HEARTBEAT_TIMEOUT_MS - 1;
		assert!(daemon_registry_heartbeat_expired(&registry));
		registry.heartbeat_unix_ms = registry_heartbeat_unix_ms();
		assert!(!daemon_registry_heartbeat_expired(&registry));
	}

	#[test]
	fn atomic_registry_update_replaces_a_complete_entry() {
		let dir = tempfile::tempdir().expect("tempdir");
		let path = dir.path().join("workspace.json");
		let initial = entry("same-daemon", 111);
		atomic_write_registry_entry(&path, &initial).expect("write initial entry");
		let updated = DaemonRegistryEntry {
			heartbeat_unix_ms: initial.heartbeat_unix_ms + 1,
			..initial.clone()
		};
		atomic_write_registry_entry(&path, &updated).expect("write updated entry");
		let read: DaemonRegistryEntry =
			serde_json::from_str(&fs::read_to_string(path).expect("read entry")).expect("json");
		assert_eq!(read, updated);
	}

	#[test]
	fn registry_claim_keeps_the_final_path_hidden_until_publication() {
		let dir = tempfile::tempdir().expect("tempdir");
		let path = dir.path().join("workspace.json");
		let owner = entry("owner-token", 111);
		let timeout = std::time::Duration::from_secs(5);
		let (prepared_tx, prepared_rx) = std::sync::mpsc::sync_channel(1);
		let (publish_tx, publish_rx) = std::sync::mpsc::sync_channel(1);
		let claim_path = path.clone();
		let claim_owner = owner.clone();
		let claimant = std::thread::spawn(move || {
			claim_registry_file_before_publish(&claim_path, &claim_owner, || {
				prepared_tx.send(()).expect("signal prepared claim");
				publish_rx
					.recv_timeout(timeout)
					.expect("receive publication release");
			})
		});

		prepared_rx
			.recv_timeout(timeout)
			.expect("claimant prepares the complete temporary entry");
		assert!(
			!path.exists(),
			"the final registry path must stay hidden while its complete temporary is pending"
		);
		publish_tx.send(()).expect("release registry publication");
		assert!(
			claimant
				.join()
				.expect("join claimant")
				.expect("claim registry"),
			"the first complete entry must win the registry claim"
		);

		let published: DaemonRegistryEntry = serde_json::from_str(
			&fs::read_to_string(&path).expect("read published registry entry"),
		)
		.expect("published registry entry is complete JSON");
		assert_eq!(published, owner);
	}

	#[test]
	fn concurrent_registry_claims_publish_exactly_one_complete_owner() {
		let dir = tempfile::tempdir().expect("tempdir");
		let path = dir.path().join("workspace.json");
		let first = entry("first-token", 111);
		let second = entry("second-token", 222);
		let timeout = std::time::Duration::from_secs(5);
		let (prepared_tx, prepared_rx) = std::sync::mpsc::sync_channel(2);
		let spawn_claimant = |claim: DaemonRegistryEntry| {
			let claim_path = path.clone();
			let claimant_prepared_tx = prepared_tx.clone();
			let (publish_tx, publish_rx) = std::sync::mpsc::sync_channel(1);
			let claimant = std::thread::spawn(move || {
				claim_registry_file_before_publish(&claim_path, &claim, || {
					claimant_prepared_tx
						.send(())
						.expect("signal prepared claim");
					publish_rx
						.recv_timeout(timeout)
						.expect("receive publication release");
				})
			});
			(claimant, publish_tx)
		};
		let (first_claimant, first_publish_tx) = spawn_claimant(first.clone());
		let (second_claimant, second_publish_tx) = spawn_claimant(second.clone());

		for _ in 0..2 {
			prepared_rx
				.recv_timeout(timeout)
				.expect("claimant prepares a complete temporary entry");
		}
		assert!(
			!path.exists(),
			"neither prepared claimant may expose the final path before publication"
		);
		first_publish_tx
			.send(())
			.expect("release first registry publication");
		second_publish_tx
			.send(())
			.expect("release second registry publication");
		let first_won = first_claimant
			.join()
			.expect("join first claimant")
			.expect("first claim");
		let second_won = second_claimant
			.join()
			.expect("join second claimant")
			.expect("second claim");
		assert_ne!(first_won, second_won, "exactly one claimant must win");

		let published: DaemonRegistryEntry = serde_json::from_str(
			&fs::read_to_string(&path).expect("read published registry entry"),
		)
		.expect("published registry entry is complete JSON");
		assert_eq!(published, if first_won { first } else { second });
		assert_eq!(
			fs::read_dir(dir.path()).expect("list registry dir").count(),
			1,
			"claim publication must not leave temporary files"
		);
	}

	#[cfg(unix)]
	#[test]
	fn permission_denied_pid_is_alive_but_missing_pid_is_dead() {
		assert!(kill_result_means_alive(-1, Some(libc::EPERM)));
		assert!(!kill_result_means_alive(-1, Some(libc::ESRCH)));
		assert!(kill_result_means_alive(0, None));
	}

	#[cfg(windows)]
	#[test]
	fn current_windows_process_is_alive() {
		assert!(pid_is_alive(std::process::id()));
	}

	#[test]
	fn registry_directory_honors_the_environment_override() {
		let expected = tempfile::tempdir()
			.expect("registry tempdir")
			.path()
			.join("custom-registry");
		let status = std::process::Command::new(std::env::current_exe().expect("test binary"))
			.args([
				"--exact",
				"discovery::tests::registry_directory_environment_child",
				"--ignored",
			])
			.env("CODE_MONIKER_REGISTRY_DIR", &expected)
			.env("CODE_MONIKER_REGISTRY_TEST_EXPECTED", &expected)
			.status()
			.expect("run registry environment child");
		assert!(status.success());
	}

	#[test]
	#[ignore = "subprocess fixture"]
	fn registry_directory_environment_child() {
		let Some(expected) = std::env::var_os("CODE_MONIKER_REGISTRY_TEST_EXPECTED") else {
			return;
		};
		assert_eq!(registry_dir(), PathBuf::from(expected));
	}

	#[test]
	fn registry_lock_serializes_processes() {
		let dir = tempfile::tempdir().expect("tempdir");
		let target = dir.path().join("workspace.json");
		let started = dir.path().join("child-started");
		let acquired = dir.path().join("child-acquired");
		let mut child = None;

		with_registry_lock(&target, || {
			child = Some(
				std::process::Command::new(std::env::current_exe()?)
					.args([
						"--exact",
						"discovery::tests::registry_lock_child",
						"--nocapture",
					])
					.env("CODE_MONIKER_LOCK_TEST_TARGET", &target)
					.env("CODE_MONIKER_LOCK_TEST_STARTED", &started)
					.env("CODE_MONIKER_LOCK_TEST_ACQUIRED", &acquired)
					.spawn()?,
			);
			let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
			while !started.exists() && std::time::Instant::now() < deadline {
				std::thread::sleep(std::time::Duration::from_millis(10));
			}
			assert!(started.exists(), "child process did not reach the lock");
			std::thread::sleep(std::time::Duration::from_millis(200));
			assert!(
				!acquired.exists(),
				"child acquired the registry lock before its owner released it"
			);
			Ok(())
		})
		.expect("hold parent lock");

		let status = child
			.expect("child process")
			.wait()
			.expect("wait for child");
		assert!(status.success(), "child lock process failed: {status}");
		assert!(acquired.exists(), "child never acquired the released lock");
	}

	#[test]
	fn registry_lock_child() {
		let (Some(target), Some(started), Some(acquired)) = (
			std::env::var_os("CODE_MONIKER_LOCK_TEST_TARGET"),
			std::env::var_os("CODE_MONIKER_LOCK_TEST_STARTED"),
			std::env::var_os("CODE_MONIKER_LOCK_TEST_ACQUIRED"),
		) else {
			return;
		};
		fs::write(&started, b"started").expect("announce child");
		with_registry_lock(Path::new(&target), || {
			fs::write(&acquired, b"acquired")?;
			Ok(())
		})
		.expect("acquire child lock");
	}
}