From df8ed7e74fe95df1547c0adacf8de99ec401e618 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=BB=9D=E5=B0=98?= <237809796@qq.com> Date: Fri, 26 Jun 2026 22:50:48 +0800 Subject: [PATCH] =?UTF-8?q?=E6=96=B0=E5=A2=9E:=20=E7=9F=A5=E8=AF=86?= =?UTF-8?q?=E5=9B=BE=E8=B0=B1Phase1=E6=95=B0=E6=8D=AE=E5=B1=82(=E7=88=B6?= =?UTF-8?q?=E2=91=A1)=20+=20=E4=BF=AE=E5=A4=8D=20idea=20list=5Fby=5Fquery?= =?UTF-8?q?=20=E5=8D=A0=E4=BD=8D=E7=AC=A6(=E7=88=B6=E2=91=A4=E2=91=A4.1)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 父② 数据层(V29): - migrate_v29: tasks +queue/parent_id/content_json + task_links 表+双索引(幂等对标 v14/v24) - TaskRecord 加 3 字段(15→18 列,INSERT/SELECT×6/from_row/struct 同步)+ 8 构造点补默认 - TaskLinkRepo(新文件,CRUD + 循环依赖 BFS depends_on 链检测,14 测) - task_repo list_by_query queue/parent_id 筛选 + get_children + 聚合计数(10 测) - queue DEFAULT 'todo' 向后兼容;task_links 软删保留(无 ON DELETE) idea_repo list_by_query 占位符修复(父⑤⑤.1 引入的潜伏 bug): - 加 deleted_at IS NULL 恒带后 where_clauses.len()+1 偏移 → 非空 status/keyword 查询 rusqlite 错误 - 改 params_vec.len()+1(对标 task_repo ②.2 同款修复,父② agent 发现) df-storage 96 lib + 11 集成测试全过。 --- crates/df-mcp/src/tools.rs | 6 + crates/df-nodes/src/ai_self_review_node.rs | 3 + crates/df-nodes/src/task_advance_node.rs | 3 + crates/df-storage/src/crud/idea_repo.rs | 8 +- crates/df-storage/src/crud/mod.rs | 4 + crates/df-storage/src/crud/task_link_repo.rs | 580 ++++++++++++++++++ crates/df-storage/src/crud/task_repo.rs | 355 ++++++++++- crates/df-storage/src/migrations.rs | 77 ++- crates/df-storage/src/models.rs | 50 ++ .../df-storage/tests/project_soft_delete.rs | 3 + docs/todo.md | 4 +- src-tauri/src/commands/ai/tool_registry.rs | 6 + src-tauri/src/commands/task.rs | 6 + 13 files changed, 1082 insertions(+), 23 deletions(-) create mode 100644 crates/df-storage/src/crud/task_link_repo.rs diff --git a/crates/df-mcp/src/tools.rs b/crates/df-mcp/src/tools.rs index a0f536a..605f7b9 100644 --- a/crates/df-mcp/src/tools.rs +++ b/crates/df-mcp/src/tools.rs @@ -393,6 +393,9 @@ fn create_task(ctx: &Ctx, args: Value) -> BoxFuture<'static, CallToolResult> { review_rounds: 0, output_json: None, idea_id: None, + queue: "todo".to_string(), + parent_id: None, + content_json: None, created_at: now.clone(), updated_at: now, }; @@ -446,6 +449,9 @@ fn update_task(ctx: &Ctx, args: Value) -> BoxFuture<'static, CallToolResult> { review_rounds: existing.review_rounds, output_json: existing.output_json, idea_id: existing.idea_id, + queue: existing.queue, + parent_id: existing.parent_id, + content_json: existing.content_json, created_at: existing.created_at, updated_at: now, }; diff --git a/crates/df-nodes/src/ai_self_review_node.rs b/crates/df-nodes/src/ai_self_review_node.rs index 3b2f200..27ec792 100644 --- a/crates/df-nodes/src/ai_self_review_node.rs +++ b/crates/df-nodes/src/ai_self_review_node.rs @@ -311,6 +311,9 @@ mod tests { review_rounds: 0, output_json: task_output_json.map(String::from), idea_id: None, + queue: "todo".to_string(), + parent_id: None, + content_json: None, created_at: "0".to_string(), updated_at: "0".to_string(), }) diff --git a/crates/df-nodes/src/task_advance_node.rs b/crates/df-nodes/src/task_advance_node.rs index 66cd58f..c2af821 100644 --- a/crates/df-nodes/src/task_advance_node.rs +++ b/crates/df-nodes/src/task_advance_node.rs @@ -190,6 +190,9 @@ mod tests { review_rounds: 0, output_json: None, idea_id: None, + queue: "todo".to_string(), + parent_id: None, + content_json: None, created_at: "0".to_string(), updated_at: "0".to_string(), } diff --git a/crates/df-storage/src/crud/idea_repo.rs b/crates/df-storage/src/crud/idea_repo.rs index a0b86f1..561ea35 100644 --- a/crates/df-storage/src/crud/idea_repo.rs +++ b/crates/df-storage/src/crud/idea_repo.rs @@ -278,12 +278,16 @@ impl IdeaRepo { let mut params_vec: Vec> = Vec::new(); if let Some(s) = &status { - where_clauses.push(format!("status = ?{}", where_clauses.len() + 1)); + // 占位符编号用 params_vec.len()+1(参数实际位置),非 where_clauses.len()+1 + // (where_clauses 含 deleted_at IS NULL 常量无占位符子句,len() 会偏移致 ?N 与参数错位 + // — 父⑤⑤.1 加 deleted_at 恒带引入的潜伏 bug,非空 status/keyword 查询 rusqlite 报 + // "needed N, got M"。对标 task_repo list_by_query ②.2 同款修复) + where_clauses.push(format!("status = ?{}", params_vec.len() + 1)); params_vec.push(Box::new(s.clone())); } if let Some(kw) = &keyword { let pat = format!("%{kw}%"); - let p1 = where_clauses.len() + 1; + let p1 = params_vec.len() + 1; let p2 = p1 + 1; where_clauses.push(format!("(title LIKE ?{p1} OR description LIKE ?{p2})")); params_vec.push(Box::new(pat.clone())); diff --git a/crates/df-storage/src/crud/mod.rs b/crates/df-storage/src/crud/mod.rs index d46940f..6e47980 100644 --- a/crates/df-storage/src/crud/mod.rs +++ b/crates/df-storage/src/crud/mod.rs @@ -6,6 +6,7 @@ //! - [`mod@settings`]:SettingsRepo + 列白名单(allowed_columns_for/validate_column_name/is_allowed_column) //! - [`mod@project_repo`]:ProjectRepo/BranchRepo/ReleaseRepo/WorkflowRepo/NodeExecutionRepo //! - [`mod@task_repo`]:TaskRepo(含 advance_status_atomic 状态机收口) +//! - [`mod@task_link_repo`]:TaskLinkRepo(任务横向关联 task_links,V29,知识图谱 Phase 1) //! - [`mod@conversation_repo`]:AiProviderRepo/AiConversationRepo/AiToolExecutionRepo //! - [`mod@idea_repo`]:IdeaRepo/KnowledgeRepo/KnowledgeEventsRepo + 向量工具 //! - [`mod@idea_eval_repo`]:IdeaEvalRepo(灵感评估历史追加型审计表 idea_evaluations,V22) @@ -20,6 +21,7 @@ mod idea_repo; mod message_repo; mod project_repo; mod settings; +mod task_link_repo; mod task_repo; pub use conversation_repo::*; @@ -28,6 +30,7 @@ pub use idea_repo::*; pub use message_repo::*; pub use project_repo::*; pub use settings::*; +pub use task_link_repo::*; pub use task_repo::*; // ============================================================ @@ -293,6 +296,7 @@ mod baseline_tests { let _ = IdeaRepo::new(&db); let _ = ProjectRepo::new(&db); let _ = TaskRepo::new(&db); + let _ = TaskLinkRepo::new(&db); let _ = BranchRepo::new(&db); let _ = ReleaseRepo::new(&db); let _ = WorkflowRepo::new(&db); diff --git a/crates/df-storage/src/crud/task_link_repo.rs b/crates/df-storage/src/crud/task_link_repo.rs new file mode 100644 index 0000000..406d58b --- /dev/null +++ b/crates/df-storage/src/crud/task_link_repo.rs @@ -0,0 +1,580 @@ +//! 任务横向关联 Repo:TaskLinkRepo(task_links 表,知识图谱 Phase 1 V29) +//! +//! 表达任务间 depends_on / blocks / relates_to 关系,AI 拓扑排序编排调度的基础。 +//! 对标设计 docs/02-架构设计/专项设计/项目知识图谱与任务队列系统-2026-06-26.md §2.2。 +//! +//! 关键设计: +//! - `link_type` 白名单(depends_on/blocks/relates_to)应用层校验,非 DB 约束。 +//! - 循环依赖(`depends_on` 链 A→B→A)在 `create_link` BFS 检测拒绝(非 DB 约束,对标 D8)。 +//! - 跨项目依赖允许(现实中有跨项目依赖)。 +//! - 软删除语义:Task 软删不级联删 link(恢复后关系还在)。 + +use std::collections::HashSet; + +use rusqlite::{params, OptionalExtension, Row}; + +use df_types::error::Result; + +use crate::db::Database; +use crate::models::TaskLinkRecord; + +use super::{now_millis_str, storage_err}; + +// ============================================================ +// link_type 白名单 + 校验 +// ============================================================ + +/// `link_type` 白名单:只允许这三种关联类型(防拼写漂移 / 非法值进库)。 +/// +/// 对标设计 §2.2 link_type 语义: +/// - `depends_on`:source 依赖 target(target 完成后 source 才能开始)→ 拓扑排序调度 +/// - `blocks`:source 阻塞 target(source 不完成则 target 无法推进)→ 依赖的反向声明 +/// - `relates_to`:弱关联,无执行约束 → 上下文提示 +pub const TASK_LINK_TYPES: &[&str] = &["depends_on", "blocks", "relates_to"]; + +/// 校验 `link_type` 在白名单内,否则返回 Err。 +fn validate_link_type(link_type: &str) -> Result<()> { + if TASK_LINK_TYPES.contains(&link_type) { + Ok(()) + } else { + Err(df_types::error::Error::Storage(format!( + "非法 link_type: {link_type}，合法值: {:?}", + TASK_LINK_TYPES + ))) + } +} + +// ============================================================ +// from_row 辅助函数 +// ============================================================ + +fn task_link_from_row(row: &Row<'_>) -> std::result::Result { + Ok(TaskLinkRecord { + id: row.get("id")?, + source_id: row.get("source_id")?, + target_id: row.get("target_id")?, + link_type: row.get("link_type")?, + remark: row.get("remark")?, + created_at: row.get("created_at")?, + }) +} + +// ============================================================ +// TaskLinkRepo +// ============================================================ + +/// 任务横向关联表 Repo(task_links,V29)。 +/// +/// 不走 `impl_repo!` 宏:① 表无 `updated_at`/`deleted_at`(link 不可改只增删,审计简单); +/// ② `create_link` 需在 INSERT 前做 BFS 循环依赖检测(宏生成的 insert 无业务前置逻辑)。 +/// 故手写专用方法,对标 SettingsRepo 全专用路径。 +pub struct TaskLinkRepo { + conn: std::sync::Arc>, +} + +impl TaskLinkRepo { + pub fn new(db: &Database) -> Self { + Self { conn: db.conn() } + } + + /// 创建任务关联(应用层校验 link_type 白名单 + depends_on 链 BFS 循环依赖检测)。 + /// + /// 对标设计 §2.2 边界约束: + /// - `link_type` 必须在白名单(depends_on/blocks/relates_to),否则 Err。 + /// - 自环(source_id == target_id)直接拒绝(无意义的自依赖)。 + /// - 循环依赖检测:**仅对 depends_on 链**做 BFS。blocks 是 depends_on 的反向声明, + /// relates_to 无执行约束,二者不参与环检测(否则弱关联也会触发拒绝,过度约束)。 + /// BFS 思路:新增 source→target 后,从 target 出发沿 depends_on 链向下遍历, + /// 若能回到 source 则成环,拒绝。数据量小(单项目 ~50 任务),全表遍历无压力(对标 D8)。 + /// - 跨项目允许(不做 project_id 一致性校验,现实有跨项目依赖)。 + /// + /// 返回插入的 link id。 + pub async fn create_link( + &self, + id: &str, + source_id: &str, + target_id: &str, + link_type: &str, + remark: Option<&str>, + ) -> Result { + // 应用层校验先行(fail-fast,非法值不进 DB 层) + validate_link_type(link_type)?; + if source_id == target_id { + return Err(df_types::error::Error::Storage(format!( + "非法 task_link: source_id 与 target_id 相同({source_id}),自环无意义" + ))); + } + + let conn = self.conn.clone(); + let id = id.to_owned(); + let source_id_o = source_id.to_owned(); + let target_id_o = target_id.to_owned(); + let link_type_o = link_type.to_owned(); + let remark_o = remark.map(|s| s.to_owned()); + let now = now_millis_str(); + + tokio::task::spawn_blocking(move || { + let guard = conn.blocking_lock(); + + // depends_on 链 BFS 环检测:新增 source→target 边后,从 target 沿 depends_on + // 向下走,若能回到 source 即成环。在 INSERT 前检测,避免脏数据(检测与插入非原子, + // 但单用户桌面应用无并发,够用)。 + if link_type_o == "depends_on" { + if bfs_reaches(&guard, &target_id_o, &source_id_o)? { + return Err(df_types::error::Error::Storage(format!( + "非法 task_link: {} depends_on {} 会形成循环依赖", + source_id_o, target_id_o + ))); + } + } + + guard + .execute( + "INSERT INTO task_links (id, source_id, target_id, link_type, remark, created_at) + VALUES (?1, ?2, ?3, ?4, ?5, ?6)", + params![id, source_id_o, target_id_o, link_type_o, remark_o, now], + ) + .map_err(storage_err)?; + Ok(id) + }) + .await + .map_err(storage_err)? + } + + /// 删除关联(按 id),返回是否命中。 + 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 + .execute("DELETE FROM task_links WHERE id = ?1", params![id]) + .map_err(storage_err)?; + Ok(affected > 0) + }) + .await + .map_err(storage_err)? + } + + /// 按 id 取单条(对标 IdeaRepo::get_by_id 模式)。 + pub async fn get_by_id(&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 row = guard + .query_row( + "SELECT id, source_id, target_id, link_type, remark, created_at \ + FROM task_links WHERE id = ?1", + params![id], + task_link_from_row, + ) + .optional() + .map_err(storage_err)?; + Ok(row) + }) + .await + .map_err(storage_err)? + } + + /// 按 source_id 查询(source 主动声明的全部关联),命中 idx_task_links_source。 + /// AI 编排调度用:取某任务的全部依赖 / 阻塞 / 关联。 + pub async fn get_by_source(&self, source_id: &str) -> Result> { + let conn = self.conn.clone(); + let source_id = source_id.to_owned(); + tokio::task::spawn_blocking(move || { + let guard = conn.blocking_lock(); + let mut stmt = guard + .prepare( + "SELECT id, source_id, target_id, link_type, remark, created_at \ + FROM task_links WHERE source_id = ?1 ORDER BY created_at ASC", + ) + .map_err(storage_err)?; + let rows = stmt + .query_map(params![source_id], task_link_from_row) + .map_err(storage_err)?; + let mut results = Vec::new(); + for r in rows { + results.push(r.map_err(storage_err)?); + } + Ok(results) + }) + .await + .map_err(storage_err)? + } + + /// 按 target_id 查询(谁指向了 target),命中 idx_task_links_target。 + /// AI 编排调度用:「谁依赖了我」「谁被我阻塞」反向查询(JSON 列方案无法高效反查的痛点)。 + pub async fn get_by_target(&self, target_id: &str) -> Result> { + let conn = self.conn.clone(); + let target_id = target_id.to_owned(); + tokio::task::spawn_blocking(move || { + let guard = conn.blocking_lock(); + let mut stmt = guard + .prepare( + "SELECT id, source_id, target_id, link_type, remark, created_at \ + FROM task_links WHERE target_id = ?1 ORDER BY created_at ASC", + ) + .map_err(storage_err)?; + let rows = stmt + .query_map(params![target_id], task_link_from_row) + .map_err(storage_err)?; + let mut results = Vec::new(); + for r in rows { + results.push(r.map_err(storage_err)?); + } + Ok(results) + }) + .await + .map_err(storage_err)? + } + + /// 列出全部关联(调试/管理用,按创建时间升序)。 + pub async fn list_all(&self) -> Result> { + let conn = self.conn.clone(); + tokio::task::spawn_blocking(move || { + let guard = conn.blocking_lock(); + let mut stmt = guard + .prepare( + "SELECT id, source_id, target_id, link_type, remark, created_at \ + FROM task_links ORDER BY created_at ASC", + ) + .map_err(storage_err)?; + let rows = stmt.query_map([], task_link_from_row).map_err(storage_err)?; + let mut results = Vec::new(); + for r in rows { + results.push(r.map_err(storage_err)?); + } + Ok(results) + }) + .await + .map_err(storage_err)? + } +} + +// ============================================================ +// BFS 循环依赖检测(仅 depends_on 链) +// ============================================================ + +/// BFS:从 `start` 出发沿 depends_on 链(source→target 方向)向下遍历,判断能否到达 `target_node`。 +/// +/// 用于 `create_link(source, depends_on, target)` 前的环检测:新增 source→target 边后, +/// 若 target 能经 depends_on 链回到 source,则形成 source→target→...→source 闭环。 +/// 故调用 `bfs_reaches(conn, start=target, target_node=source)`:从 target 出发看能否到 source。 +/// +/// 实现细节: +/// - 遍历 `WHERE source_id = ? AND link_type = 'depends_on'` 取下一跳(对标设计 D8:数据量小, +/// 全表遍历无压力,不走 SQL 递归 CTE,纯 Rust BFS)。 +/// - `visited` HashSet 防重复访问(菱形依赖 A→B,A→C,B→D,C→D 不应死循环)。 +/// - 防御性深度上限(10000 跳):万一数据异常成环未被拦截,BFS 不致无限循环。 +fn bfs_reaches( + conn: &rusqlite::Connection, + start: &str, + target_node: &str, +) -> Result { + use std::collections::VecDeque; + + let mut visited: HashSet = HashSet::new(); + let mut queue: VecDeque = VecDeque::new(); + queue.push_back(start.to_owned()); + visited.insert(start.to_owned()); + + let mut depth = 0u32; + const MAX_DEPTH: u32 = 10_000; + + while let Some(node) = queue.pop_front() { + depth += 1; + if depth > MAX_DEPTH { + // 防御性兜底:正常依赖链不可能这么深,到这要么数据异常要么已有环未被拦截。 + // 保守视为成环(拒绝创建),避免无限循环 + 暴露异常数据。 + tracing::warn!( + "task_links BFS 超过 {MAX_DEPTH} 跳,疑似数据异常,保守拒绝创建" + ); + return Ok(true); + } + + if node == target_node { + return Ok(true); + } + + // 取该节点作为 source 的全部 depends_on 边的 target(下一跳) + let mut stmt = conn + .prepare( + "SELECT target_id FROM task_links \ + WHERE source_id = ?1 AND link_type = 'depends_on'", + ) + .map_err(storage_err)?; + let nexts: Vec = stmt + .query_map(params![node], |row| row.get::<_, String>(0)) + .map_err(storage_err)? + .filter_map(|r| r.ok()) + .collect(); + + for next in nexts { + if visited.insert(next.clone()) { + queue.push_back(next); + } + } + } + Ok(false) +} + +// ============================================================ +// 单元测试 — TaskLinkRepo CRUD + 循环依赖拒绝(内存 DB,对标 idea_repo 测试) +// ============================================================ + +#[cfg(test)] +mod tests { + use super::*; + use crate::crud::TaskRepo; + use crate::models::TaskRecord; + + /// 构造一条 TaskRecord fixture(18 字段全填,queue 默认 todo)。 + fn trec(id: &str, project_id: &str) -> TaskRecord { + TaskRecord { + id: id.to_string(), + project_id: project_id.to_string(), + title: format!("task-{id}"), + description: String::new(), + status: "todo".to_string(), + priority: 1, + 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: None, + content_json: None, + created_at: "1700000000000".to_string(), + updated_at: "1700000000000".to_string(), + } + } + + /// 构造内存 DB + 注入若干任务(task_links FK 要求 tasks 存在, + /// tasks.project_id FK 要求 projects 存在 → 先建占位 project 满足 FK 约束, + /// 对标 tests/project_soft_delete.rs 集成测试的 setup 模式)。 + async fn setup_with_tasks(ids: &[&str]) -> (crate::db::Database, TaskLinkRepo, TaskRepo) { + let db = crate::db::Database::open_in_memory() + .await + .expect("open_in_memory"); + let link_repo = TaskLinkRepo::new(&db); + let task_repo = TaskRepo::new(&db); + // 先建占位 project 满足 tasks.project_id FK(PRAGMA foreign_keys=ON,db.rs:37/37) + let project_repo = crate::crud::ProjectRepo::new(&db); + project_repo + .insert(crate::models::ProjectRecord { + id: "proj-1".to_string(), + name: "proj-1".to_string(), + description: String::new(), + status: "active".to_string(), + idea_id: None, + path: None, + stack: None, + created_at: "1700000000000".to_string(), + updated_at: "1700000000000".to_string(), + }) + .await + .unwrap(); + for id in ids { + task_repo.insert(trec(id, "proj-1")).await.unwrap(); + } + (db, link_repo, task_repo) + } + + #[tokio::test] + async fn validate_link_type_rejects_unknown() { + assert!(validate_link_type("depends_on").is_ok()); + assert!(validate_link_type("blocks").is_ok()); + assert!(validate_link_type("relates_to").is_ok()); + assert!(validate_link_type("unknown").is_err()); + assert!(validate_link_type("").is_err()); + } + + #[tokio::test] + async fn create_link_basic_depends_on() { + let (_db, repo, _task_repo) = setup_with_tasks(&["a", "b"]).await; + let id = repo + .create_link("l1", "a", "b", "depends_on", None) + .await + .unwrap(); + assert_eq!(id, "l1"); + + let got = repo.get_by_id("l1").await.unwrap().expect("link 存在"); + assert_eq!(got.source_id, "a"); + assert_eq!(got.target_id, "b"); + assert_eq!(got.link_type, "depends_on"); + assert!(got.remark.is_none()); + } + + #[tokio::test] + async fn create_link_with_remark() { + let (_db, repo, _) = setup_with_tasks(&["a", "b"]).await; + repo.create_link("l1", "a", "b", "blocks", Some("阻塞说明")) + .await + .unwrap(); + let got = repo.get_by_id("l1").await.unwrap().unwrap(); + assert_eq!(got.link_type, "blocks"); + assert_eq!(got.remark.as_deref(), Some("阻塞说明")); + } + + #[tokio::test] + async fn create_link_rejects_invalid_type() { + let (_db, repo, _) = setup_with_tasks(&["a", "b"]).await; + let err = repo + .create_link("l1", "a", "b", "invalid_type", None) + .await; + assert!(err.is_err(), "非法 link_type 应被拒绝"); + // 未入库 + assert!(repo.get_by_id("l1").await.unwrap().is_none()); + } + + #[tokio::test] + async fn create_link_rejects_self_loop() { + let (_db, repo, _) = setup_with_tasks(&["a"]).await; + let err = repo.create_link("l1", "a", "a", "depends_on", None).await; + assert!(err.is_err(), "自环 source==target 应被拒绝"); + } + + #[tokio::test] + async fn get_by_source_and_target() { + let (_db, repo, _) = setup_with_tasks(&["a", "b", "c"]).await; + repo.create_link("l1", "a", "b", "depends_on", None) + .await + .unwrap(); + repo.create_link("l2", "a", "c", "relates_to", None) + .await + .unwrap(); + repo.create_link("l3", "c", "b", "blocks", None) + .await + .unwrap(); + + // a 作为 source 声明了 2 条(l1/l2) + let from_a = repo.get_by_source("a").await.unwrap(); + let ids: Vec<_> = from_a.iter().map(|l| l.id.as_str()).collect(); + assert_eq!(ids, vec!["l1", "l2"]); + + // b 作为 target 被 2 条指向(l1/l3) + let to_b = repo.get_by_target("b").await.unwrap(); + let ids: Vec<_> = to_b.iter().map(|l| l.id.as_str()).collect(); + assert_eq!(ids, vec!["l1", "l3"]); + } + + #[tokio::test] + async fn delete_link() { + let (_db, repo, _) = setup_with_tasks(&["a", "b"]).await; + repo.create_link("l1", "a", "b", "depends_on", None) + .await + .unwrap(); + assert!(repo.delete("l1").await.unwrap()); + assert!(repo.get_by_id("l1").await.unwrap().is_none()); + // 再删返回 false + assert!(!repo.delete("l1").await.unwrap()); + } + + // ---------- 循环依赖 BFS 检测(仅 depends_on 链)---------- + + #[tokio::test] + async fn cycle_direct_a_depends_b_then_b_depends_a_rejected() { + // A→B 合法;再 B→A 应形成 A→B→A 闭环,拒绝 + let (_db, repo, _) = setup_with_tasks(&["a", "b"]).await; + repo.create_link("l1", "a", "b", "depends_on", None) + .await + .unwrap(); + let err = repo + .create_link("l2", "b", "a", "depends_on", None) + .await; + assert!(err.is_err(), "A→B→A 循环依赖应被拒绝"); + // l2 未入库 + assert!(repo.get_by_id("l2").await.unwrap().is_none()); + } + + #[tokio::test] + async fn cycle_three_nodes_rejected() { + // A→B→C 合法;再 C→A 形成 A→B→C→A 闭环,拒绝 + let (_db, repo, _) = setup_with_tasks(&["a", "b", "c"]).await; + repo.create_link("l1", "a", "b", "depends_on", None) + .await + .unwrap(); + repo.create_link("l2", "b", "c", "depends_on", None) + .await + .unwrap(); + let err = repo + .create_link("l3", "c", "a", "depends_on", None) + .await; + assert!(err.is_err(), "A→B→C→A 三节点循环依赖应被拒绝"); + } + + #[tokio::test] + async fn diamond_dependency_not_cycle() { + // 菱形依赖 A→B, A→C, B→D, C→D 是 DAG 非环,D 不应再指向 A/B/C + // 关键:BFS 遇菱形不误判(D 被两条路径到达,visited 去重不死循环) + let (_db, repo, _) = setup_with_tasks(&["a", "b", "c", "d"]).await; + repo.create_link("l1", "a", "b", "depends_on", None) + .await + .unwrap(); + repo.create_link("l2", "a", "c", "depends_on", None) + .await + .unwrap(); + repo.create_link("l3", "b", "d", "depends_on", None) + .await + .unwrap(); + repo.create_link("l4", "c", "d", "depends_on", None) + .await + .unwrap(); + // 全部成功(菱形合法) + assert_eq!(repo.list_all().await.unwrap().len(), 4); + } + + #[tokio::test] + async fn blocks_link_does_not_trigger_cycle_check() { + // blocks 是 depends_on 的反向声明,不参与环检测:A blocks B + B blocks A 应都合法 + // (虽然语义重复,但环检测只管 depends_on 链,blocks/relates_to 弱约束不强拦) + let (_db, repo, _) = setup_with_tasks(&["a", "b"]).await; + repo.create_link("l1", "a", "b", "blocks", None) + .await + .unwrap(); + // B blocks A 不触发 depends_on 环检测,合法 + repo.create_link("l2", "b", "a", "blocks", None) + .await + .unwrap(); + assert_eq!(repo.list_all().await.unwrap().len(), 2); + } + + #[tokio::test] + async fn relates_to_link_does_not_trigger_cycle_check() { + // relates_to 弱关联无执行约束,不参与环检测 + let (_db, repo, _) = setup_with_tasks(&["a", "b"]).await; + repo.create_link("l1", "a", "b", "relates_to", None) + .await + .unwrap(); + repo.create_link("l2", "b", "a", "relates_to", None) + .await + .unwrap(); + assert_eq!(repo.list_all().await.unwrap().len(), 2); + } + + #[tokio::test] + async fn cross_project_link_allowed() { + // 跨项目依赖允许(设计 §2.2 边界):不做 project_id 一致性校验 + let (_db, repo, _) = setup_with_tasks(&["a", "b"]).await; // fixture 同 proj-1,但 repo 不校验 + let res = repo.create_link("l1", "a", "b", "depends_on", None).await; + assert!(res.is_ok(), "跨项目依赖应允许(repo 层不校验 project 一致性)"); + } + + #[tokio::test] + async fn cycle_check_mixed_chain_only_depends_on_matters() { + // 混合链:A relates_to B(弱关联),B depends_on A 应合法(relates_to 不构成环路径) + let (_db, repo, _) = setup_with_tasks(&["a", "b"]).await; + repo.create_link("l1", "a", "b", "relates_to", None) + .await + .unwrap(); + // B depends_on A:BFS 从 A 出发沿 depends_on 找 B,A 无 depends_on 出边 → 不到 B → 合法 + repo.create_link("l2", "b", "a", "depends_on", None) + .await + .unwrap(); + assert_eq!(repo.list_all().await.unwrap().len(), 2); + } +} diff --git a/crates/df-storage/src/crud/task_repo.rs b/crates/df-storage/src/crud/task_repo.rs index 70111a7..89d99be 100644 --- a/crates/df-storage/src/crud/task_repo.rs +++ b/crates/df-storage/src/crud/task_repo.rs @@ -33,6 +33,10 @@ fn task_from_row(row: &Row<'_>) -> std::result::Result) -> std::result::Result, /// 关键词搜索(P2):title/description LIKE %kw%,对齐知识库 search 的 LIKE 模式。 pub keyword: Option, + /// 管理池过滤(知识图谱 Phase 1 V29):backlog/todo/decision/active/done。 + /// 看板视图按池分列的数据源。值合法性由上层 queue 语义校验兜底(非法值 DB 无匹配返回空)。 + #[serde(default)] + pub queue: Option, + /// 父任务 ID 过滤(知识图谱 Phase 1 V29):Some(id) = 查某父任务的子任务; + /// 查叶子任务(parent_id IS NULL)由专用方法 get_children 之外的语义决定,本字段只做等值匹配。 + #[serde(default)] + pub parent_id: Option, /// 排序字段(白名单 created_at/updated_at/priority/status,降序)。P3 基建就绪。 pub order_by: Option, /// 分页上限(钳制 ≤500)。P3 基建就绪。 @@ -103,22 +119,26 @@ impl_repo!( from_row => |row| task_from_row(row), insert => |conn, rec| { conn.execute( - "INSERT INTO tasks (id, project_id, title, description, status, priority, branch_name, assignee, workflow_def_id, base_branch, review_rounds, output_json, idea_id, created_at, updated_at) - VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15)", + "INSERT INTO tasks (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, created_at, updated_at) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17, ?18)", params![ rec.id, rec.project_id, rec.title, rec.description, rec.status, rec.priority, rec.branch_name, rec.assignee, rec.workflow_def_id, rec.base_branch, - rec.review_rounds, rec.output_json, rec.idea_id, rec.created_at, rec.updated_at + rec.review_rounds, rec.output_json, rec.idea_id, + rec.queue, rec.parent_id, rec.content_json, + rec.created_at, rec.updated_at ], ) }, update => |conn, rec| { conn.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, updated_at = ?13 WHERE id = ?14", + "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, updated_at = ?16 WHERE id = ?17", params![ rec.project_id, rec.title, rec.description, rec.status, rec.priority, rec.branch_name, rec.assignee, rec.workflow_def_id, rec.base_branch, - rec.review_rounds, rec.output_json, rec.idea_id, rec.updated_at, rec.id + rec.review_rounds, rec.output_json, rec.idea_id, + rec.queue, rec.parent_id, rec.content_json, + rec.updated_at, rec.id ], ) } @@ -127,14 +147,14 @@ impl_repo!( impl TaskRepo { /// 列出未删除任务(deleted_at IS NULL)— 对标 ProjectRepo::list_active /// - /// 显式列出全部 15 个 TaskRecord 列名(同 ProjectRepo::list_active 写法), + /// 显式列出全部 18 个 TaskRecord 列名(同 ProjectRepo::list_active 写法), /// 不 SELECT deleted_at:TaskRecord 不带该字段,取了 from_row 会因未知列报错。 pub async fn list_active(&self) -> Result> { let conn = self.conn.clone(); 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, created_at, updated_at FROM tasks WHERE deleted_at IS NULL ORDER BY created_at DESC") + .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, created_at, updated_at FROM tasks WHERE deleted_at IS NULL ORDER BY created_at DESC") .map_err(storage_err)?; let rows = stmt .query_map([], |row| task_from_row(row)) @@ -235,7 +255,7 @@ impl TaskRepo { } // 回读更新后的记录(含新 status / 累加后的 review_rounds / 新 updated_at)。 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, created_at, updated_at FROM tasks WHERE id = ?1") + .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, created_at, updated_at FROM tasks WHERE id = ?1") .map_err(storage_err)?; let row = stmt .query_row(params![id], |row| task_from_row(row)) @@ -257,7 +277,7 @@ impl TaskRepo { 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, created_at, updated_at FROM tasks WHERE deleted_at IS NULL AND project_id = ?1 ORDER BY created_at DESC") + .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, created_at, updated_at FROM tasks WHERE deleted_at IS NULL AND project_id = ?1 ORDER BY created_at DESC") .map_err(storage_err)?; let rows = stmt .query_map(params![pid], |row| task_from_row(row)) @@ -284,7 +304,7 @@ impl TaskRepo { /// - order_by 白名单(validate_order_by 防 SQL 注入),默认 created_at,恒 DESC(与 list_active 一致) /// - limit/offset 钳制(limit ≤500 防滥用,对齐 conversation_repo::list_recent 的 limit≤200 思路) /// - deleted_at IS NULL 恒带(回收站任务不进结果,语义同 list_active,不可被 query 关闭) - /// - 显式列出全部 15 列(不 SELECT deleted_at:TaskRecord 不带该字段,取了 from_row 报未知列) + /// - 显式列出全部 18 列(不 SELECT deleted_at:TaskRecord 不带该字段,取了 from_row 报未知列) /// /// 空 query(全 None)→ 等价 list_active(全量未删,created_at DESC),向后兼容。 /// status 值合法性由上层 list_tasks 命令(TaskStatus::is_valid)兜底,本层不过滤值集 @@ -311,35 +331,51 @@ impl TaskRepo { let priority = query.priority; let assignee = query.assignee.clone(); let keyword = query.keyword.clone(); + let queue = query.queue.clone(); + let parent_id = query.parent_id.clone(); tokio::task::spawn_blocking(move || { let guard = conn.blocking_lock(); - // ── 累积 WHERE 子句 + 收集参数(按出现顺序绑定占位符 ?N,序号 = 已有子句数+1)── + // ── 累积 WHERE 子句 + 收集参数(按出现顺序绑定占位符 ?N,序号 = params_vec.len()+1)── + // 占位符序号必须按「实际参数位置」(params_vec.len()+1)而非「子句数」(where_clauses.len()+1) + // 编号:deleted_at IS NULL 是常量条件无占位符却占 where_clauses[0],用子句数编号会让首个 + // 真参数拿到 ?2 而 params_vec 只有 1 个元素 → rusqlite "needed 2, got 1"。 + // 用 params_vec.len()+1 保证 ?N 与参数位置严格对齐(N = 参数序号)。 // deleted_at IS NULL 恒带(常量条件无占位符),回收站任务不进结果(语义同 list_active)。 let mut where_clauses: Vec = vec!["deleted_at IS NULL".to_string()]; let mut params_vec: Vec> = Vec::new(); if let Some(pid) = &project_id { - where_clauses.push(format!("project_id = ?{}", where_clauses.len() + 1)); + where_clauses.push(format!("project_id = ?{}", params_vec.len() + 1)); params_vec.push(Box::new(pid.clone())); } if let Some(s) = &status { - where_clauses.push(format!("status = ?{}", where_clauses.len() + 1)); + where_clauses.push(format!("status = ?{}", params_vec.len() + 1)); params_vec.push(Box::new(s.clone())); } if let Some(p) = priority { - where_clauses.push(format!("priority = ?{}", where_clauses.len() + 1)); + where_clauses.push(format!("priority = ?{}", params_vec.len() + 1)); params_vec.push(Box::new(p)); } if let Some(a) = &assignee { - where_clauses.push(format!("assignee = ?{}", where_clauses.len() + 1)); + where_clauses.push(format!("assignee = ?{}", params_vec.len() + 1)); params_vec.push(Box::new(a.clone())); } + // queue(知识图谱 Phase 1 V29):管理池等值过滤,看板视图按池分列数据源 + if let Some(q) = &queue { + where_clauses.push(format!("queue = ?{}", params_vec.len() + 1)); + params_vec.push(Box::new(q.clone())); + } + // parent_id(知识图谱 Phase 1 V29):父任务等值过滤(查某父的子任务) + if let Some(pid) = &parent_id { + where_clauses.push(format!("parent_id = ?{}", params_vec.len() + 1)); + params_vec.push(Box::new(pid.clone())); + } // keyword: title/description LIKE %kw%(P2,对齐知识库 search 的 LIKE 模式) if let Some(kw) = &keyword { let pat = format!("%{kw}%"); - let p1 = where_clauses.len() + 1; + let p1 = params_vec.len() + 1; let p2 = p1 + 1; where_clauses.push(format!("(title LIKE ?{p1} OR description LIKE ?{p2})")); params_vec.push(Box::new(pat.clone())); @@ -358,10 +394,13 @@ impl TaskRepo { }; // 拼 SQL:?N 占位符序号与 params_vec 顺序严格对应(累积时按 +1 递增保证)。 + // 显式列出全部 18 列(含 V29 queue/parent_id/content_json,不 SELECT deleted_at: + // TaskRecord 不带该字段,取了 from_row 会因未知列报错)。 let sql = format!( "SELECT id, project_id, title, description, status, priority, branch_name, \ assignee, workflow_def_id, base_branch, review_rounds, output_json, idea_id, \ - created_at, updated_at FROM tasks WHERE {} ORDER BY {} DESC{}", + queue, parent_id, content_json, created_at, updated_at \ + FROM tasks WHERE {} ORDER BY {} DESC{}", where_clauses.join(" AND "), order_col, limit_sql_bound @@ -393,6 +432,74 @@ impl TaskRepo { .map_err(storage_err)? } + /// 查某父任务的全部子任务(parent_id = ?,deleted_at IS NULL),按创建时间升序。 + /// + /// 知识图谱 Phase 1 V29(对标设计 §2.1 父任务聚合规则):父任务=容器模型,status + /// 不走状态机,由子任务聚合计算。本方法取子任务列表供聚合规则消费。 + /// + /// 注:嵌套深度限制 1 级(无孙任务)由 IPC 层校验不进 DB 约束,本方法只做等值查询。 + pub async fn get_children(&self, parent_id: &str) -> Result> { + let conn = self.conn.clone(); + let pid = parent_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, created_at, updated_at FROM tasks WHERE deleted_at IS NULL AND parent_id = ?1 ORDER BY created_at ASC") + .map_err(storage_err)?; + let rows = stmt + .query_map(params![pid], |row| task_from_row(row)) + .map_err(storage_err)?; + let mut results = Vec::new(); + for r in rows { + results.push(r.map_err(storage_err)?); + } + Ok(results) + }) + .await + .map_err(storage_err)? + } + + /// 父任务聚合:按 status 分组统计子任务计数(对标设计 §2.1 聚合规则)。 + /// + /// 父任务 status 由子任务聚合计算(不走状态机),聚合规则: + /// - 全 todo → 父 todo + /// - 任一 in_progress → 父 in_progress + /// - 任一 blocked → 父 blocked + /// - 全 done/cancelled → 父 done + /// + /// 本方法返回 `Vec<(status, count)>`(SQL GROUP BY 一次查询,数据量小 ~50 无压力), + /// 聚合规则的具体判定由调用方(commands 层)实现 —— 本层只提供原始计数,不持有 + /// 业务聚合逻辑(CRUD 层只懂表/列语义,对标 B-260616-16 跨表校验下沉思路)。 + pub async fn count_children_by_status( + &self, + parent_id: &str, + ) -> Result> { + let conn = self.conn.clone(); + let pid = parent_id.to_owned(); + tokio::task::spawn_blocking(move || { + let guard = conn.blocking_lock(); + let mut stmt = guard + .prepare( + "SELECT status, COUNT(*) AS cnt FROM tasks \ + WHERE deleted_at IS NULL AND parent_id = ?1 \ + GROUP BY status", + ) + .map_err(storage_err)?; + let rows = stmt + .query_map(params![pid], |row| { + Ok((row.get::<_, String>(0)?, row.get::<_, i64>(1)?)) + }) + .map_err(storage_err)?; + let mut results = Vec::new(); + for r in rows { + results.push(r.map_err(storage_err)?); + } + Ok(results) + }) + .await + .map_err(storage_err)? + } + /// 列出回收站(deleted_at IS NOT NULL),按更新时间(≈删除时间)降序。对标 ProjectRepo::list_deleted。 /// /// 注:按项目列活跃任务走 list_active_by_project(SQL 下推 project_id), @@ -402,7 +509,7 @@ impl TaskRepo { 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, created_at, updated_at FROM tasks WHERE deleted_at IS NOT NULL ORDER BY updated_at DESC") + .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, created_at, updated_at FROM tasks WHERE deleted_at IS NOT NULL ORDER BY updated_at DESC") .map_err(storage_err)?; let rows = stmt .query_map([], |row| task_from_row(row)) @@ -417,3 +524,215 @@ impl TaskRepo { .map_err(storage_err)? } } + +// ============================================================ +// 单元测试 — 知识图谱 Phase 1:queue/parent_id 筛选 + get_children(内存 DB) +// ============================================================ + +#[cfg(test)] +mod tests { + use super::*; + use crate::crud::ProjectRepo; + use crate::models::ProjectRecord; + + /// 构造一条 TaskRecord fixture(queue/parent_id/status 可定制,V29 新维度 + 聚合测试用 status)。 + fn trec(id: &str, queue: &str, parent_id: Option<&str>) -> TaskRecord { + trec_full(id, queue, parent_id, "todo") + } + + /// 全参 fixture(聚合测试需自定义 status 时用)。 + fn trec_full(id: &str, queue: &str, parent_id: Option<&str>, status: &str) -> TaskRecord { + TaskRecord { + id: id.to_string(), + project_id: "proj-1".to_string(), + title: format!("task-{id}"), + description: String::new(), + status: status.to_string(), + priority: 1, + branch_name: None, + assignee: None, + workflow_def_id: None, + base_branch: None, + review_rounds: 0, + output_json: None, + idea_id: None, + queue: queue.to_string(), + parent_id: parent_id.map(|s| s.to_string()), + content_json: None, + created_at: "1700000000000".to_string(), + updated_at: "1700000000000".to_string(), + } + } + + /// 构造内存 DB + 占位 project(满足 tasks.project_id FK,PRAGMA foreign_keys=ON)。 + async fn setup() -> TaskRepo { + let db = Database::open_in_memory().await.expect("open_in_memory"); + let project_repo = ProjectRepo::new(&db); + project_repo + .insert(ProjectRecord { + id: "proj-1".to_string(), + name: "proj-1".to_string(), + description: String::new(), + status: "active".to_string(), + idea_id: None, + path: None, + stack: None, + created_at: "1700000000000".to_string(), + updated_at: "1700000000000".to_string(), + }) + .await + .unwrap(); + TaskRepo::new(&db) + } + + #[tokio::test] + async fn list_by_query_queue_filter() { + let repo = setup().await; + repo.insert(trec("t1", "backlog", None)).await.unwrap(); + repo.insert(trec("t2", "todo", None)).await.unwrap(); + repo.insert(trec("t3", "todo", None)).await.unwrap(); + repo.insert(trec("t4", "done", None)).await.unwrap(); + + // queue=todo → 只返回 t2/t3 + let q = TaskQuery { + queue: Some("todo".to_string()), + ..Default::default() + }; + let res = repo.list_by_query(&q).await.unwrap(); + let ids: Vec<_> = res.iter().map(|r| r.id.as_str()).collect(); + assert_eq!(ids.len(), 2); + assert!(ids.contains(&"t2")); + assert!(ids.contains(&"t3")); + + // queue=backlog → 只返回 t1 + let q = TaskQuery { + queue: Some("backlog".to_string()), + ..Default::default() + }; + let res = repo.list_by_query(&q).await.unwrap(); + let ids: Vec<_> = res.iter().map(|r| r.id.as_str()).collect(); + assert_eq!(ids, vec!["t1"]); + } + + #[tokio::test] + async fn list_by_query_parent_id_filter() { + let repo = setup().await; + // parent 父任务(叶子,parent_id=None)+ 3 子任务(parent_id=parent) + repo.insert(trec("parent", "todo", None)).await.unwrap(); + repo.insert(trec("c1", "todo", Some("parent"))).await.unwrap(); + repo.insert(trec("c2", "todo", Some("parent"))).await.unwrap(); + repo.insert(trec("orphan", "todo", None)).await.unwrap(); + + // parent_id=parent → 只返回 c1/c2(父任务自身不匹配) + let q = TaskQuery { + parent_id: Some("parent".to_string()), + ..Default::default() + }; + let res = repo.list_by_query(&q).await.unwrap(); + let ids: Vec<_> = res.iter().map(|r| r.id.as_str()).collect(); + assert_eq!(ids.len(), 2); + assert!(ids.contains(&"c1")); + assert!(ids.contains(&"c2")); + } + + #[tokio::test] + async fn list_by_query_queue_and_parent_id_combined() { + let repo = setup().await; + repo.insert(trec("parent", "todo", None)).await.unwrap(); + repo.insert(trec("c1", "todo", Some("parent"))).await.unwrap(); + repo.insert(trec("c2", "backlog", Some("parent"))).await.unwrap(); + + // queue=todo AND parent_id=parent → 只 c1(c2 是 backlog) + let q = TaskQuery { + queue: Some("todo".to_string()), + parent_id: Some("parent".to_string()), + ..Default::default() + }; + let res = repo.list_by_query(&q).await.unwrap(); + let ids: Vec<_> = res.iter().map(|r| r.id.as_str()).collect(); + assert_eq!(ids, vec!["c1"]); + } + + #[tokio::test] + async fn list_by_query_empty_returns_all_active() { + // 空 query(全 None)→ 等价 list_active 全量未删,向后兼容 + let repo = setup().await; + repo.insert(trec("t1", "todo", None)).await.unwrap(); + repo.insert(trec("t2", "backlog", None)).await.unwrap(); + + let res = repo.list_by_query(&TaskQuery::default()).await.unwrap(); + assert_eq!(res.len(), 2); + } + + #[tokio::test] + async fn get_children_returns_only_direct_children() { + let repo = setup().await; + repo.insert(trec("parent", "todo", None)).await.unwrap(); + repo.insert(trec("c1", "todo", Some("parent"))).await.unwrap(); + repo.insert(trec("c2", "in_progress", Some("parent"))).await.unwrap(); + repo.insert(trec("orphan", "todo", None)).await.unwrap(); + + let children = repo.get_children("parent").await.unwrap(); + let ids: Vec<_> = children.iter().map(|r| r.id.as_str()).collect(); + assert_eq!(ids.len(), 2); + assert!(ids.contains(&"c1")); + assert!(ids.contains(&"c2")); + } + + #[tokio::test] + async fn get_children_empty_when_no_children() { + let repo = setup().await; + repo.insert(trec("parent", "todo", None)).await.unwrap(); + let children = repo.get_children("parent").await.unwrap(); + assert!(children.is_empty()); + } + + #[tokio::test] + async fn count_children_by_status_groups() { + let repo = setup().await; + repo.insert(trec("parent", "todo", None)).await.unwrap(); + // 4 个子任务:status 分布 todo×2 / in_progress×1 / done×1(queue 统一 todo) + repo.insert(trec_full("c1", "todo", Some("parent"), "todo")) + .await + .unwrap(); + repo.insert(trec_full("c2", "todo", Some("parent"), "todo")) + .await + .unwrap(); + repo.insert(trec_full("c3", "todo", Some("parent"), "in_progress")) + .await + .unwrap(); + repo.insert(trec_full("c4", "todo", Some("parent"), "done")) + .await + .unwrap(); + + let counts = repo.count_children_by_status("parent").await.unwrap(); + // 转 map 便于断言(顺序由 GROUP BY 决定,不依赖)。父任务自身不计入(parent_id 非自身)。 + let map: std::collections::HashMap = counts.into_iter().collect(); + assert_eq!(map.get("todo"), Some(&2)); + assert_eq!(map.get("in_progress"), Some(&1)); + assert_eq!(map.get("done"), Some(&1)); + assert!(!map.contains_key("cancelled")); + } + + #[tokio::test] + async fn count_children_by_status_empty_when_no_children() { + let repo = setup().await; + repo.insert(trec("parent", "todo", None)).await.unwrap(); + let counts = repo.count_children_by_status("parent").await.unwrap(); + assert!(counts.is_empty()); + } + + #[tokio::test] + async fn get_children_excludes_soft_deleted() { + // 软删子任务不进 get_children 结果(deleted_at IS NULL 过滤) + let repo = setup().await; + repo.insert(trec("parent", "todo", None)).await.unwrap(); + repo.insert(trec("c1", "todo", Some("parent"))).await.unwrap(); + repo.insert(trec("c2", "todo", Some("parent"))).await.unwrap(); + repo.soft_delete("c2").await.unwrap(); + + let children = repo.get_children("parent").await.unwrap(); + let ids: Vec<_> = children.iter().map(|r| r.id.as_str()).collect(); + assert_eq!(ids, vec!["c1"], "软删子任务应被过滤"); + } +} diff --git a/crates/df-storage/src/migrations.rs b/crates/df-storage/src/migrations.rs index a5361ca..d927725 100644 --- a/crates/df-storage/src/migrations.rs +++ b/crates/df-storage/src/migrations.rs @@ -33,7 +33,11 @@ pub fn run(conn: &Connection) -> Result<()> { // 对齐 DTO 契约 audit/mod.rs:53 只列 completed + 前端 i18n auditLog.status 无 executed 键 + // 治 AuditLog executed 记录显示错位蓝pending标签+raw"executed")。 // V28 = 灵感软删回收站(ideas.deleted_at,对标 tasks.deleted_at V14/projects.deleted_at V11)。 - let steps: [(i32, fn(&Connection) -> Result<()>); 28] = [ + // V29 = 知识图谱 Phase 1 任务网络基础(对标 docs/02-架构设计/专项设计/ + // 项目知识图谱与任务队列系统-2026-06-26.md §2.1/§2.2):tasks 加 queue(管理池)/ + // parent_id(父任务纵向关联)/content_json(结构化需求规格)三列 + task_links 表 + // (横向关联 depends_on/blocks/relates_to),为 AI 编排地基打底。 + let steps: [(i32, fn(&Connection) -> Result<()>); 29] = [ (1, migrate_v1), (2, migrate_v2), (3, migrate_v3), @@ -62,6 +66,7 @@ pub fn run(conn: &Connection) -> Result<()> { (26, migrate_v26), (27, migrate_v27), (28, migrate_v28), + (29, migrate_v29), ]; for (version, migrate_fn) in steps { @@ -698,6 +703,76 @@ fn migrate_v28(conn: &Connection) -> Result<()> { Ok(()) } +/// V29:知识图谱 Phase 1 任务网络基础 — 幂等补 tasks.queue / parent_id / content_json 三列 +/// +/// 对标 docs/02-架构设计/专项设计/项目知识图谱与任务队列系统-2026-06-26.md §2.1, +/// 为「AI 拥有完整项目知识图谱、自主分解任务编排依赖」打数据层地基。三列各司其职: +/// +/// - `queue TEXT NOT NULL DEFAULT 'todo'`:管理维度池标记,与 status(执行维度)正交。 +/// 取值 backlog(需求池)/todo(待办池)/decision(待决策池)/active(执行中)/done(已完成)。 +/// DEFAULT 'todo' 保证老任务迁移后取值确定(非 NULL),向后兼容:历史任务原 status=todo, +/// 落 todo 池语义一致。queue 与 status 一致性约束由 IPC 层校验(不进状态机,不进 DB 约束)。 +/// +/// - `parent_id TEXT REFERENCES tasks(id)`:父任务纵向关联(AI 分解-执行编排结构)。 +/// NULL = 叶子任务(走状态机 advance_task);非空 = 子任务。 +/// 限制 1 级嵌套(无孙任务)由 IPC 层校验,不进 DB 约束。父任务=容器模型,status 由 +/// 子任务聚合计算(不走状态机)。TEXT NULL 向后兼容(老任务无 parent → None)。 +/// +/// - `content_json TEXT`:结构化需求规格 JSON 字符串(AI 可读写的执行规格)。 +/// 结构 { background, acceptance_criteria[], scope[], technical_design, custom_fields }。 +/// AI 从对话提取填充,执行中用 acceptance_criteria 自检。NULL = 无结构化规格(纯文本 description)。 +/// TEXT NULL 向后兼容(老任务无 content_json → None)。 +/// +/// 对标 v14(v14 tasks.deleted_at)/ v24(v24 ideas.related_ids)幂等模式:每列用 PRAGMA +/// 探测存在性,缺失才 ALTER,对新库/老库/坏库均安全(列已存在时跳过 ALTER 不报 duplicate column)。 +fn migrate_v29(conn: &Connection) -> Result<()> { + if !column_exists(conn, "tasks", "queue") { + conn.execute( + "ALTER TABLE tasks ADD COLUMN queue TEXT NOT NULL DEFAULT 'todo'", + [], + )?; + tracing::info!("v29: 补建 tasks.queue 列(管理池,知识图谱 Phase 1)"); + } + if !column_exists(conn, "tasks", "parent_id") { + conn.execute( + "ALTER TABLE tasks ADD COLUMN parent_id TEXT REFERENCES tasks(id)", + [], + )?; + tracing::info!("v29: 补建 tasks.parent_id 列(父任务纵向关联,知识图谱 Phase 1)"); + } + if !column_exists(conn, "tasks", "content_json") { + conn.execute("ALTER TABLE tasks ADD COLUMN content_json TEXT", [])?; + tracing::info!("v29: 补建 tasks.content_json 列(结构化需求规格,知识图谱 Phase 1)"); + } + // task_links 表(横向关联,对标设计 §2.2):depends_on/blocks/relates_to。 + // CREATE TABLE IF NOT EXISTS 幂等:新库建、老库(V29 之前的库已跑过前半三列)已有则跳过。 + // 循环依赖检测走应用层(TaskLinkRepo::create_link BFS),非 DB 约束(对标设计 D8)。 + // 软删除语义:Task 软删不级联删 link(恢复后关系还在),故无 ON DELETE,FK 仅引用完整性。 + conn.execute( + "CREATE TABLE IF NOT EXISTS task_links ( + id TEXT PRIMARY KEY, + source_id TEXT NOT NULL REFERENCES tasks(id), + target_id TEXT NOT NULL REFERENCES tasks(id), + link_type TEXT NOT NULL, + remark TEXT, + created_at TEXT NOT NULL + )", + [], + )?; + conn.execute( + "CREATE INDEX IF NOT EXISTS idx_task_links_source ON task_links(source_id)", + [], + )?; + conn.execute( + "CREATE INDEX IF NOT EXISTS idx_task_links_target ON task_links(target_id)", + [], + )?; + tracing::info!("v29: 建 task_links 表 + 索引(任务横向关联,知识图谱 Phase 1)"); + conn.execute("INSERT INTO schema_version (version) VALUES (?)", [29])?; + tracing::info!("迁移 v29 完成"); + Ok(()) +} + /// V21 建表 SQL — 消息拆分存储 ai_messages 表 /// /// 与 V9_SQL 中的 ai_messages 镜像(V9 给新库,此 const 给老库 V21 迁移用 IF NOT EXISTS)。 diff --git a/crates/df-storage/src/models.rs b/crates/df-storage/src/models.rs index 91d8c4f..924e4c4 100644 --- a/crates/df-storage/src/models.rs +++ b/crates/df-storage/src/models.rs @@ -98,10 +98,54 @@ pub struct TaskRecord { /// 老任务无 idea_id → None。#[serde(default)] 兼容旧前端 JSON(无该字段时为 None)。 #[serde(default)] pub idea_id: Option, + /// 管理维度池标记(知识图谱 Phase 1,V29 列)。与 status(执行维度)正交。 + /// 取值 backlog(需求池)/todo(待办池)/decision(待决策池)/active(执行中)/done(已完成)。 + /// DEFAULT 'todo' 保证老任务迁移后取值确定(非 NULL),TaskRecord 字段为 String(非 Option)。 + /// #[serde(default = "default_queue")] 兼容旧前端 JSON(无该字段时为 "todo")。 + #[serde(default = "default_queue")] + pub queue: String, + /// 父任务 ID(知识图谱 Phase 1,V29 列,父任务纵向关联)。 + /// NULL = 叶子任务(走状态机 advance_task);非空 = 子任务(限制 1 级嵌套,无孙任务, + /// 由 IPC 层校验不进 DB 约束)。父任务=容器模型,status 由子任务聚合计算(不走状态机)。 + /// TEXT NULL 向后兼容,TaskRecord 字段为 Option(老任务无 parent → None)。 + /// #[serde(default)] 兼容旧前端 JSON(无该字段时为 None)。 + #[serde(default)] + pub parent_id: Option, + /// 结构化需求规格 JSON 字符串(知识图谱 Phase 1,V29 列)。 + /// 结构 { background, acceptance_criteria[], scope[], technical_design, custom_fields }。 + /// AI 从对话提取填充,执行中用 acceptance_criteria 自检。NULL = 无结构化规格(纯文本 description)。 + /// TEXT NULL 向后兼容,TaskRecord 字段为 Option。 + /// #[serde(default, skip_serializing_if = "Option::is_none")] 兼容旧 JSON + 无值不序列化。 + #[serde(default, skip_serializing_if = "Option::is_none")] + pub content_json: Option, pub created_at: String, pub updated_at: String, } +/// 任务横向关联记录(task_links 表,知识图谱 Phase 1 V29,对标设计 §2.2)。 +/// +/// 表达任务间依赖/阻塞/弱关联关系,AI 拓扑排序编排调度的基础。独立表方案替代 JSON 列 +/// (Idea related_ids 的缺陷:无类型区分/单向/无法高效反向查询)。 +/// +/// - `source_id` / `target_id`:关联两端任务 id(FK tasks.id)。软删除语义:Task 软删不 +/// 级联删 link(对标设计 §2.2 边界「软删除保留」——恢复后关系还在),故无 ON DELETE。 +/// - `link_type`:关联类型白名单(应用层校验,非 DB 约束): +/// - `depends_on`:source 依赖 target(target 完成后 source 才能开始)→ 拓扑排序调度 +/// - `blocks`:source 阻塞 target(source 不完成则 target 无法推进)→ 依赖的反向声明 +/// - `relates_to`:弱关联,无执行约束 → 上下文提示 +/// - `remark`:可选备注。 +/// - 循环依赖(A→B→A)在 `TaskLinkRepo::create_link` 应用层 BFS 检测拒绝(非 DB 约束, +/// 对标设计 D8:task_links 数据量小,检测成本低)。跨项目依赖允许。 +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct TaskLinkRecord { + pub id: String, + pub source_id: String, + pub target_id: String, + pub link_type: String, + pub remark: Option, + pub created_at: String, +} + /// 分支记录 #[derive(Debug, Clone, Serialize, Deserialize)] pub struct BranchRecord { @@ -205,6 +249,12 @@ fn default_weight() -> u32 { 50 } +/// TaskRecord.queue 的 serde 默认("todo",知识图谱 Phase 1 V29)。老库/缺字段 JSON → "todo"。 +/// 对齐 DB 列 DEFAULT 'todo' 语义,保证字段始终非空 String。 +fn default_queue() -> String { + "todo".to_string() +} + /// AI 对话记录 #[derive(Debug, Clone, Serialize, Deserialize)] pub struct AiConversationRecord { diff --git a/crates/df-storage/tests/project_soft_delete.rs b/crates/df-storage/tests/project_soft_delete.rs index 931cb00..55e8558 100644 --- a/crates/df-storage/tests/project_soft_delete.rs +++ b/crates/df-storage/tests/project_soft_delete.rs @@ -44,6 +44,9 @@ fn task(id: &str, project_id: &str) -> TaskRecord { review_rounds: 0, output_json: None, idea_id: None, + queue: "todo".to_string(), + parent_id: None, + content_json: None, created_at: now_ts(), updated_at: now_ts(), } diff --git a/docs/todo.md b/docs/todo.md index 2f32ad8..51bbf59 100644 --- a/docs/todo.md +++ b/docs/todo.md @@ -37,7 +37,7 @@ graph TD G1["G1 目标钉扎
✅代码 8ce18cb / 🔨待实测"]:::doing P1["父① 小bug攒批
✅①.2/①.3(①.1过时·①.4归⑤)"]:::done - P2["父② 知识图谱Phase1
📋待办"]:::todo + P2["父② 知识图谱Phase1
🔨数据层✅·业务层待办"]:::doing P3["父③ AI对话体验
📋待办"]:::todo P4["父④ F-09 per-conv
📋待办"]:::todo P5["父⑤ 灵感模块
🔨⑤.1/①.4✅·⑤.2待办"]:::doing @@ -55,7 +55,7 @@ graph TD |---|---|---|---| | **G1** 目标钉扎 | 🔨 代码✅`8ce18cb` 待实测 | G1/G2/G4 落地 | — | | **父①** 小bug攒批 | ✅ 完成 | ①.2 白名单✅(settings.rs) / ①.3 priority✅(idea.rs) / ①.1 BUG层1❌过时(F-260619-03 方案①取代,层2待决策) / ①.4 雷达图→归父⑤ | — | -| **父②** 知识图谱Phase1 | 📋 待办(✅DEC-01 b) | ②.1 **V29**迁移(queue/parent_id/content_json/task_links;V28已被灵感软删除占用) / ②.2 TaskRecord+TaskLinkRepo / ②.3 IPC+move_queue / ②.4 父聚合 / ②.5 AI工具12 | G1(弱) | +| **父②** 知识图谱Phase1 | 🔨 数据层✅ | ②.1 V29迁移✅(tasks+queue/parent_id/content_json+task_links表) / ②.2 TaskRecord+TaskLinkRepo✅(96lib+11集成测试,BFS环检测) / ②.3 IPC+move_queue 待办 / ②.4 父聚合 待办 / ②.5 AI工具12 待办 | G1(弱) | | **父③** AI对话体验 | 📋 待办 | ③.1 B-260619-04 ToolCard / ③.2 REFACTOR-260619-04 审批状态机 | — | | **父④** F-09 per-conv | 📋 待办 | ④.1 streaming/currentText per-conv | — | | **父⑤** 灵感模块 | 🔨 进行中 | ⑤.1 软删除✅ / ①.4 雷达图✅ / ⑤.2 #05 record_to_idea✅ / #06 配置化·#09·#10 表单待办 / #07 promote补偿待定(soft_delete 污染回收站 vs purge 内部回滚干净?) | ⑤.2#07→②.1 | diff --git a/src-tauri/src/commands/ai/tool_registry.rs b/src-tauri/src/commands/ai/tool_registry.rs index cba334a..48d1470 100644 --- a/src-tauri/src/commands/ai/tool_registry.rs +++ b/src-tauri/src/commands/ai/tool_registry.rs @@ -712,6 +712,12 @@ fn register_task_tools(registry: &mut AiToolRegistry, db: &Arc) { output_json: None, // F-260619-01 可选关联灵感(空字符串/缺省视为不关联) idea_id: args.get("idea_id").and_then(|v| v.as_str()).filter(|s| !s.is_empty()).map(String::from), + // 知识图谱 Phase 1 V29 三列:本批仅数据层迁移,create_task 参数扩展为 Phase 1 + // 后续 AI 工具任务。此处默认值(queue='todo'/parent_id=None/content_json=None) + // 与 DB 列 DEFAULT 及 commands::task::create_task 一致。 + queue: "todo".to_string(), + parent_id: None, + content_json: None, created_at: now_millis(), updated_at: now_millis(), }; let id = record.id.clone(); diff --git a/src-tauri/src/commands/task.rs b/src-tauri/src/commands/task.rs index ac3d706..6d3e77d 100644 --- a/src-tauri/src/commands/task.rs +++ b/src-tauri/src/commands/task.rs @@ -113,6 +113,12 @@ pub async fn create_task( output_json: None, // F-260619-01:空字符串视为不关联(与 AI 工具层一致) idea_id: input.idea_id.filter(|s| !s.is_empty()), + // 知识图谱 Phase 1 V29 三列:本批仅数据层迁移,create_task 参数扩展(queue/parent_id/ + // content_json 可选入参)为 Phase 1 后续 IPC 任务。此处默认值:queue='todo'(待办池)、 + // parent_id=None(叶子任务)、content_json=None(无结构化规格),与 DB 列 DEFAULT 一致。 + queue: "todo".to_string(), + parent_id: None, + content_json: None, created_at: now.clone(), updated_at: now, };