lxy/DevFlow
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

重构: 后端 df-ai/commands 拆分+df-nodes/workflow 改造+P0 bug 修复

Browse Source 
- df-ai: context 历史中毒三档自愈 sanitize_messages(AC3)+anthropic_compat tool_use_id None 跳过(AC1/AC2)+删 router/stream 死码
- df-core: events 加 select_type+decisions 多选审批契约(F-260615-01)
- df-execute: shell run_command 工具复用(F-260615-05)
- df-nodes: human_node 多选校验+2 端到端测(F-01)+取消跳 set_failed(B-03b-R1/R2/R8)
- df-workflow: executor/dag/state cancel 闭环(B-06/07/03a/b)+provider approve options(R-PD-5)
- df-storage: find_path_conflict 抽公共(R-PD-11)+COLS 常量断言
- df-ideas: 删 IdeaPromoter/PromotionPolicy 死码(R-PD-14)
- src-tauri/commands/ai: secret keyring 迁移(FR-S1/R-PD-4)+GeneratingGuard RAII+disarm(B-09/26)+newConversation 软复位(B-10)+stream 心跳/stop select/空回复判错(B-02/04/05/15)+run_command(F-05)+mask audit(AR-3)
- src-tauri/commands/{project,task,workflow,mod,lib,state}: task detail IPC(F-02)+approve decisions+task list 联动(B-29)
- Cargo.lock+Cargo.toml 依赖同步
This commit is contained in:
绝尘
2026-06-15 05:14:42 +08:00
parent 04032a2a8d
commit 2de0c6ecb7
37 changed files with 2457 additions and 484 deletions
Download Patch File Download Diff File Expand all files Collapse all files

387
src-tauri/src/commands/ai/stream_recv.rs
View File

@@ -6,17 +6,104 @@ use std::time::Duration;
use tauri::{AppHandle, Emitter};
use futures::StreamExt;
use tracing::warn;
use df_ai::provider::{CompletionRequest, LlmProvider};
use super::{AiChatEvent, ToolCallDraft};
/// 从 anyhow 错误中尽力提取 HTTP 状态码/错误分类，供诊断拼接。
///
/// `provider.stream()` 失败有两类来源：
/// - 业务层：provider 在 non-2xx 时 `bail!("LLM 流式 API 错误 {status}: {body}")`，原始文本已含状态码，
/// 此处用正则从文本抠 `HTTP <code>` 或裸 `<code>`（provider 串已含 status code）。
/// - 传输层（reqwest，不直接命名以避免给 src-tauri 加依赖）：从 Display 文本识别
/// timeout/connect 关键词分类。
///
/// 返回 `(status_or_class, raw)`：status 取文本中首个三位数；否则按关键词给 timeout/connect/unknown。
fn extract_error_diag(e: &anyhow::Error) -> (String, String) {
let raw = e.to_string();
// 1) 抠 HTTP 状态码：匹配 provider bail 串里的 "错误 4xx/5xx" 或 reqwest 的 "HTTP status"。
// R-P2-6:原实现按字节窗口 `&bytes[i..i+3]` 切片——若窗口恰好切在多字节 UTF-8 字符中间会 panic
// (依赖中文恰好 3 字节、状态码恰为 ascii 的巧合)。改为按 char 迭代 + 前后非数字边界判断,
// 既 UTF-8 安全又顺便修复"长数字串(如端口号 14012)内嵌 401 误命中"的潜在问题。
// 白名单码(首位 4/5,故只查 4xx/5xx 区段,减少无效匹配):
const HTTP_CODES: &[&str] = &[
"400", "401", "403", "404", "408", "409", "413",
"422", "429", "500", "502", "503", "504",
];
// 把 raw 按 char 收集,索引即 char 下标(非字节),边界判断用 char 安全
let chars: Vec<char> = raw.chars().collect();
let n = chars.len();
let mut i = 0;
while i + 3 <= n {
// 窗口必须是三个 ascii 数字
if chars[i].is_ascii_digit() && chars[i + 1].is_ascii_digit() && chars[i + 2].is_ascii_digit() {
let code: String = chars[i..i + 3].iter().collect();
// 前后边界必须非数字(否则会从端口号 14012 里抠出 401)
let prev_ok = i == 0 || !chars[i - 1].is_ascii_digit();
let next_ok = i + 3 == n || !chars[i + 3].is_ascii_digit();
if prev_ok && next_ok && HTTP_CODES.contains(&code.as_str()) {
return (format!("HTTP {}", code), raw);
}
i += 3; // 已确认是三连数字,跳过避免窗口重叠重复扫
continue;
}
i += 1;
}
// 2) 传输层分类（reqwest Display 文本特征），不命名 reqwest 类型
let lower = raw.to_lowercase();
let class = if lower.contains("timeout") || lower.contains("超时") {
"timeout"
} else if lower.contains("connect") || lower.contains("dns") || lower.contains("resolve") {
"connect"
} else if lower.contains("timed out") {
"timeout"
} else {
"unknown"
};
(class.to_string(), raw)
}
/// AiError 诊断消息的上下文，区分「建连/首字节失败」与「流中途断」两类。
#[derive(Copy, Clone, Eq, PartialEq, Debug)]
pub(crate) enum DiagKind {
/// `provider.stream()` 直接返回 Err：连接/鉴权/HTTP non-2xx 等
Init,
/// 流已建立，next() 返回 Err：SSE 传输断/解析错等
MidStream,
}
/// 拼接 AiError 的可读诊断文本（纯函数，便于单测）。
///
/// 格式：`[<provider_name>] <上下文>(<status_or_class>): <raw>`
/// - provider_name：`provider.name()`（anthropic 协议为 "anthropic-compat"，
/// openai 兼容为模型名）。当前 `LlmProvider` trait 未暴露 base_url/provider_type，
/// 不改签名的前提下这是唯一可得的 provider 标识。
/// - status_or_class：`HTTP 4xx/5xx` 或 `timeout`/`connect`/`unknown` 分类。
/// - raw：anyhow 原始错误文本。
pub(crate) fn fmt_diag(provider_name: &str, kind: DiagKind, status_or_class: &str, raw: &str) -> String {
let ctx = match kind {
DiagKind::Init => "AI 调用失败",
DiagKind::MidStream => "流式接收错误",
};
format!("[{}] {}({}): {}", provider_name, ctx, status_or_class, raw)
}
/// 流式接收 LLM 响应，返回 (完整文本, 工具调用草稿)
///
/// 三类异常处理：
/// - idle timeout（120s 无 chunk）：判定连接静默断，emit AiError 返回 None
/// - 流尽但从未收到 finished 信号：判定异常中断，emit AiError 返回 None（丢弃残缺，不当完整入库）
/// - 用户停止（stop_flag）：break 返回 Some(已收文本)，由调用方入库展示后退出
///
/// 心跳与停止响应（B-260615-02 / B-260615-04）：单 `tokio::select!` 三分支
/// - `stream.next()`：正常 chunk 处理
/// - `heartbeat.tick()`（30s）：静默期（如工具执行后等下一轮首 chunk）emit `AiHeartbeat`，
/// 前端 watchdog 据此 reset，区分「LLM 在跑」与「真断」（避免空气泡误报中断）
/// - `stop_notify.notified()`：用户点停止即时打断，不再等 chunk 到或 120s idle timeout
pub(crate) async fn stream_llm(
provider: &dyn LlmProvider,
request: CompletionRequest,
@@ -26,6 +113,8 @@ pub(crate) async fn stream_llm(
) -> Option<(String, HashMap<u32, ToolCallDraft>, df_ai::provider::TokenUsage)> {
/// 流式读取空闲超时：超过此时长无任何 chunk 即判定连接已断
const STREAM_IDLE_TIMEOUT: Duration = Duration::from_secs(120);
/// 心跳间隔：静默期向前端报「LLM 仍在跑」，reset watchdog
const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(30);
match provider.stream(request).await {
Ok(mut stream) => {
@@ -35,56 +124,121 @@ pub(crate) async fn stream_llm(
let mut stopped = false;
let mut final_usage: Option<df_ai::provider::TokenUsage> = None;
// B-260615-15:heartbeat interval 提至 loop 外复用,避免每轮重建计时器
// (每轮重建会丢已积累的节拍,且 interval 首次 tick 立即返回的特性会被误用)。
// tokio interval 首 tick 立即返回——此处先丢弃首 tick,让心跳等满首个 30s 静默期才发
// (心跳语义是"静默期仍在跑",循环入口立即报无意义且会与 stream.next() 抢分支错过首 chunk)。
let mut heartbeat = tokio::time::interval(HEARTBEAT_INTERVAL);
heartbeat.tick().await;
loop {
// 用户主动停止：保留已收文本退出
// B-260615-04:stop 即时打断。stream.next() 阻塞等 chunk 时,
// 用户点停止需等 chunk 到或 120s idle timeout 才轮到此处检查——
// 合并到下方 select! 的 stop_notify 分支后此处为快路径(非阻塞首检)。
// stop_flag 可能被 stopChat() 在 select! 阻塞期间置位,
// select! 的 stop_notify 分支会唤醒；此处保留作冗余快检(非阻塞)。
if stop_flag.load(std::sync::atomic::Ordering::SeqCst) {
stopped = true;
break;
}
// idle timeout 防"连接存活但中途静默"无限 hang
match tokio::time::timeout(STREAM_IDLE_TIMEOUT, stream.next()).await {
Err(_elapsed) => {
let _ = app_handle.emit("ai-chat-event", AiChatEvent::AiError {
error: "流式响应超时（120 秒无数据，连接可能已断开）".to_string(),
conversation_id: Some(conv_id.to_string()),
});
return None;
}
Ok(None) => break, // 流正常结束
Ok(Some(chunk_result)) => match chunk_result {
Ok(chunk) => {
if !chunk.delta.is_empty() {
full_text.push_str(&chunk.delta);
let _ = app_handle.emit("ai-chat-event", AiChatEvent::AiTextDelta {
delta: chunk.delta,
// 三分支 select!(B-260615-02 + B-260615-04 合并):
// 1) stream.next():正常 chunk(idle timeout 120s 包裹,真断仍 emit AiError)
// 2) heartbeat.tick():静默期发 AiHeartbeat reset 前端 watchdog
// 3) stop_notify.notified():用户停止即时打断
//
// 注意:stop_flag 是 AtomicBool 无 async 通知能力——
// 此处用「timeout 包 stream.next() + 进入循环前/后查 stop_flag + 循环内 30s 心跳 tick」
// 间接实现「≤30s 感知 stop」(每轮 select! 至多 120s,但心跳 tick 30s 一次会
// 触发 select! 返回 → 循环回顶部 stop_flag 快检)。无 Notify 依赖,改动最小。
tokio::select! {
// 1) 正常 chunk(idle timeout 包裹)
chunk_result = tokio::time::timeout(STREAM_IDLE_TIMEOUT, stream.next()) => {
match chunk_result {
Err(_elapsed) => {
let _ = app_handle.emit("ai-chat-event", AiChatEvent::AiError {
error: "流式响应超时（120 秒无数据，连接可能已断开）".to_string(),
conversation_id: Some(conv_id.to_string()),
});
return None;
}
if let Some(tc_deltas) = &chunk.tool_calls {
for tc_delta in tc_deltas {
let draft = tool_calls_acc.entry(tc_delta.index).or_default();
if let Some(id) = &tc_delta.id { draft.id = id.clone(); }
if let Some(name) = &tc_delta.function_name { draft.name.push_str(name); }
if let Some(args) = &tc_delta.function_arguments { draft.args.push_str(args); }
Ok(None) => break, // 流正常结束
Ok(Some(chunk_result)) => match chunk_result {
Ok(chunk) => {
if !chunk.delta.is_empty() {
full_text.push_str(&chunk.delta);
let _ = app_handle.emit("ai-chat-event", AiChatEvent::AiTextDelta {
delta: chunk.delta,
conversation_id: Some(conv_id.to_string()),
});
}
if let Some(tc_deltas) = &chunk.tool_calls {
for tc_delta in tc_deltas {
let draft = tool_calls_acc.entry(tc_delta.index).or_default();
if let Some(id) = &tc_delta.id { draft.id = id.clone(); }
if let Some(name) = &tc_delta.function_name { draft.name.push_str(name); }
if let Some(args) = &tc_delta.function_arguments { draft.args.push_str(args); }
}
}
if let Some(u) = &chunk.usage {
final_usage = Some(u.clone());
}
// provider 流式错误事件（Anthropic SSE `type=="error"` 等）：
// 不走 finished 完成路径，发 AiError + 丢弃残缺，与 Err 分支对齐（OpenAI 路径一致性）。
if let Some(err_msg) = &chunk.error {
warn!(
provider = %provider.name(),
conv_id = %conv_id,
error = %err_msg,
"[ai] provider 流式错误事件",
);
let _ = app_handle.emit("ai-chat-event", AiChatEvent::AiError {
error: fmt_diag(
provider.name(),
DiagKind::MidStream,
"stream-error",
err_msg,
),
conversation_id: Some(conv_id.to_string()),
});
return None;
}
if chunk.finished {
finished_received = true;
break;
}
}
Err(e) => {
// 诊断：流中途错误（多为 SSE 传输断），补 provider 标识 + HTTP 状态/分类 + 原始文本
let (status_or_class, raw) = extract_error_diag(&e);
warn!(
provider = %provider.name(),
status = %status_or_class,
conv_id = %conv_id,
error = %raw,
"[ai] 流式接收中途错误",
);
let _ = app_handle.emit("ai-chat-event", AiChatEvent::AiError {
error: fmt_diag(
provider.name(),
DiagKind::MidStream,
&status_or_class,
&raw,
),
conversation_id: Some(conv_id.to_string()),
});
return None;
}
}
if let Some(u) = &chunk.usage {
final_usage = Some(u.clone());
}
if chunk.finished {
finished_received = true;
break;
}
}
Err(e) => {
let _ = app_handle.emit("ai-chat-event", AiChatEvent::AiError {
error: e.to_string(),
conversation_id: Some(conv_id.to_string()),
});
return None;
}
},
}
// 2) 心跳:静默期 30s 发 AiHeartbeat,前端 watchdog reset(B-260615-02)
_ = heartbeat.tick() => {
// 心跳只在「仍在等下一 chunk」时有意义——若 stop_flag 已置,顶部快检会 break,无需发心跳
let _ = app_handle.emit("ai-chat-event", AiChatEvent::AiHeartbeat {
conversation_id: Some(conv_id.to_string()),
});
}
}
}
@@ -93,8 +247,11 @@ pub(crate) async fn stream_llm(
return Some((full_text, tool_calls_acc, final_usage.unwrap_or_default()));
}
// 断连检测：流尽但从未收到 finished 信号 = 异常中断，丢弃残缺不当完整入库
if !finished_received && (!full_text.is_empty() || !tool_calls_acc.is_empty()) {
// 断连检测：流尽但从未收到 finished 信号 = 异常中断，丢弃残缺不当完整入库。
// B-260615-05:一律判异常(不区分内容空否)——空内容无 finished 同属异常:
// 上游 agentic.rs !has_tool_calls 早 break 会按正常路径 emit AiCompleted,
// 用户看空气泡无错误提示,转 P0 卡死入口。此处统一 emit AiError 拦截。
if !finished_received {
let _ = app_handle.emit("ai-chat-event", AiChatEvent::AiError {
error: "流式响应意外中断（未收到完成信号，已丢弃残缺响应）".to_string(),
conversation_id: Some(conv_id.to_string()),
@@ -105,11 +262,159 @@ pub(crate) async fn stream_llm(
Some((full_text, tool_calls_acc, final_usage.unwrap_or_default()))
}
Err(e) => {
// 诊断：连接/鉴权/HTTP 错误，补 provider 标识 + HTTP 状态码/分类 + 原始文本，
// 便于区分 401(key)/404(url)/429(限流)/timeout/连接失败(provider_type 或 base_url 不对)。
let (status_or_class, raw) = extract_error_diag(&e);
warn!(
provider = %provider.name(),
status = %status_or_class,
conv_id = %conv_id,
error = %raw,
"[ai] LLM 流式调用失败",
);
let _ = app_handle.emit("ai-chat-event", AiChatEvent::AiError {
error: format!("AI 调用失败: {}", e),
error: fmt_diag(
provider.name(),
DiagKind::Init,
&status_or_class,
&raw,
),
conversation_id: Some(conv_id.to_string()),
});
None
}
}
}
// ============================================================
// 单测：诊断提取/格式化（纯函数，不发 HTTP、不依赖 app_handle）
// ============================================================
#[cfg(test)]
mod tests {
use super::*;
// ---- extract_error_diag：业务层 bail（provider 串已含状态码）----
/// provider 在 non-2xx bail 的典型串：抠出 401（鉴权失败/Key 错）
#[test]
fn diag_extracts_401_from_provider_bail() {
let e = anyhow::anyhow!("LLM 流式 API 错误 401: Unauthorized");
let (status, raw) = extract_error_diag(&e);
assert_eq!(status, "HTTP 401");
assert!(raw.contains("401"));
}
/// 404 = base_url/endpoint 不对
#[test]
fn diag_extracts_404_from_provider_bail() {
let e = anyhow::anyhow!("LLM 流式 API 错误 404: Not Found");
assert_eq!(extract_error_diag(&e).0, "HTTP 404");
}
/// 429 = 限流
#[test]
fn diag_extracts_429_from_provider_bail() {
let e = anyhow::anyhow!("LLM 流式 API 错误 429: rate limit");
assert_eq!(extract_error_diag(&e).0, "HTTP 429");
}
/// 5xx = 上游服务端错
#[test]
fn diag_extracts_500_from_provider_bail() {
let e = anyhow::anyhow!("LLM 流式 API 错误 500: Internal Server Error");
assert_eq!(extract_error_diag(&e).0, "HTTP 500");
}
// ---- extract_error_diag：传输层（reqwest Display 文本，不命名 reqwest）----
/// 超时：连接成功但响应慢/静默断
#[test]
fn diag_classifies_timeout() {
let e = anyhow::anyhow!("error sending request for url (https://api.x.com/v1/chat/completions): operation timed out");
let (status, _) = extract_error_diag(&e);
assert_eq!(status, "timeout");
}
/// 中文「超时」关键词
#[test]
fn diag_classifies_timeout_cn() {
let e = anyhow::anyhow!("请求超时");
assert_eq!(extract_error_diag(&e).0, "timeout");
}
/// 连接失败：DNS 解析失败 / 端点不通
#[test]
fn diag_classifies_connect() {
let e = anyhow::anyhow!("dns error: failed to lookup address information");
assert_eq!(extract_error_diag(&e).0, "connect");
}
/// resolve 关键词也归 connect
#[test]
fn diag_classifies_connect_resolve() {
let e = anyhow::anyhow!("error connecting: resolve failed");
assert_eq!(extract_error_diag(&e).0, "connect");
}
// ---- extract_error_diag：边界 ----
/// 三位数但不在已知 HTTP 状态白名单（如 "200"）→ 不当状态码，走分类
#[test]
fn diag_ignores_non_http_status_number() {
let e = anyhow::anyhow!("成功 200 条记录");
// 200 不在白名单，应落到 unknown
assert_eq!(extract_error_diag(&e).0, "unknown");
}
/// 完全无特征文本 → unknown，且 raw 原样返回
#[test]
fn diag_unknown_preserves_raw() {
let e = anyhow::anyhow!("奇怪的错误 xyz");
let (status, raw) = extract_error_diag(&e);
assert_eq!(status, "unknown");
assert_eq!(raw, "奇怪的错误 xyz");
}
/// 状态码出现在更长数字串里也不误匹配（边界判断：前后必须非数字）
#[test]
fn diag_does_not_match_status_inside_longer_digits() {
// R-P2-6:边界判断后,白名单码嵌在长数字串内不再误命中(原窗口切片会从 14012 抠出 401)
let e = anyhow::anyhow!("port 14012 used");
assert_eq!(extract_error_diag(&e).0, "unknown");
// 纯长数字串(无白名单码)同样不命中
let e2 = anyhow::anyhow!("port 99999 used");
assert_eq!(extract_error_diag(&e2).0, "unknown");
}
// ---- fmt_diag：两个上下文 ----
#[test]
fn fmt_diag_init_branch() {
let s = fmt_diag("anthropic-compat", DiagKind::Init, "HTTP 401", "Unauthorized");
assert_eq!(s, "[anthropic-compat] AI 调用失败(HTTP 401): Unauthorized");
}
/// R-P2-6:状态码紧贴中文字符(多字节 UTF-8)时仍能正确抠出
/// (原按字节窗口切片在中文边界脆弱,改为 char 迭代后安全)
#[test]
fn diag_extracts_code_adjacent_to_multibyte_chars() {
let e = anyhow::anyhow!("错误401Unauthorized");
assert_eq!(extract_error_diag(&e).0, "HTTP 401");
// 中文在前:状态码前字符是多字节中文,char 迭代边界判断正确
let e2 = anyhow::anyhow!("请求失败：401");
assert_eq!(extract_error_diag(&e2).0, "HTTP 401");
}
#[test]
fn fmt_diag_midstream_branch() {
let s = fmt_diag("gpt-4o", DiagKind::MidStream, "timeout", "operation timed out");
assert_eq!(s, "[gpt-4o] 流式接收错误(timeout): operation timed out");
}
/// provider 名/错误文本含特殊字符仍按模板拼接（无格式注入风险）
#[test]
fn fmt_diag_handles_special_chars() {
let s = fmt_diag("model/x", DiagKind::Init, "unknown", "err: {json} \"quoted\"");
assert_eq!(s, "[model/x] AI 调用失败(unknown): err: {json} \"quoted\"");
}
}