diff --git a/crates/arc-workflows/src/condition.rs b/crates/arc-workflows/src/condition.rs index 5d994f91d..3fe8ad9e1 100644 --- a/crates/arc-workflows/src/condition.rs +++ b/crates/arc-workflows/src/condition.rs @@ -145,13 +145,7 @@ mod tests { fn make_outcome(status: StageStatus) -> Outcome { Outcome { status, - preferred_label: None, - suggested_next_ids: Vec::new(), - context_updates: std::collections::HashMap::new(), - notes: None, - failure: None, - usage: None, - files_touched: Vec::new(), + ..Outcome::success() } } diff --git a/crates/arc-workflows/src/engine.rs b/crates/arc-workflows/src/engine.rs index 87212f6bd..3f3f70575 100644 --- a/crates/arc-workflows/src/engine.rs +++ b/crates/arc-workflows/src/engine.rs @@ -1502,16 +1502,32 @@ impl WorkflowRunEngine { } // Step 5: Select next edge (done before checkpoint so we can store next_node_id) - let next_edge = select_edge(&node.id, &outcome, &context, graph); - if let Some(edge) = next_edge { + // If the handler specified a direct jump (e.g., parallel -> fan-in), + // bypass edge selection entirely. + let (next_edge, jump_target) = if let Some(ref target) = outcome.jump_to_node { self.services.emitter.emit(&WorkflowRunEvent::EdgeSelected { from_node: node.id.clone(), - to_node: edge.to.clone(), - label: edge.label().map(String::from), - condition: edge.condition().map(String::from), + to_node: target.clone(), + label: None, + condition: None, }); - } - let next_node_id_for_checkpoint = next_edge.map(|e| e.to.clone()); + (None, Some(target.clone())) + } else { + let edge = select_edge(&node.id, &outcome, &context, graph); + if let Some(ref e) = edge { + self.services.emitter.emit(&WorkflowRunEvent::EdgeSelected { + from_node: node.id.clone(), + to_node: e.to.clone(), + label: e.label().map(String::from), + condition: e.condition().map(String::from), + }); + } + (edge, None) + }; + let next_node_id_for_checkpoint = jump_target + .as_ref() + .cloned() + .or_else(|| next_edge.map(|e| e.to.clone())); // Step 6: Save checkpoint with all state let mut checkpoint = Checkpoint::from_context( @@ -1648,7 +1664,12 @@ impl WorkflowRunEngine { } } - // Step 7: Follow selected edge + // Step 7: Follow selected edge (or direct jump) + if let Some(target) = jump_target { + incoming_edge = None; + current_node_id = target; + continue; + } match next_edge { None => { // Gap #1: Failure routing -- when FAIL and no matching edge, diff --git a/crates/arc-workflows/src/error.rs b/crates/arc-workflows/src/error.rs index ef42f71f1..172813c38 100644 --- a/crates/arc-workflows/src/error.rs +++ b/crates/arc-workflows/src/error.rs @@ -399,13 +399,8 @@ impl ArcError { }; crate::outcome::Outcome { status: crate::outcome::StageStatus::Fail, - preferred_label: None, - suggested_next_ids: Vec::new(), - context_updates: std::collections::HashMap::new(), - notes: None, failure: Some(failure), - usage: None, - files_touched: Vec::new(), + ..crate::outcome::Outcome::success() } } } diff --git a/crates/arc-workflows/src/handler/parallel.rs b/crates/arc-workflows/src/handler/parallel.rs index b01ba230c..f8d986f4f 100644 --- a/crates/arc-workflows/src/handler/parallel.rs +++ b/crates/arc-workflows/src/handler/parallel.rs @@ -686,15 +686,13 @@ impl Handler for ParallelHandler { } }; - // Build suggested_next_ids from branch targets - let branch_ids: Vec = results.iter().map(|r| r.id.clone()).collect(); + // Find the join/convergence node: follow each branch's outgoing edges + // and find the common downstream target (typically the fan-in node). + let join_node = find_join_node(&results, graph); let is_fail = status == StageStatus::Fail; let mut outcome = Outcome { status, - preferred_label: None, - suggested_next_ids: branch_ids, - context_updates: std::collections::HashMap::new(), notes: Some(format!( "Parallel node dispatched {total} branches ({success_count} succeeded, {fail_count} failed)" )), @@ -706,8 +704,8 @@ impl Handler for ParallelHandler { } else { None }, - usage: None, - files_touched: Vec::new(), + jump_to_node: if is_fail { None } else { join_node }, + ..Outcome::success() }; if is_fail { @@ -718,6 +716,39 @@ impl Handler for ParallelHandler { } } +/// Find the convergence (join/fan-in) node by following each branch's outgoing edges +/// and finding the first node reachable from all branches. +fn find_join_node(results: &[BranchResult], graph: &Graph) -> Option { + if results.is_empty() { + return None; + } + + // Collect outgoing targets for each branch + let mut target_sets: Vec> = Vec::new(); + for result in results { + let targets: std::collections::HashSet = graph + .outgoing_edges(&result.id) + .into_iter() + .map(|e| e.to.clone()) + .collect(); + target_sets.push(targets); + } + + // Find the intersection — nodes reachable from ALL branches + let Some(first) = target_sets.first() else { + return None; + }; + let common: std::collections::HashSet<&String> = first + .iter() + .filter(|id| target_sets.iter().all(|set| set.contains(*id))) + .collect(); + + // Return the first common target (lexically sorted for determinism) + let mut common_sorted: Vec<&String> = common.into_iter().collect(); + common_sorted.sort(); + common_sorted.first().map(|id| (*id).clone()) +} + #[cfg(test)] mod tests { use super::*; diff --git a/crates/arc-workflows/src/outcome.rs b/crates/arc-workflows/src/outcome.rs index ec2b10fd1..158892ada 100644 --- a/crates/arc-workflows/src/outcome.rs +++ b/crates/arc-workflows/src/outcome.rs @@ -98,6 +98,10 @@ pub struct Outcome { pub usage: Option, #[serde(default, skip_serializing_if = "Vec::is_empty")] pub files_touched: Vec, + /// When set, the engine bypasses edge selection and jumps directly to this node. + /// Used by the parallel handler to skip re-executing branch nodes. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub jump_to_node: Option, } impl Outcome { @@ -112,6 +116,7 @@ impl Outcome { failure: None, usage: None, files_touched: Vec::new(), + jump_to_node: None, } } @@ -126,6 +131,7 @@ impl Outcome { failure: Some(FailureDetail::new(reason, FailureClass::Deterministic)), usage: None, files_touched: Vec::new(), + jump_to_node: None, } } @@ -142,6 +148,7 @@ impl Outcome { failure: Some(FailureDetail::new(reason, failure_class)), usage: None, files_touched: Vec::new(), + jump_to_node: None, } } @@ -158,6 +165,7 @@ impl Outcome { failure: Some(FailureDetail::new(reason, failure_class)), usage: None, files_touched: Vec::new(), + jump_to_node: None, } } @@ -181,6 +189,7 @@ impl Outcome { failure: None, usage: None, files_touched: Vec::new(), + jump_to_node: None, } }