1use crate::apr_bin::apr_binary;
23use crate::types::{ContentBlock, ToolCallResult};
24use std::ffi::OsStr;
25use std::io::{BufRead, BufReader, Read};
26use std::process::{Command, Stdio};
27use std::sync::mpsc::{Receiver, TryRecvError};
28use std::time::{Duration, Instant};
29
30pub const CANCEL_GRACE_MS: u64 = 30_000;
36
37const POLL_INTERVAL: Duration = Duration::from_millis(10);
39
40fn quote_arg(arg: &str) -> String {
46 let safe = !arg.is_empty()
47 && arg
48 .bytes()
49 .all(|b| b.is_ascii_alphanumeric() || b"@%+=:,./-_".contains(&b));
50 if safe {
51 arg.to_string()
52 } else {
53 format!("'{}'", arg.replace('\'', r"'\''"))
54 }
55}
56
57fn failure_result(cmd_display: &str, code: i32, stdout: &str, stderr: &str) -> ToolCallResult {
63 let summary = if stderr.trim().is_empty() {
64 stdout.to_string()
65 } else {
66 stderr.to_string()
67 };
68 let mut content = vec![ContentBlock::text(format!(
69 "`{cmd_display}` failed (exit {code}): {summary}"
70 ))];
71 if !stderr.trim().is_empty() && !stdout.trim().is_empty() {
72 content.push(ContentBlock::text(stdout.to_string()));
73 }
74 ToolCallResult {
75 content,
76 is_error: Some(true),
77 }
78}
79
80#[must_use]
88pub fn run_apr(args: &[&str]) -> ToolCallResult {
89 run_program(apr_binary(), args)
90}
91
92#[must_use]
96pub fn run_program<P: AsRef<OsStr>>(program: P, args: &[&str]) -> ToolCallResult {
97 let program = program.as_ref();
98 let cmd_display = display_cmd(program, args);
99 let output = match Command::new(program).args(args).output() {
100 Ok(o) => o,
101 Err(e) => {
102 return ToolCallResult::error(format!("Failed to spawn `{cmd_display}`: {e}"));
103 }
104 };
105
106 let stdout = String::from_utf8_lossy(&output.stdout).into_owned();
107 let stderr = String::from_utf8_lossy(&output.stderr).into_owned();
108
109 if output.status.success() {
110 if stdout.trim().is_empty() {
111 ToolCallResult::error(format!("`{cmd_display}` produced no output"))
112 } else {
113 ToolCallResult::success(stdout)
114 }
115 } else {
116 let code = output.status.code().unwrap_or(-1);
117 failure_result(&cmd_display, code, &stdout, &stderr)
118 }
119}
120
121fn display_cmd(program: &OsStr, args: &[&str]) -> String {
132 let mut out = quote_arg(&program.to_string_lossy());
133 for a in args {
134 out.push(' ');
135 out.push_str("e_arg(a));
136 }
137 out
138}
139
140#[must_use]
151pub fn run_apr_cancellable(
152 args: &[&str],
153 cancel_rx: &Receiver<()>,
154 grace_ms: u64,
155) -> ToolCallResult {
156 spawn_cancellable(apr_binary(), args, cancel_rx, grace_ms)
157}
158
159#[must_use]
163pub fn spawn_cancellable<P: AsRef<OsStr>>(
164 program: P,
165 args: &[&str],
166 cancel_rx: &Receiver<()>,
167 grace_ms: u64,
168) -> ToolCallResult {
169 let program = program.as_ref();
170 let cmd_display = display_cmd(program, args);
171
172 let mut child = match Command::new(program)
173 .args(args)
174 .stdout(Stdio::piped())
175 .stderr(Stdio::piped())
176 .spawn()
177 {
178 Ok(c) => c,
179 Err(e) => {
180 return ToolCallResult::error(format!("Failed to spawn `{cmd_display}`: {e}"));
181 }
182 };
183
184 let pid = child.id();
185
186 let wait_status = loop {
190 match child.try_wait() {
191 Ok(Some(status)) => break Ok(status),
192 Ok(None) => {}
193 Err(e) => {
194 return ToolCallResult::error(format!("Failed to poll `{cmd_display}`: {e}"));
195 }
196 }
197
198 match cancel_rx.try_recv() {
199 Ok(()) => break Err(CancelReason::Signalled),
200 Err(TryRecvError::Empty) => {}
201 Err(TryRecvError::Disconnected) => {
202 }
205 }
206
207 std::thread::sleep(POLL_INTERVAL);
208 };
209
210 match wait_status {
211 Ok(status) => {
212 let stdout = drain(&mut child.stdout.take());
214 let stderr = drain(&mut child.stderr.take());
215 if status.success() {
216 if stdout.trim().is_empty() {
217 ToolCallResult::error(format!("`{cmd_display}` produced no output"))
218 } else {
219 ToolCallResult::success(stdout)
220 }
221 } else {
222 let code = status.code().unwrap_or(-1);
223 failure_result(&cmd_display, code, &stdout, &stderr)
224 }
225 }
226 Err(CancelReason::Signalled) => {
227 send_sigterm(pid);
229 let deadline = Instant::now() + Duration::from_millis(grace_ms);
230 let mut escalated = false;
231 loop {
232 match child.try_wait() {
233 Ok(Some(_)) => break,
234 Ok(None) => {}
235 Err(_) => break,
236 }
237 if Instant::now() >= deadline {
238 if !escalated {
239 let _ = child.kill();
243 escalated = true;
244 } else {
245 break;
249 }
250 }
251 std::thread::sleep(POLL_INTERVAL);
252 }
253 let _ = child.wait();
255
256 let stdout = drain(&mut child.stdout.take());
257 let preview = truncate_for_preview(&stdout);
258 ToolCallResult::error(format!(
259 "Cancelled: `{cmd_display}` terminated by notifications/cancelled; partial stdout: {preview}"
260 ))
261 }
262 }
263}
264
265enum CancelReason {
266 Signalled,
267}
268
269fn drain<R: Read>(reader: &mut Option<R>) -> String {
270 let mut buf = String::new();
271 if let Some(r) = reader.as_mut() {
272 let _ = r.read_to_string(&mut buf);
273 }
274 buf
275}
276
277fn truncate_for_preview(s: &str) -> String {
278 const MAX: usize = 512;
279 if s.len() <= MAX {
280 s.to_string()
281 } else {
282 let truncated: String = s.chars().take(MAX).collect();
283 format!("{truncated}… (truncated)")
284 }
285}
286
287#[cfg(unix)]
288fn send_sigterm(pid: u32) {
289 use nix::sys::signal::{kill, Signal};
290 use nix::unistd::Pid;
291
292 #[allow(clippy::cast_possible_wrap)]
297 let raw = pid as i32;
298 let _ = kill(Pid::from_raw(raw), Signal::SIGTERM);
299}
300
301#[cfg(not(unix))]
302fn send_sigterm(_pid: u32) {
303 }
306
307#[must_use]
319pub fn run_apr_streaming<F>(args: &[&str], on_line: F) -> ToolCallResult
320where
321 F: FnMut(&str),
322{
323 spawn_streaming(apr_binary(), args, on_line)
324}
325
326#[must_use]
329pub fn spawn_streaming<P: AsRef<OsStr>, F>(
330 program: P,
331 args: &[&str],
332 mut on_line: F,
333) -> ToolCallResult
334where
335 F: FnMut(&str),
336{
337 let program = program.as_ref();
338 let cmd_display = display_cmd(program, args);
339
340 let mut child = match Command::new(program)
341 .args(args)
342 .stdout(Stdio::piped())
343 .stderr(Stdio::piped())
344 .spawn()
345 {
346 Ok(c) => c,
347 Err(e) => {
348 return ToolCallResult::error(format!("Failed to spawn `{cmd_display}`: {e}"));
349 }
350 };
351
352 let stdout_pipe = match child.stdout.take() {
355 Some(p) => p,
356 None => {
357 let _ = child.wait();
358 return ToolCallResult::error(format!("Failed to capture stdout of `{cmd_display}`"));
359 }
360 };
361
362 let mut accumulated = String::new();
363 let reader = BufReader::new(stdout_pipe);
364 for line in reader.lines() {
365 match line {
366 Ok(text) => {
367 on_line(&text);
368 accumulated.push_str(&text);
369 accumulated.push('\n');
370 }
371 Err(e) => {
372 let _ = child.wait();
375 return ToolCallResult::error(format!(
376 "Failed to read stdout of `{cmd_display}`: {e}"
377 ));
378 }
379 }
380 }
381
382 let status = match child.wait() {
385 Ok(s) => s,
386 Err(e) => {
387 return ToolCallResult::error(format!("Failed to reap `{cmd_display}`: {e}"));
388 }
389 };
390
391 let stderr = drain(&mut child.stderr.take());
392
393 if status.success() {
394 if accumulated.trim().is_empty() {
395 ToolCallResult::error(format!("`{cmd_display}` produced no output"))
396 } else {
397 ToolCallResult::success(accumulated)
398 }
399 } else {
400 let code = status.code().unwrap_or(-1);
401 let detail = if stderr.trim().is_empty() {
402 accumulated
403 } else {
404 stderr
405 };
406 ToolCallResult::error(format!("`{cmd_display}` failed (exit {code}): {detail}"))
407 }
408}
409
410#[cfg(test)]
411#[allow(clippy::disallowed_methods)] mod tests {
413 use super::*;
414 use std::sync::mpsc;
415 use std::thread;
416
417 #[test]
421 fn failure_keeps_the_stdout_report_when_stderr_also_spoke() {
422 let report = r#"{"passed":false,"gates":[{"name":"ollama_parity","passed":false}]}"#;
423 let result = failure_result(
424 "apr qa m.gguf --json",
425 5,
426 report,
427 "error: Validation failed",
428 );
429
430 assert_eq!(result.is_error, Some(true));
431 let whole: String = result
432 .content
433 .iter()
434 .map(|b| b.text.as_str())
435 .collect::<Vec<_>>()
436 .join("\n");
437 assert!(
438 whole.contains("ollama_parity"),
439 "gate report must reach the client, got: {whole}"
440 );
441 assert!(
442 whole.contains("failed (exit 5)"),
443 "summary line must survive too, got: {whole}"
444 );
445 }
446
447 #[test]
456 fn cancellable_failure_carries_both_streams() {
457 let (_tx, rx) = mpsc::channel::<()>();
458 let result = spawn_cancellable(
459 "sh",
460 &[
461 "-c",
462 "printf '{\"gates\":\"REP\"}\\nORT\\n'; echo SUMMARY >&2; exit 5",
463 ],
464 &rx,
465 CANCEL_GRACE_MS,
466 );
467 assert_eq!(result.is_error, Some(true));
468 assert!(
469 result.content[0].text.contains("SUMMARY"),
470 "stderr dropped: {}",
471 result.content[0].text
472 );
473 assert_eq!(
474 result.content.len(),
475 2,
476 "stdout report dropped, only got: {:?}",
477 result.content
478 );
479 assert!(
480 result.content[1].text.contains("{\"gates\":\"REP\"}\nORT"),
481 "stdout report mangled: {}",
482 result.content[1].text
483 );
484 }
485
486 #[test]
489 fn failure_with_empty_stderr_reports_stdout_once() {
490 let result = failure_result("apr qa m.gguf", 1, "only-stdout", " \n");
491 assert_eq!(result.content.len(), 1);
492 assert!(result.content[0].text.contains("only-stdout"));
493 }
494
495 #[test]
499 fn echoed_command_is_shell_quoted() {
500 let cmd = display_cmd(
501 OsStr::new("apr"),
502 &["run", "m.gguf", "--prompt", "What is 2+2?"],
503 );
504 assert_eq!(cmd, "apr run m.gguf --prompt 'What is 2+2?'");
505 }
506
507 #[test]
510 fn quoting_leaves_safe_args_alone_and_escapes_quotes() {
511 assert_eq!(quote_arg("--max-tokens"), "--max-tokens");
512 assert_eq!(
513 quote_arg("/home/noah/models/a.gguf"),
514 "/home/noah/models/a.gguf"
515 );
516 assert_eq!(quote_arg(""), "''");
517 assert_eq!(quote_arg("it's"), r"'it'\''s'");
518 }
519
520 #[test]
523 fn spawn_failure_maps_to_tool_error() {
524 let result = run_apr(&["this-subcommand-does-not-exist"]);
525 assert_eq!(result.is_error, Some(true));
526 }
527
528 #[test]
544 #[cfg(unix)]
545 fn falsify_2384_run_apr_executes_the_resolved_binary() {
546 use std::io::Write;
547 use std::os::unix::fs::PermissionsExt;
548
549 let dir =
552 std::env::temp_dir().join(format!("aprender-mcp-2384-run-apr-{}", std::process::id()));
553 let _ = std::fs::remove_dir_all(&dir);
554 std::fs::create_dir_all(&dir).expect("mkdir scratch");
555 let shim = dir.join("apr");
556 {
557 let mut f = std::fs::File::create(&shim).expect("create shim");
558 writeln!(f, "#!/bin/sh").expect("shebang");
559 writeln!(f, "if [ \"$1\" = \"validate\" ]; then").expect("if");
560 writeln!(f, " echo '{{\"marker\":\"APR-BIN-RESOLVED-SHIM\"}}'").expect("body");
561 writeln!(f, " exit 0").expect("ok");
562 writeln!(f, "fi").expect("fi");
563 writeln!(f, "exit 2").expect("unknown subcommand");
564 f.sync_all().expect("sync");
565 }
566 let mut perms = std::fs::metadata(&shim).expect("stat").permissions();
567 perms.set_mode(0o755);
568 std::fs::set_permissions(&shim, perms).expect("chmod");
569
570 std::env::set_var(crate::apr_bin::APR_BIN_ENV, &shim);
572 let result = run_apr(&["validate", "/dev/null", "--json"]);
573 std::env::remove_var(crate::apr_bin::APR_BIN_ENV);
574
575 assert!(
576 result.is_error.is_none(),
577 "resolved shim should succeed, got: {}",
578 result.content[0].text
579 );
580 assert!(
581 result.content[0].text.contains("APR-BIN-RESOLVED-SHIM"),
582 "run_apr must execute the resolved binary; got: {}",
583 result.content[0].text
584 );
585 }
586
587 #[test]
590 fn cancellable_natural_exit_matches_run_apr() {
591 let (_tx, rx) = mpsc::channel::<()>();
592 let result = spawn_cancellable("echo", &["hello"], &rx, CANCEL_GRACE_MS);
593 assert!(result.is_error.is_none(), "echo should succeed");
594 assert!(result.content[0].text.contains("hello"));
595 }
596
597 #[test]
601 fn cancellable_disconnected_channel_is_noop() {
602 let (tx, rx) = mpsc::channel::<()>();
603 drop(tx);
604 let result = spawn_cancellable("echo", &["world"], &rx, CANCEL_GRACE_MS);
605 assert!(result.is_error.is_none());
606 assert!(result.content[0].text.contains("world"));
607 }
608
609 #[test]
611 fn cancellable_spawn_failure_maps_to_error() {
612 let (_tx, rx) = mpsc::channel::<()>();
613 let result = spawn_cancellable(
614 "/this/binary/does/not/exist/apr-mcp-test",
615 &[],
616 &rx,
617 CANCEL_GRACE_MS,
618 );
619 assert_eq!(result.is_error, Some(true));
620 assert!(result.content[0].text.contains("Failed to spawn"));
621 }
622
623 #[test]
626 fn streaming_invokes_callback_per_line() {
627 let lines = std::sync::Mutex::new(Vec::<String>::new());
628 let result = spawn_streaming("printf", &["line1\nline2\nline3\n"], |line| {
629 lines
630 .lock()
631 .expect("test mutex not poisoned")
632 .push(line.to_string());
633 });
634 assert!(result.is_error.is_none(), "printf should succeed");
635
636 let captured = lines.lock().expect("mutex").clone();
637 assert_eq!(captured, vec!["line1", "line2", "line3"]);
638 assert!(result.content[0].text.contains("line1"));
639 assert!(result.content[0].text.contains("line3"));
640 }
641
642 #[test]
645 fn streaming_spawn_failure_does_not_call_callback() {
646 let called = std::sync::Mutex::new(false);
647 let result = spawn_streaming(
648 "/this/binary/does/not/exist/apr-mcp-streaming-test",
649 &[],
650 |_| {
651 *called.lock().expect("mutex") = true;
652 },
653 );
654 assert_eq!(result.is_error, Some(true));
655 assert!(!*called.lock().expect("mutex"));
656 assert!(result.content[0].text.contains("Failed to spawn"));
657 }
658
659 #[test]
661 #[cfg(unix)]
662 fn streaming_nonzero_exit_is_error() {
663 let result = spawn_streaming("sh", &["-c", "echo partial; exit 3"], |_| {});
664 assert_eq!(result.is_error, Some(true));
665 assert!(
666 result.content[0].text.contains("exit 3"),
667 "message should include exit code: {}",
668 result.content[0].text
669 );
670 }
671
672 #[test]
677 #[cfg(unix)]
678 fn cancellable_stops_long_running_subprocess_within_grace() {
679 let (tx, rx) = mpsc::channel::<()>();
680
681 let handle = thread::spawn(move || {
684 thread::sleep(Duration::from_millis(100));
685 let _ = tx.send(());
686 });
687
688 let t0 = Instant::now();
689 let result = spawn_cancellable("sleep", &["60"], &rx, 2_000);
692 let elapsed = t0.elapsed();
693
694 handle.join().expect("cancel-sender thread joins");
695
696 assert_eq!(result.is_error, Some(true), "cancelled calls are errors");
697 assert!(
698 result.content[0].text.starts_with("Cancelled:"),
699 "message should indicate cancellation, got: {}",
700 result.content[0].text
701 );
702 assert!(
705 elapsed < Duration::from_millis(2_500),
706 "cancel should finish within grace + slack, took {elapsed:?}"
707 );
708 assert!(
711 elapsed < Duration::from_secs(5),
712 "cancelled call must return far before sleep 60's natural exit"
713 );
714 }
715}