diff --git a/lib/crates/fabro-cli/src/commands/run/run_progress/event.rs b/lib/crates/fabro-cli/src/commands/run/run_progress/event.rs index f950f2216..20f5d98a0 100644 --- a/lib/crates/fabro-cli/src/commands/run/run_progress/event.rs +++ b/lib/crates/fabro-cli/src/commands/run/run_progress/event.rs @@ -510,6 +510,7 @@ mod tests { node_id: "plan".into(), name: "Plan".into(), index: 0, + visit: 1, duration_ms: 5000, status: "success".into(), preferred_label: None, diff --git a/lib/crates/fabro-cli/src/commands/run/run_progress/mod.rs b/lib/crates/fabro-cli/src/commands/run/run_progress/mod.rs index 0354d9ead..63507433e 100644 --- a/lib/crates/fabro-cli/src/commands/run/run_progress/mod.rs +++ b/lib/crates/fabro-cli/src/commands/run/run_progress/mod.rs @@ -487,6 +487,7 @@ mod tests { node_id: node_id.into(), name: name.into(), index: 0, + visit: 1, handler_type: String::new(), attempt: 1, max_attempts: 1, @@ -510,6 +511,7 @@ mod tests { node_id: node_id.into(), name: name.into(), index: 0, + visit: 1, duration_ms: 5000, status: "success".into(), preferred_label: None, @@ -561,6 +563,8 @@ mod tests { emit( &mut ui, Event::ParallelBranchStarted { + parallel_group_id: "fork1@1".into(), + parallel_branch_id: "fork1@1:0".into(), branch: "security".into(), index: 0, }, @@ -576,6 +580,8 @@ mod tests { emit( &mut ui, Event::ParallelBranchCompleted { + parallel_group_id: "fork1@1".into(), + parallel_branch_id: "fork1@1:0".into(), branch: "security".into(), index: 0, duration_ms: 2000, @@ -607,6 +613,8 @@ mod tests { emit( &mut ui, Event::ParallelBranchStarted { + parallel_group_id: "fork1@1".into(), + parallel_branch_id: "fork1@1:0".into(), branch: "security".into(), index: 0, }, @@ -700,6 +708,7 @@ mod tests { node_id: "code".into(), name: "Code".into(), index: 0, + visit: 1, attempt: 2, max_attempts: 3, delay_ms: 1500, @@ -944,6 +953,7 @@ mod tests { node_id: "code".into(), name: "Code".into(), index: 0, + visit: 1, attempt: 2, max_attempts: 3, delay_ms: 1500, @@ -1108,6 +1118,8 @@ mod tests { emit( &mut ui, Event::ParallelBranchStarted { + parallel_group_id: "fork1@1".into(), + parallel_branch_id: "fork1@1:0".into(), branch: "security".into(), index: 0, }, @@ -1115,6 +1127,8 @@ mod tests { emit( &mut ui, Event::ParallelBranchCompleted { + parallel_group_id: "fork1@1".into(), + parallel_branch_id: "fork1@1:0".into(), branch: "security".into(), index: 0, duration_ms: 500, @@ -1144,6 +1158,7 @@ mod tests { node_id: "code".into(), name: "Code".into(), index: 0, + visit: 1, handler_type: "agent".into(), attempt: 1, max_attempts: 1, diff --git a/lib/crates/fabro-cli/src/commands/store/dump.rs b/lib/crates/fabro-cli/src/commands/store/dump.rs index 611f14c04..3f25253a5 100644 --- a/lib/crates/fabro-cli/src/commands/store/dump.rs +++ b/lib/crates/fabro-cli/src/commands/store/dump.rs @@ -599,6 +599,7 @@ mod tests { node_id: "code".to_string(), name: "Code".to_string(), index: 1, + visit: 2, duration_ms: 250, status: "partial_success".to_string(), preferred_label: None, diff --git a/lib/crates/fabro-workflow/src/error.rs b/lib/crates/fabro-workflow/src/error.rs index 942b442f1..b81f690a6 100644 --- a/lib/crates/fabro-workflow/src/error.rs +++ b/lib/crates/fabro-workflow/src/error.rs @@ -1692,6 +1692,7 @@ mod tests { node_id: "code".into(), name: "code".into(), index: 0, + visit: 1, failure: failure.clone(), will_retry: false, }; diff --git a/lib/crates/fabro-workflow/src/event.rs b/lib/crates/fabro-workflow/src/event.rs index 220d58ee0..7bcf75194 100644 --- a/lib/crates/fabro-workflow/src/event.rs +++ b/lib/crates/fabro-workflow/src/event.rs @@ -137,6 +137,7 @@ pub enum Event { node_id: String, name: String, index: usize, + visit: u32, handler_type: String, attempt: usize, max_attempts: usize, @@ -145,6 +146,7 @@ pub enum Event { node_id: String, name: String, index: usize, + visit: u32, duration_ms: u64, status: String, preferred_label: Option, @@ -175,6 +177,7 @@ pub enum Event { node_id: String, name: String, index: usize, + visit: u32, failure: FailureDetail, will_retry: bool, }, @@ -182,6 +185,7 @@ pub enum Event { node_id: String, name: String, index: usize, + visit: u32, attempt: usize, max_attempts: usize, delay_ms: u64, @@ -193,10 +197,14 @@ pub enum Event { join_policy: String, }, ParallelBranchStarted { + parallel_group_id: String, + parallel_branch_id: String, branch: String, index: usize, }, ParallelBranchCompleted { + parallel_group_id: String, + parallel_branch_id: String, branch: String, index: usize, duration_ms: u64, @@ -676,6 +684,7 @@ impl Event { index, failure, will_retry, + .. } => { let error_msg = &failure.message; if *will_retry { @@ -705,6 +714,7 @@ impl Event { attempt, max_attempts, delay_ms, + .. } => { warn!( node_id, @@ -723,7 +733,7 @@ impl Event { } => { debug!(branch_count, join_policy, "Parallel execution started"); } - Self::ParallelBranchStarted { branch, index } => { + Self::ParallelBranchStarted { branch, index, .. } => { debug!(branch, index, "Parallel branch started"); } Self::ParallelBranchCompleted { @@ -2759,6 +2769,7 @@ mod tests { node_id: "plan".to_string(), name: "Plan".to_string(), index: 0, + visit: 1, duration_ms: 5000, status: "success".to_string(), preferred_label: None, @@ -2797,6 +2808,7 @@ mod tests { node_id: "plan".to_string(), name: "Plan".to_string(), index: 0, + visit: 1, duration_ms: 5000, status: "success".to_string(), preferred_label: None, @@ -2831,6 +2843,7 @@ mod tests { node_id: "code".to_string(), name: "Code".to_string(), index: 1, + visit: 1, failure: FailureDetail::new( "lint failed", crate::outcome::FailureCategory::Deterministic, @@ -3016,6 +3029,8 @@ mod tests { ); assert_eq!( event_name(&Event::ParallelBranchStarted { + parallel_group_id: "plan@1".to_string(), + parallel_branch_id: "plan@1:0".to_string(), branch: "fork".to_string(), index: 0, }), diff --git a/lib/crates/fabro-workflow/src/git.rs b/lib/crates/fabro-workflow/src/git.rs index 24ccf9fac..504bd03e5 100644 --- a/lib/crates/fabro-workflow/src/git.rs +++ b/lib/crates/fabro-workflow/src/git.rs @@ -449,6 +449,7 @@ mod tests { node_id: "work".into(), name: "Work".into(), index: 2, + visit: 2, duration_ms: 100, status: "success".into(), preferred_label: None, diff --git a/lib/crates/fabro-workflow/src/handler/parallel.rs b/lib/crates/fabro-workflow/src/handler/parallel.rs index 5c7b8158a..bec2bb5d8 100644 --- a/lib/crates/fabro-workflow/src/handler/parallel.rs +++ b/lib/crates/fabro-workflow/src/handler/parallel.rs @@ -132,6 +132,7 @@ impl Handler for ParallelHandler { struct BranchSetup { target_id: String, branch_index: usize, + parallel_branch_id: String, branch_context: Context, sandbox: Arc, worktree_path: Option, @@ -150,9 +151,12 @@ impl Handler for ParallelHandler { .unwrap_or("wait_all"), ); + let parallel_visit = u32::try_from(visit_from_context(context)).unwrap_or(u32::MAX); + let parallel_group_id = format!("{}@{}", node.id, parallel_visit); + services.emitter.emit(&Event::ParallelStarted { node_id: node.id.clone(), - visit: u32::try_from(visit_from_context(context)).unwrap_or(u32::MAX), + visit: parallel_visit, branch_count: branches.len(), join_policy: join_policy.to_string(), }); @@ -253,9 +257,11 @@ impl Handler for ParallelHandler { (Arc::clone(&services.sandbox), None) }; + let parallel_branch_id = format!("{parallel_group_id}:{branch_index}"); branch_setups.push(BranchSetup { target_id, branch_index, + parallel_branch_id, branch_context, sandbox: branch_sandbox, worktree_path, @@ -283,6 +289,7 @@ impl Handler for ParallelHandler { .as_ref() .map(|gs| gs.git_author.clone()) .unwrap_or_default(); + let group_id = parallel_group_id.clone(); let handle = tokio::spawn(async move { let _permit = sem @@ -291,6 +298,8 @@ impl Handler for ParallelHandler { .map_err(|e| FabroError::handler(format!("semaphore error: {e}")))?; emitter.emit(&Event::ParallelBranchStarted { + parallel_group_id: group_id.clone(), + parallel_branch_id: setup.parallel_branch_id.clone(), branch: setup.target_id.clone(), index: setup.branch_index, }); @@ -302,6 +311,8 @@ impl Handler for ParallelHandler { setup.target_id )); emitter.emit(&Event::ParallelBranchCompleted { + parallel_group_id: group_id.clone(), + parallel_branch_id: setup.parallel_branch_id.clone(), branch: setup.target_id.clone(), index: setup.branch_index, duration_ms: millis_u64(branch_start.elapsed()), @@ -386,6 +397,8 @@ impl Handler for ParallelHandler { }; emitter.emit(&Event::ParallelBranchCompleted { + parallel_group_id: group_id.clone(), + parallel_branch_id: setup.parallel_branch_id.clone(), branch: setup.target_id.clone(), index: setup.branch_index, duration_ms: millis_u64(branch_start.elapsed()), diff --git a/lib/crates/fabro-workflow/src/lifecycle/event.rs b/lib/crates/fabro-workflow/src/lifecycle/event.rs index dbb4a5a2b..f6cbae9ed 100644 --- a/lib/crates/fabro-workflow/src/lifecycle/event.rs +++ b/lib/crates/fabro-workflow/src/lifecycle/event.rs @@ -79,6 +79,11 @@ fn response_from_outcome(node_id: &str, outcome: &Outcome) -> Option { .and_then(|value| value.as_str().map(ToOwned::to_owned)) } +fn stage_visit(state: &WfRunState, node_id: &str) -> u32 { + let visits = state.node_visits.get(node_id).copied().unwrap_or(1); + u32::try_from(visits.max(1)).unwrap_or(u32::MAX) +} + #[async_trait] impl RunLifecycle for EventLifecycle { async fn on_run_start(&self, _graph: &WorkflowGraph, _state: &WfRunState) -> CoreResult<()> { @@ -120,12 +125,14 @@ impl RunLifecycle for EventLifecycle { } let gv = node.inner(); let stage_index = state.stage_index; + let visit = stage_visit(state, &gv.id); let (loop_failure_signatures, restart_failure_signatures) = snapshot_failure_signatures(&self.circuit_breaker); self.emitter.emit(&Event::StageStarted { node_id: gv.id.clone(), name: gv.label().to_string(), index: stage_index, + visit, handler_type: gv.handler_type().unwrap_or_default().to_string(), attempt: 1, max_attempts: 1, @@ -134,6 +141,7 @@ impl RunLifecycle for EventLifecycle { node_id: gv.id.clone(), name: gv.label().to_string(), index: stage_index, + visit, duration_ms: 0, status: StageStatus::Success.to_string(), preferred_label: None, @@ -167,6 +175,7 @@ impl RunLifecycle for EventLifecycle { node_id: gv.id.clone(), name: gv.label().to_string(), index: state.stage_index, + visit: stage_visit(state, &gv.id), handler_type: gv.handler_type().unwrap_or_default().to_string(), attempt: ctx.attempt as usize, max_attempts: ctx.max_attempts as usize, @@ -183,11 +192,13 @@ impl RunLifecycle for EventLifecycle { let gv = ctx.node.inner(); let outcome = &ctx.result.outcome; let stage_index = state.stage_index; + let visit = stage_visit(state, &gv.id); self.emitter.emit(&Event::StageFailed { node_id: gv.id.clone(), name: gv.label().to_string(), index: stage_index, + visit, failure: outcome.failure.clone().unwrap_or_else(|| { FailureDetail::new("handler failed", FailureCategory::TransientInfra) }), @@ -198,6 +209,7 @@ impl RunLifecycle for EventLifecycle { node_id: gv.id.clone(), name: gv.label().to_string(), index: stage_index, + visit, attempt: ctx.attempt as usize, max_attempts: ctx.result.max_attempts as usize, delay_ms: ctx @@ -221,6 +233,7 @@ impl RunLifecycle for EventLifecycle { } let gv = node.inner(); let stage_index = state.stage_index; + let visit = stage_visit(state, &gv.id); let duration_ms = u64::try_from(result.duration.as_millis()).unwrap(); let (loop_failure_signatures, restart_failure_signatures) = snapshot_failure_signatures(&self.circuit_breaker); @@ -230,6 +243,7 @@ impl RunLifecycle for EventLifecycle { node_id: gv.id.clone(), name: gv.label().to_string(), index: stage_index, + visit, failure: outcome.failure.clone().unwrap_or_else(|| { FailureDetail::new("handler failed", FailureCategory::Deterministic) }), @@ -240,6 +254,7 @@ impl RunLifecycle for EventLifecycle { node_id: gv.id.clone(), name: gv.label().to_string(), index: stage_index, + visit, duration_ms, status: outcome.status.to_string(), preferred_label: outcome.preferred_label.clone(), diff --git a/lib/crates/fabro-workflow/src/pipeline/pull_request.rs b/lib/crates/fabro-workflow/src/pipeline/pull_request.rs index 914a9daf5..0f8631b79 100644 --- a/lib/crates/fabro-workflow/src/pipeline/pull_request.rs +++ b/lib/crates/fabro-workflow/src/pipeline/pull_request.rs @@ -1197,6 +1197,7 @@ mod tests { node_id: "plan".to_string(), name: "plan".to_string(), index: 0, + visit: 1, duration_ms: 1, status: "success".to_string(), preferred_label: None,