新增: 多Agent数据层(V36迁移+3新表+Repo CRUD+7个测试全绿)

- V36: ai_plans/ai_subtasks/ai_conflicts 3 新表 + ai_messages/ai_tool_executions 加 subtask_id - models: PlanRecord/SubTaskRecord/ConflictRecord 结构体 - PlanRepo/SubTaskRepo/ConflictRepo CRUD(insert/get/list_by_*/update_status/resolve) - 7个 Repo 单元测试全部通过(内存 SQLite + 全迁移) - 顺带修复 task_repo.rs 预存测试编译错误(trec_full 参数类型+缺失 await)
2026-07-01 21:50:16 +08:00
parent f4fc0feff5
commit ae6d3d0043
6 changed files with 735 additions and 7 deletions
@@ -17,6 +17,9 @@
 //! re-export(`pub use ...::*`)保持 `df_storage::crud::XxxRepo` /
 //! `df_storage::crud::is_allowed_column` 路径不变,**调用方零改动**。

+mod module_dependency_repo;
+mod plan_repo;
+
 mod conversation_repo;
 mod idea_eval_repo;
 mod idea_repo;
@@ -28,7 +31,6 @@ mod project_service_repo;
 mod settings;
 mod task_link_repo;
 mod task_repo;
-mod module_dependency_repo;

 pub use conversation_repo::*;
 pub use idea_eval_repo::*;
@@ -42,6 +44,7 @@ pub use settings::*;
 pub use task_link_repo::*;
 pub use task_repo::*;
 pub use module_dependency_repo::*;
+pub use plan_repo::*;

 // ============================================================
 // 辅助宏 — 消除多个 Repo 的重复样板
@@ -0,0 +1,579 @@
+//! 多 Agent 并行执行 Repo — ai_plans / ai_subtasks / ai_conflicts 表 CRUD(V36)
+//!
+//! 设计依据:docs/02-架构设计/专项设计/多Agent并行执行与仲裁合并设计-2026-07-01.md §二
+
+use std::sync::Arc;
+
+use rusqlite::{params, OptionalExtension, Row};
+use tokio::sync::Mutex;
+
+use crate::db::Database;
+use crate::models::{ConflictRecord, PlanRecord, SubTaskRecord};
+
+use super::storage_err;
+
+type Result<T> = std::result::Result<T, df_types::error::Error>;
+
+fn plan_from_row(row: &Row<'_>) -> rusqlite::Result<PlanRecord> {
+    Ok(PlanRecord {
+        id: row.get("id")?,
+        conversation_id: row.get("conversation_id")?,
+        user_message_id: row.get("user_message_id")?,
+        status: row.get("status")?,
+        subtask_count: row.get("subtask_count")?,
+        created_at: row.get("created_at")?,
+        completed_at: row.get("completed_at")?,
+    })
+}
+
+fn subtask_from_row(row: &Row<'_>) -> rusqlite::Result<SubTaskRecord> {
+    Ok(SubTaskRecord {
+        id: row.get("id")?,
+        plan_id: row.get("plan_id")?,
+        persona_id: row.get("persona_id")?,
+        intent: row.get("intent")?,
+        status: row.get("status")?,
+        layer: row.get("layer")?,
+        deps: row.get("deps")?,
+        branch: row.get("branch")?,
+        created_at: row.get("created_at")?,
+        completed_at: row.get("completed_at")?,
+    })
+}
+
+fn conflict_from_row(row: &Row<'_>) -> rusqlite::Result<ConflictRecord> {
+    Ok(ConflictRecord {
+        id: row.get("id")?,
+        plan_id: row.get("plan_id")?,
+        file_path: row.get("file_path")?,
+        conflict_type: row.get("conflict_type")?,
+        subtask_a: row.get("subtask_a")?,
+        subtask_b: row.get("subtask_b")?,
+        diff_a: row.get("diff_a")?,
+        diff_b: row.get("diff_b")?,
+        resolution: row.get("resolution")?,
+        resolved_by: row.get("resolved_by")?,
+        created_at: row.get("created_at")?,
+        resolved_at: row.get("resolved_at")?,
+    })
+}
+
+// ============================================================
+// PlanRepo
+// ============================================================
+
+pub struct PlanRepo {
+    conn: Arc<Mutex<rusqlite::Connection>>,
+}
+
+impl PlanRepo {
+    pub fn new(db: &Database) -> Self {
+        Self { conn: db.conn() }
+    }
+
+    pub async fn insert(&self, record: PlanRecord) -> Result<String> {
+        let conn = self.conn.clone();
+        tokio::task::spawn_blocking(move || {
+            let guard = conn.blocking_lock();
+            guard
+                .execute(
+                    "INSERT INTO ai_plans \
+                     (id, conversation_id, user_message_id, status, subtask_count, created_at, completed_at) \
+                     VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
+                    params![
+                        record.id, record.conversation_id, record.user_message_id,
+                        record.status, record.subtask_count, record.created_at, record.completed_at,
+                    ],
+                )
+                .map_err(storage_err)?;
+            Ok(record.id)
+        })
+        .await
+        .map_err(storage_err)?
+    }
+
+    pub async fn get(&self, id: &str) -> Result<Option<PlanRecord>> {
+        let conn = self.conn.clone();
+        let id = id.to_owned();
+        tokio::task::spawn_blocking(move || {
+            let guard = conn.blocking_lock();
+            let row = guard
+                .query_row("SELECT * FROM ai_plans WHERE id = ?1", params![id], plan_from_row)
+                .optional()
+                .map_err(storage_err)?;
+            Ok(row)
+        })
+        .await
+        .map_err(storage_err)?
+    }
+
+    pub async fn update_status(
+        &self,
+        id: &str,
+        status: &str,
+        completed_at: Option<&str>,
+    ) -> Result<bool> {
+        let conn = self.conn.clone();
+        let id = id.to_owned();
+        let status = status.to_owned();
+        let completed_at = completed_at.map(|s| s.to_owned());
+        tokio::task::spawn_blocking(move || {
+            let guard = conn.blocking_lock();
+            let affected = guard
+                .execute(
+                    "UPDATE ai_plans SET status = ?1, completed_at = ?2 WHERE id = ?3",
+                    params![status, completed_at, id],
+                )
+                .map_err(storage_err)?;
+            Ok(affected > 0)
+        })
+        .await
+        .map_err(storage_err)?
+    }
+
+    pub async fn list_by_conversation(&self, conv_id: &str) -> Result<Vec<PlanRecord>> {
+        let conn = self.conn.clone();
+        let conv_id = conv_id.to_owned();
+        tokio::task::spawn_blocking(move || {
+            let guard = conn.blocking_lock();
+            let mut stmt = guard
+                .prepare("SELECT * FROM ai_plans WHERE conversation_id = ?1 ORDER BY created_at DESC")
+                .map_err(storage_err)?;
+            let rows = stmt.query_map(params![conv_id], plan_from_row).map_err(storage_err)?;
+            let mut results = Vec::new();
+            for r in rows {
+                results.push(r.map_err(storage_err)?);
+            }
+            Ok(results)
+        })
+        .await
+        .map_err(storage_err)?
+    }
+}
+
+// ============================================================
+// SubTaskRepo
+// ============================================================
+
+pub struct SubTaskRepo {
+    conn: Arc<Mutex<rusqlite::Connection>>,
+}
+
+impl SubTaskRepo {
+    pub fn new(db: &Database) -> Self {
+        Self { conn: db.conn() }
+    }
+
+    pub async fn insert(&self, record: SubTaskRecord) -> Result<String> {
+        let conn = self.conn.clone();
+        tokio::task::spawn_blocking(move || {
+            let guard = conn.blocking_lock();
+            guard
+                .execute(
+                    "INSERT INTO ai_subtasks \
+                     (id, plan_id, persona_id, intent, status, layer, deps, branch, created_at, completed_at) \
+                     VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)",
+                    params![
+                        record.id, record.plan_id, record.persona_id, record.intent,
+                        record.status, record.layer, record.deps, record.branch,
+                        record.created_at, record.completed_at,
+                    ],
+                )
+                .map_err(storage_err)?;
+            Ok(record.id)
+        })
+        .await
+        .map_err(storage_err)?
+    }
+
+    pub async fn list_by_plan(&self, plan_id: &str) -> Result<Vec<SubTaskRecord>> {
+        let conn = self.conn.clone();
+        let plan_id = plan_id.to_owned();
+        tokio::task::spawn_blocking(move || {
+            let guard = conn.blocking_lock();
+            let mut stmt = guard
+                .prepare("SELECT * FROM ai_subtasks WHERE plan_id = ?1 ORDER BY layer ASC, created_at ASC")
+                .map_err(storage_err)?;
+            let rows = stmt.query_map(params![plan_id], subtask_from_row).map_err(storage_err)?;
+            let mut results = Vec::new();
+            for r in rows {
+                results.push(r.map_err(storage_err)?);
+            }
+            Ok(results)
+        })
+        .await
+        .map_err(storage_err)?
+    }
+
+    pub async fn update_status(
+        &self,
+        id: &str,
+        status: &str,
+        completed_at: Option<&str>,
+    ) -> Result<bool> {
+        let conn = self.conn.clone();
+        let id = id.to_owned();
+        let status = status.to_owned();
+        let completed_at = completed_at.map(|s| s.to_owned());
+        tokio::task::spawn_blocking(move || {
+            let guard = conn.blocking_lock();
+            let affected = guard
+                .execute(
+                    "UPDATE ai_subtasks SET status = ?1, completed_at = ?2 WHERE id = ?3",
+                    params![status, completed_at, id],
+                )
+                .map_err(storage_err)?;
+            Ok(affected > 0)
+        })
+        .await
+        .map_err(storage_err)?
+    }
+}
+
+// ============================================================
+// ConflictRepo
+// ============================================================
+
+pub struct ConflictRepo {
+    conn: Arc<Mutex<rusqlite::Connection>>,
+}
+
+impl ConflictRepo {
+    pub fn new(db: &Database) -> Self {
+        Self { conn: db.conn() }
+    }
+
+    pub async fn insert(&self, record: ConflictRecord) -> Result<String> {
+        let conn = self.conn.clone();
+        tokio::task::spawn_blocking(move || {
+            let guard = conn.blocking_lock();
+            guard
+                .execute(
+                    "INSERT INTO ai_conflicts \
+                     (id, plan_id, file_path, conflict_type, subtask_a, subtask_b, diff_a, diff_b, \
+                     resolution, resolved_by, created_at, resolved_at) \
+                     VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12)",
+                    params![
+                        record.id, record.plan_id, record.file_path, record.conflict_type,
+                        record.subtask_a, record.subtask_b, record.diff_a, record.diff_b,
+                        record.resolution, record.resolved_by, record.created_at, record.resolved_at,
+                    ],
+                )
+                .map_err(storage_err)?;
+            Ok(record.id)
+        })
+        .await
+        .map_err(storage_err)?
+    }
+
+    pub async fn list_pending(&self, plan_id: &str) -> Result<Vec<ConflictRecord>> {
+        let conn = self.conn.clone();
+        let plan_id = plan_id.to_owned();
+        tokio::task::spawn_blocking(move || {
+            let guard = conn.blocking_lock();
+            let mut stmt = guard
+                .prepare("SELECT * FROM ai_conflicts WHERE plan_id = ?1 AND resolution = 'pending' ORDER BY created_at ASC")
+                .map_err(storage_err)?;
+            let rows = stmt.query_map(params![plan_id], conflict_from_row).map_err(storage_err)?;
+            let mut results = Vec::new();
+            for r in rows {
+                results.push(r.map_err(storage_err)?);
+            }
+            Ok(results)
+        })
+        .await
+        .map_err(storage_err)?
+    }
+
+    pub async fn resolve(
+        &self,
+        id: &str,
+        resolution: &str,
+        resolved_by: &str,
+        resolved_at: &str,
+    ) -> Result<bool> {
+        let conn = self.conn.clone();
+        let id = id.to_owned();
+        let resolution = resolution.to_owned();
+        let resolved_by = resolved_by.to_owned();
+        let resolved_at = resolved_at.to_owned();
+        tokio::task::spawn_blocking(move || {
+            let guard = conn.blocking_lock();
+            let affected = guard
+                .execute(
+                    "UPDATE ai_conflicts SET resolution = ?1, resolved_by = ?2, resolved_at = ?3 WHERE id = ?4",
+                    params![resolution, resolved_by, resolved_at, id],
+                )
+                .map_err(storage_err)?;
+            Ok(affected > 0)
+        })
+        .await
+        .map_err(storage_err)?
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::db::Database;
+
+    async fn setup() -> Database {
+        let db = Database::open_in_memory().await.expect("open_in_memory");
+        let conn = db.conn();
+        tokio::task::spawn_blocking(move || {
+            crate::migrations::run(&conn.blocking_lock()).expect("migrations");
+        })
+        .await
+        .expect("spawn_blocking");
+        db
+    }
+
+    #[tokio::test]
+    async fn repo_01_plan_insert_and_get() {
+        let db = setup().await;
+        let repo = PlanRepo::new(&db);
+        let plan = PlanRecord {
+            id: "plan-001".into(),
+            conversation_id: "conv-001".into(),
+            user_message_id: Some("msg-001".into()),
+            status: "planning".into(),
+            subtask_count: 2,
+            created_at: "2026-07-01T00:00:00Z".into(),
+            completed_at: None,
+        };
+        let id = repo.insert(plan.clone()).await.unwrap();
+        assert_eq!(id, "plan-001");
+        let got = repo.get("plan-001").await.unwrap().unwrap();
+        assert_eq!(got.status, "planning");
+        assert_eq!(got.subtask_count, 2);
+    }
+
+    #[tokio::test]
+    async fn repo_02_plan_update_status() {
+        let db = setup().await;
+        let repo = PlanRepo::new(&db);
+        repo.insert(PlanRecord {
+            id: "plan-002".into(),
+            conversation_id: "conv-001".into(),
+            user_message_id: None,
+            status: "planning".into(),
+            subtask_count: 0,
+            created_at: "2026-07-01T00:00:00Z".into(),
+            completed_at: None,
+        })
+        .await
+        .unwrap();
+        assert!(repo.update_status("plan-002", "executing", None).await.unwrap());
+        let got = repo.get("plan-002").await.unwrap().unwrap();
+        assert_eq!(got.status, "executing");
+    }
+
+    #[tokio::test]
+    async fn repo_03_subtask_insert_and_list() {
+        let db = setup().await;
+        let plan_repo = PlanRepo::new(&db);
+        let st_repo = SubTaskRepo::new(&db);
+        plan_repo
+            .insert(PlanRecord {
+                id: "plan-003".into(),
+                conversation_id: "conv-001".into(),
+                user_message_id: None,
+                status: "planning".into(),
+                subtask_count: 2,
+                created_at: "2026-07-01T00:00:00Z".into(),
+                completed_at: None,
+            })
+            .await
+            .unwrap();
+        st_repo
+            .insert(SubTaskRecord {
+                id: "st-001".into(),
+                plan_id: "plan-003".into(),
+                persona_id: Some("coder".into()),
+                intent: "读取代码".into(),
+                status: "pending".into(),
+                layer: 0,
+                deps: None,
+                branch: Some("subtask/plan-003/st-001".into()),
+                created_at: "2026-07-01T00:00:00Z".into(),
+                completed_at: None,
+            })
+            .await
+            .unwrap();
+        st_repo
+            .insert(SubTaskRecord {
+                id: "st-002".into(),
+                plan_id: "plan-003".into(),
+                persona_id: Some("coder".into()),
+                intent: "修改代码".into(),
+                status: "pending".into(),
+                layer: 1,
+                deps: Some(r#"["st-001"]"#.into()),
+                branch: Some("subtask/plan-003/st-002".into()),
+                created_at: "2026-07-01T00:00:00Z".into(),
+                completed_at: None,
+            })
+            .await
+            .unwrap();
+        let list = st_repo.list_by_plan("plan-003").await.unwrap();
+        assert_eq!(list.len(), 2);
+        assert_eq!(list[0].layer, 0); // layer 排序
+        assert_eq!(list[1].layer, 1);
+    }
+
+    #[tokio::test]
+    async fn repo_04_subtask_update_status() {
+        let db = setup().await;
+        let plan_repo = PlanRepo::new(&db);
+        let st_repo = SubTaskRepo::new(&db);
+        plan_repo
+            .insert(PlanRecord {
+                id: "plan-004".into(),
+                conversation_id: "conv-001".into(),
+                user_message_id: None,
+                status: "planning".into(),
+                subtask_count: 1,
+                created_at: "2026-07-01T00:00:00Z".into(),
+                completed_at: None,
+            })
+            .await
+            .unwrap();
+        st_repo
+            .insert(SubTaskRecord {
+                id: "st-004".into(),
+                plan_id: "plan-004".into(),
+                persona_id: None,
+                intent: "test".into(),
+                status: "pending".into(),
+                layer: 0,
+                deps: None,
+                branch: None,
+                created_at: "2026-07-01T00:00:00Z".into(),
+                completed_at: None,
+            })
+            .await
+            .unwrap();
+        assert!(st_repo.update_status("st-004", "done", Some("2026-07-01T01:00:00Z")).await.unwrap());
+        let list = st_repo.list_by_plan("plan-004").await.unwrap();
+        assert_eq!(list[0].status, "done");
+    }
+
+    #[tokio::test]
+    async fn repo_05_conflict_insert_and_list_pending() {
+        let db = setup().await;
+        let plan_repo = PlanRepo::new(&db);
+        let conf_repo = ConflictRepo::new(&db);
+        plan_repo
+            .insert(PlanRecord {
+                id: "plan-005".into(),
+                conversation_id: "conv-001".into(),
+                user_message_id: None,
+                status: "merging".into(),
+                subtask_count: 2,
+                created_at: "2026-07-01T00:00:00Z".into(),
+                completed_at: None,
+            })
+            .await
+            .unwrap();
+        conf_repo
+            .insert(ConflictRecord {
+                id: "conf-001".into(),
+                plan_id: "plan-005".into(),
+                file_path: "src/main.rs".into(),
+                conflict_type: "file".into(),
+                subtask_a: Some("st-a".into()),
+                subtask_b: Some("st-b".into()),
+                diff_a: Some("-old\n+new_a".into()),
+                diff_b: Some("-old\n+new_b".into()),
+                resolution: "pending".into(),
+                resolved_by: None,
+                created_at: "2026-07-01T00:00:00Z".into(),
+                resolved_at: None,
+            })
+            .await
+            .unwrap();
+        let pending = conf_repo.list_pending("plan-005").await.unwrap();
+        assert_eq!(pending.len(), 1);
+        assert_eq!(pending[0].file_path, "src/main.rs");
+    }
+
+    #[tokio::test]
+    async fn repo_06_conflict_resolve() {
+        let db = setup().await;
+        let plan_repo = PlanRepo::new(&db);
+        let conf_repo = ConflictRepo::new(&db);
+        plan_repo
+            .insert(PlanRecord {
+                id: "plan-006".into(),
+                conversation_id: "conv-001".into(),
+                user_message_id: None,
+                status: "merging".into(),
+                subtask_count: 1,
+                created_at: "2026-07-01T00:00:00Z".into(),
+                completed_at: None,
+            })
+            .await
+            .unwrap();
+        conf_repo
+            .insert(ConflictRecord {
+                id: "conf-002".into(),
+                plan_id: "plan-006".into(),
+                file_path: "src/lib.rs".into(),
+                conflict_type: "semantic".into(),
+                subtask_a: None,
+                subtask_b: None,
+                diff_a: None,
+                diff_b: None,
+                resolution: "pending".into(),
+                resolved_by: None,
+                created_at: "2026-07-01T00:00:00Z".into(),
+                resolved_at: None,
+            })
+            .await
+            .unwrap();
+        assert!(
+            conf_repo
+                .resolve("conf-002", "merged", "reviewer", "2026-07-01T02:00:00Z")
+                .await
+                .unwrap()
+        );
+        let pending = conf_repo.list_pending("plan-006").await.unwrap();
+        assert_eq!(pending.len(), 0, "解决后 pending 列表应为空");
+    }
+
+    #[tokio::test]
+    async fn repo_07_subtask_branch_null_for_non_git() {
+        let db = setup().await;
+        let plan_repo = PlanRepo::new(&db);
+        let st_repo = SubTaskRepo::new(&db);
+        plan_repo
+            .insert(PlanRecord {
+                id: "plan-007".into(),
+                conversation_id: "conv-001".into(),
+                user_message_id: None,
+                status: "planning".into(),
+                subtask_count: 1,
+                created_at: "2026-07-01T00:00:00Z".into(),
+                completed_at: None,
+            })
+            .await
+            .unwrap();
+        st_repo
+            .insert(SubTaskRecord {
+                id: "st-007".into(),
+                plan_id: "plan-007".into(),
+                persona_id: None,
+                intent: "非 Git 工程".into(),
+                status: "pending".into(),
+                layer: 0,
+                deps: None,
+                branch: None, // 非 Git 工程
+                created_at: "2026-07-01T00:00:00Z".into(),
+                completed_at: None,
+            })
+            .await
+            .unwrap();
+        let list = st_repo.list_by_plan("plan-007").await.unwrap();
+        assert!(list[0].branch.is_none(), "非 Git 工程 branch 应为 NULL");
+    }
+}
@@ -784,16 +784,16 @@ mod tests {
        let repo = setup().await;
        repo.insert(trec("parent", "todo", None)).await.unwrap();
        // 4 个子任务:status 分布 todo×2 / in_progress×1 / done×1(queue 统一 todo)
-        repo.insert(trec_full("c1", "todo", Some("parent"), "todo"))
+        repo.insert(trec_full("c1", "todo", Some("parent"), TaskStatus::Todo))
            .await
            .unwrap();
-        repo.insert(trec_full("c2", "todo", Some("parent"), "todo"))
+        repo.insert(trec_full("c2", "todo", Some("parent"), TaskStatus::Todo))
            .await
            .unwrap();
-        repo.insert(trec_full("c3", "todo", Some("parent"), "in_progress"))
+        repo.insert(trec_full("c3", "todo", Some("parent"), TaskStatus::InProgress))
            .await
            .unwrap();
-        repo.insert(trec_full("c4", "todo", Some("parent"), "done"))
+        repo.insert(trec_full("c4", "todo", Some("parent"), TaskStatus::Done))
            .await
            .unwrap();

@@ -841,7 +841,7 @@ mod tests {
        repo.insert(trec_full("parent", "todo", None, TaskStatus::Todo))
            .await
            .unwrap();
-        let ok = repo.set_status_for_aggregation("parent", "in_progress");
+        let ok = repo.set_status_for_aggregation("parent", "in_progress").await.unwrap();
        assert!(ok, "应命中写入");
        let after = repo.get_by_id("parent").await.unwrap().unwrap();
        assert_eq!(after.status.as_str(), "in_progress", "status 应被聚合写入更新");