diff --git a/crates/df-ai/src/ai_tools.rs b/crates/df-ai/src/ai_tools.rs index 5f4c3bd..f5e57d8 100644 --- a/crates/df-ai/src/ai_tools.rs +++ b/crates/df-ai/src/ai_tools.rs @@ -66,7 +66,8 @@ pub struct AiTool { // - approval_hint:审批卡文案模板 + 关键参数键 // - display_verb:轻量中文动作标签 // - goal:从工具调用推理目标描述(动词 + 参数键 + 取值风格 + 缺参兜底文案) -// - sig:漂移检测签名取哪些参数组(组内多键为取值回退序,数值键按 JSON 形态序列化) +// - sig:漂移检测签名取哪些参数组(组内多键为取值回退序,数值键按 JSON 形态序列化; +// 注册点未覆盖但换值即换操作的键由 SIG_EXTRA_GROUPS 补参,见 tool_signature) // ============================================================ /// 取值风格(goal 维度):从参数值到展示文本的取法 @@ -98,14 +99,16 @@ pub struct GoalMeta { pub enum SigValKind { /// 字符串取值(as_str,缺失为空串) Str, - /// Value::to_string 形态(数字 50 -> "50",缺失为空串) + /// Value::to_string 形态(数字 50 -> "50",布尔 true -> "true");组内全键参与归一化 + /// (排序拼接,缺失键空占位) Json, } /// 漂移检测签名元数据:tool_call_signature 消费 #[derive(Debug, Clone, Copy, PartialEq)] pub struct SigMeta { - /// 有序参数组;每组内多键表示取值回退序(如 pattern 缺失落 query) + /// 有序参数组;组内多键:Str 组为取值回退序(如 pattern 缺失落 query),Json 组为 + /// 全键参与(排序拼接,防同组其他键换值被压成同签名) pub groups: &'static [(&'static [&'static str], SigValKind)], } @@ -227,36 +230,135 @@ pub fn goal_text(goal: GoalMeta, args: Option<&Value>) -> Option { rendered.or_else(|| goal.fallback.map(|f| f.to_string())) } -/// 签名体:每组渲染 k=v,组间逗号连接。标签恒为首键(与旧实现 pair("pattern", ...) 一致, -/// 取值来自回退键时标签不变)。Str 组取组内首个字符串值键的值,全非字符串落末键; -/// Json 组按 Value::to_string(缺失为空串)。 -pub fn sig_body(sig: SigMeta, args: &Value) -> String { - let arg_str = |key: &str| -> String { - args.get(key).and_then(|v| v.as_str()).map(|s| s.to_string()).unwrap_or_default() - }; - sig.groups +/// 组渲染口径:Always = 基础组逐组渲染,缺键空值占位(消费方按固定串断言,形态不可漂移); +/// Present = 补参口径,只渲染 args 中实际出现的键,键未出现视同未换参,不渲染空占位。 +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum GroupPresence { + Always, + Present, +} + +/// 组渲染:每组产出 k=v 片段,组间逗号连接。Str 组标签恒为首键(与旧实现 +/// pair("pattern", ...) 一致,取值来自回退键时标签不变),取组内首个字符串值键的值, +/// 全非字符串落末键。Json 组全键参与归一化:数值/布尔键换值即换操作,只取 keys[0] 会把 +/// 同组其他键换值的不同调用压成同一签名(G2 熔断以签名为重复判据,压缩即误判重复), +/// 排序后拼接保证与键序无关的确定性。 +fn render_groups( + groups: &'static [(&'static [&'static str], SigValKind)], + args: &Value, + presence: GroupPresence, +) -> String { + groups .iter() - .map(|(keys, kind)| { - let value = match kind { - SigValKind::Json => args.get(keys[0]).map(|v| v.to_string()).unwrap_or_default(), - SigValKind::Str => { - let key = keys - .iter() - .find(|k| args.get(**k).and_then(|v| v.as_str()).is_some()) - .unwrap_or(&keys[keys.len() - 1]); - arg_str(key) + .filter_map(|(keys, kind)| match kind { + SigValKind::Json => { + let mut parts: Vec = keys + .iter() + .copied() + .filter(|k| presence == GroupPresence::Always || args.get(*k).is_some()) + .map(|k| match args.get(k) { + Some(v) => format!("{}={}", k, v.to_string()), + None => format!("{}=", k), + }) + .collect(); + parts.sort_unstable(); + if parts.is_empty() { + None + } else { + Some(parts.join("|")) } - }; - format!("{}={}", keys[0], value) + } + SigValKind::Str => { + if presence == GroupPresence::Present { + // 补参组恒单键:键未出现整组跳过,值非字符串视同缺失 + return args + .get(keys[0]) + .and_then(|v| v.as_str()) + .map(|s| format!("{}={}", keys[0], s)); + } + let key = keys + .iter() + .find(|k| args.get(**k).and_then(|v| v.as_str()).is_some()) + .unwrap_or(&keys[keys.len() - 1]); + Some(format!( + "{}={}", + keys[0], + args.get(*key).and_then(|v| v.as_str()).unwrap_or("") + )) + } }) .collect::>() .join(",") } -/// 工具调用签名:有 SigMeta 元数据按组渲染,未注册回退「工具名 + 全参 JSON」保守兜底。 +/// 签名体:基础组按 Always 口径渲染。 +pub fn sig_body(sig: SigMeta, args: &Value) -> String { + render_groups(sig.groups, args, GroupPresence::Always) +} + +/// 签名补参表:注册点 SigMeta 组未覆盖、但「换值即换操作」的键,按工具名追加进签名。 +/// +/// 为什么在 df-ai 签名层:G2 熔断以签名为重复判据,签名过粗会把合法换参误判成 +/// 「同一操作反复」(同工具换参是不同操作)。注册点 SigMeta 声明主参数组,换参敏感 +/// 但未覆盖的键由本表统一补参;若日后注册点把这些键并入 groups,须同步从本表移除 +/// (双源会重复渲染同键,不改变同异判定但产生冗余片段)。 +/// +/// 渲染走 Present 口径(键出现在 args 才进签名):补参键缺省 = 未换参,既有签名逐字 +/// 不变。纯格式/开关类键(-n/-C/context_chars/timeout_secs)不补,变体不构成意图变化, +/// 补进签名只会稀释重复判据。 +static SIG_EXTRA_GROUPS: &[(&str, &'static [(&'static [&'static str], SigValKind)])] = &[ + // read_file:同 path 换 search 是换词定位,非重复读 + ("read_file", &[(&["search"], SigValKind::Str)]), + // search_files:换 recursive 换搜索范围,换 offset/limit 是翻页续搜 + // (schema 无 -i/output_mode,大小写不敏感是内建语义非参数) + ( + "search_files", + &[ + (&["recursive"], SigValKind::Json), + (&["offset"], SigValKind::Json), + (&["limit"], SigValKind::Json), + ], + ), + // grep:换 -i 是敏感/不敏感两种匹配,换 output_mode 是行内容/文件名/计数三种形态, + // 换 max_results 是扩容重取(schema 无 offset/limit,数量参数实为 max_results) + ( + "grep", + &[ + (&["-i"], SigValKind::Json), + (&["output_mode"], SigValKind::Str), + (&["max_results"], SigValKind::Json), + ], + ), + // run_command:同命令不同 working_dir 是在不同目录执行 + ("run_command", &[(&["working_dir"], SigValKind::Str)]), +]; + +/// 按工具名取补参组(未列名返回 None) +fn sig_extra_groups(name: &str) -> Option<&'static [(&'static [&'static str], SigValKind)]> { + SIG_EXTRA_GROUPS + .iter() + .find(|(n, _)| *n == name) + .map(|(_, g)| *g) +} + +/// 工具调用签名:有 SigMeta 元数据按组渲染并追加补参组,未注册回退「工具名 + 全参 JSON」 +/// 保守兜底(全参已含补参键,无需追加)。 +/// +/// G2 判定语义:同工具同参连续 = 同签名,计重复;同工具换参 = 补参键进签名,签名不同 +/// 不累计(正常探索)。 pub fn tool_signature(name: &str, args: &Value) -> String { match sig_of(name) { - Some(sig) => format!("{}:{}", name, sig_body(sig, args)), + Some(sig) => { + let mut body = render_groups(sig.groups, args, GroupPresence::Always); + let extra = sig_extra_groups(name) + .map(|g| render_groups(g, args, GroupPresence::Present)) + .unwrap_or_default(); + if !extra.is_empty() { + body.push(','); + body.push_str(&extra); + } + format!("{}:{}", name, body) + } None => format!("{}:name={},args={}", name, name, args.to_string()), } } @@ -545,4 +647,119 @@ mod tests { let sig = tool_signature("custom_tool", &json!({"x": 1})); assert!(sig.starts_with("custom_tool:name=custom_tool,args=")); } + + /// Json 组全键参与:同组多键排序拼接(与 args 键序无关),任一键换值即换片段。 + #[test] + fn sig_body_json_group_includes_all_keys() { + use serde_json::json; + let page = SigMeta { groups: &[(&["offset", "limit"], SigValKind::Json)] }; + // 排序拼接:limit < offset 字典序,与 args 内键书写顺序无关 + assert_eq!(sig_body(page, &json!({"offset": 50, "limit": 100})), "limit=100|offset=50"); + assert_eq!(sig_body(page, &json!({"limit": 100, "offset": 50})), "limit=100|offset=50"); + // 任一键换值 → 不同片段 + assert_ne!( + sig_body(page, &json!({"offset": 50, "limit": 100})), + sig_body(page, &json!({"offset": 60, "limit": 100})) + ); + // 全缺失逐键空占位(既有缺省口径);布尔键按 JSON 形态 + assert_eq!(sig_body(page, &json!({})), "limit=|offset="); + let flag = SigMeta { groups: &[(&["-i"], SigValKind::Json)] }; + assert_eq!(sig_body(flag, &json!({"-i": true})), "-i=true"); + } + + /// 注册与生产同形态的探针工具。df-ai 单测进程 GLOBAL_TOOL_META 无生产注册, + /// 不种入则 tool_signature 走未注册兜底,测不到「注册点组 + 补参」合流分支。 + /// 进程级表跨测试共享,各处注册值恒相同,幂等无顺序依赖。 + fn seed_signature_probe_tools() { + let read_sig = SigMeta { + groups: &[ + (&["path"], SigValKind::Str), + (&["offset"], SigValKind::Json), + (&["limit"], SigValKind::Json), + ], + }; + let search_sig = SigMeta { + groups: &[ + (&["pattern", "query"], SigValKind::Str), + (&["path", "glob"], SigValKind::Str), + ], + }; + let cmd_sig = SigMeta { groups: &[(&["command"], SigValKind::Str)] }; + let mut reg = AiToolRegistry::new(); + reg.register_meta("read_file", "读取", object_schema(vec![]), RiskLevel::Low, + ToolMeta { sig: Some(read_sig), ..Default::default() }, dummy_handler()); + reg.register_meta("search_files", "搜索文件", object_schema(vec![]), RiskLevel::Low, + ToolMeta { sig: Some(search_sig), ..Default::default() }, dummy_handler()); + reg.register_meta("grep", "跨文件内容搜索", object_schema(vec![]), RiskLevel::Low, + ToolMeta { sig: Some(search_sig), ..Default::default() }, dummy_handler()); + reg.register_meta("run_command", "执行命令", object_schema(vec![]), RiskLevel::High, + ToolMeta { sig: Some(cmd_sig), ..Default::default() }, dummy_handler()); + } + + /// 补参:合法换参 → 签名不同(G2 目标语义:换参不共享签名,不再误判重复); + /// 同参 → 签名相同(重复判据正例仍成立)。 + #[test] + fn tool_signature_extra_params_distinguish_calls() { + use serde_json::json; + seed_signature_probe_tools(); + + // read_file 同 path 换 search:不同签名 + let s1 = tool_signature("read_file", &json!({"path": "a.rs", "search": "fn sig"})); + let s2 = tool_signature("read_file", &json!({"path": "a.rs", "search": "struct cfg"})); + assert_ne!(s1, s2, "read_file 换 search 应不同签名: {} vs {}", s1, s2); + assert!(s1.contains("search=fn sig"), "read_file 签名应含 search: {}", s1); + + // search_files 只差 offset(翻页续搜):不同签名 + let s1 = tool_signature("search_files", &json!({"pattern": "todo", "path": "src", "offset": 0})); + let s2 = tool_signature("search_files", &json!({"pattern": "todo", "path": "src", "offset": 50})); + assert_ne!(s1, s2, "search_files 换 offset 应不同签名: {} vs {}", s1, s2); + + // grep 只差 -i / output_mode / max_results(schema 无 offset/limit,数量参数实为 + // max_results):换匹配方式/输出形态/扩容重取都是不同操作 + let base = tool_signature("grep", &json!({"pattern": "err", "path": "src"})); + let ci = tool_signature("grep", &json!({"pattern": "err", "path": "src", "-i": true})); + let mode = tool_signature("grep", &json!({"pattern": "err", "path": "src", "output_mode": "files_with_matches"})); + let capped = tool_signature("grep", &json!({"pattern": "err", "path": "src", "max_results": 200})); + assert_ne!(base, ci, "grep 换 -i 应不同签名: {} vs {}", base, ci); + assert_ne!(base, mode, "grep 换 output_mode 应不同签名: {} vs {}", base, mode); + assert_ne!(base, capped, "grep 换 max_results 应不同签名: {} vs {}", base, capped); + + // run_command 同命令换 working_dir:不同签名 + let s1 = tool_signature("run_command", &json!({"command": "cargo test"})); + let s2 = tool_signature("run_command", &json!({"command": "cargo test", "working_dir": "crates/df-ai"})); + assert_ne!(s1, s2, "run_command 换 working_dir 应不同签名: {} vs {}", s1, s2); + + // 同工具同参:签名相同(同签名连续才算 G2 重复) + assert_eq!( + tool_signature("grep", &json!({"pattern": "err", "path": "src"})), + base + ); + } + + /// 补参键缺席时签名与既有格式逐字一致(向后兼容:消费方按固定串断言,空补参不占位)。 + #[test] + fn tool_signature_without_extra_keys_unchanged() { + use serde_json::json; + seed_signature_probe_tools(); + assert_eq!( + tool_signature("read_file", &json!({"path": "big.log", "offset": 50, "limit": 100})), + "read_file:path=big.log,offset=50,limit=100" + ); + assert_eq!( + tool_signature("read_file", &json!({})), + "read_file:path=,offset=,limit=" + ); + assert_eq!( + tool_signature("grep", &json!({"pattern": "MAX", "path": "src", "case_sensitive": true})), + "grep:pattern=MAX,path=src" + ); + assert_eq!( + tool_signature("search_files", &json!({"query": "TODO", "glob": "**/*.rs"})), + "search_files:pattern=TODO,path=**/*.rs" + ); + assert_eq!( + tool_signature("run_command", &json!({"command": "ls -la", "timeout": 5000})), + "run_command:command=ls -la" + ); + } } diff --git a/crates/df-ai/src/anthropic_compat.rs b/crates/df-ai/src/anthropic_compat.rs index e063312..4c56148 100644 --- a/crates/df-ai/src/anthropic_compat.rs +++ b/crates/df-ai/src/anthropic_compat.rs @@ -626,16 +626,23 @@ impl LlmProvider for AnthropicCompatProvider { error!(%status, %text, "Anthropic API 调用失败(不可重试)"); return AttemptOutcome::Fatal(msg); } - let resp: AnthropicResponse = match resp.json().await { - Ok(r) => r, - // serde 结构错属 Fatal(重试同败);body 读取中途断连同归 is_decode,可重试。 + // body 解析:text 先取再 serde 分层判定:读取中断(text Err,reqwest 归 + // is_decode 类,complete 重发幂等安全)可重试;JSON 结构错属 Fatal(响应已 + // 完整送达,重试同败)。Response::json 把 serde 错同包成 is_decode,先 text + // 后 from_str 才能区分二者,避免确定性结构错被重试白烧预算。 + let text = match resp.text().await { + Ok(t) => t, Err(e) => { - if e.is_decode() { - return AttemptOutcome::Retryable(format!("响应解码失败(可重试): {}", e)); + if is_reqwest_error_retryable(&e) || e.is_decode() { + return AttemptOutcome::Retryable(format!("响应读取失败(可重试): {}", e)); } - return AttemptOutcome::Fatal(format!("响应解析失败: {}", e)); + return AttemptOutcome::Fatal(format!("响应读取失败(不可重试): {}", e)); } }; + let resp: AnthropicResponse = match serde_json::from_str(&text) { + Ok(r) => r, + Err(e) => return AttemptOutcome::Fatal(format!("响应解析失败: {}", e)), + }; // content 块中拼接 text,收集 tool_use,累积 thinking let mut text = String::new(); let mut reasoning = String::new(); diff --git a/crates/df-ai/src/context/manager_tests.rs b/crates/df-ai/src/context/manager_tests.rs index 617aa6b..2e0ffce 100644 --- a/crates/df-ai/src/context/manager_tests.rs +++ b/crates/df-ai/src/context/manager_tests.rs @@ -51,7 +51,7 @@ fn estimate_message_counts_parts_tokens() { mm_tokens ); - // url 模式(无字节)也按 URL 长度估算,不爆 + // url 模式(无字节)按每图 ~1600 token 地板估(生效于 ×0.35 折扣之后,防预算漏裁) let url_msg = ChatMessage::user_parts( "t", vec![crate::provider::ContentPart::image_url("https://example.com/x.png")], @@ -777,6 +777,34 @@ fn topic_marker_take_is_idempotent_and_clears() { assert!(mgr.take_topic_marker().is_none(), "marker 一次性消费, 二次 take 必空"); } +#[test] +fn topic_marker_reset_after_restore_from_messages() { + // restore_from_messages(DB 加载)重放历史时 push 会照常跑主题切换检测, + // 历史里"当时已发生"的切换不得残留成 pending marker——否则加载后首轮 take + // 拿到假话题锚点,后续话题归属被误判为切换、多插一次系统标记。 + let history = vec![ + ChatMessage::user("帮我重构这段代码"), // topic=code(高置信) + ChatMessage::assistant("好的"), + ChatMessage::user("创建项目并绑定目录"), // topic=project(高置信)→ 重放中会置位 marker + ]; + let mut mgr = ContextManager::new(cfg(100_000)); + mgr.restore_from_messages(history); + assert!( + mgr.take_topic_marker().is_none(), + "重放历史不得残留 pending_topic_marker(假话题锚点)" + ); + // restore 之后的实时 push 检测不受影响:末条 user topic=project,新 push code → 触发 + mgr.push(ChatMessage::user("帮我重构这段代码")); + let marker = mgr + .take_topic_marker() + .expect("restore 后的实时切换仍应触发"); + assert!( + marker.contains("project") && marker.contains("code"), + "marker: {}", + marker + ); +} + // ============================================================ // msg-split-phase1:增量写路径基线(persisted_msg_count)+ 全量重写标志(needs_full_rewrite) // ============================================================ diff --git a/crates/df-ai/src/context/mod.rs b/crates/df-ai/src/context/mod.rs index 0c3272e..2f33431 100644 --- a/crates/df-ai/src/context/mod.rs +++ b/crates/df-ai/src/context/mod.rs @@ -316,6 +316,10 @@ impl ContextManager { } self.persisted_msg_count = len; self.needs_full_rewrite = false; + // 重放历史时 push 会照常跑主题切换检测,历史里"当时已发生"的切换会残留进 marker; + // 留着它,加载后首轮 take_topic_marker 就拿到假话题锚点、误插一次切换标记。 + // 恢复即重置:切换检测只对 restore 之后的实时 push 生效。 + self.pending_topic_marker = None; } /// 就地替换某条 tool_result 的内容(兼容审批 replace_tool_result) diff --git a/crates/df-ai/src/context/sanitize.rs b/crates/df-ai/src/context/sanitize.rs index 3017e3e..de08498 100644 --- a/crates/df-ai/src/context/sanitize.rs +++ b/crates/df-ai/src/context/sanitize.rs @@ -196,8 +196,10 @@ pub fn sanitize_messages(messages: Vec) -> Vec { /// 占位与非占位统一改"补头自愈":任何工具结果都不丢(除非无 tool_call_id 的异常数据,无法 /// 补头配对,只能保留——由协议层兜底)。 /// -/// **未闭合 vs TOOL_MISSING_PREFIX 头**:已带 TOOL_MISSING_PREFIX 的 id 视为已配对(出口断言 -/// 补的占位头),不重复补头(否则同 id 双头致 400)。 +/// **未闭合 vs TOOL_MISSING_PREFIX 头**:判定只看 head_ids.contains——真配对(占位头在)由 +/// contains 覆盖,不会重复补头;头不在的 prefix id result(前轮占位头被裁剪丢)即 orphan, +/// 同样补头自愈。补出的头 id 为前缀叠加(tool_missing_tool_missing_X,双层前缀),头与 +/// result 改写后 id 一致,唯一配对闭合(双层前缀为已知无害副作用)。 pub fn drop_reverse_orphans(messages: Vec) -> Vec { use std::collections::HashSet; // 收集所有保留 assistant 头的 tool_call.id(正向 orphan 处理后残留的头里的 id)。 @@ -227,16 +229,14 @@ pub fn drop_reverse_orphans(messages: Vec) -> Vec { out.push(m); continue; }; - // 已配对(头在)→ 原样保留。 + // 已配对(头在)→ 原样保留。TOOL_MISSING_PREFIX 前缀 id 无需特判:真配对(占位头在) + // 已被 contains 覆盖;头不在(前轮占位头被裁剪丢掉)即是 orphan,同样走补头自愈。 + // 补出的头 id 为前缀叠加(tool_missing_tool_missing_X,双层前缀),头+result 改写后 + // id 一致,唯一配对闭合,功能正确(双层前缀为已知无害副作用)。 if head_ids.contains(id) { out.push(m); continue; } - // TOOL_MISSING_PREFIX 占位头 id:已被出口断言/上一轮补过,视为配对,不重复补头。 - if id.starts_with(TOOL_MISSING_PREFIX) { - out.push(m); - continue; - } // orphan tool_result(无头)→ **补占位头自愈**(不丢 result)。 // 头 id 复用 result 的 tool_call_id 精确配对(同 assert_placeholder_pairing 自愈模式)。 let head_id = format!("{}{}", TOOL_MISSING_PREFIX, id); @@ -306,13 +306,16 @@ pub fn assert_placeholder_pairing( .collect(); // 找 orphan tool_result(id 不在 head_ids)。占位/非占位都需补头(否则 400)。 + // 注:TOOL_MISSING_PREFIX 前缀 id 不豁免——真配对(占位头在)已被 head_ids.contains + // 覆盖;头不在的 prefix id result(前轮占位头被裁剪丢)正是 orphan,必须补头闭合。 + // 补出的头 id 为前缀叠加(tool_missing_tool_missing_X,双层前缀),唯一配对,功能正确。 // 单次遍历产出新 Vec:orphan result 前插一个占位头(同 id),其余原样保留保序。 let mut healed = 0u32; let mut out: Vec = Vec::with_capacity(messages.len()); for m in messages.into_iter() { if matches!(m.role, MessageRole::Tool) { if let Some(id) = m.tool_call_id.as_deref() { - if !head_ids.contains(id) && !id.starts_with(TOOL_MISSING_PREFIX) { + if !head_ids.contains(id) { // orphan result → 前补占位头(TOOL_MISSING_PREFIX + 原 id,确保唯一且与 result 配对) let head_id = format!("{}{}", TOOL_MISSING_PREFIX, id); let head = ChatMessage::assistant_with_tools( @@ -419,6 +422,11 @@ pub fn ensure_sequence_legal(messages: Vec) -> Vec { last.tool_calls.get_or_insert_with(Vec::new).extend(calls); } } + // 多模态 parts 随合并迁移:空 content 纯图消息的 content 拼接为空操作, + // parts 不迁移则图片片随整条消息丢弃,图对 LLM 隐身 + if let Some(parts) = m.parts { + last.parts.get_or_insert_with(Vec::new).extend(parts); + } merged += 1; continue; } @@ -921,9 +929,11 @@ mod tests { } #[test] - fn drop_reverse_orphans_no_redundant_head_for_tool_missing_prefix_id() { - // 边界:orphan result 的 id 已是 TOOL_MISSING_PREFIX(出口断言/上轮补过)→ 视为已配对, - // 不重复补头(否则同 id 双头致 400)。 + fn drop_reverse_orphans_heals_prefix_id_orphan_with_nested_prefix_head() { + // 边界(盲区修复回归):orphan result 的 id 已是 TOOL_MISSING_PREFIX 但占位头不在 + // (前轮占位头被裁剪丢)→ 即 orphan,同样补头自愈(旧"前缀即已配对"豁免会吞掉此场景, + // 三元组不闭合直送 provider 400)。补出的头 id 为双层前缀叠加 + // (tool_missing_tool_missing_X),头与 result 改写后 id 一致,唯一配对闭合。 let msgs = vec![ ChatMessage::user("问题"), ChatMessage::tool_result("tool_missing_already", "已补过头的结果"), @@ -936,17 +946,39 @@ mod tests { .filter(|m| matches!(m.role, MessageRole::Tool)) .count(), 1, - "TOOL_MISSING_PREFIX id 的 result 应保留" + "TOOL_MISSING_PREFIX id 的 orphan result 应保留(补头自愈)" ); - // 不补新头(原样保留,无新 TOOL_MISSING_PREFIX 占位头插入) - let new_heads = sanitized + // 补双层前缀占位头(头 id 含原 id 且以前缀开头) + let healed_heads: Vec<&str> = sanitized .iter() .filter(|m| matches!(m.role, MessageRole::Assistant)) .filter_map(|m| m.tool_calls.as_ref()) .flatten() - .filter(|c| c.id.starts_with(TOOL_MISSING_PREFIX)) - .count(); - assert_eq!(new_heads, 0, "TOOL_MISSING_PREFIX id result 不应重复补头"); + .map(|c| c.id.as_str()) + .collect(); + assert_eq!( + healed_heads.len(), + 1, + "prefix id orphan 应补 1 个占位头, 实际 {:?}", + healed_heads + ); + assert!( + healed_heads[0].starts_with(TOOL_MISSING_PREFIX) + && healed_heads[0][TOOL_MISSING_PREFIX.len()..].starts_with(TOOL_MISSING_PREFIX), + "补的头 id 应为双层前缀叠加(tool_missing_tool_missing_X), 实际 {}", + healed_heads[0] + ); + // result 的 tool_call_id 改写为占位头 id(闭合配对) + let result_id = sanitized + .iter() + .find(|m| matches!(m.role, MessageRole::Tool)) + .and_then(|m| m.tool_call_id.clone()) + .expect("应有 tool_result"); + assert_eq!( + result_id, + healed_heads[0], + "result id 应与补的头 id 一致(闭合配对)" + ); } #[test] @@ -1205,4 +1237,21 @@ mod tests { sanitized.first().map(|m| format!("{:?}", m.role)) ); } + + /// 连续同 role 合并须迁移多模态 parts:空 content 纯图消息与相邻同 role 消息 + /// 合并时,图片片并入前一条(此前只拼 content,纯图消息整条丢弃,图对 LLM 隐身) + #[test] + fn sequence_merge_moves_multimodal_parts() { + let pure_image = ChatMessage::user_parts( + "", + vec![crate::provider::ContentPart::image_url("https://x/a.png")], + ); + let fixed = ensure_sequence_legal(vec![pure_image, ChatMessage::user("继续")]); + + assert_eq!(fixed.len(), 1, "连续 user 应合并为一条"); + let parts = fixed[0].parts.as_ref().expect("图片片应迁移到合并结果"); + assert_eq!(parts.len(), 1); + assert!(parts[0].is_image(), "迁移的应是图片片"); + assert_eq!(fixed[0].content, "继续"); + } } diff --git a/crates/df-ai/src/context_helpers.rs b/crates/df-ai/src/context_helpers.rs index c8a23ad..8f4bb1a 100644 --- a/crates/df-ai/src/context_helpers.rs +++ b/crates/df-ai/src/context_helpers.rs @@ -48,6 +48,8 @@ impl Default for TokenEstimator { // 2026-08-05(F-260801 Phase4):url 模式无字节可估,仅按 URL 长度估(几十字符≈几 token)会致含图 // 消息 history_tokens 严重低估 → 预算裁剪不触发 → 多图/长会话超 provider 上限 400。地板按每图 // ~1600 token 保守估(对齐 Anthropic 图片 token 成本量级),宁可高估触发裁剪也不低估漏裁。 +// 注意单位是 token 不是字符:地板必须在 ×chars_ratio 折扣之后直接加 token,若混进 char_count +// 会被 0.35 稀释成 560,保底名存实亡。 const URL_IMAGE_TOKEN_FLOOR: usize = 1600; impl TokenEstimator { @@ -59,6 +61,10 @@ impl TokenEstimator { /// 这里把 parts 的文本/base64 按同一 chars_ratio 粗估累加(base64 视为密集字符,0.35 偏保守)。 pub fn estimate_message(&self, msg: &ChatMessage) -> u32 { let mut char_count = msg.content.chars().count(); + // url 图地板单独按 token 累加:地板语义是"每图至少 ~1600 token",必须生效于 + // ×chars_ratio 折扣之后——若折进 char_count 会被 0.35 稀释成 560,1600 保底失效, + // 含 url 图消息的 history_tokens 低估 → 预算裁剪漏触发 → provider 超限 400。 + let mut url_image_floor_tokens: u32 = 0; if let Some(parts) = &msg.parts { for p in parts { match p { @@ -66,19 +72,20 @@ impl TokenEstimator { crate::provider::ContentPart::Image { base64, url, .. } => { // 0.35 按 base64 字节数粗估,显著高于厂商实际(OpenAI 按像素非字节)。 // 偏保守致含图消息 token 高估、过度裁剪;降值(如 0.10~0.15)需独立评估裁剪边界,本次不改值仅标注。 - // base64 优先(多模态主载荷),url 次之;url 模式无字节,仅按 URL 长度估 + // base64 优先(多模态主载荷),url 次之;url 模式无字节,按地板直接计 token if let Some(b) = base64 { char_count += b.chars().count(); } else if url.is_some() { // F-260801 Phase4:url 模式按地板估(URL 长度估严重低估,见 URL_IMAGE_TOKEN_FLOOR 注)。 - char_count += URL_IMAGE_TOKEN_FLOOR; + url_image_floor_tokens = url_image_floor_tokens + .saturating_add(URL_IMAGE_TOKEN_FLOOR as u32); } } } } } let content_tokens = (char_count as f32 * self.chars_ratio).ceil() as u32; - let mut total = content_tokens + self.per_message_overhead; + let mut total = content_tokens + url_image_floor_tokens + self.per_message_overhead; // tool_calls 的 JSON 结构开销(role=Assistant 时可能有) if let Some(ref calls) = msg.tool_calls { @@ -833,8 +840,14 @@ impl WorkingContext { parts.push(format!("当前步骤: {} [{:?}]", step.step_name, step.status)); } - // L1b: 过去 N 轮内有变更才注入 - let threshold = self.version.current_turn.saturating_sub(recent_turns); + // L1b: 过去 N 轮内有变更才注入。窗口语义 = 最近 N 轮 [current_turn-N+1, current_turn]: + // 变更落在第 current_turn-N 轮(即 N+1 轮前)已出窗。原实现下界取 current_turn-N 且 + // 比较 >=,把 N+1 轮前的变更也划进窗口(off-by-one 多注一轮,脏段晚一轮才过期)。 + let threshold = self + .version + .current_turn + .saturating_sub(recent_turns) + .saturating_add(1); let has_active = self.goals.iter().any(|g| g.status == GoalStatus::Active); if has_active && self.version.goals_changed_at >= threshold { let active: Vec<&str> = self.goals.iter() @@ -899,6 +912,12 @@ impl WorkingContext { if !items.is_empty() { self.decisions = items; } + } else { + // 解析失败不吞:压缩后决策回填静默丢失会让"决策段凭空消失"无从排查,warn 留痕 + tracing::warn!( + "sync_after_compress: 摘要 JSON 卡片解析失败,decisions 未恢复(卡片 {} 字节)", + json_card.len() + ); } self.version.reset_all(json_card, self.version.current_turn); } @@ -1674,4 +1693,41 @@ mod tests { let out = ctx.build_injection(3); assert!(out.is_empty(), "过期后所有段均不注入, 实际: {}", out); } + + #[test] + fn working_context_injection_window_is_exactly_n_turns() { + // 边界:脏窗口 = 最近 N 轮。turn=1 的变更、窗口 3: + // current_turn=3 → 第 1 轮是第 3 轮前,恰在窗内(1,2,3 共 3 轮),注入; + // current_turn=4 → 第 1 轮是第 4 轮前,出窗不注入(off-by-one bug 下会误注)。 + let mut ctx = WorkingContext::new(); + ctx.begin_turn("窗口边界目标", 100); // current_turn=1, goals_changed_at=1 + ctx.version.current_turn = 3; + assert!( + ctx.build_injection(3).contains("窗口边界目标"), + "第 N 轮前的变更恰在窗内,应注入" + ); + ctx.version.current_turn = 4; + assert!( + ctx.build_injection(3).is_empty(), + "第 N+1 轮前的变更应出窗,不注入" + ); + } + + #[test] + fn url_image_token_floor_survives_ratio_discount() { + // url 图地板语义是"每图至少 ~1600 token",必须生效于 ×chars_ratio 折扣之后: + // 若地板折进 char_count 再乘 0.35,1600 被稀释成 ~560,保底失效 → 预算裁剪漏触发。 + let est = TokenEstimator::default(); + let msg = ChatMessage::user_parts( + "t", + vec![crate::provider::ContentPart::image_url("https://example.com/x.png")], + ); + let tokens = est.estimate_message(&msg); + assert!( + tokens >= URL_IMAGE_TOKEN_FLOOR as u32, + "url 图消息 token({}) 应 >= 地板 {},稀释 bug 下仅 ~565", + tokens, + URL_IMAGE_TOKEN_FLOOR + ); + } } diff --git a/crates/df-ai/src/intent.rs b/crates/df-ai/src/intent.rs index 65d0495..ab065b3 100644 --- a/crates/df-ai/src/intent.rs +++ b/crates/df-ai/src/intent.rs @@ -420,7 +420,7 @@ fn best_in_group(message: &str, group: &[IntentGroup]) -> Option<(Intent, f32)> for rule in cand.rules.iter() { // 英文关键词用小写匹配;中文不受 to_lowercase 影响(no-op)。 let needle = rule.kw.to_lowercase(); - if lower.contains(&needle) { + if kw_matches(&lower, &needle) { score += rule.weight; matched = true; } @@ -437,6 +437,26 @@ fn best_in_group(message: &str, group: &[IntentGroup]) -> Option<(Intent, f32)> best } +/// 关键词命中判定:ASCII 关键词要求两侧词边界(词字符 = ASCII 字母数字 + 下划线,与正则 +/// `\b` 语义一致),防子串误判——"history" 不命中 "hi"(误判 Chat)、"rapid" 不命中 +/// "api"(误判 Http)。非 ASCII(中文)关键词保持子串语义:中文无词边界概念,直接 contains。 +fn kw_matches(haystack: &str, needle: &str) -> bool { + if !needle.is_ascii() { + return haystack.contains(needle); + } + let is_word_char = |c: char| c.is_ascii_alphanumeric() || c == '_'; + haystack.match_indices(needle).any(|(pos, m)| { + haystack[..pos] + .chars() + .next_back() + .map_or(true, |c| !is_word_char(c)) + && haystack[pos + m.len()..] + .chars() + .next() + .map_or(true, |c| !is_word_char(c)) + }) +} + // ---- Intent → 工具子集映射 -------------------------------------------------- /// 按 Intent 返回工具名子集。 @@ -854,6 +874,38 @@ mod tests { assert_eq!(i, Intent::Http); } + // --- ASCII 关键词词边界(防子串误判) --- + + #[test] + fn word_boundary_history_not_chat() { + // "history" 内嵌 "hi" 不再判 Chat(无其他关键词 → Unknown) + let (i, c) = IntentRecognizer::recognize("show me the history"); + assert_eq!(i, Intent::Unknown, "history 不应命中 hi, 实际 conf={}", c); + } + + #[test] + fn word_boundary_rapid_not_http() { + // "rapid" 内嵌 "api" 不再判 Http + let (i, c) = IntentRecognizer::recognize("rapid growth"); + assert_eq!(i, Intent::Unknown, "rapid 不应命中 api, 实际 conf={}", c); + } + + #[test] + fn standalone_english_keywords_still_hit() { + // 词边界不误伤独立词命中 + let (i, _) = IntentRecognizer::recognize("call the api"); + assert_eq!(i, Intent::Http); + let (i, _) = IntentRecognizer::recognize("hi"); + assert_eq!(i, Intent::Chat); + } + + #[test] + fn cjk_adjacent_ascii_keyword_still_hits() { + // 中文紧邻 ASCII 关键词:非 ASCII 字符算边界,不破坏中文场景命中 + let (i, _) = IntentRecognizer::recognize("调用这个api"); + assert_eq!(i, Intent::Http); + } + // --- tool_subset_for --- #[test] diff --git a/crates/df-ai/src/openai_compat.rs b/crates/df-ai/src/openai_compat.rs index 2b8941b..bf495d1 100644 --- a/crates/df-ai/src/openai_compat.rs +++ b/crates/df-ai/src/openai_compat.rs @@ -522,17 +522,23 @@ impl LlmProvider for OpenAICompatProvider { error!(%status, %body, "LLM API 调用失败(不可重试)"); return AttemptOutcome::Fatal(msg); } - // body 解析: serde 结构错属 Fatal(响应已成功送达,重试同样失败); - // body 读取中途断连在 reqwest 同归 is_decode,重发幂等安全 → 可重试。 - let body: OpenAiResponse = match resp.json().await { - Ok(b) => b, + // body 解析:text 先取再 serde 分层判定:读取中断(text Err,reqwest 归 + // is_decode 类,重发幂等安全)可重试;JSON 结构错属 Fatal(响应已完整送达, + // 重试同败)。Response::json 把 serde 错同包成 is_decode,先 text 后 + // from_str 才能区分二者,避免确定性结构错被重试白烧预算。 + let text = match resp.text().await { + Ok(t) => t, Err(e) => { - if e.is_decode() { - return AttemptOutcome::Retryable(format!("响应解码失败(可重试): {}", e)); + if is_reqwest_error_retryable(&e) || e.is_decode() { + return AttemptOutcome::Retryable(format!("响应读取失败(可重试): {}", e)); } - return AttemptOutcome::Fatal(format!("响应解析失败: {}", e)); + return AttemptOutcome::Fatal(format!("响应读取失败(不可重试): {}", e)); } }; + let body: OpenAiResponse = match serde_json::from_str(&text) { + Ok(b) => b, + Err(e) => return AttemptOutcome::Fatal(format!("响应解析失败: {}", e)), + }; let choice = match body.choices.into_iter().next() { Some(c) => c, None => return AttemptOutcome::Fatal("LLM 响应无 choices".to_string()), @@ -858,6 +864,39 @@ mod tests { assert_eq!(c.error.as_deref(), Some("stream error"), "无 message 字段应兜底"); } + /// 标准 error 帧不带 choices 键 → 仍须解析出 error(此前 choices 必填, + /// 整帧 serde 失败走空 chunk 分支,error 被吞、流呈无声截断) + #[test] + fn openai_sse_error_frame_without_choices_key_parses() { + let mut acc: Option = None; + let data = r#"{"error":{"message":"insufficient quota","type":"insufficient_quota"}}"#; + let c = apply_openai_sse(data, &mut acc); + assert!(!c.finished, "error 帧不应走 finished 完成路径"); + assert_eq!(c.delta, ""); + assert_eq!(c.error.as_deref(), Some("insufficient quota")); + assert!(c.usage.is_none()); + } + + /// usage 尾帧不带 choices 键(OpenAI include_usage 标准形态)→ 累加进 usage, + /// [DONE] 终态带出(此前 choices 必填,该帧被整帧丢弃,usage 永远拿不到) + #[test] + fn openai_sse_usage_tail_frame_without_choices_key_accumulates() { + let mut acc: Option = None; + let data = r#"{"usage":{"prompt_tokens":7,"completion_tokens":3,"total_tokens":10}}"#; + let c = apply_openai_sse(data, &mut acc); + assert_eq!(c.delta, ""); + assert!(!c.finished); + let acc_after_tail = acc.as_ref().expect("usage 尾帧应写入累加器"); + assert_eq!(acc_after_tail.prompt_tokens, 7); + assert_eq!(acc_after_tail.completion_tokens, 3); + + let done = apply_openai_sse("[DONE]", &mut acc); + assert!(done.finished); + let usage = done.usage.expect("[DONE] 应带出累计 usage"); + assert_eq!(usage.prompt_tokens, 7); + assert_eq!(usage.total_tokens, 10); + } + // ---------- 多模态 convert_request ---------- /// 含图消息 → content 数组(text + image_url data URI);纯文本 → 字符串简写 diff --git a/crates/df-ai/src/openai_helpers.rs b/crates/df-ai/src/openai_helpers.rs index d0d2019..c3a5529 100644 --- a/crates/df-ai/src/openai_helpers.rs +++ b/crates/df-ai/src/openai_helpers.rs @@ -191,6 +191,10 @@ impl From for OpenAiUsage { /// SSE 流式响应 chunk #[derive(Debug, Deserialize)] pub(crate) struct OpenAiStreamChunk { + /// 流式 choice 数组。标准 error 帧(`{"error":{..}}`)与 usage 尾帧均不带 choices 键, + /// 缺省为空数组才能整帧解析成功;必填会让这类合法帧 serde 失败走空 chunk 分支, + /// 流呈无声截断。 + #[serde(default)] pub choices: Vec, /// 末 chunk(choices 为空)携带的累计 usage #[serde(default)] @@ -204,11 +208,13 @@ pub(crate) struct OpenAiStreamChunk { #[derive(Debug, Deserialize)] pub(crate) struct OpenAiStreamChoice { + /// 部分 usage 尾帧的 choice 仅带 finish_reason 无 delta 键,缺省按空 delta 处理 + #[serde(default)] pub delta: OpenAiStreamDelta, pub finish_reason: Option, } -#[derive(Debug, Deserialize)] +#[derive(Debug, Default, Deserialize)] pub(crate) struct OpenAiStreamDelta { pub content: Option, pub tool_calls: Option>, @@ -219,6 +225,8 @@ pub(crate) struct OpenAiStreamDelta { #[derive(Debug, Deserialize)] pub(crate) struct OpenAiStreamToolCall { + /// 部分端点 tool_call 增量省略 index,缺省 0(单工具调用场景语义等价) + #[serde(default)] pub index: u32, pub id: Option, pub function: Option, @@ -299,10 +307,13 @@ pub(crate) fn apply_openai_sse(data: &str, usage_accum: &mut Option) } if let Some(choice) = chunk.choices.into_iter().next() { let delta_text = choice.delta.content.unwrap_or_default(); - // "length" = max_tokens 截断,属正常终止(非断连),纳入 finished - let finished = choice.finish_reason.as_deref() == Some("stop") - || choice.finish_reason.as_deref() == Some("tool_calls") - || choice.finish_reason.as_deref() == Some("length"); + // 终态 finish_reason:stop(正常结束)/ tool_calls(转交工具执行)/ + // length(max_tokens 截断,属正常终止非断连)/ content_filter(内容审查拦截)。 + // content_filter 的流部分 provider 不发 [DONE] 帧,不纳入终态会一直悬挂 + let finished = matches!( + choice.finish_reason.as_deref(), + Some("stop") | Some("tool_calls") | Some("length") | Some("content_filter") + ); let tool_calls = choice.delta.tool_calls.map(|tcs| { tcs.into_iter() @@ -411,4 +422,21 @@ mod tests { assert_eq!(usage.prompt_cache_hit_tokens, 64); assert_eq!(usage.prompt_cache_miss_tokens, 36); } + + /// content_filter 属终态:敏感内容拦截的流部分 provider 不发 [DONE] 帧, + /// 必须靠 finish_reason 收敛,否则悬挂;未列入终态的 finish_reason 保持非终态 + #[test] + fn finish_reason_content_filter_is_terminal() { + let data = r#"{"choices":[{"delta":{"content":""},"finish_reason":"content_filter"}]}"#; + let mut accum = None; + let chunk = apply_openai_sse(data, &mut accum); + assert!(chunk.finished, "content_filter 应收敛为终态,不再悬挂"); + + let mut accum2 = None; + let chunk2 = apply_openai_sse( + r#"{"choices":[{"delta":{"content":"x"},"finish_reason":null}]}"#, + &mut accum2, + ); + assert!(!chunk2.finished, "无终态 finish_reason 的普通 delta 不应终结"); + } } diff --git a/crates/df-ai/src/sse_parser.rs b/crates/df-ai/src/sse_parser.rs index 4774023..1cb5761 100644 --- a/crates/df-ai/src/sse_parser.rs +++ b/crates/df-ai/src/sse_parser.rs @@ -17,7 +17,8 @@ //! data: 前缀提取) //! - 多字节续接:事件体只在定位到完整分隔符后才解码,跨 chunk 的多字节字符在 //! buffer 中天然拼接复原(分隔符为 ASCII,不落在多字节序列中间) -//! - 容错:解析失败时跳过该事件继续,不中断流;buffer 设 `BUF_MAX` 上限防 OOM +//! - 容错:解析失败时跳过该事件继续,不中断流;buffer 设 `BUF_MAX` 上限防 OOM, +//! 超限时按事件边界对齐保尾裁剪(丢最旧的超限事件整体),不产出半截破损事件 //! - 返回 Vec(每个元素是一个事件 data 字段拼接内容) use futures::Stream; @@ -35,6 +36,9 @@ const BUF_MAX: usize = 1 << 20; // 1 MiB pub struct SseStream { inner: S, buffer: Vec, + /// 超限裁剪置位:无边界的超限事件已整体丢弃,但其尾部字节仍在后续 chunk 中, + /// 到达收尾 boundary 前的数据须静默吞掉,避免残缺事件体被当完整事件产出 + skip_partial: bool, } impl SseStream @@ -45,6 +49,7 @@ where Self { inner, buffer: Vec::with_capacity(8192), + skip_partial: false, } } @@ -62,6 +67,17 @@ where /// 字符在 buffer 中天然拼接复原(分隔符为 ASCII,不会落在多字节序列中间)。 /// 解码用 `String::from_utf8`(失败再降级 lossy),避免把合法多字节误判为残缺。 fn parse_events(&mut self) -> Vec { + // 超限裁剪收尾:被整体丢弃的残缺事件,其后续字节以首个 boundary 收尾, + // 此处静默吞掉不产出,之后的正常事件恢复正常解析 + if self.skip_partial { + match Self::find_event_boundary(&self.buffer) { + Some((pos, len)) => { + self.buffer.drain(..pos + len); + self.skip_partial = false; + } + None => return Vec::new(), + } + } let mut events = Vec::new(); loop { // 同时识别三种分隔符,取最早出现的位置与该分隔符的字节长度 @@ -149,10 +165,23 @@ where Poll::Ready(Some(Ok(chunk))) => { self.buffer.extend_from_slice(&chunk); // 防御:缓冲区不应无限增长。正常情况下 parse_events 会持续消费, - // 仅当上游持续不发分隔符(畸形流)时才触发,此处裁掉头部旧数据避免 OOM。 + // 超限只发生在单个事件体超过 BUF_MAX 的场景(畸形上游不发分隔符, + // 或超大单事件)。按字节盲截头部会把半截事件体留给后续解析,产出 + // 破损 JSON 且无法复原;改为对齐事件边界的保尾裁剪: + // - 已有完整分隔符 → 最旧的一个事件(超限源头)整体丢弃,保留的 + // 数据从干净的事件起点开始,后续事件解析不受损; + // - 全程无分隔符 → 整体丢弃并置 skip_partial,该事件的收尾字节 + // 到达时静默吞掉。 if self.buffer.len() > BUF_MAX { - let drop_n = self.buffer.len() - BUF_MAX; - self.buffer.drain(..drop_n); + match Self::find_event_boundary(&self.buffer) { + Some((pos, len)) => { + self.buffer.drain(..pos + len); + } + None => { + self.buffer.clear(); + self.skip_partial = true; + } + } } continue; } @@ -160,8 +189,9 @@ where return Poll::Ready(Some(Err(format!("SSE 流读取错误: {}", e)))); } Poll::Ready(None) => { - // 流结束,处理 buffer 中的剩余数据(可能没有 \n\n 结束的最后一段) - if !self.buffer.is_empty() { + // 流结束,处理 buffer 中的剩余数据(可能没有 \n\n 结束的最后一段); + // skip_partial 置位时剩余数据是被丢弃残缺事件的尾段,不得溢出为事件 + if !self.buffer.is_empty() && !self.skip_partial { // 解码:优先严格 UTF-8(EOF 无下一 chunk 可拼接, // 残缺多字节尾部只能尽力而为,降级 lossy 不崩) let mut remaining = match String::from_utf8(self.buffer.clone()) { @@ -185,3 +215,58 @@ where } } } + +#[cfg(test)] +mod tests { + use super::*; + use futures::stream; + + /// 构造只含 Ok chunk 的内存流,跑完整个 SseStream,返回拍平的事件序列 + fn run_chunks(chunks: Vec) -> Vec { + use futures::StreamExt; + let items: Vec> = chunks + .into_iter() + .map(|c| Ok(bytes::Bytes::from(c))) + .collect(); + let mut s = SseStream::new(stream::iter(items)); + futures::executor::block_on(async { + let mut out = Vec::new(); + while let Some(item) = s.next().await { + out.extend(item.expect("测试流不应产生错误")); + } + out + }) + } + + /// 回归:正常多事件流照常解析 + #[test] + fn normal_events_parse() { + let events = run_chunks(vec![ + "data: {\"a\":1}\n\n".into(), + "data: {\"b\":2}\n\n".into(), + ]); + assert_eq!(events, vec!["{\"a\":1}".to_string(), "{\"b\":2}".to_string()]); + } + + /// 超限单事件(末尾带完整分隔符)整体丢弃:不产出半截破损事件,后续正常事件照常解析。 + /// 旧按字节截头实现会把残留尾部 "data: TAIL" 行当伪事件产出 + #[test] + fn oversize_event_with_boundary_dropped_whole() { + let giant = format!("data: {}\ndata: TAIL\n\n", "x".repeat(BUF_MAX)); + let events = run_chunks(vec![giant, "data: {\"ok\":1}\n\n".into()]); + assert_eq!(events, vec!["{\"ok\":1}".to_string()]); + } + + /// 验收:超限注入(无边界大块)丢弃后,残缺事件收尾静默吞掉,后续正常事件仍可解析。 + /// 旧按字节截头实现会把 STALE-TAIL 残段当伪事件产出 + #[test] + fn oversize_boundaryless_blob_then_normal_events_still_parse() { + let events = run_chunks(vec![ + "x".repeat(BUF_MAX + 100), + "data: STALE-TAIL\n\n".into(), + "data: {\"ok\":1}\n\n".into(), + "data: [DONE]\n\n".into(), + ]); + assert_eq!(events, vec!["{\"ok\":1}".to_string(), "[DONE]".to_string()]); + } +}