diff --git a/crates/df-types/src/events.rs b/crates/df-types/src/events.rs index d2adbf7..24326d7 100644 --- a/crates/df-types/src/events.rs +++ b/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, }, diff --git a/crates/df-workflow/src/executor.rs b/crates/df-workflow/src/executor.rs index d3a7f05..ec5d9cc 100644 --- a/crates/df-workflow/src/executor.rs +++ b/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; diff --git a/src-tauri/src/commands/workflow.rs b/src-tauri/src/commands/workflow.rs index e5e93cc..d4cb5b1 100644 --- a/src-tauri/src/commands/workflow.rs +++ b/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, })