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

修复: 连接池收敛与传输错误分类重试

Browse Source 
- 连接池空闲十五秒,陈旧连接复用致重置根治
- 流式发送错误改谓词分类,openai 路径不再裸奔
- decode 类失败纳入重试,正文中断可自愈
- 状态分类合并响应体,激活休眠词表分支
- 手动压缩失败降级兜底,不再弹红卡
This commit is contained in:
lxy
2026-10-01 19:35:22 +08:00
parent 4417b3ddd3
commit 7dcbdb1feb
5 changed files with 343 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src-tauri/src/commands/ai/commands/chat.rs
+127 -11
View File
@@ -666,8 +666,51 @@ pub async fn ai_chat_send(
);
}
}
// 共用核心收口(先例 execute_run_workflow_for_tool):provider 解析 → conv 占用检查 →
// push user 消息 → system prompt 组装 → loop_epoch → spawn run_agentic_loop。
// IPC 返回值语义不变:Ok 映射回 "ok"(前端契约),conv_id 仅供编排节点侧消费。
spawn_agentic_session(
&app,
&state,
message,
language,
skill,
model_override,
conversation_id,
parts,
mention_spans,
true,
)
.await
.map(|_| "ok".to_string())
}
/// 发送消息并启动 agentic 会话的共用核心(纯搬迁自 ai_chat_send,行为逐字等价)。
///
/// 调用方:ai_chat_send IPC(前端对话,pin_active=true)/ 工作流 AiAgentNode 编排会话
/// (orchestration_session_spawner,pin_active=false)。两方共用同一 provider 解析/conv 占用/
/// system prompt 组装/loop_epoch/spawn 链路,不重复实现。
///
/// `pin_active` 唯一行为开关:true 时保持原「active_conversation_id 同步到 target」语义
/// (用户当前面板即 target);false(编排会话,显式传独立新 conv_id)跳过同步,不劫持用户
/// 活跃面板。其余路径与原实现逐字一致。
///
/// 返回 Ok(conv_id):编排节点用它追踪会话/等待完成通知;ai_chat_send 映射回 "ok"。
#[allow(clippy::too_many_arguments)]
pub(crate) async fn spawn_agentic_session(
app: &AppHandle,
state: &AppState,
message: String,
language: Option<String>,
skill: Option<String>,
model_override: Option<String>,
conversation_id: Option<String>,
parts: Option<Vec<ContentPart>>,
mention_spans: Option<Vec<MentionSpanDto>>,
pin_active: bool,
) -> Result<String, String> {
// 获取活跃提供商（只读，失败可直接返回，不影响生成标志）
let provider_config = super::super::prompt::get_active_provider(&state).await?;
let provider_config = super::super::prompt::get_active_provider(state).await?;
// anthropic 协议需 url→base64 预拉(base64 内嵌,URL 不可达;OpenAI 兼容原生 URL 直传零改动)。
// 拉取在 lock 之前做,不持锁网络 I/O;结果供 lock 内 user_parts 使用。
@@ -704,7 +747,8 @@ pub async fn ai_chat_send(
};
// active_conversation_id 同步到 target(用户当前面板即 target,路由键跟随)。
// 决策 e:旧 conv 的后台 loop 不受 active 变更影响(写自己的 per_conv)。
if session.active_conversation_id.as_deref() != Some(target.as_str()) {
// pin_active=false(工作流编排会话)跳过:编排 conv 不劫持用户活跃面板。
if pin_active && session.active_conversation_id.as_deref() != Some(target.as_str()) {
session.active_conversation_id = Some(target.clone());
// 切到已有 conv(非本次懒建)时,确保其 active_conv_created_at 已设(switch 路径已设)。
if session.active_conv_created_at.is_none() {
@@ -827,6 +871,8 @@ pub async fn ai_chat_send(
let app_handle = app.clone();
let knowledge_config = state.knowledge_config.lock().await.clone();
let llm_concurrency = state.llm_concurrency.clone();
// 返回值用 clone(conv_id 本体 move 进 spawn 闭包,搬运自原实现保持不动)
let conv_id_return = conv_id.clone();
// loop 入口 load 快照，当前 loop 锁定边界（热改下次发消息生效）
let max_iterations = state.agent_max_iterations.load(Ordering::SeqCst);
// 流式失败重试次数快照
@@ -842,7 +888,53 @@ pub async fn ai_chat_send(
run_agentic_loop(session_arc, tools_arc, db, app_handle, provider_config, system_prompt, volatile_tail, conv_id, knowledge_config, llm_concurrency, max_iterations, max_retries, 0, conv_model_override, loop_epoch).await;
});
Ok("ok".to_string())
// 返回 conv_id(共用核心:IPC 层映射回 "ok",编排节点消费 conv_id 等待完成通知)
Ok(conv_id_return)
}
/// 工作流 AiAgentNode 的编排会话启动通道(节点工厂经 build_registry 注入,见 state.rs)。
///
/// 流程:取进程级 AppHandle(spawn_agent 子代理同款 OnceLock,lib.rs setup 注入)→ 生成
/// 新独立 conv_id(不与用户活跃会话共用,防 epoch 互踩/active 面板劫持)→ 注册完成通知桥
/// (AppState.agent_completion_bridge)→ 复用 [`spawn_agentic_session`] 起独立 agentic 会话
/// → 返回 (conv_id, 完成通知接收端)。
///
/// 注册先于 spawn:完成通知最早发生在 loop 收敛出口(秒级之后),不存在注册前触发竞态;
/// 顺序对调则有「先收敛后注册」丢通知窗口。返回的接收端由节点 select! 等待;发送端挂在
/// 通知桥,agentic loop 收敛统一出口(emit_ai_completed_once)投递最终 assistant 报告。
pub(crate) fn orchestration_session_spawner() -> df_nodes::ai_agent_node::SpawnFn {
std::sync::Arc::new(|task: String| -> df_nodes::ai_agent_node::SpawnFuture {
Box::pin(async move {
let app = super::super::tools::spawn_agent::app_handle()
.ok_or_else(|| "应用句柄未就绪(启动初始化未完成)".to_string())?;
let conv_id = new_id();
let (tx, rx) = tokio::sync::oneshot::channel::<String>();
let state = app.state::<AppState>();
state.register_agent_completion(conv_id.clone(), tx);
match spawn_agentic_session(
&app,
&state,
task,
// 编排会话:无语言/技能/模型覆盖/多模态/@mention;pin_active=false 不动用户活跃面板
None,
None,
None,
Some(conv_id.clone()),
None,
None,
false,
)
.await
{
Ok(_) => Ok((conv_id, rx)),
// 启动失败:接收端随本 Err 返回被丢弃,通知桥条目同步清掉(防残留死条目)
Err(e) => {
state.notify_agent_completion(&conv_id, String::new());
Err(e)
}
}
})
})
}
/// 批准/拒绝挂起的工具调用
@@ -1608,12 +1700,14 @@ pub async fn ai_chat_clear_context(
/// 手动 LLM 压缩上下文(F-15 阶段2):摘要保护区外消息并插入首位。
///
/// 流程(失败不阻塞,消息状态不变):
/// 流程(失败非阻断降级):
/// 1. emit AiCompressing + set_compressing(true) 防重入
/// 2. 算 compress_end(保护区外,同 clear_context 的 protect_start)
/// 3. 先取保护区外 active 消息的克隆(读不改)→ 喂 compress_via_llm
/// 4. LLM 成功 → compress_old_messages 标 compressed(扣 token) + insert_at(0, system 摘要)
/// 5. LLM 失败 → set_compressing(false) + emit AiError(不含 api_key) + 返回 Err;消息状态不变
/// 5. LLM 失败 → 与自动压缩同款兜底:标 compressed 释放 token + 关键词摘要插首位 +
/// toast(AiManualCompressed),不发 AiError 红卡(前端 AiError 无条件复位 streaming,
/// 手动压缩失败也非生成期错误)
///
/// 空会话/无 active 可压缩消息 → set_compressing(false) 后 emit AiCompressing noop 返回 Ok。
#[tauri::command]
@@ -1707,6 +1801,9 @@ pub async fn ai_chat_compress_context(
}
};
let llm_concurrency = state.llm_concurrency.clone();
// LLM 失败兜底文本须在 active_msgs move 进 compress_via_llm 前算好
// (同 context_lifecycle 自动压缩先算兜底再喂 LLM 的次序)。
let keyword_fallback = df_ai::context_helpers::extract_keyword_summary(&active_msgs);
let summary = match super::super::compress::compress_via_llm(
provider.as_ref(),
&provider_config,
@@ -1717,19 +1814,38 @@ pub async fn ai_chat_compress_context(
).await {
Ok(s) => s.nl_summary,
Err(e) => {
// LLM 失败:不阻塞,set_compressing(false),消息状态不变(未调 compress_old_messages)
// LLM 失败非阻断降级(与自动压缩 context_lifecycle 同款兜底):标 compressed 释放
// token + 关键词摘要插首位作续接锚点,不再 emit AiError 红卡(前端 AiError 无条件
// 复位 streaming 误判生成结束,且手动压缩失败非生成期错误)。toast 走
// AiManualCompressed 告知兜底结果。
tracing::warn!(
conv_id = %conv_id,
error = %e,
fallback_len = keyword_fallback.len(),
"[ai] 手动压缩 LLM 失败,降级关键词摘要兜底(非阻断)"
);
let __lock_t1295 = std::time::Instant::now();
let mut session = state.ai_session.lock().await;
session.conv(&conv_id).messages.set_compressing(false);
let conv = session.conv(&conv_id);
let _compressed = conv.messages.compress_old_messages(compress_end);
if !keyword_fallback.is_empty() {
conv.messages.insert_at(0, ChatMessage::system(&keyword_fallback));
}
conv.messages.set_compressing(false);
log_slow_lock(__lock_t1295, "ai_chat_compress_context:1295");
drop(session);
let ev = AiChatEvent::AiError {
error: format!("压缩失败: {}", e),
error_type: Some(super::super::ErrorType::Unknown),
save_conversation(&state.ai_session, &state.db, &conv_id, None, None, false).await;
let toast_summary = if keyword_fallback.is_empty() {
"压缩失败,历史消息已直接释放(可重试)".to_string()
} else {
keyword_fallback.clone()
};
let ev = AiChatEvent::AiManualCompressed {
conversation_id: Some(conv_id),
summary: toast_summary,
};
emit_ai_chat_event(&app, ev);
return Err(format!("压缩失败: {}", e));
return Ok(());
}
};
src-tauri/src/commands/ai/stream_recv.rs
+56 -2
View File
@@ -582,7 +582,8 @@ pub(crate) async fn stream_llm(
/// - HTTP 429 / 5xx → retryable=true(复用 retry::is_status_retryable 逻辑)
/// - HTTP 4xx(非429) → retryable=false Fatal(401/403/404/422 等)
/// - timeout / connect → retryable=true(瞬态,可重试)
/// - unknown / 其他 → retryable=true(保守可重试,避免误判 Fatal 错杀)
/// - unknown → 交 classify_status_with_body 合并 body 按传输关键词(reset/decode/eof 等)判定,
/// 无命中则 Fatal(对齐 retry::is_status_retryable 未知不可重试语义)
///
/// 抽公共分类函数(决策 F-260616-07 a1 "复用 retry.rs 错误分类逻辑或镜像"),
/// 避免 stream_recv 与 agentic 各自重写状态码解析。
@@ -614,7 +615,17 @@ fn classify_status_or_class(status_or_class: &str) -> bool {
// BUG-2026-07-17 根治:SSE 解码/传输错误(error decoding response body / Transport error)
// 是传输层错误,不是 Fatal。原实现未识别这些关键词,返 unknown→Fatal,致连续重试 4 次均
// 立即放弃。现补全传输层关键词,让这类错误走可重试路径。
let lower = status_or_class.to_lowercase();
transport_keyword_hit(&status_or_class.to_lowercase())
}
/// 传输层可重试关键词判定(小写文本)。
///
/// 词表与 df-ai crate::send_error_class 的类词契约对应(改任一侧需同步)。
/// 抽独立纯函数供 classify_status_or_class 与 classify_status_with_body 的 unknown 分支共用:
/// body 原文常含任意数字串(os error 10054/时间戳),过上方三位码扫描器会被误当 HTTP 码
/// (如 "100" → is_status_retryable(100)=false)——这正是历史 reset/decode 关键词分支
/// 永远够不着的第二层死因,unknown 类需绕过扫描器直接走词表。
fn transport_keyword_hit(lower: &str) -> bool {
lower.contains("timeout")
|| lower.contains("connect")
|| lower.contains("transport")
@@ -647,10 +658,19 @@ fn is_quota_exhaustion(combined: &str) -> bool {
/// 额度耗尽型 429 → Fatal(确定性,不重试);否则交 classify_status_or_class 正常判定。
/// 入参 body:错误体原文(provider bail 文本 / error 帧 message)。
///
/// 传输层关键词(reset/decode/eof/broken pipe 等)只出现在 body 原文,而 status_or_class
/// 经 extract_error_diag_from_str 只会是 HTTP 码/timeout/connect/unknown——只喂 status 时
/// 关键词分支永远够不着(dead code,历史修复未生效)。unknown(无码无类)时合并 body 直接走
/// 传输词表(绕过三位码扫描器,body 的 os error 码/时间戳会被误当 HTTP 码);HTTP 码/
/// timeout/connect 显式类保持只喂 status,body 不反转既有判定(401 带 reset 字样仍 Fatal)。
fn classify_status_with_body(status_or_class: &str, body: &str) -> bool {
if is_quota_exhaustion(&format!("{status_or_class} {body}")) {
return false;
}
if status_or_class == "unknown" {
return transport_keyword_hit(&format!("{status_or_class} {body}").to_lowercase());
}
classify_status_or_class(status_or_class)
}
@@ -834,6 +854,40 @@ mod tests {
assert!(classify_status_with_body("HTTP 500", "Internal Server Error"));
}
// ---- classify_status_with_body：unknown 类合并 body 喂入,激活传输关键词分支 ----
/// unknown 类 + body 含传输关键词 → 可重试(原只喂 status 恒 Fatal 的 dead code 根治:
/// reset/decode/eof/broken pipe 只出现在 body 原文,status_or_class 永远不带)。
#[test]
fn classify_unknown_with_transport_body_retryable() {
assert!(classify_status_with_body("unknown", "Connection reset by peer (os error 10054)"));
assert!(classify_status_with_body("unknown", "error decoding response body"));
assert!(classify_status_with_body("unknown", "stream ended unexpectedly: unexpected eof"));
assert!(classify_status_with_body("unknown", "failed to send request: broken pipe"));
}
/// unknown 类 + body 无传输关键词 → 仍 Fatal(不回退到无脑可重试空耗)。
#[test]
fn classify_unknown_without_transport_keywords_fatal() {
assert!(!classify_status_with_body("unknown", "服务开小差了"));
assert!(!classify_status_with_body("unknown", ""));
}
/// 显式 HTTP 4xx 即便 body 带传输关键词也保持 Fatal(body 不反转状态码判定)。
#[test]
fn classify_4xx_transport_body_still_fatal() {
assert!(!classify_status_with_body("HTTP 401", "Connection reset by peer"));
assert!(!classify_status_with_body("HTTP 403", "error decoding response body"));
assert!(!classify_status_with_body("HTTP 404", "broken pipe"));
}
/// timeout/connect 显式类不受 body 影响(timeout 串碰含 404 字样的 body 仍可重试)。
#[test]
fn classify_explicit_class_unaffected_by_body() {
assert!(classify_status_with_body("timeout", "HTTP 404 not found"));
assert!(classify_status_with_body("connect", "connection refused"));
}
/// 普通瞬时限流 429 的标准 type=rate_limit_error 不应误判额度耗尽(仍可重试,防误伤)
#[test]
fn standard_rate_limit_error_not_quota() {