From 7dcbdb1febe67961dc86297140f295872dc28803 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=BB=9D=E5=B0=98?= <237809796@qq.com> Date: Thu, 1 Oct 2026 19:35:22 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D:=20=E8=BF=9E=E6=8E=A5?= =?UTF-8?q?=E6=B1=A0=E6=94=B6=E6=95=9B=E4=B8=8E=E4=BC=A0=E8=BE=93=E9=94=99?= =?UTF-8?q?=E8=AF=AF=E5=88=86=E7=B1=BB=E9=87=8D=E8=AF=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 连接池空闲十五秒,陈旧连接复用致重置根治 - 流式发送错误改谓词分类,openai 路径不再裸奔 - decode 类失败纳入重试,正文中断可自愈 - 状态分类合并响应体,激活休眠词表分支 - 手动压缩失败降级兜底,不再弹红卡 --- crates/df-ai/src/anthropic_compat.rs | 21 +++- crates/df-ai/src/lib.rs | 57 +++++++++ crates/df-ai/src/openai_compat.rs | 90 +++++++++++++- src-tauri/src/commands/ai/commands/chat.rs | 138 +++++++++++++++++++-- src-tauri/src/commands/ai/stream_recv.rs | 58 ++++++++- 5 files changed, 343 insertions(+), 21 deletions(-) diff --git a/crates/df-ai/src/anthropic_compat.rs b/crates/df-ai/src/anthropic_compat.rs index 425f010..589cadc 100644 --- a/crates/df-ai/src/anthropic_compat.rs +++ b/crates/df-ai/src/anthropic_compat.rs @@ -600,7 +600,9 @@ impl LlmProvider for AnthropicCompatProvider { source = ?std::error::Error::source(&e), "Anthropic 同步 send 失败" ); - if is_reqwest_error_retryable(&e) { + // is_decode 补可重试:body 中途断连在 reqwest 归 decode 类,complete 重发 + // 幂等安全(受 3 次/30s 预算约束,真正格式错重试同败,预算内止损)。 + if is_reqwest_error_retryable(&e) || e.is_decode() { return AttemptOutcome::Retryable(format!("请求失败(可重试): {}", e)); } return AttemptOutcome::Fatal(format!( @@ -626,7 +628,13 @@ impl LlmProvider for AnthropicCompatProvider { } let resp: AnthropicResponse = match resp.json().await { Ok(r) => r, - Err(e) => return AttemptOutcome::Fatal(format!("响应解析失败: {}", e)), + // serde 结构错属 Fatal(重试同败);body 读取中途断连同归 is_decode,可重试。 + Err(e) => { + if e.is_decode() { + return AttemptOutcome::Retryable(format!("响应解码失败(可重试): {}", e)); + } + return AttemptOutcome::Fatal(format!("响应解析失败: {}", e)); + } }; // content 块中拼接 text，收集 tool_use let mut text = String::new(); @@ -714,17 +722,24 @@ impl LlmProvider for AnthropicCompatProvider { let resp = match tokio::time::timeout(Duration::from_secs(60), send_future).await { Ok(Ok(r)) => r, Ok(Err(e)) => { + // 谓词结构化分类(对齐 openai 路径,共用 crate::send_error_class)。原 bail 串 + // [timeout={} connect={}=...] 的字面词让一切错误(含 builder 错)侥幸可重试。 + // 类词改由谓词决定置串首;谓词转储缩写为 t/c/b/r 防误触分类词表,source 链 + // 保留真实传输词(reset/broken pipe 等)作串兜底。 + let class = crate::send_error_class(crate::SendErrPreds::of(&e)); tracing::error!( is_timeout = e.is_timeout(), is_connect = e.is_connect(), is_body = e.is_body(), is_request = e.is_request(), + class = class, url = %self.messages_url(), source = ?std::error::Error::source(&e), "Anthropic 流式 send 失败" ); anyhow::bail!( - "send 失败: {} [timeout={} connect={} body={} request={} src={:?}]", + "send 失败({}): {} [t={} c={} b={} r={} src={:?}]", + class, e, e.is_timeout(), e.is_connect(), diff --git a/crates/df-ai/src/lib.rs b/crates/df-ai/src/lib.rs index 67d52d5..5a93a95 100644 --- a/crates/df-ai/src/lib.rs +++ b/crates/df-ai/src/lib.rs @@ -64,6 +64,11 @@ pub use df_ai_core; pub(crate) fn build_provider_client() -> Client { Client::builder() .connect_timeout(std::time::Duration::from_secs(30)) + // 连接池收敛:idle 连接 15s 弃用 + 每主机最多存 2 条。reqwest 默认 idle 90s 且 + // 无上限,服务端/前置 keepalive 更短时复用陈旧连接首包即被 RST(ConnectionReset + // 10054),呈现「隔几十秒一发就失败」的间歇形态;15s 保守低于常见网关 60s keepalive。 + .pool_idle_timeout(std::time::Duration::from_secs(15)) + .pool_max_idle_per_host(2) .build() .unwrap_or_else(|e| { tracing::warn!("reqwest builder 失败，回退默认 client: {}", e); @@ -71,6 +76,58 @@ pub(crate) fn build_provider_client() -> Client { }) } +/// 流式 send 阶段 reqwest 错误的谓词快照。 +/// +/// reqwest::Error 无公开构造器,抽谓词结构体过纯函数,单测才能锁定分类 +/// (send 阶段 ConnectionReset → 可重试类词)。 +#[derive(Copy, Clone, Debug, Default, PartialEq, Eq)] +pub(crate) struct SendErrPreds { + pub is_timeout: bool, + pub is_connect: bool, + pub is_decode: bool, + pub is_body: bool, + pub is_request: bool, + pub is_builder: bool, +} + +impl SendErrPreds { + pub(crate) fn of(e: &reqwest::Error) -> Self { + Self { + is_timeout: e.is_timeout(), + is_connect: e.is_connect(), + is_decode: e.is_decode(), + is_body: e.is_body(), + is_request: e.is_request(), + is_builder: e.is_builder(), + } + } +} + +/// 流式 send 阶段错误分类,返回类词(置于 bail 串首位供下游按串分类)。 +/// +/// reqwest 0.12 的 Display 只给 "error sending request for url",不含 source 关键词, +/// 纯 Display 串分类必得 unknown → Fatal 零重试。此处用谓词确定性给类词: +/// timeout/connect/decode/body 各归本类;is_request(非 builder)覆盖陈旧池化连接 +/// 首包 RST(ConnectionReset 10054)等传输层发送错,归 connect(与 retry 的 +/// is_reqwest_error_retryable 对 is_request 可重试口径一致);builder 构造错归 +/// request(不可重试)。类词须落在 stream_recv classify_status_or_class 传输词表内, +/// 改任一侧需同步。 +pub(crate) fn send_error_class(p: SendErrPreds) -> &'static str { + if p.is_timeout { + "timeout" + } else if p.is_connect { + "connect" + } else if p.is_decode { + "decode" + } else if p.is_body { + "body" + } else if p.is_request && !p.is_builder { + "connect" + } else { + "request" + } +} + /// 按 provider_type 构建 LLM Provider 实例（统一选择逻辑，消除调用方重复 match） /// /// `anthropic` 协议走 AnthropicCompatProvider（GLM 订阅端点 / Claude 官方）， diff --git a/crates/df-ai/src/openai_compat.rs b/crates/df-ai/src/openai_compat.rs index 1dab082..85508a5 100644 --- a/crates/df-ai/src/openai_compat.rs +++ b/crates/df-ai/src/openai_compat.rs @@ -502,7 +502,9 @@ impl LlmProvider for OpenAICompatProvider { let resp = match resp { Ok(r) => r, Err(e) => { - if is_reqwest_error_retryable(&e) { + // is_decode 补可重试:body 中途断连在 reqwest 归 decode 类,complete 重发 + // 幂等安全(受 3 次/30s 预算约束,真正格式错重试同败,预算内止损)。 + if is_reqwest_error_retryable(&e) || e.is_decode() { return AttemptOutcome::Retryable(format!("请求失败(可重试): {}", e)); } return AttemptOutcome::Fatal(format!("请求失败(不可重试): {}", e)); @@ -520,10 +522,16 @@ impl LlmProvider for OpenAICompatProvider { error!(%status, %body, "LLM API 调用失败(不可重试)"); return AttemptOutcome::Fatal(msg); } - // body 解析: 解析错属 Fatal(响应已成功送达，重试也会因同样格式失败) + // body 解析: serde 结构错属 Fatal(响应已成功送达,重试同样失败); + // body 读取中途断连在 reqwest 同归 is_decode,重发幂等安全 → 可重试。 let body: OpenAiResponse = match resp.json().await { Ok(b) => b, - Err(e) => return AttemptOutcome::Fatal(format!("响应解析失败: {}", e)), + Err(e) => { + if e.is_decode() { + return AttemptOutcome::Retryable(format!("响应解码失败(可重试): {}", e)); + } + return AttemptOutcome::Fatal(format!("响应解析失败: {}", e)); + } }; let choice = match body.choices.into_iter().next() { Some(c) => c, @@ -570,8 +578,23 @@ impl LlmProvider for OpenAICompatProvider { let resp = match tokio::time::timeout(Duration::from_secs(60), send_future).await { Ok(Ok(r)) => r, Ok(Err(e)) => { - tracing::error!(error = %e, is_timeout = e.is_timeout(), "OpenAI 流式 send 失败"); - return Err(e.into()); + // 谓词结构化分类:Display 串不含 source 关键词("error sending request for url"), + // 纯串分类必 unknown → Fatal 零重试(陈旧连接首包 RST 就此裸奔)。类词置串首, + // source 链兜底真实传输词(reset/broken pipe 等),下游按类词走可重试。 + let class = crate::send_error_class(crate::SendErrPreds::of(&e)); + tracing::error!( + error = %e, + class = class, + url = %self.chat_url(), + source = ?std::error::Error::source(&e), + "OpenAI 流式 send 失败" + ); + anyhow::bail!( + "流式 send 失败({}): {} [src={:?}]", + class, + e, + std::error::Error::source(&e) + ); } Err(_elapsed) => { tracing::error!(url = %self.chat_url(), "OpenAI 流式 send 超时(60s 未返回响应头)"); @@ -1279,4 +1302,61 @@ mod tests { let tc4 = c4.tool_calls.as_ref().expect("应有 tool_calls").first().unwrap(); assert_eq!(tc4.id.as_deref(), Some("call_xyz"), "非空 id 原样透传"); } + + // ---------- 流式 send 错误分类(谓词纯函数,reqwest::Error 无公开构造器以快照模拟) ---------- + + /// send 阶段陈旧连接首包 RST(reqwest 形态 is_request=true + 源 ConnectionReset) + /// → connect 类词 → 下游可重试。根治 Display 串无 source 关键词被分 unknown 零重试。 + #[test] + fn send_error_connection_reset_is_retryable_class() { + let class = crate::send_error_class(crate::SendErrPreds { + is_request: true, + ..Default::default() + }); + assert_eq!(class, "connect"); + } + + /// timeout/connect/decode/body/request 形态各归本类,且类词均须落在下游传输可重试 + /// 词表内(镜像 src-tauri stream_recv classify_status_or_class 词表,改任一侧需同步)。 + #[test] + fn send_error_transport_classes_are_retryable_words() { + const RETRYABLE_WORDS: &[&str] = &[ + "timeout", + "connect", + "transport", + "decoding", + "decode", + "body", + "eof", + "reset", + "broken pipe", + ]; + let cases = [ + crate::SendErrPreds { is_timeout: true, ..Default::default() }, + crate::SendErrPreds { is_connect: true, ..Default::default() }, + crate::SendErrPreds { is_decode: true, ..Default::default() }, + crate::SendErrPreds { is_body: true, ..Default::default() }, + crate::SendErrPreds { is_request: true, ..Default::default() }, + ]; + for p in cases { + let class = crate::send_error_class(p); + assert!( + RETRYABLE_WORDS.contains(&class), + "类词 {} 应在传输可重试词表内", + class + ); + } + } + + /// builder 构造错(请求本体问题,重试同败)→ request 类词,不在词表 → 下游 Fatal, + /// 与 retry::is_reqwest_error_retryable 的 is_builder 排除口径一致。 + #[test] + fn send_error_builder_is_fatal_class() { + let class = crate::send_error_class(crate::SendErrPreds { + is_request: true, + is_builder: true, + ..Default::default() + }); + assert_eq!(class, "request"); + } } diff --git a/src-tauri/src/commands/ai/commands/chat.rs b/src-tauri/src/commands/ai/commands/chat.rs index b0c293a..53467ee 100644 --- a/src-tauri/src/commands/ai/commands/chat.rs +++ b/src-tauri/src/commands/ai/commands/chat.rs @@ -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, + skill: Option, + model_override: Option, + conversation_id: Option, + parts: Option>, + mention_spans: Option>, + pin_active: bool, +) -> Result { // 获取活跃提供商（只读，失败可直接返回，不影响生成标志） - 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::(); + let state = app.state::(); + 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(()); } }; diff --git a/src-tauri/src/commands/ai/stream_recv.rs b/src-tauri/src/commands/ai/stream_recv.rs index fc895cb..950a833 100644 --- a/src-tauri/src/commands/ai/stream_recv.rs +++ b/src-tauri/src/commands/ai/stream_recv.rs @@ -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() {