lxy/DevFlow
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

新增: 知识图谱Phase1数据层(父②) + 修复 idea list_by_query 占位符(父⑤⑤.1)

Browse Source 
父② 数据层(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 集成测试全过。
This commit is contained in:
绝尘
2026-06-26 22:50:48 +08:00
parent 549f198cea
commit df8ed7e74f
13 changed files with 1082 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
crates/df-mcp/src/tools.rs
View File

@@ -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,
};

3
crates/df-nodes/src/ai_self_review_node.rs
View File

@@ -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(),
})

3
crates/df-nodes/src/task_advance_node.rs
View File

@@ -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(),
}

8
crates/df-storage/src/crud/idea_repo.rs
View File

@@ -278,12 +278,16 @@ impl IdeaRepo {
let mut params_vec: Vec<Box<dyn rusqlite::ToSql>> = 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()));

4
crates/df-storage/src/crud/mod.rs
View File

@@ -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);

580
crates/df-storage/src/crud/task_link_repo.rs Normal file
View File

@@ -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<TaskLinkRecord, rusqlite::Error> {
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<tokio::sync::Mutex<rusqlite::Connection>>,
}
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<String> {
// 应用层校验先行(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<bool> {
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<Option<TaskLinkRecord>> {
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<Vec<TaskLinkRecord>> {
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<Vec<TaskLinkRecord>> {
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<Vec<TaskLinkRecord>> {
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<bool> {
use std::collections::VecDeque;
let mut visited: HashSet<String> = HashSet::new();
let mut queue: VecDeque<String> = 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<String> = 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);
}
}

355
crates/df-storage/src/crud/task_repo.rs
View File

@@ -33,6 +33,10 @@ fn task_from_row(row: &Row<'_>) -> std::result::Result<TaskRecord, rusqlite::Err
review_rounds: row.get("review_rounds")?,
output_json: row.get("output_json")?,
idea_id: row.get("idea_id")?,
// 知识图谱 Phase 1 V29 三列(queue/parent_id/content_json),15→18 列同步之一。
queue: row.get("queue")?,
parent_id: row.get("parent_id")?,
content_json: row.get("content_json")?,
created_at: row.get("created_at")?,
updated_at: row.get("updated_at")?,
})
@@ -54,6 +58,10 @@ fn task_from_row(row: &Row<'_>) -> std::result::Result<TaskRecord, rusqlite::Err
///
/// 当前仅 status(P1 下沉)+ keyword(P2 LIKE)在视图链路使用;project_id/priority/assignee/
/// order_by/limit/offset 为基建就绪(视图暂不用,P3 排序分页待数据量增长)。
///
/// 知识图谱 Phase 1 V29(对标设计 §2.1)新增 queue / parent_id 两个筛选维度:
/// - `queue`:管理池筛选(backlog/todo/decision/active/done),看板视图按池分列的数据源;
/// - `parent_id`:父任务筛选(Some(id)=某父的子任务;特殊语义 None 仅叶子 vs 全部 由调用方拼)。
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct TaskQuery {
/// 项目 ID 过滤(SQL 下推,命中 idx_tasks_project_id)
@@ -66,6 +74,14 @@ pub struct TaskQuery {
pub assignee: Option<String>,
/// 关键词搜索(P2):title/description LIKE %kw%,对齐知识库 search 的 LIKE 模式。
pub keyword: Option<String>,
/// 管理池过滤(知识图谱 Phase 1 V29):backlog/todo/decision/active/done。
/// 看板视图按池分列的数据源。值合法性由上层 queue 语义校验兜底(非法值 DB 无匹配返回空)。
#[serde(default)]
pub queue: Option<String>,
/// 父任务 ID 过滤(知识图谱 Phase 1 V29):Some(id) = 查某父任务的子任务;
/// 查叶子任务(parent_id IS NULL)由专用方法 get_children 之外的语义决定,本字段只做等值匹配。
#[serde(default)]
pub parent_id: Option<String>,
/// 排序字段(白名单 created_at/updated_at/priority/status,降序)。P3 基建就绪。
pub order_by: Option<String>,
/// 分页上限(钳制 ≤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<Vec<TaskRecord>> {
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<String> = vec!["deleted_at IS NULL".to_string()];
let mut params_vec: Vec<Box<dyn rusqlite::ToSql>> = 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<Vec<TaskRecord>> {
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<Vec<(String, i64)>> {
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<String, i64> = 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"], "软删子任务应被过滤");
}
}

77
crates/df-storage/src/migrations.rs
View File

@@ -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)。

50
crates/df-storage/src/models.rs
View File

@@ -98,10 +98,54 @@ pub struct TaskRecord {
/// 老任务无 idea_id → None。#[serde(default)] 兼容旧前端 JSON(无该字段时为 None)。
#[serde(default)]
pub idea_id: Option<String>,
/// 管理维度池标记(知识图谱 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<String>(老任务无 parent → None)。
/// #[serde(default)] 兼容旧前端 JSON(无该字段时为 None)。
#[serde(default)]
pub parent_id: Option<String>,
/// 结构化需求规格 JSON 字符串(知识图谱 Phase 1,V29 列)。
/// 结构 { background, acceptance_criteria[], scope[], technical_design, custom_fields }。
/// AI 从对话提取填充,执行中用 acceptance_criteria 自检。NULL = 无结构化规格(纯文本 description)。
/// TEXT NULL 向后兼容,TaskRecord 字段为 Option<String>。
/// #[serde(default, skip_serializing_if = "Option::is_none")] 兼容旧 JSON + 无值不序列化。
#[serde(default, skip_serializing_if = "Option::is_none")]
pub content_json: Option<String>,
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<String>,
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 {

3
crates/df-storage/tests/project_soft_delete.rs
View File

@@ -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(),
}