From 2255d2e774e8f266a7f347ce849e964dcde3463f Mon Sep 17 00:00:00 2001 From: oudecheng <13802883547@139.com> Date: Wed, 5 Aug 2026 09:42:53 +0800 Subject: [PATCH] =?UTF-8?q?fix(tools):=20=E4=BF=AE=E5=A4=8D=20shell=20?= =?UTF-8?q?=E5=B7=A5=E5=85=B7=E8=B6=85=E6=97=B6=E6=9C=AA=E7=94=9F=E6=95=88?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 根因 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 --- src/tools/bash.rs | 210 ++++++++++++++++------------------------------ 1 file changed, 73 insertions(+), 137 deletions(-) diff --git a/src/tools/bash.rs b/src/tools/bash.rs index d9e6236..98d07f6 100644 --- a/src/tools/bash.rs +++ b/src/tools/bash.rs @@ -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>, - stderr_buf: &Arc>, -) { - 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> = Some(rx_inner); + let mut read_tasks: Vec> = 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 + )); } } }