refactor(workflow): consolidate list_events walk and dedupe test helpers

FINALIZE loaded the run event log twice: once via build_conclusion_from_store
for stage durations, then again to count ArtifactCaptured events. Merged into
a single walk feeding both the conclusion and the artifact count.

Collapsed six near-identical pipeline::execute + emit_terminal + flush blocks
in test_support into one execute_and_emit_terminal helper. Also trimmed
narrative comments that described caller ordering, control flow, or the fix
commit rather than non-obvious invariants.

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

lib/crates/fabro-cli/tests/it/cmd/runner.rs

@@ -917,7 +917,7 @@ retros = true
         .expect("retro.completed should be present");
     assert!(
         retro_completed_index < run_completed_index,
-        "retro.completed must precede run.completed: run.completed is the terminal event and retro runs before FINALIZE"
+        "retro.completed must precede run.completed"
     );
 }
 

lib/crates/fabro-workflow/src/pipeline/execute/tests.rs

@@ -788,8 +788,6 @@ async fn execute_cancelled_mid_run_persists_cancelled_status() {
 
     let executed = execute_test_run_with_options(run_options, g, Some(Arc::new(registry))).await;
 
-    // The status update is driven by the terminal event, which FINALIZE emits.
-    // This test only runs EXECUTE, so we assert on the cancellation outcome.
     assert!(matches!(executed.outcome, Err(Error::Cancelled)));
 }
 

lib/crates/fabro-workflow/src/pipeline/finalize.rs

@@ -101,12 +101,9 @@ fn build_conclusion_from_parts(
     run_duration_ms: u64,
     final_git_commit_sha: Option<String>,
 ) -> Conclusion {
-    // Dedupe by node id: looping workflows push a duplicate into
+    // Looping workflows revisit nodes; `completed_nodes` accumulates duplicates
-    // `completed_nodes` on every revisit, but `node_outcomes`,
+    // while the other checkpoint maps are keyed by node_id. Dedupe to one row
-    // `node_retries`, and `stage_durations` are all keyed by node_id with
+    // per node so the stages table matches the deduped billing total.
-    // 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();
@@ -152,11 +149,8 @@ fn build_conclusion_from_parts(
     }
 }
 
-/// Write a finalize projection snapshot commit to the metadata branch.
+/// `conclusion` is injected because the terminal event hasn't been emitted
-///
+/// yet — the run store's `projection.conclusion` is still `None` at this point.
-/// `conclusion` is injected into the projection copy: the terminal event
-/// hasn't been emitted yet (FINALIZE emits it after this commit lands), so
-/// the run store's `projection.conclusion` is still `None`.
 pub async fn write_finalize_commit(
     run_options: &RunOptions,
     run_store: &RunStoreHandle,
@@ -198,12 +192,8 @@ pub async fn write_finalize_commit(
     .await;
 }
 
-/// Compute the diff between the run's base sha and the workspace head.
+/// Failed and cancelled runs use a shorter diff timeout so a corrupted
-///
+/// workspace can't stall downstream consumers waiting on the terminal event.
-/// 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,
@@ -230,13 +220,8 @@ async fn compute_final_patch(
     }
 }
 
-/// Billing aggregate for the run. Iterates `node_outcomes.values()` to give
+/// Iterates `node_outcomes.values()` rather than `completed_nodes` to avoid
-/// one entry per unique node — `completed_nodes` contains duplicates for
+/// over-counting the last visit's usage on looping workflows.
-/// 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
@@ -285,9 +270,6 @@ pub(crate) fn build_terminal_event(
         };
     }
 
-    // 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(
@@ -338,11 +320,8 @@ async fn cleanup_sandbox(
 /// FINALIZE phase: build conclusion, write the meta branch, emit the terminal
 /// `WorkflowRunCompleted`/`WorkflowRunFailed` event.
 ///
-/// The terminal event MUST be emitted from here, not from the executor's
+/// The terminal event is emitted here (not from `on_run_end`) so observers
-/// `on_run_end` lifecycle hook. Observers (CLI attach, daemon SSE) treat the
+/// can't act on "done" before the meta branch writes are flushed.
-/// event as the run's "done" signal — emitting it earlier means they can
-/// observe terminal state and act on it (e.g. delete the meta branch in
-/// recovery tests) before this function's writes are flushed.
 ///
 /// # Errors
 ///
@@ -361,27 +340,32 @@ pub async fn finalize(retroed: Retroed, options: &FinalizeOptions) -> Result<Con
     } = retroed;
 
     let (final_status, failure_reason, _run_status) = classify_engine_result(&outcome);
-    let conclusion = build_conclusion_from_store(
+
-        &options.run_store,
+    let events = options.run_store.list_events().await.unwrap_or_default();
+    let stage_durations = crate::extract_stage_durations_from_events(&events);
+    let artifact_count = events
+        .iter()
+        .filter(|envelope| matches!(envelope.event.body, EventBody::ArtifactCaptured(_)))
+        .count();
+    let checkpoint = options
+        .run_store
+        .state()
+        .await
+        .ok()
+        .and_then(|state| state.checkpoint);
+    let conclusion = build_conclusion_from_parts(
+        checkpoint.as_ref(),
+        &stage_durations,
         final_status.clone(),
         failure_reason,
         duration_ms,
         options.last_git_sha.clone(),
-    )
+    );
-    .await;
 
     let final_patch = compute_final_patch(&run_options, &*sandbox, final_status, &emitter).await;
 
     write_finalize_commit(&run_options, &options.run_store, &conclusion).await;
 
-    let artifact_count = options
-        .run_store
-        .list_events()
-        .await
-        .unwrap_or_default()
-        .iter()
-        .filter(|envelope| matches!(envelope.event.body, EventBody::ArtifactCaptured(_)))
-        .count();
     let terminal_event = build_terminal_event(
         &outcome,
         duration_ms,

lib/crates/fabro-workflow/src/test_support.rs

@@ -20,16 +20,17 @@ 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
+/// These helpers stop at EXECUTE, so they emit the terminal event here to
-/// EXECUTE, so they emit it here to keep test consumers seeing the same
+/// keep test consumers seeing the same end-of-run signal as production
-/// end-of-run signal — including billing derived from recorded node usage.
+/// (FINALIZE).
 ///
-/// Flushes `store_logger` first: `StoreProgressLogger` forwards events
+/// The first flush is needed because `StoreProgressLogger` forwards events
-/// through an mpsc channel, so the projection returned by `run_store.state()`
+/// through an mpsc channel — without it, billing would read from a stale
-/// can still be missing `StageCompleted` entries when EXECUTE's `await`
+/// checkpoint. The second flush ensures the just-emitted terminal event is
-/// returns. Without this, billing reads from a stale checkpoint.
+/// persisted before tests reopen the run store.
-async fn emit_test_terminal_event(executed: &Executed, store_logger: &StoreProgressLogger) {
+async fn execute_and_emit_terminal(initialized: InitializedState) -> Executed {
-    store_logger.flush().await;
+    let executed = Box::pin(pipeline::execute(initialized.initialized)).await;
+    initialized.store_logger.flush().await;
     let state = executed.run_store.state().await.ok();
     let billing = state
         .as_ref()
@@ -44,6 +45,8 @@ async fn emit_test_terminal_event(executed: &Executed, store_logger: &StoreProgr
         billing,
     );
     executed.emitter.emit(&event);
+    initialized.store_logger.flush().await;
+    executed
 }
 
 pub fn test_store_dir(run_dir: &std::path::Path) -> PathBuf {
@@ -191,11 +194,7 @@ pub async fn run_graph(
         },
     )
     .await;
-    let executed = pipeline::execute(initialized.initialized).await;
+    let executed = execute_and_emit_terminal(initialized).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;
     executed.outcome
 }
 
@@ -219,9 +218,7 @@ pub async fn run_graph_with_state(
         },
     )
     .await;
-    let executed = pipeline::execute(initialized.initialized).await;
+    let executed = execute_and_emit_terminal(initialized).await;
-    emit_test_terminal_event(&executed, &initialized.store_logger).await;
-    initialized.store_logger.flush().await;
     let outcome = executed.outcome?;
     let state = executed
         .run_store
@@ -253,9 +250,7 @@ pub async fn run_graph_with_hooks(
         },
     )
     .await;
-    let executed = pipeline::execute(initialized.initialized).await;
+    let executed = execute_and_emit_terminal(initialized).await;
-    emit_test_terminal_event(&executed, &initialized.store_logger).await;
-    initialized.store_logger.flush().await;
     executed.outcome
 }
 
@@ -281,9 +276,7 @@ pub async fn run_graph_with_hooks_and_state(
         },
     )
     .await;
-    let executed = pipeline::execute(initialized.initialized).await;
+    let executed = execute_and_emit_terminal(initialized).await;
-    emit_test_terminal_event(&executed, &initialized.store_logger).await;
-    initialized.store_logger.flush().await;
     let outcome = executed.outcome?;
     let state = executed
         .run_store
@@ -314,9 +307,7 @@ pub async fn run_graph_from_checkpoint(
         },
     )
     .await;
-    let executed = pipeline::execute(initialized.initialized).await;
+    let executed = execute_and_emit_terminal(initialized).await;
-    emit_test_terminal_event(&executed, &initialized.store_logger).await;
-    initialized.store_logger.flush().await;
     executed.outcome
 }
 
@@ -341,9 +332,7 @@ pub async fn run_graph_from_checkpoint_with_state(
         },
     )
     .await;
-    let executed = pipeline::execute(initialized.initialized).await;
+    let executed = execute_and_emit_terminal(initialized).await;
-    emit_test_terminal_event(&executed, &initialized.store_logger).await;
-    initialized.store_logger.flush().await;
     let outcome = executed.outcome?;
     let state = executed
         .run_store