Simplify SSH access API and add parallel fan-in jump_to_node

- Simplify create_ssh_access to return String directly
- Add jump_to_node field to Outcome for parallel handler fan-in
- Find convergence node from branch outgoing edges
- Use ..Outcome::success() spread in tests and error paths

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

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()
         }
     }
 

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,

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()
         }
     }
 }

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<String> = 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<String> {
+    if results.is_empty() {
+        return None;
+    }
+
+    // Collect outgoing targets for each branch
+    let mut target_sets: Vec<std::collections::HashSet<String>> = Vec::new();
+    for result in results {
+        let targets: std::collections::HashSet<String> = 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::*;

crates/arc-workflows/src/outcome.rs

@@ -98,6 +98,10 @@ pub struct Outcome {
     pub usage: Option<StageUsage>,
     #[serde(default, skip_serializing_if = "Vec::is_empty")]
     pub files_touched: Vec<String>,
+    /// 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<String>,
 }
 
 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,
         }
     }