From c623bf7432b8c2a95ffe8854b0faf05f8da230c8 Mon Sep 17 00:00:00 2001 From: andremw Date: Wed, 22 Jul 2026 17:50:04 +0930 Subject: [PATCH] Finalize parallel branch stages instead of leaving them Running Parallel branches run in spawned tasks that bypass the engine's StageStarted/StageCompleted lifecycle, so a branch stage was created Running by its first branch-scoped event and never reached a terminal state. On a successful run nothing swept it (only RunFailed does), so the fan-out rows spun forever with a `--` duration even after the run and its fan-in finished. Fold ParallelBranchStarted/ParallelBranchCompleted in the projection: seed started_at for the live timer, then set the terminal state and wall-time from the branch's own completion event. Co-Authored-By: Claude Opus 4.8 (1M context) --- lib/crates/fabro-store/src/run_state.rs | 77 ++++++++++++++++++++++++- 1 file changed, 75 insertions(+), 2 deletions(-) diff --git a/lib/crates/fabro-store/src/run_state.rs b/lib/crates/fabro-store/src/run_state.rs index 8335d277c..0b0c464da 100644 --- a/lib/crates/fabro-store/src/run_state.rs +++ b/lib/crates/fabro-store/src/run_state.rs @@ -513,6 +513,38 @@ impl RunProjectionReducer for RunProjection { }; stage.parallel_results = Some(parallel_results); } + EventBody::ParallelBranchStarted(_) => { + // Branches bypass the engine's StageStarted/StageCompleted + // lifecycle. Seed started_at so the branch stage drives a live + // wall-clock timer while it runs (the entry is created Running). + let Some(stage) = stage_at_stored_or_current_visit(self, stored, event.seq) else { + return Ok(()); + }; + if stage.started_at.is_none() { + stage.started_at = Some(ts); + } + stage.state = StageState::Running; + } + EventBody::ParallelBranchCompleted(props) => { + // A branch never emits its own StageCompleted, so finalize it + // here; otherwise the stage spins Running forever after the run + // (and the fan-in) is done. + let outcome = + StageOutcome::from_str(&props.status).unwrap_or(StageOutcome::Failed { + retry_requested: false, + }); + let Some(stage) = stage_at_stored_or_current_visit(self, stored, event.seq) else { + return Ok(()); + }; + stage.completion = Some(StageCompletion { + outcome, + notes: None, + failure_reason: None, + timestamp: ts, + }); + stage.timing = Some(fabro_types::StageTiming::wall_only(props.duration_ms)); + stage.state = StageState::from(outcome); + } EventBody::TodoCreated(props) => { let Some(stage) = stage_at_stored_or_current_visit(self, stored, event.seq) else { return Ok(()); @@ -1259,8 +1291,9 @@ mod tests { AgentSubFailedProps, AgentSubSpawnedProps, AgentToolCategory, AgentToolSource, AgentToolStartedProps, AgentToolSummary, AgentToolsAvailableProps, CheckpointCompletedProps, InterviewCompletedProps, InterviewOption, InterviewStartedProps, - RunCompletedProps, RunControlEffectProps, StageCompletedProps, StageFailedProps, - StagePromptProps, StageRetryingProps, StageStartedProps, + ParallelBranchCompletedProps, ParallelBranchStartedProps, RunCompletedProps, + RunControlEffectProps, StageCompletedProps, StageFailedProps, StagePromptProps, + StageRetryingProps, StageStartedProps, }; use fabro_types::settings::run::{DockerfileSource, EnvironmentProvider}; use fabro_types::{ @@ -1992,6 +2025,46 @@ mod tests { assert_eq!(stage.prompt.as_deref(), Some("prompt")); } + #[test] + fn parallel_branch_completed_finalizes_branch_stage() { + // A parallel branch never runs through the engine's StageStarted/ + // StageCompleted lifecycle: its stage entry is created Running by the + // first branch-scoped event, and only ParallelBranchCompleted marks it + // terminal. Guards against branches spinning Running forever. + let mut state = initialized_projection(); + let branch = StageId::new("review_ux", 1); + + state + .apply_event(&test_stage_event( + 3, + EventBody::ParallelBranchStarted(ParallelBranchStartedProps { index: 0 }), + branch.clone(), + )) + .unwrap(); + assert_eq!(state.stage(&branch).unwrap().state, StageState::Running); + + state + .apply_event(&test_stage_event( + 4, + EventBody::ParallelBranchCompleted(ParallelBranchCompletedProps { + index: 0, + duration_ms: 1234, + status: "succeeded".to_string(), + head_sha: None, + }), + branch.clone(), + )) + .unwrap(); + + let stage = state.stage(&branch).unwrap(); + assert_eq!(stage.state, StageState::Succeeded); + assert_eq!(stage.timing.unwrap().wall_time_ms, 1234); + assert_eq!( + stage.completion.as_ref().unwrap().outcome, + StageOutcome::Succeeded + ); + } + fn start_stage(state: &mut RunProjection, stage_id: &StageId) { state .apply_event(&test_stage_event(