优化: df-ai/df-storage/df-nodes 底层加固: 上下文/模型/节点/仓储

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。
This commit is contained in:
lxy
2026-10-09 22:54:25 +08:00
parent d7e626a813
commit f4ca01f2f2
8 changed files with 296 additions and 36 deletions
+18
View File
@@ -136,6 +136,24 @@
"context_window": 262144, "context_window": 262144,
"temperature_supported": false "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", "model_id": "deepseek-v4-flash",
"enabled": true, "enabled": true,
+26 -22
View File
@@ -1045,9 +1045,12 @@ fn phase1_append_after_dirty_stays_dirty_until_save() {
// ============================================================ // ============================================================
#[test] #[test]
fn eviction_inserts_summary_message_at_view_head() { fn eviction_inserts_summary_message_at_view_tail() {
// 触发淘汰 → 视图头部应有一条 "[早期对话概要]" user 摘要(补偿无声丢弃的早期主题), // 触发淘汰 → 视图尾部应有一条 "[早期对话概要]" user 摘要(补偿无声丢弃的早期主题),
// 概要 ≤ EVICTION_SUMMARY_MAX_CHARS,保护消息完好,且纯视图(不污染内存全量/不落库)。 // 概要 ≤ EVICTION_SUMMARY_MAX_CHARS,保护消息完好,且纯视图(不污染内存全量/不落库)。
// 缓存友好(2026-10):摘要改尾部追加(push)而非头部插入(insert(0))——头部插入位移
// 整条前缀字节,使 DeepSeek 等自动前缀缓存 provider 全部缓存失效(命中率掉到个位数)。
// 尾部追加让稳定前缀(system + 早期 history)逐轮不变,命中率最大化。
let mut mgr = ContextManager::new(cfg(200)); let mut mgr = ContextManager::new(cfg(200));
for i in 0..20 { for i in 0..20 {
if i % 2 == 0 { if i % 2 == 0 {
@@ -1063,46 +1066,47 @@ fn eviction_inserts_summary_message_at_view_head() {
let (msgs, trimmed) = mgr.build_for_request(0); let (msgs, trimmed) = mgr.build_for_request(0);
assert!(trimmed, "超预算应触发裁剪"); assert!(trimmed, "超预算应触发裁剪");
// 视图头部 = 补偿摘要(role=user,"[早期对话概要]" 前缀 + 高频主题) // 视图尾部 = 补偿摘要(role=user,"[早期对话概要]" 前缀 + 高频主题)
let head = msgs.first().expect("视图不应为空"); let tail = msgs.last().expect("视图不应为空");
assert!( assert!(
matches!(head.role, MessageRole::User), matches!(tail.role, MessageRole::User),
"视图头部应为 user 摘要, 实际 {:?}", "视图尾部应为 user 摘要, 实际 {:?}",
head.role tail.role
); );
assert!( assert!(
head.content.starts_with("[早期对话概要] 高频主题: "), tail.content.starts_with("[早期对话概要] 高频主题: "),
"摘要格式不符, 实际 {}", "摘要格式不符, 实际 {}",
head.content tail.content
); );
assert!( assert!(
head.content.contains("审批"), tail.content.contains("审批"),
"摘要应含被淘汰区高频主题词'审批', 实际 {}", "摘要应含被淘汰区高频主题词'审批', 实际 {}",
head.content tail.content
); );
assert!( assert!(
head.content.chars().count() <= EVICTION_SUMMARY_MAX_CHARS, tail.content.chars().count() <= EVICTION_SUMMARY_MAX_CHARS,
"摘要应 ≤ {} 字符, 实际 {}", "摘要应 ≤ {} 字符, 实际 {}",
EVICTION_SUMMARY_MAX_CHARS, EVICTION_SUMMARY_MAX_CHARS,
head.content.chars().count() tail.content.chars().count()
); );
// 保护消息完好:末条(i=19 assistant)保留;摘要 user 与首轮真实 user 相邻,由 // 保护消息完好:末条(i=19 assistant)保留;摘要 user 追加在尾部,与末尾 assistant
// sanitize 的 ensure_sequence_legal 合并(摘要在前,换行接原文),故 msgs[1] 为 i=1 的 assistant // 不同 role 不合并,故 msgs[倒数第2] 仍为 i=19 的 assistant
assert_eq!( assert_eq!(
msgs.last().unwrap().content, msgs[msgs.len() - 2].content,
"第19条回复内容同样较长用于撑爆上下文预算触发视图裁剪逻辑", "第19条回复内容同样较长用于撑爆上下文预算触发视图裁剪逻辑",
"保护区最新消息被误裁" "保护区最新消息被误裁"
); );
assert!( assert!(
matches!(msgs[1].role, MessageRole::Assistant), matches!(msgs[msgs.len() - 2].role, MessageRole::Assistant),
"摘要与首轮 user 合并后,第二条应为 assistant, 实际 {:?}", "摘要前一条应为 assistant(i=19), 实际 {:?}",
msgs[1].role msgs[msgs.len() - 2].role
); );
// 首条仍是保留区真实首条 user(摘要不再抢占头部,不位移前缀)
assert!( assert!(
msgs[0].content.contains("审批流程与审批门禁需要适配"), matches!(msgs[0].role, MessageRole::User),
"合并不应丢失被保留的首轮真实 user 文本, 实际 {}", "视图头部应为真实首条 user(摘要已移尾部), 实际 {:?}",
msgs[0].content msgs[0].role
); );
// 纯视图:摘要不进内存全量,被淘汰原文存储态不变 // 纯视图:摘要不进内存全量,被淘汰原文存储态不变
+11 -9
View File
@@ -246,16 +246,18 @@ impl ContextManager {
.collect(); .collect();
// 淘汰补偿摘要:视图裁剪无声丢弃被淘汰内容,LLM 从此看不到早期主题,与压缩 // 淘汰补偿摘要:视图裁剪无声丢弃被淘汰内容,LLM 从此看不到早期主题,与压缩
// 路径(有摘要补偿)行为不一致。此处把被淘汰文本提炼成一短段摘要插回发送 // 路径(有摘要补偿)行为不一致。此处把被淘汰文本提炼成一短段摘要插回发送视图。
// 视图最前部(淘汰区头部、保护消息之前)。角色必须 user:role=system 会被 // **缓存友好修正(2026-10)**:原 insert(0) 头部插入会位移整条前缀字节,使
// anthropic_compat 折叠进带 cache_control 的顶层 system 段,而本摘要纯视图 // DeepSeek 等自动前缀缓存 provider 的稳定前缀(顶层 system + 早期 history)
// 每轮即时再生成(淘汰区单调增长致内容逐轮漂移),落 system 即逐轮改写缓存 // 全部缓存失效——命中率掉到个位数。改尾部追加(push):摘要本身每轮即时再生成
// 指纹使全部前缀缓存失效;连续 role 由既有 sanitize/协议层自愈兜底。摘要只进 // (淘汰区单调增长致内容逐轮漂移)本就无法命中缓存,但放尾部不再拖累稳定前缀,
// messages 易变区,严禁拼进 system_prompt/任何稳定段;纯视图语义:不落库、 // 命中率最大化。角色必须 user:role=system 会被 anthropic_compat 折叠进带
// 不改 conv.messages 存储、不标记 compressed,被淘汰原文存储态不变。 // cache_control 的顶层 system 段,落 system 即逐轮改写缓存指纹;user 放尾部
// 极短/空淘汰文本提炼为空 → 跳过(不值得插)。 // 由 sanitize/协议层自愈兜底。摘要只进 messages 易变区,严禁拼进 system_prompt/
// 任何稳定段;纯视图语义:不落库、不改 conv.messages 存储、不标记 compressed,
// 被淘汰原文存储态不变。极短/空淘汰文本提炼为空 → 跳过(不值得插)。
if let Some(text) = build_eviction_summary_text(&evicted) { if let Some(text) = build_eviction_summary_text(&evicted) {
msgs.insert(0, ChatMessage::user(text)); msgs.push(ChatMessage::user(text));
} }
tracing::info!( tracing::info!(
+51
View File
@@ -211,6 +211,10 @@ impl OpenAICompatProvider {
// 模型不支持 temperature(如商汤 sensenova 仅接受 1.0)→ 省略该字段,避免上游 HTTP 400。 // 模型不支持 temperature(如商汤 sensenova 仅接受 1.0)→ 省略该字段,避免上游 HTTP 400。
// 判定走 model_probe(命中 presets 表;未命中默认支持,保持原样透传零回归)。 // 判定走 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() let temperature = if req.temperature.is_some()
&& !crate::model_probe::probe(&model).temperature_supported && !crate::model_probe::probe(&model).temperature_supported
{ {
@@ -1018,6 +1022,53 @@ mod tests {
assert_eq!(msg.content, serde_json::Value::String("hello".into())); 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)---------- // ---------- ensure_leading_user(首条非 user/system → 补 user 占位,OpenAI 对称 Anthropic)----------
/// 首条 assistant → 补 user 占位(对齐 Anthropic)。上游绕过 sanitize 的 /// 首条 assistant → 补 user 占位(对齐 Anthropic)。上游绕过 sanitize 的
+122 -5
View File
@@ -67,6 +67,26 @@ pub async fn advance_task_atomic(
.await? .await?
.ok_or_else(|| df_types::error::Error::NotFound(format!("任务 {} 不存在", id)))?; .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. 状态机校验(三类拒绝,错误类型区分供前端分辨): // 3. 状态机校验(三类拒绝,错误类型区分供前端分辨):
// - 同态(from==to):Validation「相同状态,无需推进」(空操作,非状态机违例) // - 同态(from==to):Validation「相同状态,无需推进」(空操作,非状态机违例)
// - 非法转换(跳态/终态无后继等):InvalidState「非法状态转换 X→Y」 // - 非法转换(跳态/终态无后继等):InvalidState「非法状态转换 X→Y」
@@ -112,11 +132,45 @@ pub async fn advance_task_atomic(
.advance_status_atomic(id, from, target_status, bump) .advance_status_atomic(id, from, target_status, bump)
.await?; .await?;
// 5. CAS 失败:并发已改动 status(或任务被删),拒绝静默成功。 // 5. CAS 失败:affected==0 有四种可能,回查区分,给出精确错误(不再一律误报「并发」)。
updated.ok_or_else(|| df_types::error::Error::InvalidState { //
current: "已变更(并发推进或旁路修改)".to_string(), // CAS SQL 条件:`WHERE id=? AND status=? AND deleted_at IS NULL AND archived=0`。
expected: from.to_string(), // 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 失败"); 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] #[tokio::test]
async fn blocked_round_trip_does_not_bump() { async fn blocked_round_trip_does_not_bump() {
let repo = setup().await; let repo = setup().await;
@@ -1094,4 +1094,43 @@ mod tests {
.expect("snapshot"); .expect("snapshot");
assert_eq!(snapshot, r#"[{"seq":0}]"#, "冲突时应保留首次快照"); 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");
}
} }
+1
View File
@@ -72,6 +72,7 @@ macro_rules! impl_repo {
update => |$u_conn:ident, $u_rec:ident| $update_body:expr update => |$u_conn:ident, $u_rec:ident| $update_body:expr
) => { ) => {
$(#[$meta])* $(#[$meta])*
#[derive(Clone)]
pub struct $name { pub struct $name {
conn: Arc<Mutex<Connection>>, conn: Arc<Mutex<Connection>>,
} }
+28
View File
@@ -249,6 +249,34 @@ impl TaskRepo {
.map_err(storage_err)? .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<bool> {
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`)。 /// 按 id 查未删除任务(`WHERE id AND deleted_at IS NULL`)。
/// ///
/// 专供 [`validate_parent_id`](super::task_validation) 的父任务存在性检查: /// 专供 [`validate_parent_id`](super::task_validation) 的父任务存在性检查: