572 lines
26 KiB
Rust
572 lines
26 KiB
Rust
//! Shell 执行器 — 通过 tokio::process 执行 shell 命令
|
||
|
||
use serde::{Deserialize, Serialize};
|
||
use std::process::Stdio;
|
||
|
||
use tokio::io::{AsyncBufReadExt, BufReader};
|
||
|
||
/// Shell 命令执行结果
|
||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||
pub struct ShellResult {
|
||
/// 标准输出
|
||
pub stdout: String,
|
||
/// 标准错误
|
||
pub stderr: String,
|
||
/// 退出码
|
||
pub exit_code: Option<i32>,
|
||
/// 执行耗时（毫秒）
|
||
pub duration_ms: u64,
|
||
}
|
||
|
||
/// Shell 类型选择
|
||
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
|
||
#[serde(rename_all = "snake_case")]
|
||
pub enum ShellType {
|
||
/// Windows cmd.exe (默认 Windows)
|
||
Cmd,
|
||
/// PowerShell 5.x(Windows 自带,不支持 && 运算符)
|
||
PowerShell,
|
||
/// PowerShell 7(pwsh,支持 && 运算符;运行时探测,未装回退 PowerShell)
|
||
Pwsh,
|
||
/// Unix sh (默认非 Windows)
|
||
Sh,
|
||
}
|
||
|
||
impl Default for ShellType {
|
||
fn default() -> Self {
|
||
// L1 环境感知:Windows 默认 PowerShell 系(非 Cmd)。PowerShell 对引号/$变量/Unicode 处理
|
||
// 远优于 cmd,从根上避 kms 类引号转义地狱。AI 写文件执行见 env_profile。
|
||
// 优先 pwsh(PS7,支持 && 运算符)——LLM 训练数据 Unix 多,普遍生成 `cd x && y`,
|
||
// PS5 不支持 && 致命令失败(实测会话 6acb7f9b `cd ... && git init` InvalidEndOfLine)。
|
||
// 探测失败(未装 pwsh)回退 PS5。探测结果缓存于模块级 PWSH_CACHE(只探一次)。
|
||
//
|
||
// 【序约束 / 死缓存修复】Default 为同步签名,读模块级单源 PWSH_CACHE(由 probe_pwsh 写入)。
|
||
// 缓存有两个填充点(均走 probe_pwsh_blocking 单真相源,无漂移):
|
||
// 1) env_snapshot::detect_shell → EnvSnapshot::detect()(启动期 spawn_blocking 内,先于 execute);
|
||
// 2) execute()/execute_streaming() 异步入口 → probe_pwsh().await(spawn_blocking + 3s 超时)。
|
||
// 通常 detect_shell 先跑(EnvSnapshot::detect 在 run_agentic_loop 早期被 await),首次 execute 时
|
||
// probe_pwsh 直接命中缓存。若 detect 尚未填充,probe_pwsh 自带探测兜底。任一时刻 build_command →
|
||
// ShellType::default() 读到缓存值;未探测(缓存空)返回 PowerShell(PS5)。绝不在 Default 同步路径内 spawn 探测。
|
||
//
|
||
// 历史 bug(2026-08 修复):原 probe_pwsh_cached() 与 probe_pwsh() 各持一个独立 static OnceLock,
|
||
// Default 读的那个永不被填充 → Pwsh 全死代码,Windows 永走 PS5(&& 必失败)。根因:两个 OnceLock
|
||
// 非单源 + env_snapshot 另持 probe_command_success 独立探测。根本修:合并为模块级 PWSH_CACHE 单源,
|
||
// 写读同源;detect_shell 复用 probe_pwsh_blocking(单一探测实现)。
|
||
if cfg!(windows) {
|
||
if cache_get().unwrap_or(false) { ShellType::Pwsh } else { ShellType::PowerShell }
|
||
} else {
|
||
ShellType::Sh
|
||
}
|
||
}
|
||
}
|
||
|
||
/// pwsh(PowerShell 7)可用性的全局单源缓存。
|
||
///
|
||
/// 由异步 `probe_pwsh()`(execute 路径)或同步 `probe_pwsh_blocking()`(detect_shell 路径)探测,
|
||
/// 同步路径 `ShellType::default()` / `current_shell()` 读取——写读同源,
|
||
/// 杜绝历史上「两个 OnceLock 各填各的、Default 读的永空」死缓存 bug。
|
||
///
|
||
/// 语义:`None` 表示尚未探测(首次启动 / Windows 未装 pwsh 也仅表示探测未跑或返 false);
|
||
/// `Some(true)` 表示已探测且 pwsh 可用。
|
||
///
|
||
/// 用 `Mutex<Option<bool>>` 而非 OnceLock:OnceLock 写入后进程级不可清,写缓存的测试
|
||
/// 与依赖缓存一致性的测试(env_snapshot 漂移测试)全量并行跑时互踩。Mutex 化后测试可
|
||
/// 备份/复位隔离(配 PWSH_CACHE_TEST_LOCK 串行闸);生产读写语义不变,有缓存读缓存、
|
||
/// 无缓存探测后写。
|
||
static PWSH_CACHE: std::sync::Mutex<Option<bool>> = std::sync::Mutex::new(None);
|
||
|
||
/// 读缓存当前值(毒化容错:持锁测试 panic 后其余测试仍可继续,行为不变)。
|
||
fn cache_get() -> Option<bool> {
|
||
*PWSH_CACHE.lock().unwrap_or_else(|e| e.into_inner())
|
||
}
|
||
|
||
/// 写缓存当前值(探测产物写入 / 测试复位共用,覆盖写)。
|
||
fn cache_set(v: Option<bool>) {
|
||
*PWSH_CACHE.lock().unwrap_or_else(|e| e.into_inner()) = v;
|
||
}
|
||
|
||
/// 测试串行闸:缓存为进程级单源,写缓存的测试与依赖缓存一致性的测试并行互踩,
|
||
/// 凡改写缓存或依赖缓存/EnvSnapshot 一致性的测试须持锁(仅测试隔离用,非生产路径)。
|
||
/// 对齐 path_auth.rs 的 BLACKLIST_BYPASS_TEST_LOCK 模式;env_snapshot 漂移测试持同闸。
|
||
#[cfg(test)]
|
||
pub(crate) static PWSH_CACHE_TEST_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(());
|
||
|
||
/// 同步探测 pwsh(PowerShell 7)是否可用(std::Command::status),成功返回 true。
|
||
///
|
||
/// 这是阻塞 IO 的「单真相源」实现——`probe_pwsh()`(异步,spawn_blocking + 超时)与
|
||
/// env_snapshot::detect_shell(windows 同步路径)都调用本函数,二者探测逻辑永远一致,
|
||
/// 杜绝「提示告诉 LLM 用 pwsh,执行却走 powershell」的两套逻辑漂移。
|
||
///
|
||
/// 实现等价于原 `probe_pwsh()` 内的闭包:`pwsh -NoProfile -Command exit 0` 成功即 true;
|
||
/// Windows 加 CREATE_NO_WINDOW(0x0800_0000)防黑窗闪现。
|
||
///
|
||
/// 注:此处无超时——`probe_pwsh()` 在调用方包 `tokio::time::timeout`,
|
||
/// `detect_shell()` 则由外层 `EnvSnapshot::detect()` 的 spawn_blocking 5s 超时兜底。
|
||
/// 故本函数本身只负责「同步 spawn + status」,超时治理在调用点。
|
||
pub(crate) fn probe_pwsh_blocking() -> bool {
|
||
let mut cmd = std::process::Command::new("pwsh");
|
||
cmd.arg("-NoProfile").arg("-Command").arg("exit 0");
|
||
cmd.stdout(Stdio::null()).stderr(Stdio::null());
|
||
#[cfg(windows)]
|
||
{
|
||
use std::os::windows::process::CommandExt;
|
||
cmd.creation_flags(0x0800_0000); // CREATE_NO_WINDOW
|
||
}
|
||
cmd.status().map(|s| s.success()).unwrap_or(false)
|
||
}
|
||
|
||
/// 探测 pwsh(PowerShell 7)是否可用(模块级缓存,3s 超时,只探一次)。
|
||
///
|
||
/// LLM 普遍生成 `&&`(Unix 习惯),仅 PS7+ 支持,Windows 自带 PS5 不支持。
|
||
/// 探测:成功 spawn `pwsh -Command exit 0` 即可用。Windows 加 CREATE_NO_WINDOW 防黑窗闪现。
|
||
///
|
||
/// 异步化 + 超时治理:在异步上下文中通过 `tokio::task::spawn_blocking` 执行阻塞探测,
|
||
/// 外包 `tokio::time::timeout(3s)`。结果写入模块级单源 `PWSH_CACHE`,同步 `Default` 路径共享。
|
||
///
|
||
/// 超时/panic 时**不写入 PWSH_CACHE**:
|
||
/// - 超时根因往往是 Windows Store Alias / 杀软 hook 拦截 `pwsh` 命令(status() 永不返回);
|
||
/// 若错误地把 false 缓存,后续 execute() 会一直走 PS5(&& 必失败),把瞬时环境问题冻结成
|
||
/// 「永不可用」错误判定。不 set → 每次 execute 重试,环境恢复后自动回正。
|
||
/// - panic(join err)同理,可能是临时线程池异常,不应冻结判定。
|
||
/// - 仅「正常完成(Ok(Ok(_)))」时 set 缓存(此时结果是可信的探测产物)。
|
||
/// 对齐 env_snapshot.rs:64-80 的「spawn_blocking + timeout + 超时不 set」模式。
|
||
async fn probe_pwsh() -> bool {
|
||
if let Some(cached) = cache_get() {
|
||
return cached;
|
||
}
|
||
match tokio::time::timeout(
|
||
std::time::Duration::from_secs(3),
|
||
tokio::task::spawn_blocking(probe_pwsh_blocking),
|
||
).await {
|
||
Ok(Ok(result)) => {
|
||
// 正常完成:写入缓存(多任务竞态覆盖写,各写者探测值等价)
|
||
cache_set(Some(result));
|
||
result
|
||
}
|
||
Ok(Err(join_err)) => {
|
||
eprintln!(
|
||
"[shell] probe_pwsh spawn_blocking 异常(不缓存,下次 execute 重试): {}",
|
||
join_err
|
||
);
|
||
false
|
||
}
|
||
Err(_elapsed) => {
|
||
eprintln!(
|
||
"[shell] probe_pwsh 3s 超时(疑似 Windows Store Alias / 杀软 hook 拦截 pwsh, \
|
||
不缓存以免冻结错误判定,下次 execute 重试)"
|
||
);
|
||
false
|
||
}
|
||
}
|
||
}
|
||
|
||
/// 当前 shell 类型(同步读取模块级 PWSH_CACHE 单源)。
|
||
///
|
||
/// 探测在 `probe_pwsh()`(异步,带 3s 超时)或 `probe_pwsh_blocking()`(detect_shell 同步路径)中执行并填充 PWSH_CACHE;此处仅读。
|
||
/// 缓存空(未探测 / 探测超时未 set)时返回 `ShellType::default()`(Windows → PowerShell,Unix → Sh)。
|
||
/// 单源语义:env_snapshot::detect_shell 与 build_command 共用此判定,杜绝「探测与执行两套逻辑漂移」。
|
||
pub fn current_shell() -> ShellType {
|
||
ShellType::default()
|
||
}
|
||
|
||
/// 剥离 ANSI 转义序列(#8 出口兜底)。
|
||
///
|
||
/// 源头抑制(build_command 注入 NO_COLOR/TERM=dumb)覆盖主流工具链,但总有漏网
|
||
/// (不读环境变量的工具/嵌套脚本转发色码)。本纯函数兜底 CSI/OSC 两类常见序列:
|
||
/// - CSI:ESC [ 参数字母(颜色/光标/擦除,如 \x1b[7m \x1b[0K)
|
||
/// - OSC:ESC ] ... BEL 或 ESC \(终端标题/超链接,罕见但同样乱码)
|
||
/// 不碰普通文本与换行,幂等(已剥离文本再剥无变化)。
|
||
pub fn strip_ansi(text: &str) -> String {
|
||
let mut out = String::with_capacity(text.len());
|
||
let mut chars = text.char_indices().peekable();
|
||
while let Some((i, c)) = chars.next() {
|
||
if c != '\x1b' {
|
||
out.push(c);
|
||
continue;
|
||
}
|
||
// c == ESC:按下一字符分派序列类型
|
||
match text[i + 1..].chars().next() {
|
||
Some('[') => {
|
||
// CSI:跳到终止字母(@-~)为止
|
||
let rest = &text[i + 2..];
|
||
match rest.find(|ch: char| ('\u{40}'..='\u{7e}').contains(&ch)) {
|
||
Some(end) => {
|
||
// 跳过整个序列: advance 到 i+2+end 的下一字符
|
||
let skip = 2 + end + 1;
|
||
for _ in 1..skip {
|
||
chars.next();
|
||
}
|
||
}
|
||
None => {
|
||
// 流式截断的残缺序列(无终止符):丢弃剩余全部
|
||
for _ in (i + 1)..text.len() {
|
||
chars.next();
|
||
}
|
||
}
|
||
}
|
||
}
|
||
Some(']') => {
|
||
// OSC:到 BEL(\x07) 或 ST(ESC \)为止
|
||
let rest = &text[i + 2..];
|
||
let bel = rest.find('\x07');
|
||
let st = rest.find("\x1b\\");
|
||
match (bel, st) {
|
||
(Some(b), None) => {
|
||
// 已消费 ESC,再消费 i+1..=i+2+b(含 BEL)共 b+2 个
|
||
for _ in 0..(2 + b) {
|
||
chars.next();
|
||
}
|
||
}
|
||
(_, Some(s)) => {
|
||
// 再消费 i+1..=i+2+s+1(含 ST 两字符)共 s+3 个
|
||
for _ in 0..(2 + s + 1) {
|
||
chars.next();
|
||
}
|
||
}
|
||
(None, None) => {
|
||
for _ in (i + 1)..text.len() {
|
||
chars.next();
|
||
}
|
||
}
|
||
}
|
||
}
|
||
_ => {
|
||
// 两字符转义(ESC \ ESC ( 等):丢 ESC 保留后字符?保守双丢后单字符
|
||
chars.next();
|
||
}
|
||
}
|
||
}
|
||
out
|
||
}
|
||
|
||
#[cfg(test)]
|
||
mod tests {
|
||
use super::*;
|
||
|
||
/// #8 ANSI 剥离:CSI 色码/擦除/流式残缺序列/OSC/普通文本不动
|
||
#[test]
|
||
fn strip_ansi_basic() {
|
||
assert_eq!(strip_ansi("\x1b[7ma\x1b[0mb"), "ab");
|
||
assert_eq!(strip_ansi("\x1b[2K\x1b[1Gdone"), "done");
|
||
assert_eq!(strip_ansi("plain text\nline2"), "plain text\nline2");
|
||
// 流式残缺(无终止符):丢残序列
|
||
assert_eq!(strip_ansi("ok\x1b[3"), "ok");
|
||
// OSC BEL 终止
|
||
assert_eq!(strip_ansi("\x1b]0;title\x07body"), "body");
|
||
// 幂等
|
||
let once = strip_ansi("\x1b[32mgreen\x1b[0m");
|
||
assert_eq!(strip_ansi(&once), once);
|
||
}
|
||
|
||
/// 死缓存回归测试:probe_pwsh 写入后,Default 同步路径必须读到同一缓存值。
|
||
///
|
||
/// 历史 bug:probe_pwsh_cached() 与 probe_pwsh() 各持独立 OnceLock,Default 读的永空。
|
||
/// 本测试经 probe_pwsh 写模块级 PWSH_CACHE 后断言 default() 与探测值一致,锁定「写读同源」不变量。
|
||
///
|
||
/// 隔离:持串行闸,进时备份、断言前复位。OnceLock 时代写入后全进程不可清,
|
||
/// 会污染并行的 env_snapshot 漂移测试(其快照先以空缓存创建,断言即漂移)。
|
||
#[cfg(windows)]
|
||
#[tokio::test]
|
||
async fn probe_pwsh_cache_shared_with_default() {
|
||
let _gate = PWSH_CACHE_TEST_LOCK.lock().unwrap_or_else(|e| e.into_inner());
|
||
let backup = cache_get();
|
||
// 探测一次填充缓存(无论机器是否装 pwsh,只要写读同源即应一致)。
|
||
// 注:正常完成路径才会写缓存;本机 pwsh 探测不超时/不 panic,故 cached 应与 probed 一致。
|
||
let probed = probe_pwsh().await;
|
||
let cached = cache_get();
|
||
// Default 必须读到与 probe 一致的判定:Pwsh ↔ true,PowerShell ↔ false
|
||
let default_shell = ShellType::default();
|
||
// 复位放在断言前:中途 panic 不残留污染态(对齐 path_auth.rs 先复位再断言)
|
||
cache_set(backup);
|
||
// 探测完成后缓存必已填充(同步路径由此读到)
|
||
assert_eq!(cached, Some(probed));
|
||
match probed {
|
||
true => assert_eq!(default_shell, ShellType::Pwsh),
|
||
false => assert_eq!(default_shell, ShellType::PowerShell),
|
||
}
|
||
}
|
||
|
||
/// 不挂死不变量回归:probe_pwsh 必须在有限时间内返回(自带 3s 超时 + spawn_blocking)。
|
||
///
|
||
/// 历史 bug:probe_pwsh 仅 spawn_blocking 无超时,Windows Store Alias / 杀软 hook 拦截 pwsh 时
|
||
/// status() 永不返回 → spawn_blocking 线程永不返回 → probe_pwsh().await 永久挂 →
|
||
/// execute()/execute_streaming() 卡死 → run_agentic_loop 死锁(与 env_snapshot::detect 同源 bug)。
|
||
/// 本测试外包 10s timeout(远大于 3s 内部超时),无论探测成败都应在 10s 内返回,锁定「不挂死」不变量。
|
||
///
|
||
/// 隔离:持串行闸,探测在缓存为空时会写入缓存,备份/复位防污染并行测试。
|
||
#[cfg(windows)]
|
||
#[tokio::test]
|
||
async fn probe_pwsh_completes_within_timeout() {
|
||
let _gate = PWSH_CACHE_TEST_LOCK.lock().unwrap_or_else(|e| e.into_inner());
|
||
let backup = cache_get();
|
||
// 10s >> probe_pwsh 内部 3s 超时;若 10s 仍未返回 → 探测挂死,不变量被破坏。
|
||
let result = tokio::time::timeout(
|
||
std::time::Duration::from_secs(10),
|
||
probe_pwsh(),
|
||
).await;
|
||
cache_set(backup);
|
||
assert!(result.is_ok(), "probe_pwsh 必须在 10s 内返回(内部 3s 超时已兜底),不应挂死");
|
||
}
|
||
}
|
||
|
||
/// Shell 命令执行请求
|
||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||
pub struct ShellRequest {
|
||
/// 要执行的命令
|
||
pub command: String,
|
||
/// 工作目录
|
||
pub working_dir: Option<String>,
|
||
/// 环境变量
|
||
pub env: std::collections::HashMap<String, String>,
|
||
/// 超时时间（秒），None 表示不超时
|
||
pub timeout_secs: Option<u64>,
|
||
/// Shell 类型（默认：Windows→Cmd, 非Windows→Sh）
|
||
#[serde(default)]
|
||
pub shell_type: Option<ShellType>,
|
||
}
|
||
|
||
/// 输出流类型(回调 on_output 用)
|
||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||
pub enum StreamKind {
|
||
Stdout,
|
||
Stderr,
|
||
}
|
||
|
||
impl StreamKind {
|
||
/// 序列化为稳定字符串标识(emit 事件 stream 字段用)
|
||
pub fn as_str(&self) -> &'static str {
|
||
match self {
|
||
StreamKind::Stdout => "stdout",
|
||
StreamKind::Stderr => "stderr",
|
||
}
|
||
}
|
||
}
|
||
|
||
/// 构造已配置好(stdio piped + kill_on_drop + CREATE_NO_WINDOW + cwd + env)的子进程 Command。
|
||
///
|
||
/// execute() 与 execute_streaming() 共用同一构造逻辑(单真相源,DRY):
|
||
/// shell 类型选择 / kill_on_drop / Windows 无窗 / cwd / env 全在此。差异仅在后续如何消费 stdout/stderr。
|
||
fn build_command(request: ShellRequest) -> tokio::process::Command {
|
||
let shell_type = request.shell_type.unwrap_or_default();
|
||
let mut cmd = match shell_type {
|
||
ShellType::PowerShell => {
|
||
let mut c = tokio::process::Command::new("powershell");
|
||
c.arg("-NoProfile").arg("-Command").arg(&request.command);
|
||
c.stdout(Stdio::piped()).stderr(Stdio::piped());
|
||
c
|
||
}
|
||
ShellType::Pwsh => {
|
||
// PS7(支持 && 运算符),同 PS5 参数语义。-NoProfile 避加载用户 profile(速度+确定性)
|
||
let mut c = tokio::process::Command::new("pwsh");
|
||
c.arg("-NoProfile").arg("-Command").arg(&request.command);
|
||
c.stdout(Stdio::piped()).stderr(Stdio::piped());
|
||
c
|
||
}
|
||
ShellType::Cmd => {
|
||
let mut c = tokio::process::Command::new("cmd");
|
||
c.arg("/C").arg(&request.command);
|
||
c.stdout(Stdio::piped()).stderr(Stdio::piped());
|
||
c
|
||
}
|
||
ShellType::Sh => {
|
||
let mut c = tokio::process::Command::new("sh");
|
||
c.arg("-c").arg(&request.command);
|
||
c.stdout(Stdio::piped()).stderr(Stdio::piped());
|
||
c
|
||
}
|
||
};
|
||
|
||
// kill_on_drop(true) 让 Command 被 drop 时主动 kill 子进程。
|
||
// 配合 tokio::time::timeout 超时场景:超时 drop future → Command 析构 → kill 子进程,
|
||
// 不再让超时后的命令变孤儿继续后台跑(长 hang 命令/死循环仍占资源)。
|
||
// 对齐 tool_registry.rs:514「进程已终止」文案名副其实。tokio 1.52.3 支持。
|
||
cmd.kill_on_drop(true);
|
||
|
||
// Windows 下创建子进程默认弹控制台窗口(cmd/powershell 黑窗闪现)。
|
||
// CREATE_NO_WINDOW(0x0800_0000) 标志抑制窗口创建,后台静默执行。
|
||
// tokio::process::Command 在 Windows 自带 creation_flags 方法(无需 std CommandExt trait)。
|
||
#[cfg(windows)]
|
||
{
|
||
cmd.creation_flags(0x0800_0000);
|
||
}
|
||
|
||
if let Some(dir) = &request.working_dir {
|
||
cmd.current_dir(dir);
|
||
}
|
||
|
||
for (key, value) in &request.env {
|
||
cmd.env(key, value);
|
||
}
|
||
// #8 ANSI 裸露根治(源头抑制):非交互管道下多数 CLI 仍探测 TERM 发色码,cargo/ls 等
|
||
// 输出 \x1b[Nm 原样透传前端显示为 ␛[7m 乱码。注入 NO_COLOR(https://no-color.org 标准)+
|
||
// TERM=dumb+FORCE_COLOR=0 三重信号,主流工具链(cargo/npm/git/ls/pytest)管道下停发色码。
|
||
// 调用方显式传同名 env 时以调用方为准(后写覆盖,见上循环顺序)。
|
||
cmd.env("NO_COLOR", "1");
|
||
cmd.env("TERM", "dumb");
|
||
cmd.env("FORCE_COLOR", "0");
|
||
cmd
|
||
}
|
||
|
||
/// 执行 Shell 命令(等 exit 一次性返回,非流式)
|
||
///
|
||
/// 支持超时(timeout_secs)、环境变量(env)、工作目录(working_dir),
|
||
/// kill_on_drop(true) 保证超时后子进程不残留,shell_type 可选 Cmd/PowerShell/Sh。
|
||
///
|
||
/// 需要执行中实时获取 stdout/stderr 行(如 run_command 进度展示)用 [`execute_streaming`]。
|
||
pub async fn execute(request: ShellRequest) -> anyhow::Result<ShellResult> {
|
||
let start = std::time::Instant::now();
|
||
|
||
// 探测 pwsh(惰性 + OnceLock 全局缓存,只探一次),使后续 ShellType::default() 可读取缓存
|
||
#[cfg(windows)]
|
||
let _ = probe_pwsh().await;
|
||
|
||
// 先取走 build_command 之外的引用字段(超时错误信息 + timeout 判定),再 move request
|
||
let command_for_err = request.command.clone();
|
||
let timeout_secs = request.timeout_secs;
|
||
let mut cmd = build_command(request);
|
||
|
||
let output = match timeout_secs {
|
||
Some(secs) => tokio::time::timeout(
|
||
std::time::Duration::from_secs(secs),
|
||
cmd.output(),
|
||
)
|
||
.await
|
||
.map_err(|_| anyhow::anyhow!("命令执行超时({}s): {}", secs, command_for_err))??,
|
||
None => cmd.output().await?,
|
||
};
|
||
|
||
let duration = start.elapsed().as_millis() as u64;
|
||
|
||
Ok(ShellResult {
|
||
stdout: strip_ansi(&String::from_utf8_lossy(&output.stdout)),
|
||
stderr: strip_ansi(&String::from_utf8_lossy(&output.stderr)),
|
||
exit_code: output.status.code(),
|
||
duration_ms: duration,
|
||
})
|
||
}
|
||
|
||
/// 流式执行 Shell 命令 —— spawn 后逐行读 stdout/stderr,每行回调 on_output。
|
||
///
|
||
/// 治 run_command 执行中黑盒:execute() 等 exit 才返回整块 stdout/stderr,长命令(cargo/npm 构建)
|
||
/// 期间前端只看 Started→Completed,中间进度不可见。本函数 spawn 子进程后并发逐行读两条流,
|
||
/// 每读到一行回调 `on_output(kind, line)`(调用方可 emit 事件给前端实时展示),仍等进程 exit
|
||
/// 后返回完整 ShellResult(与 execute() 同形,调用方无需感知差异)。
|
||
///
|
||
/// 4性:
|
||
/// - 合理机制:spawn + BufReader::lines() 逐行,不丢未换行结尾的末段(read_to_end 兜底)
|
||
/// - 简洁:与 execute() 共用 build_command(单真相源,shell/kill_on_drop/cwd/env 不重复)
|
||
/// - 可靠兜底:timeout_secs 仍生效(超时 drop future → kill_on_drop 杀进程,返回 Err);
|
||
/// on_output 回调 Err 不影响主流程(调用方 emit 失败静默吞)
|
||
/// - 健壮边界:stdout/stderr 各独立任务并发读,互不阻塞;无管道死锁(piped + 同时消费)
|
||
pub async fn execute_streaming<F>(request: ShellRequest, mut on_output: F) -> anyhow::Result<ShellResult>
|
||
where
|
||
F: FnMut(StreamKind, &str) + Send,
|
||
{
|
||
let start = std::time::Instant::now();
|
||
|
||
#[cfg(windows)]
|
||
let _ = probe_pwsh().await;
|
||
|
||
// 先取走引用字段,再 move request 进 build_command
|
||
let command_for_err = request.command.clone();
|
||
let timeout_secs = request.timeout_secs;
|
||
let mut cmd = build_command(request);
|
||
let inner = async {
|
||
let mut child = cmd.spawn()?;
|
||
// 取出 piped 的 stdout/stderr handle(None → 视为已关,读为空,不影响主流程)
|
||
let stdout = child.stdout.take();
|
||
let stderr = child.stderr.take();
|
||
|
||
// mpsc 通道:读 task 把 (kind, line) 推过来,主 task 在 wait 期间 drain 并调 on_output。
|
||
// 用通道而非直接共享 on_output:FnMut 不可 clone,两读 task 无法各持一份;通道解耦读写,
|
||
// 回调集中在主 task 单点调用(顺序确定、无锁、回调内阻塞不影响读循环)。
|
||
let (tx, mut rx) = tokio::sync::mpsc::channel::<(StreamKind, String)>(64);
|
||
|
||
let mut tasks: Vec<tokio::task::JoinHandle<()>> = Vec::with_capacity(2);
|
||
if let Some(out) = stdout {
|
||
let tx = tx.clone();
|
||
tasks.push(tokio::spawn(async move {
|
||
let mut reader = BufReader::new(out).lines();
|
||
while let Ok(Some(line)) = reader.next_line().await {
|
||
if tx.send((StreamKind::Stdout, line)).await.is_err() {
|
||
break; // 接收端 drop(主 task 结束)→ 停止读
|
||
}
|
||
}
|
||
}));
|
||
}
|
||
if let Some(err) = stderr {
|
||
let tx = tx.clone();
|
||
tasks.push(tokio::spawn(async move {
|
||
let mut reader = BufReader::new(err).lines();
|
||
while let Ok(Some(line)) = reader.next_line().await {
|
||
if tx.send((StreamKind::Stderr, line)).await.is_err() {
|
||
break;
|
||
}
|
||
}
|
||
}));
|
||
}
|
||
// 主 task 不再 send → drop tx(读 task send 失败即退出)
|
||
drop(tx);
|
||
|
||
// 完整输出累积(主 task 单点写,无锁)。
|
||
let mut stdout_buf = String::new();
|
||
let mut stderr_buf = String::new();
|
||
|
||
// wait + drain 并行:边等进程退出边消费输出行(防管道写满阻塞致子进程 hang)。
|
||
let wait_fut = child.wait();
|
||
tokio::pin!(wait_fut);
|
||
let status: std::process::ExitStatus = loop {
|
||
tokio::select! {
|
||
// 进程退出 → 跳出循环,继续 drain 通道内残余行
|
||
status = &mut wait_fut => {
|
||
let status = status?;
|
||
// drain 剩余行(读 task 在管道 EOF 后 send 完最后批次即退出,rx 返 None 闭合)
|
||
while let Some((kind, line)) = rx.recv().await {
|
||
let clean = strip_ansi(&line);
|
||
match kind {
|
||
StreamKind::Stdout => { stdout_buf.push_str(&clean); stdout_buf.push('\n'); }
|
||
StreamKind::Stderr => { stderr_buf.push_str(&clean); stderr_buf.push('\n'); }
|
||
}
|
||
on_output(kind, &clean);
|
||
}
|
||
break status;
|
||
}
|
||
// 收到一行 → 累积 + 回调
|
||
Some((kind, line)) = rx.recv() => {
|
||
let clean = strip_ansi(&line);
|
||
match kind {
|
||
StreamKind::Stdout => { stdout_buf.push_str(&clean); stdout_buf.push('\n'); }
|
||
StreamKind::Stderr => { stderr_buf.push_str(&clean); stderr_buf.push('\n'); }
|
||
}
|
||
on_output(kind, &clean);
|
||
}
|
||
}
|
||
};
|
||
// 防御性 join 读 task(此时必已 EOF 退出,仅保险;失败静默不阻断)
|
||
for t in tasks {
|
||
let _ = t.await;
|
||
}
|
||
Ok::<ShellResult, anyhow::Error>(ShellResult {
|
||
stdout: stdout_buf,
|
||
stderr: stderr_buf,
|
||
exit_code: status.code(),
|
||
duration_ms: 0, // 外层统一填
|
||
})
|
||
};
|
||
|
||
let result = match timeout_secs {
|
||
Some(secs) => tokio::time::timeout(std::time::Duration::from_secs(secs), inner)
|
||
.await
|
||
.map_err(|_| anyhow::anyhow!("命令执行超时({}s): {}", secs, command_for_err))??,
|
||
None => inner.await?,
|
||
};
|
||
|
||
let duration = start.elapsed().as_millis() as u64;
|
||
Ok(ShellResult {
|
||
stdout: result.stdout,
|
||
stderr: result.stderr,
|
||
exit_code: result.exit_code,
|
||
duration_ms: duration,
|
||
})
|
||
}
|