fix(tools): 修复 shell 工具超时未生效
根因 A：子进程派生守护进程（如 adb daemon）继承 stdout 管道写端，父进程退出后 EOF 永不到达，child.wait() 分支进入无界 drain 循环，select! 已退出导致 deadline 失效，永久阻塞。 根因 B：deadline 分支调用 is_process_waiting_on_stdin 误判 socket 等待为 stdin 等待，且 should_return_pending 关键词过宽，绕过硬 kill 转 pending 会话，表面未超时实则进程未终止。 修复：- 新增 STREAM_DRAIN_MS=1000，用 tokio::time::timeout 包裹 drain 循环 - 新增 kill_and_reap：start_kill → 5s wait 超时 → kill().await + 3s 兜底 - 收集 read_tasks JoinHandles，返回前 abort 避免任务泄漏 - deadline 分支移除绕过逻辑，一律硬 kill 并返回超时错误 - timeout 参数改用 .clamp(1, MAX_TIMEOUT_SECS) 防御 timeout:0
This commit is contained in:
oudecheng
parent
1e9075e1ed
commit
2255d2e774
@ -17,6 +17,12 @@ use crate::tools::{check_null_args, extract_bool, extract_u64};
|
||||
|
||||
const MAX_TIMEOUT_SECS: u64 = 600;
|
||||
const MAX_OUTPUT_CHARS: usize = 50_000;
|
||||
/// 子进程退出后，等待 read_stream 把管道残余输出排空的最长时间。
|
||||
///
|
||||
/// 不能无界等待 EOF：若子进程派生了继承 stdout 管道的守护进程
|
||||
/// （如 `adb start-server` 启动的 adb daemon），EOF 永不到达，
|
||||
/// 会永久阻塞并绕过 deadline（select! 已由 child.wait() 分支退出）。
|
||||
const STREAM_DRAIN_MS: u64 = 1000;
|
||||
const PENDING_USER_ACTION_MARKER: &str = "__PICOBOT_PENDING_USER_ACTION__";
|
||||
const INTERACTIVE_HINT: &str =
|
||||
"进程正在等待输入。请使用 session_id 和 stdin_input 参数回复交互内容。";
|
||||
@ -203,7 +209,7 @@ impl BashTool {
|
||||
)
|
||||
}
|
||||
|
||||
fn should_return_pending(&self, interactive: bool, output: &str) -> bool {
|
||||
fn should_return_pending(&self, _interactive: bool, output: &str) -> bool {
|
||||
let normalized = output.to_lowercase();
|
||||
let has_auth_phrase = [
|
||||
// 中文 — 原有
|
||||
@ -254,7 +260,10 @@ impl BashTool {
|
||||
.iter()
|
||||
.any(|pattern| normalized.contains(pattern));
|
||||
|
||||
has_auth_phrase || (interactive && !output.trim().is_empty())
|
||||
// 仅 auth 短语命中才转 pending（超时前早退，不构成 deadline 绕过）。
|
||||
// 此前的 `|| (interactive && !output.trim().is_empty())` 会在 interactive=true
|
||||
// 时对任意输出转 pending，绕过超时，与严格硬超时冲突，已移除。
|
||||
has_auth_phrase
|
||||
}
|
||||
}
|
||||
|
||||
@ -272,38 +281,6 @@ async fn drain_available_chunks(
|
||||
}
|
||||
}
|
||||
|
||||
/// 自适应 drain：循环读取直到输出稳定，确保进程的所有提示内容都被捕获
|
||||
///
|
||||
/// - 每次 drain 后等待 200ms 再检查是否有新数据
|
||||
/// - 最多循环 10 次（即最多等待 2 秒）
|
||||
/// - 如果连续一次 drain 没有新数据，立即返回
|
||||
async fn drain_until_stable(
|
||||
rx: &mut mpsc::UnboundedReceiver<(bool, String)>,
|
||||
stdout_buf: &Arc<Mutex<String>>,
|
||||
stderr_buf: &Arc<Mutex<String>>,
|
||||
) {
|
||||
const DRAIN_INTERVAL_MS: u64 = 200;
|
||||
const MAX_DRAIN_ROUNDS: u32 = 10;
|
||||
|
||||
for _ in 0..MAX_DRAIN_ROUNDS {
|
||||
let prev_stdout_len = stdout_buf.lock().await.len();
|
||||
let prev_stderr_len = stderr_buf.lock().await.len();
|
||||
|
||||
drain_available_chunks(rx, stdout_buf, stderr_buf).await;
|
||||
|
||||
let new_stdout_len = stdout_buf.lock().await.len();
|
||||
let new_stderr_len = stderr_buf.lock().await.len();
|
||||
|
||||
// 如果没有新数据，说明输出已稳定
|
||||
if new_stdout_len == prev_stdout_len && new_stderr_len == prev_stderr_len {
|
||||
break;
|
||||
}
|
||||
|
||||
// 有新数据，等待后再次检查
|
||||
tokio::time::sleep(Duration::from_millis(DRAIN_INTERVAL_MS)).await;
|
||||
}
|
||||
}
|
||||
|
||||
impl Default for BashTool {
|
||||
fn default() -> Self {
|
||||
Self::new(Arc::new(ShellSessionManager::new()))
|
||||
@ -402,7 +379,7 @@ impl Tool for BashTool {
|
||||
|
||||
let timeout_secs = extract_u64(&args, "timeout")
|
||||
.unwrap_or(self.timeout_secs)
|
||||
.min(MAX_TIMEOUT_SECS);
|
||||
.clamp(1, MAX_TIMEOUT_SECS); // 下界 1 防止 timeout:0 误杀刚 spawn 的子进程；上界 600s（与 schema minimum/maximum 对齐）
|
||||
let interactive = extract_bool(&args, "interactive").unwrap_or(false);
|
||||
|
||||
let cwd = self
|
||||
@ -430,6 +407,22 @@ impl Tool for BashTool {
|
||||
}
|
||||
|
||||
impl BashTool {
|
||||
/// 强制终止子进程并回收，避免 `wait()` 永久挂起导致超时未生效。
|
||||
///
|
||||
/// `start_kill` 在 Unix 发 SIGKILL、在 Windows 调 TerminateProcess（均为强制终止）。
|
||||
/// 用 `timeout` 包裹 `wait()` 防止 reap 在异常情况下永久挂起；5s 内未回收则再
|
||||
/// `kill().await`（重发信号并等待）+ 3s 兜底，保证工具调用必然返回。
|
||||
async fn kill_and_reap(child: &mut tokio::process::Child) {
|
||||
let _ = child.start_kill();
|
||||
if tokio::time::timeout(Duration::from_secs(5), child.wait())
|
||||
.await
|
||||
.is_err()
|
||||
{
|
||||
let _ = child.kill().await;
|
||||
let _ = tokio::time::timeout(Duration::from_secs(3), child.wait()).await;
|
||||
}
|
||||
}
|
||||
|
||||
async fn run_command(
|
||||
&self,
|
||||
command: &str,
|
||||
@ -453,11 +446,12 @@ impl BashTool {
|
||||
let (tx, rx_inner) = mpsc::unbounded_channel::<(bool, String)>();
|
||||
let mut rx: Option<mpsc::UnboundedReceiver<(bool, String)>> = Some(rx_inner);
|
||||
|
||||
let mut read_tasks: Vec<tokio::task::JoinHandle<()>> = Vec::new();
|
||||
if let Some(stdout) = stdout {
|
||||
tokio::spawn(read_stream(stdout, false, tx.clone()));
|
||||
read_tasks.push(tokio::spawn(read_stream(stdout, false, tx.clone())));
|
||||
}
|
||||
if let Some(stderr) = stderr {
|
||||
tokio::spawn(read_stream(stderr, true, tx.clone()));
|
||||
read_tasks.push(tokio::spawn(read_stream(stderr, true, tx.clone())));
|
||||
}
|
||||
drop(tx);
|
||||
|
||||
@ -471,14 +465,31 @@ impl BashTool {
|
||||
let status = status.map_err(|e| format!("Failed to wait: {}", e))?;
|
||||
let mut rx_val = rx.take().unwrap();
|
||||
drain_available_chunks(&mut rx_val, &stdout_buf, &stderr_buf).await;
|
||||
while let Some((is_stderr, chunk)) = rx_val.recv().await {
|
||||
if is_stderr {
|
||||
stderr_buf.lock().await.push_str(&chunk);
|
||||
} else {
|
||||
stdout_buf.lock().await.push_str(&chunk);
|
||||
}
|
||||
// 子进程已退出：给 read_stream 一个短窗口排空残余输出，但不等待 EOF。
|
||||
// 见 STREAM_DRAIN_MS 注释——守护进程（如 adb daemon）持管道时 EOF 永不到达，
|
||||
// 无界等待会永久阻塞并绕过 deadline（select! 已由本分支退出）。
|
||||
let _ = tokio::time::timeout(
|
||||
Duration::from_millis(STREAM_DRAIN_MS),
|
||||
async {
|
||||
while let Some((is_stderr, chunk)) = rx_val.recv().await {
|
||||
if is_stderr {
|
||||
stderr_buf.lock().await.push_str(&chunk);
|
||||
} else {
|
||||
stdout_buf.lock().await.push_str(&chunk);
|
||||
}
|
||||
}
|
||||
},
|
||||
)
|
||||
.await;
|
||||
// 终止可能仍阻塞在 reader.read() 上的 read_stream 任务，避免泄漏。
|
||||
for t in &read_tasks {
|
||||
t.abort();
|
||||
}
|
||||
let output = format_command_output(&stdout_buf.lock().await, &stderr_buf.lock().await, Some(status.code().unwrap_or(-1)));
|
||||
let output = format_command_output(
|
||||
&stdout_buf.lock().await,
|
||||
&stderr_buf.lock().await,
|
||||
Some(status.code().unwrap_or(-1)),
|
||||
);
|
||||
return Ok(self.truncate_output(&output));
|
||||
}
|
||||
Some((is_stderr, chunk)) = async {
|
||||
@ -498,7 +509,7 @@ impl BashTool {
|
||||
let mut rx_val = rx.take().unwrap();
|
||||
drain_available_chunks(&mut rx_val, &stdout_buf, &stderr_buf).await;
|
||||
let combined = format_command_output(&stdout_buf.lock().await, &stderr_buf.lock().await, None);
|
||||
// Try to save as interactive session
|
||||
// 保存为交互式会话：read_stream 任务作为会话生产者，必须继续运行，不 abort。
|
||||
if let Some(stdin) = child_stdin {
|
||||
let session_id = self.session_manager.save_session(
|
||||
child, stdin, rx_val,
|
||||
@ -507,107 +518,32 @@ impl BashTool {
|
||||
).await;
|
||||
return Ok(self.pending_output(&combined, Some(&session_id)));
|
||||
}
|
||||
let _ = child.start_kill();
|
||||
let _ = child.wait().await;
|
||||
return Ok(self.pending_output(&combined, None));
|
||||
}
|
||||
}
|
||||
_ = tokio::time::sleep(Duration::from_secs(2)) => {
|
||||
// Periodic safety net: check OS-level process state
|
||||
if let Some(pid) = child.id() {
|
||||
if crate::platform::is_process_waiting_on_stdin(pid) == Some(true) {
|
||||
// 自适应 drain，等待输出稳定
|
||||
if let Some(rx_ref) = rx.as_mut() {
|
||||
drain_until_stable(rx_ref, &stdout_buf, &stderr_buf).await;
|
||||
}
|
||||
let combined = format_command_output(&stdout_buf.lock().await, &stderr_buf.lock().await, None);
|
||||
// 始终创建 session，即使输出为空（进程可能还没写出提示）
|
||||
if let Some(stdin) = child_stdin {
|
||||
if let Some(rx_val) = rx.take() {
|
||||
let session_id = self.session_manager.save_session(
|
||||
child, stdin, rx_val,
|
||||
stdout_buf.lock().await.clone(),
|
||||
stderr_buf.lock().await.clone(),
|
||||
).await;
|
||||
return Ok(self.pending_output(&combined, Some(&session_id)));
|
||||
}
|
||||
}
|
||||
let _ = child.start_kill();
|
||||
let _ = child.wait().await;
|
||||
return Ok(self.pending_output(&combined, None));
|
||||
// 无 stdin 可存会话：硬终止并回收，终止 read_stream 任务避免泄漏。
|
||||
Self::kill_and_reap(&mut child).await;
|
||||
for t in &read_tasks {
|
||||
t.abort();
|
||||
}
|
||||
}
|
||||
|
||||
let combined = format_command_output(&stdout_buf.lock().await, &stderr_buf.lock().await, None);
|
||||
if self.should_return_pending(interactive, &combined) {
|
||||
if let Some(rx_ref) = rx.as_mut() {
|
||||
drain_available_chunks(rx_ref, &stdout_buf, &stderr_buf).await;
|
||||
}
|
||||
let combined = format_command_output(&stdout_buf.lock().await, &stderr_buf.lock().await, None);
|
||||
if let Some(stdin) = child_stdin {
|
||||
if let Some(rx_val) = rx.take() {
|
||||
let session_id = self.session_manager.save_session(
|
||||
child, stdin, rx_val,
|
||||
stdout_buf.lock().await.clone(),
|
||||
stderr_buf.lock().await.clone(),
|
||||
).await;
|
||||
return Ok(self.pending_output(&combined, Some(&session_id)));
|
||||
}
|
||||
}
|
||||
let _ = child.start_kill();
|
||||
let _ = child.wait().await;
|
||||
return Ok(self.pending_output(&combined, None));
|
||||
}
|
||||
}
|
||||
_ = sleep_until(deadline) => {
|
||||
// 严格硬超时：deadline 到达一律 kill 并返回超时错误，不再转 pending 会话。
|
||||
if let Some(rx_ref) = rx.as_mut() {
|
||||
drain_available_chunks(rx_ref, &stdout_buf, &stderr_buf).await;
|
||||
}
|
||||
let combined = format_command_output(&stdout_buf.lock().await, &stderr_buf.lock().await, None);
|
||||
|
||||
// OS-level check: if blocked on stdin, save as session
|
||||
if let Some(pid) = child.id() {
|
||||
if crate::platform::is_process_waiting_on_stdin(pid) == Some(true) {
|
||||
// 自适应 drain，等待输出稳定
|
||||
if let Some(rx_ref) = rx.as_mut() {
|
||||
drain_until_stable(rx_ref, &stdout_buf, &stderr_buf).await;
|
||||
}
|
||||
let combined = format_command_output(&stdout_buf.lock().await, &stderr_buf.lock().await, None);
|
||||
if let Some(stdin) = child_stdin {
|
||||
if let Some(rx_val) = rx.take() {
|
||||
let session_id = self.session_manager.save_session(
|
||||
child, stdin, rx_val,
|
||||
stdout_buf.lock().await.clone(),
|
||||
stderr_buf.lock().await.clone(),
|
||||
).await;
|
||||
return Ok(self.pending_output(&combined, Some(&session_id)));
|
||||
}
|
||||
}
|
||||
let _ = child.start_kill();
|
||||
let _ = child.wait().await;
|
||||
return Ok(self.pending_output(&combined, None));
|
||||
}
|
||||
let combined = format_command_output(
|
||||
&stdout_buf.lock().await,
|
||||
&stderr_buf.lock().await,
|
||||
None,
|
||||
);
|
||||
Self::kill_and_reap(&mut child).await;
|
||||
for t in &read_tasks {
|
||||
t.abort();
|
||||
}
|
||||
|
||||
if self.should_return_pending(interactive, &combined) {
|
||||
if let Some(stdin) = child_stdin {
|
||||
if let Some(rx_val) = rx.take() {
|
||||
let session_id = self.session_manager.save_session(
|
||||
child, stdin, rx_val,
|
||||
stdout_buf.lock().await.clone(),
|
||||
stderr_buf.lock().await.clone(),
|
||||
).await;
|
||||
return Ok(self.pending_output(&combined, Some(&session_id)));
|
||||
}
|
||||
}
|
||||
let _ = child.start_kill();
|
||||
let _ = child.wait().await;
|
||||
return Ok(self.pending_output(&combined, None));
|
||||
}
|
||||
|
||||
let _ = child.start_kill();
|
||||
let _ = child.wait().await;
|
||||
return Err(format!("Command timed out after {} seconds", timeout_secs));
|
||||
return Err(format!(
|
||||
"Command timed out after {} seconds\n{}",
|
||||
timeout_secs, combined
|
||||
));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user