优化: run_command 实时流式 + 审批浮窗修复 + 文档探索闭环(fetch_url+obscura引导+model_fetch兼容+prompt策略+厂商预设)
This commit is contained in:
+168
-12
@@ -3,6 +3,8 @@
|
||||
use serde::{Deserialize, Serialize};
|
||||
use std::process::Stdio;
|
||||
|
||||
use tokio::io::{AsyncBufReadExt, BufReader};
|
||||
|
||||
/// Shell 命令执行结果
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct ShellResult {
|
||||
@@ -100,17 +102,28 @@ pub struct ShellRequest {
|
||||
pub shell_type: Option<ShellType>,
|
||||
}
|
||||
|
||||
/// 执行 Shell 命令
|
||||
/// 输出流类型(回调 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。
|
||||
///
|
||||
/// 支持超时(timeout_secs)、环境变量(env)、工作目录(working_dir),
|
||||
/// kill_on_drop(true) 保证超时后子进程不残留,shell_type 可选 Cmd/PowerShell/Sh。
|
||||
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;
|
||||
|
||||
/// 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 => {
|
||||
@@ -161,14 +174,34 @@ pub async fn execute(request: ShellRequest) -> anyhow::Result<ShellResult> {
|
||||
for (key, value) in &request.env {
|
||||
cmd.env(key, value);
|
||||
}
|
||||
cmd
|
||||
}
|
||||
|
||||
let output = match request.timeout_secs {
|
||||
/// 执行 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, request.command))??,
|
||||
.map_err(|_| anyhow::anyhow!("命令执行超时({}s): {}", secs, command_for_err))??,
|
||||
None => cmd.output().await?,
|
||||
};
|
||||
|
||||
@@ -181,3 +214,126 @@ pub async fn execute(request: ShellRequest) -> anyhow::Result<ShellResult> {
|
||||
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 {
|
||||
match kind {
|
||||
StreamKind::Stdout => { stdout_buf.push_str(&line); stdout_buf.push('\n'); }
|
||||
StreamKind::Stderr => { stderr_buf.push_str(&line); stderr_buf.push('\n'); }
|
||||
}
|
||||
on_output(kind, &line);
|
||||
}
|
||||
break status;
|
||||
}
|
||||
// 收到一行 → 累积 + 回调
|
||||
Some((kind, line)) = rx.recv() => {
|
||||
match kind {
|
||||
StreamKind::Stdout => { stdout_buf.push_str(&line); stdout_buf.push('\n'); }
|
||||
StreamKind::Stderr => { stderr_buf.push_str(&line); stderr_buf.push('\n'); }
|
||||
}
|
||||
on_output(kind, &line);
|
||||
}
|
||||
}
|
||||
};
|
||||
// 防御性 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,
|
||||
})
|
||||
}
|
||||
|
||||
@@ -9,7 +9,7 @@
|
||||
//!
|
||||
//! 注:execute 逻辑本身未改动,此文件为零行为变更的纯新增测试。
|
||||
|
||||
use df_execute::shell::{execute, ShellRequest, ShellType};
|
||||
use df_execute::shell::{execute, execute_streaming, ShellRequest, ShellType, StreamKind};
|
||||
use std::collections::HashMap;
|
||||
|
||||
/// 平台默认 ShellType(对齐 shell.rs:31 Default impl:Windows→Cmd, 非 Windows→Sh)
|
||||
@@ -159,3 +159,81 @@ async fn execute_working_dir() {
|
||||
// 清理
|
||||
let _ = std::fs::remove_dir_all(&tmp_for_cleanup);
|
||||
}
|
||||
|
||||
// ============================================================
|
||||
// execute_streaming 流式测试
|
||||
// ============================================================
|
||||
|
||||
/// 流式:stdout 多行逐行回调,且 ShellResult 完整(行数对齐 + exit_code=0)。
|
||||
///
|
||||
/// 治 run_command 黑盒:验证 spawn 后逐行回调 vs 一次性返回的等价性(行内容 + 完整结果)。
|
||||
#[tokio::test]
|
||||
async fn streaming_stdout_lines_callback() {
|
||||
// 多行输出:Cmd 用多个 echo(用 & 串联无依赖),Sh 用 printf 多行
|
||||
let cmd = if cfg!(windows) {
|
||||
"@echo line1 & @echo line2 & @echo line3"
|
||||
} else {
|
||||
"printf 'line1\\nline2\\nline3\\n'"
|
||||
};
|
||||
let mut lines: Vec<(StreamKind, String)> = Vec::new();
|
||||
let res = execute_streaming(req(cmd), |kind, line| {
|
||||
lines.push((kind, line.to_string()));
|
||||
})
|
||||
.await
|
||||
.expect("execute_streaming 应返回 Ok");
|
||||
|
||||
assert_eq!(res.exit_code, Some(0), "成功命令 exit_code 应为 0");
|
||||
// stdout 应含三行(line1/line2/line3)
|
||||
assert!(res.stdout.contains("line1"), "stdout 应含 line1,实际: {:?}", res.stdout);
|
||||
assert!(res.stdout.contains("line3"), "stdout 应含 line3,实际: {:?}", res.stdout);
|
||||
// 回调收到的 stdout 行应含三行(过滤 stderr 干扰:Cmd 无 stderr,Sh 无 stderr)
|
||||
let stdout_lines: Vec<&String> = lines.iter()
|
||||
.filter(|(k, _)| *k == StreamKind::Stdout)
|
||||
.map(|(_, l)| l)
|
||||
.collect();
|
||||
assert!(
|
||||
stdout_lines.iter().any(|l| l.contains("line1")),
|
||||
"回调应收到含 line1 的 stdout 行,实际: {:?}", stdout_lines
|
||||
);
|
||||
assert!(
|
||||
stdout_lines.iter().any(|l| l.contains("line3")),
|
||||
"回调应收到含 line3 的 stdout 行,实际: {:?}", stdout_lines
|
||||
);
|
||||
}
|
||||
|
||||
/// 流式:超时仍生效(timeout_secs=1 + 长睡命令,返回 Err)。
|
||||
#[tokio::test]
|
||||
async fn streaming_timeout_returns_err() {
|
||||
let sleep_cmd = if cfg!(windows) {
|
||||
"ping -n 5 127.0.0.1 > nul".to_string()
|
||||
} else {
|
||||
"sleep 5".to_string()
|
||||
};
|
||||
let request = ShellRequest {
|
||||
command: sleep_cmd,
|
||||
working_dir: None,
|
||||
env: HashMap::new(),
|
||||
timeout_secs: Some(1),
|
||||
shell_type: Some(default_shell()),
|
||||
};
|
||||
let result = execute_streaming(request, |_, _| {}).await;
|
||||
assert!(result.is_err(), "超时应返回 Err,实际: {:?}", result.as_ref().err());
|
||||
let msg = result.unwrap_err().to_string();
|
||||
assert!(
|
||||
msg.contains("超时") || msg.to_lowercase().contains("timeout"),
|
||||
"错误信息应含超时提示,实际: {}",
|
||||
msg
|
||||
);
|
||||
}
|
||||
|
||||
/// 流式:非零退出仍返回 Ok + exit_code 非 0(对齐 execute 语义)。
|
||||
#[tokio::test]
|
||||
async fn streaming_nonzero_exit() {
|
||||
let mut callbacks = 0u32;
|
||||
let res = execute_streaming(req("exit 1"), |_, _| { callbacks += 1; })
|
||||
.await
|
||||
.expect("非零退出应仍返回 Ok");
|
||||
assert_ne!(res.exit_code, Some(0), "exit 1 的 exit_code 应非 0");
|
||||
// exit 1 无输出,回调可为 0 次(无行)——不强制断言次数,只确认无 panic
|
||||
let _ = callbacks;
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user