mirror of
https://github.com/fabro-sh/fabro.git
synced 2026-10-09 03:20:56 +00:00
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) <noreply@anthropic.com>
This commit is contained in:
parent
daa6f4cdbf
commit
c623bf7432
1 changed files with 75 additions and 2 deletions
|
|
@ -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(
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue