From 499aa49e4908a6321d400940e9b3c1d97ec9177c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=BB=9D=E5=B0=98?= <237809796@qq.com> Date: Sat, 3 Oct 2026 15:34:55 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BC=98=E5=8C=96:=20=E5=AD=98=E5=82=A8?= =?UTF-8?q?=E5=B1=82=E6=B8=85=E5=80=BA=E4=B8=8E=E5=AD=90=E8=BD=A8=E8=BF=B9?= =?UTF-8?q?=E6=9D=A5=E6=BA=90=E8=BF=81=E7=A7=BB?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - V44 加 ai_conversations.origin 列(子代理轨迹来源标记) - V21 断点续迁改造(逐对话进度判定,崩溃重跑补漏) - 手写事务改 guard.transaction,panic 不残留打开事务 - 软删 CAS 补 deleted_at IS NULL(任务/项目/灵感三处) - 父任务校验拒回收站挂子,删模块先清依赖边 - 列缺失与未知 status 降级补 warn 留痕,limit=0 返空集 --- .../df-storage/src/crud/conversation_repo.rs | 123 ++++-- crates/df-storage/src/crud/idea_repo.rs | 37 +- crates/df-storage/src/crud/message_repo.rs | 34 +- .../src/crud/project_module_repo.rs | 86 ++++- crates/df-storage/src/crud/project_repo.rs | 40 +- .../src/crud/project_service_repo.rs | 41 ++ crates/df-storage/src/crud/task_repo.rs | 64 +++- crates/df-storage/src/crud/task_validation.rs | 95 ++++- crates/df-storage/src/migrations.rs | 357 ++++++++++++++++-- crates/df-storage/src/models.rs | 3 + 10 files changed, 796 insertions(+), 84 deletions(-) diff --git a/crates/df-storage/src/crud/conversation_repo.rs b/crates/df-storage/src/crud/conversation_repo.rs index 5b76c69..5efa8e8 100644 --- a/crates/df-storage/src/crud/conversation_repo.rs +++ b/crates/df-storage/src/crud/conversation_repo.rs @@ -40,8 +40,31 @@ fn ai_provider_from_row(row: &Row<'_>) -> std::result::Result("enabled") { + Ok(v) => v != 0, + Err(e) => { + tracing::warn!("ai_providers[{id}] enabled 列读取失败({e}),降级默认 enabled=true"); + true + } + }; + // weight 读侧 clamp [0,100]:与 insert/update_full 落库的 `.min(100)` 对齐, + // 防老库(clamp 落地前写入的)或外部直改 DB 产生的越界值污染路由权重语义。 + let weight = match row.get::<_, i32>("weight") { + Ok(v) => v.clamp(0, 100) as u32, + Err(e) => { + tracing::warn!("ai_providers[{id}] weight 列读取失败({e}),降级默认 weight=50"); + 50 + } + }; Ok(AiProviderRecord { - id: row.get("id")?, + id, name: row.get("name")?, provider_type: row.get("provider_type")?, api_key: row.get("api_key")?, @@ -53,14 +76,8 @@ fn ai_provider_from_row(row: &Row<'_>) -> std::result::Result("enabled").unwrap_or(1) != 0, - // weight 读侧 clamp [0,100]:与 insert/update_full 落库的 `.min(100)` 对齐, - // 防老库(clamp 落地前写入的)或外部直改 DB 产生的越界值污染路由权重语义。 - weight: row.get::<_, i32>("weight") - .unwrap_or(50) - .clamp(0, 100) as u32, + enabled, + weight, }) } @@ -88,6 +105,9 @@ fn ai_conversation_from_row(row: &Row<'_>) -> std::result::Result |row| ai_conversation_from_row(row), insert => |conn, rec| { conn.execute( - "INSERT INTO ai_conversations (id, title, messages, provider_id, model, models, archived, pinned, prompt_tokens, completion_tokens, pinned_goals, pending_approvals, created_at, updated_at) - VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14)", + "INSERT INTO ai_conversations (id, title, messages, provider_id, model, models, archived, pinned, prompt_tokens, completion_tokens, pinned_goals, pending_approvals, created_at, updated_at, origin) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15)", params![ rec.id, rec.title, rec.messages, rec.provider_id, rec.model, rec.models, rec.archived, if rec.pinned { 1i32 } else { 0i32 }, rec.prompt_tokens, rec.completion_tokens, - rec.pinned_goals, rec.pending_approvals, rec.created_at, rec.updated_at + rec.pinned_goals, rec.pending_approvals, rec.created_at, rec.updated_at, rec.origin ], ) }, @@ -261,33 +281,28 @@ impl AiToolExecutionRepo { return Ok(()); } let conn = self.conn.clone(); - tokio::task::spawn_blocking(move || { - let guard = conn.blocking_lock(); - guard.execute_batch("BEGIN").map_err(storage_err)?; - let result = (|| -> std::result::Result<(), rusqlite::Error> { - let mut stmt = guard.prepare( + tokio::task::spawn_blocking(move || -> Result<()> { + let mut guard = conn.blocking_lock(); + // transaction() guard 范式(同 clear_conversation_atomic / message_repo::insert_batch): + // rusqlite Transaction 在 Drop 未 commit 时自动回滚,中途 panic/早退不残留打开事务; + // 原手写 BEGIN/COMMIT 无此保证,panic 后连接带开事务,后续写全部静默失败。 + let tx = guard.transaction().map_err(storage_err)?; + { + let mut stmt = tx.prepare( "INSERT INTO ai_tool_executions (id, conversation_id, message_id, tool_call_id, tool_name, arguments, result, status, risk_level, requested_at, executed_at, decided_by) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12)", - )?; + ) + .map_err(storage_err)?; for rec in &records { stmt.execute(params![ rec.id, rec.conversation_id, rec.message_id, &*rec.tool_call_id, rec.tool_name, rec.arguments, rec.result, rec.status, rec.risk_level, rec.requested_at, rec.executed_at, rec.decided_by - ])?; - } - Ok(()) - })(); - match result { - Ok(()) => { - guard.execute_batch("COMMIT").map_err(storage_err)?; - } - Err(e) => { - // 回滚失败静默(尽量保一致性;rollback 失败通常是连接已坏,交给上层) - let _ = guard.execute_batch("ROLLBACK"); - return Err(storage_err(e)); + ]) + .map_err(storage_err)?; } } + tx.commit().map_err(storage_err)?; Ok(()) }) .await @@ -606,6 +621,12 @@ impl AiConversationRepo { /// 参数化绑定(替代原调用方的 format! 拼 SQL + execute_batch),防 snapshot 含引号/ /// 特殊字符致注入或损坏。列对齐 conversation_checkpoints(id, conv_id, snapshot, /// token_total, created_at)。 + /// + /// 冲突语义:调用方 ck_id 由 (conv, 消息数) 确定性生成,同轮重复 save 撞同 id 是 + /// 预期幂等跳过;但静默会掩盖另一情形——会话回退/重生成回同消息数且内容已变, + /// 新 snapshot 被旧 checkpoint 挡住未落库。affected=0 时 warn 留痕,两种情形都 + /// 可观测(是否异常由调用方日志上下文判别)。签名保 Result<()> 不外泄跳过信息: + /// 跳过非错误,调用方对 Err 已按 warn 处理,无需改调用点。 pub async fn insert_checkpoint( &self, id: &str, @@ -619,20 +640,25 @@ impl AiConversationRepo { let conv_id = conv_id.to_string(); let snapshot = snapshot.to_string(); let created_at = created_at.to_string(); - let _ = tokio::task::spawn_blocking(move || { + tokio::task::spawn_blocking(move || -> Result<()> { let guard = conn.blocking_lock(); - guard + let affected = guard .execute( "INSERT OR IGNORE INTO conversation_checkpoints \ (id, conv_id, snapshot, token_total, created_at) \ VALUES (?1, ?2, ?3, ?4, ?5)", params![id, conv_id, snapshot, token_total, created_at], ) - .map_err(storage_err) + .map_err(storage_err)?; + if affected == 0 { + tracing::warn!( + "checkpoint 幂等跳过(同 id 已存在,新 snapshot 未落库): id={id} conv={conv_id}" + ); + } + Ok(()) }) .await - .map_err(storage_err)??; - Ok(()) + .map_err(storage_err)? } /// 清空对话消息内容(保留 conversation 记录本身,只清 messages JSON + 清零 token 计数) @@ -943,4 +969,31 @@ mod tests { let got2 = repo.find_by_conv_tool_args("conv1", "write_file", "{\"path\":\"a.go\"}").await.expect("query"); assert_eq!(got2.unwrap().id, "a2", "同三元组应返回最新落定记录"); } + + /// insert_checkpoint 冲突语义:同 id 二次写入 Ok 不报错(幂等跳过,affected=0 走 warn), + /// 保留首次 snapshot。锁死 OR IGNORE 行为,防未来误改成 REPLACE 静默覆盖快照。 + #[tokio::test] + async fn insert_checkpoint_conflict_skips_and_keeps_first() { + let db = Database::open_in_memory().await.expect("open_in_memory"); + let repo = AiConversationRepo::new(&db); + + repo.insert_checkpoint("ck_1", "conv1", r#"[{"seq":0}]"#, 1, "100") + .await + .expect("首次写入"); + // 同 id 不同内容:IGNORE 生效应 Ok,首次 snapshot 保留 + repo.insert_checkpoint("ck_1", "conv1", r#"[{"seq":0},{"seq":1}]"#, 2, "200") + .await + .expect("冲突写入应幂等跳过不报错"); + + let conn = db.conn(); + let g = conn.lock().await; + let count: i64 = g + .query_row("SELECT COUNT(*) FROM conversation_checkpoints WHERE id = 'ck_1'", [], |r| r.get(0)) + .expect("count"); + assert_eq!(count, 1, "同 id 只应有一行"); + let snapshot: String = g + .query_row("SELECT snapshot FROM conversation_checkpoints WHERE id = 'ck_1'", [], |r| r.get(0)) + .expect("snapshot"); + assert_eq!(snapshot, r#"[{"seq":0}]"#, "冲突时应保留首次快照"); + } } diff --git a/crates/df-storage/src/crud/idea_repo.rs b/crates/df-storage/src/crud/idea_repo.rs index 33dd571..dc00b9a 100644 --- a/crates/df-storage/src/crud/idea_repo.rs +++ b/crates/df-storage/src/crud/idea_repo.rs @@ -104,7 +104,16 @@ fn idea_from_row(row: &Row<'_>) -> std::result::Result v, + None => { + // 未知 status 静默降级默认值后,后续 update_full 回写会把库里的未知值 + // 覆盖成默认值(读→改→写放大静默),warn 留痕让脏数据可发现。 + let id: String = row.get("id")?; + tracing::warn!("ideas.id={id} 含未知 status {s:?},按默认值降级(回写会覆盖原值,请核查)"); + Default::default() + } + } }, priority: row.get("priority")?, score: row.get("score")?, @@ -613,7 +622,7 @@ impl IdeaRepo { /// /// 不动 GUI 的无条件 [`update_full`](无 expected 语义,保持现状);本方法仅服务 /// 需要乐观锁版本的调用方(df-mcp update_idea / score_idea)。 - /// `Ok(false)` = id 不存在或版本冲突。 + /// `Ok(false)` = id 不存在、已进回收站(deleted_at IS NULL 守卫,V28 灵感软删)或版本冲突。 pub async fn update_full_cas( &self, record: &IdeaRecord, @@ -626,7 +635,8 @@ impl IdeaRepo { let guard = conn.blocking_lock(); let affected = guard .execute( - "UPDATE ideas SET title = ?1, description = ?2, status = ?3, priority = ?4, score = ?5, tags = ?6, source = ?7, promoted_to = ?8, ai_analysis = ?9, scores = ?10, related_ids = ?11, updated_at = ?12 WHERE id = ?13 AND updated_at = ?14", + // deleted_at IS NULL:回收站灵感不可被 CAS 覆盖(同 tasks/projects 同族守卫) + "UPDATE ideas SET title = ?1, description = ?2, status = ?3, priority = ?4, score = ?5, tags = ?6, source = ?7, promoted_to = ?8, ai_analysis = ?9, scores = ?10, related_ids = ?11, updated_at = ?12 WHERE id = ?13 AND updated_at = ?14 AND deleted_at IS NULL", params![ rec.title, rec.description, rec.status.as_str(), rec.priority, rec.score, rec.tags, rec.source, rec.promoted_to, rec.ai_analysis, @@ -1992,6 +2002,27 @@ mod tests { assert_eq!(after.updated_at, "1800000000000"); } + #[tokio::test] + async fn idea_update_full_cas_skips_soft_deleted() { + // 回收站灵感不可被 CAS 命中(deleted_at IS NULL 守卫,V28 软删) + let repo = setup_idea_repo().await; + repo.insert(irec("i1", "原标题")).await.unwrap(); + repo.soft_delete("i1").await.unwrap(); + // soft_delete 刷新了 updated_at,取当前值作 expected(排除版本不匹配干扰) + let current = repo.get_by_id("i1").await.unwrap().unwrap(); + let mut rec = current.clone(); + rec.title = "回收站里改标题".to_string(); + rec.updated_at = "1900000000000".to_string(); + let ok = repo + .update_full_cas(&rec, ¤t.updated_at) + .await + .unwrap(); + assert!(!ok, "软删灵感 CAS 应拒绝(deleted_at 守卫)"); + let after = repo.get_by_id("i1").await.unwrap().unwrap(); + assert_eq!(after.title, "原标题", "回收站灵感内容不应被改写"); + assert_eq!(after.updated_at, current.updated_at, "updated_at 不应抖动"); + } + #[tokio::test] async fn idea_update_full_cas_conflict_when_expected_stale() { let repo = setup_idea_repo().await; diff --git a/crates/df-storage/src/crud/message_repo.rs b/crates/df-storage/src/crud/message_repo.rs index 158c6ef..cad232b 100644 --- a/crates/df-storage/src/crud/message_repo.rs +++ b/crates/df-storage/src/crud/message_repo.rs @@ -147,6 +147,10 @@ impl AiMessageRepo { /// list_by_conversation 一致的 seq 升序)。`before_seq` 可选:指定后只取 seq < before_seq /// 的消息(滚顶加载更多时的游标,取下一页更早的历史)。 /// + /// `limit = 0` 契约:返空集(SQLite LIMIT 0 合法且恒 0 行),不做旧版的 `.max(1)` + /// 静默钳制——请求 0 条却回 1 条会让调用方误判页内有数据。当前唯一调用方传 + /// 常量 PAGE_SIZE=50 恒 >0,0 只可能来自未来调用方失误,空集是最不易误导的语义。 + /// /// 典型用法: /// - 首次切入对话:list_recent(conv_id, 50, None) → 最近 50 条 /// - 滚顶加载更多:list_recent(conv_id, 50, Some(最早已加载消息的 seq)) → 再加载 50 条更早的 @@ -160,8 +164,9 @@ impl AiMessageRepo { let conv_id = conversation_id.to_owned(); tokio::task::spawn_blocking(move || { let guard = conn.blocking_lock(); - // before_seq 有无分两个 SQL(参数化 LIMIT 必须用固定占位,Rust 侧 clamp 防 0) - let limit = limit.max(1) as i64; + // before_seq 有无分两个 SQL(参数化 LIMIT 必须用固定占位);limit=0 直传 + // LIMIT 0 返空集(契约见方法文档),不再钳成 1 + let limit = limit as i64; let sql = if before_seq.is_some() { "SELECT id, conversation_id, seq, role, content, parts, tool_call_id, tool_calls, model, status, reasoning_content, timestamp, created_at, @@ -681,4 +686,29 @@ mod tests { let got = repo.list_by_conversation("c").await.expect("list"); assert_eq!(got.len(), 1, "重复 replace 不应叠加"); } + + /// limit=0 契约:返空集(直传 SQLite LIMIT 0),不再旧版 max(1) 钳成 1 条。 + /// 锁死语义防回归:请求 0 条却回 1 条会让调用方误判页内有数据。 + #[tokio::test] + async fn list_recent_limit_zero_returns_empty() { + let db = Database::open_in_memory().await.expect("open_in_memory"); + let repo = AiMessageRepo::new(&db); + repo.insert_batch(vec![ + mk_msg("m0", "c", 0, "user", "0"), + mk_msg("m1", "c", 1, "user", "1"), + ]) + .await + .expect("insert"); + + let got = repo.list_recent("c", 0, None).await.expect("list limit=0"); + assert!(got.is_empty(), "limit=0 无游标应返空集"); + + let got2 = repo.list_recent("c", 0, Some(5)).await.expect("list limit=0 带游标"); + assert!(got2.is_empty(), "limit=0 带游标也应返空集"); + + // 邻位 sanity:limit=1 行为不受影响,仍返最近 1 条 + let got3 = repo.list_recent("c", 1, None).await.expect("list limit=1"); + assert_eq!(got3.len(), 1); + assert_eq!(got3[0].seq, 1, "limit=1 应取 seq 最大(最新)一条"); + } } diff --git a/crates/df-storage/src/crud/project_module_repo.rs b/crates/df-storage/src/crud/project_module_repo.rs index 42639a2..20c1de4 100644 --- a/crates/df-storage/src/crud/project_module_repo.rs +++ b/crates/df-storage/src/crud/project_module_repo.rs @@ -169,17 +169,28 @@ impl ProjectModuleRepo { } /// 按 id 硬删工程(工程不需要软删审计追溯)。 + /// + /// 单事务先清 module_dependencies 依赖边(from/to 任一端引用本工程)再删工程: + /// module_dependencies 的 FK 无 ON DELETE 级联且 PRAGMA foreign_keys=ON,带着依赖边 + /// 直接删工程必 FK 违例回滚(调用侧 module.rs/task_graph 各自清边的兜底收口到本层)。 pub async fn delete(&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 affected = guard + let mut guard = conn.blocking_lock(); + let tx = guard.transaction().map_err(storage_err)?; + tx.execute( + "DELETE FROM module_dependencies WHERE from_module_id = ?1 OR to_module_id = ?1", + params![id], + ) + .map_err(storage_err)?; + let affected = tx .execute( "DELETE FROM project_modules WHERE id = ?1", params![id], ) .map_err(storage_err)?; + tx.commit().map_err(storage_err)?; Ok(affected > 0) }) .await @@ -187,17 +198,26 @@ impl ProjectModuleRepo { } /// 按项目删除全部工程(删项目时级联清理,硬删)。 + /// + /// 单事务先清本项目全部依赖边再删工程(顺序同 [`delete`],防 FK 违例)。 pub async fn delete_by_project(&self, project_id: &str) -> Result { let conn = self.conn.clone(); let project_id = project_id.to_owned(); tokio::task::spawn_blocking(move || { - let guard = conn.blocking_lock(); - let affected = guard + let mut guard = conn.blocking_lock(); + let tx = guard.transaction().map_err(storage_err)?; + tx.execute( + "DELETE FROM module_dependencies WHERE project_id = ?1", + params![project_id], + ) + .map_err(storage_err)?; + let affected = tx .execute( "DELETE FROM project_modules WHERE project_id = ?1", params![project_id], ) .map_err(storage_err)?; + tx.commit().map_err(storage_err)?; Ok(affected > 0) }) .await @@ -338,4 +358,62 @@ mod tests { assert!(hit); assert!(repo.list_by_project(&pid).await.unwrap().is_empty()); } + + // ============================================================ + // 删工程先清 module_dependencies 依赖边(⑦走查项:FK 无级联,带边删必违例) + // ============================================================ + + /// 建一条 a→b 依赖边(模块 id 规则见 mrec:mod-a / mod-b)。 + async fn seed_edge(db: &Database, pid: &str) -> crate::crud::ModuleDependencyRepo { + let dep_repo = crate::crud::ModuleDependencyRepo::new(db); + dep_repo + .insert(crate::models::ModuleDependencyRecord { + id: "dep-1".to_string(), + project_id: pid.to_string(), + from_module_id: "mod-a".to_string(), + to_module_id: "mod-b".to_string(), + dep_type: "library".to_string(), + label: None, + created_at: "0".to_string(), + }) + .await + .expect("insert dependency edge"); + dep_repo + } + + #[tokio::test] + async fn delete_clears_dependency_edges_first() { + let (_db, repo, pid) = setup().await; + repo.insert(mrec(&pid, "a", "/p/a")).await.unwrap(); + repo.insert(mrec(&pid, "b", "/p/b")).await.unwrap(); + let dep_repo = seed_edge(&_db, &pid).await; + + // 修前行为对照:带边直接删工程会 FK 违例报错;修后先清边再删,单事务原子 + let hit = repo.delete("mod-a").await.expect("delete 应成功不报 FK"); + assert!(hit); + assert!( + dep_repo.list_by_field("from_module_id", "mod-a").await.unwrap().is_empty() + && dep_repo.list_by_field("to_module_id", "mod-a").await.unwrap().is_empty(), + "引用被删工程的依赖边应一并清除" + ); + assert!(repo.get_by_id("mod-a").await.unwrap().is_none()); + // 另一端工程不受影响 + assert!(repo.get_by_id("mod-b").await.unwrap().is_some()); + } + + #[tokio::test] + async fn delete_by_project_clears_dependency_edges() { + let (_db, repo, pid) = setup().await; + repo.insert(mrec(&pid, "a", "/p/a")).await.unwrap(); + repo.insert(mrec(&pid, "b", "/p/b")).await.unwrap(); + let dep_repo = seed_edge(&_db, &pid).await; + + let hit = repo.delete_by_project(&pid).await.expect("delete_by_project 应成功不报 FK"); + assert!(hit); + assert!( + dep_repo.list_by_field("project_id", &pid).await.unwrap().is_empty(), + "项目级删除应先清全部依赖边" + ); + assert!(repo.list_by_project(&pid).await.unwrap().is_empty()); + } } diff --git a/crates/df-storage/src/crud/project_repo.rs b/crates/df-storage/src/crud/project_repo.rs index ca05f71..3df4f90 100644 --- a/crates/df-storage/src/crud/project_repo.rs +++ b/crates/df-storage/src/crud/project_repo.rs @@ -111,7 +111,16 @@ fn project_from_row(row: &Row<'_>) -> std::result::Result v, + None => { + // 未知 status 静默降级默认值后,后续 update_full 回写会把库里的未知值 + // 覆盖成默认值(读→改→写放大静默),warn 留痕让脏数据可发现。 + let id: String = row.get("id")?; + tracing::warn!("projects.id={id} 含未知 status {s:?},按默认值降级(回写会覆盖原值,请核查)"); + Default::default() + } + } }, idea_id: row.get("idea_id")?, path: row.get("path")?, @@ -436,7 +445,8 @@ impl ProjectRepo { /// 而非静默覆盖(旧 `update_full` 无条件覆盖,存在跨进程互相覆盖竞态)。 /// /// 不动 GUI 的无条件 [`update_full`](无 expected 语义,保持现状);本方法仅服务 - /// 需要乐观锁版本的调用方(df-mcp update_project)。`Ok(false)` = id 不存在或版本冲突。 + /// 需要乐观锁版本的调用方(df-mcp update_project)。`Ok(false)` = id 不存在、已进回收站 + /// (deleted_at IS NULL 守卫)或版本冲突。 pub async fn update_full_cas( &self, record: &ProjectRecord, @@ -449,7 +459,9 @@ impl ProjectRepo { let guard = conn.blocking_lock(); let affected = guard .execute( - "UPDATE projects SET name = ?1, description = ?2, status = ?3, idea_id = ?4, path = ?5, stack = ?6, updated_at = ?7 WHERE id = ?8 AND updated_at = ?9", + // deleted_at IS NULL:回收站项目不可被 CAS 覆盖(MCP 持旧 expected 也能 + // 命中已软删行,造成回收站内容被静默改写),与 advance 类守卫同族。 + "UPDATE projects SET name = ?1, description = ?2, status = ?3, idea_id = ?4, path = ?5, stack = ?6, updated_at = ?7 WHERE id = ?8 AND updated_at = ?9 AND deleted_at IS NULL", params![ rec.name, rec.description, rec.status.as_str(), rec.idea_id, rec.path, rec.stack, rec.updated_at, rec.id, expected @@ -755,4 +767,26 @@ mod tests { assert_eq!(after.name, "他人已改", "CAS 冲突不得覆盖他人修改"); assert_eq!(after.updated_at, "1800000000000"); } + + #[tokio::test] + async fn update_full_cas_skips_soft_deleted() { + // 回收站项目不可被 CAS 命中(deleted_at IS NULL 守卫):即使 expected 与 DB + // updated_at 一致,已软删项目也拒绝写入,防 MCP 对回收站项目静默改写。 + let repo = setup().await; + repo.insert(prec("p1", "原项目")).await.unwrap(); + repo.soft_delete("p1").await.unwrap(); + // soft_delete 刷新了 updated_at,取当前值作 expected(排除版本不匹配干扰) + let current = repo.get_by_id("p1").await.unwrap().unwrap(); + let mut rec = current.clone(); + rec.name = "回收站里改名".to_string(); + rec.updated_at = "1900000000000".to_string(); + let ok = repo + .update_full_cas(&rec, ¤t.updated_at) + .await + .unwrap(); + assert!(!ok, "软删项目 CAS 应拒绝(deleted_at 守卫)"); + let after = repo.get_by_id("p1").await.unwrap().unwrap(); + assert_eq!(after.name, "原项目", "回收站项目内容不应被改写"); + assert_eq!(after.updated_at, current.updated_at, "updated_at 不应抖动"); + } } diff --git a/crates/df-storage/src/crud/project_service_repo.rs b/crates/df-storage/src/crud/project_service_repo.rs index a7352f3..dafbe07 100644 --- a/crates/df-storage/src/crud/project_service_repo.rs +++ b/crates/df-storage/src/crud/project_service_repo.rs @@ -136,6 +136,10 @@ impl_repo!( "project_services", from_row => |row| project_service_from_row(row), insert => |conn, rec| { + // 宏生成的 insert/update_full 是公开写入路径,SQL 体由本调用点整体传入——校验就挂 + // 在体首(等价 insert_validated/update_full_validated 的 validate_record 收口), + // 堵住旁路宏方法把非法 service_type / 含凭证字段写进库的 D10 缺口。 + validate_record(rec)?; conn.execute( "INSERT INTO project_services (id, project_id, name, service_type, endpoint, config_json, environment, remark, created_at, updated_at) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)", @@ -147,6 +151,8 @@ impl_repo!( ) }, update => |conn, rec| { + // 同 insert:宏 update_full 的校验收口在体首(update_full_validated 同源校验)。 + validate_record(rec)?; conn.execute( "UPDATE project_services SET project_id = ?1, name = ?2, service_type = ?3, endpoint = ?4, config_json = ?5, environment = ?6, remark = ?7, updated_at = ?8 WHERE id = ?9", params![ @@ -436,6 +442,41 @@ mod tests { assert_eq!(got.config_json.as_deref(), Some(r#"{"pool_size":10}"#)); } + // ============================================================ + // 宏生成的 insert/update_full 校验收口(⑥走查项:宏写入路径同过 validate_record) + // ============================================================ + + #[tokio::test] + async fn macro_insert_rejects_invalid_service_type() { + // 旁路 insert_validated 直接走宏 insert,非法 service_type 同样被拒 + let (_db, repo) = setup().await; + let err = repo + .insert(srec("s1", "elasticsearch", "development")) + .await; + assert!(err.is_err(), "宏 insert 应过 validate_record 拒非法类型"); + assert!(repo.get_by_id("s1").await.unwrap().is_none(), "拒后不应入库"); + } + + #[tokio::test] + async fn macro_update_full_rejects_sensitive_config() { + // 旁路 update_full_validated 直接走宏 update_full,凭证审查同样生效 + let (_db, repo) = setup().await; + repo.insert_validated(srec("s1", "mysql", "development")) + .await + .unwrap(); + + let mut bad = srec("s1", "mysql", "development"); + bad.endpoint = Some("https://api.example.com?token=abc".to_string()); + let err = repo.update_full(&bad).await; + assert!(err.is_err(), "宏 update_full 应过 D10 凭证审查"); + let got = repo.get_by_id("s1").await.unwrap().unwrap(); + assert_eq!( + got.endpoint.as_deref(), + Some("localhost:3306"), + "被拒后原记录不应被改写" + ); + } + #[tokio::test] async fn update_full_validated_changes_fields() { let (_db, repo) = setup().await; diff --git a/crates/df-storage/src/crud/task_repo.rs b/crates/df-storage/src/crud/task_repo.rs index 187bb44..75782cb 100644 --- a/crates/df-storage/src/crud/task_repo.rs +++ b/crates/df-storage/src/crud/task_repo.rs @@ -27,7 +27,16 @@ fn task_from_row(row: &Row<'_>) -> std::result::Result v, + None => { + // 未知 status 静默降级默认值后,后续 update_full 回写会把库里的未知值 + // 覆盖成默认值(读→改→写放大静默),warn 留痕让脏数据可发现。 + let id: String = row.get("id")?; + tracing::warn!("tasks.id={id} 含未知 status {s:?},按默认值降级(回写会覆盖原值,请核查)"); + Default::default() + } + } }, priority: row.get("priority")?, branch_name: row.get("branch_name")?, @@ -219,6 +228,30 @@ impl TaskRepo { .map_err(storage_err)? } + /// 按 id 查未删除任务(`WHERE id AND deleted_at IS NULL`)。 + /// + /// 专供 [`validate_parent_id`](super::task_validation) 的父任务存在性检查: + /// 回收站父任务(soft_delete 已标 deleted_at)对"挂子任务"视角等同不存在, + /// 通用 get_by_id 不带软删过滤会让子任务挂到回收站父下(⑤走查项)。 + /// 列清单与 get_children 一致(TaskRecord 不带 deleted_at,不 SELECT 该列)。 + pub async fn get_by_id_active(&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 id, project_id, title, description, status, priority, branch_name, assignee, workflow_def_id, base_branch, review_rounds, output_json, idea_id, queue, parent_id, content_json, module_id, created_at, updated_at FROM tasks WHERE id = ?1 AND deleted_at IS NULL") + .map_err(storage_err)?; + let row = stmt + .query_row(params![id], |row| task_from_row(row)) + .optional() + .map_err(storage_err)?; + Ok(row) + }) + .await + .map_err(storage_err)? + } + /// 原子推进任务状态(任务推进链唯一 status 写入路径) /// /// 下沉 SQL `WHERE id=? AND status=?expected` 做 CAS(Compare-And-Swap)防 TOCTOU: @@ -784,7 +817,8 @@ impl TaskRepo { /// 而非静默覆盖。对齐 [`advance_status_atomic`](同款 `WHERE ... = ?expected` 原子条件写)。 /// /// 不动 GUI 的无条件 [`update_full`](无 expected 语义,保持现状);本方法仅服务 - /// 需要乐观锁版本的调用方(df-mcp update_task)。`Ok(false)` = id 不存在或版本冲突。 + /// 需要乐观锁版本的调用方(df-mcp update_task)。`Ok(false)` = id 不存在、已进回收站 + /// (deleted_at IS NULL 守卫)或版本冲突。 pub async fn update_full_cas( &self, record: &TaskRecord, @@ -797,7 +831,9 @@ impl TaskRepo { let guard = conn.blocking_lock(); let affected = guard .execute( - "UPDATE tasks SET project_id = ?1, title = ?2, description = ?3, status = ?4, priority = ?5, branch_name = ?6, assignee = ?7, workflow_def_id = ?8, base_branch = ?9, review_rounds = ?10, output_json = ?11, idea_id = ?12, queue = ?13, parent_id = ?14, content_json = ?15, module_id = ?16, updated_at = ?17 WHERE id = ?18 AND updated_at = ?19", + // deleted_at IS NULL:回收站任务不可被 CAS 覆盖——MCP 拿旧 expected 也能 + // 命中已软删行,造成回收站内容被改且 updated_at 抖动,调用方无从感知。 + "UPDATE tasks SET project_id = ?1, title = ?2, description = ?3, status = ?4, priority = ?5, branch_name = ?6, assignee = ?7, workflow_def_id = ?8, base_branch = ?9, review_rounds = ?10, output_json = ?11, idea_id = ?12, queue = ?13, parent_id = ?14, content_json = ?15, module_id = ?16, updated_at = ?17 WHERE id = ?18 AND updated_at = ?19 AND deleted_at IS NULL", params![ rec.project_id, rec.title, rec.description, rec.status.as_str(), rec.priority, rec.branch_name, rec.assignee, rec.workflow_def_id, rec.base_branch, @@ -1299,6 +1335,28 @@ mod tests { assert_eq!(after.updated_at, "1800000000000"); } + #[tokio::test] + async fn update_full_cas_skips_soft_deleted() { + // 回收站任务不可被 CAS 命中(deleted_at IS NULL 守卫):即使 expected 与 DB + // updated_at 一致,已软删行也拒绝写入,防 MCP 对回收站任务静默改写。 + let repo = setup().await; + repo.insert(trec("t1", "todo", None)).await.unwrap(); + repo.soft_delete("t1").await.unwrap(); + // soft_delete 刷新了 updated_at,取当前值作 expected(排除版本不匹配干扰) + let current = repo.get_by_id("t1").await.unwrap().unwrap(); + let mut rec = current.clone(); + rec.title = "回收站里改标题".to_string(); + rec.updated_at = "1900000000000".to_string(); + let ok = repo + .update_full_cas(&rec, ¤t.updated_at) + .await + .unwrap(); + assert!(!ok, "软删任务 CAS 应拒绝(deleted_at 守卫)"); + let after = repo.get_by_id("t1").await.unwrap().unwrap(); + assert_eq!(after.title, "task-t1", "回收站任务内容不应被改写"); + assert_eq!(after.updated_at, current.updated_at, "updated_at 不应抖动"); + } + // ============================================================ // LW-5 count_by_query 维度对齐 list_by_query(防 count/list 漂移翻页) // ============================================================ diff --git a/crates/df-storage/src/crud/task_validation.rs b/crates/df-storage/src/crud/task_validation.rs index 12371c5..8860596 100644 --- a/crates/df-storage/src/crud/task_validation.rs +++ b/crates/df-storage/src/crud/task_validation.rs @@ -45,7 +45,7 @@ pub fn validate_content_json(content: Option<&str>) -> Result, St } } -/// 校验 parent_id 1 级嵌套铁律:父任务存在 + 父任务自身不是子任务(防孙任务)。 +/// 校验 parent_id 1 级嵌套铁律:父任务存在(未软删) + 父任务自身不是子任务(防孙任务)。 /// 返回归一化 parent_id(空串/缺省 → None)。 pub async fn validate_parent_id( repo: &TaskRepo, @@ -55,11 +55,13 @@ pub async fn validate_parent_id( .map(|s| s.trim().to_string()) .filter(|s| !s.is_empty()); if let Some(pid) = &pid { + // get_by_id_active 带 deleted_at IS NULL:回收站父任务拒挂子,否则子任务 + // 挂在已删父下,父恢复前子任务在树/看板视图两不可见(get_by_id 无软删过滤)。 let parent = repo - .get_by_id(pid) + .get_by_id_active(pid) .await .map_err(|e| e.to_string())? - .ok_or_else(|| format!("父任务 {pid} 不存在,无法创建子任务"))?; + .ok_or_else(|| format!("父任务 {pid} 不存在(或已在回收站),无法创建子任务"))?; if parent.parent_id.is_some() { return Err(format!( "违反 1 级嵌套铁律:父任务 {pid} 自身是子任务(parent_id={:?}),不允许在其下创建孙任务", @@ -80,3 +82,90 @@ pub fn normalize_priority(priority: Option) -> Result { )), } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::crud::ProjectRepo; + use crate::db::Database; + use crate::models::{ProjectRecord, TaskRecord}; + use df_types::types::ProjectStatus; + + /// 内存 DB + 占位 project(tasks.project_id FK 要求)+ TaskRepo。 + async fn setup() -> TaskRepo { + let db = Database::open_in_memory().await.expect("open_in_memory"); + ProjectRepo::new(&db) + .insert(ProjectRecord { + id: "proj-1".to_string(), + name: "proj-1".to_string(), + description: String::new(), + status: ProjectStatus::Planning, + idea_id: None, + path: None, + stack: None, + created_at: "1700000000000".to_string(), + updated_at: "1700000000000".to_string(), + }) + .await + .unwrap(); + TaskRepo::new(&db) + } + + fn trec(id: &str, parent_id: Option<&str>) -> TaskRecord { + TaskRecord { + id: id.to_string(), + project_id: "proj-1".to_string(), + title: format!("task-{id}"), + description: String::new(), + status: df_types::types::TaskStatus::Todo, + priority: 2, + branch_name: None, + assignee: None, + workflow_def_id: None, + base_branch: None, + review_rounds: 0, + output_json: None, + idea_id: None, + queue: "todo".to_string(), + parent_id: parent_id.map(|s| s.to_string()), + content_json: None, + module_id: None, + created_at: "1700000000000".to_string(), + updated_at: "1700000000000".to_string(), + } + } + + #[tokio::test] + async fn parent_in_trash_rejected() { + // 回收站父任务拒挂子(deleted_at IS NULL 存在性检查,⑤走查项) + let repo = setup().await; + repo.insert(trec("parent", None)).await.unwrap(); + repo.soft_delete("parent").await.unwrap(); + let err = validate_parent_id(&repo, Some("parent")).await.unwrap_err(); + assert!(err.contains("回收站"), "软删父应报「不存在(或已在回收站)」: {err}"); + } + + #[tokio::test] + async fn active_parent_accepted_and_grandchild_still_rejected() { + let repo = setup().await; + repo.insert(trec("parent", None)).await.unwrap(); + // 活跃父:校验通过,归一化保留 pid + let pid = validate_parent_id(&repo, Some(" parent ")).await.unwrap(); + assert_eq!(pid.as_deref(), Some("parent"), "trim 后应保留 pid"); + // 1 级嵌套铁律不回归:父是子任务时仍拒孙 + repo.insert(trec("child", Some("parent"))).await.unwrap(); + let err = validate_parent_id(&repo, Some("child")).await.unwrap_err(); + assert!(err.contains("嵌套"), "孙任务应被拒: {err}"); + } + + #[tokio::test] + async fn missing_and_empty_parent() { + let repo = setup().await; + // 不存在的父:报不存在 + let err = validate_parent_id(&repo, Some("ghost")).await.unwrap_err(); + assert!(err.contains("不存在")); + // 空串/缺省:归一化为 None 直接放行 + assert_eq!(validate_parent_id(&repo, None).await.unwrap(), None); + assert_eq!(validate_parent_id(&repo, Some(" ")).await.unwrap(), None); + } +} diff --git a/crates/df-storage/src/migrations.rs b/crates/df-storage/src/migrations.rs index 9b4a6e0..6d796a7 100644 --- a/crates/df-storage/src/migrations.rs +++ b/crates/df-storage/src/migrations.rs @@ -47,7 +47,11 @@ pub fn run(conn: &Connection) -> Result<()> { // 重启后从 DB 恢复 pending_approvals 内存态,使待审批不丢。 // V41 = 任务关联工程模块:tasks.module_id 列(工程系统打底,项目多工程下任务落到具体 module)。 // V42 = ai_messages 加 is_estimated 列(消息级估算标记,reload 逐条回显「估算」标注)。 - let steps: [(i32, fn(&Connection) -> Result<()>); 42] = [ + // V43 = 存量知识一次性晋升(决策单 B4①:candidate + confidence='high' → published, + // 纯数据 UPDATE 不经过 promote 函数,不触发嵌入生成)。 + // V44 = 子代理轨迹来源标记(ai_conversations.origin TEXT,spawn_agent 出口整轨迹落库, + // 桌面侧栏/miniapp 列表按 origin 过滤子会话)。 + let steps: [(i32, fn(&Connection) -> Result<()>); 44] = [ (1, migrate_v1), (2, migrate_v2), (3, migrate_v3), @@ -90,6 +94,8 @@ pub fn run(conn: &Connection) -> Result<()> { (40, migrate_v40), (41, migrate_v41), (42, migrate_v42), + (43, migrate_v43), + (44, migrate_v44), ]; for (version, migrate_fn) in steps { @@ -433,11 +439,13 @@ fn migrate_v20(conn: &Connection) -> Result<()> { /// **一次原子迁移**(决策 V21 合并,不拆 V21a/V21b): /// 1. 建表 ai_messages(IF NOT EXISTS 幂等,新库空表/老库均安全) /// 2. 幂等补 ai_tool_executions.message_id 列(消息级溯源 audit) -/// 3. COUNT 探测 ai_messages 已有数据 → 跳过数据迁移(仅写版本号,防重复迁移) -/// 4. 遍历 ai_conversations.messages JSON → 逐条提取到 ai_messages(分批 commit) +/// 3. 遍历 ai_conversations.messages JSON → 逐条提取到 ai_messages(分批 commit) +/// 4. 逐对话进度判定:已落库数 == JSON 消息数 → 跳过该对话(防重复迁移) /// /// 设计要点(详见消息拆分存储设计 §4.2): -/// - **幂等安全**:COUNT 探测 + INSERT OR IGNORE,中途崩溃重跑跳过已迁移数据 +/// - **幂等安全**:逐对话进度判定 + INSERT OR IGNORE,中途崩溃重跑只补未迁对话。 +/// 不用全局 COUNT 判完成——批量迁移每批一事务,批次中途崩溃后 ai_messages 已有 +/// 数据,COUNT>0 会整段跳过,剩余对话永不再迁而版本号照样补齐,残缺被永久掩盖 /// - **分批 commit**:每 50 对话一批,避免长事务持有 SQLite 写锁 /// - **迁移期 ID**:`msg_migrated_{conv_id}_{seq}` —— 天然唯一(UNIQUE 是 conv_id+seq)、零依赖 /// - **裸 JSON 提取**:用 `serde_json::Value` 而非 ChatMessage(df-storage 不依赖 df-ai-core) @@ -471,20 +479,7 @@ fn migrate_v21(conn: &Connection) -> Result<()> { tracing::info!("v21: 补建 ai_tool_executions.message_id 列(消息级溯源 audit)"); } - // 3. COUNT 探测:ai_messages 已有数据 → 跳过迁移只写版本号(幂等) - // INSERT OR IGNORE 防崩溃重跑(schema_version PK 冲突):run() 正常流程只调 - // 一次 migrate_v21(current_version<21),但崩溃重跑/手动重调时 version=21 - // 可能已存在,IGNORE 保证幂等不报错。 - let existing: i64 = conn.query_row( - "SELECT COUNT(*) FROM ai_messages", [], |row| row.get(0), - )?; - if existing > 0 { - tracing::info!("v21: ai_messages 已有 {} 条,跳过数据迁移", existing); - conn.execute("INSERT OR IGNORE INTO schema_version (version) VALUES (?)", [21])?; - return Ok(()); - } - - // 4. 遍历 ai_conversations,反序列化 messages JSON → 逐条写入 ai_messages + // 3. 遍历 ai_conversations,反序列化 messages JSON → 逐条写入 ai_messages let mut stmt = conn.prepare( "SELECT id, messages, created_at FROM ai_conversations", )?; @@ -497,13 +492,13 @@ fn migrate_v21(conn: &Connection) -> Result<()> { })?; let all_rows: Vec<(String, String, String)> = rows.collect::, _>>()?; - // 5. 分批 commit(每 50 个对话一批,避免长事务持有写锁) + // 4. 分批 commit(每 50 个对话一批,避免长事务持有写锁) const BATCH_SIZE: usize = 50; let mut migrated_count: usize = 0; for (batch_idx, batch) in all_rows.chunks(BATCH_SIZE).enumerate() { let tx = conn.unchecked_transaction()?; for (conv_id, messages_json, conv_created_at) in batch { - // 6. 逐对话反序列化 messages JSON → Vec + // 5. 逐对话反序列化 messages JSON → Vec // (用裸 JSON 而非 ChatMessage,因 df-storage 不依赖 df-ai-core) let messages: Vec = match serde_json::from_str(messages_json) { Ok(v) => v, @@ -513,8 +508,23 @@ fn migrate_v21(conn: &Connection) -> Result<()> { } }; + // 逐对话进度判定(幂等核心):批事务保证对话级原子——一个对话的全部消息 + // 在同一事务内写入,要么全落库要么全无,不存在半迁持久态。已落库数等于 + // JSON 消息数即该对话已完整迁移,跳过;否则(未迁/断点残缺)整对话重跑, + // 行级 INSERT OR IGNORE 兜底去重。中途崩溃重跑只补未迁对话,不再被 + // 全局 COUNT>0 整段短路。 + let json_count = messages.len() as i64; + let db_count: i64 = tx.query_row( + "SELECT COUNT(*) FROM ai_messages WHERE conversation_id = ?1", + rusqlite::params![conv_id], + |row| row.get(0), + )?; + if json_count > 0 && db_count == json_count { + continue; + } + for (seq, msg) in messages.iter().enumerate() { - // 7. 逐条消息提取字段 → INSERT INTO ai_messages + // 6. 逐条消息提取字段 → INSERT INTO ai_messages // 字段名硬编码("role"/"content" 等)——ChatMessage 改名会漏数据! // 迁移期 ID 天然唯一(UNIQUE 是 conv_id+seq),人类可读,零依赖 let id = format!("msg_migrated_{}_{}", conv_id, seq); @@ -1275,6 +1285,55 @@ fn migrate_v42(conn: &Connection) -> Result<()> { Ok(()) } +/// V43: 存量知识一次性晋升 — candidate 且 confidence='high' 批量转 published +/// +/// 决策单 B4①(docs/02-架构设计/专项设计/aichat上下文管理-派工与决策单-2026-09-29.md ①, +/// 2026-09-29)。DB 实测:knowledges 表 candidate 967 条(其中 confidence='high' 854 条) +/// 长期滞留,而检索侧(search/vector/top_reused)只认 published(53 条)——AI 提炼的知识 +/// 从未进入检索面,知识库沉淀近乎零生效。本迁移把存量高置信 candidate 一次性晋升, +/// 与代码侧「复用即信任」(KnowledgeRepo::auto_publish_if_reused)互补:前者清历史存量, +/// 后者管增量沉淀。 +/// +/// 设计要点: +/// - **纯数据 UPDATE,不经过 promote 函数**:不触发嵌入生成(854 条不会在启动时批量调 +/// 嵌入 API);embedding NULL 的条目检索走 LIKE 降级,向量面由后续复用/晋升路径逐步补 +/// ——这是有意为之,非遗漏。也不写 updated_at(非内容变更,不伪造新近度)。 +/// - **幂等双保险**:schema_version 版本号保证只跑一次(run() 按 current_version<43 跳过); +/// 即便崩溃重跑/版本号回退重调本函数,WHERE 条件天然幂等(已晋升行不再命中,UPDATE +/// 影响 0 行),版本号 INSERT OR IGNORE 防 PK 冲突(同 V21/V32 崩溃重跑防御模式)。 +/// - **可逆方式**(备用,当前不执行):本迁移不落版本特征列,knowledges.updated_at 由业务 +/// 路径维护、无法精确圈定本次影响集,可逆按晋升前后 confidence 不变这一事实圈定: +/// `UPDATE knowledges SET status='candidate' WHERE status='published' AND confidence='high'` +/// ——注意此语句会连带回退「非本迁移产生的 published+high」条目(人工放行/复用达标晋升的 +/// 高置信条目),执行前需以 knowledge_events 审计表核对时间窗。 +fn migrate_v43(conn: &Connection) -> Result<()> { + let affected = conn.execute( + "UPDATE knowledges SET status = 'published' \ + WHERE status = 'candidate' AND confidence = 'high'", + [], + )?; + conn.execute("INSERT OR IGNORE INTO schema_version (version) VALUES (?)", [43])?; + tracing::info!("迁移 v43 完成(存量知识 candidate+confidence=high 晋升 published,共 {} 条)", affected); + Ok(()) +} + +/// V44: 子代理轨迹来源标记 — ai_conversations 加 origin TEXT 列。 +/// +/// 子代理会话(spawn_agent)从「纯内存零落库」改为出口整轨迹落库(二期池①),落库时 +/// origin='sub_agent' 随首插写入,桌面侧栏与 miniapp 会话列表按此过滤——对话本体可追溯 +/// 而不污染主会话列表。主会话 origin 恒为 NULL(业务层 None),无其他写入点。 +/// - **幂等**:column_exists 守卫(同 V42 模式),崩溃重跑/全链两遍不炸; +/// - **可逆**:`ALTER TABLE ai_conversations DROP COLUMN origin`(SQLite 3.35+)。 +fn migrate_v44(conn: &Connection) -> Result<()> { + if !column_exists(conn, "ai_conversations", "origin") { + conn.execute("ALTER TABLE ai_conversations ADD COLUMN origin TEXT", [])?; + tracing::info!("v44: ai_conversations 加 origin 列(子代理轨迹来源标记)"); + } + conn.execute("INSERT OR IGNORE INTO schema_version (version) VALUES (?)", [44])?; + tracing::info!("迁移 v44 完成: ai_conversations 加 origin 列"); + Ok(()) +} + /// V21 建表 SQL — 消息拆分存储 ai_messages 表 /// /// 与 V9_SQL 中的 ai_messages 镜像(V9 给新库,此 const 给老库 V21 迁移用 IF NOT EXISTS)。 @@ -1742,7 +1801,7 @@ mod tests { assert_eq!(bad_count, 0); } - /// 幂等重跑:第二次 migrate_v21 不重复迁移(COUNT 探测跳过) + /// 幂等重跑:第二次 migrate_v21 按逐对话进度判定跳过已迁对话,不重复迁移 #[test] fn v21_idempotent_rerun() { let conn = setup_legacy_db(); @@ -1759,7 +1818,7 @@ mod tests { .unwrap(); assert_eq!(count_after_first, 1); - // 第二次跑:COUNT 探测 > 0 → 跳过数据迁移,不重复 + // 第二次跑:逐对话进度判定命中 → 跳过,不重复 migrate_v21(&conn).expect("二次迁移应幂等成功"); let count_after_second: i64 = conn .query_row("SELECT COUNT(*) FROM ai_messages", [], |r| r.get(0)) @@ -1778,6 +1837,141 @@ mod tests { assert_eq!(v_count, 1, "版本号 21 应只写一次"); } + /// 断点续迁:批次中途崩溃后 ai_messages 已有部分对话数据、版本号未写, + /// 重跑必须补齐未迁对话,不得因「表已有数据」整段跳过。 + /// 旧实现全局 COUNT>0 即跳过全部数据迁移,此场景下 conv_b 永不再迁而版本号照样补齐。 + #[test] + fn v21_crash_midway_resumes_missing_convs() { + let conn = setup_legacy_db(); + let msgs_a = serde_json::json!([ + {"role": "user", "content": "a1"}, + {"role": "assistant", "content": "a2"} + ]) + .to_string(); + let msgs_b = serde_json::json!([{"role": "user", "content": "b1"}]).to_string(); + conn.execute( + "INSERT INTO ai_conversations (id, messages, created_at, updated_at) VALUES ('conv_a', ?1, '0', '0')", + rusqlite::params![msgs_a], + ) + .unwrap(); + conn.execute( + "INSERT INTO ai_conversations (id, messages, created_at, updated_at) VALUES ('conv_b', ?1, '0', '0')", + rusqlite::params![msgs_b], + ) + .unwrap(); + + // 模拟崩溃断点:表已建(V21_SQL,首次跑 migrate_v21 的产物)、conv_a 已按迁移格式 + // 落库(其批次已 commit),conv_b 未迁,版本号未写 + conn.execute_batch(V21_SQL).unwrap(); + for seq in 0..2i64 { + conn.execute( + "INSERT INTO ai_messages (id, conversation_id, seq, role, content, status, created_at) + VALUES (?1, 'conv_a', ?2, 'user', ?3, 'active', '0')", + rusqlite::params![ + format!("msg_migrated_conv_a_{}", seq), + seq, + format!("a{}", seq) + ], + ) + .unwrap(); + } + let version_before: i64 = conn + .query_row( + "SELECT COUNT(*) FROM schema_version WHERE version = 21", + [], + |r| r.get(0), + ) + .unwrap(); + assert_eq!(version_before, 0, "崩溃断点:版本号尚未写入"); + + migrate_v21(&conn).expect("重跑应补齐未迁对话"); + + // conv_b 被补迁(旧实现此处为 0 —— 被 COUNT>0 整段跳过) + let b_count: i64 = conn + .query_row( + "SELECT COUNT(*) FROM ai_messages WHERE conversation_id = 'conv_b'", + [], + |r| r.get(0), + ) + .unwrap(); + assert_eq!(b_count, 1, "未迁对话应被补齐"); + + // conv_a 不重复迁移 + let a_count: i64 = conn + .query_row( + "SELECT COUNT(*) FROM ai_messages WHERE conversation_id = 'conv_a'", + [], + |r| r.get(0), + ) + .unwrap(); + assert_eq!(a_count, 2, "已迁对话不应重复插入"); + + let v: i64 = conn + .query_row("SELECT MAX(version) FROM schema_version", [], |r| r.get(0)) + .unwrap(); + assert_eq!(v, 21, "重跑后版本号应补齐"); + } + + /// 批次边界断点续迁:55 个对话分两批(每批 50),首批已 commit 后崩溃, + /// 重跑应只补剩余批次,全部对话迁移完整且无重复。 + #[test] + fn v21_batch_boundary_crash_resumes_remaining_batch() { + let conn = setup_legacy_db(); + let total = 55usize; + let batch_done = 50usize; + for i in 0..total { + let msgs = serde_json::json!([ + {"role": "user", "content": format!("q{}", i)}, + {"role": "assistant", "content": format!("r{}", i)} + ]) + .to_string(); + conn.execute( + "INSERT INTO ai_conversations (id, messages, created_at, updated_at) VALUES (?1, ?2, '0', '0')", + rusqlite::params![format!("conv_{:03}", i), msgs], + ) + .unwrap(); + } + // 模拟崩溃断点:表已建(V21_SQL)、首批 50 对话已按迁移格式落库(批次已 commit),其余未迁 + conn.execute_batch(V21_SQL).unwrap(); + for i in 0..batch_done { + for seq in 0..2i64 { + conn.execute( + "INSERT INTO ai_messages (id, conversation_id, seq, role, content, status, created_at) + VALUES (?1, ?2, ?3, 'user', '', 'active', '0')", + rusqlite::params![ + format!("msg_migrated_conv_{:03}_{}", i, seq), + format!("conv_{:03}", i), + seq + ], + ) + .unwrap(); + } + } + + migrate_v21(&conn).expect("重跑应补齐剩余批次"); + + let total_count: i64 = conn + .query_row("SELECT COUNT(*) FROM ai_messages", [], |r| r.get(0)) + .unwrap(); + assert_eq!( + total_count as usize, + total * 2, + "55 对话 × 2 条,无重复无缺失" + ); + + // 逐对话完整性:不存在条数不足的残缺对话 + let incomplete: i64 = conn + .query_row( + "SELECT COUNT(*) FROM ( + SELECT conversation_id FROM ai_messages GROUP BY conversation_id HAVING COUNT(*) < 2 + )", + [], + |r| r.get(0), + ) + .unwrap(); + assert_eq!(incomplete, 0, "不应存在残缺对话"); + } + /// ai_tool_executions.message_id 列补建(老库已有表无该列) #[test] fn v21_adds_message_id_column_to_tool_executions() { @@ -1962,7 +2156,108 @@ mod tests { } // ============================================================ - // 全量迁移测试 — 新库从零跑完整 V1-V41 路径 + // V43 迁移(存量知识一次性晋升) + // ============================================================ + + /// 构造最小 knowledges 表(schema_version 同建),只含 V43 触及的列。 + fn setup_knowledges_db() -> Connection { + let conn = Connection::open_in_memory().expect("open in-memory db"); + conn.execute_batch( + "CREATE TABLE schema_version (version INTEGER PRIMARY KEY); + CREATE TABLE knowledges ( + id TEXT PRIMARY KEY, + status TEXT NOT NULL DEFAULT 'candidate', + confidence TEXT + );", + ) + .expect("create knowledges table"); + conn + } + + /// V43:仅 candidate + confidence='high' 晋升,其余状态/置信度不动 + #[test] + fn v43_promotes_only_candidate_high_confidence() { + let conn = setup_knowledges_db(); + conn.execute_batch( + "INSERT INTO knowledges (id, status, confidence) VALUES + ('k1', 'candidate', 'high'), + ('k2', 'candidate', 'high'), + ('k3', 'candidate', 'medium'), + ('k4', 'candidate', 'low'), + ('k5', 'candidate', NULL), + ('k6', 'pending_review', 'high'), + ('k7', 'published', 'high'), + ('k8', 'archived', 'high');", + ) + .unwrap(); + + migrate_v43(&conn).expect("v43 应成功晋升"); + + // k1/k2 转 published + let published_high: i64 = conn + .query_row( + "SELECT COUNT(*) FROM knowledges WHERE status = 'published' AND confidence = 'high'", + [], |r| r.get(0), + ) + .unwrap(); + assert_eq!(published_high, 3, "2 条存量晋升 + 原 k7 = 3 条 published+high"); + + // 非 candidate+high 一律不动 + let untouched: i64 = conn + .query_row( + "SELECT COUNT(*) FROM knowledges WHERE id IN ('k3','k4','k5','k6','k8') \ + AND status NOT IN ('candidate','pending_review','archived')", + [], |r| r.get(0), + ) + .unwrap(); + assert_eq!(untouched, 0, "medium/low/NULL 置信与 pending_review/archived 不应被晋升"); + + // candidate 剩余:k3/k4/k5(2 条 high 已走) + let candidates_left: i64 = conn + .query_row( + "SELECT COUNT(*) FROM knowledges WHERE status = 'candidate'", + [], |r| r.get(0), + ) + .unwrap(); + assert_eq!(candidates_left, 3, "仅剩 medium/low/NULL 置信的 candidate"); + + let v: i64 = conn + .query_row("SELECT MAX(version) FROM schema_version", [], |r| r.get(0)) + .unwrap(); + assert_eq!(v, 43, "应写入版本号 43"); + } + + /// V43 幂等:重跑 UPDATE 影响 0 行(WHERE 天然幂等,版本号防重入) + #[test] + fn v43_idempotent_rerun_zero_affected() { + let conn = setup_knowledges_db(); + conn.execute( + "INSERT INTO knowledges (id, status, confidence) VALUES ('k1', 'candidate', 'high')", + [], + ) + .unwrap(); + + migrate_v43(&conn).expect("首次晋升"); + migrate_v43(&conn).expect("重跑应幂等成功(UPDATE 0 行)"); + + let published: i64 = conn + .query_row( + "SELECT COUNT(*) FROM knowledges WHERE status = 'published'", + [], |r| r.get(0), + ) + .unwrap(); + assert_eq!(published, 1, "重跑不应产生重复变更"); + let v_count: i64 = conn + .query_row( + "SELECT COUNT(*) FROM schema_version WHERE version = 43", + [], |r| r.get(0), + ) + .unwrap(); + assert_eq!(v_count, 1, "版本号 43 应只写一次"); + } + + // ============================================================ + // 全量迁移测试 — 新库从零跑完整 V1-V43 路径 // ------------------------------------------------------------ // 目的:某 migrate_vN 的 SQL 手滑写错(列名/类型/缺索引/缺表)只能等运行时暴露, // 此测试一次性覆盖全部迁移路径。任何一条迁移 SQL 写错、列名拼错、缺建表 @@ -1986,7 +2281,7 @@ mod tests { ); } - /// 全量迁移:新库从零跑完 V1-V41,验证关键表齐全 + 列数 > 0 + 关键列存在。 + /// 全量迁移:新库从零跑完 V1-V43,验证关键表齐全 + 列数 > 0 + 关键列存在。 /// /// 覆盖至少:task / ai_conversations / ai_messages / ai_tool_executions / /// conversation_checkpoints / ai_providers / projects / ideas。 @@ -2043,7 +2338,7 @@ mod tests { "tasks.module_id 列缺失(V41 加)" ); - // 3. schema_version 应推进到 42(全量迁移成功落版本号) + // 3. schema_version 应推进到 43(全量迁移成功落版本号) let max_version: i64 = conn .query_row( "SELECT COALESCE(MAX(version), 0) FROM schema_version", @@ -2052,8 +2347,8 @@ mod tests { ) .expect("查 schema_version 应成功"); assert_eq!( - max_version, 42, - "全量迁移后 schema_version 应为 42(实际 {}),说明某条 migrate_vN 链路断在中间", + max_version, 44, + "全量迁移后 schema_version 应为 44(实际 {}),说明某条 migrate_vN 链路断在中间", max_version ); @@ -2114,7 +2409,7 @@ mod tests { assert_eq!(v_count, 1, "版本号 32 应只写一次"); } - /// G5.5: 全链幂等不变量——V1-V41 每步执行两遍不抛错。 + /// G5.5: 全链幂等不变量——V1-V43 每步执行两遍不抛错。 /// /// 首轮 run() 建全 schema;清空 schema_version 强制下一轮从 V1 重跑每步 /// (模拟存量库 + 崩溃重跑/版本号回退)。任何 migrate_vN 的裸 ALTER(无 column_exists @@ -2130,7 +2425,7 @@ mod tests { |r| r.get(0), ) .unwrap(); - assert_eq!(max_v, 42, "首轮应推进到 42"); + assert_eq!(max_v, 44, "首轮应推进到 44"); // 清空版本表强制全链第二遍(每步 execute 第二次) conn.execute("DELETE FROM schema_version", []).unwrap(); @@ -2142,7 +2437,7 @@ mod tests { |r| r.get(0), ) .unwrap(); - assert_eq!(max_v2, 42, "重跑后应重新推进到 42"); + assert_eq!(max_v2, 44, "重跑后应重新推进到 44"); } // ============================================================ diff --git a/crates/df-storage/src/models.rs b/crates/df-storage/src/models.rs index 0a0b722..1972135 100644 --- a/crates/df-storage/src/models.rs +++ b/crates/df-storage/src/models.rs @@ -371,6 +371,9 @@ pub struct AiConversationRecord { pub pending_approvals: Option, // 挂起审批快照(JSON 对象,以 tool_call_id 为键,默认'{}') pub created_at: String, pub updated_at: String, + /// 会话来源标记(V44):None=主会话;Some("sub_agent")=子代理轨迹(spawn_agent 出口落库), + /// 桌面侧栏/miniapp 列表按此过滤。不可变列(UPDATE 不写)。 + pub origin: Option, } /// AI 工具执行审计记录