watchexec-cli 2.7.4

use std::{
	borrow::Cow,
	collections::HashMap,
	env::{var, var_os},
	ffi::OsStr,
	fmt,
	fs::File,
	io::{IsTerminal, Write},
	iter::once,
	path::{Path, PathBuf},
	process::{ExitCode, Stdio},
	sync::{
		atomic::{AtomicBool, AtomicU8, Ordering},
		Arc, Mutex,
	},
	time::{Duration, Instant},
};

use clearscreen::ClearScreen;
use miette::{IntoDiagnostic, Report, Result};
use notify_rust::Notification;
use termcolor::{Color, ColorChoice, ColorSpec, StandardStream, WriteColor};
use tokio::{process::Command as TokioCommand, time::sleep};
use tracing::{debug, debug_span, error, instrument, trace, trace_span, warn, Instrument};
use watchexec::{
	action::ActionHandler,
	command::{Command, Program, Shell, SpawnOptions},
	error::RuntimeError,
	job::{CommandState, Job},
	sources::fs::Watcher,
	Config, ErrorHook, Id,
};
use watchexec_events::{Event, KeyCode, Keyboard, Priority, ProcessEnd, Tag};
use watchexec_signals::Signal;

use crate::{
	args::{
		command::{EnvVar, WrapMode},
		events::{EmitEvents, OnBusyUpdate, SignalMapping},
		output::{ClearMode, ColourMode, NotifyMode},
		Args,
	},
	emits::events_to_simple_format,
	socket::Sockets,
	state::State,
};

#[derive(Clone, Copy, Debug)]
struct OutputFlags {
	quiet: bool,
	colour: ColorChoice,
	timings: bool,
	bell: bool,
	notify: Option<NotifyMode>,
}

#[derive(Clone, Copy, Debug)]
struct TimeoutConfig {
	/// The maximum duration the command is allowed to run
	timeout: Option<Duration>,
	/// Signal to send for graceful stop (used when timeout fires)
	stop_signal: Signal,
	/// Grace period after stop signal before force kill
	stop_timeout: Duration,
}

const RUNTIME_ERROR_SAMPLE_INTERVAL: Duration = Duration::from_secs(1);
const RUNTIME_ERROR_SAMPLE_CAPACITY: usize = 256;

struct RuntimeErrorSample {
	last_emitted: Instant,
	last_seen: Instant,
}

struct RuntimeErrorSampler {
	samples: HashMap<String, RuntimeErrorSample>,
	interval: Duration,
	capacity: usize,
}

impl RuntimeErrorSampler {
	fn new(interval: Duration, capacity: usize) -> Self {
		Self {
			samples: HashMap::new(),
			interval,
			capacity,
		}
	}

	fn should_emit(&mut self, message: String, now: Instant) -> bool {
		if self.capacity == 0 {
			return true;
		}

		if let Some(sample) = self.samples.get_mut(&message) {
			let should_emit = now.saturating_duration_since(sample.last_emitted) >= self.interval;
			sample.last_seen = now;
			if should_emit {
				sample.last_emitted = now;
			}
			return should_emit;
		}

		self.samples
			.retain(|_, sample| now.saturating_duration_since(sample.last_seen) < self.interval);

		if self.samples.len() >= self.capacity {
			// Bounded state takes priority over the sampling window, so an evicted
			// message will be reported if it occurs again.
			let oldest = self
				.samples
				.iter()
				.min_by_key(|(_, sample)| sample.last_seen)
				.map(|(message, _)| message.clone());
			if let Some(oldest) = oldest {
				self.samples.remove(&oldest);
			}
		}

		self.samples.insert(
			message,
			RuntimeErrorSample {
				last_emitted: now,
				last_seen: now,
			},
		);
		true
	}
}

pub fn make_config(args: &Args, state: &State) -> Result<Config> {
	let _span = debug_span!("args-runtime").entered();
	let mut config = Config::default();
	config.signal_job_control = true;
	let runtime_error_sampler = Mutex::new(RuntimeErrorSampler::new(
		RUNTIME_ERROR_SAMPLE_INTERVAL,
		RUNTIME_ERROR_SAMPLE_CAPACITY,
	));
	config.on_error(move |err: ErrorHook| {
		if let RuntimeError::IoError {
			about: "waiting on process group",
			..
		} = err.error
		{
			// "No child processes" and such
			// these are often spurious, so condemn them to -v only
			error!("{}", err.error);
			return;
		}

		let should_emit = {
			let mut sampler = runtime_error_sampler
				.lock()
				.unwrap_or_else(std::sync::PoisonError::into_inner);
			sampler.should_emit(err.error.to_string(), Instant::now())
		};
		if !should_emit {
			return;
		}

		if cfg!(debug_assertions) {
			eprintln!("[[{:?}]]", err.error);
		}

		eprintln!("[[Error (not fatal)]]\n{}", Report::new(err.error));
	});

	config.pathset(args.filtering.paths.clone());

	config.throttle(args.events.debounce.0);
	config.keyboard_events(args.events.stdin_quit || args.events.interactive);
	if let Some(interval) = args.events.poll {
		config.file_watcher(Watcher::Poll(interval.0));
	}

	if args.filtering.no_follow_symlinks {
		config.follow_symlinks(false);
	}

	let once = args.once;
	let clear = args.output.screen_clear;

	let emit_events_to = args.events.emit_events_to;
	let state = state.clone();

	if args.only_emit_events {
		config.on_action(move |mut action| {
			#[cfg(unix)]
			if action.signals().any(|sig| sig == Signal::TerminalSuspend) {
				suspend_self();
			}

			// if we got a terminate or interrupt signal, quit
			if action
				.signals()
				.any(|sig| sig == Signal::Terminate || sig == Signal::Interrupt)
			{
				// no need to be graceful as there's no commands
				action.quit();
				return action;
			}

			// clear the screen before printing events
			if let Some(mode) = clear {
				match mode {
					ClearMode::Clear => {
						clearscreen::clear().ok();
					}
					ClearMode::Reset => {
						reset_screen();
					}
				}
			}

			match emit_events_to {
				EmitEvents::Stdio => {
					println!(
						"{}",
						events_to_simple_format(action.events.as_ref()).unwrap_or_default()
					);
				}
				EmitEvents::JsonStdio => {
					for event in action.events.iter().filter(|e| !e.is_empty()) {
						println!("{}", serde_json::to_string(event).unwrap_or_default());
					}
				}
				other => unreachable!(
					"emit_events_to should have been validated earlier: {:?}",
					other
				),
			}

			action
		});

		return Ok(config);
	}

	let delay_run = args.command.delay_run.map(|ts| ts.0);
	let on_busy = args.events.on_busy_update;
	let stdin_quit = args.events.stdin_quit;
	let interactive = args.events.interactive;
	let exit_on_error = args.events.exit_on_error;

	let signal = args.events.signal;
	let stop_signal = args.command.stop_signal;
	let stop_timeout = args.command.stop_timeout.0;

	let print_events = args.logging.print_events;
	let outflags = OutputFlags {
		quiet: args.output.quiet,
		colour: match args.output.color {
			ColourMode::Auto if !std::io::stdin().is_terminal() => ColorChoice::Never,
			ColourMode::Auto => ColorChoice::Auto,
			ColourMode::Always => ColorChoice::Always,
			ColourMode::Never => ColorChoice::Never,
		},
		timings: args.output.timings,
		bell: args.output.bell,
		notify: args.output.notify,
	};

	let timeout_config = TimeoutConfig {
		timeout: args.command.timeout.map(|ts| ts.0),
		stop_signal: stop_signal.unwrap_or(Signal::Terminate),
		stop_timeout,
	};

	let workdir = Arc::new(args.command.workdir.clone());

	let add_envs: Arc<[EnvVar]> = args.command.env.clone().into();
	debug!(
		envs=?args.command.env,
		"additional environment variables to add to command"
	);

	let id = Id::default();
	let command = interpret_command_args(args)?;

	let signal_map: Arc<HashMap<Signal, Option<Signal>>> = Arc::new(
		args.events
			.signal_map
			.iter()
			.copied()
			.map(|SignalMapping { from, to }| (from, to))
			.collect(),
	);

	let queued = Arc::new(AtomicBool::new(false));
	let quit_again = Arc::new(AtomicU8::new(0));
	let paused = Arc::new(AtomicBool::new(false));
	let should_quit = Arc::new(AtomicBool::new(false));

	config.on_action_async(move |mut action| {
		let add_envs = add_envs.clone();
		let command = command.clone();
		let state = state.clone();
		let queued = queued.clone();
		let quit_again = quit_again.clone();
		let paused = paused.clone();
		let should_quit = should_quit.clone();
		let signal_map = signal_map.clone();
		let workdir = workdir.clone();
		Box::new(
			async move {
				trace!(events=?action.events, "handling action");

				let add_envs = add_envs.clone();
				let command = command.clone();
				let queued = queued.clone();
				let quit_again = quit_again.clone();
				let paused = paused.clone();
				let should_quit = should_quit.clone();
				let signal_map = signal_map.clone();
				let workdir = workdir.clone();

				trace!("set spawn hook for workdir and environment variables");
				let job = action.get_or_create_job(id, move || command.clone());
				let events = action.events.clone();
				job.set_spawn_hook({
					let state = state.clone();
					move |command, _| {
						let add_envs = add_envs.clone();
						let state = state.clone();
						let events = events.clone();

						if let Some(ref workdir) = workdir.as_ref() {
							debug!(?workdir, "set command workdir");
							command.command_mut().current_dir(workdir);
						}

						if let Some(ref socket_set) = state.socket_set {
							for env in socket_set.envs() {
								command.command_mut().env(env.key, env.value);
							}
						}

						emit_events_to_command(
							command.command_mut(),
							events,
							state,
							emit_events_to,
							add_envs,
						);
					}
				});

				let show_events = {
					let events = action.events.clone();
					move || {
						if print_events {
							trace!("print events to stderr");
							for (n, event) in events.iter().enumerate() {
								eprintln!("[EVENT {n}] {event}");
							}
						}
					}
				};

				let clear_screen = {
					let events = action.events.clone();
					move || {
						if let Some(mode) = clear {
							match mode {
								ClearMode::Clear => {
									clearscreen::clear().ok();
									debug!("cleared screen");
								}
								ClearMode::Reset => {
									reset_screen();
									debug!("hard-reset screen");
								}
							}
						}

						// re-show events after clearing
						if print_events {
							trace!("print events to stderr");
							for (n, event) in events.iter().enumerate() {
								eprintln!("[EVENT {n}] {event}");
							}
						}
					}
				};

				let quit = |mut action: ActionHandler| {
					match quit_again.fetch_add(1, Ordering::Relaxed) {
						0 => {
							if stop_timeout > Duration::ZERO
								&& action.list_jobs().any(|(_, job)| job.is_running())
							{
								eprintln!("[Waiting {stop_timeout:?} for processes to exit before stopping...]");
							}
							// eprintln!("[Waiting {stop_timeout:?} for processes to exit before stopping... Ctrl-C again to exit faster]");
							// see TODO in action/worker.rs
							action.quit_gracefully(
								stop_signal.unwrap_or(Signal::Terminate),
								stop_timeout,
							);
						}
						1 => {
							action.quit_gracefully(Signal::ForceStop, Duration::ZERO);
						}
						_ => {
							action.quit();
						}
					}

					action
				};

				// Check if we should quit due to command failure (--exit-on-error)
				if should_quit.load(Ordering::SeqCst) {
					debug!("command failed with --exit-on-error, quitting");
					return quit(action);
				}

				if once {
					debug!("debug mode: run once and quit");
					show_events();

					if let Some(delay) = delay_run {
						job.run_async(move |_| {
							Box::new(async move {
								sleep(delay).await;
							})
						});
					}

					// this blocks the event loop, but also this is a debug feature so i don't care
					job.start().await;
					let timed_out = if let Some(timeout) = timeout_config.timeout {
						tokio::select! {
							_ = job.to_wait() => false,
							_ = tokio::time::sleep(timeout) => {
								if cfg!(windows) {
									job.stop().await;
								} else {
									job.stop_with_signal(timeout_config.stop_signal, timeout_config.stop_timeout).await;
								}
								true
							}
						}
					} else {
						job.to_wait().await;
						false
					};
					job.run({
						let state = state.clone();
						move |context| {
							if let Some(end) = end_of_process(context.current, outflags, timed_out)
							{
								*state.exit_code.lock().unwrap() = ExitCode::from(
									end.into_exitstatus()
										.code()
										.unwrap_or(0)
										.try_into()
										.unwrap_or(1),
								);
							}
						}
					})
					.await;
					return quit(action);
				}

				let is_keyboard_eof = action
					.events
					.iter()
					.any(|e| e.tags.contains(&Tag::Keyboard(Keyboard::Eof)));
				if stdin_quit && is_keyboard_eof {
					debug!("keyboard EOF, quit");
					show_events();
					return quit(action);
				}

				if interactive {
					for event in action.events.iter() {
						for tag in &event.tags {
							match tag {
								Tag::Keyboard(Keyboard::Eof) => {
									debug!("interactive: Ctrl-C/D, quit");
									return quit(action);
								}
								Tag::Keyboard(Keyboard::Key { key, .. }) => match key {
									KeyCode::Char('q') => {
										debug!("interactive: quit");
										return quit(action);
									}
									KeyCode::Char('p') => {
										let was_paused = paused.fetch_xor(true, Ordering::SeqCst);
										if was_paused {
											debug!("interactive: unpause");
											eprintln!("[Unpaused]");
										} else {
											debug!("interactive: pause");
											eprintln!("[Paused]");
										}
										return action;
									}
									KeyCode::Char('s') => {
										debug!("interactive: stop");
										if cfg!(windows) {
											job.stop();
										} else {
											job.stop_with_signal(
												stop_signal.unwrap_or(Signal::Terminate),
												stop_timeout,
											);
										}
										return action;
									}
									KeyCode::Char('r') => {
										debug!("interactive: restart");
										clear_screen();
										if cfg!(windows) {
											job.restart();
										} else {
											job.restart_with_signal(
												stop_signal.unwrap_or(Signal::Terminate),
												stop_timeout,
											);
										}
										job.run({
											let job = job.clone();
											let should_quit = should_quit.clone();
											let state = state.clone();
											move |context| {
												setup_process(
													job.clone(),
													context.command.clone(),
													outflags,
													timeout_config,
													exit_on_error,
													should_quit.clone(),
													state.clone(),
												);
											}
										});
										return action;
									}
									_ => {}
								},
								_ => {}
							}
						}
					}
				}

				let signals: Vec<Signal> = action.signals().collect();
				trace!(?signals, "received some signals");

				// if we got a terminate or interrupt signal and they're not mapped, quit
				if (signals.contains(&Signal::Terminate)
					&& !signal_map.contains_key(&Signal::Terminate))
					|| (signals.contains(&Signal::Interrupt)
						&& !signal_map.contains_key(&Signal::Interrupt))
				{
					debug!("unmapped terminate or interrupt signal, quit");
					show_events();
					return quit(action);
				}

				// pass all other signals on
				for signal in signals {
					match signal_map.get(&signal) {
						Some(Some(mapped)) => {
							debug!(?signal, ?mapped, "passing mapped signal");
							job.signal(*mapped);
						}
						Some(None) => {
							debug!(?signal, "discarding signal");
						}
						None => {
							debug!(?signal, "passing signal on");
							#[cfg(unix)]
							if signal == Signal::TerminalSuspend {
								job.signal(signal).await;
								suspend_self();
							} else {
								job.signal(signal);
							}
							#[cfg(not(unix))]
							job.signal(signal);
						}
					}
				}

				// only filesystem events below here (or empty synthetic events)
				if action.paths().next().is_none()
					&& !action.events.iter().any(watchexec_events::Event::is_empty)
				{
					debug!("no filesystem or synthetic events, skip without doing more");
					show_events();
					return action;
				}

				if interactive && paused.load(Ordering::SeqCst) {
					debug!("interactive: paused, ignoring filesystem event");
					return action;
				}

				show_events();

				if let Some(delay) = delay_run {
					trace!("delaying run by sleeping inside the job");
					job.run_async(move |_| {
						Box::new(async move {
							sleep(delay).await;
						})
					});
				}

				trace!("querying job state via run_async");
				job.run_async({
					let job = job.clone();
					let should_quit = should_quit.clone();
					let state = state.clone();
					move |context| {
						let job = job.clone();
						let should_quit = should_quit.clone();
						let state = state.clone();
						let is_running = matches!(context.current, CommandState::Running { .. });
						Box::new(async move {
							let innerjob = job.clone();
							let should_quit = should_quit.clone();
							let state = state.clone();
							if is_running {
								trace!(?on_busy, "job is running, decide what to do");
								match on_busy {
									OnBusyUpdate::DoNothing => {}
									OnBusyUpdate::Signal => {
										job.signal(if cfg!(windows) {
											Signal::ForceStop
										} else {
											on_busy_signal(signal, stop_signal)
										});
									}
									OnBusyUpdate::Restart if cfg!(windows) => {
										job.restart();
										job.run({
											let should_quit = should_quit.clone();
											let state = state.clone();
											move |context| {
												clear_screen();
												setup_process(
													innerjob.clone(),
													context.command.clone(),
													outflags,
													timeout_config,
													exit_on_error,
													should_quit.clone(),
													state.clone(),
												);
											}
										});
									}
									OnBusyUpdate::Restart => {
										job.restart_with_signal(
											stop_signal.unwrap_or(Signal::Terminate),
											stop_timeout,
										);
										job.run({
											let should_quit = should_quit.clone();
											let state = state.clone();
											move |context| {
												clear_screen();
												setup_process(
													innerjob.clone(),
													context.command.clone(),
													outflags,
													timeout_config,
													exit_on_error,
													should_quit.clone(),
													state.clone(),
												);
											}
										});
									}
									OnBusyUpdate::Queue => {
										let job = job.clone();
										let already_queued =
											queued.fetch_or(true, Ordering::SeqCst);
										if already_queued {
											debug!("next start is already queued, do nothing");
										} else {
											debug!("queueing next start of job");
											tokio::spawn({
												let queued = queued.clone();
												let should_quit = should_quit.clone();
												let state = state.clone();
												async move {
													trace!("waiting for job to finish");
													job.to_wait().await;
													trace!("job finished, starting queued");
													job.start();
													job.run({
														let should_quit = should_quit.clone();
														let state = state.clone();
														move |context| {
															clear_screen();
															setup_process(
																innerjob.clone(),
																context.command.clone(),
																outflags,
																timeout_config,
																exit_on_error,
																should_quit.clone(),
																state.clone(),
															);
														}
													})
													.await;
													trace!("resetting queued state");
													queued.store(false, Ordering::SeqCst);
												}
											});
										}
									}
								}
							} else {
								trace!("job is not running, start it");
								job.start();
								job.run({
									let should_quit = should_quit.clone();
									let state = state.clone();
									move |context| {
										clear_screen();
										setup_process(
											innerjob.clone(),
											context.command.clone(),
											outflags,
											timeout_config,
											exit_on_error,
											should_quit.clone(),
											state.clone(),
										);
									}
								});
							}
						})
					}
				});

				action
			}
			.instrument(trace_span!("action handler")),
		)
	});

	Ok(config)
}

/// The signal to send in the `signal` mode of `--on-busy-update`.
///
/// `--stop-signal` documents that it's used by the `signal` mode "unless `--signal` is provided",
/// so `--signal` wins when both are given, and `--stop-signal` is only the fallback.
const fn on_busy_signal(signal: Option<Signal>, stop_signal: Option<Signal>) -> Signal {
	match (signal, stop_signal) {
		(Some(sig), _) | (None, Some(sig)) => sig,
		(None, None) => Signal::Terminate,
	}
}

/// `$SHELL` as set inside Git Bash / MSYS2 (and a plain `--shell=/usr/bin/bash`)
/// is a POSIX-style path. Windows process creation doesn't understand those,
/// so spawning would fail with "the system cannot find the path specified"
/// even though the shell is right there. Resolve it by file name via `PATH`
/// instead, the same way `pwsh`/`powershell` are already detected above.
///
/// One thing to dodge: Windows ships its own `%SystemRoot%\System32\bash.exe`,
/// a launcher stub for WSL. If that happens to come before the real Git Bash
/// on `PATH`, naively resolving "bash" would silently run the command inside
/// WSL instead of the Git Bash the user actually has, so it's skipped.
#[cfg(windows)]
fn resolve_shell_prog(prog: &str) -> String {
	let system32 = var("SystemRoot")
		.map(|root| PathBuf::from(root).join("System32"))
		.ok();
	let cwd = std::env::current_dir().unwrap_or_default();
	resolve_shell_prog_in(prog, var_os("PATH").as_deref(), system32.as_deref(), &cwd)
}

/// Does the actual work of [`resolve_shell_prog`], with the `PATH` list and
/// the WSL stub's directory passed in rather than read from the process
/// environment, so tests can exercise this without touching global state.
#[cfg(windows)]
fn resolve_shell_prog_in(
	prog: &str,
	search_path: Option<&OsStr>,
	system32: Option<&Path>,
	cwd: &Path,
) -> String {
	if !prog.starts_with('/') || Path::new(prog).exists() {
		return prog.to_string();
	}

	let Some(name) = Path::new(prog).file_name() else {
		return prog.to_string();
	};

	// The WSL launcher stub only shadows `bash`; skipping System32 for any
	// other program name would wrongly discard a legitimate candidate that
	// happens to live there.
	let is_bash = name.eq_ignore_ascii_case("bash") || name.eq_ignore_ascii_case("bash.exe");

	which::which_in_all(name, search_path, cwd)
		.ok()
		.and_then(|mut candidates| {
			candidates.find(|c| !is_bash || system32.is_none_or(|s32| c.parent() != Some(s32)))
		})
		.map_or_else(|| prog.to_string(), |p| p.to_string_lossy().into_owned())
}

#[cfg(not(windows))]
fn resolve_shell_prog(prog: &str) -> String {
	prog.to_string()
}

#[cfg(test)]
mod on_busy_signal_tests {
	use super::on_busy_signal;
	use watchexec_signals::Signal;

	/// '--stop-signal' documents that it is used by the 'signal' mode of '--on-busy-update'
	/// "(unless '--signal' is provided)", and '--signal' documents that the signal it names is the
	/// one sent "when it's still running". So '--signal' has to win over '--stop-signal'.
	#[test]
	fn signal_wins_over_stop_signal() {
		assert_eq!(
			on_busy_signal(Some(Signal::User1), Some(Signal::Hangup)),
			Signal::User1
		);
	}

	#[test]
	fn stop_signal_is_the_fallback() {
		assert_eq!(
			on_busy_signal(None, Some(Signal::Hangup)),
			Signal::Hangup,
			"--stop-signal alone should still be honoured"
		);
		assert_eq!(
			on_busy_signal(Some(Signal::User1), None),
			Signal::User1,
			"--signal alone should be used"
		);
		assert_eq!(
			on_busy_signal(None, None),
			Signal::Terminate,
			"neither given means the default"
		);
	}
}

#[cfg(test)]
mod runtime_error_sampler_tests {
	use std::time::{Duration, Instant};

	use super::RuntimeErrorSampler;

	const WINDOW: Duration = Duration::from_secs(1);

	fn should_emit(sampler: &mut RuntimeErrorSampler, message: &str, now: Instant) -> bool {
		sampler.should_emit(message.to_string(), now)
	}

	#[test]
	fn first_occurrence_emits_and_duplicate_suppresses() {
		let now = Instant::now();
		let mut sampler = RuntimeErrorSampler::new(WINDOW, 4);

		assert!(should_emit(&mut sampler, "error", now));
		assert!(!should_emit(
			&mut sampler,
			"error",
			now + Duration::from_millis(500)
		));
	}

	#[test]
	fn distinct_messages_emit_independently() {
		let now = Instant::now();
		let mut sampler = RuntimeErrorSampler::new(WINDOW, 4);

		assert!(should_emit(&mut sampler, "first", now));
		assert!(should_emit(&mut sampler, "second", now));
	}

	#[test]
	fn suppressed_occurrences_do_not_extend_the_window() {
		let now = Instant::now();
		let mut sampler = RuntimeErrorSampler::new(WINDOW, 4);

		assert!(should_emit(&mut sampler, "error", now));
		assert!(!should_emit(
			&mut sampler,
			"error",
			now + Duration::from_millis(900)
		));
		assert!(should_emit(&mut sampler, "error", now + WINDOW));
	}

	#[test]
	fn emission_starts_a_new_window() {
		let now = Instant::now();
		let mut sampler = RuntimeErrorSampler::new(WINDOW, 4);

		assert!(should_emit(&mut sampler, "error", now));
		assert!(should_emit(
			&mut sampler,
			"error",
			now + Duration::from_millis(1_100)
		));
		assert!(!should_emit(
			&mut sampler,
			"error",
			now + Duration::from_millis(1_500)
		));
		assert!(should_emit(
			&mut sampler,
			"error",
			now + Duration::from_millis(2_100)
		));
	}

	#[test]
	fn new_messages_prune_stale_entries() {
		let now = Instant::now();
		let mut sampler = RuntimeErrorSampler::new(WINDOW, 2);

		assert!(should_emit(&mut sampler, "stale", now));
		assert!(should_emit(
			&mut sampler,
			"fresh",
			now + Duration::from_millis(800)
		));
		assert!(should_emit(
			&mut sampler,
			"new",
			now + Duration::from_millis(1_100)
		));

		assert_eq!(sampler.samples.len(), 2);
		assert!(!sampler.samples.contains_key("stale"));
		assert!(sampler.samples.contains_key("fresh"));
		assert!(sampler.samples.contains_key("new"));
	}

	#[test]
	fn capacity_evicts_the_least_recently_seen_entry() {
		let now = Instant::now();
		let mut sampler = RuntimeErrorSampler::new(Duration::from_secs(10), 2);

		assert!(should_emit(&mut sampler, "first", now));
		assert!(should_emit(
			&mut sampler,
			"second",
			now + Duration::from_millis(100)
		));
		assert!(!should_emit(
			&mut sampler,
			"first",
			now + Duration::from_millis(200)
		));
		assert!(should_emit(
			&mut sampler,
			"third",
			now + Duration::from_millis(300)
		));

		assert_eq!(sampler.samples.len(), 2);
		assert!(sampler.samples.contains_key("first"));
		assert!(!sampler.samples.contains_key("second"));
		assert!(sampler.samples.contains_key("third"));
	}

	#[test]
	fn evicted_messages_fail_open() {
		let now = Instant::now();
		let mut sampler = RuntimeErrorSampler::new(Duration::from_secs(10), 1);

		assert!(should_emit(&mut sampler, "first", now));
		assert!(should_emit(
			&mut sampler,
			"second",
			now + Duration::from_millis(100)
		));
		assert!(should_emit(
			&mut sampler,
			"first",
			now + Duration::from_millis(200)
		));
		assert_eq!(sampler.samples.len(), 1);
	}

	#[test]
	fn zero_capacity_fails_open() {
		let now = Instant::now();
		let mut sampler = RuntimeErrorSampler::new(WINDOW, 0);

		assert!(should_emit(&mut sampler, "error", now));
		assert!(should_emit(&mut sampler, "error", now));
		assert!(sampler.samples.is_empty());
	}
}

#[cfg(all(test, windows))]
mod resolve_shell_prog_tests {
	use super::resolve_shell_prog_in;
	use std::{env::current_dir, fs::File, path::PathBuf};

	fn make_fake_exe(dir: &std::path::Path, name: &str) -> PathBuf {
		let path = dir.join(name);
		File::create(&path).unwrap();
		path
	}

	#[test]
	fn leaves_non_posix_prog_untouched() {
		assert_eq!(
			resolve_shell_prog_in("pwsh", None, None, &current_dir().unwrap()),
			"pwsh"
		);
		assert_eq!(
			resolve_shell_prog_in(
				r"C:\Program Files\Git\bin\bash.exe",
				None,
				None,
				&current_dir().unwrap()
			),
			r"C:\Program Files\Git\bin\bash.exe"
		);
	}

	#[test]
	fn resolves_posix_style_path_via_path_lookup() {
		let dir = tempfile::tempdir().unwrap();
		make_fake_exe(dir.path(), "myshell.exe");

		let resolved = resolve_shell_prog_in(
			"/usr/bin/myshell.exe",
			Some(dir.path().as_os_str()),
			None,
			dir.path(),
		);

		assert_eq!(resolved, dir.path().join("myshell.exe").to_string_lossy());
	}

	#[test]
	fn falls_back_to_original_when_nothing_found() {
		let dir = tempfile::tempdir().unwrap();
		// dir is empty: no "myshell.exe" anywhere on the given search path.
		let resolved = resolve_shell_prog_in(
			"/usr/bin/myshell.exe",
			Some(dir.path().as_os_str()),
			None,
			dir.path(),
		);
		assert_eq!(resolved, "/usr/bin/myshell.exe");
	}

	#[test]
	fn skips_the_wsl_launcher_stub_in_system32() {
		let root = tempfile::tempdir().unwrap();
		let system32 = root.path().join("System32");
		std::fs::create_dir(&system32).unwrap();
		make_fake_exe(&system32, "bash.exe");

		let real_git_bin = root.path().join("Git").join("bin");
		std::fs::create_dir_all(&real_git_bin).unwrap();
		let real_bash = make_fake_exe(&real_git_bin, "bash.exe");

		// System32 comes first on PATH, same as it typically does on a real
		// Windows install, so a naive lookup would find the WSL stub first.
		let search_path = format!("{};{}", system32.display(), real_git_bin.display());

		let resolved = resolve_shell_prog_in(
			"/usr/bin/bash",
			Some(std::ffi::OsStr::new(&search_path)),
			Some(&system32),
			root.path(),
		);

		assert_eq!(resolved, real_bash.to_string_lossy());
	}

	#[test]
	fn does_not_skip_system32_for_non_bash_programs() {
		let root = tempfile::tempdir().unwrap();
		let system32 = root.path().join("System32");
		std::fs::create_dir(&system32).unwrap();
		let real_tool = make_fake_exe(&system32, "sometool.exe");

		let resolved = resolve_shell_prog_in(
			"/usr/bin/sometool",
			Some(system32.as_os_str()),
			Some(&system32),
			root.path(),
		);

		// Unlike bash, sometool.exe has no WSL-stub problem, so a legitimate
		// candidate living in System32 must still be found.
		assert_eq!(resolved, real_tool.to_string_lossy());
	}
}

#[instrument(level = "debug")]
fn interpret_command_args(args: &Args) -> Result<Arc<Command>> {
	let mut cmd = args.program.clone();
	assert!(!cmd.is_empty(), "(clap) Bug: command is not present");

	let shell = if args.command.no_shell {
		None
	} else {
		// surrounding whitespace is part of neither the shell program nor its options, and
		// a whitespace-only value is an empty shell, handled below like an empty string
		let shell = args
			.command
			.shell
			.clone()
			.or_else(|| var("SHELL").ok())
			.map(|shell| shell.trim().to_owned());
		match shell
			.as_deref()
			.or_else(|| {
				if cfg!(not(windows)) {
					Some("sh")
				} else if var("POWERSHELL_DISTRIBUTION_CHANNEL").is_ok()
					&& (which::which("pwsh").is_ok() || which::which("pwsh.exe").is_ok())
				{
					trace!("detected pwsh");
					Some("pwsh")
				} else if var("PSModulePath").is_ok()
					&& (which::which("powershell").is_ok()
						|| which::which("powershell.exe").is_ok())
				{
					trace!("detected powershell");
					Some("powershell")
				} else {
					Some("cmd")
				}
			})
			.or(Some("default"))
		{
			Some("") => return Err(RuntimeError::CommandShellEmptyShell).into_diagnostic(),

			Some("none") | None => None,

			#[cfg(windows)]
			Some("cmd") | Some("cmd.exe") | Some("CMD") | Some("CMD.EXE") => {
				Some(Shell::cmd_with_quoting(args.command.quote))
			}

			Some(other) => {
				let (shell_path, shell_options) = parse_path(other)?;

				Some(Shell {
					prog: shell_path.into(),
					options: shell_options,
					program_option: Some(Cow::Borrowed(OsStr::new("-c"))),
					quote: args.command.quote,
				})
			}
		}
	};

	let program = if let Some(shell) = shell {
		Program::Shell {
			shell,
			command: cmd.join(" "),
			args: Vec::new(),
		}
	} else {
		Program::Exec {
			prog: cmd.remove(0).into(),
			args: cmd,
		}
	};

	Ok(Arc::new(Command {
		program,
		options: SpawnOptions {
			grouped: matches!(args.command.wrap_process, WrapMode::Group),
			session: matches!(args.command.wrap_process, WrapMode::Session),
			..Default::default()
		},
	}))
}

/// Returns a [`(String, Vec<String>)`] containing the parsed shell path and optional shell arguments, respectively.
/// This logic handles potential spaces in the shell path as well as potential arguments by splitting shell_path by whitespace,
/// then iterating over the vector backwards by splitting the last item off of the vector.
/// For each item we concatenate the remaining items into a single string, and check if it resolves to a valid path.
/// If it doesn't, then the next item should also be considered an argument.
///
/// This scenario is possible on Windows when users use git-bash which is located in "C:/Program Files/Git/usr/bin/bash.exe".
/// This scenario is also theoretically possible in Linux if a user, though unlikely.
/// Due to it theoretically being possible on Linux, and an unknown number of other use cases on Windows, this logic was written to be capable of handling many such inputs such as :
/// - "/bin/bash"
/// - "/bin/bash --arg1 --arg2"
/// - "C:/Program Files/Git/usr/bin/bash.exe"
/// - "C:/Program Files/Git/usr/bin/bash.exe --arg1 --arg2"
fn parse_path(original_path: &str) -> Result<(String, Vec<String>)> {
	parse_path_with(original_path, resolve_shell_prog)
}

/// Does the actual work of [`parse_path`], with the POSIX-path resolution passed
/// in rather than called directly, so tests can exercise the order of the two
/// without depending on which shells happen to be installed.
fn parse_path_with(
	original_path: &str,
	resolve: impl Fn(&str) -> String,
) -> Result<(String, Vec<String>)> {
	debug!(?original_path, "before parsing for potential shell options");

	let mut shell_path = original_path.to_string();
	let mut shell_options: Vec<String> = vec![];

	if shell_path.contains(' ') && !Path::new(&shell_path).exists() {
		let split_shell_parts = shell_path
			.split_ascii_whitespace()
			.map(String::from)
			.collect::<Vec<_>>();

		debug!(
			?split_shell_parts,
			"after splitting original shell path by whitespace"
		);

		let (mut last_shell_part, mut remaining_shell_parts) =
			split_shell_parts.split_last().unwrap();

		shell_path = remaining_shell_parts.join(" ");
		shell_options.push(last_shell_part.clone());

		while !Path::new(&shell_path).exists() {
			(last_shell_part, remaining_shell_parts) = remaining_shell_parts.split_last().unwrap();

			if remaining_shell_parts.is_empty() {
				// If we get here and the path still isn't valid, it doesn't really matter what we do.
				break;
			}

			shell_options.push(last_shell_part.clone());
			shell_path = remaining_shell_parts.join(" ");
		}
	}

	// Reversing the shell options because we add them in reverse order.
	shell_options.reverse();

	debug!(?shell_path, "after splitting options");
	debug!(?shell_options, "after splitting options");

	// Git Bash / MSYS2 set $SHELL to a POSIX path, which Windows can't spawn
	// and which neither the exists() check nor `which` below can resolve. Map
	// it back to the native path first, so the validity check sees the shell
	// the user actually has. No-op everywhere else.
	shell_path = resolve(&shell_path);

	let mut is_valid = Path::new(&shell_path).exists();

	if !is_valid {
		// It's possible that at this point "shell_path" is something like "pwsh" which while not a valid path,
		// is runnable when passed into a program as is if "which" is able to resolve it from the path.
		is_valid = which::which(&shell_path).is_ok();
	}

	if !is_valid {
		return Err(miette::miette!(
			"the parsed shell path \"{shell_path}\" does not exist on this system, and could not be resolved from your PATH"
		));
	}

	Ok((shell_path, shell_options))
}

#[cfg(test)]
fn test_parse_path_for_shell(expected_path: &str) {
	if Path::new(expected_path).exists() {
		let expected_empty_args: Vec<String> = vec![];
		let expected_args = ["--arg1", "--arg2"];
		let (mut actual_path, mut actual_args): (String, Vec<String>);

		(actual_path, actual_args) = parse_path(expected_path)
			.unwrap_or_else(|error| panic!("failed to parse shell {expected_path:?}: {error}"));

		assert_eq!(actual_path, *expected_path);
		assert_eq!(actual_args, expected_empty_args);

		(actual_path, actual_args) =
			parse_path(&format!("{} {}", expected_path, expected_args.join(" ")))
				.unwrap_or_else(|error| panic!("failed to parse shell {expected_path:?}: {error}"));

		assert_eq!(actual_path, *expected_path);
		assert_eq!(actual_args, expected_args);
	}
}

/// A POSIX shell path can't pass the validity check at the end of [`parse_path`]
/// on Windows, so the resolution has to happen before it rather than at the call
/// site: with the two the other way round the check is reached with the original
/// path and kills the process.
#[test]
#[cfg(test)]
fn test_parse_path_resolves_before_validating() {
	let native = std::env::current_exe()
		.unwrap()
		.to_string_lossy()
		.into_owned();

	let (path, options) = parse_path_with("/usr/bin/bash", |_| native.clone())
		.unwrap_or_else(|error| panic!("failed to parse shell: {error}"));
	assert_eq!(path, native);
	assert!(options.is_empty());

	let (path, options) = parse_path_with("/usr/bin/bash --norc", |p| {
		assert_eq!(p, "/usr/bin/bash");
		native.clone()
	})
	.unwrap_or_else(|error| panic!("failed to parse shell: {error}"));
	assert_eq!(path, native);
	assert_eq!(options, ["--norc"]);
}

#[test]
#[cfg(test)]
fn test_shell_spec_whitespace_is_trimmed() {
	use clap::Parser;

	for spec in ["sh ", " sh", "  sh  "] {
		let args = Args::parse_from(["watchexec", &format!("--shell={spec}"), "echo", "hi"]);
		let cmd = interpret_command_args(&args)
			.unwrap_or_else(|err| panic!("--shell={spec:?} should be usable, got {err:?}"));
		let Program::Shell { shell, .. } = &cmd.program else {
			panic!(
				"--shell={spec:?} should give a shelled program, got {:?}",
				cmd.program
			);
		};
		assert_eq!(shell.prog, PathBuf::from("sh"), "for --shell={spec:?}");
		assert!(
			shell.options.is_empty(),
			"for --shell={spec:?}: whitespace should not become a shell option, got {:?}",
			shell.options
		);
	}

	// A whitespace-only shell is an empty shell, not a program named "".
	let args = Args::parse_from(["watchexec", "--shell=  ", "echo", "hi"]);
	assert!(
		interpret_command_args(&args).is_err(),
		"a whitespace-only --shell should error like an empty one"
	);
}

/// A shell spec that is neither an existing path nor resolvable from PATH used
/// to assert (hard panic) in parse_path; it must instead surface as a regular
/// configuration error, same as an empty shell.
#[test]
#[cfg(test)]
fn test_shell_spec_that_cannot_resolve_is_an_error_not_a_panic() {
	use clap::Parser;

	for spec in ["notashell", "@bash", "/usr/bin/definitely-not-a-shell-xyz"] {
		let args = Args::parse_from(["watchexec", &format!("--shell={spec}"), "echo", "hi"]);
		assert!(
			interpret_command_args(&args).is_err(),
			"--shell={spec:?} should be a configuration error, not a panic"
		);
	}
}

#[test]
#[cfg(test)]
fn test_parse_path() {
	#[cfg(windows)]
	{
		test_parse_path_for_shell("C:\\Program Files\\Git\\usr\\bin\\bash.exe");
		test_parse_path_for_shell("C:\\Program Files\\nu\\bin\\nu.exe");
		// This is the newer PowerShell 7 which does not ship with Windows.
		test_parse_path_for_shell("C:\\Program Files\\PowerShell\\7\\pwsh.exe");
		test_parse_path_for_shell("C:\\Windows\\System32\\cmd.exe");
		// This is the older PowerShell 5.1 which ships with Windows.
		test_parse_path_for_shell("C:\\Windows\\System32\\WindowsPowerShell\\v1.0\\powershell.exe");
		// This is PowerShell 5.1 (x86).
		test_parse_path_for_shell("C:\\Windows\\SysWOW64\\WindowsPowerShell\\v1.0\\powershell.exe");
	}

	#[cfg(unix)]
	{
		test_parse_path_for_shell("/bin/bash");
		test_parse_path_for_shell("/bin/sh");
		test_parse_path_for_shell("/usr/bin/nu");
	}
}

#[instrument(level = "trace")]
fn setup_process(
	job: Job,
	command: Arc<Command>,
	outflags: OutputFlags,
	timeout_config: TimeoutConfig,
	exit_on_error: bool,
	should_quit: Arc<AtomicBool>,
	state: State,
) {
	if outflags.notify.is_some_and(|m| m.on_start()) {
		Notification::new()
			.summary("Watchexec: change detected")
			.body(&format!("Running {command}"))
			.show()
			.map_or_else(
				|err| {
					eprintln!("[[Failed to send desktop notification: {err}]]");
				},
				drop,
			);
	}

	if !outflags.quiet {
		let mut stderr = StandardStream::stderr(outflags.colour);
		stderr.reset().ok();
		stderr
			.set_color(ColorSpec::new().set_fg(Some(Color::Green)))
			.ok();
		writeln!(&mut stderr, "[Running: {command}]").ok();
		stderr.reset().ok();
	}

	let send_quit_event = Arc::new(AtomicBool::new(false));
	tokio::spawn({
		let send_quit_event = send_quit_event.clone();
		let state_for_event = state.clone();
		async move {
			let timed_out = if let Some(timeout) = timeout_config.timeout {
				tokio::select! {
					_ = job.to_wait() => false,
					_ = tokio::time::sleep(timeout) => {
						if cfg!(windows) {
							job.stop().await;
						} else {
							job.stop_with_signal(timeout_config.stop_signal, timeout_config.stop_timeout).await;
						}
						true
					}
				}
			} else {
				job.to_wait().await;
				false
			};

			job.run({
				let send_quit_event = send_quit_event.clone();
				move |context| {
					if let Some(status) = end_of_process(context.current, outflags, timed_out) {
						// Store exit code in state
						*state.exit_code.lock().unwrap() = ExitCode::from(
							status
								.into_exitstatus()
								.code()
								.unwrap_or(0)
								.try_into()
								.unwrap_or(1),
						);

						// If exit_on_error is enabled and command failed, signal quit
						if exit_on_error && !matches!(status, ProcessEnd::Success) {
							debug!("command failed, setting should_quit flag for --exit-on-error");
							should_quit.store(true, Ordering::SeqCst);
							send_quit_event.store(true, Ordering::SeqCst);
						}
					}
				}
			})
			.await;

			// Send a synthetic event to trigger the action handler to check should_quit
			// This ensures we quit immediately instead of waiting for the next file event
			if send_quit_event.load(Ordering::SeqCst) {
				if let Some(wx) = state_for_event.watchexec.get() {
					debug!("sending synthetic event to trigger quit");
					if let Err(e) = wx.send_event(Event::default(), Priority::Urgent).await {
						error!("failed to send synthetic quit event: {e}");
					}
				}
			}
		}
	});
}

fn format_duration(duration: Duration) -> impl fmt::Display {
	fmt::from_fn(move |f| {
		let secs = duration.as_secs();
		if secs > 0 {
			write!(f, "{secs}s")
		} else {
			write!(f, "{}ms", duration.subsec_millis())
		}
	})
}

#[instrument(level = "trace")]
fn end_of_process(
	state: &CommandState,
	outflags: OutputFlags,
	timed_out: bool,
) -> Option<ProcessEnd> {
	let CommandState::Finished {
		status,
		started,
		finished,
	} = state
	else {
		return None;
	};

	let duration = *finished - *started;
	let duration_display = format_duration(duration);
	let timing = if outflags.timings {
		format!(", lasted {duration_display}")
	} else {
		String::new()
	};

	// Show timeout message and return early - no need for redundant status message
	if timed_out {
		if outflags.notify.is_some_and(|m| m.on_end()) {
			Notification::new()
				.summary("Watchexec: command timed out")
				.body(&format!("Command timed out after {duration_display}"))
				.show()
				.map_or_else(
					|err| {
						eprintln!("[[Failed to send desktop notification: {err}]]");
					},
					drop,
				);
		}

		if !outflags.quiet {
			let mut stderr = StandardStream::stderr(outflags.colour);
			stderr.reset().ok();
			stderr
				.set_color(ColorSpec::new().set_fg(Some(Color::Yellow)))
				.ok();
			writeln!(&mut stderr, "[Command timed out after {duration_display}]").ok();
			stderr.reset().ok();
		}

		if outflags.bell {
			let mut stdout = std::io::stdout();
			stdout.write_all(b"\x07").ok();
			stdout.flush().ok();
		}

		return Some(*status);
	}

	let (msg, fg) = match status {
		ProcessEnd::ExitError(code) => (format!("Command exited with {code}{timing}"), Color::Red),
		ProcessEnd::ExitSignal(sig) => {
			(format!("Command killed by {sig:?}{timing}"), Color::Magenta)
		}
		ProcessEnd::ExitStop(sig) => (format!("Command stopped by {sig:?}{timing}"), Color::Blue),
		ProcessEnd::Continued => (format!("Command continued{timing}"), Color::Cyan),
		ProcessEnd::Exception(ex) => (
			format!("Command ended by exception {ex:#x}{timing}"),
			Color::Yellow,
		),
		ProcessEnd::Success => (format!("Command was successful{timing}"), Color::Green),
	};

	if outflags.notify.is_some_and(|m| m.on_end()) {
		Notification::new()
			.summary("Watchexec: command ended")
			.body(&msg)
			.show()
			.map_or_else(
				|err| {
					eprintln!("[[Failed to send desktop notification: {err}]]");
				},
				drop,
			);
	}

	if !outflags.quiet {
		let mut stderr = StandardStream::stderr(outflags.colour);
		stderr.reset().ok();
		stderr.set_color(ColorSpec::new().set_fg(Some(fg))).ok();
		writeln!(&mut stderr, "[{msg}]").ok();
		stderr.reset().ok();
	}

	if outflags.bell {
		let mut stdout = std::io::stdout();
		stdout.write_all(b"\x07").ok();
		stdout.flush().ok();
	}

	Some(*status)
}

#[instrument(level = "trace")]
fn emit_events_to_command(
	command: &mut TokioCommand,
	events: Arc<[Event]>,
	state: State,
	emit_events_to: EmitEvents,
	add_envs: Arc<[EnvVar]>,
) {
	use crate::emits::{emits_to_environment, emits_to_file, emits_to_json_file};

	let mut stdin = None;

	let add_envs = add_envs.clone();
	let mut envs = Box::new(add_envs.into_iter().cloned()) as Box<dyn Iterator<Item = EnvVar>>;

	match emit_events_to {
		EmitEvents::Environment => {
			envs = Box::new(envs.chain(emits_to_environment(&events)));
		}
		EmitEvents::Stdio => match emits_to_file(&state.emit_file, &events)
			.and_then(|path| File::open(path).into_diagnostic())
		{
			Ok(file) => {
				stdin.replace(Stdio::from(file));
			}
			Err(err) => {
				error!("Failed to write events to stdin, continuing without it: {err}");
			}
		},
		EmitEvents::File => match emits_to_file(&state.emit_file, &events) {
			Ok(path) => {
				envs = Box::new(envs.chain(once(EnvVar {
					key: "WATCHEXEC_EVENTS_FILE".into(),
					value: path.into(),
				})));
			}
			Err(err) => {
				error!("Failed to write WATCHEXEC_EVENTS_FILE, continuing without it: {err}");
			}
		},
		EmitEvents::JsonStdio => match emits_to_json_file(&state.emit_file, &events)
			.and_then(|path| File::open(path).into_diagnostic())
		{
			Ok(file) => {
				stdin.replace(Stdio::from(file));
			}
			Err(err) => {
				error!("Failed to write events to stdin, continuing without it: {err}");
			}
		},
		EmitEvents::JsonFile => match emits_to_json_file(&state.emit_file, &events) {
			Ok(path) => {
				envs = Box::new(envs.chain(once(EnvVar {
					key: "WATCHEXEC_EVENTS_FILE".into(),
					value: path.into(),
				})));
			}
			Err(err) => {
				error!("Failed to write WATCHEXEC_EVENTS_FILE, continuing without it: {err}");
			}
		},
		EmitEvents::None => {}
	}

	for var in envs {
		debug!(?var, "inserting environment variable");
		command.env(var.key, var.value);
	}

	if let Some(stdin) = stdin {
		debug!("set command stdin");
		command.stdin(stdin);
	}
}

pub fn reset_screen() {
	for cs in [
		ClearScreen::WindowsCooked,
		ClearScreen::VtLeaveAlt,
		ClearScreen::VtWellDone,
		ClearScreen::default(),
	] {
		cs.clear().ok();
	}
}

#[cfg(unix)]
fn suspend_self() {
	// SAFETY: raise takes a valid signal number and reports failure by return
	// value, which is checked. SIGSTOP cannot be caught or ignored, so the
	// process stops here and resumes on SIGCONT; no state is expected to
	// survive across the stop.
	if unsafe { libc::raise(libc::SIGSTOP) } != 0 {
		warn!(error = ?std::io::Error::last_os_error(), "failed to suspend watchexec");
	}
}