scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-08-28 05:27:41 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

fix(workflow): dedupe stages/billing and surface real errors in terminal event

Browse source 
Follow-ups to the FINALIZE terminal-event refactor, surfaced during
review:

- build_terminal_event: drop re-wrapping Err outcomes in Error::engine,
  which doubled the "Engine error: " prefix on display. Surface the
  original error directly.
- Unify loop billing: move billing aggregation into a shared
  billing_from_checkpoint helper iterating node_outcomes.values() once
  per unique node. Both Conclusion.billing and the emitted terminal
  event use it, so the persisted metadata snapshot and the run.completed
  event can't disagree.
- Dedupe conclusion.stages by node id while preserving execution order.
  completed_nodes has duplicates for looping workflows, but
  node_outcomes, node_retries, and stage_durations are all keyed by
  node_id with overwrite semantics, so duplicate StageSummary rows
  carried identical latest-visit values and inflated total_retries /
  the PR Fabro Details table.
- test_support: flush StoreProgressLogger before reading state.
  StoreProgressLogger forwards events via mpsc, so state() right after
  execute could miss StageCompleted entries and return stale billing.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Bryan Helmkamp 2026-04-23 18:44:41 -04:00
parent 4585f9874c
commit e8a89ac393
No known key found for this signature in database
3 changed files with 91 additions and 87 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

144
lib/crates/fabro-workflow/src/pipeline/finalize.rs
View file

@ -101,12 +101,21 @@ fn build_conclusion_from_parts(
run_duration_ms: u64,
final_git_commit_sha: Option<String>,
) -> Conclusion {
let (stages, billing, total_retries) = if let Some(cp) = checkpoint {
// Dedupe by node id: looping workflows push a duplicate into
// `completed_nodes` on every revisit, but `node_outcomes`,
// `node_retries`, and `stage_durations` are all keyed by node_id with
// overwrite semantics, so duplicate rows would carry identical
// (latest-visit) values. One row per unique node keeps the table
// consistent with the deduped `billing` total below.
let (stages, total_retries) = if let Some(cp) = checkpoint {
let mut stages = Vec::new();
let mut seen = std::collections::HashSet::new();
let mut retries_sum: u32 = 0;
let mut billed_usage = Vec::new();
for node_id in &cp.completed_nodes {
if !seen.insert(node_id.as_str()) {
continue;
}
let outcome = cp.node_outcomes.get(node_id);
let retries = cp
.node_retries
@ -116,10 +125,6 @@ fn build_conclusion_from_parts(
.saturating_sub(1);
retries_sum += retries;
if let Some(usage) = outcome.and_then(|o| o.usage.as_ref()) {
billed_usage.push(usage.clone());
}
stages.push(StageSummary {
stage_id: node_id.clone(),
stage_label: node_id.clone(),
@ -130,13 +135,9 @@ fn build_conclusion_from_parts(
retries,
});
}
(
stages,
(!billed_usage.is_empty()).then(|| BilledTokenCounts::from_billed_usage(&billed_usage)),
retries_sum,
)
(stages, retries_sum)
} else {
(vec![], None, 0)
(vec![], 0)
};
Conclusion {
@ -146,7 +147,7 @@ fn build_conclusion_from_parts(
failure_reason,
final_git_commit_sha,
stages,
billing,
billing: checkpoint.and_then(billing_from_checkpoint),
total_retries,
}
}
@ -199,9 +200,10 @@ pub async fn write_finalize_commit(
/// Compute the diff between the run's base sha and the workspace head.
///
/// Failed runs use a shorter timeout: a corrupted workspace must not stall
/// the terminal event downstream consumers (Slack, SSE, CI hooks) are waiting
/// for.
/// Failed and cancelled runs use a shorter timeout: a corrupted workspace
/// must not stall the terminal event downstream consumers (Slack, SSE, CI
/// hooks) are waiting for. The diff is still captured for cancelled runs so
/// partial work stays visible after sandbox cleanup.
async fn compute_final_patch(
run_options: &RunOptions,
sandbox: &dyn fabro_agent::Sandbox,
@ -228,47 +230,31 @@ async fn compute_final_patch(
}
}
/// Build the terminal `WorkflowRunCompleted`/`WorkflowRunFailed` event.
/// Billing aggregate for the run. Iterates `node_outcomes.values()` to give
/// one entry per unique node — `completed_nodes` contains duplicates for
/// looping workflows and would over-count the last visit's usage.
///
/// Used by both `build_conclusion_from_parts` (persisted into
/// `Conclusion.billing`) and `build_terminal_event` so the metadata snapshot
/// and the emitted `run.completed`/`run.failed` event can never disagree.
pub(crate) fn billing_from_checkpoint(cp: &Checkpoint) -> Option<BilledTokenCounts> {
let usage: Vec<_> = cp
.node_outcomes
.values()
.filter_map(|o| o.usage.clone())
.collect();
(!usage.is_empty()).then(|| BilledTokenCounts::from_billed_usage(&usage))
}
pub(crate) fn build_terminal_event(
outcome: &Result<Outcome, Error>,
duration_ms: u64,
artifact_count: usize,
final_git_commit_sha: Option<String>,
final_patch: Option<String>,
state: Option<&fabro_store::RunProjection>,
billing: Option<BilledTokenCounts>,
) -> Event {
let cancelled = matches!(outcome, Err(Error::Cancelled));
let outcome_status = outcome
.as_ref()
.map_or(StageStatus::Fail, |o| o.status.clone());
let billed_usage: Vec<_> = state
.and_then(|s| s.checkpoint.as_ref())
.map(|cp| {
cp.node_outcomes
.values()
.filter_map(|o| o.usage.clone())
.collect()
})
.unwrap_or_default();
let billing =
(!billed_usage.is_empty()).then(|| BilledTokenCounts::from_billed_usage(&billed_usage));
let total_usd_micros = billing
.as_ref()
.and_then(|b| b.total_usd_micros)
.or_else(|| {
let mut total = 0_i64;
let mut has_total = false;
for usage in &billed_usage {
if let Some(value) = usage.total_usd_micros {
total += value;
has_total = true;
}
}
has_total.then_some(total)
});
if cancelled {
if matches!(outcome, Err(Error::Cancelled)) {
return Event::WorkflowRunFailed {
error: Error::Cancelled,
duration_ms,
@ -278,8 +264,13 @@ pub(crate) fn build_terminal_event(
};
}
let outcome_status = outcome
.as_ref()
.map_or(StageStatus::Fail, |o| o.status.clone());
if outcome_status == StageStatus::Success || outcome_status == StageStatus::PartialSuccess {
Event::WorkflowRunCompleted {
let total_usd_micros = billing.as_ref().and_then(|b| b.total_usd_micros);
return Event::WorkflowRunCompleted {
duration_ms,
artifact_count,
status: outcome_status.to_string(),
@ -291,26 +282,26 @@ pub(crate) fn build_terminal_event(
final_git_commit_sha,
final_patch,
billing,
}
} else {
let error_msg = outcome
.as_ref()
.err()
.map(ToString::to_string)
.or_else(|| {
outcome
.as_ref()
.ok()
.and_then(|o| o.failure.as_ref().map(|f| f.message.clone()))
})
.unwrap_or_else(|| "run failed".to_string());
Event::WorkflowRunFailed {
error: Error::engine(error_msg),
duration_ms,
reason: FailureReason::WorkflowError,
git_commit_sha: final_git_commit_sha,
final_patch,
}
};
}
// Err(Cancelled) was handled above, so Err here is a real failure: surface
// it directly without re-wrapping in Error::engine, which would double the
// "Engine error: " prefix.
let error = match outcome {
Err(err) => err.clone(),
Ok(o) => Error::engine(
o.failure
.as_ref()
.map_or_else(|| "run failed".to_string(), |f| f.message.clone()),
),
};
Event::WorkflowRunFailed {
error,
duration_ms,
reason: FailureReason::WorkflowError,
git_commit_sha: final_git_commit_sha,
final_patch,
}
}
@ -383,20 +374,21 @@ pub async fn finalize(retroed: Retroed, options: &FinalizeOptions) -> Result<Con
write_finalize_commit(&run_options, &options.run_store, &conclusion).await;
let events = options.run_store.list_events().await.unwrap_or_default();
let artifact_count = events
let artifact_count = options
.run_store
.list_events()
.await
.unwrap_or_default()
.iter()
.filter(|envelope| matches!(envelope.event.body, EventBody::ArtifactCaptured(_)))
.count();
let state_for_event = options.run_store.state().await.ok();
let terminal_event = build_terminal_event(
&outcome,
duration_ms,
artifact_count,
options.last_git_sha.clone(),
final_patch,
state_for_event.as_ref(),
conclusion.billing.clone(),
);
emitter.emit(&terminal_event);

4
lib/crates/fabro-workflow/src/pipeline/mod.rs
View file

@ -11,7 +11,9 @@ mod validate;
pub use execute::execute;
pub use fabro_types::PullRequestRecord;
pub(crate) use finalize::{build_conclusion_from_store, build_terminal_event};
pub(crate) use finalize::{
billing_from_checkpoint, build_conclusion_from_store, build_terminal_event,
};
pub use finalize::{classify_engine_result, finalize, write_finalize_commit};
pub use initialize::initialize;
pub use parse::parse;

30
lib/crates/fabro-workflow/src/test_support.rs
View file

@ -15,23 +15,33 @@ use crate::event::{Emitter, Event, StoreProgressLogger, append_event};
use crate::handler::HandlerRegistry;
use crate::outcome::Outcome;
use crate::pipeline;
use crate::pipeline::build_terminal_event;
use crate::pipeline::types::{Executed, Initialized};
use crate::pipeline::{billing_from_checkpoint, build_terminal_event};
use crate::records::Checkpoint;
use crate::run_options::RunOptions;
/// FINALIZE emits the terminal event in production. These helpers stop at
/// EXECUTE, so they emit it here to keep test consumers seeing the same
/// end-of-run signal.
async fn emit_test_terminal_event(executed: &Executed) {
/// end-of-run signal — including billing derived from recorded node usage.
///
/// Flushes `store_logger` first: `StoreProgressLogger` forwards events
/// through an mpsc channel, so the projection returned by `run_store.state()`
/// can still be missing `StageCompleted` entries when EXECUTE's `await`
/// returns. Without this, billing reads from a stale checkpoint.
async fn emit_test_terminal_event(executed: &Executed, store_logger: &StoreProgressLogger) {
store_logger.flush().await;
let state = executed.run_store.state().await.ok();
let billing = state
.as_ref()
.and_then(|s| s.checkpoint.as_ref())
.and_then(billing_from_checkpoint);
let event = build_terminal_event(
&executed.outcome,
executed.duration_ms,
0,
None,
None,
state.as_ref(),
billing,
);
executed.emitter.emit(&event);
}
@ -182,7 +192,7 @@ pub async fn run_graph(
)
.await;
let executed = pipeline::execute(initialized.initialized).await;
emit_test_terminal_event(&executed).await;
emit_test_terminal_event(&executed, &initialized.store_logger).await;
// Tests often reopen the run store immediately after `run()` returns.
// Flush the async store logger first so they don't observe partial state.
initialized.store_logger.flush().await;
@ -210,7 +220,7 @@ pub async fn run_graph_with_state(
)
.await;
let executed = pipeline::execute(initialized.initialized).await;
emit_test_terminal_event(&executed).await;
emit_test_terminal_event(&executed, &initialized.store_logger).await;
initialized.store_logger.flush().await;
let outcome = executed.outcome?;
let state = executed
@ -244,7 +254,7 @@ pub async fn run_graph_with_hooks(
)
.await;
let executed = pipeline::execute(initialized.initialized).await;
emit_test_terminal_event(&executed).await;
emit_test_terminal_event(&executed, &initialized.store_logger).await;
initialized.store_logger.flush().await;
executed.outcome
}
@ -272,7 +282,7 @@ pub async fn run_graph_with_hooks_and_state(
)
.await;
let executed = pipeline::execute(initialized.initialized).await;
emit_test_terminal_event(&executed).await;
emit_test_terminal_event(&executed, &initialized.store_logger).await;
initialized.store_logger.flush().await;
let outcome = executed.outcome?;
let state = executed
@ -305,7 +315,7 @@ pub async fn run_graph_from_checkpoint(
)
.await;
let executed = pipeline::execute(initialized.initialized).await;
emit_test_terminal_event(&executed).await;
emit_test_terminal_event(&executed, &initialized.store_logger).await;
initialized.store_logger.flush().await;
executed.outcome
}
@ -332,7 +342,7 @@ pub async fn run_graph_from_checkpoint_with_state(
)
.await;
let executed = pipeline::execute(initialized.initialized).await;
emit_test_terminal_event(&executed).await;
emit_test_terminal_event(&executed, &initialized.store_logger).await;
initialized.store_logger.flush().await;
let outcome = executed.outcome?;
let state = executed