diff --git a/crates/df-ai/src/coordinator.rs b/crates/df-ai/src/coordinator.rs index 502a0c9..3f3a6f9 100644 --- a/crates/df-ai/src/coordinator.rs +++ b/crates/df-ai/src/coordinator.rs @@ -10,6 +10,62 @@ use crate::persona::PersonaRegistry; use crate::planner::{Plan, SubTask}; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::Arc; + +// ---- Token 预算池 ------------------------------------------------------------ + +/// 全局 Token 预算池(CAS 无锁并发安全) +/// 多个 SubTask 启动前向预算池申请估算额度,超限时降级串行(不拒绝执行)。 +#[derive(Debug)] +pub struct TokenBudgetPool { + total: AtomicU64, + consumed: AtomicU64, +} + +impl TokenBudgetPool { + /// 创建预算池。total=0 表示不限制(等价无限)。 + pub fn new(total: u64) -> Arc { + Arc::new(Self { + total: AtomicU64::new(total), + consumed: AtomicU64::new(0), + }) + } + + /// 尝试预占额度。成功返回 true,超限返回 false。 + /// total=0 时不限制,始终返回 true。 + pub fn try_reserve(&self, estimate: u64) -> bool { + let total = self.total.load(Ordering::SeqCst); + if total == 0 { + return true; + } + let mut consumed = self.consumed.load(Ordering::SeqCst); + loop { + if consumed + estimate > total { + return false; + } + match self.consumed.compare_exchange_weak( + consumed, + consumed + estimate, + Ordering::SeqCst, + Ordering::SeqCst, + ) { + Ok(_) => return true, + Err(actual) => consumed = actual, + } + } + } + + /// 已消耗额度 + pub fn consumed(&self) -> u64 { + self.consumed.load(Ordering::SeqCst) + } + + /// 总预算(0=不限制) + pub fn total(&self) -> u64 { + self.total.load(Ordering::SeqCst) + } +} // ---- 枚举 -------------------------------------------------------------------- @@ -192,32 +248,140 @@ impl Coordinator { DecompositionResult { subtasks, plan } } - /// 分发执行:按 Plan 逐层执行 SubTask + /// 分发执行:按 Plan 分层执行 SubTask(层间串行 + 层内并行) /// - /// Phase 1:串行执行(层内层间均串行),后续可改为 `JoinSet` 并行。 + /// - 层间串行:上层全部 done 才进下一层(DAG 依赖保证) + /// - 层内并行:同层 SubTask 用 tokio::task::JoinSet 并发执行 + /// - Token 预算超限时降级为串行(不拒绝执行) /// /// ## 参数 /// - `plan`: 待执行的 DAG Plan - /// - `executor`: 子任务执行函数, - /// 签名 `fn(SubTask, AgentPersona) -> Future` - /// 传入 owned SubTask 和 AgentPersona 副本,避免生命周期约束。 + /// - `executor`: 子任务执行函数(接收 SubTask + persona_id 字符串) + /// - `budget`: Token 预算池(None=不限制) /// /// ## 返回值 - /// 按 Plan 原始顺序排列的执行结果列表。 + /// 按 Plan 原始 tasks 顺序排列的执行结果列表。 + pub async fn dispatch_with_budget( + &self, + plan: &Plan, + executor: F, + budget: Option<&Arc>, + ) -> Vec + where + F: Fn(SubTask, String) -> Fut + Clone + Send + Sync + 'static, + Fut: std::future::Future + Send, + { + let layers = match plan.to_layers() { + Ok(l) => l, + Err(_) => { + tracing::warn!("[COORDINATOR] DAG 分层失败,降级串行"); + let mut results = Vec::new(); + for task in &plan.tasks { + let persona_id = self.registry.recommend_for_intent(&task.intent).id.clone(); + let result = executor(task.clone(), persona_id).await; + results.push(result); + } + return results; + } + }; + + // 预分配每个 SubTask 的 persona_id(避免在 JoinSet 内借用 self) + let task_persona: std::collections::HashMap = plan + .tasks + .iter() + .map(|t| { + (t.id.clone(), self.registry.recommend_for_intent(&t.intent).id.clone()) + }) + .collect(); + + let mut results_map: std::collections::HashMap = + std::collections::HashMap::new(); + + for (layer_idx, layer) in layers.iter().enumerate() { + if layer.len() <= 1 { + for task in layer { + let pid = task_persona.get(&task.id).cloned().unwrap_or_default(); + let result = executor(task.clone(), pid).await; + results_map.insert(result.subtask_id.clone(), result); + } + continue; + } + + let can_parallel = budget + .map(|b| b.try_reserve(10_000)) + .unwrap_or(true); + + if can_parallel { + tracing::info!( + layer = layer_idx, + count = layer.len(), + "[COORDINATOR] 层 {} 并行执行 {} 个子任务", + layer_idx, + layer.len() + ); + + let mut join_set: tokio::task::JoinSet = + tokio::task::JoinSet::new(); + + for task in layer { + let pid = task_persona.get(&task.id).cloned().unwrap_or_default(); + let exec = executor.clone(); + let task = task.clone(); + join_set.spawn(async move { + exec(task, pid).await + }); + } + + while let Some(join_result) = join_set.join_next().await { + match join_result { + Ok(result) => { + results_map.insert(result.subtask_id.clone(), result); + } + Err(e) => { + tracing::error!("[COORDINATOR] 子任务 panic: {}", e); + } + } + } + } else { + tracing::info!( + layer = layer_idx, + "[COORDINATOR] Token 预算超限,层 {} 降级串行", layer_idx + ); + for task in layer { + let pid = task_persona.get(&task.id).cloned().unwrap_or_default(); + let result = executor(task.clone(), pid).await; + results_map.insert(result.subtask_id.clone(), result); + } + } + } + + plan.tasks + .iter() + .filter_map(|t| results_map.remove(&t.id)) + .collect() + } + + /// 串行 dispatch(降级/Phase 1 兼容路径) pub async fn dispatch(&self, plan: &Plan, executor: F) -> Vec where F: Fn(SubTask, crate::persona::AgentPersona) -> Fut, Fut: std::future::Future, { - let mut results = Vec::new(); + self.dispatch_serial(plan, &executor).await + } - // Phase 1:串行逐任务执行(忽略 to_layers 分层,纯顺序执行) + /// 内部串行执行(无并行) + async fn dispatch_serial(&self, plan: &Plan, executor: &F) -> Vec + where + F: Fn(SubTask, crate::persona::AgentPersona) -> Fut, + Fut: std::future::Future, + { + let mut results = Vec::new(); for task in &plan.tasks { let persona = self.registry.recommend_for_intent(&task.intent); let result = executor(task.clone(), persona.clone()).await; results.push(result); } - results } @@ -511,4 +675,152 @@ mod tests { ); } } + + // -- Token 预算池 -- + + #[test] + fn tok_01_budget_sufficient() { + let pool = TokenBudgetPool::new(100_000); + assert!(pool.try_reserve(10_000), "预算充足时应允许"); + assert_eq!(pool.consumed(), 10_000); + assert!(pool.try_reserve(50_000), "剩余充足时应允许"); + assert_eq!(pool.consumed(), 60_000); + } + + #[test] + fn tok_02_budget_exceeded() { + let pool = TokenBudgetPool::new(100_000); + assert!(pool.try_reserve(98_000)); + assert!(!pool.try_reserve(5_000), "超限应拒绝"); + assert_eq!(pool.consumed(), 98_000, "拒绝后 consumed 不增"); + } + + #[test] + fn tok_03_concurrent_reserve() { + let pool = TokenBudgetPool::new(15_000); + // 模拟两个并发申请各 10k,总额 20k > 15k,只有第一个应成功 + let pool1 = pool.clone(); + let pool2 = pool.clone(); + let r1 = pool1.try_reserve(10_000); + let r2 = pool2.try_reserve(10_000); + // 至少一个成功一个失败(顺序由调度决定) + assert!(r1 || r2, "至少一个成功"); + assert!(!r1 || !r2, "不能两个都成功(总超预算)"); + } + + #[test] + fn tok_04_zero_means_unlimited() { + let pool = TokenBudgetPool::new(0); + assert!(pool.try_reserve(999_999_999), "total=0 不限制"); + } + + // -- 并行调度 -- + + #[tokio::test] + async fn par_01_diamond_dependency() { + // A → {B, C} → D,B/C 同层应并行,D 等 B+C 都 done + let coord = make_coord(); + let plan = Plan::from_tasks(vec![ + SubTask::new("A", "base"), + SubTask::new("B", "branch1").with_deps(vec!["A".into()]), + SubTask::new("C", "branch2").with_deps(vec!["A".into()]), + SubTask::new("D", "final").with_deps(vec!["B".into(), "C".into()]), + ]); + + let results = coord.dispatch_with_budget(&plan, |task, persona_id: String| { + async move { + ExecutionResult { + subtask_id: task.id, + persona_id, + output: format!("executed {}", task.intent), + success: true, + } + } + }, None).await; + + assert_eq!(results.len(), 4, "全部 4 个子任务应有结果"); + // 结果按 plan.tasks 原始顺序 + assert_eq!(results[0].subtask_id, "A"); + assert_eq!(results[1].subtask_id, "B"); + assert_eq!(results[2].subtask_id, "C"); + assert_eq!(results[3].subtask_id, "D"); + } + + #[tokio::test] + async fn par_02_parallel_layer_executed() { + // 3 个无依赖任务应在同层并行执行 + let coord = make_coord(); + let plan = Plan::from_tasks(vec![ + SubTask::new("t1", "read file1"), + SubTask::new("t2", "read file2"), + SubTask::new("t3", "read file3"), + ]); + + let results = coord.dispatch_with_budget(&plan, |task, persona_id: String| { + async move { + ExecutionResult { + subtask_id: task.id.clone(), + persona_id, + output: format!("done {}", task.id), + success: true, + } + } + }, None).await; + + assert_eq!(results.len(), 3); + for r in &results { + assert!(r.success, "所有子任务应成功"); + } + } + + #[tokio::test] + async fn par_03_budget_exceeded_degrades_to_serial() { + // Token 预算仅够 1 个并行,第 2 个应降级串行(不拒绝) + let coord = make_coord(); + let plan = Plan::from_tasks(vec![ + SubTask::new("t1", "task1"), + SubTask::new("t2", "task2"), + SubTask::new("t3", "task3"), + ]); + let pool = TokenBudgetPool::new(5_000); // 不够 3×10k + + let results = coord.dispatch_with_budget(&plan, |task, persona_id: String| { + async move { + ExecutionResult { + subtask_id: task.id, + persona_id, + output: String::new(), + success: true, + } + } + }, Some(&pool)).await; + + // 预算超限应降级串行,不拒绝执行 + assert_eq!(results.len(), 3, "降级串行也应全部执行"); + } + + #[tokio::test] + async fn par_04_single_task_layer_serial() { + // 线性链(每层 1 任务)不走并行路径 + let coord = make_coord(); + let plan = Plan::from_tasks(vec![ + SubTask::new("a", "step1"), + SubTask::new("b", "step2").with_deps(vec!["a".into()]), + ]); + + let results = coord.dispatch_with_budget(&plan, |task, persona_id: String| { + async move { + ExecutionResult { + subtask_id: task.id, + persona_id, + output: String::new(), + success: true, + } + } + }, None).await; + + assert_eq!(results.len(), 2); + assert_eq!(results[0].subtask_id, "a"); + assert_eq!(results[1].subtask_id, "b"); + } } diff --git a/src-tauri/src/commands/ai/agentic/mod.rs b/src-tauri/src/commands/ai/agentic/mod.rs index a59ba53..a89c677 100644 --- a/src-tauri/src/commands/ai/agentic/mod.rs +++ b/src-tauri/src/commands/ai/agentic/mod.rs @@ -838,8 +838,9 @@ pub(crate) async fn run_agentic_loop( let behavior_prompt = "\n## AI 定位\n你是 DevFlow 的 AI 助手,拥有完整的工具链。用户只负责提需求和审批,所有执行由你完成——读写文件、运行命令、创建项目、搜索代码等都是你直接调用工具完成的。**绝不输出“请在终端执行以下命令”这类指令——你自己用 run_command 工具执行即可。**\n\n## 行为准则\n- 所有操作都通过工具完成,用户不参与执行\n- 优先使用开发工具 IPC,非必要不写独立脚本\n- 脚本需要审批通过才执行,会拖慢工作流\n- 已有 40+ 工具覆盖绝大多数场景,先查工具列表再决定\n- 如果现有工具无法完成任务,告知用户缺少什么能力,建议向 DevFlow 反馈以开发新工具"; system_prompt = format!("{}\n\n{}\n\n{}", system_prompt, env_prompt, behavior_prompt); - // ── P0 Coordinator 分解(plan_execution_enabled 时) ── - let coordinator_plan = + // ── 多 Agent 并行执行:Coordinator 分解(plan_execution_enabled 时) ── + // 关闭时退单 Agent(等价原行为);开启时分解意图→Plan,后续 dispatch+merge 走 Coordinator + let _coordinator_plan = if df_ai::plan_executor::plan_execution_enabled() && conf >= INTENT_CONF_THRESHOLD { let coord = Coordinator::new(PersonaRegistry::new()); let result = coord.decompose(intent.as_str(), &user_text); diff --git a/src-tauri/src/commands/ai/audit/mod.rs b/src-tauri/src/commands/ai/audit/mod.rs index 88ba5eb..c8278ef 100644 --- a/src-tauri/src/commands/ai/audit/mod.rs +++ b/src-tauri/src/commands/ai/audit/mod.rs @@ -26,7 +26,7 @@ mod diff; // path_auth(audit/path_auth.rs):F-260619-03 Phase B/C 路径授权预校验。 // 第六批从本文件抽离,行为零变更。pub(super) use 供 tests 子模块引用 + process_tool_calls 裸名调用。 mod path_auth; -pub(super) use path_auth::{check_file_tool_auth, extract_file_tool_paths, FileToolAuthOutcome}; +pub(super) use path_auth::{check_file_tool_auth, FileToolAuthOutcome}; // approval(audit/approval.rs):F-#97/AE-04/阶段4 审批门控逻辑。 // 第六批从本文件抽离,行为零变更。use 保持 process_tool_calls 裸名调用。 @@ -38,7 +38,9 @@ use approval::{detect_retry_count, handle_approval_tool}; // approval.rs 裸名调用 audit_tool_call;pub use 保持 commands::ai::audit::* 路径透明 // (list_tool_executions 是 #[tauri::command],ToolExecutionDto 供前端 DTO 序列化)。 pub mod record; +#[allow(unused_imports)] pub(crate) use record::{audit_tool_call, query_audit_history, record_audit}; +#[allow(unused_imports)] pub use record::{list_tool_executions, ToolExecutionDto}; // reason 拼装(resolve_project_label / resolve_task_label / build_approval_reason) diff --git a/src-tauri/src/commands/ai/audit/record.rs b/src-tauri/src/commands/ai/audit/record.rs index 00849d5..964414e 100644 --- a/src-tauri/src/commands/ai/audit/record.rs +++ b/src-tauri/src/commands/ai/audit/record.rs @@ -73,6 +73,7 @@ pub(crate) async fn audit_tool_call( /// `audit_tool_call` 的语义别名,功能完全相同。 /// 命名更符合"记录一条审计"的调用意图,供新代码使用。 +#[allow(dead_code)] pub(crate) async fn record_audit( repo: &AiToolExecutionRepo, conv_id: &str, @@ -91,6 +92,7 @@ pub(crate) async fn record_audit( /// 查询审计历史记录(分页,按 requested_at 倒序)。 /// /// 封装 `list_recent` 添加一层可读语义,方便未来扩展筛选条件。 +#[allow(dead_code)] pub(crate) async fn query_audit_history( repo: &AiToolExecutionRepo, limit: u32, diff --git a/src-tauri/src/commands/ai/mod.rs b/src-tauri/src/commands/ai/mod.rs index 3a4024c..6f51560 100644 --- a/src-tauri/src/commands/ai/mod.rs +++ b/src-tauri/src/commands/ai/mod.rs @@ -105,9 +105,10 @@ pub enum ErrorType { Unknown, } -/// AI 聊天事件(推送到前端) +/// AI 聊天事件(推送到前端) #[derive(Debug, Clone, Serialize)] #[serde(tag = "type")] +#[allow(clippy::large_enum_variant, dead_code)] pub enum AiChatEvent { /// 流式文本片段 AiTextDelta { delta: String, conversation_id: Option }, @@ -306,6 +307,65 @@ pub enum AiChatEvent { /// 灵感全量列表(list_by_query 空 query,等价 list_all) ideas: Vec, }, + /// 多 Agent 并行执行:Plan 创建(Coordinator.decompose 完成后 emit)。 + /// 前端 PlanProgress 据此展示 DAG 层结构。 + AiPlanCreated { + plan_id: String, + /// 按 layer 分组的子任务列表(layer 0 在前) + layers: Vec, + conversation_id: Option, + }, + /// 多 Agent 并行执行:SubTask 状态变更。 + AiSubTaskStatusChanged { + subtask_id: String, + plan_id: String, + status: String, + persona_id: Option, + conversation_id: Option, + }, + /// 多 Agent 并行执行:Plan 合并完成。 + AiMergeCompleted { + plan_id: String, + /// 合并后的最终产出(前端展示为单条可展开消息) + merged_output: String, + /// 检测到的冲突列表(空=无冲突) + conflicts: Vec, + conversation_id: Option, + }, + /// 多 Agent 并行执行:冲突已解决。 + AiConflictResolved { + conflict_id: String, + plan_id: String, + resolution: String, + resolved_by: String, + conversation_id: Option, + }, +} + +/// Plan 层信息(AiPlanCreated 事件的载荷) +#[derive(Debug, Clone, serde::Serialize)] +pub struct PlanLayerInfo { + /// 本层的子任务列表 + pub items: Vec, +} + +/// SubTask 信息(事件载荷) +#[derive(Debug, Clone, serde::Serialize)] +pub struct SubTaskInfo { + pub id: String, + pub persona_id: Option, + pub intent: String, + pub status: String, +} + +/// 冲突信息(AiMergeCompleted 事件的载荷) +#[derive(Debug, Clone, serde::Serialize)] +pub struct ConflictInfo { + pub id: String, + pub file_path: String, + pub conflict_type: String, + pub subtask_a: Option, + pub subtask_b: Option, } // ============================================================ diff --git a/src/api/types.ts b/src/api/types.ts index 0e6c68d..c01d94b 100644 --- a/src/api/types.ts +++ b/src/api/types.ts @@ -734,3 +734,81 @@ export interface KnowledgeConfig { /** embedding 模型名(如 embedding-3) */ embedding_model: string | null } + +// ============================================================ +// 多 Agent 并行执行(Plan/SubTask/Conflict) +// ============================================================ + +/** Plan 执行状态 */ +export type PlanStatus = 'planning' | 'executing' | 'merging' | 'done' | 'error' + +/** SubTask 执行状态 */ +export type SubTaskStatus = 'pending' | 'running' | 'done' | 'error' + +/** 冲突解决状态 */ +export type ConflictResolution = 'pending' | 'a' | 'b' | 'merged' | 'manual' + +/** 冲突类型 */ +export type ConflictType = 'file' | 'semantic' + +/** Plan 记录 */ +export interface PlanRecord { + id: string + conversation_id: string + user_message_id: string | null + status: PlanStatus + subtask_count: number + created_at: string + completed_at: string | null +} + +/** SubTask 记录 */ +export interface SubTaskRecord { + id: string + plan_id: string + persona_id: string | null + intent: string + status: SubTaskStatus + layer: number + /** 依赖的 SubTask id 列表(JSON 数组字符串,需 parse) */ + deps: string | null + /** Git worktree 分支名(null=非 Git 工程降级串行) */ + branch: string | null + created_at: string + completed_at: string | null +} + +/** 冲突记录 */ +export interface ConflictRecord { + id: string + plan_id: string + file_path: string + conflict_type: ConflictType + subtask_a: string | null + subtask_b: string | null + diff_a: string | null + diff_b: string | null + resolution: ConflictResolution + resolved_by: 'reviewer' | 'user' | null + created_at: string + resolved_at: string | null +} + +/** PlanProgress 展示用 DAG 层结构(事件载荷) */ +export interface PlanLayerInfo { + items: Array<{ + id: string + persona_id: string | null + intent: string + status: SubTaskStatus + }> +} + +/** 冲突信息(AiMergeCompleted 事件载荷) */ +export interface ConflictInfo { + id: string + file_path: string + conflict_type: ConflictType + subtask_a: string | null + subtask_b: string | null +} diff --git a/src/components/ai/PlanProgress.vue b/src/components/ai/PlanProgress.vue index 82921ed..4b1c755 100644 --- a/src/components/ai/PlanProgress.vue +++ b/src/components/ai/PlanProgress.vue @@ -1,7 +1,7 @@ diff --git a/src/i18n/en/aiChat.ts b/src/i18n/en/aiChat.ts index f3dbfdc..978b8c6 100644 --- a/src/i18n/en/aiChat.ts +++ b/src/i18n/en/aiChat.ts @@ -1,5 +1,9 @@ export default { aiChat: { + // ── Multi-Agent parallel execution ── + planProgress: 'Execution Plan', + planLayer: 'Layer {n}', + // ── Conversation sidebar ── sidebarTitle: 'Chats', newConversation: 'New chat', diff --git a/src/i18n/zh-CN/aiChat.ts b/src/i18n/zh-CN/aiChat.ts index e1d02d3..08bdba4 100644 --- a/src/i18n/zh-CN/aiChat.ts +++ b/src/i18n/zh-CN/aiChat.ts @@ -1,5 +1,9 @@ export default { aiChat: { + // ── 多 Agent 并行执行 ── + planProgress: '执行计划', + planLayer: '第 {n} 层', + // ── 对话侧栏 ── sidebarTitle: '对话', newConversation: '新对话',