use std::collections::BTreeMap;
use std::path::Path;
use std::process::Stdio;
use std::time::Duration;
use tokio::io::AsyncReadExt;
use tokio_util::sync::CancellationToken;
use theway_core::AgentToolError;
pub(crate) mod process_group {
use std::time::Duration;
use theway_core::AgentToolError;
use tokio::time::timeout;
#[cfg(windows)]
use std::process::Stdio;
#[cfg(windows)]
const CREATE_NO_WINDOW: u32 = 0x0800_0000;
pub(crate) fn prepare_command(cmd: &mut tokio::process::Command) {
#[cfg(unix)]
{
unsafe {
cmd.pre_exec(|| {
if libc::setsid() == -1 {
return Err(std::io::Error::last_os_error());
}
Ok(())
});
}
}
#[cfg(not(unix))]
let _ = cmd;
}
pub(crate) async fn kill(pid: u32) -> Result<(), AgentToolError> {
#[cfg(unix)]
{
unsafe {
libc::killpg(pid as libc::pid_t, libc::SIGKILL);
}
Ok(())
}
#[cfg(windows)]
{
let mut cmd = tokio::process::Command::new("taskkill");
cmd.arg("/PID")
.arg(pid.to_string())
.arg("/T")
.arg("/F")
.stdout(Stdio::null())
.stderr(Stdio::null())
.creation_flags(CREATE_NO_WINDOW);
let status = timeout(Duration::from_secs(5), cmd.status())
.await
.map_err(|_| AgentToolError::from("taskkill timed out"))?
.map_err(|e| AgentToolError::from(format!("taskkill spawn: {e}")))?;
if !status.success() {
return Err(AgentToolError::from(format!(
"taskkill failed with {status}"
)));
}
Ok(())
}
#[cfg(not(any(unix, windows)))]
{
let _ = pid;
Ok(())
}
}
pub(crate) async fn terminate_child_tree(child: &mut tokio::process::Child, pid: Option<u32>) {
if let Some(pid) = pid {
let _ = kill(pid).await;
}
let _ = child.start_kill();
let _ = timeout(Duration::from_secs(2), child.wait()).await;
}
}
#[derive(Debug)]
pub struct RunOutcome {
pub stdout: String,
pub stderr: String,
pub exit_code: Option<i32>,
pub stderr_suffix: Option<String>,
pub kill_reason: Option<KillReason>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum KillReason {
TimedOut { secs: u64 },
Cancelled,
}
impl RunOutcome {
pub fn rendered_exit(&self) -> i32 {
self.exit_code.unwrap_or(-1)
}
}
pub async fn run_with_kill_on_timeout_or_cancel(
command: &str,
timeout: Option<Duration>,
cwd: Option<&Path>,
envs: Option<&BTreeMap<String, String>>,
cancel: &CancellationToken,
) -> Result<RunOutcome, AgentToolError> {
let mut cmd = tokio::process::Command::new("sh");
cmd.arg("-c")
.arg(command)
.stdin(Stdio::null())
.stdout(Stdio::piped())
.stderr(Stdio::piped());
if let Some(dir) = cwd {
cmd.current_dir(dir);
}
if let Some(envs) = envs {
cmd.envs(envs);
}
cmd
.kill_on_drop(true);
process_group::prepare_command(&mut cmd);
let mut child = cmd
.spawn()
.map_err(|e| AgentToolError::from(format!("spawn: {e}")))?;
let child_pid = child.id();
let stdout = child.stdout.take();
let stderr = child.stderr.take();
let drain_handle = tokio::spawn(async move {
let stdout_task = async move {
let mut s = String::new();
if let Some(mut h) = stdout {
let _ = h.read_to_string(&mut s).await;
}
s
};
let stderr_task = async move {
let mut s = String::new();
if let Some(mut h) = stderr {
let _ = h.read_to_string(&mut s).await;
}
s
};
tokio::join!(stdout_task, stderr_task)
});
let select_outcome: SelectOutcome;
let exit_code: Option<i32>;
{
let wait = child.wait();
tokio::pin!(wait);
let timeout_future =
tokio::time::sleep(timeout.unwrap_or(Duration::from_secs(u64::MAX / 2)));
tokio::pin!(timeout_future);
let has_timeout = timeout.is_some();
let (kr, code) = tokio::select! {
biased;
_ = cancel.cancelled() => (SelectOutcome::Cancelled, None),
_ = &mut timeout_future, if has_timeout => (
SelectOutcome::TimedOut { secs: timeout.expect("guarded by has_timeout").as_secs() },
None,
),
status = &mut wait => {
let c = status.ok().and_then(|s| s.code());
(SelectOutcome::Finished, c)
}
};
select_outcome = kr;
exit_code = code;
}
if !matches!(select_outcome, SelectOutcome::Finished) {
process_group::terminate_child_tree(&mut child, child_pid).await;
}
let drain_result = tokio::time::timeout(Duration::from_secs(2), drain_handle).await;
let (stdout, stderr) = match drain_result {
Ok(Ok((o, e))) => (o, e),
_ => (String::new(), String::new()),
};
let stderr_suffix = match select_outcome {
SelectOutcome::Finished => None,
SelectOutcome::Cancelled => Some("[aborted]".into()),
SelectOutcome::TimedOut { secs } => Some(format!("[timed out after {secs}s]")),
};
Ok(RunOutcome {
stdout,
stderr,
exit_code,
stderr_suffix,
kill_reason: match select_outcome {
SelectOutcome::Finished => None,
SelectOutcome::Cancelled => Some(KillReason::Cancelled),
SelectOutcome::TimedOut { secs } => Some(KillReason::TimedOut { secs }),
},
})
}
#[derive(Clone, Copy)]
enum SelectOutcome {
Finished,
TimedOut { secs: u64 },
Cancelled,
}