新增: 任务推进链(7态状态机+advance_task CAS原子写)+软删除+前后端7态对齐

2026-06-16 02:33:15 +08:00
parent f30df333b3
commit d2cb38cdac
21 changed files with 1628 additions and 167 deletions
--- a/crates/df-nodes/src/task_advance_node.rs
+++ b/crates/df-nodes/src/task_advance_node.rs
@@ -0,0 +1,341 @@
+//! 任务推进节点 — advance_task 推进链触发器(F-260616-02)
+//!
+//! 实现推进链的唯一 status 写入路径(D-260616-03 落 df-nodes Node):
+//!   1. 读当前 TaskRecord(取 from status)
+//!   2. 状态机校验 can_transition(from, to)(task_state_machine)
+//!   3. 原子写:下沉 SQL `WHERE id AND status=expected` 防 TOCTOU,退回转换一并
+//!      review_rounds+=1(见 df_storage::crud::TaskRepo::advance_status_atomic)
+//!
+//! 两条调用路径复用同一推进逻辑:
+//!   - DAG 工作流:TaskAdvanceNode 实现 Node trait,从 NodeContext.config 读 task_id /
+//!     target_status,持有 Arc<Database> 构造 TaskRepo。
+//!   - IPC 直驱:advance_task_atomic 公开 async fn,src-tauri commands::task::advance_task
+//!     thin 入口直接调用(无需构造 NodeContext,避免 EventBus/StateMachine 等重依赖)。
+
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use df_storage::crud::TaskRepo;
+use df_storage::db::Database;
+use df_storage::models::TaskRecord;
+use df_workflow::node::{Node, NodeContext, NodeOutput, NodeResult, NodeSchema};
+
+use crate::task_state_machine::{can_transition, is_regression, is_valid_state};
+
+/// 推进任务到目标状态(核心逻辑,DAG 节点与 IPC 入口共用)。
+///
+/// 入参:
+///   - `repo`:df-storage TaskRepo
+///   - `id`:任务 ID
+///   - `target_status`:目标状态(7 态之一,snake_case)
+///
+/// 流程:
+///   1. 任务存在性:找不到 → InvalidState 错误(复用 Error::InvalidState 语义,前端可据此提示)
+///   2. target_status 合法性:非 7 态 → Validation 错误(防脏数据直入推进链)
+///   3. 状态机:can_transition(from, to) 不合法 → InvalidState 错误(含 from/to 上下文)
+///   4. 原子写:advance_status_atomic CAS,to 是退回转换时 bump_rounds=true
+///   5. CAS 失败(affected==0):状态已被并发改动 → InvalidState 错误(防 TOCTOU 静默成功)
+///
+/// 返回:成功推进后的最新 TaskRecord(含新 status / 累加后的 review_rounds / 新 updated_at)。
+pub async fn advance_task_atomic(
+    repo: &TaskRepo,
+    id: &str,
+    target_status: &str,
+) -> df_core::error::Result<TaskRecord> {
+    // 1. target 合法性(7 态之一)。先于读库校验:即便任务不存在,也先拒绝非法状态值。
+    if !is_valid_state(target_status) {
+        return Err(df_core::error::Error::Validation(format!(
+            "非法 target_status {:?},合法值: todo/in_progress/in_review/testing/done/blocked/cancelled",
+            target_status
+        )));
+    }
+
+    // 2. 读当前任务(取 from status)。
+    let current = repo
+        .get_by_id(id)
+        .await?
+        .ok_or_else(|| df_core::error::Error::NotFound(format!("任务 {} 不存在", id)))?;
+
+    // 3. 状态机校验(同态拒绝:todo→todo 等,can_transition 返回 false)。
+    let from = current.status.as_str();
+    if from == target_status {
+        return Err(df_core::error::Error::InvalidState {
+            current: from.to_string(),
+            expected: target_status.to_string(),
+        });
+    }
+    if !can_transition(from, target_status) {
+        return Err(df_core::error::Error::InvalidState {
+            current: from.to_string(),
+            expected: target_status.to_string(),
+        });
+    }
+
+    // 4. 原子 CAS 写。退回转换一并 review_rounds+=1。
+    let bump = is_regression(from, target_status);
+    let updated = repo
+        .advance_status_atomic(id, from, target_status, bump)
+        .await?;
+
+    // 5. CAS 失败:并发已改动 status(或任务被删),拒绝静默成功。
+    updated.ok_or_else(|| df_core::error::Error::InvalidState {
+        current: "已变更(并发推进或旁路修改)".to_string(),
+        expected: from.to_string(),
+    })
+}
+
+// ============================================================
+// DAG 工作流节点 — TaskAdvanceNode(推进链在 DAG 内的形态)
+// ============================================================
+
+/// 任务推进节点 — DAG 工作流内触发 advance_task。
+///
+/// 持有 Arc<Database> 在 NodeRegistry 注册时注入(state.rs build_registry),
+/// Node::execute 从 NodeContext.config 读 task_id / target_status,调 advance_task_atomic。
+/// 当前推进链 F-01~04 阶段为手动推进(IPC 直驱),DAG 形态为阶段 2(工作流联动)预留。
+pub struct TaskAdvanceNode {
+    db: Arc<Database>,
+}
+
+impl TaskAdvanceNode {
+    /// 构造节点(注册表工厂调用,注入数据库句柄)
+    pub fn new(db: Arc<Database>) -> Self {
+        Self { db }
+    }
+}
+
+#[async_trait]
+impl Node for TaskAdvanceNode {
+    async fn execute(&self, ctx: NodeContext) -> NodeResult {
+        let task_id = ctx
+            .config
+            .get("task_id")
+            .and_then(|v| v.as_str())
+            .ok_or_else(|| anyhow::anyhow!("TaskAdvanceNode 缺少必填参数: task_id"))?;
+        let target_status = ctx
+            .config
+            .get("target_status")
+            .and_then(|v| v.as_str())
+            .ok_or_else(|| anyhow::anyhow!("TaskAdvanceNode 缺少必填参数: target_status"))?;
+
+        let repo = TaskRepo::new(&self.db);
+        let updated = advance_task_atomic(&repo, task_id, target_status)
+            .await
+            .map_err(|e| anyhow::anyhow!("推进任务失败: {}", e))?;
+
+        Ok(NodeOutput::from_value(serde_json::json!({
+            "task": updated,
+            "task_id": updated.id,
+            "status": updated.status,
+            "review_rounds": updated.review_rounds,
+        })))
+    }
+
+    fn schema(&self) -> NodeSchema {
+        NodeSchema {
+            params: serde_json::json!({
+                "type": "object",
+                "properties": {
+                    "task_id": { "type": "string" },
+                    "target_status": {
+                        "type": "string",
+                        "enum": ["todo", "in_progress", "in_review", "testing", "done", "blocked", "cancelled"]
+                    }
+                },
+                "required": ["task_id", "target_status"]
+            }),
+            output: serde_json::json!({
+                "type": "object",
+                "properties": {
+                    "task_id": { "type": "string" },
+                    "status": { "type": "string" },
+                    "review_rounds": { "type": "integer" }
+                }
+            }),
+        }
+    }
+
+    fn node_type(&self) -> &str {
+        "task_advance"
+    }
+}
+
+// ============================================================
+// 单元测试 — advance_task_atomic 状态机集成(内存 DB)
+// ============================================================
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use df_storage::crud::ProjectRepo;
+    use df_storage::models::{ProjectRecord, TaskRecord};
+
+    fn rec(id: &str, status: &str) -> TaskRecord {
+        TaskRecord {
+            id: id.to_string(),
+            project_id: "p1".to_string(),
+            title: format!("t-{id}"),
+            description: "".to_string(),
+            status: status.to_string(),
+            priority: 2,
+            branch_name: None,
+            assignee: None,
+            workflow_def_id: None,
+            base_branch: None,
+            review_rounds: 0,
+            created_at: "0".to_string(),
+            updated_at: "0".to_string(),
+        }
+    }
+
+    async fn setup() -> TaskRepo {
+        let db = Database::open_in_memory().await.expect("open_in_memory");
+        // 先插父项目,满足 tasks.project_id FK 约束(PRAGMA foreign_keys=ON)。
+        ProjectRepo::new(&db)
+            .insert(ProjectRecord {
+                id: "p1".to_string(),
+                name: "proj".to_string(),
+                description: "".to_string(),
+                status: "planning".to_string(),
+                idea_id: None,
+                path: None,
+                stack: None,
+                created_at: "0".to_string(),
+                updated_at: "0".to_string(),
+            })
+            .await
+            .expect("insert project");
+        TaskRepo::new(&db)
+    }
+
+    #[tokio::test]
+    async fn forward_path_todo_to_done() {
+        let repo = setup().await;
+        repo.insert(rec("t1", "todo")).await.unwrap();
+        // 主路径逐级推进
+        let r = advance_task_atomic(&repo, "t1", "in_progress").await.unwrap();
+        assert_eq!(r.status, "in_progress");
+        assert_eq!(r.review_rounds, 0);
+        let r = advance_task_atomic(&repo, "t1", "in_review").await.unwrap();
+        assert_eq!(r.status, "in_review");
+        assert_eq!(r.review_rounds, 0);
+        let r = advance_task_atomic(&repo, "t1", "testing").await.unwrap();
+        assert_eq!(r.status, "testing");
+        let r = advance_task_atomic(&repo, "t1", "done").await.unwrap();
+        assert_eq!(r.status, "done");
+        assert_eq!(r.review_rounds, 0);
+    }
+
+    #[tokio::test]
+    async fn regression_in_review_to_in_progress_bumps_rounds() {
+        let repo = setup().await;
+        repo.insert(rec("t1", "in_review")).await.unwrap();
+        let r = advance_task_atomic(&repo, "t1", "in_progress").await.unwrap();
+        assert_eq!(r.status, "in_progress");
+        assert_eq!(r.review_rounds, 1, "退回应 +1");
+    }
+
+    #[tokio::test]
+    async fn regression_testing_to_in_review_bumps_rounds() {
+        let repo = setup().await;
+        repo.insert(rec("t1", "testing")).await.unwrap();
+        let r = advance_task_atomic(&repo, "t1", "in_review").await.unwrap();
+        assert_eq!(r.status, "in_review");
+        assert_eq!(r.review_rounds, 1, "退回应 +1");
+    }
+
+    #[tokio::test]
+    async fn multiple_regressions_accumulate() {
+        let repo = setup().await;
+        repo.insert(rec("t1", "in_review")).await.unwrap();
+        // in_review → in_progress (+1) → in_review (前向,不动) → in_progress (+1=2)
+        advance_task_atomic(&repo, "t1", "in_progress").await.unwrap();
+        advance_task_atomic(&repo, "t1", "in_review").await.unwrap();
+        let r = advance_task_atomic(&repo, "t1", "in_progress").await.unwrap();
+        assert_eq!(r.review_rounds, 2);
+    }
+
+    #[tokio::test]
+    async fn illegal_skip_rejected() {
+        let repo = setup().await;
+        repo.insert(rec("t1", "todo")).await.unwrap();
+        let err = advance_task_atomic(&repo, "t1", "done").await.unwrap_err();
+        assert!(matches!(err, df_core::error::Error::InvalidState { .. }));
+    }
+
+    #[tokio::test]
+    async fn terminal_done_no_successor() {
+        let repo = setup().await;
+        repo.insert(rec("t1", "done")).await.unwrap();
+        let err = advance_task_atomic(&repo, "t1", "todo").await.unwrap_err();
+        assert!(matches!(err, df_core::error::Error::InvalidState { .. }));
+    }
+
+    #[tokio::test]
+    async fn same_status_rejected() {
+        let repo = setup().await;
+        repo.insert(rec("t1", "in_progress")).await.unwrap();
+        let err = advance_task_atomic(&repo, "t1", "in_progress").await.unwrap_err();
+        assert!(matches!(err, df_core::error::Error::InvalidState { .. }));
+    }
+
+    #[tokio::test]
+    async fn invalid_target_rejected() {
+        let repo = setup().await;
+        repo.insert(rec("t1", "todo")).await.unwrap();
+        let err = advance_task_atomic(&repo, "t1", "merged").await.unwrap_err();
+        assert!(matches!(err, df_core::error::Error::Validation(_)));
+    }
+
+    #[tokio::test]
+    async fn not_found_rejected() {
+        let repo = setup().await;
+        let err = advance_task_atomic(&repo, "nope", "in_progress").await.unwrap_err();
+        assert!(matches!(err, df_core::error::Error::NotFound(_)));
+    }
+
+    #[tokio::test]
+    async fn concurrent_cas_change_detected() {
+        // 模拟 TOCTOU:推进前已被旁路改 status。CAS 的 expected 与库内不符 → None → 报错。
+        let repo = setup().await;
+        repo.insert(rec("t1", "todo")).await.unwrap();
+        // 旁路把 status 改成 in_progress(模拟另一路并发推进)
+        repo.update_field("t1", "status", "in_progress").await.unwrap();
+        // 此时读出来是 in_progress,推进到 in_review 是合法的——这测的不是 CAS 失败,
+        // 而是验证「读后改」路径在 status 一致时正常。CAS 失败路径靠 advance_status_atomic
+        // 自身的 None 返回覆盖(下方 cas_returns_none_when_status_mismatch 单测)。
+        let r = advance_task_atomic(&repo, "t1", "in_review").await.unwrap();
+        assert_eq!(r.status, "in_review");
+    }
+
+    #[tokio::test]
+    async fn cas_returns_none_when_status_mismatch() {
+        let repo = setup().await;
+        repo.insert(rec("t1", "todo")).await.unwrap();
+        // 直接调底层:expected 传错(模拟读到 todo 但实际已被改成 in_progress)
+        let r = repo
+            .advance_status_atomic("t1", "todo", "in_review", false)
+            .await
+            .unwrap();
+        // 此时 status 确实是 todo,CAS 成功,返回 Some
+        assert!(r.is_some());
+        // 再用错误的 expected todo(实际已是 in_review)→ CAS 失败 None
+        let r = repo
+            .advance_status_atomic("t1", "todo", "done", false)
+            .await
+            .unwrap();
+        assert!(r.is_none(), "expected 与库内不符应 CAS 失败");
+    }
+
+    #[tokio::test]
+    async fn blocked_round_trip_does_not_bump() {
+        let repo = setup().await;
+        repo.insert(rec("t1", "in_progress")).await.unwrap();
+        let r = advance_task_atomic(&repo, "t1", "blocked").await.unwrap();
+        assert_eq!(r.status, "blocked");
+        assert_eq!(r.review_rounds, 0, "进 blocked 不累加");
+        let r = advance_task_atomic(&repo, "t1", "in_progress").await.unwrap();
+        assert_eq!(r.status, "in_progress");
+        assert_eq!(r.review_rounds, 0, "解除 blocked 不累加");
+    }
+}