1use std::io::{Read, Write};
11use std::path::{Path, PathBuf};
12use std::process::{Command, Stdio};
13use std::sync::atomic::{AtomicBool, Ordering};
14use std::sync::{Arc, Mutex};
15use std::time::{Duration, Instant};
16
17use crate::error::{Result, SparError};
18
19pub const DEFAULT_TIMEOUT_SECS: u64 = 3600;
26
27const _: () = assert!(DEFAULT_TIMEOUT_SECS >= 3600);
31
32#[derive(Debug, Clone)]
33pub struct ExecOpts {
34 pub cwd: Option<PathBuf>,
35 pub timeout: Duration,
36 pub check: bool,
38 pub env: Vec<(String, String)>,
39 pub stdin: Option<String>,
40 pub stop_descendants: bool,
44}
45
46impl Default for ExecOpts {
47 fn default() -> Self {
48 Self {
49 cwd: None,
50 timeout: Duration::from_secs(DEFAULT_TIMEOUT_SECS),
51 check: true,
52 env: Vec::new(),
53 stdin: None,
54 stop_descendants: false,
55 }
56 }
57}
58
59impl ExecOpts {
60 pub fn new() -> Self {
61 Self::default()
62 }
63
64 pub fn cwd(mut self, path: impl AsRef<Path>) -> Self {
65 self.cwd = Some(path.as_ref().to_path_buf());
66 self
67 }
68
69 pub fn cwd_opt(mut self, path: Option<PathBuf>) -> Self {
70 self.cwd = path;
71 self
72 }
73
74 pub fn timeout_secs(mut self, secs: u64) -> Self {
75 self.timeout = Duration::from_secs(secs);
76 self
77 }
78
79 pub fn check(mut self, value: bool) -> Self {
80 self.check = value;
81 self
82 }
83
84 pub fn env(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
85 self.env.push((key.into(), value.into()));
86 self
87 }
88
89 pub fn stdin(mut self, text: impl Into<String>) -> Self {
90 self.stdin = Some(text.into());
91 self
92 }
93
94 pub fn stop_descendants(mut self, value: bool) -> Self {
95 self.stop_descendants = value;
96 self
97 }
98}
99
100#[derive(Debug, Clone)]
101pub struct Output {
102 pub stdout: String,
103 pub stdout_bytes: Vec<u8>,
105 pub stderr: String,
106 pub code: i32,
107}
108
109impl Output {
110 pub fn ok(&self) -> bool {
111 self.code == 0
112 }
113}
114
115pub fn exec(argv: &[String], opts: &ExecOpts) -> Result<Output> {
120 let input = opts.stdin.as_deref().map(str::as_bytes);
121 exec_with_input(argv, opts, input)
122}
123
124fn exec_with_input(argv: &[String], opts: &ExecOpts, input: Option<&[u8]>) -> Result<Output> {
125 let program = argv
126 .first()
127 .ok_or_else(|| SparError::new("cannot run an empty command"))?;
128
129 let mut command = Command::new(program);
130 command
131 .args(&argv[1..])
132 .stdout(Stdio::piped())
133 .stderr(Stdio::piped());
134
135 if input.is_some() {
136 command.stdin(Stdio::piped());
137 } else {
138 command.stdin(Stdio::null());
140 }
141
142 if let Some(dir) = &opts.cwd {
143 command.current_dir(dir);
144 }
145 for (key, value) in &opts.env {
146 command.env(key, value);
147 }
148
149 #[cfg(unix)]
150 if opts.stop_descendants {
151 use std::os::unix::process::CommandExt;
152 command.process_group(0);
153 }
154
155 let mut child = command
156 .spawn()
157 .map_err(|e| SparError::new(format!("could not run `{}`: {e}", abbreviate(argv))))?;
158
159 if let Some(bytes) = input {
160 if let Some(mut pipe) = child.stdin.take() {
161 let _ = pipe.write_all(bytes);
162 }
163 }
166
167 let out_reader = Reader::spawn(child.stdout.take().expect("stdout piped"));
168 let err_reader = Reader::spawn(child.stderr.take().expect("stderr piped"));
169
170 let deadline = Instant::now() + opts.timeout;
171 let mut poll = Duration::from_millis(5);
172 let mut timed_out = false;
173 let mut quiet_error = None;
174 let status = loop {
175 match child.try_wait() {
176 Ok(Some(status)) => break Some(status),
177 Ok(None) => {
178 if Instant::now() >= deadline {
179 if let Err(error) = stop_and_reap(&mut child, opts.stop_descendants) {
180 quiet_error = Some(error);
181 }
182 timed_out = true;
183 break None;
184 }
185 std::thread::sleep(poll);
186 poll = (poll * 2).min(Duration::from_millis(100));
189 }
190 Err(error) => {
191 return Err(poll_failure(error, argv, || {
192 stop_and_reap(&mut child, opts.stop_descendants)
193 }));
194 }
195 }
196 };
197
198 #[cfg(unix)]
199 if !timed_out && opts.stop_descendants {
200 if let Err(error) = stop_process_group(child.id()) {
201 quiet_error = Some(error);
202 }
203 }
204
205 let stdout = out_reader.collect(DRAIN_GRACE);
213 let stderr = err_reader.collect(DRAIN_GRACE);
214
215 if let Some(error) = quiet_error {
216 return Err(quiet_point_failure(error, argv));
217 }
218
219 if timed_out {
220 #[cfg(not(unix))]
221 if opts.stop_descendants {
222 return Err(SparError::uncertain_write(format!(
223 "timed out after {}s: {}\nThe direct process was stopped, but remaining child \
224 processes could not be stopped on this platform. The worktree must be inspected \
225 before retrying.",
226 opts.timeout.as_secs(),
227 abbreviate(argv)
228 )));
229 }
230 return Err(SparError::timed_out(format!(
231 "timed out after {}s: {}\nRaise `timeout` on this agent in spar.toml if the model \
232 legitimately needs longer. Not retried: asking again would wait exactly as long a \
233 second time.",
234 opts.timeout.as_secs(),
235 abbreviate(argv)
236 )));
237 }
238
239 Ok(Output {
240 stdout: String::from_utf8_lossy(&stdout).into_owned(),
241 stdout_bytes: stdout,
242 stderr: String::from_utf8_lossy(&stderr).into_owned(),
243 code: status.and_then(|s| s.code()).unwrap_or(-1),
244 })
245}
246
247fn stop_and_reap(child: &mut std::process::Child, stop_descendants: bool) -> std::io::Result<()> {
248 #[cfg(not(unix))]
249 let _ = stop_descendants;
250
251 #[cfg(unix)]
252 if stop_descendants {
253 if let Err(group_error) = stop_process_group(child.id()) {
254 if child.kill().is_ok() {
255 let _ = child.wait();
256 }
257 return Err(group_error);
258 }
259 return child.wait().map(|_| ());
260 }
261
262 match child.kill() {
263 Ok(()) => child.wait().map(|_| ()),
264 Err(kill_error) => match child.try_wait() {
265 Ok(Some(_)) => Ok(()),
266 Ok(None) => Err(kill_error),
267 Err(poll_error) => Err(std::io::Error::new(
268 poll_error.kind(),
269 format!("could not stop the direct process: {kill_error}; {poll_error}"),
270 )),
271 },
272 }
273}
274
275fn poll_failure(
276 error: std::io::Error,
277 argv: &[String],
278 stop_and_reap: impl FnOnce() -> std::io::Result<()>,
279) -> SparError {
280 let cleanup = stop_and_reap()
281 .err()
282 .map(|cleanup| format!(" Cleanup also failed: {cleanup}."))
283 .unwrap_or_default();
284 SparError::uncertain_write(format!(
285 "could not confirm whether `{}` stopped: {error}. SPAR attempted to stop and reap the \
286 direct process and its descendants.{cleanup} The worktree must be inspected before \
287 retrying.",
288 abbreviate(argv),
289 ))
290}
291
292fn quiet_point_failure(error: std::io::Error, argv: &[String]) -> SparError {
293 SparError::uncertain_write(format!(
294 "could not establish a quiet point after `{}`: {error}. The worktree must be inspected \
295 before retrying.",
296 abbreviate(argv)
297 ))
298}
299
300#[cfg(unix)]
301fn stop_process_group(id: u32) -> std::io::Result<()> {
302 let id = i32::try_from(id).map_err(|_| {
303 std::io::Error::new(
304 std::io::ErrorKind::InvalidInput,
305 "the child process id does not fit a platform process-group id",
306 )
307 })?;
308 if unsafe { libc::kill(-id, libc::SIGKILL) } == 0 {
312 return Ok(());
313 }
314 let error = std::io::Error::last_os_error();
315 if error.raw_os_error() == Some(libc::ESRCH) {
316 return Ok(());
317 }
318 Err(error)
319}
320
321const DRAIN_GRACE: Duration = Duration::from_secs(3);
325
326struct Reader {
331 buf: Arc<Mutex<Vec<u8>>>,
332 done: Arc<AtomicBool>,
333}
334
335impl Reader {
336 fn spawn<R: Read + Send + 'static>(mut pipe: R) -> Self {
337 let buf = Arc::new(Mutex::new(Vec::new()));
338 let done = Arc::new(AtomicBool::new(false));
339 let (buf_w, done_w) = (Arc::clone(&buf), Arc::clone(&done));
340 std::thread::spawn(move || {
341 let mut chunk = [0u8; 16 * 1024];
342 loop {
343 match pipe.read(&mut chunk) {
344 Ok(0) | Err(_) => break,
345 Ok(n) => buf_w
346 .lock()
347 .unwrap_or_else(|e| e.into_inner())
348 .extend_from_slice(&chunk[..n]),
349 }
350 }
351 done_w.store(true, Ordering::Release);
352 });
353 Self { buf, done }
354 }
355
356 fn len(&self) -> usize {
357 self.buf.lock().unwrap_or_else(|e| e.into_inner()).len()
358 }
359
360 fn collect(&self, grace: Duration) -> Vec<u8> {
368 let mut last_len = self.len();
369 let mut quiet_since = Instant::now();
370 let mut poll = Duration::from_micros(100);
371 while !self.done.load(Ordering::Acquire) {
372 let now_len = self.len();
373 if now_len != last_len {
374 last_len = now_len;
375 quiet_since = Instant::now();
376 } else if quiet_since.elapsed() >= grace {
377 break;
378 }
379 std::thread::sleep(poll);
380 poll = (poll * 2).min(Duration::from_millis(10));
381 }
382 self.buf.lock().unwrap_or_else(|e| e.into_inner()).clone()
383 }
384}
385
386pub fn run(argv: &[String], opts: &ExecOpts) -> Result<String> {
389 let out = exec(argv, opts)?;
390 if opts.check && !out.ok() {
391 return Err(SparError::call_failed(failure_message(argv, &out)));
392 }
393 Ok(out.stdout)
394}
395
396pub(crate) fn run_with_input_bytes(
397 argv: &[String],
398 opts: &ExecOpts,
399 input: &[u8],
400) -> Result<String> {
401 let out = exec_with_input(argv, opts, Some(input))?;
402 if opts.check && !out.ok() {
403 return Err(SparError::call_failed(failure_message(argv, &out)));
404 }
405 Ok(out.stdout)
406}
407
408pub fn run_bytes(argv: &[String], opts: &ExecOpts) -> Result<Vec<u8>> {
410 let out = exec(argv, opts)?;
411 if opts.check && !out.ok() {
412 return Err(SparError::call_failed(failure_message(argv, &out)));
413 }
414 Ok(out.stdout_bytes)
415}
416
417pub fn run_str(argv: &[&str], opts: &ExecOpts) -> Result<String> {
419 let owned: Vec<String> = argv.iter().map(|s| (*s).to_string()).collect();
420 run(&owned, opts)
421}
422
423pub fn abbreviate(argv: &[String]) -> String {
425 argv.iter()
426 .map(|arg| {
427 let one_line = arg.split_whitespace().collect::<Vec<_>>().join(" ");
428 if one_line.chars().count() <= 60 {
429 one_line
430 } else {
431 let head: String = one_line.chars().take(57).collect();
432 format!("{head}...")
433 }
434 })
435 .collect::<Vec<_>>()
436 .join(" ")
437}
438
439pub fn failure_message(argv: &[String], out: &Output) -> String {
444 let mut parts = vec![format!(
445 "command failed ({}): {}",
446 out.code,
447 abbreviate(argv)
448 )];
449 for (label, stream) in [("stderr", &out.stderr), ("stdout", &out.stdout)] {
450 let text = stream.trim();
451 if !text.is_empty() {
452 parts.push(format!("--- {label} ---\n{}", tail(text, 1500)));
453 }
454 }
455 if parts.len() == 1 {
456 parts.push("(no output on either stream)".to_string());
457 }
458 parts.join("\n")
459}
460
461fn tail(text: &str, max: usize) -> &str {
463 let count = text.chars().count();
464 if count <= max {
465 return text;
466 }
467 let start = text
468 .char_indices()
469 .nth(count - max)
470 .map(|(i, _)| i)
471 .unwrap_or(0);
472 &text[start..]
473}
474
475pub fn which(program: &str) -> Option<PathBuf> {
477 if program.contains(std::path::MAIN_SEPARATOR) {
478 let path = PathBuf::from(program);
479 return is_executable(&path).then_some(path);
480 }
481 let paths = std::env::var_os("PATH")?;
482 std::env::split_paths(&paths).find_map(|dir| {
483 let candidate = dir.join(program);
484 is_executable(&candidate).then_some(candidate)
485 })
486}
487
488pub fn is_executable(path: &Path) -> bool {
489 #[cfg(unix)]
490 {
491 use std::os::unix::fs::PermissionsExt;
492 match std::fs::metadata(path) {
493 Ok(meta) => meta.is_file() && meta.permissions().mode() & 0o111 != 0,
494 Err(_) => false,
495 }
496 }
497 #[cfg(not(unix))]
498 {
499 path.is_file()
500 }
501}
502
503pub fn expand_tilde(path: &str) -> PathBuf {
506 if path == "~" {
507 if let Some(home) = home_dir() {
508 return home;
509 }
510 }
511 if let Some(rest) = path.strip_prefix("~/") {
512 if let Some(home) = home_dir() {
513 return home.join(rest);
514 }
515 }
516 PathBuf::from(path)
517}
518
519pub fn home_dir() -> Option<PathBuf> {
520 std::env::var_os("HOME").map(PathBuf::from)
521}
522
523#[cfg(test)]
524mod tests {
525 use super::*;
526
527 fn proc(out: &str, err: &str, code: i32) -> Output {
528 Output {
529 stdout: out.into(),
530 stdout_bytes: out.as_bytes().to_vec(),
531 stderr: err.into(),
532 code,
533 }
534 }
535
536 fn argv(parts: &[&str]) -> Vec<String> {
537 parts.iter().map(|s| s.to_string()).collect()
538 }
539
540 #[test]
541 fn a_poll_failure_attempts_cleanup_and_is_uncertain() {
542 let stopped = std::cell::Cell::new(false);
543 let error = poll_failure(
544 std::io::Error::other("poll failed"),
545 &argv(&["editor"]),
546 || {
547 stopped.set(true);
548 Ok(())
549 },
550 );
551
552 assert!(stopped.get());
553 assert_eq!(crate::error::ErrorKind::UncertainWrite, error.kind());
554 assert!(error.to_string().contains("poll failed"), "{error}");
555 }
556
557 #[test]
558 fn a_stop_or_reap_failure_is_uncertain() {
559 let error =
560 quiet_point_failure(std::io::Error::other("could not reap"), &argv(&["editor"]));
561
562 assert_eq!(crate::error::ErrorKind::UncertainWrite, error.kind());
563 assert!(error.to_string().contains("could not reap"), "{error}");
564 }
565
566 #[test]
567 fn stdin_reaches_the_child_byte_for_byte() {
568 let body = "line one \n\nline three\n";
569 let out = run_str(&["/bin/sh", "-c", "cat"], &ExecOpts::new().stdin(body))
570 .expect("the child reads stdin");
571 assert_eq!(body, out);
572 }
573
574 #[test]
575 fn byte_input_reaches_the_child_without_text_conversion() {
576 let input = [b'f', 0xff, 0];
577 let command = argv(&["/bin/sh", "-c", "cat"]);
578 let out = exec_with_input(&command, &ExecOpts::new(), Some(&input))
579 .expect("the child reads byte input");
580
581 assert_eq!(input, out.stdout_bytes.as_slice());
582 }
583
584 #[test]
585 fn stdout_used_when_stderr_is_empty() {
586 let msg = failure_message(&argv(&["claude"]), &proc("You've hit your limit.", "", 1));
587 assert!(msg.contains("hit your limit"), "{msg}");
588 }
589
590 #[test]
591 fn stderr_shown_when_present() {
592 let msg = failure_message(&argv(&["gh"]), &proc("noise", "real reason", 1));
593 assert!(msg.contains("real reason"), "{msg}");
594 }
595
596 #[test]
597 fn both_streams_are_shown_not_just_one() {
598 let msg = failure_message(&argv(&["gh"]), &proc("on stdout", "on stderr", 1));
599 assert!(
600 msg.contains("on stdout") && msg.contains("on stderr"),
601 "{msg}"
602 );
603 }
604
605 #[test]
606 fn says_something_when_both_are_empty() {
607 assert!(failure_message(&argv(&["x"]), &proc("", "", 2)).contains("no output"));
608 }
609
610 #[test]
611 fn long_arguments_are_abbreviated() {
612 let long = "word ".repeat(500);
613 let out = abbreviate(&argv(&["claude", "-p", &long]));
614 assert!(out.len() < 200, "{}", out.len());
615 }
616
617 #[test]
618 fn newlines_in_arguments_do_not_break_the_line() {
619 assert!(!abbreviate(&argv(&["claude", "a\nb\nc"])).contains('\n'));
620 }
621
622 #[test]
623 fn short_arguments_survive_intact() {
624 assert_eq!(
625 "gh pr merge 17",
626 abbreviate(&argv(&["gh", "pr", "merge", "17"]))
627 );
628 }
629
630 #[test]
631 fn abbreviation_never_splits_a_character() {
632 let wide = "\u{1f600}".repeat(200);
635 let out = abbreviate(&argv(&[&wide]));
636 assert!(out.ends_with("..."));
637 }
638
639 #[test]
640 fn exit_code_is_reported() {
641 let out = exec(
642 &argv(&["sh", "-c", "exit 3"]),
643 &ExecOpts::new().check(false),
644 )
645 .unwrap();
646 assert_eq!(3, out.code);
647 }
648
649 #[test]
650 fn check_false_returns_stdout_on_failure() {
651 let text = run(
652 &argv(&["sh", "-c", "echo partial; exit 1"]),
653 &ExecOpts::new().check(false),
654 )
655 .unwrap();
656 assert_eq!("partial\n", text);
657 }
658
659 #[test]
660 fn check_true_fails_loudly() {
661 let err = run(
662 &argv(&["sh", "-c", "echo why >&2; exit 1"]),
663 &ExecOpts::new(),
664 )
665 .unwrap_err();
666 assert!(err.to_string().contains("why"), "{err}");
667 }
668
669 #[test]
670 fn large_output_does_not_deadlock() {
671 let text = run(
673 &argv(&["sh", "-c", "yes hello | head -c 400000"]),
674 &ExecOpts::new().timeout_secs(60),
675 )
676 .unwrap();
677 assert_eq!(400_000, text.len());
678 }
679
680 #[test]
684 fn a_surviving_grandchild_holding_the_pipe_cannot_hang_the_timeout() {
685 let start = Instant::now();
686 let err = run(
687 &argv(&["sh", "-c", "sleep 120 & echo parent-output; sleep 60"]),
688 &ExecOpts::new().timeout_secs(1),
689 )
690 .unwrap_err();
691 let elapsed = start.elapsed();
692
693 assert!(err.to_string().contains("timed out"), "{err}");
694 assert!(
695 elapsed < Duration::from_secs(20),
696 "the timeout did not bound the call: {elapsed:?}"
697 );
698 }
699
700 #[test]
701 fn a_surviving_grandchild_does_not_hang_a_normal_exit_either() {
702 let start = Instant::now();
703 let out = run(
704 &argv(&["sh", "-c", "sleep 120 & echo done"]),
705 &ExecOpts::new().timeout_secs(60),
706 )
707 .unwrap();
708 assert!(out.contains("done"), "{out:?}");
709 assert!(
710 start.elapsed() < Duration::from_secs(20),
711 "waited on a grandchild that will never exit"
712 );
713 }
714
715 #[cfg(unix)]
716 #[test]
717 fn a_stopped_process_group_cannot_edit_after_the_parent_exits() {
718 let late = std::env::temp_dir().join(format!(
719 "spar-late-child-{}-{}",
720 std::process::id(),
721 std::thread::current().name().unwrap_or("test")
722 ));
723 let _ = std::fs::remove_file(&late);
724 let out = run(
725 &argv(&[
726 "/bin/sh",
727 "-c",
728 "(sleep 1; touch \"$1\") & echo done",
729 "sh",
730 late.to_str().unwrap(),
731 ]),
732 &ExecOpts::new().stop_descendants(true),
733 )
734 .unwrap();
735
736 assert!(out.contains("done"), "{out:?}");
737 std::thread::sleep(Duration::from_millis(1_200));
738 assert!(!late.exists(), "a descendant edited after the quiet point");
739 }
740
741 #[test]
742 fn timeout_kills_and_explains() {
743 let err = run(
744 &argv(&["sh", "-c", "sleep 30"]),
745 &ExecOpts::new().timeout_secs(1),
746 )
747 .unwrap_err();
748 assert!(err.to_string().contains("timed out"), "{err}");
749 }
750
751 #[test]
752 fn missing_binary_names_the_command() {
753 let err = exec(
754 &argv(&["spar-definitely-not-a-real-binary"]),
755 &ExecOpts::new(),
756 )
757 .unwrap_err();
758 assert!(
759 err.to_string()
760 .contains("spar-definitely-not-a-real-binary"),
761 "{err}"
762 );
763 }
764
765 #[test]
766 fn tilde_expands_against_home() {
767 std::env::set_var("HOME", "/home/someone");
768 assert_eq!(PathBuf::from("/home/someone/bin"), expand_tilde("~/bin"));
769 assert_eq!(PathBuf::from("/absolute"), expand_tilde("/absolute"));
770 assert_eq!(PathBuf::from("~notauser/x"), expand_tilde("~notauser/x"));
771 }
772}
773
774#[cfg(test)]
775mod timeout_kind_tests {
776 use super::*;
777 use crate::error::ErrorKind;
778
779 #[test]
783 fn a_timeout_is_marked_as_one_and_is_not_worth_retrying() {
784 let err = run(
785 &["sh".to_string(), "-c".to_string(), "sleep 30".to_string()],
786 &ExecOpts::new().timeout_secs(1),
787 )
788 .unwrap_err();
789
790 assert_eq!(ErrorKind::TimedOut, err.kind());
791 assert!(!err.worth_retrying());
792 assert!(err.to_string().contains("Not retried"), "{err}");
793 }
794
795 #[test]
799 fn a_non_zero_exit_is_the_call_failing_rather_than_the_answer() {
800 let err = run(
801 &["sh".to_string(), "-c".to_string(), "exit 1".to_string()],
802 &ExecOpts::new(),
803 )
804 .unwrap_err();
805
806 assert_eq!(ErrorKind::CallFailed, err.kind());
807 assert!(err.worth_retrying());
811 }
812}