diff --git a/crates/df-ai/src/coordinator.rs b/crates/df-ai/src/coordinator.rs index 3fe692a..f7f19f1 100644 --- a/crates/df-ai/src/coordinator.rs +++ b/crates/df-ai/src/coordinator.rs @@ -508,16 +508,52 @@ impl Coordinator { }); } + // 记录最近一次 JoinError 详情:JoinError 不携带 subtask 身份, + // 缺失任务靠循环结束后按 id 对账补齐,错误文本由此带出 + let mut join_err_msg: Option = None; + while let Some(join_result) = join_set.join_next().await { match join_result { Ok(result) => { results_map.insert(result.subtask_id.clone(), result); } Err(e) => { - tracing::error!("[COORDINATOR] 子任务 panic: {}", e); + // panic 与取消分开留痕:panic 是执行体自身崩溃需排查,取消是外部干预 + if e.is_panic() { + tracing::error!("[COORDINATOR] 子任务 panic: {}", e); + } else { + tracing::warn!("[COORDINATOR] 子任务被取消: {}", e); + } + join_err_msg = Some(e.to_string()); } } } + + // panic/取消的子任务不能凭空消失:返回契约是与 plan.tasks 一一对应, + // 缺失会被下游误读为"无此任务";此处按该任务失败语义补失败结果收敛, + // merge 按 success 过滤自然跳过,不扩散中断整层 + for task in layer { + if !results_map.contains_key(&task.id) { + tracing::error!( + subtask = %task.id, + "[COORDINATOR] 子任务执行体崩溃,按失败结果收敛(不扩散中断)" + ); + results_map.insert( + task.id.clone(), + ExecutionResult { + subtask_id: task.id.clone(), + persona_id: task_persona + .get(&task.id) + .cloned() + .unwrap_or_default(), + output: join_err_msg + .clone() + .unwrap_or_else(|| "子任务 panic/被取消".to_string()), + success: false, + }, + ); + } + } } else { tracing::info!( layer = layer_idx, @@ -1415,6 +1451,39 @@ mod tests { assert_eq!(results[1].subtask_id, "b"); } + #[tokio::test] + async fn par_05_panicked_task_converges_to_failed_result() { + // 同层并行中某子任务 panic:JoinSet 返回 JoinError(panic 变体), + // 应收敛为该任务的失败结果,而不是让它从结果列表凭空消失 + let coord = make_coord(); + let plan = Plan::from_tasks(vec![ + SubTask::new("t1", "task1"), + SubTask::new("t2", "task2"), + SubTask::new("t3", "task3"), + ]); + + let results = coord + .dispatch_with_budget(&plan, |task, persona_id: String| async move { + if task.id == "t2" { + panic!("模拟子任务执行体崩溃"); + } + ExecutionResult { + subtask_id: task.id, + persona_id, + output: format!("done {}", task.intent), + success: true, + } + }, None) + .await; + + assert_eq!(results.len(), 3, "panic 子任务也应占一个结果位"); + let t2 = results.iter().find(|r| r.subtask_id == "t2").expect("t2 应有失败结果"); + assert!(!t2.success, "panic 子任务应收敛为失败而非成功"); + assert!(!t2.output.is_empty(), "失败结果应带 JoinError 详情"); + assert!(results.iter().filter(|r| r.subtask_id != "t2").all(|r| r.success), + "其余子任务不受 panic 扩散影响"); + } + // -- Plan-driven LLM 规划开关 -- #[test] diff --git a/crates/df-ai/src/lib.rs b/crates/df-ai/src/lib.rs index 5a93a95..a11ad00 100644 --- a/crates/df-ai/src/lib.rs +++ b/crates/df-ai/src/lib.rs @@ -134,6 +134,10 @@ pub(crate) fn send_error_class(p: SendErrPreds) -> &'static str { /// 其余(openai / glm / deepseek 等 OpenAI 兼容)走 OpenAICompatProvider。 /// 调用方(AI Chat 的 run_agentic_loop、df-nodes 的 AiNode)统一引用此工厂, /// 新增 provider 只改这一处。 +/// +/// 未知 provider_type 无法走显式报错(工厂签名不动,调用方横跨 src-tauri/df-nodes), +/// 兜底 OpenAI 兼容协议但 warn 留痕,不再静默——协议错配的请求失败可溯源到配置项。 +/// anthropic_family 口径与 prompt.rs cache_split / chat.rs needs_url_prefetch 对齐。 pub fn build_provider( provider_type: &str, base_url: &str, @@ -141,11 +145,22 @@ pub fn build_provider( model: &str, ) -> Box { match provider_type { - "anthropic" => Box::new(anthropic_compat::AnthropicCompatProvider::new( - base_url, api_key, model, - )), - _ => Box::new(openai_compat::OpenAICompatProvider::new( - base_url, api_key, model, - )), + "anthropic" | "anthropic_compat" => Box::new( + anthropic_compat::AnthropicCompatProvider::new(base_url, api_key, model), + ), + "openai_compat" | "openai" | "glm" | "deepseek" => { + Box::new(openai_compat::OpenAICompatProvider::new( + base_url, api_key, model, + )) + } + unknown => { + tracing::warn!( + "未知 provider_type {:?},按 OpenAI 兼容协议兜底;请检查 ai_providers 配置的 provider_type", + unknown + ); + Box::new(openai_compat::OpenAICompatProvider::new( + base_url, api_key, model, + )) + } } } diff --git a/crates/df-ai/src/model_probe.rs b/crates/df-ai/src/model_probe.rs index 9383538..92722cc 100644 --- a/crates/df-ai/src/model_probe.rs +++ b/crates/df-ai/src/model_probe.rs @@ -333,6 +333,36 @@ mod tests { assert_eq!(m.cost_tier, CostTier::Medium); } + // ── 启发式:版本号后缀不误判 Vision ── + + #[test] + fn heuristic_version_suffix_no_vision() { + // "-v" 后接数字属版本号(deepseek-v3 为纯文本模型),不得误判 Vision + for name in ["deepseek-v3", "acme-chat-v2", "acme-v12"] { + let m = probe(name); + assert_eq!(m.probe_source, Some(ProbeSource::Heuristic)); + assert!( + !m.modalities.contains(&Modality::Vision), + "{name} 版本号后缀不应误判 Vision: {:?}", + m.modalities + ); + } + } + + #[test] + fn heuristic_v_morpheme_tail_still_vision() { + // v 词素尾仍判视觉:独立 v 段(qwen-v-plus 形态)与段尾 v(glm-4v 形态,假名避开预设表) + for name in ["acme-chat-v", "acme-4v-plus"] { + let m = probe(name); + assert_eq!(m.probe_source, Some(ProbeSource::Heuristic)); + assert!( + m.modalities.contains(&Modality::Vision), + "{name} 段尾 v 词素应判 Vision: {:?}", + m.modalities + ); + } + } + // ── 启发式:默认(Standard/Medium/Text/ToolUse) ── #[test] diff --git a/crates/df-ai/src/model_probe_helpers.rs b/crates/df-ai/src/model_probe_helpers.rs index 87ebd0b..ab38ee0 100644 --- a/crates/df-ai/src/model_probe_helpers.rs +++ b/crates/df-ai/src/model_probe_helpers.rs @@ -112,8 +112,11 @@ pub(super) fn heuristic_infer(model_id: &str) -> ModelConfig { // —— 词素判定助手(命名约定式启发式,集中维护便于增删) —— -/// Vision 词素:`-v`(尾缀如 glm-4v)、以 `v` 结尾、含 `vl` / `vision`。 -/// 不直接 `contains('v')`(避免误伤 review/verbal 等)。 +/// Vision 词素:`vision` / `vl` 子串、`v` 词素尾(任一分隔段以 v 结尾:glm-4v / qwen-v-plus)。 +/// 裸 "v" 必须是词素尾:`contains("-v")` 会把版本号误判成视觉(deepseek-v3 纯文本, +/// "-v" 后接数字属版本号而非视觉词素),按段尾判定后 v3/v2 不命中、4v/v 段仍命中。 pub(super) fn has_vision_token(name: &str) -> bool { - name.contains("vision") || name.contains("vl") || name.ends_with('v') || name.contains("-v") + name.contains("vision") + || name.contains("vl") + || name.split(['-', '.', '_']).any(|seg| seg.ends_with('v')) } diff --git a/crates/df-ai/src/namespace_store.rs b/crates/df-ai/src/namespace_store.rs index f314c04..a132ed6 100644 --- a/crates/df-ai/src/namespace_store.rs +++ b/crates/df-ai/src/namespace_store.rs @@ -81,6 +81,10 @@ impl NamespaceStore { } /// 存入 namespace。返回引用路径字符串。 + /// + /// 超限行为分两档:常规超限 LRU 淘汰旧条目腾位(既有语义);单条本身超过 + /// max_bytes 时淘汰清空也放不下,拒绝入库只返回引用路径——展开点对 + /// read_only()=None 统一落 EVICTED_PLACEHOLDER,与 LRU 淘汰后的不可解析引用同语义。 pub fn store(&mut self, tool_name: &str, content: &str) -> String { let key = simple_hash(content); let path = format!("{}{}/{}", NAMESPACE_REF_PREFIX, tool_name, key); @@ -93,8 +97,15 @@ impl NamespaceStore { return path; } - // 检查容量,超限淘汰 let new_bytes = content.len(); + + // 单条超过总容量:硬入库会把 current_bytes 永久顶破上限(evict 清空后仍无条件 + // insert 的旧路径),拒绝入库保住 current_bytes <= max_bytes 不变式 + if new_bytes > self.max_bytes { + return path; + } + + // 检查容量,超限淘汰 self.evict_if_needed(new_bytes); let entry = NamespaceEntry { @@ -268,6 +279,23 @@ mod tests { } } + #[test] + fn store_oversized_single_entry_rejected_cap_intact() { + // 回归:单条超过 max_bytes 时旧路径淘汰清空后仍无条件入库,current_bytes + // 永久突破上限。现拒绝入库,仅返回引用路径,展开点落 EVICTED_PLACEHOLDER。 + let mut ns = NamespaceStore::new(100); + let path = ns.store("t1", &"x".repeat(200)); + assert_eq!(ns.len(), 0, "超容量单条应拒绝入库"); + assert_eq!(ns.current_bytes, 0, "字节账目不得突破上限"); + assert!(is_namespace_ref(&path), "返回值仍是引用路径"); + assert_eq!(ns.read_only(&path), None, "不可解析 → 既有淘汰占位语义"); + + // 拒绝后不污染状态,常规条目仍可正常入 + let p2 = ns.store("t2", "ok"); + assert_eq!(ns.read(&p2).map(str::to_owned), Some("ok".to_string())); + assert_eq!(ns.current_bytes, 2); + } + #[test] fn clear_resets_all() { let mut ns = NamespaceStore::new(100_000); diff --git a/crates/df-ai/src/openai_compat.rs b/crates/df-ai/src/openai_compat.rs index bf495d1..e723065 100644 --- a/crates/df-ai/src/openai_compat.rs +++ b/crates/df-ai/src/openai_compat.rs @@ -310,7 +310,15 @@ impl OpenAICompatProvider { .into_iter() .enumerate() .map(|(i, c)| { + let missing_id = c.id.is_empty(); let id = tool_call_id_or_fallback(&c.id, i, "gen_tool"); + if missing_id { + tracing::warn!( + function = %c.function.name, + fallback_id = %id, + "[OpenAI] 同步响应 tool_call 缺/空 id,已生成 fallback id(防结果路由串号)" + ); + } ToolCall::new(id, c.function.name, c.function.arguments) }) .collect() @@ -1293,6 +1301,26 @@ mod tests { assert_eq!(ids.len(), unique.len(), "id 应全部唯一,实际 {:?}", ids); } + /// 同步响应 tool_call 缺 id/arguments 字段:serde default 兜底(空串/空串), + /// 缺 id 经 parse_tool_calls 生成 fallback。旧行为:id 必填,单条缺字段令整个 + /// 响应 serde 失败 → complete 报"响应解析失败"整批丢弃。 + #[test] + fn openai_sync_tool_call_missing_fields_default_instead_of_failing() { + let calls: Vec = serde_json::from_str( + r#"[{"type":"function","function":{"name":"list_dir"}}]"#, + ) + .expect("缺 id/arguments 的 tool_call 应可反序列化,不再整响应解析失败"); + let parsed = OpenAICompatProvider::parse_tool_calls(calls); + assert_eq!(parsed.len(), 1); + assert!( + parsed[0].id.starts_with("gen_tool_"), + "缺 id → fallback id, got: {}", + parsed[0].id + ); + assert_eq!(parsed[0].function.name, "list_dir"); + assert_eq!(parsed[0].function.arguments, "", "缺 arguments 缺省空串"); + } + /// CR-空 id 流式:SSE chunk 携带 `"id":""`(SenseNova 兼容缺陷)→ ToolCallDelta.id /// 转为 `gen_stream_{n}` fallback(n 取自全局计数器,跨轮跨 assistant 递增,非 None), /// 保证下游 accumulate_tool_calls 写入 draft.id 非空。chunk 完全无 id 字段(None)保持 diff --git a/crates/df-ai/src/openai_helpers.rs b/crates/df-ai/src/openai_helpers.rs index c3a5529..b77a19b 100644 --- a/crates/df-ai/src/openai_helpers.rs +++ b/crates/df-ai/src/openai_helpers.rs @@ -102,10 +102,14 @@ pub(crate) struct OpenAiMessageResp { #[derive(Debug, Deserialize)] pub(crate) struct OpenAiToolCallResp { + /// 缺省容错:个别兼容端点同步响应省略 id → 缺省空串,由 parse_tool_calls 的 + /// tool_call_id_or_fallback 生成唯一 fallback。此前必填,单条 tool_call 缺 id 令 + /// 整个响应 serde 失败 → complete 报"响应解析失败"整批丢弃(流式路径早已容错)。 + #[serde(default)] pub id: String, // 反序列化 #[serde(rename="type")] 字段,保留以对齐 OpenAI 响应结构,标注意图消除 dead_code warning #[allow(dead_code)] - #[serde(rename = "type")] + #[serde(rename = "type", default)] pub call_type: String, pub function: OpenAiFunctionResp, } @@ -113,6 +117,9 @@ pub(crate) struct OpenAiToolCallResp { #[derive(Debug, Deserialize)] pub(crate) struct OpenAiFunctionResp { pub name: String, + /// 缺省容错:arguments 缺失 → 空串(对齐 anthropic 同步 tool_use input 缺失 args="" 口径), + /// 防单条缺字段令整个响应解析失败;name 保持必填(无名 tool_call 无法派发,无安全缺省)。 + #[serde(default)] pub arguments: String, } @@ -130,8 +137,13 @@ pub(crate) struct PromptTokensDetails { /// 嵌套 cached_tokens 回填,消费方一律读归一后的扁平字段。 #[derive(Debug, Deserialize)] pub(crate) struct OpenAiUsageRaw { + /// 三个计数缺省 0:部分兼容网关 usage 只报部分字段(如仅 total_tokens),必填会让 + /// 整帧 serde 失败走空 chunk 分支 — 同帧 content 一并蒸发、usage 归零无痕。 + #[serde(default)] pub prompt_tokens: u32, + #[serde(default)] pub completion_tokens: u32, + #[serde(default)] pub total_tokens: u32, /// DeepSeek 扩展:缓存命中 token(低价,deepseek-chat/reasoner)。无此字段 → default 0。 #[serde(default)] @@ -288,6 +300,14 @@ pub(crate) fn apply_openai_sse(data: &str, usage_accum: &mut Option) } // 提取 usage(带 include_usage 时末段 chunk 携带,覆盖累积) if let Some(u) = chunk.usage { + // 缺省计数的留痕:两个分项均 0 基本可断定 provider 上报不完整(真实完成 + // 必有 completion 计数),非流路径下游拿到零 usage 无从察觉。 + if u.prompt_tokens == 0 && u.completion_tokens == 0 { + tracing::warn!( + total = u.total_tokens, + "[OpenAI] usage 缺 prompt/completion 分项(按 0 兜底),provider 上报不完整" + ); + } tracing::info!( prompt = u.prompt_tokens, completion = u.completion_tokens, @@ -316,6 +336,21 @@ pub(crate) fn apply_openai_sse(data: &str, usage_accum: &mut Option) ); let tool_calls = choice.delta.tool_calls.map(|tcs| { + // 缺 index 的增量缺省 0(单工具调用语义等价);但同帧多 call 且 index + // 冲突时缺省会互相踩(下游按 index 累积 → 参数拼错位),warn 留痕定位 + // provider 兼容缺陷,不做静默改写。 + if tcs.len() > 1 { + let mut idxs: Vec = tcs.iter().map(|tc| tc.index).collect(); + idxs.sort_unstable(); + idxs.dedup(); + if idxs.len() < tcs.len() { + tracing::warn!( + calls = tcs.len(), + unique = idxs.len(), + "[OpenAI] 同帧多 tool_call 增量 index 冲突(缺省 0 互踩),按 index 累积将错位" + ); + } + } tcs.into_iter() .map(|tc| { // CR-空 id:流式 chunk 的 id 可能为 Some("")(SenseNova 兼容缺陷)。 @@ -423,6 +458,35 @@ mod tests { assert_eq!(usage.prompt_cache_miss_tokens, 36); } + /// 部分兼容网关 usage 只报部分字段:缺省 0 入累加器,同帧 content 不再因整帧 + /// serde 失败被吞(旧行为:必填字段缺失 → 空 chunk → 文本蒸发 + usage 归零无痕) + #[test] + fn sse_partial_usage_frame_parses_with_defaults() { + let mut accum = None; + let data = r#"{"choices":[{"delta":{"content":"hi"}}],"usage":{"total_tokens":10}}"#; + let c = apply_openai_sse(data, &mut accum); + assert_eq!(c.delta, "hi", "usage 缺分项不应令整帧解析失败丢 content"); + let u = accum.expect("部分 usage 应写入累加器"); + assert_eq!(u.total_tokens, 10); + assert_eq!(u.prompt_tokens, 0, "缺省分项按 0 兜底"); + assert_eq!(u.completion_tokens, 0); + } + + /// 同帧多 tool_call 增量缺 index:缺省 0 触发 index 冲突路径(仅 warn 留痕,不改写 + /// 缺省策略);解析本身不失败,增量仍逐条带出。 + #[test] + fn sse_tool_calls_missing_index_in_multi_call_frame_still_parse() { + let mut accum = None; + let data = r#"{"choices":[{"delta":{"tool_calls":[ + {"id":"call_a","function":{"name":"ta","arguments":"{"}}, + {"id":"call_b","function":{"name":"tb","arguments":"}"}} + ]}}]}"#; + let c = apply_openai_sse(data, &mut accum); + let tcs = c.tool_calls.expect("缺 index 的多 call 帧应可解析"); + assert_eq!(tcs.len(), 2); + assert!(tcs.iter().all(|tc| tc.index == 0), "缺 index 缺省 0"); + } + /// content_filter 属终态:敏感内容拦截的流部分 provider 不发 [DONE] 帧, /// 必须靠 finish_reason 收敛,否则悬挂;未列入终态的 finish_reason 保持非终态 #[test] diff --git a/crates/df-ai/src/plan_executor.rs b/crates/df-ai/src/plan_executor.rs index 84be5d2..8b2f283 100644 --- a/crates/df-ai/src/plan_executor.rs +++ b/crates/df-ai/src/plan_executor.rs @@ -2,7 +2,8 @@ //! //! 按 Plan::to_layers 输出的层级顺序执行子任务: //! - 层间串行:上层全部完成才进下一层(deps 保证) -//! - 层内并行:同层 SubTask 用 tokio::JoinSet 并发执行 +//! - 层内并行:同层 SubTask 用 tokio::JoinSet 并发执行(设计目标; +//! 当前为串行循环模拟,见 execute 内 TODO,本执行器未接生产链路) //! //! **当前状态**:骨架实现,PLAN_EXECUTION_ENABLED 门控(默认关)。 //! 未接入 agentic loop 主路径,翻 true 启用后仍需与 run_agentic_loop 对接。 @@ -16,7 +17,9 @@ use crate::planner::{Plan, SubTask}; /// Plan 执行总开关。false = 不启用(走单链 ReAct 旧行为)。 /// /// 运行时原子门控,支持热切换(通过 IPC 或前端设置开关)。 -/// 默认关,翻 true 后 process_tool_calls 内以 JoinSet 并行执行同层工具。 +/// 默认关。现状:PlanExecutor::execute 零生产调用点(仅测试),本执行器未接入主链; +/// agentic loop 读本开关走的是 coordinator::dispatch(Phase 1 串行桩),与本文件无关。 +/// JoinSet 真并行留 Phase 3 对接时接通(见 execute 内 TODO)。 /// 与 PLANNING_ENABLED 解耦:plan_hint 独立使能,编排可见性始终可开。 use std::sync::atomic::{AtomicBool, Ordering}; pub static PLAN_EXECUTION_ENABLED: AtomicBool = AtomicBool::new(false); @@ -87,16 +90,16 @@ impl PlanExecutor { layer.len() ); - // 层内并行:JoinSet 收集 - let _join_set: tokio::task::JoinSet = tokio::task::JoinSet::new(); let summary_clone = parent_summary.clone(); for task in layer { let task = Arc::new(task.clone()); let summary = summary_clone.clone(); - // 注意:run_subtask 是 FnMut,不能直接 move 进 JoinSet(多任务并发调用)。 - // Phase 3 对接时,run_subtask 改为 Arc 或 channel 模式。 - // 当前骨架用串行执行模拟(层内不并行),验证分层逻辑正确性。 + // TODO(Phase 3 真并行): 当前为串行循环模拟并行(假并行),仅验证分层逻辑。 + // run_subtask 是 FnMut,不能直接 move 进 JoinSet(多任务并发调用); + // 接真并行的改法:run_subtask 换 Arc 或 channel 模式, + // 本层全部 spawn 进 JoinSet 统一收集后再汇总 parent_summary,替换此串行 await。 + // 现状无用户可见影响:本执行器未接生产链路(execute 零调用点)。 let result = run_subtask(task.clone(), summary).await; let sr = match result { Ok(output) => SubTaskResult {