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 let out_reader = Reader::spawn(child.stdout.take().expect("stdout piped"));
169 let err_reader = Reader::spawn(child.stderr.take().expect("stderr piped"));
170
171 if let Some(bytes) = input {
172 if let Some(mut pipe) = child.stdin.take() {
173 let owned = bytes.to_vec();
174 std::thread::spawn(move || {
179 let _ = pipe.write_all(&owned);
180 });
181 }
182 }
183
184 let deadline = Instant::now() + opts.timeout;
185 let mut poll = Duration::from_millis(5);
186 let mut timed_out = false;
187 let mut quiet_error = None;
188 let status = loop {
189 match child.try_wait() {
190 Ok(Some(status)) => break Some(status),
191 Ok(None) => {
192 if Instant::now() >= deadline {
193 if let Err(error) = stop_and_reap(&mut child, opts.stop_descendants) {
194 quiet_error = Some(error);
195 }
196 timed_out = true;
197 break None;
198 }
199 std::thread::sleep(poll);
200 poll = (poll * 2).min(Duration::from_millis(100));
203 }
204 Err(error) => {
205 return Err(poll_failure(error, argv, || {
206 stop_and_reap(&mut child, opts.stop_descendants)
207 }));
208 }
209 }
210 };
211
212 #[cfg(unix)]
213 if !timed_out && opts.stop_descendants {
214 if let Err(error) = stop_process_group(child.id()) {
215 quiet_error = Some(error);
216 }
217 }
218
219 let stdout = out_reader.collect(DRAIN_GRACE);
227 let stderr = err_reader.collect(DRAIN_GRACE);
228
229 if let Some(error) = quiet_error {
230 return Err(quiet_point_failure(error, argv));
231 }
232
233 if timed_out {
234 #[cfg(not(unix))]
235 if opts.stop_descendants {
236 return Err(SparError::uncertain_write(format!(
237 "timed out after {}s: {}\nThe direct process was stopped, but remaining child \
238 processes could not be stopped on this platform. The worktree must be inspected \
239 before retrying.",
240 opts.timeout.as_secs(),
241 abbreviate(argv)
242 )));
243 }
244 return Err(SparError::timed_out(format!(
245 "timed out after {}s: {}\nRaise `timeout` on this agent in spar.toml if the model \
246 legitimately needs longer. Not retried: asking again would wait exactly as long a \
247 second time.",
248 opts.timeout.as_secs(),
249 abbreviate(argv)
250 )));
251 }
252
253 Ok(Output {
254 stdout: String::from_utf8_lossy(&stdout).into_owned(),
255 stdout_bytes: stdout,
256 stderr: String::from_utf8_lossy(&stderr).into_owned(),
257 code: status.and_then(|s| s.code()).unwrap_or(-1),
258 })
259}
260
261fn stop_and_reap(child: &mut std::process::Child, stop_descendants: bool) -> std::io::Result<()> {
262 #[cfg(not(unix))]
263 let _ = stop_descendants;
264
265 #[cfg(unix)]
266 if stop_descendants {
267 if let Err(group_error) = stop_process_group(child.id()) {
268 if child.kill().is_ok() {
269 let _ = child.wait();
270 }
271 return Err(group_error);
272 }
273 return child.wait().map(|_| ());
274 }
275
276 match child.kill() {
277 Ok(()) => child.wait().map(|_| ()),
278 Err(kill_error) => match child.try_wait() {
279 Ok(Some(_)) => Ok(()),
280 Ok(None) => Err(kill_error),
281 Err(poll_error) => Err(std::io::Error::new(
282 poll_error.kind(),
283 format!("could not stop the direct process: {kill_error}; {poll_error}"),
284 )),
285 },
286 }
287}
288
289fn poll_failure(
290 error: std::io::Error,
291 argv: &[String],
292 stop_and_reap: impl FnOnce() -> std::io::Result<()>,
293) -> SparError {
294 let cleanup = stop_and_reap()
295 .err()
296 .map(|cleanup| format!(" Cleanup also failed: {cleanup}."))
297 .unwrap_or_default();
298 SparError::uncertain_write(format!(
299 "could not confirm whether `{}` stopped: {error}. SPAR attempted to stop and reap the \
300 direct process and its descendants.{cleanup} The worktree must be inspected before \
301 retrying.",
302 abbreviate(argv),
303 ))
304}
305
306fn quiet_point_failure(error: std::io::Error, argv: &[String]) -> SparError {
307 SparError::uncertain_write(format!(
308 "could not establish a quiet point after `{}`: {error}. The worktree must be inspected \
309 before retrying.",
310 abbreviate(argv)
311 ))
312}
313
314#[cfg(unix)]
315fn stop_process_group(id: u32) -> std::io::Result<()> {
316 let id = i32::try_from(id).map_err(|_| {
317 std::io::Error::new(
318 std::io::ErrorKind::InvalidInput,
319 "the child process id does not fit a platform process-group id",
320 )
321 })?;
322 if unsafe { libc::kill(-id, libc::SIGKILL) } == 0 {
326 return Ok(());
327 }
328 let error = std::io::Error::last_os_error();
329 if error.raw_os_error() == Some(libc::ESRCH) {
330 return Ok(());
331 }
332 Err(error)
333}
334
335const DRAIN_GRACE: Duration = Duration::from_secs(3);
339
340struct Reader {
345 buf: Arc<Mutex<Vec<u8>>>,
346 done: Arc<AtomicBool>,
347}
348
349impl Reader {
350 fn spawn<R: Read + Send + 'static>(mut pipe: R) -> Self {
351 let buf = Arc::new(Mutex::new(Vec::new()));
352 let done = Arc::new(AtomicBool::new(false));
353 let (buf_w, done_w) = (Arc::clone(&buf), Arc::clone(&done));
354 std::thread::spawn(move || {
355 let mut chunk = [0u8; 16 * 1024];
356 loop {
357 match pipe.read(&mut chunk) {
358 Ok(0) | Err(_) => break,
359 Ok(n) => buf_w
360 .lock()
361 .unwrap_or_else(|e| e.into_inner())
362 .extend_from_slice(&chunk[..n]),
363 }
364 }
365 done_w.store(true, Ordering::Release);
366 });
367 Self { buf, done }
368 }
369
370 fn len(&self) -> usize {
371 self.buf.lock().unwrap_or_else(|e| e.into_inner()).len()
372 }
373
374 fn collect(&self, grace: Duration) -> Vec<u8> {
382 let mut last_len = self.len();
383 let mut quiet_since = Instant::now();
384 let mut poll = Duration::from_micros(100);
385 while !self.done.load(Ordering::Acquire) {
386 let now_len = self.len();
387 if now_len != last_len {
388 last_len = now_len;
389 quiet_since = Instant::now();
390 } else if quiet_since.elapsed() >= grace {
391 break;
392 }
393 std::thread::sleep(poll);
394 poll = (poll * 2).min(Duration::from_millis(10));
395 }
396 self.buf.lock().unwrap_or_else(|e| e.into_inner()).clone()
397 }
398}
399
400pub fn run(argv: &[String], opts: &ExecOpts) -> Result<String> {
403 let out = exec(argv, opts)?;
404 if opts.check && !out.ok() {
405 return Err(SparError::call_failed(failure_message(argv, &out)));
406 }
407 Ok(out.stdout)
408}
409
410pub(crate) fn run_with_input_bytes(
411 argv: &[String],
412 opts: &ExecOpts,
413 input: &[u8],
414) -> Result<String> {
415 let out = exec_with_input(argv, opts, Some(input))?;
416 if opts.check && !out.ok() {
417 return Err(SparError::call_failed(failure_message(argv, &out)));
418 }
419 Ok(out.stdout)
420}
421
422pub fn run_bytes(argv: &[String], opts: &ExecOpts) -> Result<Vec<u8>> {
424 let out = exec(argv, opts)?;
425 if opts.check && !out.ok() {
426 return Err(SparError::call_failed(failure_message(argv, &out)));
427 }
428 Ok(out.stdout_bytes)
429}
430
431pub fn run_str(argv: &[&str], opts: &ExecOpts) -> Result<String> {
433 let owned: Vec<String> = argv.iter().map(|s| (*s).to_string()).collect();
434 run(&owned, opts)
435}
436
437pub fn abbreviate(argv: &[String]) -> String {
439 argv.iter()
440 .map(|arg| {
441 let one_line = arg.split_whitespace().collect::<Vec<_>>().join(" ");
442 if one_line.chars().count() <= 60 {
443 one_line
444 } else {
445 let head: String = one_line.chars().take(57).collect();
446 format!("{head}...")
447 }
448 })
449 .collect::<Vec<_>>()
450 .join(" ")
451}
452
453pub fn failure_message(argv: &[String], out: &Output) -> String {
458 let mut parts = vec![format!(
459 "command failed ({}): {}",
460 out.code,
461 abbreviate(argv)
462 )];
463 for (label, stream) in [("stderr", &out.stderr), ("stdout", &out.stdout)] {
464 let text = stream.trim();
465 if !text.is_empty() {
466 parts.push(format!("--- {label} ---\n{}", tail(text, 1500)));
467 }
468 }
469 if parts.len() == 1 {
470 parts.push("(no output on either stream)".to_string());
471 }
472 parts.join("\n")
473}
474
475fn tail(text: &str, max: usize) -> &str {
477 let count = text.chars().count();
478 if count <= max {
479 return text;
480 }
481 let start = text
482 .char_indices()
483 .nth(count - max)
484 .map(|(i, _)| i)
485 .unwrap_or(0);
486 &text[start..]
487}
488
489pub fn which(program: &str) -> Option<PathBuf> {
491 if program.contains(std::path::MAIN_SEPARATOR) {
492 let path = PathBuf::from(program);
493 return is_executable(&path).then_some(path);
494 }
495 let paths = std::env::var_os("PATH")?;
496 std::env::split_paths(&paths).find_map(|dir| {
497 let candidate = dir.join(program);
498 is_executable(&candidate).then_some(candidate)
499 })
500}
501
502pub fn is_executable(path: &Path) -> bool {
503 #[cfg(unix)]
504 {
505 use std::os::unix::fs::PermissionsExt;
506 match std::fs::metadata(path) {
507 Ok(meta) => meta.is_file() && meta.permissions().mode() & 0o111 != 0,
508 Err(_) => false,
509 }
510 }
511 #[cfg(not(unix))]
512 {
513 path.is_file()
514 }
515}
516
517pub fn expand_tilde(path: &str) -> PathBuf {
520 if path == "~" {
521 if let Some(home) = home_dir() {
522 return home;
523 }
524 }
525 if let Some(rest) = path.strip_prefix("~/") {
526 if let Some(home) = home_dir() {
527 return home.join(rest);
528 }
529 }
530 PathBuf::from(path)
531}
532
533pub fn home_dir() -> Option<PathBuf> {
534 std::env::var_os("HOME").map(PathBuf::from)
535}
536
537#[cfg(test)]
538mod tests {
539 use super::*;
540
541 fn proc(out: &str, err: &str, code: i32) -> Output {
542 Output {
543 stdout: out.into(),
544 stdout_bytes: out.as_bytes().to_vec(),
545 stderr: err.into(),
546 code,
547 }
548 }
549
550 fn argv(parts: &[&str]) -> Vec<String> {
551 parts.iter().map(|s| s.to_string()).collect()
552 }
553
554 #[test]
555 fn a_poll_failure_attempts_cleanup_and_is_uncertain() {
556 let stopped = std::cell::Cell::new(false);
557 let error = poll_failure(
558 std::io::Error::other("poll failed"),
559 &argv(&["editor"]),
560 || {
561 stopped.set(true);
562 Ok(())
563 },
564 );
565
566 assert!(stopped.get());
567 assert_eq!(crate::error::ErrorKind::UncertainWrite, error.kind());
568 assert!(error.to_string().contains("poll failed"), "{error}");
569 }
570
571 #[test]
572 fn a_stop_or_reap_failure_is_uncertain() {
573 let error =
574 quiet_point_failure(std::io::Error::other("could not reap"), &argv(&["editor"]));
575
576 assert_eq!(crate::error::ErrorKind::UncertainWrite, error.kind());
577 assert!(error.to_string().contains("could not reap"), "{error}");
578 }
579
580 #[test]
581 fn stdin_reaches_the_child_byte_for_byte() {
582 let body = "line one \n\nline three\n";
583 let out = run_str(&["/bin/sh", "-c", "cat"], &ExecOpts::new().stdin(body))
584 .expect("the child reads stdin");
585 assert_eq!(body, out);
586 }
587
588 #[test]
589 fn byte_input_reaches_the_child_without_text_conversion() {
590 let input = [b'f', 0xff, 0];
591 let command = argv(&["/bin/sh", "-c", "cat"]);
592 let out = exec_with_input(&command, &ExecOpts::new(), Some(&input))
593 .expect("the child reads byte input");
594
595 assert_eq!(input, out.stdout_bytes.as_slice());
596 }
597
598 #[test]
599 fn stdout_used_when_stderr_is_empty() {
600 let msg = failure_message(&argv(&["claude"]), &proc("You've hit your limit.", "", 1));
601 assert!(msg.contains("hit your limit"), "{msg}");
602 }
603
604 #[test]
605 fn stderr_shown_when_present() {
606 let msg = failure_message(&argv(&["gh"]), &proc("noise", "real reason", 1));
607 assert!(msg.contains("real reason"), "{msg}");
608 }
609
610 #[test]
611 fn both_streams_are_shown_not_just_one() {
612 let msg = failure_message(&argv(&["gh"]), &proc("on stdout", "on stderr", 1));
613 assert!(
614 msg.contains("on stdout") && msg.contains("on stderr"),
615 "{msg}"
616 );
617 }
618
619 #[test]
620 fn says_something_when_both_are_empty() {
621 assert!(failure_message(&argv(&["x"]), &proc("", "", 2)).contains("no output"));
622 }
623
624 #[test]
625 fn long_arguments_are_abbreviated() {
626 let long = "word ".repeat(500);
627 let out = abbreviate(&argv(&["claude", "-p", &long]));
628 assert!(out.len() < 200, "{}", out.len());
629 }
630
631 #[test]
632 fn newlines_in_arguments_do_not_break_the_line() {
633 assert!(!abbreviate(&argv(&["claude", "a\nb\nc"])).contains('\n'));
634 }
635
636 #[test]
637 fn short_arguments_survive_intact() {
638 assert_eq!(
639 "gh pr merge 17",
640 abbreviate(&argv(&["gh", "pr", "merge", "17"]))
641 );
642 }
643
644 #[test]
645 fn abbreviation_never_splits_a_character() {
646 let wide = "\u{1f600}".repeat(200);
649 let out = abbreviate(&argv(&[&wide]));
650 assert!(out.ends_with("..."));
651 }
652
653 #[test]
654 fn exit_code_is_reported() {
655 let out = exec(
656 &argv(&["sh", "-c", "exit 3"]),
657 &ExecOpts::new().check(false),
658 )
659 .unwrap();
660 assert_eq!(3, out.code);
661 }
662
663 #[test]
664 fn check_false_returns_stdout_on_failure() {
665 let text = run(
666 &argv(&["sh", "-c", "echo partial; exit 1"]),
667 &ExecOpts::new().check(false),
668 )
669 .unwrap();
670 assert_eq!("partial\n", text);
671 }
672
673 #[test]
674 fn check_true_fails_loudly() {
675 let err = run(
676 &argv(&["sh", "-c", "echo why >&2; exit 1"]),
677 &ExecOpts::new(),
678 )
679 .unwrap_err();
680 assert!(err.to_string().contains("why"), "{err}");
681 }
682
683 #[test]
684 fn large_output_does_not_deadlock() {
685 let text = run(
687 &argv(&["sh", "-c", "yes hello | head -c 400000"]),
688 &ExecOpts::new().timeout_secs(60),
689 )
690 .unwrap();
691 assert_eq!(400_000, text.len());
692 }
693
694 #[test]
702 fn large_input_and_large_output_together_do_not_deadlock() {
703 let line = "a".repeat(199);
704 let input = std::iter::repeat_n(line.as_str(), 5_000)
705 .collect::<Vec<_>>()
706 .join("\n");
707 assert!(input.len() > 900_000, "the input has to outgrow the pipe");
708
709 let start = Instant::now();
710 let out = run(
713 &argv(&["sh", "-c", "tee /dev/stderr"]),
714 &ExecOpts::new().timeout_secs(60).stdin(input.clone()),
715 )
716 .unwrap();
717
718 assert_eq!(input.len(), out.trim_end().len());
719 assert!(
720 start.elapsed() < Duration::from_secs(30),
721 "the call did not return promptly: {:?}",
722 start.elapsed()
723 );
724 }
725
726 #[test]
730 fn a_surviving_grandchild_holding_the_pipe_cannot_hang_the_timeout() {
731 let start = Instant::now();
732 let err = run(
733 &argv(&["sh", "-c", "sleep 120 & echo parent-output; sleep 60"]),
734 &ExecOpts::new().timeout_secs(1),
735 )
736 .unwrap_err();
737 let elapsed = start.elapsed();
738
739 assert!(err.to_string().contains("timed out"), "{err}");
740 assert!(
741 elapsed < Duration::from_secs(20),
742 "the timeout did not bound the call: {elapsed:?}"
743 );
744 }
745
746 #[test]
747 fn a_surviving_grandchild_does_not_hang_a_normal_exit_either() {
748 let start = Instant::now();
749 let out = run(
750 &argv(&["sh", "-c", "sleep 120 & echo done"]),
751 &ExecOpts::new().timeout_secs(60),
752 )
753 .unwrap();
754 assert!(out.contains("done"), "{out:?}");
755 assert!(
756 start.elapsed() < Duration::from_secs(20),
757 "waited on a grandchild that will never exit"
758 );
759 }
760
761 #[cfg(unix)]
762 #[test]
763 fn a_stopped_process_group_cannot_edit_after_the_parent_exits() {
764 let late = std::env::temp_dir().join(format!(
765 "spar-late-child-{}-{}",
766 std::process::id(),
767 std::thread::current().name().unwrap_or("test")
768 ));
769 let _ = std::fs::remove_file(&late);
770 let out = run(
771 &argv(&[
772 "/bin/sh",
773 "-c",
774 "(sleep 1; touch \"$1\") & echo done",
775 "sh",
776 late.to_str().unwrap(),
777 ]),
778 &ExecOpts::new().stop_descendants(true),
779 )
780 .unwrap();
781
782 assert!(out.contains("done"), "{out:?}");
783 std::thread::sleep(Duration::from_millis(1_200));
784 assert!(!late.exists(), "a descendant edited after the quiet point");
785 }
786
787 #[test]
788 fn timeout_kills_and_explains() {
789 let err = run(
790 &argv(&["sh", "-c", "sleep 30"]),
791 &ExecOpts::new().timeout_secs(1),
792 )
793 .unwrap_err();
794 assert!(err.to_string().contains("timed out"), "{err}");
795 }
796
797 #[test]
798 fn missing_binary_names_the_command() {
799 let err = exec(
800 &argv(&["spar-definitely-not-a-real-binary"]),
801 &ExecOpts::new(),
802 )
803 .unwrap_err();
804 assert!(
805 err.to_string()
806 .contains("spar-definitely-not-a-real-binary"),
807 "{err}"
808 );
809 }
810
811 #[test]
812 fn tilde_expands_against_home() {
813 std::env::set_var("HOME", "/home/someone");
814 assert_eq!(PathBuf::from("/home/someone/bin"), expand_tilde("~/bin"));
815 assert_eq!(PathBuf::from("/absolute"), expand_tilde("/absolute"));
816 assert_eq!(PathBuf::from("~notauser/x"), expand_tilde("~notauser/x"));
817 }
818}
819
820#[cfg(test)]
821mod timeout_kind_tests {
822 use super::*;
823 use crate::error::ErrorKind;
824
825 #[test]
829 fn a_timeout_is_marked_as_one_and_is_not_worth_retrying() {
830 let err = run(
831 &["sh".to_string(), "-c".to_string(), "sleep 30".to_string()],
832 &ExecOpts::new().timeout_secs(1),
833 )
834 .unwrap_err();
835
836 assert_eq!(ErrorKind::TimedOut, err.kind());
837 assert!(!err.worth_retrying());
838 assert!(err.to_string().contains("Not retried"), "{err}");
839 }
840
841 #[test]
845 fn a_non_zero_exit_is_the_call_failing_rather_than_the_answer() {
846 let err = run(
847 &["sh".to_string(), "-c".to_string(), "exit 1".to_string()],
848 &ExecOpts::new(),
849 )
850 .unwrap_err();
851
852 assert_eq!(ErrorKind::CallFailed, err.kind());
853 assert!(err.worth_retrying());
857 }
858}