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

新增: 任务灵感归档后端

Browse Source
This commit is contained in:
lxy
2026-10-05 00:35:01 +08:00
parent c8c13d302a
commit f4d31e884a
17 changed files with 354 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/df-storage/src/crud/idea_repo.rs
+89 -13
View File
@@ -44,11 +44,12 @@ const KNOWLEDGE_COLS_WITH_EMBEDDING: &str = concat!(
/// 定义 + 配合 `IDEA_COL_COUNT` 断言 + 测试 `test_idea_cols_matches_record`,加列漏改即捕获。
///
/// V24 新增 related_ids 列(灵感间关联关系持久化打底),已纳入列清单 + 计数。
/// V47 新增 archived 列(灵感归档,归档功能 P1),已纳入列清单 + 计数(14→15)。
#[cfg_attr(not(test), allow(dead_code))]
const IDEA_COLS: &str = "id,title,description,status,priority,score,tags,source,promoted_to,ai_analysis,scores,related_ids,created_at,updated_at";
const IDEA_COLS: &str = "id,title,description,status,priority,score,tags,source,promoted_to,ai_analysis,scores,related_ids,archived,created_at,updated_at";
/// `IdeaRecord` 字段数(与上面列清单的逗号分隔项数一致,被测试断言)。
#[cfg_attr(not(test), allow(dead_code))]
const IDEA_COL_COUNT: usize = 14;
const IDEA_COL_COUNT: usize = 15;
// ============================================================
// 向量工具 — embedding BLOB 序列化 + 余弦相似度
@@ -123,6 +124,8 @@ fn idea_from_row(row: &Row<'_>) -> std::result::Result<IdeaRecord, rusqlite::Err
ai_analysis: row.get("ai_analysis")?,
scores: row.get("scores")?,
related_ids: row.get("related_ids")?,
// 归档功能 P1 V47:15 列同步之一(SQLite 无真 BOOLEAN,0→false/非0→true)。
archived: row.get::<_, i32>("archived")? != 0,
created_at: row.get("created_at")?,
updated_at: row.get("updated_at")?,
})
@@ -218,22 +221,22 @@ impl_repo!(
from_row => |row| idea_from_row(row),
insert => |conn, rec| {
conn.execute(
"INSERT INTO ideas (id, title, description, status, priority, score, tags, source, promoted_to, ai_analysis, scores, related_ids, created_at, updated_at)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14)",
"INSERT INTO ideas (id, title, description, status, priority, score, tags, source, promoted_to, ai_analysis, scores, related_ids, archived, created_at, updated_at)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15)",
params![
rec.id, rec.title, rec.description, rec.status.as_str(), rec.priority,
rec.score, rec.tags, rec.source, rec.promoted_to, rec.ai_analysis,
rec.scores, rec.related_ids, rec.created_at, rec.updated_at
rec.scores, rec.related_ids, rec.archived as i32, rec.created_at, rec.updated_at
],
)
},
update => |conn, rec| {
conn.execute(
"UPDATE ideas SET title = ?1, description = ?2, status = ?3, priority = ?4, score = ?5, tags = ?6, source = ?7, promoted_to = ?8, ai_analysis = ?9, scores = ?10, related_ids = ?11, updated_at = ?12 WHERE id = ?13",
"UPDATE ideas SET title = ?1, description = ?2, status = ?3, priority = ?4, score = ?5, tags = ?6, source = ?7, promoted_to = ?8, ai_analysis = ?9, scores = ?10, related_ids = ?11, archived = ?12, updated_at = ?13 WHERE id = ?14",
params![
rec.title, rec.description, rec.status.as_str(), rec.priority,
rec.score, rec.tags, rec.source, rec.promoted_to, rec.ai_analysis,
rec.scores, rec.related_ids, rec.updated_at, rec.id
rec.scores, rec.related_ids, rec.archived as i32, rec.updated_at, rec.id
],
)
}
@@ -306,7 +309,7 @@ impl IdeaRepo {
// 累积 WHERE 子句 + 收集参数(按出现顺序绑定占位符)。
// deleted_at IS NULL 恒带(常量条件无占位符),回收站任务不进结果
// (对标 task_repo list_by_query 同款模式,防回收站泄漏,不可被 query 关闭)。
let mut where_clauses: Vec<String> = vec!["deleted_at IS NULL".to_string()];
let mut where_clauses: Vec<String> = vec!["deleted_at IS NULL".to_string(), "archived = 0".to_string()];
let mut params_vec: Vec<Box<dyn rusqlite::ToSql>> = Vec::new();
if let Some(s) = &status {
@@ -428,7 +431,7 @@ impl IdeaRepo {
validate_column_name(field, "ideas")?;
let conn = self.conn.clone();
let sql = format!(
"UPDATE ideas SET {} = ?1, updated_at = ?2 WHERE id = ?3 AND deleted_at IS NULL",
"UPDATE ideas SET {} = ?1, updated_at = ?2 WHERE id = ?3 AND deleted_at IS NULL AND archived = 0",
field
);
let id = id.to_owned();
@@ -445,6 +448,54 @@ impl IdeaRepo {
.map_err(storage_err)?
}
/// 归档/取消归档(归档功能 P1,V47)。
///
/// 纯元数据标记,仅作用于未软删灵感(回收站优先级高于归档,先恢复再操作)。
/// archived=true 后灵感退出主视图(list_by_query 恒带 archived = 0),仅在归档视图可见。
/// 对标 TaskRepo::set_archived。返回是否命中。
pub async fn set_archived(&self, id: &str, archived: bool) -> Result<bool> {
let conn = self.conn.clone();
let id = id.to_owned();
let now = now_millis_str();
tokio::task::spawn_blocking(move || {
let guard = conn.blocking_lock();
let affected = guard
.execute(
"UPDATE ideas SET archived = ?1, updated_at = ?2 WHERE id = ?3 AND deleted_at IS NULL",
params![if archived { 1 } else { 0 }, now, id],
)
.map_err(storage_err)?;
Ok(affected > 0)
})
.await
.map_err(storage_err)?
}
/// 列出归档灵感(archived = 1 且未软删),按更新时间降序(最近归档在前)。
///
/// 归档视图数据源,与 list_by_query 对偶(恒带 archived = 1)。
pub async fn list_archived(&self) -> Result<Vec<IdeaRecord>> {
let conn = self.conn.clone();
tokio::task::spawn_blocking(move || {
let guard = conn.blocking_lock();
let mut stmt = guard
.prepare(&format!(
"SELECT {IDEA_COLS} FROM ideas WHERE deleted_at IS NULL AND archived = 1 ORDER BY updated_at DESC"
))
.map_err(storage_err)?;
let rows = stmt
.query_map([], |row| idea_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)?
}
/// 原子「立项认领」:CAS 写回 status=promoted + promoted_to(`WHERE id AND promoted_to IS NULL`)。
///
/// promote_idea 读-改-写竞态的原子关闭。promote_idea 先建项目再回写灵感,
@@ -622,7 +673,8 @@ impl IdeaRepo {
///
/// 不动 GUI 的无条件 [`update_full`](无 expected 语义,保持现状);本方法仅服务
/// 需要乐观锁版本的调用方(df-mcp update_idea / score_idea)。
/// `Ok(false)` = id 不存在、已进回收站(deleted_at IS NULL 守卫,V28 灵感软删)或版本冲突。
/// `Ok(false)` = id 不存在、已进回收站(deleted_at IS NULL 守卫,V28 灵感软删)、
/// 已归档(archived = 0 守卫,V47 灵感归档,须先取消归档)或版本冲突。
pub async fn update_full_cas(
&self,
record: &IdeaRecord,
@@ -635,8 +687,9 @@ impl IdeaRepo {
let guard = conn.blocking_lock();
let affected = guard
.execute(
// deleted_at IS NULL:回收站灵感不可被 CAS 覆盖(同 tasks/projects 同族守卫)
"UPDATE ideas SET title = ?1, description = ?2, status = ?3, priority = ?4, score = ?5, tags = ?6, source = ?7, promoted_to = ?8, ai_analysis = ?9, scores = ?10, related_ids = ?11, updated_at = ?12 WHERE id = ?13 AND updated_at = ?14 AND deleted_at IS NULL",
// deleted_at IS NULL:回收站灵感不可被 CAS 覆盖(同 tasks/projects 同族守卫);
// archived = 0:归档灵感冻结(同 tasks update_full_cas 同族守卫,须先取消归档再改)。
"UPDATE ideas SET title = ?1, description = ?2, status = ?3, priority = ?4, score = ?5, tags = ?6, source = ?7, promoted_to = ?8, ai_analysis = ?9, scores = ?10, related_ids = ?11, updated_at = ?12 WHERE id = ?13 AND updated_at = ?14 AND deleted_at IS NULL AND archived = 0",
params![
rec.title, rec.description, rec.status.as_str(), rec.priority,
rec.score, rec.tags, rec.source, rec.promoted_to, rec.ai_analysis,
@@ -1837,7 +1890,7 @@ mod tests {
// ---------- IdeaRepo 软删回收站(对标 task_repo 软删测试,V28) ----------
/// 构造一条 IdeaRecord fixture(14 字段全填,IdeaRecord 不含 deleted_at —— 纯 SQL 过滤)。
/// 构造一条 IdeaRecord fixture(15 字段全填,IdeaRecord 不含 deleted_at —— 纯 SQL 过滤)。
fn irec(id: &str, title: &str) -> IdeaRecord {
IdeaRecord {
id: id.to_string(),
@@ -1852,6 +1905,7 @@ mod tests {
ai_analysis: None,
scores: None,
related_ids: None,
archived: false,
created_at: "1700000000000".to_string(),
updated_at: "1700000000000".to_string(),
}
@@ -2023,6 +2077,28 @@ mod tests {
assert_eq!(after.updated_at, current.updated_at, "updated_at 不应抖动");
}
#[tokio::test]
async fn idea_update_full_cas_skips_archived() {
// 归档灵感不可被 CAS 命中(archived = 0 守卫,V47 归档,对标 tasks 同族):
// 归档=冻结,须先取消归档(set_archived(false))再改,防 MCP 对归档灵感静默改写。
let repo = setup_idea_repo().await;
repo.insert(irec("i1", "原标题")).await.unwrap();
assert!(repo.set_archived("i1", true).await.unwrap());
// set_archived 刷新了 updated_at,取当前值作 expected(排除版本不匹配干扰)
let current = repo.get_by_id("i1").await.unwrap().unwrap();
let mut rec = current.clone();
rec.title = "归档里改标题".to_string();
rec.updated_at = "1900000000000".to_string();
let ok = repo
.update_full_cas(&rec, &current.updated_at)
.await
.unwrap();
assert!(!ok, "归档灵感 CAS 应拒绝(archived 守卫)");
let after = repo.get_by_id("i1").await.unwrap().unwrap();
assert_eq!(after.title, "原标题", "归档灵感内容不应被改写");
assert_eq!(after.updated_at, current.updated_at, "updated_at 不应抖动");
}
#[tokio::test]
async fn idea_update_full_cas_conflict_when_expected_stale() {
let repo = setup_idea_repo().await;
crates/df-storage/src/crud/task_link_repo.rs
+1
View File
@@ -350,6 +350,7 @@ mod tests {
parent_id: None,
content_json: None,
module_id: None,
archived: false,
created_at: "1700000000000".to_string(),
updated_at: "1700000000000".to_string(),
}
crates/df-storage/src/crud/task_repo.rs
+88 -28
View File
@@ -15,6 +15,20 @@ use crate::models::TaskRecord;
use super::impl_repo;
use super::{now_millis_str, storage_err, validate_column_name};
// ============================================================
// 列清单(防 COLS 漂移,对标 idea_repo.rs IDEA_COLS 先例)
// ============================================================
/// `tasks` 表 20 列(顺序与 TaskRecord 结构体一致,不含 deleted_at)。
///
/// 19→20 列(归档功能 P1 V46 加 archived)。此前散落 8 处内联列名串
/// (list_active/get_by_id_active/advance_status_atomic 回读/list_active_by_project/
/// list_by_query/get_children/move_task_queue/list_deleted),加列须处处同步,
/// 漏一处只在运行时 rusqlite 报错。集中一处 + from_row 按名取列,加列漏改测试期捕获。
pub(crate) const TASK_COLS: &str = "id, project_id, title, description, status, priority, \
branch_name, assignee, workflow_def_id, base_branch, review_rounds, output_json, \
idea_id, queue, parent_id, content_json, module_id, archived, created_at, updated_at";
// ============================================================
// from_row 辅助函数
// ============================================================
@@ -52,6 +66,8 @@ fn task_from_row(row: &Row<'_>) -> std::result::Result<TaskRecord, rusqlite::Err
content_json: row.get("content_json")?,
// 工程系统 V41:任务关联具体工程(module_id),18→19 列同步之一。
module_id: row.get("module_id")?,
// 归档功能 P1 V46:20 列同步之一(SQLite 无真 BOOLEAN,0→false/非0→true)。
archived: row.get::<_, i32>("archived")? != 0,
created_at: row.get("created_at")?,
updated_at: row.get("updated_at")?,
})
@@ -138,26 +154,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, queue, parent_id, content_json, module_id, created_at, updated_at)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17, ?18, ?19)",
"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, module_id, archived, created_at, updated_at)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17, ?18, ?19, ?20)",
params![
rec.id, rec.project_id, rec.title, rec.description, rec.status.as_str(), rec.priority,
rec.branch_name, rec.assignee, rec.workflow_def_id, rec.base_branch,
rec.review_rounds, rec.output_json, rec.idea_id,
rec.queue, rec.parent_id, rec.content_json, rec.module_id,
rec.created_at, rec.updated_at
rec.archived as i32, 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, queue = ?13, parent_id = ?14, content_json = ?15, module_id = ?16, updated_at = ?17 WHERE id = ?18",
"UPDATE tasks SET project_id = ?1, title = ?2, description = ?3, status = ?4, priority = ?5, branch_name = ?6, assignee = ?7, workflow_def_id = ?8, base_branch = ?9, review_rounds = ?10, output_json = ?11, idea_id = ?12, queue = ?13, parent_id = ?14, content_json = ?15, module_id = ?16, archived = ?17, updated_at = ?18 WHERE id = ?19",
params![
rec.project_id, rec.title, rec.description, rec.status.as_str(), rec.priority,
rec.branch_name, rec.assignee, rec.workflow_def_id, rec.base_branch,
rec.review_rounds, rec.output_json, rec.idea_id,
rec.queue, rec.parent_id, rec.content_json, rec.module_id,
rec.updated_at, rec.id
rec.archived as i32, rec.updated_at, rec.id
],
)
}
@@ -173,7 +189,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, queue, parent_id, content_json, module_id, created_at, updated_at FROM tasks WHERE deleted_at IS NULL ORDER BY created_at DESC")
.prepare(&format!("SELECT {TASK_COLS} FROM tasks WHERE deleted_at IS NULL AND archived = 0 ORDER BY created_at DESC"))
.map_err(storage_err)?;
let rows = stmt
.query_map([], |row| task_from_row(row))
@@ -240,7 +256,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, queue, parent_id, content_json, module_id, created_at, updated_at FROM tasks WHERE id = ?1 AND deleted_at IS NULL")
.prepare(&format!("SELECT {TASK_COLS} FROM tasks WHERE id = ?1 AND deleted_at IS NULL AND archived = 0"))
.map_err(storage_err)?;
let row = stmt
.query_row(params![id], |row| task_from_row(row))
@@ -285,10 +301,10 @@ impl TaskRepo {
// 返回 None,杜绝回收站任务被推进(D-02 软删语义收口,一处关闭)。
let sql = if bump_rounds {
"UPDATE tasks SET status = ?1, review_rounds = review_rounds + 1, updated_at = ?2 \
WHERE id = ?3 AND status = ?4 AND deleted_at IS NULL"
WHERE id = ?3 AND status = ?4 AND deleted_at IS NULL AND archived = 0"
} else {
"UPDATE tasks SET status = ?1, updated_at = ?2 \
WHERE id = ?3 AND status = ?4 AND deleted_at IS NULL"
WHERE id = ?3 AND status = ?4 AND deleted_at IS NULL AND archived = 0"
};
let affected = guard
.execute(sql, params![new_status, now, id, expected])
@@ -298,7 +314,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, queue, parent_id, content_json, module_id, created_at, updated_at FROM tasks WHERE id = ?1")
.prepare(&format!("SELECT {TASK_COLS} FROM tasks WHERE id = ?1"))
.map_err(storage_err)?;
let row = stmt
.query_row(params![id], |row| task_from_row(row))
@@ -320,7 +336,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, queue, parent_id, content_json, module_id, created_at, updated_at FROM tasks WHERE deleted_at IS NULL AND project_id = ?1 ORDER BY created_at DESC")
.prepare(&format!("SELECT {TASK_COLS} FROM tasks WHERE deleted_at IS NULL AND archived = 0 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))
@@ -389,7 +405,7 @@ impl TaskRepo {
// 真参数拿到 ?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 where_clauses: Vec<String> = vec!["deleted_at IS NULL".to_string(), "archived = 0".to_string()];
let mut params_vec: Vec<Box<dyn rusqlite::ToSql>> = Vec::new();
if let Some(pid) = &project_id {
@@ -454,9 +470,7 @@ impl TaskRepo {
// 显式列出全部 19 列(含 V29 queue/parent_id/content_json 与 V41 module_id,
// 不 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, \
queue, parent_id, content_json, module_id, created_at, updated_at \
"SELECT {TASK_COLS} \
FROM tasks WHERE {} ORDER BY {} DESC{}",
where_clauses.join(" AND "),
order_col,
@@ -511,7 +525,7 @@ impl TaskRepo {
tokio::task::spawn_blocking(move || {
let guard = conn.blocking_lock();
let mut where_clauses: Vec<String> = vec!["deleted_at IS NULL".to_string()];
let mut where_clauses: Vec<String> = vec!["deleted_at IS NULL".to_string(), "archived = 0".to_string()];
let mut params_vec: Vec<Box<dyn rusqlite::ToSql>> = Vec::new();
if let Some(ref pid) = project_id {
@@ -581,7 +595,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, queue, parent_id, content_json, module_id, created_at, updated_at FROM tasks WHERE deleted_at IS NULL AND parent_id = ?1 ORDER BY created_at ASC")
.prepare(&format!("SELECT {TASK_COLS} FROM tasks WHERE deleted_at IS NULL AND archived = 0 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))
@@ -618,7 +632,7 @@ impl TaskRepo {
let mut stmt = guard
.prepare(
"SELECT status, COUNT(*) AS cnt FROM tasks \
WHERE deleted_at IS NULL AND parent_id = ?1 \
WHERE deleted_at IS NULL AND archived = 0 AND parent_id = ?1 \
GROUP BY status",
)
.map_err(storage_err)?;
@@ -660,7 +674,7 @@ impl TaskRepo {
let affected = guard
.execute(
"UPDATE tasks SET status = ?1, updated_at = ?2 \
WHERE id = ?3 AND deleted_at IS NULL",
WHERE id = ?3 AND deleted_at IS NULL AND archived = 0",
params![new_status, now, id],
)
.map_err(storage_err)?;
@@ -716,11 +730,6 @@ impl TaskRepo {
let id = id.to_owned();
let new_queue = new_queue.to_owned();
let now = now_millis_str();
// 显式列出全部 19 列(同 from_row 消费列,不 SELECT deleted_at:
// TaskRecord 不带该字段,取了 from_row 会因未知列报错)。
const TASK_COLS: &str = "id, project_id, title, description, status, priority, branch_name, \
assignee, workflow_def_id, base_branch, review_rounds, output_json, \
idea_id, queue, parent_id, content_json, module_id, created_at, updated_at";
tokio::task::spawn_blocking(move || {
let mut guard = conn.blocking_lock();
let tx = guard.transaction().map_err(storage_err)?;
@@ -729,7 +738,7 @@ impl TaskRepo {
let current: Option<TaskRecord> = {
let mut stmt = tx
.prepare(&format!(
"SELECT {TASK_COLS} FROM tasks WHERE id = ?1 AND deleted_at IS NULL"
"SELECT {TASK_COLS} FROM tasks WHERE id = ?1 AND deleted_at IS NULL AND archived = 0"
))
.map_err(storage_err)?;
stmt.query_row(params![id], |row| task_from_row(row))
@@ -793,7 +802,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, queue, parent_id, content_json, module_id, created_at, updated_at FROM tasks WHERE deleted_at IS NOT NULL ORDER BY updated_at DESC")
.prepare(&format!("SELECT {TASK_COLS} 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))
@@ -833,7 +842,7 @@ impl TaskRepo {
.execute(
// deleted_at IS NULL:回收站任务不可被 CAS 覆盖——MCP 拿旧 expected 也能
// 命中已软删行,造成回收站内容被改且 updated_at 抖动,调用方无从感知。
"UPDATE tasks SET project_id = ?1, title = ?2, description = ?3, status = ?4, priority = ?5, branch_name = ?6, assignee = ?7, workflow_def_id = ?8, base_branch = ?9, review_rounds = ?10, output_json = ?11, idea_id = ?12, queue = ?13, parent_id = ?14, content_json = ?15, module_id = ?16, updated_at = ?17 WHERE id = ?18 AND updated_at = ?19 AND deleted_at IS NULL",
"UPDATE tasks SET project_id = ?1, title = ?2, description = ?3, status = ?4, priority = ?5, branch_name = ?6, assignee = ?7, workflow_def_id = ?8, base_branch = ?9, review_rounds = ?10, output_json = ?11, idea_id = ?12, queue = ?13, parent_id = ?14, content_json = ?15, module_id = ?16, updated_at = ?17 WHERE id = ?18 AND updated_at = ?19 AND deleted_at IS NULL AND archived = 0",
params![
rec.project_id, rec.title, rec.description, rec.status.as_str(), rec.priority,
rec.branch_name, rec.assignee, rec.workflow_def_id, rec.base_branch,
@@ -859,7 +868,7 @@ impl TaskRepo {
validate_column_name(field, "tasks")?;
let conn = self.conn.clone();
let sql = format!(
"UPDATE tasks SET {} = ?1, updated_at = ?2 WHERE id = ?3 AND deleted_at IS NULL",
"UPDATE tasks SET {} = ?1, updated_at = ?2 WHERE id = ?3 AND deleted_at IS NULL AND archived = 0",
field
);
let id = id.to_owned();
@@ -875,6 +884,56 @@ impl TaskRepo {
.await
.map_err(storage_err)?
}
/// 归档/取消归档(归档功能 P1,V46)。
///
/// 纯元数据标记(对标 ConversationRepo::set_archived),仅作用于未软删任务:
/// 软删(回收站)优先级高于归档——回收站任务不可直接归档/取消归档,先恢复再操作。
/// archived=true 后任务冻结(活跃查询恒过滤,不可推进/跨池/字段更新),仅在归档视图可见。
/// 返回是否命中。
pub async fn set_archived(&self, id: &str, archived: bool) -> Result<bool> {
let conn = self.conn.clone();
let id = id.to_owned();
let now = now_millis_str();
tokio::task::spawn_blocking(move || {
let guard = conn.blocking_lock();
let affected = guard
.execute(
"UPDATE tasks SET archived = ?1, updated_at = ?2 WHERE id = ?3 AND deleted_at IS NULL",
params![if archived { 1 } else { 0 }, now, id],
)
.map_err(storage_err)?;
Ok(affected > 0)
})
.await
.map_err(storage_err)?
}
/// 列出归档任务(archived = 1 且未软删),按更新时间降序(最近归档在前)。
///
/// 归档视图数据源。与 list_active 对偶:活跃查询恒带 archived = 0,
/// 本方法恒带 archived = 1,两集合互斥且并集 = 全部未删任务。
pub async fn list_archived(&self) -> Result<Vec<TaskRecord>> {
let conn = self.conn.clone();
tokio::task::spawn_blocking(move || {
let guard = conn.blocking_lock();
let mut stmt = guard
.prepare(&format!(
"SELECT {TASK_COLS} FROM tasks WHERE deleted_at IS NULL AND archived = 1 ORDER BY updated_at DESC"
))
.map_err(storage_err)?;
let rows = stmt
.query_map([], |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)?
}
}
#[cfg(test)]
@@ -909,6 +968,7 @@ mod tests {
parent_id: parent_id.map(|s| s.to_string()),
content_json: None,
module_id: None,
archived: false,
created_at: "1700000000000".to_string(),
updated_at: "1700000000000".to_string(),
}
crates/df-storage/src/crud/task_validation.rs
+1
View File
@@ -130,6 +130,7 @@ mod tests {
parent_id: parent_id.map(|s| s.to_string()),
content_json: None,
module_id: None,
archived: false,
created_at: "1700000000000".to_string(),
updated_at: "1700000000000".to_string(),
}
crates/df-storage/src/migrations.rs
+65 -5
View File
@@ -53,7 +53,9 @@ pub fn run(conn: &Connection) -> Result<()> {
// 桌面侧栏/miniapp 列表按 origin 过滤子会话)。
// V45 = 无人值守任务边界(ai_conversations.boundary_json TEXT,TaskBoundary JSON 快照,
// 发起时一次声明存续期生效,随会话同亡)。
let steps: [(i32, fn(&Connection) -> Result<()>); 45] = [
// V46/V47 = 任务/灵感归档(archived INTEGER NOT NULL DEFAULT 0,归档功能 P1,
// 对标 ai_conversations.archived V4;归档与软删正交:回收站=删除缓冲,归档=低频可见保检索)。
let steps: [(i32, fn(&Connection) -> Result<()>); 47] = [
(1, migrate_v1),
(2, migrate_v2),
(3, migrate_v3),
@@ -99,6 +101,10 @@ pub fn run(conn: &Connection) -> Result<()> {
(43, migrate_v43),
(44, migrate_v44),
(45, migrate_v45),
// V46 = 任务归档(tasks.archived 列,归档功能 P1,对标 ai_conversations.archived V4);
// V47 = 灵感归档(ideas.archived 列,同款)。归档与软删正交:回收站=删除缓冲,归档=低频可见保检索。
(46, migrate_v46),
(47, migrate_v47),
];
for (version, migrate_fn) in steps {
@@ -1354,6 +1360,50 @@ fn migrate_v45(conn: &Connection) -> Result<()> {
Ok(())
}
/// V46: 幂等补 tasks.archived 列(任务归档,归档功能 P1)
///
/// 归档语义对标 ai_conversations.archived(V4):纯元数据标记,与 status/queue/deleted_at 正交。
/// 「作废」走 status=cancelled,「完结」走 done/cancelled——归档是另一种动作:
/// 低频完结任务移出主视图但保留检索(回收站=删除缓冲不可见,归档=低频可见保检索)。
/// - `INTEGER NOT NULL DEFAULT 0`:老库行迁移后全为 0(未归档),行为零变化;
/// - 不进 update_field 白名单(settings.rs):归档走专用 set_archived(带埋点 + 布尔语义),
/// 对标 status 收口思路防旁路写;
/// - 活跃查询(list_active/list_by_query/count_by_query/list_active_by_project/get_children/
/// advance_status_atomic/move_task_queue)恒带 `archived = 0` 过滤,归档任务退出主视图与
/// 状态机(冻结:不可推进/不可跨池/不可字段更新,需先取消归档)。
/// - **幂等**:column_exists 守卫(同 V45 模式);**可逆**:`ALTER TABLE tasks DROP COLUMN archived`。
fn migrate_v46(conn: &Connection) -> Result<()> {
if !column_exists(conn, "tasks", "archived") {
conn.execute(
"ALTER TABLE tasks ADD COLUMN archived INTEGER NOT NULL DEFAULT 0",
[],
)?;
tracing::info!("v46: 补建 tasks.archived 列(任务归档)");
}
conn.execute("INSERT OR IGNORE INTO schema_version (version) VALUES (?)", [46])?;
tracing::info!("迁移 v46 完成: tasks 加 archived 列");
Ok(())
}
/// V47: 幂等补 ideas.archived 列(灵感归档,归档功能 P1)
///
/// 对标 V46 tasks.archived:rejected/deferred/已晋升灵感一键归档移出主视图,保留检索。
/// 灵感无状态机收口(status 走 idea_repo 通用路径),但归档同样不进 update_field 白名单,
/// 走专用 set_archived(布尔语义 + 防旁路)。活跃查询恒带 `archived = 0`。
/// - **幂等**:column_exists 守卫;**可逆**:`ALTER TABLE ideas DROP COLUMN archived`。
fn migrate_v47(conn: &Connection) -> Result<()> {
if !column_exists(conn, "ideas", "archived") {
conn.execute(
"ALTER TABLE ideas ADD COLUMN archived INTEGER NOT NULL DEFAULT 0",
[],
)?;
tracing::info!("v47: 补建 ideas.archived 列(灵感归档)");
}
conn.execute("INSERT OR IGNORE INTO schema_version (version) VALUES (?)", [47])?;
tracing::info!("迁移 v47 完成: ideas 加 archived 列");
Ok(())
}
/// V21 建表 SQL — 消息拆分存储 ai_messages 表
///
/// 与 V9_SQL 中的 ai_messages 镜像(V9 给新库,此 const 给老库 V21 迁移用 IF NOT EXISTS)。
@@ -2357,6 +2407,16 @@ mod tests {
column_exists(&conn, "tasks", "module_id"),
"tasks.module_id 列缺失(V41 加)"
);
// tasks.archived(V46 归档功能 P1)
assert!(
column_exists(&conn, "tasks", "archived"),
"tasks.archived 列缺失(V46 加)"
);
// ideas.archived(V47 归档功能 P1)
assert!(
column_exists(&conn, "ideas", "archived"),
"ideas.archived 列缺失(V47 加)"
);
// 3. schema_version 应推进到 45(全量迁移成功落版本号)
let max_version: i64 = conn
@@ -2367,8 +2427,8 @@ mod tests {
)
.expect("查 schema_version 应成功");
assert_eq!(
max_version, 45,
"全量迁移后 schema_version 应为 45(实际 {}),说明某条 migrate_vN 链路断在中间",
max_version, 47,
"全量迁移后 schema_version 应为 47(实际 {}),说明某条 migrate_vN 链路断在中间",
max_version
);
@@ -2445,7 +2505,7 @@ mod tests {
|r| r.get(0),
)
.unwrap();
assert_eq!(max_v, 45, "首轮应推进到 45");
assert_eq!(max_v, 47, "首轮应推进到 47");
// 清空版本表强制全链第二遍(每步 execute 第二次)
conn.execute("DELETE FROM schema_version", []).unwrap();
@@ -2457,7 +2517,7 @@ mod tests {
|r| r.get(0),
)
.unwrap();
assert_eq!(max_v2, 45, "重跑后应重新推进到 45");
assert_eq!(max_v2, 47, "重跑后应重新推进到 47");
}
// ============================================================
crates/df-storage/src/models.rs
+9 -1
View File
@@ -23,6 +23,10 @@ pub struct IdeaRecord {
pub ai_analysis: Option<String>, // AI 分析结果 JSON
pub scores: Option<String>, // 多维评分 JSON (feasibility/impact/urgency/overall)
pub related_ids: Option<String>, // 关联灵感 id JSON 数组(同 tags 模式,为关联关系 UI 打底)
/// 归档标记(归档功能 P1,V47 列)。纯元数据,archived=true 的灵感退出主视图,仅在归档视图可见。
/// DEFAULT 0 保证老库行非 NULL;#[serde(default)] 兼容旧前端 JSON(缺省 false)。
#[serde(default)]
pub archived: bool,
pub created_at: String,
pub updated_at: String,
}
@@ -123,9 +127,13 @@ pub struct TaskRecord {
/// 一个项目可含多个工程(Monorepo/微服务),本字段让任务落到具体工程,是后续
/// IPC / 前端 / AI 按工程筛选任务的数据基础。外键 REFERENCES project_modules(id)
/// ON DELETE SET NULL(module 删除时任务自动解关联)。可空(老任务无关联 → None)。
/// #[serde(default)] 兼容旧前端 JSON(无该字段时为 None)。
#[serde(default)]
pub module_id: Option<String>,
/// 归档标记(归档功能 P1,V46 列)。与 status/queue/deleted_at 正交的纯元数据:
/// archived=true 的任务退出主视图/状态机(冻结,不可推进/跨池/字段更新),仅在归档视图可见。
/// DEFAULT 0 保证老库行非 NULL;#[serde(default)] 兼容旧前端 JSON(缺省 false)。
#[serde(default)]
pub archived: bool,
pub created_at: String,
pub updated_at: String,
}
crates/df-storage/tests/project_soft_delete.rs
+1
View File
@@ -56,6 +56,7 @@ fn task(id: &str, project_id: &str) -> TaskRecord {
parent_id: None,
content_json: None,
module_id: None,
archived: false,
created_at: now_ts(),
updated_at: now_ts(),
}