From 6dc359f867d0f7c750969caa1c6b40a93838d8cc Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Wed, 4 Mar 2026 08:58:44 -0500 Subject: [PATCH] Show parallel branches in progress UI Handle ParallelBranchStarted/Completed events as nested entries under the fork stage, using the same indented spinner styling as tool calls. Branch bars persist after the stage completes so they remain visible in the final output. Co-Authored-By: Claude Opus 4.6 --- crates/arc-workflows/src/cli/progress.rs | 234 ++++++++++++++++++++++- 1 file changed, 232 insertions(+), 2 deletions(-) diff --git a/crates/arc-workflows/src/cli/progress.rs b/crates/arc-workflows/src/cli/progress.rs index a11e8c60e..7467262f4 100644 --- a/crates/arc-workflows/src/cli/progress.rs +++ b/crates/arc-workflows/src/cli/progress.rs @@ -149,6 +149,7 @@ struct ToolCallEntry { tool_call_id: String, status: ToolCallStatus, bar: ProgressBar, + is_branch: bool, } // ── Active stage ──────────────────────────────────────────────────────── @@ -182,6 +183,7 @@ pub struct ProgressUI { sandbox_bar: Option, setup_bar: Option, any_stage_started: bool, + parallel_parent: Option, } impl ProgressUI { @@ -200,6 +202,7 @@ impl ProgressUI { sandbox_bar: None, setup_bar: None, any_stage_started: false, + parallel_parent: None, } } @@ -216,7 +219,11 @@ impl ProgressUI { pub fn finish(&mut self) { for (_id, stage) in self.active_stages.drain() { for entry in &stage.tool_calls { - entry.bar.finish_and_clear(); + if entry.is_branch { + entry.bar.abandon(); + } else { + entry.bar.finish_and_clear(); + } } stage.spinner.finish_and_clear(); } @@ -281,6 +288,30 @@ impl ProgressUI { WorkflowRunEvent::StageFailed { node_id, name, .. } => { self.finish_stage(node_id, name, red_cross(), ""); } + WorkflowRunEvent::ParallelStarted { .. } => { + // The fork stage is the (only) active stage at this point. + // In Plain mode active_stages is empty, so use a sentinel. + self.parallel_parent = self + .active_stages + .keys() + .next() + .cloned() + .or_else(|| Some(String::new())); + } + WorkflowRunEvent::ParallelBranchStarted { branch, .. } => { + self.on_parallel_branch_started(branch); + } + WorkflowRunEvent::ParallelBranchCompleted { + branch, + duration_ms, + status, + .. + } => { + self.on_parallel_branch_completed(branch, *duration_ms, status); + } + WorkflowRunEvent::ParallelCompleted { .. } => { + self.parallel_parent = None; + } WorkflowRunEvent::Agent { stage, event } => { self.on_agent_event(stage, event); } @@ -435,7 +466,12 @@ impl ProgressUI { ProgressRenderer::Tty(_) => { if let Some(stage) = self.active_stages.remove(node_id) { for entry in &stage.tool_calls { - entry.bar.finish_and_clear(); + if entry.is_branch { + // Already finished by on_parallel_branch_completed; keep visible + entry.bar.abandon(); + } else { + entry.bar.finish_and_clear(); + } } stage.spinner.set_style(style_stage_done()); stage.spinner.set_prefix(prefix.to_string()); @@ -523,11 +559,78 @@ impl ProgressUI { tool_call_id: tool_call_id.to_string(), status: ToolCallStatus::Running, bar, + is_branch: false, }); } } } + // ── Parallel branches ───────────────────────────────────────────── + + fn on_parallel_branch_started(&mut self, branch: &str) { + let parent_id = match &self.parallel_parent { + Some(id) => id.clone(), + None => return, + }; + + if let ProgressRenderer::Tty(tty) = &self.renderer { + if let Some(stage) = self.active_stages.get_mut(&parent_id) { + let bar = tty.multi.insert_after( + stage.tool_calls.back().map_or(&stage.spinner, |e| &e.bar), + ProgressBar::new_spinner(), + ); + bar.set_style(style_tool_running()); + bar.set_message(branch.to_string()); + bar.enable_steady_tick(Duration::from_millis(100)); + stage.tool_calls.push_back(ToolCallEntry { + display_name: branch.to_string(), + tool_call_id: branch.to_string(), + status: ToolCallStatus::Running, + bar, + is_branch: true, + }); + } + } + } + + fn on_parallel_branch_completed(&mut self, branch: &str, duration_ms: u64, status: &str) { + let succeeded = matches!(status, "success" | "partial_success"); + let glyph = if succeeded { green_check() } else { red_cross() }; + let dur = format_duration_ms(duration_ms); + + let parent_id = match &self.parallel_parent { + Some(id) => id.clone(), + None => return, + }; + + match &self.renderer { + ProgressRenderer::Tty(_) => { + if let Some(stage) = self.active_stages.get_mut(&parent_id) { + if let Some(entry) = stage + .tool_calls + .iter_mut() + .find(|e| e.tool_call_id == branch) + { + entry.status = if succeeded { + ToolCallStatus::Succeeded + } else { + ToolCallStatus::Failed + }; + let elapsed = format_duration_short(entry.bar.elapsed()); + entry.bar.set_style(style_tool_done()); + entry.bar.set_prefix(elapsed); + entry + .bar + .finish_with_message(format!("{glyph} {}", entry.display_name)); + } + } + } + ProgressRenderer::Plain => { + eprintln!(" {glyph} {branch} {dur}"); + } + } + } + fn on_tool_call_completed(&mut self, stage_node_id: &str, tool_call_id: &str, is_error: bool) { if let ProgressRenderer::Tty(_) = &self.renderer { if let Some(stage) = self.active_stages.get_mut(stage_node_id) { @@ -609,3 +712,130 @@ impl Interviewer for ProgressAwareInterviewer { self.show_bars(); } } + +#[cfg(test)] +mod tests { + use super::*; + + fn stage_started(node_id: &str, name: &str) -> WorkflowRunEvent { + WorkflowRunEvent::StageStarted { + node_id: node_id.into(), + name: name.into(), + index: 0, + handler_type: None, + script: None, + attempt: 1, + max_attempts: 1, + } + } + + #[test] + fn parallel_branches_tracked_as_tool_calls() { + let mut ui = ProgressUI::new(true); + + ui.handle_event(&stage_started("fork1", "Fork Analysis")); + assert!(ui.active_stages.contains_key("fork1")); + assert!(ui.parallel_parent.is_none()); + + ui.handle_event(&WorkflowRunEvent::ParallelStarted { + branch_count: 2, + join_policy: "wait_all".into(), + error_policy: "continue".into(), + }); + assert_eq!(ui.parallel_parent.as_deref(), Some("fork1")); + + // Branch started → creates a tool_call entry + ui.handle_event(&WorkflowRunEvent::ParallelBranchStarted { + branch: "security".into(), + index: 0, + }); + let stage = ui.active_stages.get("fork1").unwrap(); + assert_eq!(stage.tool_calls.len(), 1); + assert_eq!(stage.tool_calls[0].tool_call_id, "security"); + assert!(matches!(stage.tool_calls[0].status, ToolCallStatus::Running)); + + // Branch completed → marks entry as succeeded + ui.handle_event(&WorkflowRunEvent::ParallelBranchCompleted { + branch: "security".into(), + index: 0, + duration_ms: 2000, + status: "success".into(), + }); + let stage = ui.active_stages.get("fork1").unwrap(); + assert!(matches!( + stage.tool_calls[0].status, + ToolCallStatus::Succeeded + )); + + // Second branch + ui.handle_event(&WorkflowRunEvent::ParallelBranchStarted { + branch: "quality".into(), + index: 1, + }); + ui.handle_event(&WorkflowRunEvent::ParallelBranchCompleted { + branch: "quality".into(), + index: 1, + duration_ms: 3000, + status: "success".into(), + }); + let stage = ui.active_stages.get("fork1").unwrap(); + assert_eq!(stage.tool_calls.len(), 2); + + // Parallel completed → clears parent + ui.handle_event(&WorkflowRunEvent::ParallelCompleted { + duration_ms: 3000, + success_count: 2, + failure_count: 0, + }); + assert!(ui.parallel_parent.is_none()); + } + + #[test] + fn parallel_branch_failure_tracked() { + let mut ui = ProgressUI::new(true); + + ui.handle_event(&stage_started("fork1", "Fork")); + ui.handle_event(&WorkflowRunEvent::ParallelStarted { + branch_count: 1, + join_policy: "wait_all".into(), + error_policy: "continue".into(), + }); + ui.handle_event(&WorkflowRunEvent::ParallelBranchStarted { + branch: "risky".into(), + index: 0, + }); + ui.handle_event(&WorkflowRunEvent::ParallelBranchCompleted { + branch: "risky".into(), + index: 0, + duration_ms: 500, + status: "fail".into(), + }); + + let stage = ui.active_stages.get("fork1").unwrap(); + assert!(matches!( + stage.tool_calls[0].status, + ToolCallStatus::Failed + )); + } + + #[test] + fn plain_mode_sets_parallel_parent() { + let mut ui = ProgressUI::new(false); + + ui.handle_event(&stage_started("fork1", "Fork")); + ui.handle_event(&WorkflowRunEvent::ParallelStarted { + branch_count: 2, + join_policy: "wait_all".into(), + error_policy: "continue".into(), + }); + // In Plain mode, active_stages is empty so parallel_parent is a sentinel + assert!(ui.parallel_parent.is_some()); + + ui.handle_event(&WorkflowRunEvent::ParallelCompleted { + duration_ms: 1000, + success_count: 2, + failure_count: 0, + }); + assert!(ui.parallel_parent.is_none()); + } +}