From f4ca01f2f263a4d8d765eb5a2d777566b7f095d9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=BB=9D=E5=B0=98?= <237809796@qq.com> Date: Fri, 9 Oct 2026 22:54:25 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BC=98=E5=8C=96:=20df-ai/df-storage/df-nodes?= =?UTF-8?q?=20=E5=BA=95=E5=B1=82=E5=8A=A0=E5=9B=BA:=20=E4=B8=8A=E4=B8=8B?= =?UTF-8?q?=E6=96=87/=E6=A8=A1=E5=9E=8B/=E8=8A=82=E7=82=B9/=E4=BB=93?= =?UTF-8?q?=E5=82=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit df-ai: - context/mod.rs + manager_tests.rs: 上下文管理调整 + 测试同步(48 行) - openai_compat.rs: OpenAI 兼容层增强(+51) - presets/models.json: 模型预设 +18 df-storage: - crud/conversation_repo.rs: 会话仓储(loop_state 相关) - crud/task_repo.rs: 任务仓储 - crud/mod.rs: 导出调整 df-nodes: - task_advance_node.rs: 任务推进节点 验收: cargo check -p devflow EXIT 0。 --- crates/df-ai/presets/models.json | 18 +++ crates/df-ai/src/context/manager_tests.rs | 48 ++++--- crates/df-ai/src/context/mod.rs | 20 +-- crates/df-ai/src/openai_compat.rs | 51 +++++++ crates/df-nodes/src/task_advance_node.rs | 127 +++++++++++++++++- .../df-storage/src/crud/conversation_repo.rs | 39 ++++++ crates/df-storage/src/crud/mod.rs | 1 + crates/df-storage/src/crud/task_repo.rs | 28 ++++ 8 files changed, 296 insertions(+), 36 deletions(-) diff --git a/crates/df-ai/presets/models.json b/crates/df-ai/presets/models.json index ef308e0..1a9b3e2 100644 --- a/crates/df-ai/presets/models.json +++ b/crates/df-ai/presets/models.json @@ -136,6 +136,24 @@ "context_window": 262144, "temperature_supported": false }, + { + "model_id": "sensenova-6.8-flash-lite", + "enabled": true, + "modalities": ["text", "vision"], + "capabilities": ["tool_use"], + "weight": 85, + "context_window": 262144, + "temperature_supported": false + }, + { + "model_id": "sn2-kimi", + "enabled": true, + "modalities": ["text"], + "capabilities": ["tool_use"], + "weight": 80, + "context_window": 131072, + "temperature_supported": false + }, { "model_id": "deepseek-v4-flash", "enabled": true, diff --git a/crates/df-ai/src/context/manager_tests.rs b/crates/df-ai/src/context/manager_tests.rs index 4ba62fe..dd5848b 100644 --- a/crates/df-ai/src/context/manager_tests.rs +++ b/crates/df-ai/src/context/manager_tests.rs @@ -1045,9 +1045,12 @@ fn phase1_append_after_dirty_stays_dirty_until_save() { // ============================================================ #[test] -fn eviction_inserts_summary_message_at_view_head() { - // 触发淘汰 → 视图头部应有一条 "[早期对话概要]" user 摘要(补偿无声丢弃的早期主题), +fn eviction_inserts_summary_message_at_view_tail() { + // 触发淘汰 → 视图尾部应有一条 "[早期对话概要]" user 摘要(补偿无声丢弃的早期主题), // 概要 ≤ EVICTION_SUMMARY_MAX_CHARS,保护消息完好,且纯视图(不污染内存全量/不落库)。 + // 缓存友好(2026-10):摘要改尾部追加(push)而非头部插入(insert(0))——头部插入位移 + // 整条前缀字节,使 DeepSeek 等自动前缀缓存 provider 全部缓存失效(命中率掉到个位数)。 + // 尾部追加让稳定前缀(system + 早期 history)逐轮不变,命中率最大化。 let mut mgr = ContextManager::new(cfg(200)); for i in 0..20 { if i % 2 == 0 { @@ -1063,46 +1066,47 @@ fn eviction_inserts_summary_message_at_view_head() { let (msgs, trimmed) = mgr.build_for_request(0); assert!(trimmed, "超预算应触发裁剪"); - // 视图头部 = 补偿摘要(role=user,"[早期对话概要]" 前缀 + 高频主题) - let head = msgs.first().expect("视图不应为空"); + // 视图尾部 = 补偿摘要(role=user,"[早期对话概要]" 前缀 + 高频主题) + let tail = msgs.last().expect("视图不应为空"); assert!( - matches!(head.role, MessageRole::User), - "视图头部应为 user 摘要, 实际 {:?}", - head.role + matches!(tail.role, MessageRole::User), + "视图尾部应为 user 摘要, 实际 {:?}", + tail.role ); assert!( - head.content.starts_with("[早期对话概要] 高频主题: "), + tail.content.starts_with("[早期对话概要] 高频主题: "), "摘要格式不符, 实际 {}", - head.content + tail.content ); assert!( - head.content.contains("审批"), + tail.content.contains("审批"), "摘要应含被淘汰区高频主题词'审批', 实际 {}", - head.content + tail.content ); assert!( - head.content.chars().count() <= EVICTION_SUMMARY_MAX_CHARS, + tail.content.chars().count() <= EVICTION_SUMMARY_MAX_CHARS, "摘要应 ≤ {} 字符, 实际 {}", EVICTION_SUMMARY_MAX_CHARS, - head.content.chars().count() + tail.content.chars().count() ); - // 保护消息完好:末条(i=19 assistant)保留;摘要 user 与首轮真实 user 相邻,由 - // sanitize 的 ensure_sequence_legal 合并(摘要在前,换行接原文),故 msgs[1] 为 i=1 的 assistant + // 保护消息完好:末条(i=19 assistant)保留;摘要 user 追加在尾部,与末尾 assistant + // 不同 role 不合并,故 msgs[倒数第2] 仍为 i=19 的 assistant assert_eq!( - msgs.last().unwrap().content, + msgs[msgs.len() - 2].content, "第19条回复内容同样较长用于撑爆上下文预算触发视图裁剪逻辑", "保护区最新消息被误裁" ); assert!( - matches!(msgs[1].role, MessageRole::Assistant), - "摘要与首轮 user 合并后,第二条应为 assistant, 实际 {:?}", - msgs[1].role + matches!(msgs[msgs.len() - 2].role, MessageRole::Assistant), + "摘要前一条应为 assistant(i=19), 实际 {:?}", + msgs[msgs.len() - 2].role ); + // 首条仍是保留区真实首条 user(摘要不再抢占头部,不位移前缀) assert!( - msgs[0].content.contains("审批流程与审批门禁需要适配"), - "合并不应丢失被保留的首轮真实 user 文本, 实际 {}", - msgs[0].content + matches!(msgs[0].role, MessageRole::User), + "视图头部应为真实首条 user(摘要已移尾部), 实际 {:?}", + msgs[0].role ); // 纯视图:摘要不进内存全量,被淘汰原文存储态不变 diff --git a/crates/df-ai/src/context/mod.rs b/crates/df-ai/src/context/mod.rs index 7e1bab9..1bb407f 100644 --- a/crates/df-ai/src/context/mod.rs +++ b/crates/df-ai/src/context/mod.rs @@ -246,16 +246,18 @@ impl ContextManager { .collect(); // 淘汰补偿摘要:视图裁剪无声丢弃被淘汰内容,LLM 从此看不到早期主题,与压缩 - // 路径(有摘要补偿)行为不一致。此处把被淘汰文本提炼成一短段摘要插回发送 - // 视图最前部(淘汰区头部、保护消息之前)。角色必须 user:role=system 会被 - // anthropic_compat 折叠进带 cache_control 的顶层 system 段,而本摘要纯视图 - // 每轮即时再生成(淘汰区单调增长致内容逐轮漂移),落 system 即逐轮改写缓存 - // 指纹使全部前缀缓存失效;连续 role 由既有 sanitize/协议层自愈兜底。摘要只进 - // messages 易变区,严禁拼进 system_prompt/任何稳定段;纯视图语义:不落库、 - // 不改 conv.messages 存储、不标记 compressed,被淘汰原文存储态不变。 - // 极短/空淘汰文本提炼为空 → 跳过(不值得插)。 + // 路径(有摘要补偿)行为不一致。此处把被淘汰文本提炼成一短段摘要插回发送视图。 + // **缓存友好修正(2026-10)**:原 insert(0) 头部插入会位移整条前缀字节,使 + // DeepSeek 等自动前缀缓存 provider 的稳定前缀(顶层 system + 早期 history) + // 全部缓存失效——命中率掉到个位数。改尾部追加(push):摘要本身每轮即时再生成 + // (淘汰区单调增长致内容逐轮漂移)本就无法命中缓存,但放尾部不再拖累稳定前缀, + // 命中率最大化。角色必须 user:role=system 会被 anthropic_compat 折叠进带 + // cache_control 的顶层 system 段,落 system 即逐轮改写缓存指纹;user 放尾部 + // 由 sanitize/协议层自愈兜底。摘要只进 messages 易变区,严禁拼进 system_prompt/ + // 任何稳定段;纯视图语义:不落库、不改 conv.messages 存储、不标记 compressed, + // 被淘汰原文存储态不变。极短/空淘汰文本提炼为空 → 跳过(不值得插)。 if let Some(text) = build_eviction_summary_text(&evicted) { - msgs.insert(0, ChatMessage::user(text)); + msgs.push(ChatMessage::user(text)); } tracing::info!( diff --git a/crates/df-ai/src/openai_compat.rs b/crates/df-ai/src/openai_compat.rs index 69d9fac..cb9056d 100644 --- a/crates/df-ai/src/openai_compat.rs +++ b/crates/df-ai/src/openai_compat.rs @@ -211,6 +211,10 @@ impl OpenAICompatProvider { // 模型不支持 temperature(如商汤 sensenova 仅接受 1.0)→ 省略该字段,避免上游 HTTP 400。 // 判定走 model_probe(命中 presets 表;未命中默认支持,保持原样透传零回归)。 + // 商汤网关(token.sensenova.cn)下温度支持**按模型各异**:sensenova-6.x 与 + // sn2-kimi 实测仅接受 1(传其它值 400),部分新模型支持常规温度。 + // 故不做 base_url 一刀切,改由 presets 逐模型精确维护 temperature_supported: + // 受限模型(含实测 sn2-kimi)标 false 省略;支持温度变化的模型标 true 正常透传。 let temperature = if req.temperature.is_some() && !crate::model_probe::probe(&model).temperature_supported { @@ -1018,6 +1022,53 @@ mod tests { assert_eq!(msg.content, serde_json::Value::String("hello".into())); } + /// 温度支持按模型精确判定(presets 表,probe 命中),不按 base_url 一刀切—— + /// 商汤网关下温度支持各异:sn2-kimi 实测仅接受 temperature=1(传 0.7 → 上游 400), + /// presets 已标 temperature_supported=false,任何 provider 下都省略。 + /// 回归 B5 挂机 sn2-kimi 传 0.7 → 400。省略交上游默认,宁默认不 400。 + #[test] + fn openai_temperature_omitted_for_sn2_kimi_via_presets() { + let provider = + OpenAICompatProvider::new("https://token.sensenova.cn", "k", "sensenova-6.8-flash-lite"); + let req = CompletionRequest { + model: "sn2-kimi".into(), // presets 精确标记受限,probe 命中 false + messages: vec![ChatMessage::user("hi")], + temperature: Some(0.7), + max_tokens: None, + stream: true, + tools: None, + tool_choice: None, + reasoning_content: None, + }; + let out = provider.convert_request(req); + assert_eq!(out.temperature, None, "sn2-kimi presets 标 false,应省略 temperature"); + } + + /// 反向保证:支持常规温度的模型(presets 未标 false,如 deepseek-v4-flash) + /// 即使挂在商汤 provider 池也按模型判定透传,不因 base_url 误杀——商汤新模型 + /// 支持温度变化,一刀切会错误降级。 + #[test] + fn openai_temperature_passed_for_supported_model_on_sensenova_base() { + let provider = + OpenAICompatProvider::new("https://token.sensenova.cn", "k", "sensenova-6.8-flash-lite"); + let req = CompletionRequest { + model: "deepseek-v4-flash".into(), // presets 未标受限 → 默认支持温度 + messages: vec![ChatMessage::user("hi")], + temperature: Some(0.7), + max_tokens: None, + stream: true, + tools: None, + tool_choice: None, + reasoning_content: None, + }; + let out = provider.convert_request(req); + assert_eq!( + out.temperature, + Some(0.7), + "支持温度的模型应透传,不因 base_url 是商汤而省略" + ); + } + // ---------- ensure_leading_user（首条非 user/system → 补 user 占位,OpenAI 对称 Anthropic）---------- /// 首条 assistant → 补 user 占位(对齐 Anthropic)。上游绕过 sanitize 的 diff --git a/crates/df-nodes/src/task_advance_node.rs b/crates/df-nodes/src/task_advance_node.rs index 2c2d06d..c83c093 100644 --- a/crates/df-nodes/src/task_advance_node.rs +++ b/crates/df-nodes/src/task_advance_node.rs @@ -67,6 +67,26 @@ pub async fn advance_task_atomic( .await? .ok_or_else(|| df_types::error::Error::NotFound(format!("任务 {} 不存在", id)))?; + // 2.5 前置校验:归档任务冻结,不可推进(与 CAS SQL 的 archived=0 条件同源)。 + // + // 背景:advance_status_atomic 的 CAS SQL 带 `AND archived = 0`,归档任务必然 + // affected==0 → 返回 None → 下游误报「已变更(并发推进或旁路修改)」,诊断信息 + // 完全误导(真实原因是归档,非并发)。此处前置拦截,给出精确可操作错误。 + // + // 软删(deleted_at IS NOT NULL)任务 get_by_id 仍能读到(TASK_COLS 不 SELECT + // deleted_at,TaskRecord 无该字段),但 CAS 的 `deleted_at IS NULL` 同样必败。 + // 因 TaskRecord 不带 deleted_at,此处无法直接判定;由 advance_status_atomic + // 的 CAS 失败回查区分(见该函数注释)。 + if current.archived { + return Err(df_types::error::Error::InvalidState { + current: format!( + "任务 {} 已归档(archived=true),归档任务冻结不可推进;如需推进请先取消归档", + id + ), + expected: target_status.to_string(), + }); + } + // 3. 状态机校验(三类拒绝,错误类型区分供前端分辨): // - 同态(from==to):Validation「相同状态,无需推进」(空操作,非状态机违例) // - 非法转换(跳态/终态无后继等):InvalidState「非法状态转换 X→Y」 @@ -112,11 +132,45 @@ pub async fn advance_task_atomic( .advance_status_atomic(id, from, target_status, bump) .await?; - // 5. CAS 失败:并发已改动 status(或任务被删),拒绝静默成功。 - updated.ok_or_else(|| df_types::error::Error::InvalidState { - current: "已变更(并发推进或旁路修改)".to_string(), - expected: from.to_string(), - }) + // 5. CAS 失败:affected==0 有四种可能,回查区分,给出精确错误(不再一律误报「并发」)。 + // + // CAS SQL 条件:`WHERE id=? AND status=? AND deleted_at IS NULL AND archived=0`。 + // affected==0 时逐一排查,避免「已变更(并发推进或旁路修改)」这个误导性文案: + // a) deleted_at 非 NULL → 任务在回收站(回查 is_deleted) + // b) archived=1 → 已归档(前置 2.5 已拦,此处兜底防 TOCTOU) + // c) status 真被并发改动 → 回读当前 status 对比 from + // d) 任务被删(id 不存在)→ get_by_id 返 None + let Some(updated) = updated else { + // a) 回收站任务(最易被误报为「并发」的一类,优先排查) + if repo.is_deleted(id).await? { + return Err(df_types::error::Error::InvalidState { + current: format!( + "任务 {} 在回收站(deleted_at 非 NULL),回收站任务不可推进;如需推进请先恢复", + id + ), + expected: from.to_string(), + }); + } + // b/c/d) 回读最新记录,区分归档 / 真并发 / 已删除 + let latest = repo.get_by_id(id).await?; + return Err(match latest { + None => df_types::error::Error::NotFound(format!("任务 {} 不存在", id)), + Some(rec) if rec.archived => df_types::error::Error::InvalidState { + current: format!("任务 {} 已归档(archived=true),不可推进;如需推进请先取消归档", id), + expected: from.to_string(), + }, + Some(rec) => df_types::error::Error::InvalidState { + current: format!( + "状态已变更(并发推进或旁路修改):期望 {:?},实际 {:?}", + from, + rec.status.as_str() + ), + expected: from.to_string(), + }, + }); + }; + + Ok(updated) } // ============================================================ @@ -646,6 +700,69 @@ mod tests { assert!(r.is_none(), "expected 与库内不符应 CAS 失败"); } + // ---------- 归档/软删任务的精确错误(治「已变更(并发推进或旁路修改)」误导) ---------- + + #[tokio::test] + async fn archived_task_rejected_with_precise_error() { + // 归档任务(archived=true)推进应被前置校验拦截,给出「已归档」精确错误, + // 而非落到 CAS 失败报「已变更(并发推进或旁路修改)」误导文案。 + let repo = setup().await; + let mut r = rec("t1", TaskStatus::Todo); + r.archived = true; + repo.insert(r).await.unwrap(); + let err = advance_task_atomic(&repo, "t1", "in_progress").await.unwrap_err(); + match err { + df_types::error::Error::InvalidState { current, .. } => { + assert!( + current.contains("已归档"), + "归档任务错误应含「已归档」,实际: {current}" + ); + assert!( + !current.contains("并发推进或旁路修改"), + "归档任务不应误报并发,实际: {current}" + ); + } + other => panic!("归档任务拒绝应是 InvalidState,实际: {other:?}"), + } + } + + #[tokio::test] + async fn soft_deleted_task_rejected_with_precise_error() { + // 回收站任务(deleted_at 非 NULL)推进:get_by_id 仍能读到(不过滤软删), + // 状态机校验通过,但 CAS SQL 带 deleted_at IS NULL 必败 → 回查 is_deleted + // 区分,给出「在回收站」精确错误,而非误报「已变更(并发推进或旁路修改)」。 + let repo = setup().await; + repo.insert(rec("t1", TaskStatus::Todo)).await.unwrap(); + // 软删该任务(走 soft_delete,设 deleted_at) + let ok = repo.soft_delete("t1").await.unwrap(); + assert!(ok, "soft_delete 应命中"); + assert!(repo.is_deleted("t1").await.unwrap(), "软删后 is_deleted 应为 true"); + let err = advance_task_atomic(&repo, "t1", "in_progress").await.unwrap_err(); + match err { + df_types::error::Error::InvalidState { current, .. } => { + assert!( + current.contains("回收站"), + "软删任务错误应含「回收站」,实际: {current}" + ); + assert!( + !current.contains("并发推进或旁路修改"), + "软删任务不应误报并发,实际: {current}" + ); + } + other => panic!("软删任务拒绝应是 InvalidState,实际: {other:?}"), + } + } + + #[tokio::test] + async fn is_deleted_false_for_active_task() { + // 活跃任务 is_deleted 应为 false(不误判),保证回查区分逻辑不误伤正常任务。 + let repo = setup().await; + repo.insert(rec("t1", TaskStatus::Todo)).await.unwrap(); + assert!(!repo.is_deleted("t1").await.unwrap(), "活跃任务 is_deleted 应为 false"); + // 不存在的任务:is_deleted 返回 false(不存在 ≠ 软删) + assert!(!repo.is_deleted("nope").await.unwrap(), "不存在任务 is_deleted 应为 false"); + } + #[tokio::test] async fn blocked_round_trip_does_not_bump() { let repo = setup().await; diff --git a/crates/df-storage/src/crud/conversation_repo.rs b/crates/df-storage/src/crud/conversation_repo.rs index 4582c7b..399cfde 100644 --- a/crates/df-storage/src/crud/conversation_repo.rs +++ b/crates/df-storage/src/crud/conversation_repo.rs @@ -1094,4 +1094,43 @@ mod tests { .expect("snapshot"); assert_eq!(snapshot, r#"[{"seq":0}]"#, "冲突时应保留首次快照"); } + + /// #904 契约:set_loop_state 写 "parked" 后 get_by_id 可读回(停靠出口写侧单点)。 + /// 锁死「停靠状态写入收敛进 emit_task_parked」依赖的 repo 能力——若 set_loop_state + /// 或 from_row 的 loop_state 映射回归,停靠卡「继续」将再次撞「AI 未在暂停态」。 + #[tokio::test] + async fn set_loop_state_parked_roundtrip() { + let db = Database::open_in_memory().await.expect("open_in_memory"); + let repo = AiConversationRepo::new(&db); + repo.insert(AiConversationRecord { + id: "c1".into(), + title: None, + messages: "[]".into(), + provider_id: None, + model: None, + models: None, + archived: false, + pinned: false, + prompt_tokens: Some(0), + completion_tokens: Some(0), + pinned_goals: None, + pending_approvals: None, + created_at: "0".into(), + updated_at: "0".into(), + origin: None, + boundary_json: None, + loop_state: None, + }) + .await + .expect("insert"); + + assert!(repo.set_loop_state("c1", Some("parked")).await.expect("set")); + let got = repo.get_by_id("c1").await.expect("get").expect("row"); + assert_eq!(got.loop_state.as_deref(), Some("parked"), "写 parked 应可读回"); + + // 清态(running 收敛出口)可读回 None + assert!(repo.set_loop_state("c1", None).await.expect("clear")); + let cleared = repo.get_by_id("c1").await.expect("get").expect("row"); + assert_eq!(cleared.loop_state, None, "清 loop_state 应可读回 None"); + } } diff --git a/crates/df-storage/src/crud/mod.rs b/crates/df-storage/src/crud/mod.rs index a455a85..ae55176 100644 --- a/crates/df-storage/src/crud/mod.rs +++ b/crates/df-storage/src/crud/mod.rs @@ -72,6 +72,7 @@ macro_rules! impl_repo { update => |$u_conn:ident, $u_rec:ident| $update_body:expr ) => { $(#[$meta])* + #[derive(Clone)] pub struct $name { conn: Arc>, } diff --git a/crates/df-storage/src/crud/task_repo.rs b/crates/df-storage/src/crud/task_repo.rs index dca2723..83ff527 100644 --- a/crates/df-storage/src/crud/task_repo.rs +++ b/crates/df-storage/src/crud/task_repo.rs @@ -249,6 +249,34 @@ impl TaskRepo { .map_err(storage_err)? } + /// 判定任务是否在回收站(soft_delete 已设 deleted_at)。 + /// + /// 专供推进链 CAS 失败后的**原因区分**:`advance_status_atomic` 的 CAS SQL 带 + /// `deleted_at IS NULL` 条件,软删任务必然 affected==0 返回 None;调用方据此 + /// 回查本方法,把「已变更(并发推进或旁路修改)」这个误导性错误,精确改写为 + /// 「任务在回收站」,让 LLM/用户知道该先恢复而非重试。 + /// + /// 返回 true = 任务存在且 deleted_at 非 NULL(在回收站);false = 不存在或未删。 + /// 单列轻查询(SELECT 1),不拉全记录,失败回查开销可忽略。 + pub async fn is_deleted(&self, id: &str) -> Result { + let conn = self.conn.clone(); + let id = id.to_owned(); + tokio::task::spawn_blocking(move || { + let guard = conn.blocking_lock(); + let mut stmt = guard + .prepare("SELECT 1 FROM tasks WHERE id = ?1 AND deleted_at IS NOT NULL") + .map_err(storage_err)?; + let found = stmt + .query_row(params![id], |_| Ok(())) + .optional() + .map_err(storage_err)? + .is_some(); + Ok(found) + }) + .await + .map_err(storage_err)? + } + /// 按 id 查未删除任务(`WHERE id AND deleted_at IS NULL`)。 /// /// 专供 [`validate_parent_id`](super::task_validation) 的父任务存在性检查: