diff --git a/lib/crates/fabro-workflow/src/pipeline/finalize.rs b/lib/crates/fabro-workflow/src/pipeline/finalize.rs
index dbdee506b..45a4f082a 100644
--- a/lib/crates/fabro-workflow/src/pipeline/finalize.rs
+++ b/lib/crates/fabro-workflow/src/pipeline/finalize.rs
@@ -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);
 
diff --git a/lib/crates/fabro-workflow/src/pipeline/mod.rs b/lib/crates/fabro-workflow/src/pipeline/mod.rs
index e6e3daf1c..571cc309b 100644
--- a/lib/crates/fabro-workflow/src/pipeline/mod.rs
+++ b/lib/crates/fabro-workflow/src/pipeline/mod.rs
@@ -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;
diff --git a/lib/crates/fabro-workflow/src/test_support.rs b/lib/crates/fabro-workflow/src/test_support.rs
index b404c1c44..bd6c51203 100644
--- a/lib/crates/fabro-workflow/src/test_support.rs
+++ b/lib/crates/fabro-workflow/src/test_support.rs
@@ -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