新增: Reviewer仲裁+命令互斥锁+冲突解决IPC(批次C散项)
- coordinator: arbitrate_conflicts 方法(规则仲裁:成功方优先/双方成功推荐合并/都失败手动) - coordinator: ConflictItem 加 subtask_a/subtask_b/recommendation 字段 + ConflictResolution 枚举 - command_lock.rs: CommandLockRegistry (dir+cmd首词 Semaphore 串行化,5个测试全绿) - resolve_conflict IPC: 解决冲突→更新DB→emit AiConflictResolved 事件闭环 - 3个仲裁测试 + 5个互斥锁测试全绿 - 零编译警告
This commit is contained in:
134
src-tauri/src/commands/ai/agentic/command_lock.rs
Normal file
134
src-tauri/src/commands/ai/agentic/command_lock.rs
Normal file
@@ -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<Mutex<HashMap<String, Arc<Semaphore>>>>,
|
||||
}
|
||||
|
||||
impl CommandLockRegistry {
|
||||
#[allow(dead_code)]
|
||||
pub fn new() -> Self {
|
||||
Self {
|
||||
inner: Arc::new(Mutex::new(HashMap::new())),
|
||||
}
|
||||
}
|
||||
|
||||
/// 获取(或创建)指定 key 的信号量。
|
||||
/// 返回 Arc<Semaphore>,调用方 await acquire 后执行命令,drop permit 后释放。
|
||||
#[allow(dead_code)]
|
||||
pub fn get_or_create(&self, working_dir: &str, command: &str) -> Arc<Semaphore> {
|
||||
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);
|
||||
}
|
||||
}
|
||||
@@ -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 辅助
|
||||
|
||||
@@ -140,3 +140,45 @@ pub async fn set_plan_execution(enabled: bool) -> Result<(), String> {
|
||||
pub async fn get_plan_execution() -> Result<bool, String> {
|
||||
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<String>,
|
||||
) -> 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(())
|
||||
}
|
||||
|
||||
@@ -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");
|
||||
|
||||
Reference in New Issue
Block a user