diff --git a/crates/df-nodes/Cargo.toml b/crates/df-nodes/Cargo.toml index 5a9b045..7dc71cc 100644 --- a/crates/df-nodes/Cargo.toml +++ b/crates/df-nodes/Cargo.toml @@ -8,6 +8,7 @@ df-core = { path = "../df-core" } df-execute = { path = "../df-execute" } df-workflow = { path = "../df-workflow" } df-ai = { path = "../df-ai" } +df-storage = { path = "../df-storage" } serde = { workspace = true } serde_json = { workspace = true } tokio = { workspace = true } diff --git a/crates/df-nodes/src/lib.rs b/crates/df-nodes/src/lib.rs index 370fdf3..bc024a8 100644 --- a/crates/df-nodes/src/lib.rs +++ b/crates/df-nodes/src/lib.rs @@ -3,3 +3,5 @@ pub mod ai_node; pub mod human_node; pub mod script_node; +pub mod task_advance_node; +pub mod task_state_machine; diff --git a/crates/df-nodes/src/script_node.rs b/crates/df-nodes/src/script_node.rs index 7a8eaaf..dab4974 100644 --- a/crates/df-nodes/src/script_node.rs +++ b/crates/df-nodes/src/script_node.rs @@ -36,6 +36,7 @@ impl Node for ScriptNode { working_dir, env: std::collections::HashMap::new(), timeout_secs, + shell_type: Default::default(), }; tracing::info!("ScriptNode 执行命令: {}", command); diff --git a/crates/df-nodes/src/task_advance_node.rs b/crates/df-nodes/src/task_advance_node.rs new file mode 100644 index 0000000..34d469c --- /dev/null +++ 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 构造 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 { + // 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 在 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, +} + +impl TaskAdvanceNode { + /// 构造节点(注册表工厂调用,注入数据库句柄) + pub fn new(db: Arc) -> 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 不累加"); + } +} diff --git a/crates/df-nodes/src/task_state_machine.rs b/crates/df-nodes/src/task_state_machine.rs new file mode 100644 index 0000000..badd6f9 --- /dev/null +++ b/crates/df-nodes/src/task_state_machine.rs @@ -0,0 +1,256 @@ +//! 任务推进状态机 — 7 态合法转换定义(F-260616-01) +//! +//! 独立模块,非挂在 df-core::TaskStatus enum 上(对齐 D-260616-03「推进链业务逻辑落 +//! df-nodes」)。本模块只做「给定 from/to 是否合法」的纯函数判定,不触碰存储层 +//! (原子写 SQL 在 task_advance_node.rs 完成,见 F-260616-02)。 +//! +//! 7 态(与 df-core::TaskStatus / 前端对齐,D-260616-01): +//! todo / in_progress / in_review / testing / done / blocked / cancelled +//! +//! 闸门链主路径: todo → in_progress → in_review → testing → done +//! +//! 合法转换矩阵(以报告 docs/05-代码审查/任务执行与推进能力分析-2026-06-16.md §8 +//! 与决策 docs/02-架构设计/任务推进链实施路径-2026-06-16.md F-01 为准): +//! - todo → in_progress(开始), cancelled(取消) +//! - in_progress → in_review(提交审查), blocked(阻塞), cancelled +//! - in_review → testing(审查通过进测试), in_progress(退回修改, review_rounds+1), +//! blocked, cancelled +//! - testing → done(测试通过), in_review(退回重审, review_rounds+1), +//! blocked, cancelled +//! - done → 终态(原则上不可变;若需重开走 cancelled 或新任务) +//! - blocked → in_progress(解除阻塞继续), cancelled +//! - cancelled → 终态 + +// ============================================================ +// 状态字符串常量 — 与 df-core::TaskStatus::as_str 一一对应 +// ============================================================ +// +// 不复用 df-core::TaskStatus enum(独立模块定位 + 避免推进链判定耦合存储枚举类型), +// 但字符串值严格对齐(df-core::TaskStatus::as_str 产出的小写 snake_case), +// 保证状态机判定的 from/to 与数据库 status 列存值语义一致。 + +/// 待开始 +pub const TODO: &str = "todo"; +/// 进行中 +pub const IN_PROGRESS: &str = "in_progress"; +/// 代码审查中 +pub const IN_REVIEW: &str = "in_review"; +/// 测试中 +pub const TESTING: &str = "testing"; +/// 已完成(终态) +pub const DONE: &str = "done"; +/// 已阻塞 +pub const BLOCKED: &str = "blocked"; +/// 已取消(终态) +pub const CANCELLED: &str = "cancelled"; + +/// 全部合法状态值(供输入校验与错误提示复用) +pub const ALL_STATES: &[&str] = &[ + TODO, + IN_PROGRESS, + IN_REVIEW, + TESTING, + DONE, + BLOCKED, + CANCELLED, +]; + +/// 字符串是否为合法状态值 +pub fn is_valid_state(s: &str) -> bool { + ALL_STATES.contains(&s) +} + +/// 判定从 `from` 到 `to` 的状态转换是否合法(状态机核心)。 +/// +/// 终态(done/cancelled)无任何合法后继;非法或未知状态入参一律返回 false +/// (调用方 advance_task 在前置校验已拦截非法 status,此处保守拒绝防漏)。 +/// +/// 注意:仅判「是否合法」,不判「是否是退回」——退回(导致 review_rounds+1) +/// 的识别见 [`is_regression`],advance_task 据此决定是否一并 SET review_rounds+=1。 +pub fn can_transition(from: &str, to: &str) -> bool { + use std::sync::OnceLock; + + // 转换表静态构造一次(运行时常量,无锁开销分摊)。行=from,列=to。 + // 表条目为 true 即合法转换。未列入的 (from, to) 一律 false。 + static TABLE: OnceLock> = + OnceLock::new(); + let table = TABLE.get_or_init(|| { + let mut m = std::collections::HashMap::new(); + let allowed: &[(&str, &str)] = &[ + // todo → in_progress(开始), cancelled(取消) + (TODO, IN_PROGRESS), + (TODO, CANCELLED), + // in_progress → in_review(提交审查), blocked(阻塞), cancelled + (IN_PROGRESS, IN_REVIEW), + (IN_PROGRESS, BLOCKED), + (IN_PROGRESS, CANCELLED), + // in_review → testing(审查通过进测试), in_progress(退回修改), + // blocked, cancelled + (IN_REVIEW, TESTING), + (IN_REVIEW, IN_PROGRESS), + (IN_REVIEW, BLOCKED), + (IN_REVIEW, CANCELLED), + // testing → done(测试通过), in_review(退回重审), blocked, cancelled + (TESTING, DONE), + (TESTING, IN_REVIEW), + (TESTING, BLOCKED), + (TESTING, CANCELLED), + // done → 终态,无后继 + // blocked → in_progress(解除阻塞继续), cancelled + (BLOCKED, IN_PROGRESS), + (BLOCKED, CANCELLED), + // cancelled → 终态,无后继 + ]; + for (f, t) in allowed { + m.insert((*f, *t), true); + } + m + }); + + table.get(&(from, to)).copied().unwrap_or(false) +} + +/// 判定一次转换是否为「退回」(review_rounds 应 +1)。 +/// +/// 退回语义:任务从前向推进阶段回退到更早的推进阶段,意味着上一轮产出未过闸门、 +/// 需重做。具体两类: +/// - in_review → in_progress(审查退回修改) +/// - testing → in_review(测试退回重审) +/// +/// 其余合法转换(前向推进 / 进出 blocked / 进 cancelled)均不累加 review_rounds。 +pub fn is_regression(from: &str, to: &str) -> bool { + matches!((from, to), (IN_REVIEW, IN_PROGRESS) | (TESTING, IN_REVIEW)) +} + +// ============================================================ +// 单元测试 — 转换矩阵 + 边界 +// ============================================================ + +#[cfg(test)] +mod tests { + use super::*; + + // ---------- can_transition 闸门主路径 ---------- + + #[test] + fn main_path_todo_to_done_all_forward() { + assert!(can_transition(TODO, IN_PROGRESS)); + assert!(can_transition(IN_PROGRESS, IN_REVIEW)); + assert!(can_transition(IN_REVIEW, TESTING)); + assert!(can_transition(TESTING, DONE)); + } + + #[test] + fn cancel_from_any_non_terminal() { + for s in [TODO, IN_PROGRESS, IN_REVIEW, TESTING, BLOCKED] { + assert!(can_transition(s, CANCELLED), "应允许 {s} → cancelled"); + } + } + + #[test] + fn blocked_round_trip() { + assert!(can_transition(IN_PROGRESS, BLOCKED)); + assert!(can_transition(BLOCKED, IN_PROGRESS)); + assert!(can_transition(IN_REVIEW, BLOCKED)); + assert!(can_transition(TESTING, BLOCKED)); + } + + #[test] + fn regression_paths_allowed() { + assert!(can_transition(IN_REVIEW, IN_PROGRESS)); + assert!(can_transition(TESTING, IN_REVIEW)); + } + + // ---------- 非法转换被拒 ---------- + + #[test] + fn illegal_forward_skips_rejected() { + // 不允许跳过闸门(直奔 done) + assert!(!can_transition(TODO, DONE)); + assert!(!can_transition(TODO, TESTING)); + assert!(!can_transition(TODO, IN_REVIEW)); + assert!(!can_transition(IN_PROGRESS, DONE)); + assert!(!can_transition(IN_PROGRESS, TESTING)); + assert!(!can_transition(IN_REVIEW, DONE)); + } + + #[test] + fn terminal_states_have_no_successors() { + // done / cancelled 是终态,任何后继都拒绝 + for term in [DONE, CANCELLED] { + for to in ALL_STATES { + assert!(!can_transition(term, to), "终态 {term} 不应有后继 → {to}"); + } + } + } + + #[test] + fn backward_to_todo_rejected() { + // 不允许回退到 todo(开始即不可撤回) + for s in [IN_PROGRESS, IN_REVIEW, TESTING, BLOCKED] { + assert!(!can_transition(s, TODO), "不应允许 {s} → todo"); + } + } + + #[test] + fn blocked_only_to_in_progress_or_cancelled() { + assert!(can_transition(BLOCKED, IN_PROGRESS)); + assert!(can_transition(BLOCKED, CANCELLED)); + // 解除阻塞不能直接跳到 in_review/testing/done + assert!(!can_transition(BLOCKED, IN_REVIEW)); + assert!(!can_transition(BLOCKED, TESTING)); + assert!(!can_transition(BLOCKED, DONE)); + } + + // ---------- 未知状态 ---------- + + #[test] + fn unknown_states_rejected() { + assert!(!can_transition("unknown", TODO)); + assert!(!can_transition(TODO, "unknown")); + assert!(!can_transition("", "")); + } + + // ---------- is_regression ---------- + + #[test] + fn regression_only_on_review_back_edges() { + assert!(is_regression(IN_REVIEW, IN_PROGRESS)); + assert!(is_regression(TESTING, IN_REVIEW)); + // 前向推进不累加 + assert!(!is_regression(TODO, IN_PROGRESS)); + assert!(!is_regression(IN_PROGRESS, IN_REVIEW)); + assert!(!is_regression(IN_REVIEW, TESTING)); + assert!(!is_regression(TESTING, DONE)); + // 进出 blocked 不累加 + assert!(!is_regression(IN_PROGRESS, BLOCKED)); + assert!(!is_regression(BLOCKED, IN_PROGRESS)); + // 进 cancelled 不累加 + assert!(!is_regression(IN_REVIEW, CANCELLED)); + // 不合法的"退回"(实际上不存在的转换)也不应判定为 regression + assert!(!is_regression(TESTING, IN_PROGRESS)); + } + + // ---------- is_valid_state ---------- + + #[test] + fn is_valid_state_accepts_7_known() { + for s in ALL_STATES { + assert!(is_valid_state(s)); + } + } + + #[test] + fn is_valid_state_rejects_unknown() { + assert!(!is_valid_state("")); + assert!(!is_valid_state("unknown")); + assert!(!is_valid_state("TODO")); // 大写 + assert!(!is_valid_state("merged")); // 历史 5 态残留 + assert!(!is_valid_state("abandoned")); + } + + #[test] + fn all_states_has_seven_entries() { + assert_eq!(ALL_STATES.len(), 7); + } +} diff --git a/crates/df-storage/src/crud.rs b/crates/df-storage/src/crud.rs index 3ecad5a..067a996 100644 --- a/crates/df-storage/src/crud.rs +++ b/crates/df-storage/src/crud.rs @@ -310,7 +310,8 @@ impl SettingsRepo { /// 白名单两重作用:① 防注入(列名参数化前校验);② 按表隔离(update_task 误传 /// projects 的 "name" 会在校验阶段拒绝,而非靠 SQLite "no such column" 兜底报错)。 /// 专用更新路径的列不列入:knowledges.embedding(set_embedding)、projects.deleted_at -/// (soft_delete/restore)。未登记的表返回 None → 放行(仅靠参数化防注入,向后兼容)。 +/// (soft_delete/restore)、tasks.deleted_at(soft_delete/restore)。未登记的表返回 None +/// → 放行(仅靠参数化防注入,向后兼容)。 pub fn allowed_columns_for(table: &str) -> Option<&'static [&'static str]> { Some(match table { "ideas" => &[ @@ -322,8 +323,13 @@ pub fn allowed_columns_for(table: &str) -> Option<&'static [&'static str]> { "updated_at", ], "tasks" => &[ - "id", "project_id", "title", "description", "status", "priority", "branch_name", - "assignee", "workflow_def_id", "base_branch", "created_at", "updated_at", + // id/created_at 不列入:主键与创建时间不可通过通用 update_field 改写 + // (防篡改主键/伪造创建时间/跨项目移动)。 + "project_id", "title", "description", "status", "priority", "branch_name", + "assignee", "workflow_def_id", "base_branch", "updated_at", + // TODO(B-260616-16): project_id 跨表存在性校验待 commands/task.rs 层补。 + // 通用 CRUD 层(db repo)只懂表/列语义,不持有跨表业务约束(查 projects 表存在性)。 + // project_id 当前可在白名单内改写,合法目标存在性由上层命令层校验。 ], "releases" => &[ "id", "project_id", "version", "status", "task_ids", "changelog", "created_at", @@ -464,6 +470,7 @@ fn task_from_row(row: &Row<'_>) -> std::result::Result |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, created_at, updated_at) - VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12)", + "INSERT INTO tasks (id, project_id, title, description, status, priority, branch_name, assignee, workflow_def_id, base_branch, review_rounds, created_at, updated_at) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13)", 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.created_at, rec.updated_at + rec.review_rounds, 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, updated_at = ?10 WHERE id = ?11", + "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, updated_at = ?11 WHERE id = ?12", 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.updated_at, rec.id + rec.review_rounds, rec.updated_at, rec.id ], ) } ); +impl TaskRepo { + /// 列出未删除任务(deleted_at IS NULL)— 对标 ProjectRepo::list_active + /// + /// 显式列出 13 个 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, created_at, updated_at FROM tasks WHERE deleted_at IS NULL ORDER BY created_at DESC") + .map_err(|e| Error::Storage(e.to_string()))?; + let rows = stmt + .query_map([], |row| task_from_row(row)) + .map_err(|e| Error::Storage(e.to_string()))?; + let mut results = Vec::new(); + for r in rows { + results.push(r.map_err(|e| Error::Storage(e.to_string()))?); + } + Ok(results) + }) + .await + .map_err(|e| Error::Storage(e.to_string()))? + } + + /// 软删:标记 deleted_at(进回收站,可恢复)。仅作用于未删任务,返回是否命中。 + /// 对标 ProjectRepo::soft_delete。 + pub async fn soft_delete(&self, id: &str) -> Result { + 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 deleted_at = ?1, updated_at = ?1 WHERE id = ?2 AND deleted_at IS NULL", + params![now, id], + ) + .map_err(|e| Error::Storage(e.to_string()))?; + Ok(affected > 0) + }) + .await + .map_err(|e| Error::Storage(e.to_string()))? + } + + /// 恢复:清 deleted_at(从回收站还原)。仅作用于已删任务,返回是否命中。 + /// 对标 ProjectRepo::restore。 + pub async fn restore(&self, id: &str) -> Result { + 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 deleted_at = NULL, updated_at = ?1 WHERE id = ?2 AND deleted_at IS NOT NULL", + params![now, id], + ) + .map_err(|e| Error::Storage(e.to_string()))?; + Ok(affected > 0) + }) + .await + .map_err(|e| Error::Storage(e.to_string()))? + } + + /// 原子推进任务状态(任务推进链 F-260616-02 唯一 status 写入路径) + /// + /// 下沉 SQL `WHERE id=? AND status=?expected` 做 CAS(Compare-And-Swap)防 TOCTOU: + /// 并发推进/旁路修改若已改 status,affected_rows==0,本方法返回 None,调用方 + /// (task_advance_node)据此报「状态已变,推进中止」。`review_rounds` 不进通用 + /// update_field 白名单(收口:仅本方法可改 status 与 review_rounds)。 + /// + /// - `expected`:调用方读取的当前 status(状态机校验时的 from),CAS 前置。 + /// - `new_status`:目标 status(状态机 can_transition 已校验合法)。 + /// - `bump_rounds`:退回转换(in_review→in_progress / testing→in_review)传 true, + /// 一并 `review_rounds = review_rounds + 1`(同 UPDATE 原子,避免读改写竞争)。 + /// 前向推进 / 进出 blocked / 进 cancelled 传 false,不动 review_rounds。 + /// + /// 返回:成功推进返回更新后的 TaskRecord;affected==0(状态已变/任务不存在)返回 None。 + pub async fn advance_status_atomic( + &self, + id: &str, + expected: &str, + new_status: &str, + bump_rounds: bool, + ) -> Result> { + let conn = self.conn.clone(); + let id = id.to_owned(); + let expected = expected.to_owned(); + let new_status = new_status.to_owned(); + let now = now_millis_str(); + tokio::task::spawn_blocking(move || { + let guard = conn.blocking_lock(); + // CAS:WHERE id AND status=expected 锁定当前态;affected==0 即并发已改动。 + let sql = if bump_rounds { + "UPDATE tasks SET status = ?1, review_rounds = review_rounds + 1, updated_at = ?2 \ + WHERE id = ?3 AND status = ?4" + } else { + "UPDATE tasks SET status = ?1, updated_at = ?2 \ + WHERE id = ?3 AND status = ?4" + }; + let affected = guard + .execute(sql, params![new_status, now, id, expected]) + .map_err(|e| Error::Storage(e.to_string()))?; + if affected == 0 { + return Ok(None); + } + // 回读更新后的记录(含新 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, created_at, updated_at FROM tasks WHERE id = ?1") + .map_err(|e| Error::Storage(e.to_string()))?; + let row = stmt + .query_row(params![id], |row| task_from_row(row)) + .optional() + .map_err(|e| Error::Storage(e.to_string()))?; + Ok(row) + }) + .await + .map_err(|e| Error::Storage(e.to_string()))? + } + + /// 列出回收站(deleted_at IS NOT NULL),按更新时间(≈删除时间)降序。对标 ProjectRepo::list_deleted。 + /// + /// 注:任务表无专用 list_active_by_project 方法,按项目列活跃任务由 commands/task.rs + /// 的 list_tasks 用 list_active 后内存过滤 project_id 实现(任务量小,无需 SQL 下推)。 + pub async fn list_deleted(&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, created_at, updated_at FROM tasks WHERE deleted_at IS NOT NULL ORDER BY updated_at DESC") + .map_err(|e| Error::Storage(e.to_string()))?; + let rows = stmt + .query_map([], |row| task_from_row(row)) + .map_err(|e| Error::Storage(e.to_string()))?; + let mut results = Vec::new(); + for r in rows { + results.push(r.map_err(|e| Error::Storage(e.to_string()))?); + } + Ok(results) + }) + .await + .map_err(|e| Error::Storage(e.to_string()))? + } +} + impl_repo!( /// 分支表 CRUD BranchRepo, diff --git a/crates/df-storage/src/migrations.rs b/crates/df-storage/src/migrations.rs index 68d1432..d6606e9 100644 --- a/crates/df-storage/src/migrations.rs +++ b/crates/df-storage/src/migrations.rs @@ -84,6 +84,14 @@ pub fn run(conn: &Connection) -> Result<()> { migrate_v13(conn)?; } + if current_version < 14 { + migrate_v14(conn)?; + } + + if current_version < 15 { + migrate_v15(conn)?; + } + Ok(()) } @@ -267,6 +275,41 @@ fn migrate_v13(conn: &Connection) -> Result<()> { Ok(()) } +/// V14: 幂等补 tasks.deleted_at 列(软删回收站,对标 projects.deleted_at V11) +/// +/// 删除任务改为软删:deleted_at NULL=正常,非空=已进回收站(可恢复)。 +/// 与 projects.soft_delete 同模板:TaskRecord 不带该字段,纯靠 SQL WHERE deleted_at IS NULL +/// 过滤;子表(branches)不动,FK 仍满足,任务数据完整保留待恢复。 +/// 用 PRAGMA 探测列存在性,缺失才 ALTER(同 v4/v5/v6/v8/v10/v11 模式)。 +fn migrate_v14(conn: &Connection) -> Result<()> { + if !column_exists(conn, "tasks", "deleted_at") { + conn.execute("ALTER TABLE tasks ADD COLUMN deleted_at TEXT", [])?; + tracing::info!("v14: 补建 tasks.deleted_at 列(软删回收站)"); + } + conn.execute("INSERT INTO schema_version (version) VALUES (?)", [14])?; + tracing::info!("迁移 v14 完成"); + Ok(()) +} + +/// V15: 幂等补 tasks.review_rounds 列(review 退回累计轮数,F-260616-04) +/// +/// 任务推进链状态机退回时累加:in_review→in_progress / testing→in_review 各 +1, +/// 由 advance_task(df-nodes::task_advance_node)原子写入。默认 0(从未退回过的任务)。 +/// NOT NULL DEFAULT 0 保证老库行迁移后取值确定(非 NULL),TaskRecord 字段为 i32(非 Option)。 +/// 用 PRAGMA 探测列存在性,缺失才 ALTER(同 v4/v5/v6/v8/v10/v11/v14 模式)。 +fn migrate_v15(conn: &Connection) -> Result<()> { + if !column_exists(conn, "tasks", "review_rounds") { + conn.execute( + "ALTER TABLE tasks ADD COLUMN review_rounds INTEGER NOT NULL DEFAULT 0", + [], + )?; + tracing::info!("v15: 补建 tasks.review_rounds 列(review 退回累计轮数)"); + } + conn.execute("INSERT INTO schema_version (version) VALUES (?)", [15])?; + tracing::info!("迁移 v15 完成"); + Ok(()) +} + /// V1 建表 SQL const V1_SQL: &str = " -- 想法表 @@ -301,7 +344,8 @@ CREATE TABLE IF NOT EXISTS tasks ( title TEXT NOT NULL, description TEXT NOT NULL DEFAULT '', status TEXT NOT NULL DEFAULT 'todo', - priority INTEGER NOT NULL DEFAULT 1, + -- priority 默认 2 对齐 task.rs default_priority()=2(medium) + priority INTEGER NOT NULL DEFAULT 2, branch_name TEXT, assignee TEXT, created_at TEXT NOT NULL, diff --git a/crates/df-storage/src/models.rs b/crates/df-storage/src/models.rs index c6c4a00..94f0baa 100644 --- a/crates/df-storage/src/models.rs +++ b/crates/df-storage/src/models.rs @@ -61,6 +61,11 @@ pub struct TaskRecord { pub assignee: Option, pub workflow_def_id: Option, // 关联的工作流定义 ID pub base_branch: Option, // 基础分支 + /// review 退回累计轮数(in_review→in_progress / testing→in_review 时 +1, + /// 由 advance_task 原子写入,见 df-nodes::task_advance_node)。 + /// #[serde(default)] 兼容旧前端无该字段的 JSON(老任务记录默认 0)。 + #[serde(default)] + pub review_rounds: i32, pub created_at: String, pub updated_at: String, } diff --git a/crates/df-storage/tests/project_soft_delete.rs b/crates/df-storage/tests/project_soft_delete.rs index 3548173..2a38228 100644 --- a/crates/df-storage/tests/project_soft_delete.rs +++ b/crates/df-storage/tests/project_soft_delete.rs @@ -41,6 +41,7 @@ fn task(id: &str, project_id: &str) -> TaskRecord { assignee: None, workflow_def_id: None, base_branch: None, + review_rounds: 0, created_at: now_ts(), updated_at: now_ts(), } diff --git a/src-tauri/src/commands/ai/tool_registry.rs b/src-tauri/src/commands/ai/tool_registry.rs index c0533ca..cbd7422 100644 --- a/src-tauri/src/commands/ai/tool_registry.rs +++ b/src-tauri/src/commands/ai/tool_registry.rs @@ -2,7 +2,9 @@ use std::collections::HashMap; use std::path::{Path, PathBuf}; -use std::sync::Arc; +use std::sync::{Arc, LazyLock}; + +use tokio::sync::Mutex as TokioMutex; use df_ai::ai_tools::{AiToolRegistry, RiskLevel}; use df_execute::shell::{execute, ShellRequest}; @@ -13,6 +15,58 @@ use df_core::types::new_id; use crate::commands::now_millis; +/// CRUD list 工具的默认返回上限（防 LLM context 膨胀） +/// 用于 list_projects / list_tasks / list_ideas / list_trash +const MAX_LIST_RESULTS: usize = 50; + +/// 生成行级 unified diff（无外部依赖，基于 LCS）。 +/// 仅标 +/- 前缀，不做 hunk header（足够审批卡/审计留痕可读）。 +/// 文件改动通常集中在 old_text/new_text 局部，整体行对比可直观呈现。 +fn generate_diff(old: &str, new: &str) -> String { + let a: Vec<&str> = old.lines().collect(); + let b: Vec<&str> = new.lines().collect(); + let (n, m) = (a.len(), b.len()); + + // LCS 动态规划表（usize 即可；大文件已被 1MB 限制挡住，行数有限） + let mut dp = vec![vec![0usize; m + 1]; n + 1]; + for i in (0..n).rev() { + for j in (0..m).rev() { + dp[i][j] = if a[i] == b[j] { + dp[i + 1][j + 1] + 1 + } else { + dp[i + 1][j].max(dp[i][j + 1]) + }; + } + } + + // 回溯：相同行原样输出(空格前缀)，旧独有加 '-'，新独有加 '+' + let mut out = String::new(); + let (mut i, mut j) = (0usize, 0usize); + let mut changes = 0usize; + while i < n && j < m { + if a[i] == b[j] { + out.push_str(" "); out.push_str(a[i]); out.push('\n'); + i += 1; j += 1; + } else if dp[i + 1][j] >= dp[i][j + 1] { + out.push_str("-"); out.push_str(a[i]); out.push('\n'); + i += 1; changes += 1; + } else { + out.push_str("+"); out.push_str(b[j]); out.push('\n'); + j += 1; changes += 1; + } + } + while i < n { out.push_str("-"); out.push_str(a[i]); out.push('\n'); i += 1; changes += 1; } + while j < m { out.push_str("+"); out.push_str(b[j]); out.push('\n'); j += 1; changes += 1; } + + // 极端情况：差异行过多（>300），截断防 LLM context 爆炸 + if changes > 300 { + let kept: String = out.lines().take(300).collect::>().join("\n"); + return format!("{}\n... (diff 已截断，共 {} 处变更行)", kept, changes); + } + // 末尾多余换行清理（保持单尾换行） + out.trim_end_matches('\n').to_string() + "\n" +} + /// 验证文件路径：禁止访问系统敏感目录 fn validate_path(path: &str) -> anyhow::Result<()> { // 规范化为反斜杠：LLM 可能传正斜杠绕过黑名单（Windows tokio::fs 两种分隔符都吃） @@ -54,6 +108,13 @@ fn truncate_output(s: &str, max: usize) -> (String, bool) { ) } +/// 全局文件锁表：每个路径一把互斥锁（L1 防护，防同文件并发读写冲突） +/// +/// 唯一并行点: audit.rs join_all — Low/Medium 风险工具并行执行。 +/// 单用户桌面应用，Mutex 够用且简单（~15 行），不需要全局文件队列。 +static FILE_LOCKS: LazyLock>> = + LazyLock::new(|| TokioMutex::new(HashMap::new())); + /// workspace 根目录（项目根 = src-tauri 上两级，编译期固定） fn workspace_root() -> PathBuf { PathBuf::from(env!("CARGO_MANIFEST_DIR")) @@ -148,8 +209,8 @@ pub fn build_ai_tool_registry(db: &Arc) -> AiToolRegistry { let repo = df_storage::crud::ProjectRepo::new(&db); let mut items = repo.list_active().await?; // list_active 排除回收站(deleted_at)，防 LLM 看到已软删项目 let total = items.len(); - let truncated = total > 50; - items.truncate(50); // 防 LLM context 膨胀 + let truncated = total > MAX_LIST_RESULTS; + items.truncate(MAX_LIST_RESULTS); Ok(serde_json::json!({ "items": items, "truncated": truncated })) }) })}, @@ -167,8 +228,8 @@ pub fn build_ai_tool_registry(db: &Arc) -> AiToolRegistry { repo.list_all().await? }; let total = tasks.len(); - let truncated = total > 50; - tasks.truncate(50); // 防 LLM context 膨胀 + let truncated = total > MAX_LIST_RESULTS; + tasks.truncate(MAX_LIST_RESULTS); Ok(serde_json::json!({ "items": tasks, "truncated": truncated })) }) })}, @@ -182,8 +243,8 @@ pub fn build_ai_tool_registry(db: &Arc) -> AiToolRegistry { let repo = df_storage::crud::IdeaRepo::new(&db); let mut items = repo.list_all().await?; let total = items.len(); - let truncated = total > 50; - items.truncate(50); // 防 LLM context 膨胀 + let truncated = total > MAX_LIST_RESULTS; + items.truncate(MAX_LIST_RESULTS); Ok(serde_json::json!({ "items": items, "truncated": truncated })) }) })}, @@ -275,6 +336,7 @@ pub fn build_ai_tool_registry(db: &Arc) -> AiToolRegistry { // priority 默认 2(medium):与 commands::task::default_priority 一致,新任务默认中优先级(非 high) status: "todo".to_string(), priority: args["priority"].as_i64().unwrap_or(2) as i32, branch_name: None, assignee: None, workflow_def_id: None, base_branch: None, + review_rounds: 0, created_at: now_millis(), updated_at: now_millis(), }; let id = record.id.clone(); @@ -303,17 +365,17 @@ pub fn build_ai_tool_registry(db: &Arc) -> AiToolRegistry { }) })}, ); - // FR-D6:delete_task 对齐 commands::task::delete_task —— tasks 表无 deleted_at,走硬删 - // (与 delete_project 的软删不同:projects 有专门回收站机制,tasks 没有,只能硬删) + // FR-D6:delete_task 对齐 commands::task::delete_task(D-260616-02 软删除) + // tasks 表 V14 加 deleted_at,delete_task 改调 soft_delete 移入回收站(对标 delete_project) registry.register( - "delete_task", "删除任务（硬删,不可恢复。清理孤儿任务时务必用本工具,不要误用 delete_project）", + "delete_task", "删除任务（移入回收站，可恢复）", df_ai::ai_tools::object_schema(vec![("id", "string", true)]), RiskLevel::High, { let db = db.clone(); Box::new(move |args: serde_json::Value| { let db = db.clone(); Box::pin(async move { let id = args["id"].as_str().ok_or_else(|| anyhow::anyhow!("缺少 id"))?; let repo = df_storage::crud::TaskRepo::new(&db); - let deleted = repo.delete(id).await?; + let deleted = repo.soft_delete(id).await?; Ok(serde_json::json!({ "deleted": deleted, "id": id })) }) })}, @@ -390,26 +452,19 @@ pub fn build_ai_tool_registry(db: &Arc) -> AiToolRegistry { })}, ); registry.register( - "list_trash", "列出回收站已删除项目", + "list_trash", "列出回收站已删除项目（最多返回 50 条）", df_ai::ai_tools::object_schema(vec![]), RiskLevel::Low, { let db = db.clone(); Box::new(move |_args: serde_json::Value| { let db = db.clone(); Box::pin(async move { let repo = df_storage::crud::ProjectRepo::new(&db); let mut items = repo.list_deleted().await?; - items.truncate(50); // 防 LLM context 膨胀 - Ok(serde_json::to_value(items)?) + let truncated = items.len() > MAX_LIST_RESULTS; + items.truncate(MAX_LIST_RESULTS); + Ok(serde_json::json!({ "items": items, "truncated": truncated })) }) })}, ); - registry.register( - "run_workflow", "运行指定的工作流 DAG", - df_ai::ai_tools::object_schema(vec![("name", "string", true), ("dag", "object", true)]), RiskLevel::High, - Box::new(|_args: serde_json::Value| Box::pin(async move { - // run_workflow 需完整 DAG 执行，返回提示由前端触发 - Ok(serde_json::json!({ "note": "请通过工作流页面运行工作流", "tool": "run_workflow" })) - })), - ); registry.register( "run_command", "在指定工作目录执行 shell 命令（跑测试/构建/查看运行结果），返回 stdout/stderr/exit_code。高风险，须人工批准。命令需自包含（非交互式，避免需用户输入的程序）。默认超时 60 秒。用于验证刚写入的代码能否运行、跑测试、看报错迭代修改。", df_ai::ai_tools::object_schema(vec![ @@ -440,6 +495,7 @@ pub fn build_ai_tool_registry(db: &Arc) -> AiToolRegistry { working_dir: Some(working_dir.clone()), env: HashMap::new(), timeout_secs: Some(timeout_secs), + shell_type: Default::default(), }; let result = execute(request).await?; @@ -462,8 +518,8 @@ pub fn build_ai_tool_registry(db: &Arc) -> AiToolRegistry { // ── 文件系统 ── registry.register( - "read_file", "读取文件内容，返回文本内容。支持 offset 和 limit 参数分页读取大文件", - df_ai::ai_tools::object_schema(vec![("path", "string", true), ("offset", "integer", false), ("limit", "integer", false)]), + "read_file", "读取文件内容，返回文本内容。支持 offset/limit 分页；传入 search 则在文件内容中搜索匹配行（大小写敏感，字符串包含匹配），返回 matches 数组（限50条）", + df_ai::ai_tools::object_schema(vec![("path", "string", true), ("offset", "integer", false), ("limit", "integer", false), ("search", "string", false)]), RiskLevel::Low, Box::new(|args: serde_json::Value| Box::pin(async move { let resolved = resolve_workspace_path( @@ -492,7 +548,26 @@ pub fn build_ai_tool_registry(db: &Arc) -> AiToolRegistry { } anyhow::bail!("读取文件失败: {}", e); } - // limit 硬上限 2000 行(防 LLM 传超大 limit 读全文件,1MB 限下仍可能数万行) + // search 模式: 按行枚举收集含 search 子串的行,限 50 条 + if let Some(search) = args["search"].as_str() { + let search_max = 50usize; + let mut matches_vec: Vec = Vec::with_capacity(search_max); + for (idx, line) in content.lines().enumerate() { + if line.contains(search) { + matches_vec.push(serde_json::json!({ "line": idx + 1, "content": line })); + if matches_vec.len() >= search_max { break; } + } + } + let total = content.lines().filter(|l| l.contains(search)).count(); + return Ok(serde_json::json!({ + "path": path, "size": metadata.len(), + "search": search, + "matches": matches_vec, + "total": total, + "has_more": total > search_max, + })); + } + // 默认分页模式: limit 硬上限 2000 行(防 LLM 传超大 limit 读全文件,1MB 限下仍可能数万行) let result = if let Some(offset) = args["offset"].as_u64() { let lines: Vec<&str> = content.lines().collect(); let skip = offset as usize; @@ -523,18 +598,34 @@ pub fn build_ai_tool_registry(db: &Arc) -> AiToolRegistry { })), ); registry.register( - "write_file", "写入或创建文件，自动创建不存在的父目录", - df_ai::ai_tools::object_schema(vec![("path", "string", true), ("content", "string", true)]), + "write_file", "写入或创建文件，自动创建不存在的父目录。支持 encoding=\"base64\" 写入二进制/非 UTF-8 内容（图片/PDF/Excel 等），content 传 base64 编码字符串；默认 encoding=\"utf-8\" 传纯文本", + df_ai::ai_tools::object_schema(vec![ + ("path", "string", true), + ("content", "string", true), + ("encoding", "string", false), + ]), RiskLevel::Medium, Box::new(|args: serde_json::Value| Box::pin(async move { let resolved = resolve_workspace_path( args["path"].as_str().ok_or_else(|| anyhow::anyhow!("缺少 path 参数"))?, )?; let path = resolved.to_str().ok_or_else(|| anyhow::anyhow!("路径含非法字符"))?; - let content = args["content"].as_str().ok_or_else(|| anyhow::anyhow!("缺少 content 参数"))?; + let raw = args["content"].as_str().ok_or_else(|| anyhow::anyhow!("缺少 content 参数"))?; + // F-260615-11: encoding 参数支持 base64,写二进制(图片/PDF/Excel 等) + // 默认 utf-8:走原 &str 字节写入;base64:解码为 Vec 写字节 + let encoding = args.get("encoding").and_then(|v| v.as_str()).unwrap_or("utf-8"); + let bytes: Vec = if encoding.eq_ignore_ascii_case("base64") { + use base64::{engine::general_purpose::STANDARD, Engine as _}; + STANDARD.decode(raw) + .map_err(|e| anyhow::anyhow!("base64 解码失败: {}", e))? + } else if encoding.eq_ignore_ascii_case("utf-8") { + raw.as_bytes().to_vec() + } else { + anyhow::bail!("不支持的 encoding: {}（仅 utf-8 / base64）", encoding); + }; // 写入大小上限(FR-S2:原无限制,LLM 经审批可写超大文件撑爆磁盘/对话历史) - if content.len() > 1_048_576 { - anyhow::bail!("写入内容超过 1MB 限制 ({} 字节)", content.len()); + if bytes.len() > 1_048_576 { + anyhow::bail!("写入内容超过 1MB 限制 ({} 字节)", bytes.len()); } let target = std::path::Path::new(path); // FR-S7 覆盖防护:覆盖非空文件前自动 .bak 备份(防 LLM 误用 write_file 当 edit 致数据彻底丢失) @@ -559,7 +650,7 @@ pub fn build_ai_tool_registry(db: &Arc) -> AiToolRegistry { } // FR-S7 原子写:tmp→rename,避免写到一半崩溃留半成品(.tmp-write 同目录保证 rename 不跨卷) let tmp = format!("{}.tmp-write", path); - if let Err(e) = tokio::fs::write(&tmp, content).await { + if let Err(e) = tokio::fs::write(&tmp, &bytes).await { let _ = tokio::fs::remove_file(&tmp).await; return Err(anyhow::anyhow!("写入临时文件失败: {}", e)); } @@ -576,11 +667,140 @@ pub fn build_ai_tool_registry(db: &Arc) -> AiToolRegistry { } // FR-S7 大小异动 warn:新内容远小于旧(疑似误覆盖整文件),提示用户查 .bak if let Some(old) = old_size { - if old > 0 && (content.len() as f64 / old as f64) < 0.1 { - tracing::warn!("write_file 疑似误覆盖: {} {}→{} 字节(缩减>90%),.bak 已备份", path, old, content.len()); + if old > 0 && (bytes.len() as f64 / old as f64) < 0.1 { + tracing::warn!("write_file 疑似误覆盖: {} {}→{} 字节(缩减>90%),.bak 已备份", path, old, bytes.len()); } } - Ok(serde_json::json!({ "path": path, "bytes_written": content.len(), "old_size": old_size })) + Ok(serde_json::json!({ "path": path, "bytes_written": bytes.len(), "old_size": old_size, "encoding": encoding })) + })), + ); + + // ── 局部文件编辑 (Medium risk) ── + registry.register( + "patch_file", "局部更新文件内容。用于精确修改文件的特定部分（而非全量覆盖）。每个补丁指定 old_text（要替换的原文，必须精确匹配含空格/缩进）和 new_text（新内容）。可选 line 辅助定位。属 Medium 风险操作（修改已有文件），需人工审批。注意：若文件已被外部修改，请先重新 read_file 获取最新内容", + df_ai::ai_tools::object_schema(vec![ + ("path", "string", true), + ("old_text", "string", true), + ("new_text", "string", true), + ("line", "integer", false), + ("expected_hash", "string", false), + ]), + RiskLevel::Medium, + Box::new(|args: serde_json::Value| Box::pin(async move { + let resolved = resolve_workspace_path( + args["path"].as_str().ok_or_else(|| anyhow::anyhow!("缺少 path 参数"))?, + )?; + let path = resolved.to_str().ok_or_else(|| anyhow::anyhow!("路径含非法字符"))?; + let old_text = args["old_text"].as_str() + .ok_or_else(|| anyhow::anyhow!("缺少 old_text 参数"))?; + let new_text = args["new_text"].as_str() + .ok_or_else(|| anyhow::anyhow!("缺少 new_text 参数"))?; + + // 边界校验 + if old_text.is_empty() { + anyhow::bail!("old_text 不能为空"); + } + if old_text == new_text { + return Ok(serde_json::json!({ + "path": path, "changed": false, "warning": "new_text 与 old_text 相同，无实际更改" + })); + } + + let target = std::path::Path::new(path); + if !target.exists() { + anyhow::bail!("文件不存在: {}", path); + } + let file_meta = tokio::fs::metadata(target).await + .map_err(|e| anyhow::anyhow!("无法访问文件 {}: {}", path, e))?; + if file_meta.len() > 1_048_576 { + anyhow::bail!("文件超过 1MB 限制 ({} 字节)", file_meta.len()); + } + + // 阶段一：读文件内容 + 校验（无锁，纯读操作） + use tokio::io::AsyncReadExt; + let mut file = tokio::fs::File::open(path).await + .map_err(|e| anyhow::anyhow!("读取文件失败 {}: {}", path, e))?; + let mut content = String::new(); + file.read_to_string(&mut content).await + .map_err(|e| anyhow::anyhow!("读取文件失败: {}", e))?; + + // 二进制检测 + if content.contains('\0') { + anyhow::bail!("不支持二进制文件"); + } + + // L3: expected_hash 指纹校验（防外部修改） + if let Some(expected) = args["expected_hash"].as_str() { + let modified = file_meta.modified() + .ok().and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok()) + .map(|d| d.as_secs()); + let current_hash = format!("{}_{}", modified.unwrap_or(0), file_meta.len()); + if current_hash != expected { + anyhow::bail!( + "文件已被外部修改(hash 不匹配): 期望={} 实际={}，请重新 read_file 获取最新内容", + expected, current_hash + ); + } + } + + // L2: old_text 精确匹配（CAS 语义） + if !content.contains(old_text) { + anyhow::bail!("未找到目标文本，文件可能已被修改"); + } + + // 多匹配检测 + let match_count = content.matches(old_text).count(); + let warning = if match_count > 1 { + Some(format!("匹配到 {} 处，仅替换第 1 处", match_count)) + } else { None }; + + // 执行替换（仅替换第 1 处） + let new_content = content.replacen(old_text, new_text, 1); + + // 阶段二：L1 tokio::Mutex 保护写序列（backup → tmp write → rename → cleanup） + // tokio::sync::Mutex 的 MutexGuard 是 Send，可安全跨 await + let abs_path = target.canonicalize() + .map_err(|e| anyhow::anyhow!("路径解析失败: {}", e))?; + { + let mut locks = FILE_LOCKS.lock().await; + locks.entry(abs_path.clone()).or_insert(()); + + // .bak 备份 + let bak = format!("{}.bak", path); + tokio::fs::copy(path, &bak).await + .map_err(|e| anyhow::anyhow!("备份 .bak 失败: {}", e))?; + + // 原子写: tmp → rename + let tmp = format!("{}.tmp-write", path); + if let Err(e) = tokio::fs::write(&tmp, &new_content).await { + let _ = tokio::fs::remove_file(&tmp).await; + let _ = tokio::fs::remove_file(&bak).await; + return Err(anyhow::anyhow!("写入临时文件失败: {}", e)); + } + if let Err(e) = tokio::fs::rename(&tmp, path).await { + let _ = tokio::fs::remove_file(&tmp).await; + return Err(anyhow::anyhow!("原子替换失败: {}，备份保留在 {}", e, bak)); + } + // 成功：清理 .bak + let _ = tokio::fs::remove_file(&bak).await; + // _locks 在此 drop，释放锁 + } + + let size_diff = new_content.len() as i64 - content.len() as i64; + // 生成 unified diff 供前端审批卡/审计留痕展示 + // content（旧）与 new_content（新）作用域内可用，无需重读文件 + let diff = generate_diff(&content, &new_content); + let mut result = serde_json::json!({ + "path": path, + "changed": true, + "size_diff": size_diff, + "matches_found": match_count, + "diff": diff, + }); + if let Some(w) = warning { + result["warning"] = serde_json::Value::String(w); + } + Ok(result) })), ); @@ -647,6 +867,160 @@ pub fn build_ai_tool_registry(db: &Arc) -> AiToolRegistry { })), ); + // ── 删除文件 (High risk, F-260615-16) ── + registry.register( + "delete_file", "删除文件。默认（permanent=false）软删除：移到工作区 .trash/ 目录保留可恢复；permanent=true 硬删除不可恢复。删除前备份，回收站机制对齐 list_trash。高风险，须人工批准", + df_ai::ai_tools::object_schema(vec![ + ("path", "string", true), + ("permanent", "boolean", false), + ]), + RiskLevel::High, + Box::new(|args: serde_json::Value| Box::pin(async move { + // validate_path 前置(workspace 内)——resolve_workspace_path 含 validate_path + symlink 防逃逸 + let resolved = resolve_workspace_path( + args["path"].as_str().ok_or_else(|| anyhow::anyhow!("缺少 path 参数"))?, + )?; + let path = resolved.to_str().ok_or_else(|| anyhow::anyhow!("路径含非法字符"))?; + let permanent = args.get("permanent").and_then(|v| v.as_bool()).unwrap_or(false); + let target = std::path::Path::new(path); + if !target.exists() { + anyhow::bail!("文件不存在: {}", path); + } + // 拒绝目录:delete_file 仅删文件(删目录语义复杂+递归风险大,留 run_command 兜底) + let metadata = tokio::fs::metadata(target).await + .map_err(|e| anyhow::anyhow!("无法访问 {}: {}", path, e))?; + if metadata.is_dir() { + anyhow::bail!("delete_file 仅支持删除文件，不支持目录: {}", path); + } + let file_name = target.file_name() + .ok_or_else(|| anyhow::anyhow!("无法解析文件名: {}", path))? + .to_string_lossy().to_string(); + + if permanent { + // 硬删除:不可逆,对齐 delete_project purge 语义 + tokio::fs::remove_file(path).await + .map_err(|e| anyhow::anyhow!("删除失败: {}", e))?; + Ok(serde_json::json!({ + "path": path, + "deleted": true, + "permanent": true, + "backed_up": false, + })) + } else { + // 软删除:移到 .trash/-,可恢复(对齐 list_trash 机制) + // .trash 放 workspace_root 下,validate_path 已确认 path 在 workspace 内 + let trash_dir = workspace_root().join(".trash"); + tokio::fs::create_dir_all(&trash_dir).await + .map_err(|e| anyhow::anyhow!("创建 .trash 目录失败: {}", e))?; + let backup_name = format!("{}-{}", new_id(), file_name); + let backup_path = trash_dir.join(&backup_name); + let backup_path_str = backup_path.to_string_lossy().to_string(); + tokio::fs::rename(path, &backup_path).await + .map_err(|e| anyhow::anyhow!("移入回收站失败: {}", e))?; + Ok(serde_json::json!({ + "path": path, + "deleted": true, + "permanent": false, + "backed_up": true, + "backup_path": backup_path_str, + })) + } + })), + ); + + // ── 重命名/移动文件 (Medium risk, F-260615-15) ── + // 一个工具覆盖 rename(同目录改名) + move(跨目录移动)。绕开 shell.rs PS/cmd 中文路径 GBK 解码 + // 静默失败根因:Rust 原生 std::fs 走 Win32 UTF-16 路径,中文路径零字符集问题。 + registry.register( + "rename_file", "重命名或移动文件（同一工具覆盖 rename 和 move）。参数：from（源路径）+ to（目标路径）+ overwrite（默认 false，目标存在则拒绝）。同卷走 tokio::fs::rename（原子），跨卷自动降级 copy+remove（非原子，失败回滚删 to 保 from 完整）。须人工批准", + df_ai::ai_tools::object_schema(vec![ + ("from", "string", true), + ("to", "string", true), + ("overwrite", "boolean", false), + ]), + RiskLevel::Medium, + Box::new(|args: serde_json::Value| Box::pin(async move { + // 双向 validate_path(workspace 内)——from/to 都走 resolve_workspace_path, + // 各自含 validate_path(黑名单+.. 防遍历) + 词法 starts_with + canonicalize symlink 防逃逸 + let from_resolved = resolve_workspace_path( + args["from"].as_str().ok_or_else(|| anyhow::anyhow!("缺少 from 参数"))?, + )?; + let from_path = from_resolved.to_str().ok_or_else(|| anyhow::anyhow!("from 路径含非法字符"))?; + let to_resolved = resolve_workspace_path( + args["to"].as_str().ok_or_else(|| anyhow::anyhow!("缺少 to 参数"))?, + )?; + let to_path = to_resolved.to_str().ok_or_else(|| anyhow::anyhow!("to 路径含非法字符"))?; + let overwrite = args.get("overwrite").and_then(|v| v.as_bool()).unwrap_or(false); + + // 源存在性 + 拒目录(rename 目录语义复杂递归风险大,留 run_command 兜底) + let from_target = std::path::Path::new(from_path); + if !from_target.exists() { + anyhow::bail!("源文件不存在: {}", from_path); + } + let from_metadata = tokio::fs::metadata(from_target).await + .map_err(|e| anyhow::anyhow!("无法访问 {}: {}", from_path, e))?; + if from_metadata.is_dir() { + anyhow::bail!("rename_file 仅支持文件，不支持目录: {}", from_path); + } + let bytes_moved = from_metadata.len(); + + // 目标父目录不存在则创建(对齐 write_file L643/append_file L851,跨目录移动到不存在父目录否则 rename 失败) + let to_target = std::path::Path::new(to_path); + if let Some(parent) = to_target.parent() { + if !parent.starts_with(&workspace_root()) { + anyhow::bail!("禁止在项目目录之外创建目录"); + } + tokio::fs::create_dir_all(parent).await + .map_err(|e| anyhow::anyhow!("创建目标父目录失败: {}", e))?; + } + + // 目标存在性:overwrite=false 直接拒绝(防误覆盖);overwrite=true 先删 to + if to_target.exists() { + if !overwrite { + anyhow::bail!("目标已存在且 overwrite=false: {}", to_path); + } + tokio::fs::remove_file(to_path).await + .map_err(|e| anyhow::anyhow!("覆盖删除目标失败: {}", e))?; + } + + // 同卷:tokio::fs::rename 原子(Windows 走 MoveFileExW UTF-16,中文路径无 GBK 问题) + let rename_err = tokio::fs::rename(from_path, to_path).await.err(); + if rename_err.is_none() { + return Ok(serde_json::json!({ + "from": from_path, + "to": to_path, + "renamed": true, + "bytes_moved": bytes_moved, + "cross_volume": false, + })); + } + // rename 失败:跨卷(Windows ERROR_NOT_SAME_DEVICE 17)→ 降级 copy+remove + // 其他错误(权限/占用)直接抛,不降级 + let err = rename_err.unwrap(); + let cross_volume = err.raw_os_error() == Some(17); + if !cross_volume { + anyhow::bail!("重命名/移动失败: {}", err); + } + // 跨卷降级 copy + remove(非原子):copy 失败 from 完整(未动);copy 成功 remove 失败 + // 则 from/to 同时存在,删 to 回滚保 from 完整(对齐设计:失败回滚删 to) + if let Err(e) = tokio::fs::copy(from_path, to_path).await { + anyhow::bail!("跨卷复制失败（from 未改动）: {}", e); + } + if let Err(e) = tokio::fs::remove_file(from_path).await { + // remove 失败:回滚删 to,保 from 完整(用户可重试) + let _ = tokio::fs::remove_file(to_path).await; + anyhow::bail!("跨卷移动删除源失败已回滚（from 完整,可重试）: {}", e); + } + Ok(serde_json::json!({ + "from": from_path, + "to": to_path, + "renamed": true, + "bytes_moved": bytes_moved, + "cross_volume": true, + })) + })), + ); + // ── 文件搜索 (Low risk) ── registry.register( "search_files", "在指定目录下搜索匹配模式（字符串包含匹配）的文件名，返回路径和大小列表。支持递归搜索，结果限 50 条", @@ -726,11 +1100,65 @@ fn list_dir_recursive<'a>( }) } +/// 工具名 → 审批文案模板（供 build_approval_reason 使用）。 +/// +/// 返回 Option<(&str, &[&str])>：(动作前缀模板, 关键参数名列表)。 +/// 模板中 `{}` 占位符由 build_approval_reason 按参数列表从 args 取值填充。 +/// 未注册的工具返回 None，由调用方 fallback 到通用风险提示。 +pub(crate) fn display_hint_for_tool(name: &str) -> Option<(&'static str, &'static [&'static str])> { + let (template, keys): (&str, &[&str]) = match name { + "delete_project" => ("删除项目{}", &["id"]), + "restore_project" => ("从回收站恢复项目{}", &["id"]), + "purge_project" => ("永久删除项目及关联数据，不可恢复{}", &["id"]), + "update_project" => ("修改项目{}字段「{}」", &["id", "field"]), + "bind_directory" => ("绑定目录：{}（项目{}）", &["path", "id"]), + "create_task" => ("创建任务：{}（项目{}）", &["title", "project_id"]), + "create_project" => ("创建项目：「{}」", &["name"]), + "create_idea" => ("捕获灵感：{}", &["title"]), + "patch_file" => ("修改文件：{}", &["path"]), + "delete_file" => ("删除文件：{}（{}）", &["path", "permanent"]), + "rename_file" => ("重命名/移动：{} → {}", &["from", "to"]), + _ => return None, + }; + Some((template, keys)) +} + +/// 工具名 → 中文动作前缀（轻量映射，供 build_approval_reason 快速取可读标签）。 +/// +/// 仅返回动作动词+对象名（如 "创建任务"、"写入文件"），不含参数细节。 +/// 未注册的工具返回 None，由调用方 fallback 到 display_hint_for_tool 模板填充或硬编码。 +pub(crate) fn tool_display_hint(name: &str) -> Option<&'static str> { + match name { + "create_task" => Some("创建任务"), + "write_file" => Some("写入文件"), + "read_file" => Some("读取文件"), + "list_directory" => Some("列出目录"), + "search_files" => Some("搜索文件"), + "create_project" => Some("创建项目"), + "update_project" => Some("修改项目"), + "delete_project" => Some("删除项目"), + "restore_project" => Some("恢复项目"), + "purge_project" => Some("永久删除项目"), + "bind_directory" => Some("绑定目录"), + "create_idea" => Some("捕获灵感"), + "update_task" => Some("修改任务"), + "delete_task" => Some("删除任务"), + "update_idea" => Some("修改灵感"), + "delete_idea" => Some("删除灵感"), + "patch_file" => Some("修改文件"), + "append_file" => Some("追加写入"), + "delete_file" => Some("删除文件"), + "rename_file" => Some("重命名/移动"), + _ => None, + } +} + /// 判断是否为不应深入递归的噪音目录(构建产物/依赖/缓存等) fn is_noise_dir(name: &str) -> bool { const NOISE_DIRS: &[&str] = &[ ".git", "node_modules", "target", "dist", "build", ".next", ".cache", "__pycache__", ".venv", "venv", ".idea", + ".trash", // delete_file 软删除回收站(F-260615-16),不污染 list_directory ]; NOISE_DIRS.contains(&name) } diff --git a/src-tauri/src/commands/task.rs b/src-tauri/src/commands/task.rs index 19a6f3c..56cb644 100644 --- a/src-tauri/src/commands/task.rs +++ b/src-tauri/src/commands/task.rs @@ -27,17 +27,22 @@ fn default_priority() -> i32 { 2 // medium — 新任务默认中优先级(非 high),符合常识 } -/// 列出任务，可按 project_id 过滤 +/// 列出未删除任务(deleted_at IS NULL),可按 project_id 过滤。 +/// +/// 软删对标 projects:默认过滤回收站(语义与 list_projects 一致)。 +/// 无 project_id 时调 list_active(WHERE deleted_at IS NULL);有 project_id 时 +/// 先 list_active 再内存过滤 project_id(任务量小,无需 SQL 下推,避免新增专用查询方法)。 +/// 注意:不能用通用 query 宏——它不带 deleted_at 过滤,会把回收站任务也返回。 #[tauri::command] pub async fn list_tasks( state: State<'_, AppState>, project_id: Option, ) -> Result, String> { - let result = match project_id { - Some(pid) => state.tasks.query("project_id", &pid).await, - None => state.tasks.list_all().await, - }; - result.map_err(err_str) + let mut tasks = state.tasks.list_active().await.map_err(err_str)?; + if let Some(pid) = project_id { + tasks.retain(|t| t.project_id == pid); + } + Ok(tasks) } /// 按 id 查任务，找不到返回 Err(供前端详情页) @@ -72,6 +77,7 @@ pub async fn create_task( assignee: input.assignee, workflow_def_id: None, base_branch: None, + review_rounds: 0, created_at: now.clone(), updated_at: now, }; @@ -91,8 +97,8 @@ pub async fn update_task( field: String, value: String, ) -> Result { - // 字段名注入由 df-storage 白名单兜底；这里补 status 值校验， - // 拦截拼写错误（in-progess / "in progress" / 大小写错等）静默落库。 + // 字段名注入由 df-storage 白名单兜底；这里补 status / priority 值校验， + // 拦截拼写错误（in-progess / "in progress" / 大小写错）与越界数值（999 / "abc"）静默落库。 if field == "status" && !TaskStatus::is_valid(&value) { return Err(format!( "非法 status 值 {:?}，合法值: {:?}", @@ -100,6 +106,28 @@ pub async fn update_task( TaskStatus::valid_values() )); } + // priority 值域 0..=3（0=critical, 1=high, 2=medium, 3=low），与前端