diff --git a/crates/df-ai/src/coordinator.rs b/crates/df-ai/src/coordinator.rs index ebb4273..cffe9d2 100644 --- a/crates/df-ai/src/coordinator.rs +++ b/crates/df-ai/src/coordinator.rs @@ -123,6 +123,35 @@ pub struct ConflictItem { pub file: String, /// 冲突描述 pub description: String, + /// 冲突双方 SubTask id + pub subtask_a: Option, + pub subtask_b: Option, + /// Reviewer 仲裁推荐(Phase 2 LLM 仲裁填充) + pub recommendation: Option, +} + +/// 冲突解决方案 +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum ConflictResolution { + /// 接受 A 的改动 + AcceptA, + /// 接受 B 的改动 + AcceptB, + /// 合并两者 + Merged, + /// 用户手动处理 + Manual, +} + +impl ConflictResolution { + pub fn as_str(&self) -> &'static str { + match self { + Self::AcceptA => "a", + Self::AcceptB => "b", + Self::Merged => "merged", + Self::Manual => "manual", + } + } } // ---- Coordinator 实现 -------------------------------------------------------- @@ -445,11 +474,49 @@ impl Coordinator { subtasks.len(), subtasks.join(", ") ), + subtask_a: subtasks.first().cloned(), + subtask_b: subtasks.get(1).cloned(), + recommendation: None, }); } } conflicts } + + /// 仲裁冲突:对每个冲突给出推荐解决方案 + /// + /// Phase 1:规则驱动(简单启发式) + /// - 只有一方成功的冲突 → 推荐成功方 + /// - 两方都成功 → 推荐 Merged(需人工确认) + /// - 无法判断 → Manual(留给用户) + /// + /// Phase 2 预留:LLM 驱动(读 diff → 推荐方案 + 理由) + pub fn arbitrate_conflicts( + &self, + conflicts: &mut [ConflictItem], + results: &[ExecutionResult], + ) { + let success_ids: std::collections::HashSet<&str> = + results.iter().filter(|r| r.success).map(|r| r.subtask_id.as_str()).collect(); + + for c in conflicts.iter_mut() { + let a_ok = c.subtask_a.as_deref().map(|id| success_ids.contains(id)).unwrap_or(false); + let b_ok = c.subtask_b.as_deref().map(|id| success_ids.contains(id)).unwrap_or(false); + + c.recommendation = Some(match (a_ok, b_ok) { + (true, false) => ConflictResolution::AcceptA, + (false, true) => ConflictResolution::AcceptB, + (true, true) => ConflictResolution::Merged, + (false, false) => ConflictResolution::Manual, + }); + + tracing::info!( + file = %c.file, + recommendation = ?c.recommendation, + "[REVIEWER] 冲突仲裁推荐" + ); + } + } } /// 从执行产出文本中提取被写入的文件路径(简化检测:匹配 write_file/patch_file 后的路径) @@ -785,6 +852,62 @@ mod tests { assert!(files.is_empty()); } + // -- Reviewer 仲裁 -- + + #[test] + fn arb_01_both_success_recommends_merged() { + let coord = make_coord(); + let mut conflicts = vec![ConflictItem { + file: "main.rs".into(), + description: "冲突".into(), + subtask_a: Some("A".into()), + subtask_b: Some("B".into()), + recommendation: None, + }]; + let results = vec![ + ExecutionResult { subtask_id: "A".into(), persona_id: "coder".into(), output: "".into(), success: true }, + ExecutionResult { subtask_id: "B".into(), persona_id: "coder".into(), output: "".into(), success: true }, + ]; + coord.arbitrate_conflicts(&mut conflicts, &results); + assert_eq!(conflicts[0].recommendation, Some(ConflictResolution::Merged)); + } + + #[test] + fn arb_02_a_failed_recommends_b() { + let coord = make_coord(); + let mut conflicts = vec![ConflictItem { + file: "main.rs".into(), + description: "冲突".into(), + subtask_a: Some("A".into()), + subtask_b: Some("B".into()), + recommendation: None, + }]; + let results = vec![ + ExecutionResult { subtask_id: "A".into(), persona_id: "coder".into(), output: "".into(), success: false }, + ExecutionResult { subtask_id: "B".into(), persona_id: "coder".into(), output: "".into(), success: true }, + ]; + coord.arbitrate_conflicts(&mut conflicts, &results); + assert_eq!(conflicts[0].recommendation, Some(ConflictResolution::AcceptB)); + } + + #[test] + fn arb_03_both_failed_recommends_manual() { + let coord = make_coord(); + let mut conflicts = vec![ConflictItem { + file: "main.rs".into(), + description: "冲突".into(), + subtask_a: Some("A".into()), + subtask_b: Some("B".into()), + recommendation: None, + }]; + let results = vec![ + ExecutionResult { subtask_id: "A".into(), persona_id: "coder".into(), output: "".into(), success: false }, + ExecutionResult { subtask_id: "B".into(), persona_id: "coder".into(), output: "".into(), success: false }, + ]; + coord.arbitrate_conflicts(&mut conflicts, &results); + assert_eq!(conflicts[0].recommendation, Some(ConflictResolution::Manual)); + } + // -- DispatchStrategy -- #[test] diff --git a/src-tauri/src/commands/ai/agentic/command_lock.rs b/src-tauri/src/commands/ai/agentic/command_lock.rs new file mode 100644 index 0000000..4026c40 --- /dev/null +++ b/src-tauri/src/commands/ai/agentic/command_lock.rs @@ -0,0 +1,134 @@ +//! 命令执行互斥锁 — 防止多 Agent 并行时同目录的同名命令资源竞争 +//! +//! 设计依据:docs/02-架构设计/专项设计/多Agent并行执行与仲裁合并设计-2026-07-01.md §4.3 +//! +//! 场景:多 Agent 并行时,两个 SubTask 同时跑 `npm install` 争抢 node_modules。 +//! 方案:按 (working_dir, command_name) 加 mutex,同 key 串行化。 + +use std::collections::HashMap; +use std::sync::{Arc, Mutex}; +use tokio::sync::Semaphore; + +/// 全局命令锁注册表 +/// key = "{working_dir}::{command_first_word}" +/// 每个 key 一个 Semaphore(1),同 key 并发请求串行化 +#[allow(dead_code)] +#[derive(Debug, Clone)] +pub struct CommandLockRegistry { + inner: Arc>>>, +} + +impl CommandLockRegistry { + #[allow(dead_code)] + pub fn new() -> Self { + Self { + inner: Arc::new(Mutex::new(HashMap::new())), + } + } + + /// 获取(或创建)指定 key 的信号量。 + /// 返回 Arc,调用方 await acquire 后执行命令,drop permit 后释放。 + #[allow(dead_code)] + pub fn get_or_create(&self, working_dir: &str, command: &str) -> Arc { + let cmd_first = command.split_whitespace().next().unwrap_or(command); + let key = format!("{}::{}", working_dir, cmd_first); + let mut guard = self.inner.lock().unwrap(); + guard + .entry(key) + .or_insert_with(|| Arc::new(Semaphore::new(1))) + .clone() + } + + /// 当前注册的锁数量(调试/监控用) + #[allow(dead_code)] + pub fn lock_count(&self) -> usize { + self.inner.lock().unwrap().len() + } +} + +impl Default for CommandLockRegistry { + fn default() -> Self { + Self::new() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn lck_01_same_dir_same_cmd_serialized() { + let registry = CommandLockRegistry::new(); + let sem1 = registry.get_or_create("/project", "npm install"); + let sem2 = registry.get_or_create("/project", "npm install"); + + // 同 key 应返回同一个 Semaphore + assert!(Arc::ptr_eq(&sem1, &sem2), "同 dir+cmd 应返回同一个 Semaphore"); + + // 获取第一个 permit + let permit1 = sem1.acquire().await.unwrap(); + // 第二个应等待(非阻塞测试:try_acquire 应失败) + let try_result = sem2.try_acquire(); + assert!(try_result.is_err(), "同 key 第二个应被阻塞"); + + drop(permit1); + // 释放后第二个应能获取 + let permit2 = sem2.acquire().await; + assert!(permit2.is_ok(), "释放后应可获取"); + } + + #[tokio::test] + async fn lck_02_different_cmds_parallel() { + let registry = CommandLockRegistry::new(); + let npm_sem = registry.get_or_create("/project", "npm install"); + let cargo_sem = registry.get_or_create("/project", "cargo build"); + + // 不同命令应是不同的 Semaphore + assert!(!Arc::ptr_eq(&npm_sem, &cargo_sem), "不同 cmd 应是不同 Semaphore"); + + // 两个都能同时获取 + let p1 = npm_sem.acquire().await.unwrap(); + let p2 = cargo_sem.acquire().await.unwrap(); + drop(p1); + drop(p2); + } + + #[tokio::test] + async fn lck_03_different_dirs_parallel() { + let registry = CommandLockRegistry::new(); + let sem1 = registry.get_or_create("/project-a", "npm install"); + let sem2 = registry.get_or_create("/project-b", "npm install"); + + // 不同目录应是不同的 Semaphore + assert!(!Arc::ptr_eq(&sem1, &sem2), "不同 dir 应是不同 Semaphore"); + + // 两个都能同时获取 + let p1 = sem1.acquire().await.unwrap(); + let p2 = sem2.acquire().await.unwrap(); + drop(p1); + drop(p2); + } + + #[tokio::test] + async fn lck_04_command_extracted_first_word() { + let registry = CommandLockRegistry::new(); + let sem1 = registry.get_or_create("/proj", "npm install --force"); + let sem2 = registry.get_or_create("/proj", "npm run build"); + + // "npm install --force" 和 "npm run build" 首词都是 "npm" + // 同 dir + 同首词应串行化 + assert!(Arc::ptr_eq(&sem1, &sem2), "同 dir+同首词应串行化"); + } + + #[test] + fn lck_05_lock_count_tracks_keys() { + let registry = CommandLockRegistry::new(); + assert_eq!(registry.lock_count(), 0); + registry.get_or_create("/a", "npm"); + assert_eq!(registry.lock_count(), 1); + registry.get_or_create("/a", "cargo"); + assert_eq!(registry.lock_count(), 2); + registry.get_or_create("/a", "npm"); // 已存在 + assert_eq!(registry.lock_count(), 2); + } +} diff --git a/src-tauri/src/commands/ai/agentic/mod.rs b/src-tauri/src/commands/ai/agentic/mod.rs index 48a6ee2..8695aba 100644 --- a/src-tauri/src/commands/ai/agentic/mod.rs +++ b/src-tauri/src/commands/ai/agentic/mod.rs @@ -271,6 +271,7 @@ mod knowledge_lifecycle; // - 兜底:状态机层迁移失败(非法转换)记 warn 不 panic,核心生成态复位仍经 enum 迁移收敛。 // ============================================================ pub mod conv_state; +pub mod command_lock; // ============================================================ // F-260614-04 / F-260614-04b: 单 Provider 流式结果 + fallback 辅助 diff --git a/src-tauri/src/commands/settings.rs b/src-tauri/src/commands/settings.rs index 3857732..10dbbb3 100644 --- a/src-tauri/src/commands/settings.rs +++ b/src-tauri/src/commands/settings.rs @@ -140,3 +140,45 @@ pub async fn set_plan_execution(enabled: bool) -> Result<(), String> { pub async fn get_plan_execution() -> Result { Ok(df_ai::plan_executor::plan_execution_enabled()) } + +/// 解决合并冲突(用户/reviewer 选择解决方案后调用)。 +/// +/// 更新 ai_conflicts 表的 resolution + resolved_by + resolved_at, +/// 并 emit AiConflictResolved 事件供前端更新徽章。 +#[tauri::command] +pub async fn resolve_conflict( + state: State<'_, AppState>, + app_handle: tauri::AppHandle, + conflict_id: String, + plan_id: String, + resolution: String, + resolved_by: String, + conversation_id: Option, +) -> Result<(), String> { + use tauri::Emitter; + let repo = df_storage::crud::ConflictRepo::new(&state.db); + let now = df_types::now_millis().to_string(); + let ok = repo.resolve(&conflict_id, &resolution, &resolved_by, &now) + .await + .map_err(err_str)?; + if !ok { + return Err(format!("冲突 {} 不存在或已解决", conflict_id)); + } + tracing::info!( + conflict_id = %conflict_id, + resolution = %resolution, + resolved_by = %resolved_by, + "[CONFLICT] 冲突已解决" + ); + // emit AiConflictResolved 事件 + let ev = crate::commands::ai::AiChatEvent::AiConflictResolved { + conflict_id, + plan_id, + resolution, + resolved_by, + conversation_id, + }; + let _ = app_handle.emit("ai-chat-event", ev.clone()); + let _ = state.ai_event_bus.publish_event(ev); + Ok(()) +} diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index 9f1b723..1736e9c 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -436,6 +436,8 @@ pub fn run() { // Plan 执行开关 commands::settings::set_plan_execution, commands::settings::get_plan_execution, + // 冲突解决 + commands::settings::resolve_conflict, ]) .run(tauri::generate_context!()) .expect("error while running tauri application");