修复: B-03b-R10③工作流事件补execution_id治本并发串扰(emit填+消费exec_id守卫)

crates/df-types/src/events.rs

@@ -73,11 +73,23 @@ pub enum WorkflowEvent {
         waiting_node: NodeId,
     },
     /// 工作流执行完成
+    ///
+    /// B-03b-R10 ③(波17 治本): execution_id 字段标识本次终态事件归属的工作流执行。
+    /// 此前该变体不带 execution_id,AppState.event_bus 全局单例下并发工作流的终态事件
+    /// 会被 forward 循环的 matches! 完成判定(workflow.rs:159-163 只看变体不看 exec_id)误判,
+    /// 致任一工作流终态触发所有 forward 循环 break(过早断他人链)。
+    /// #[serde(default)] 向后兼容: 老事件/老 DB 反序列化(无 execution_id)填空串不炸。
     WorkflowCompleted {
+        #[serde(default)]
+        execution_id: String,
         total_duration_ms: u64,
     },
     /// 工作流执行失败
+    ///
+    /// B-03b-R10 ③(波17 治本): execution_id 字段标识本次终态事件归属的工作流执行(同 WorkflowCompleted)。
     WorkflowFailed {
+        #[serde(default)]
+        execution_id: String,
         error: String,
         failed_node: NodeId,
     },

crates/df-workflow/src/executor.rs

@@ -179,8 +179,11 @@ impl DagExecutor {
         }
 
         let total = start.elapsed().as_millis() as u64;
+        // B-03b-R10 ③: WorkflowCompleted 携带 execution_id,供 forward 循环按 exec_id 匹配
+        // (全局 event_bus 单例下并发工作流的终态事件不再误触发他人 forward 的完成判定)。
         self.event_bus
             .send(WorkflowEvent::WorkflowCompleted {
+                execution_id: self.execution_id.clone(),
                 total_duration_ms: total,
             })
             .await;

src-tauri/src/commands/workflow.rs

@@ -156,10 +156,19 @@ pub async fn run_workflow_inner(
         loop {
             match rx.recv().await {
                 Ok(event) => {
+                    // B-03b-R10 ③(波17 治本): 完成判定加 execution_id 匹配。
+                    // AppState.event_bus 是全局单例,并发工作流各自订阅同一总线;
+                    // 此前 matches! 只看变体不看 exec_id,任一工作流的 WorkflowCompleted/Failed
+                    // 都会触发本 forward 循环 break,过早断他人链(并发串扰根因)。
+                    // 现在按变体+exec_id 双重匹配:只处理属于本 forward_exec_id 的终态事件,
+                    // 他人终态事件忽略(emit 给前端后继续循环),不再误 break。
+                    // execution_id 字段 #[serde(default)] 向后兼容老事件(空串):
+                    // 自家终态事件本就带正确 exec_id;空串匹配只可能漏(自家老事件),不会误 break。
                     let finished = matches!(
-                        event,
-                        WorkflowEvent::WorkflowCompleted { .. }
-                            | WorkflowEvent::WorkflowFailed { .. }
+                        &event,
+                        WorkflowEvent::WorkflowCompleted { execution_id, .. }
+                            | WorkflowEvent::WorkflowFailed { execution_id, .. }
+                            if execution_id == &forward_exec_id
                     );
                     let payload = WorkflowEventPayload {
                         execution_id: forward_exec_id.clone(),
@@ -197,14 +206,17 @@ pub async fn run_workflow_inner(
                         Ok(Some(record)) => {
                             let terminal = match record.status.as_str() {
                                 "completed" => Some(WorkflowEvent::WorkflowCompleted {
+                                    execution_id: forward_exec_id.clone(),
                                     total_duration_ms: 0,
                                 }),
                                 "failed" => Some(WorkflowEvent::WorkflowFailed {
+                                    execution_id: forward_exec_id.clone(),
                                     error: "工作流执行失败（Lagged 兜底补发，详情见 DB）"
                                         .to_string(),
                                     failed_node: String::new(),
                                 }),
                                 "cancelled" => Some(WorkflowEvent::WorkflowFailed {
+                                    execution_id: forward_exec_id.clone(),
                                     error: "工作流被取消（Lagged 兜底补发）".to_string(),
                                     failed_node: String::new(),
                                 }),
@@ -350,6 +362,7 @@ pub async fn run_workflow_inner(
                 .unwrap_or_default();
             event_bus
                 .send(WorkflowEvent::WorkflowFailed {
+                    execution_id: exec_id.clone(),
                     error,
                     failed_node,
                 })