feat(workflow): carry visit + parallel group/branch ids on stage events

Adds visit: u32 to Event::StageStarted/Completed/Failed/Retrying so
stored_event_fields() can derive stage_id = "{node_id}@{visit}".
Adds parallel_group_id/parallel_branch_id to ParallelBranchStarted/
Completed Events, computed once in handler/parallel.rs from the
parent parallel node id + visit_from_context + branch index.
Emission sites in lifecycle/event.rs populate visit from
state.node_visits via a new stage_visit helper.

Stored_event_fields() still leaves stage_id and parallel ids None
pending the extraction pass in the next commit.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

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,

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,

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,

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,
         };

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<String>,
@@ -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,
             }),

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,

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<dyn Sandbox>,
             worktree_path: Option<PathBuf>,
@@ -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()),

lib/crates/fabro-workflow/src/lifecycle/event.rs

@@ -79,6 +79,11 @@ fn response_from_outcome(node_id: &str, outcome: &Outcome) -> Option<String> {
         .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<WorkflowGraph> for EventLifecycle {
     async fn on_run_start(&self, _graph: &WorkflowGraph, _state: &WfRunState) -> CoreResult<()> {
@@ -120,12 +125,14 @@ impl RunLifecycle<WorkflowGraph> 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<WorkflowGraph> 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<WorkflowGraph> 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<WorkflowGraph> 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<WorkflowGraph> 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<WorkflowGraph> 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<WorkflowGraph> 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<WorkflowGraph> 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(),

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,