From 4aa689e1103fca3bfa34c3168602e85476d8e0e8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=BB=9D=E5=B0=98?= <237809796@qq.com> Date: Sun, 14 Jun 2026 15:19:36 +0800 Subject: [PATCH] =?UTF-8?q?=E6=96=B0=E5=A2=9E:=20=E5=B7=A5=E4=BD=9C?= =?UTF-8?q?=E6=B5=81=E5=AE=A1=E6=89=B9=E5=8F=96=E6=B6=88=E7=AB=AF=E5=88=B0?= =?UTF-8?q?=E7=AB=AF=EF=BC=88StateMachine=20Arc=20=E5=85=B1=E4=BA=AB=20+?= =?UTF-8?q?=20=E5=8F=96=E6=B6=88=20IPC=20=E6=B3=A8=E5=86=8C=E8=A1=A8?= =?UTF-8?q?=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - StateMachine 内部 HashMap→Arc>，clone 共享底层 - DagExecutor 加 state_machine() 访问器 - AppState 加 workflow_state_registry（execution_id→StateMachine） - run_workflow 注册执行器状态机 + 完成清理 - cancel_workflow_node IPC 经 execution_id 取共享引用 set_cancelled，直达 HumanNode 轮询 - 前端取消按钮 + cancelHumanApproval store 方法 - 加 clone_shares 单测验证共享语义 --- crates/df-workflow/src/executor.rs | 9 +++ crates/df-workflow/src/state.rs | 103 ++++++++++++++++++++++++----- src-tauri/src/commands/workflow.rs | 36 ++++++++++ src-tauri/src/lib.rs | 1 + src-tauri/src/state.rs | 11 +++ src/stores/project.ts | 18 ++++- src/views/ProjectDetail.vue | 7 ++ 7 files changed, 166 insertions(+), 19 deletions(-) diff --git a/crates/df-workflow/src/executor.rs b/crates/df-workflow/src/executor.rs index acaed97..6e056ad 100644 --- a/crates/df-workflow/src/executor.rs +++ b/crates/df-workflow/src/executor.rs @@ -33,6 +33,15 @@ impl DagExecutor { } } + /// 返回状态机的共享引用（Arc clone，与 NodeContext.node_status 共享底层 HashMap） + /// + /// run_workflow 创建执行器后将其注册到 AppState 全局表， + /// cancel_workflow_node IPC 经 execution_id 取出此引用调 set_cancelled， + /// 写入直达运行中阻塞节点（HumanNode）的 is_cancelled 轮询。 + pub fn state_machine(&self) -> StateMachine { + self.state_machine.clone() + } + /// 执行 DAG 工作流 /// /// 同一拓扑层内的节点并发执行，层与层之间串行； diff --git a/crates/df-workflow/src/state.rs b/crates/df-workflow/src/state.rs index 4309202..d610707 100644 --- a/crates/df-workflow/src/state.rs +++ b/crates/df-workflow/src/state.rs @@ -2,29 +2,38 @@ //! //! 合法转换：Pending → Running，Running → Completed / Failed， //! 非法转换返回错误，避免状态被随意覆盖。 +//! +//! 共享语义：内部用 `Arc>`，`clone()` 为浅拷贝（Arc 引用计数 +1）， +//! 所有 clone 共享同一底层 HashMap。DagExecutor 的 `state_machine` 与其下沉到 +//! `NodeContext.node_status` 的 clone 共享底层；run_workflow 把执行器状态机注册到 +//! AppState 全局表后，`cancel_workflow_node` IPC 的 `set_cancelled` 可直达运行中 +//! 阻塞节点（如 HumanNode）的 `is_cancelled` 轮询。 use std::collections::HashMap; +use std::sync::{Arc, Mutex}; use df_core::types::{NodeId, NodeStatus}; /// 节点状态机 #[derive(Debug, Clone)] pub struct StateMachine { - /// 节点状态映射 - states: HashMap, + /// 节点状态映射（Arc 共享，clone 浅拷贝同一 HashMap） + states: Arc>>, } impl StateMachine { /// 创建状态机 pub fn new() -> Self { Self { - states: HashMap::new(), + states: Arc::new(Mutex::new(HashMap::new())), } } /// 获取节点状态 pub fn get(&self, node_id: &NodeId) -> NodeStatus { self.states + .lock() + .expect("状态机锁中毒") .get(node_id) .cloned() .unwrap_or(NodeStatus::Pending) @@ -41,8 +50,12 @@ impl StateMachine { } /// 状态转换 — 校验合法性后更新，非法转换返回错误 - pub fn transition(&mut self, node_id: NodeId, target: NodeStatus) -> anyhow::Result<()> { - let current = self.get(&node_id); + pub fn transition(&self, node_id: NodeId, target: NodeStatus) -> anyhow::Result<()> { + let mut states = self.states.lock().expect("状态机锁中毒"); + let current = states + .get(&node_id) + .cloned() + .unwrap_or(NodeStatus::Pending); if !Self::is_legal(¤t, &target) { anyhow::bail!( "节点 {} 状态转换非法：{} -> {}", @@ -51,38 +64,55 @@ impl StateMachine { target.as_str() ); } - self.states.insert(node_id, target); + states.insert(node_id, target); Ok(()) } /// 设置节点为运行中（仅允许 Pending -> Running） - pub fn set_running(&mut self, node_id: NodeId) -> anyhow::Result<()> { + pub fn set_running(&self, node_id: NodeId) -> anyhow::Result<()> { self.transition(node_id, NodeStatus::Running) } /// 设置节点为已完成（仅允许 Running -> Completed） - pub fn set_completed(&mut self, node_id: NodeId) -> anyhow::Result<()> { + pub fn set_completed(&self, node_id: NodeId) -> anyhow::Result<()> { self.transition(node_id, NodeStatus::Completed) } /// 设置节点为失败（仅允许 Running -> Failed） - pub fn set_failed(&mut self, node_id: NodeId) -> anyhow::Result<()> { + pub fn set_failed(&self, node_id: NodeId) -> anyhow::Result<()> { self.transition(node_id, NodeStatus::Failed) } /// 设置节点为等待中（暂未纳入转换校验） - pub fn set_waiting(&mut self, node_id: NodeId) { - self.states.insert(node_id, NodeStatus::Waiting); + pub fn set_waiting(&self, node_id: NodeId) { + self.states + .lock() + .expect("状态机锁中毒") + .insert(node_id, NodeStatus::Waiting); } /// 设置节点为已跳过（暂未纳入转换校验） - pub fn set_skipped(&mut self, node_id: NodeId) { - self.states.insert(node_id, NodeStatus::Skipped); + pub fn set_skipped(&self, node_id: NodeId) { + self.states + .lock() + .expect("状态机锁中毒") + .insert(node_id, NodeStatus::Skipped); } - /// 获取所有状态快照 - pub fn snapshot(&self) -> &HashMap { - &self.states + /// 设置节点为已取消（暂未纳入转换校验，同 set_waiting/set_skipped 模式） + /// + /// 人工审批取消场景由前端 cancel_workflow_node IPC 触发； + /// 阻塞节点(如 HumanNode)在 select! 轮询分支通过 is_cancelled 检测到后返回 Err。 + pub fn set_cancelled(&self, node_id: NodeId) { + self.states + .lock() + .expect("状态机锁中毒") + .insert(node_id, NodeStatus::Cancelled); + } + + /// 获取所有状态快照（clone 返回，调用方持独立副本） + pub fn snapshot(&self) -> HashMap { + self.states.lock().expect("状态机锁中毒").clone() } /// 检查节点是否被取消 @@ -103,7 +133,7 @@ mod tests { #[test] fn test_legal_transitions() { - let mut sm = StateMachine::new(); + let sm = StateMachine::new(); let id = "node-a".to_string(); // Pending -> Running -> Completed 全程合法 @@ -121,7 +151,7 @@ mod tests { #[test] fn test_illegal_transitions_rejected() { - let mut sm = StateMachine::new(); + let sm = StateMachine::new(); let id = "node-a".to_string(); // Pending -> Completed 非法 @@ -144,4 +174,41 @@ mod tests { sm.set_running(id2.clone()).unwrap(); assert!(sm.set_running(id2.clone()).is_err()); } + + #[test] + fn test_set_cancelled_marks_node_cancelled() { + // set_cancelled 直接置位，不经转换校验（同 set_waiting/set_skipped 模式） + let sm = StateMachine::new(); + let id = "human-1".to_string(); + + // 初始为 Pending，is_cancelled = false + assert_eq!(sm.get(&id), NodeStatus::Pending); + assert!(!sm.is_cancelled(&id)); + + // 取消后置位 Cancelled，is_cancelled = true + sm.set_cancelled(id.clone()); + assert_eq!(sm.get(&id), NodeStatus::Cancelled); + assert!(sm.is_cancelled(&id)); + } + + #[test] + fn test_clone_shares_underlying_state() { + // 核心机制：clone 浅拷贝 Arc，写入对另一副本可见。 + // 这让 cancel IPC 写 AppState 注册表的 StateMachine， + // 直达 HumanNode 持有的 ctx.node_status（executor.state_machine 的 clone）。 + let sm = StateMachine::new(); + let sm_clone = sm.clone(); + + // 原实例写入，clone 可见 + sm.set_cancelled("human-1".to_string()); + assert!(sm_clone.is_cancelled(&"human-1".to_string())); + + // clone 写入，原实例亦可见（双向） + sm_clone.set_running("node-a".to_string()).unwrap(); + assert_eq!(sm.get(&"node-a".to_string()), NodeStatus::Running); + + // 第三副本同样共享 + let sm3 = sm_clone.clone(); + assert!(sm3.is_cancelled(&"human-1".to_string())); + } } diff --git a/src-tauri/src/commands/workflow.rs b/src-tauri/src/commands/workflow.rs index 63194c7..d5cc7f1 100644 --- a/src-tauri/src/commands/workflow.rs +++ b/src-tauri/src/commands/workflow.rs @@ -100,8 +100,15 @@ pub async fn run_workflow( let event_bus = state.event_bus.clone(); let db = state.db.clone(); let exec_id = execution_id.clone(); + let state_registry = state.workflow_state_registry.clone(); tauri::async_runtime::spawn(async move { let mut executor = DagExecutor::new(event_bus.clone(), exec_id.clone()); + // 注册执行器状态机：StateMachine 内部 Arc，clone 共享底层 HashMap， + // cancel_workflow_node IPC 经 execution_id 取此引用 set_cancelled，直达运行中 HumanNode + state_registry + .lock() + .await + .insert(exec_id.clone(), executor.state_machine()); let result = executor.run(&runtime_dag, config).await; let (status, error) = match &result { @@ -121,6 +128,9 @@ pub async fn run_workflow( tracing::error!("更新工作流完成时间失败: {}", e); } + // 执行结束（成功/失败）清理状态注册表，防内存泄漏 + state_registry.lock().await.remove(&exec_id); + // 执行失败时补发 WorkflowFailed（执行器内部只发 NodeFailed） if let Some(error) = error { event_bus @@ -185,3 +195,29 @@ pub async fn approve_human_approval( state.event_bus.send(event).await; Ok(()) } + +/// 取消工作流节点（人工审批取消） +/// +/// 从执行器状态机注册表取出共享引用，调 `set_cancelled` 置目标节点为 Cancelled。 +/// StateMachine 内部 Arc 共享底层 HashMap，IPC 写入直达运行中阻塞节点（HumanNode） +/// 的 select! 轮询分支，其 is_cancelled 检测到后返回 Err "人工审批被取消"。 +#[tauri::command] +pub async fn cancel_workflow_node( + state: State<'_, AppState>, + execution_id: String, + node_id: String, +) -> Result<(), String> { + // 经 execution_id 取执行器状态机（与 NodeContext.node_status 共享同一 HashMap） + let registry = state.workflow_state_registry.lock().await; + let sm = match registry.get(&execution_id) { + Some(sm) => sm, + None => return Err(format!("工作流 {} 不存在或已结束", execution_id)), + }; + sm.set_cancelled(node_id.clone()); + tracing::info!( + "工作流节点取消：execution_id={}, node_id={}", + execution_id, + node_id + ); + Ok(()) +} diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index ff3b70c..c6b6a78 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -58,6 +58,7 @@ pub fn run() { commands::workflow::list_workflow_executions, commands::workflow::get_workflow_execution, commands::workflow::approve_human_approval, + commands::workflow::cancel_workflow_node, // AI 聊天 commands::ai::ai_chat_send, commands::ai::ai_chat_stop, diff --git a/src-tauri/src/state.rs b/src-tauri/src/state.rs index 89ca308..6ff6dc5 100644 --- a/src-tauri/src/state.rs +++ b/src-tauri/src/state.rs @@ -1,5 +1,6 @@ //! 应用全局状态 — 数据库、Repo、事件总线、节点注册表、AI 会话 +use std::collections::HashMap; use std::path::Path; use std::sync::Arc; @@ -16,6 +17,7 @@ use df_storage::crud::{ use df_storage::db::Database; use df_workflow::eventbus::EventBus; use df_workflow::registry::NodeRegistry; +use df_workflow::state::StateMachine; use crate::commands::ai::AiSession; @@ -172,6 +174,14 @@ pub struct AppState { // ── LLM 并发控制 ── /// LLM 调用并发上限（全局 + 单对话双层 Semaphore，运行时可调） pub llm_concurrency: LlmConcurrency, + // ── 工作流执行状态 ── + /// 工作流执行 → 节点状态机注册表 + /// + /// run_workflow 创建执行器后注册其 state_machine（StateMachine 内部 Arc， + /// clone 共享底层 HashMap，与下沉到 NodeContext.node_status 的是同一份）； + /// cancel_workflow_node IPC 经 execution_id 取出后 set_cancelled， + /// 直达运行中 HumanNode 的 is_cancelled 轮询。执行完成（成功/失败）后移除条目。 + pub workflow_state_registry: Arc>>, } impl AppState { @@ -195,6 +205,7 @@ impl AppState { knowledge_config: Arc::new(Mutex::new(KnowledgeConfig::default())), settings: SettingsRepo::new(&db), llm_concurrency: LlmConcurrency::new(3, 2), + workflow_state_registry: Arc::new(Mutex::new(HashMap::new())), db, event_bus: EventBus::new(), registry: Arc::new(build_registry()), diff --git a/src/stores/project.ts b/src/stores/project.ts index 1030c0f..3d5da4e 100644 --- a/src/stores/project.ts +++ b/src/stores/project.ts @@ -244,6 +244,22 @@ function createStore() { } } + /** 取消当前待审批节点（调 cancel_workflow_node IPC 置节点 Cancelled） */ + async function cancelHumanApproval() { + if (!state.pendingApproval) return + + try { + await invoke('cancel_workflow_node', { + execution_id: state.pendingApproval.execution_id, + node_id: state.pendingApproval.node_id, + }) + // 清除待审批状态 + state.pendingApproval = null + } catch (error) { + console.error('取消审批失败:', error) + } + } + function stopEventListener() { if (_eventUnlisten) { try { @@ -284,7 +300,7 @@ function createStore() { loadIdeas, createIdea, updateIdea, deleteIdea, evaluateIdea, promoteIdea, // workflow actions runWorkflow, loadWorkflowExecutions, startEventListener, stopEventListener, clearLiveEvents, - approveHumanApproval, pendingApproval: computed(() => state.pendingApproval), + approveHumanApproval, cancelHumanApproval, pendingApproval: computed(() => state.pendingApproval), // computed stats, }) diff --git a/src/views/ProjectDetail.vue b/src/views/ProjectDetail.vue index ec17bb7..3cd42a3 100644 --- a/src/views/ProjectDetail.vue +++ b/src/views/ProjectDetail.vue @@ -189,6 +189,7 @@
+
@@ -383,6 +384,12 @@ async function handleApproval(decision: string) { showApprovalDialog.value = false } +// ── 取消审批（调 cancel_workflow_node IPC 置节点 Cancelled）── +async function handleCancelApproval() { + await store.cancelHumanApproval() + showApprovalDialog.value = false +} + // ── 删除项目(软删 → 回收站,可恢复)── async function handleDelete() { const p = currentProject.value