use std::{
ffi::OsString,
fs,
path::PathBuf,
process::{Child, Command, ExitStatus, Stdio},
thread,
time::Duration,
};
use tracing::{debug, info, warn};
use crate::{
analysis::{CommandAdapterId, CommandAnalysisStatus, SideEffectProfile},
error::{Result, WithWatchError},
snapshot::{capture_snapshot, ChangeDetectionMode, CommandSource, SnapshotState, WatchInput},
watch::{CollectedEvents, WatchLoop},
};
const DEFAULT_POLL_TIMEOUT: Duration = Duration::from_millis(50);
const DEFAULT_DEBOUNCE_WINDOW: Duration = Duration::from_millis(200);
const WITH_WATCH_TEST_RUN_MARKER_DIR_ENV: &str = "WITH_WATCH_TEST_RUN_MARKER_DIR";
#[derive(Debug, Clone)]
pub struct ExecutionPlan {
pub source: CommandSource,
pub detection_mode: ChangeDetectionMode,
pub inputs: Vec<WatchInput>,
pub delegated_command: DelegatedCommand,
pub metadata: ExecutionMetadata,
}
impl ExecutionPlan {
pub fn passthrough(
argv: Vec<OsString>,
inputs: Vec<WatchInput>,
detection_mode: ChangeDetectionMode,
metadata: ExecutionMetadata,
) -> Self {
Self {
source: CommandSource::Argv,
detection_mode,
inputs,
delegated_command: DelegatedCommand::Argv(argv),
metadata,
}
}
pub fn shell(
expression: String,
inputs: Vec<WatchInput>,
detection_mode: ChangeDetectionMode,
metadata: ExecutionMetadata,
) -> Self {
Self {
source: CommandSource::Shell,
detection_mode,
inputs,
delegated_command: DelegatedCommand::Shell(expression),
metadata,
}
}
pub fn exec(
argv: Vec<OsString>,
inputs: Vec<WatchInput>,
detection_mode: ChangeDetectionMode,
metadata: ExecutionMetadata,
) -> Self {
Self {
source: CommandSource::Exec,
detection_mode,
inputs,
delegated_command: DelegatedCommand::Argv(argv),
metadata,
}
}
}
#[derive(Debug, Clone)]
pub struct ExecutionMetadata {
pub adapter_ids: Vec<CommandAdapterId>,
pub fallback_used: bool,
pub default_watch_root_used: bool,
pub filtered_output_count: usize,
pub side_effect_profile: SideEffectProfile,
pub status: CommandAnalysisStatus,
}
impl ExecutionMetadata {
pub fn adapter_field(&self) -> String {
self.adapter_ids
.iter()
.map(|adapter| adapter.as_str())
.collect::<Vec<_>>()
.join(",")
}
}
#[derive(Debug, Clone)]
pub enum DelegatedCommand {
Argv(Vec<OsString>),
Shell(String),
}
impl DelegatedCommand {
fn spawn_log_summary(&self) -> DelegatedCommandLogSummary {
match self {
Self::Argv(argv) => {
let program_name = argv
.first()
.map(program_name)
.unwrap_or_else(|| "<missing>".to_string());
DelegatedCommandLogSummary {
execution_kind: "argv",
program_name,
arg_count: argv.len().saturating_sub(1),
}
}
Self::Shell(_) => DelegatedCommandLogSummary {
execution_kind: "shell",
program_name: "sh".to_string(),
arg_count: 2,
},
}
}
fn display_name(&self) -> String {
match self {
Self::Argv(argv) => argv
.iter()
.map(|value| value.to_string_lossy().into_owned())
.collect::<Vec<_>>()
.join(" "),
Self::Shell(expression) => expression.clone(),
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
struct DelegatedCommandLogSummary {
execution_kind: &'static str,
program_name: String,
arg_count: usize,
}
#[derive(Debug, Clone, Copy)]
pub struct RunnerOptions {
pub debounce_window: Duration,
pub poll_timeout: Duration,
pub max_runs: Option<usize>,
}
impl Default for RunnerOptions {
fn default() -> Self {
Self {
debounce_window: DEFAULT_DEBOUNCE_WINDOW,
poll_timeout: DEFAULT_POLL_TIMEOUT,
max_runs: None,
}
}
}
impl RunnerOptions {
pub fn from_environment() -> Self {
let mut options = Self::default();
if let Ok(raw_max_runs) = std::env::var("WITH_WATCH_TEST_MAX_RUNS") {
if let Ok(parsed) = raw_max_runs.parse::<usize>() {
options.max_runs = Some(parsed);
}
}
if let Ok(raw_debounce_ms) = std::env::var("WITH_WATCH_TEST_DEBOUNCE_MS") {
if let Ok(parsed) = raw_debounce_ms.parse::<u64>() {
options.debounce_window = Duration::from_millis(parsed);
}
}
options
}
}
pub fn run(plan: ExecutionPlan, options: RunnerOptions) -> Result<i32> {
let mut watch_loop = WatchLoop::new(&plan.inputs)?;
let mut baseline = capture_snapshot(&plan.inputs, plan.detection_mode)?;
let mut child = Some(spawn_command(&plan.delegated_command)?);
let mut completed_runs = 0usize;
let mut pending_rerun = false;
let mut suppressed_self_change_snapshot = None::<SnapshotState>;
info!(
command_source = plan.source.as_str(),
detection_mode = plan.detection_mode.as_str(),
input_count = plan.inputs.len(),
adapter_id = plan.metadata.adapter_field(),
fallback_used = plan.metadata.fallback_used,
default_watch_root_used = plan.metadata.default_watch_root_used,
filtered_output_count = plan.metadata.filtered_output_count,
side_effect_profile = plan.metadata.side_effect_profile.as_str(),
analysis_status = plan.metadata.status.as_str(),
"Starting with-watch run loop"
);
loop {
if let Some(active_child) = child.as_mut() {
if let Some(status) =
active_child
.try_wait()
.map_err(|source| WithWatchError::Wait {
command: plan.delegated_command.display_name(),
source,
})?
{
completed_runs += 1;
let last_exit_code = exit_code_from_status(status);
let post_run_snapshot = capture_snapshot(&plan.inputs, plan.detection_mode)?;
let inputs_changed_since_baseline =
post_run_snapshot.is_meaningfully_different(&baseline, plan.detection_mode);
let additional_change_after_suppression = suppressed_self_change_snapshot
.as_ref()
.is_some_and(|snapshot| {
post_run_snapshot.is_meaningfully_different(snapshot, plan.detection_mode)
});
let should_rerun = if plan.metadata.side_effect_profile
== SideEffectProfile::WritesWatchedInputs
{
pending_rerun || additional_change_after_suppression
} else {
pending_rerun && inputs_changed_since_baseline
};
if pending_rerun
&& plan.metadata.side_effect_profile == SideEffectProfile::WritesWatchedInputs
{
debug!(
rerun_queued = true,
side_effect_profile = plan.metadata.side_effect_profile.as_str(),
"Queued rerun after additional changes during self-mutating command \
activity"
);
} else if additional_change_after_suppression
&& plan.metadata.side_effect_profile == SideEffectProfile::WritesWatchedInputs
{
debug!(
rerun_queued = true,
side_effect_profile = plan.metadata.side_effect_profile.as_str(),
"Queued rerun because post-run state diverged from the suppressed \
self-change snapshot"
);
} else if suppressed_self_change_snapshot.is_some()
&& plan.metadata.side_effect_profile == SideEffectProfile::WritesWatchedInputs
{
debug!(
rerun_suppressed = true,
side_effect_profile = plan.metadata.side_effect_profile.as_str(),
"Suppressing rerun after self-mutating command activity"
);
}
baseline = post_run_snapshot;
pending_rerun = false;
suppressed_self_change_snapshot = None;
child = None;
write_test_run_marker(completed_runs);
info!(
completed_runs,
last_exit_code,
command_source = plan.source.as_str(),
rerun_queued = should_rerun,
"Delegated command finished"
);
if options
.max_runs
.is_some_and(|limit| completed_runs >= limit)
{
return Ok(last_exit_code);
}
if should_rerun {
child = Some(spawn_command(&plan.delegated_command)?);
continue;
}
}
}
if let Some(events) =
watch_loop.collect_events(options.poll_timeout, options.debounce_window)
{
handle_watch_events(&events);
let current_snapshot = capture_snapshot(&plan.inputs, plan.detection_mode)?;
let reference_snapshot = if child.is_some()
&& plan.metadata.side_effect_profile == SideEffectProfile::WritesWatchedInputs
{
suppressed_self_change_snapshot
.as_ref()
.unwrap_or(&baseline)
} else {
&baseline
};
if current_snapshot.is_meaningfully_different(reference_snapshot, plan.detection_mode) {
debug!(
event_count = events.event_count,
path_count = events.path_count,
child_running = child.is_some(),
"Observed meaningful input changes"
);
if child.is_some() {
if plan.metadata.side_effect_profile == SideEffectProfile::WritesWatchedInputs
&& suppressed_self_change_snapshot.is_none()
{
suppressed_self_change_snapshot = Some(current_snapshot);
debug!(
rerun_suppressed = true,
side_effect_profile = plan.metadata.side_effect_profile.as_str(),
"Suppressed the first in-run snapshot change for a self-mutating \
command"
);
} else {
pending_rerun = true;
}
} else {
baseline = current_snapshot;
child = Some(spawn_command(&plan.delegated_command)?);
}
} else if child.is_some() {
debug!(
rerun_suppressed = true,
"Ignored non-meaningful filesystem churn"
);
}
} else if child.is_none() {
thread::sleep(Duration::from_millis(10));
}
}
}
fn handle_watch_events(events: &CollectedEvents) {
if events.error_count > 0 {
warn!(
error_count = events.error_count,
event_count = events.event_count,
path_count = events.path_count,
"Watcher reported recoverable errors; forcing a rescan"
);
} else {
debug!(
event_count = events.event_count,
path_count = events.path_count,
"Collected filesystem events"
);
}
}
fn spawn_command(command: &DelegatedCommand) -> Result<Child> {
log_delegated_command_spawn(command);
match command {
DelegatedCommand::Argv(argv) => spawn_argv(argv),
DelegatedCommand::Shell(expression) => spawn_shell(expression),
}
}
fn log_delegated_command_spawn(command: &DelegatedCommand) {
let summary = command.spawn_log_summary();
info!(
execution_kind = summary.execution_kind,
program = summary.program_name,
arg_count = summary.arg_count,
"Spawning delegated command"
);
}
fn spawn_argv(argv: &[OsString]) -> Result<Child> {
let program = argv
.first()
.cloned()
.ok_or(WithWatchError::MissingCommand)?;
Command::new(&program)
.args(argv.iter().skip(1))
.stdin(Stdio::inherit())
.stdout(Stdio::inherit())
.stderr(Stdio::inherit())
.spawn()
.map_err(|source| WithWatchError::Spawn {
command: program.to_string_lossy().into_owned(),
source,
})
}
fn spawn_shell(expression: &str) -> Result<Child> {
#[cfg(not(unix))]
{
let _ = expression;
Err(WithWatchError::UnsupportedShellPlatform)
}
#[cfg(unix)]
{
Command::new("/bin/sh")
.arg("-c")
.arg(expression)
.stdin(Stdio::inherit())
.stdout(Stdio::inherit())
.stderr(Stdio::inherit())
.spawn()
.map_err(|source| WithWatchError::Spawn {
command: expression.to_string(),
source,
})
}
}
fn program_name(program: &OsString) -> String {
std::path::Path::new(program)
.file_name()
.unwrap_or(program.as_os_str())
.to_string_lossy()
.into_owned()
}
fn exit_code_from_status(status: ExitStatus) -> i32 {
status.code().unwrap_or(1)
}
fn write_test_run_marker(completed_runs: usize) {
let Ok(marker_dir) = std::env::var(WITH_WATCH_TEST_RUN_MARKER_DIR_ENV) else {
return;
};
let marker_path = PathBuf::from(marker_dir).join(format!("run-{completed_runs}.done"));
if let Some(parent) = marker_path.parent() {
if let Err(error) = fs::create_dir_all(parent) {
warn!(
path = parent.display().to_string(),
%error,
"Failed to create test run marker directory"
);
return;
}
}
if let Err(error) = fs::write(&marker_path, completed_runs.to_string()) {
warn!(
path = marker_path.display().to_string(),
%error,
"Failed to write test run marker"
);
}
}
#[cfg(test)]
mod tests {
use std::{
ffi::OsString,
io::{self, Write},
sync::{Arc, Mutex},
};
use tracing::Level;
use super::{log_delegated_command_spawn, DelegatedCommand};
#[test]
fn argv_spawn_logging_omits_argument_values() {
let output = capture_logs(|| {
log_delegated_command_spawn(&DelegatedCommand::Argv(vec![
OsString::from("env"),
OsString::from("TOKEN=secret"),
OsString::from("cmd"),
]));
});
assert!(output.contains("execution_kind=\"argv\""));
assert!(output.contains("program=\"env\""));
assert!(output.contains("arg_count=2"));
assert!(!output.contains("TOKEN=secret"));
assert!(!output.contains("cmd"));
}
#[test]
fn shell_spawn_logging_omits_expression_text() {
let output = capture_logs(|| {
log_delegated_command_spawn(&DelegatedCommand::Shell(
"TOKEN=secret grep -f patterns.txt file.txt".to_string(),
));
});
assert!(output.contains("execution_kind=\"shell\""));
assert!(output.contains("program=\"sh\""));
assert!(output.contains("arg_count=2"));
assert!(!output.contains("TOKEN=secret"));
assert!(!output.contains("patterns.txt"));
}
fn capture_logs(callback: impl FnOnce()) -> String {
let buffer = Arc::new(Mutex::new(Vec::new()));
let writer = SharedWriter(buffer.clone());
let subscriber = tracing_subscriber::fmt()
.with_ansi(false)
.with_target(false)
.with_level(false)
.without_time()
.with_max_level(Level::INFO)
.with_writer(move || writer.clone())
.finish();
tracing::subscriber::with_default(subscriber, callback);
let output = buffer.lock().expect("lock buffer").clone();
String::from_utf8(output).expect("utf8 log output")
}
#[derive(Clone)]
struct SharedWriter(Arc<Mutex<Vec<u8>>>);
impl Write for SharedWriter {
fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
self.0
.lock()
.expect("lock log buffer")
.extend_from_slice(buf);
Ok(buf.len())
}
fn flush(&mut self) -> io::Result<()> {
Ok(())
}
}
}