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): emit terminal run event from FINALIZE, not on_run_end

Browse source 
The `WorkflowRunCompleted` / `WorkflowRunFailed` event was emitted from
`EventLifecycle::on_run_end`, a callback the executor fires at the end of
the EXECUTE phase. But the run isn't done at that point — RETRO and
FINALIZE still need to run, and FINALIZE writes the meta branch's finalize
commit. Observers that treat the event as "done" (CLI attach, daemon SSE
consumers) could observe terminal state and act on it before the worker
flushed its remaining writes.

The recovery scenario test exposed this: it deletes the meta branch
right after `fabro run` returns, then asserts the branch is empty. On
loaded CI runners the worker's finalize commit landed after the delete,
recreating the branch and failing the assertion.

Move the terminal event emission to `pipeline::finalize::finalize`, after
`write_finalize_commit`. The lifecycle's `on_run_end` overrides for event
and git become empty (deleted — the trait already provides a no-op
default). Three pieces of cross-cutting state (`final_patch`,
`captured_artifact_count`, the dead `EventLifecycle` reads of
`last_git_sha`) only existed to ferry data from EXECUTE to the terminal
event; deleted those too. The aggregator collapses to a one-line
delegate to `hook.on_run_end`.

`write_finalize_commit` now takes the conclusion as a parameter and
injects it into the projection copy, since the terminal event hasn't run
through the run store yet when the meta branch is written.

`build_terminal_event` is `pub(crate)` so `test_support` helpers (which
stop at EXECUTE) can mirror the production payload.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Bryan Helmkamp 2026-04-23 17:48:56 -04:00
parent 3f807f42a6
commit 41c47dbe12
No known key found for this signature in database
9 changed files with 227 additions and 195 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -916,8 +916,8 @@ retros = true
.position(|event| matches!(&event.body, EventBody::RetroCompleted(_)))
.expect("retro.completed should be present");
assert!(
run_completed_index < retro_completed_index,
"retro should still run after run.completed"
retro_completed_index < run_completed_index,
"retro.completed must precede run.completed: run.completed is the terminal event and retro runs before FINALIZE"
);
}

24
lib/crates/fabro-workflow/src/lifecycle/artifact.rs
View file

@ -1,5 +1,4 @@
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::Duration;
use async_trait::async_trait;
@ -34,22 +33,17 @@ const ARTIFACT_UPLOAD_RETRY_DELAYS: [Duration; 3] = [
/// Sub-lifecycle responsible for artifact collection, offloading, and syncing.
pub(crate) struct ArtifactLifecycle {
pub sandbox: Arc<dyn fabro_sandbox::Sandbox>,
pub run_store: RunStoreHandle,
pub emitter: Arc<Emitter>,
pub run_id: RunId,
pub artifact_globs: Vec<String>,
pub artifact_sink: Option<ArtifactSink>,
pub captured_artifact_count: Arc<AtomicUsize>,
pub sandbox: Arc<dyn fabro_sandbox::Sandbox>,
pub run_store: RunStoreHandle,
pub emitter: Arc<Emitter>,
pub run_id: RunId,
pub artifact_globs: Vec<String>,
pub artifact_sink: Option<ArtifactSink>,
/// Per-attempt state: epoch seconds when the attempt started.
attempt_start_epoch: std::sync::Mutex<Option<f64>>,
attempt_start_epoch: std::sync::Mutex<Option<f64>>,
}
impl ArtifactLifecycle {
#[allow(
clippy::too_many_arguments,
reason = "Artifact capture setup needs the run-scoped collaborators up front."
)]
pub(crate) fn new(
sandbox: Arc<dyn fabro_sandbox::Sandbox>,
run_store: RunStoreHandle,
@ -57,7 +51,6 @@ impl ArtifactLifecycle {
run_id: RunId,
artifact_globs: Vec<String>,
artifact_sink: Option<ArtifactSink>,
captured_artifact_count: Arc<AtomicUsize>,
) -> Self {
Self {
sandbox,
@ -66,7 +59,6 @@ impl ArtifactLifecycle {
run_id,
artifact_globs,
artifact_sink,
captured_artifact_count,
attempt_start_epoch: std::sync::Mutex::new(None),
}
}
@ -75,7 +67,6 @@ impl ArtifactLifecycle {
#[async_trait]
impl RunLifecycle<WorkflowGraph> for ArtifactLifecycle {
async fn on_run_start(&self, _graph: &WorkflowGraph, _state: &WfRunState) -> CoreResult<()> {
self.captured_artifact_count.store(0, Ordering::Relaxed);
*self.attempt_start_epoch.lock().unwrap() = None;
Ok(())
}
@ -139,7 +130,6 @@ impl RunLifecycle<WorkflowGraph> for ArtifactLifecycle {
}
let scope = stage_scope_for(state, node_id);
for asset in &summary.captured_assets {
self.captured_artifact_count.fetch_add(1, Ordering::Relaxed);
self.emitter.emit_scoped(
&Event::ArtifactCaptured {
node_id: node_id.to_string(),

105
lib/crates/fabro-workflow/src/lifecycle/event.rs
View file

@ -1,5 +1,4 @@
use std::collections::BTreeMap;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Instant;
@ -11,12 +10,11 @@ use fabro_core::lifecycle::{
};
use fabro_core::outcome::NodeResult;
use fabro_core::state::ExecutionState;
use fabro_types::{BilledTokenCounts, FailureReason, RunId, SuccessReason};
use fabro_types::RunId;
use super::circuit_breaker::CircuitBreakerLifecycle;
use super::git::GitCheckpointResult;
use crate::context::WorkflowContext;
use crate::error::Error;
use crate::event::{Emitter, Event, StageScope};
use crate::graph::{WorkflowGraph, WorkflowNode};
use crate::outcome::{BilledModelUsage, FailureCategory, FailureDetail, Outcome, StageStatus};
@ -31,25 +29,23 @@ type FailureSignatureSnapshot = (
/// Sub-lifecycle responsible for emitting workflow run events.
pub(crate) struct EventLifecycle {
pub emitter: Arc<Emitter>,
pub graph_name: String,
pub run_id: RunId,
pub run_start: Mutex<Instant>,
pub emitter: Arc<Emitter>,
pub graph_name: String,
pub run_id: RunId,
pub run_start: Mutex<Instant>,
/// Set in on_edge_selected when loop_restart approved; emitted+cleared in
/// on_run_start.
pub restarted_from: Arc<Mutex<Option<(String, String)>>>,
pub restarted_from: Arc<Mutex<Option<(String, String)>>>,
// Config for WorkflowRunStarted payload
pub base_branch: Option<String>,
pub base_sha: Option<String>,
pub run_branch: Option<String>,
pub worktree_dir: Option<String>,
pub goal: Option<String>,
pub captured_artifact_count: Arc<AtomicUsize>,
// Cross-lifecycle data
pub checkpoint_git_result: Arc<Mutex<Option<GitCheckpointResult>>>,
pub last_git_sha: Arc<Mutex<Option<String>>>,
pub final_patch: Arc<Mutex<Option<String>>>,
pub circuit_breaker: Arc<CircuitBreakerLifecycle>,
pub base_branch: Option<String>,
pub base_sha: Option<String>,
pub run_branch: Option<String>,
pub worktree_dir: Option<String>,
pub goal: Option<String>,
/// Shared git checkpoint result (written by GitLifecycle, read by
/// EventLifecycle when emitting CheckpointCompleted).
pub checkpoint_git_result: Arc<Mutex<Option<GitCheckpointResult>>>,
pub circuit_breaker: Arc<CircuitBreakerLifecycle>,
}
fn snapshot_failure_signatures(
@ -415,75 +411,4 @@ impl RunLifecycle<WorkflowGraph> for EventLifecycle {
Ok(())
}
async fn on_run_end(&self, outcome: &Outcome, state: &WfRunState) {
let duration_ms = crate::millis_u64(self.run_start.lock().unwrap().elapsed());
let artifact_count = self.captured_artifact_count.load(Ordering::Relaxed);
let last_sha = self.last_git_sha.lock().unwrap().clone();
let final_patch = self.final_patch.lock().unwrap().clone();
let run_billing_entries = state
.node_outcomes
.values()
.filter_map(|o| o.usage.clone())
.collect::<Vec<_>>();
let run_billing = (!run_billing_entries.is_empty())
.then(|| BilledTokenCounts::from_billed_usage(&run_billing_entries));
let total_usd_micros = run_billing
.as_ref()
.and_then(|billing| billing.total_usd_micros)
.or_else(|| {
let mut total = 0_i64;
let mut has_total = false;
for usage in state
.node_outcomes
.values()
.filter_map(|o| o.usage.as_ref())
{
if let Some(value) = usage.total_usd_micros {
total += value;
has_total = true;
}
}
has_total.then_some(total)
});
if state.cancelled {
self.emitter.emit(&Event::WorkflowRunFailed {
error: Error::Cancelled,
duration_ms,
reason: FailureReason::Cancelled,
git_commit_sha: last_sha,
final_patch: final_patch.clone(),
});
return;
}
if outcome.status == StageStatus::Success || outcome.status == StageStatus::PartialSuccess {
self.emitter.emit(&Event::WorkflowRunCompleted {
duration_ms,
artifact_count,
status: outcome.status.to_string(),
reason: match outcome.status {
StageStatus::PartialSuccess => SuccessReason::PartialSuccess,
_ => SuccessReason::Completed,
},
total_usd_micros,
final_git_commit_sha: last_sha,
final_patch,
billing: run_billing,
});
} else {
let error_msg = outcome
.failure
.as_ref()
.map_or_else(|| "run failed".to_string(), |f| f.message.clone());
self.emitter.emit(&Event::WorkflowRunFailed {
error: Error::engine(error_msg),
duration_ms,
reason: FailureReason::WorkflowError,
git_commit_sha: last_sha,
final_patch,
});
}
}
}

45
lib/crates/fabro-workflow/src/lifecycle/git.rs
View file

@ -14,11 +14,11 @@ use crate::event::{Emitter, Event, RunNoticeLevel};
use crate::git::MetadataStore;
use crate::graph::{WorkflowGraph, WorkflowNode};
use crate::lifecycle::event::stage_scope_for;
use crate::outcome::{BilledModelUsage, Outcome, StageStatus};
use crate::outcome::BilledModelUsage;
use crate::run_dump::RunDump;
use crate::run_options::RunOptions;
use crate::runtime_store::RunStoreHandle;
use crate::sandbox_git::{git_checkpoint, git_diff, git_diff_with_timeout, git_push_host};
use crate::sandbox_git::{git_checkpoint, git_diff, git_push_host};
type WfRunState = ExecutionState<Option<BilledModelUsage>>;
type WfNodeResult = NodeResult<Option<BilledModelUsage>>;
@ -71,7 +71,6 @@ pub(crate) struct GitLifecycle {
// Cross-lifecycle data (shared with EventLifecycle)
pub checkpoint_git_result: Arc<Mutex<Option<GitCheckpointResult>>>,
pub last_git_sha: Arc<Mutex<Option<String>>>,
pub final_patch: Arc<Mutex<Option<String>>>,
}
#[async_trait]
@ -80,7 +79,6 @@ impl RunLifecycle<WorkflowGraph> for GitLifecycle {
// Reset last_git_sha (diff base parity)
*self.last_git_sha.lock().unwrap() = None;
*self.checkpoint_git_result.lock().unwrap() = None;
*self.final_patch.lock().unwrap() = None;
// Init metadata branch (best-effort)
if let (Some(_), Some(repo_path)) = (
@ -321,43 +319,4 @@ impl RunLifecycle<WorkflowGraph> for GitLifecycle {
Ok(())
}
async fn on_run_end(&self, outcome: &Outcome, _state: &WfRunState) {
// Capture the final diff for event/store projection.
//
// Success/PartialSuccess uses the standard 30 s timeout. Failed runs
// use a shorter 10 s timeout: a pathological workspace (FS locks,
// corrupted index) must not stall terminal event emission downstream
// (Slack notifier, SSE RunFailed, CI hooks).
if self.run_options.git.is_none() {
return;
}
let timeout_ms = match outcome.status {
StageStatus::Success | StageStatus::PartialSuccess => 30_000,
_ => 10_000,
};
if let Some(base_sha) = self
.run_options
.git
.as_ref()
.and_then(|g| g.base_sha.clone())
{
match git_diff_with_timeout(&*self.sandbox, &base_sha, timeout_ms).await {
Ok(patch) if !patch.is_empty() => {
*self.final_patch.lock().unwrap() = Some(patch.clone());
}
Ok(_) => {
*self.final_patch.lock().unwrap() = None;
}
Err(err) => {
*self.final_patch.lock().unwrap() = None;
self.emitter.emit(&Event::RunNotice {
level: RunNoticeLevel::Warn,
code: "git_diff_failed".to_string(),
message: format!("final diff failed: {err}"),
});
}
}
}
}
}

41
lib/crates/fabro-workflow/src/lifecycle/mod.rs
View file

@ -8,7 +8,7 @@ pub(crate) mod hook;
use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Instant;
@ -96,8 +96,6 @@ impl WorkflowLifecycle {
let checkpoint_git_result: Arc<Mutex<Option<GitCheckpointResult>>> =
Arc::new(Mutex::new(None));
let last_git_sha: Arc<Mutex<Option<String>>> = Arc::new(Mutex::new(None));
let final_patch: Arc<Mutex<Option<String>>> = Arc::new(Mutex::new(None));
let captured_artifact_count = Arc::new(AtomicUsize::new(0));
let circuit_breaker = Arc::new(CircuitBreakerLifecycle::new(loop_restart_signature_limit));
@ -114,21 +112,18 @@ impl WorkflowLifecycle {
};
let event = EventLifecycle {
emitter: Arc::clone(emitter),
graph_name: graph.name.clone(),
run_id: run_options.run_id,
run_start: Mutex::new(Instant::now()),
restarted_from: Arc::clone(&restarted_from),
base_branch: run_options.base_branch.clone(),
base_sha: run_options.git.as_ref().and_then(|g| g.base_sha.clone()),
run_branch: run_options.git.as_ref().and_then(|g| g.run_branch.clone()),
worktree_dir: working_directory.clone(),
goal: (!graph.goal().is_empty()).then(|| graph.goal().to_string()),
captured_artifact_count: Arc::clone(&captured_artifact_count),
last_git_sha: Arc::clone(&last_git_sha),
final_patch: Arc::clone(&final_patch),
checkpoint_git_result: Arc::clone(&checkpoint_git_result),
circuit_breaker: Arc::clone(&circuit_breaker),
emitter: Arc::clone(emitter),
graph_name: graph.name.clone(),
run_id: run_options.run_id,
run_start: Mutex::new(Instant::now()),
restarted_from: Arc::clone(&restarted_from),
base_branch: run_options.base_branch.clone(),
base_sha: run_options.git.as_ref().and_then(|g| g.base_sha.clone()),
run_branch: run_options.git.as_ref().and_then(|g| g.run_branch.clone()),
worktree_dir: working_directory.clone(),
goal: (!graph.goal().is_empty()).then(|| graph.goal().to_string()),
checkpoint_git_result: Arc::clone(&checkpoint_git_result),
circuit_breaker: Arc::clone(&circuit_breaker),
};
let hook = HookLifecycle {
@ -156,8 +151,7 @@ impl WorkflowLifecycle {
run_options: Arc::clone(run_options),
start_node_id,
checkpoint_git_result: Arc::clone(&checkpoint_git_result),
last_git_sha: Arc::clone(&last_git_sha),
final_patch,
last_git_sha,
};
let artifact = ArtifactLifecycle::new(
@ -167,7 +161,6 @@ impl WorkflowLifecycle {
run_options.run_id,
run_options.artifact_globs(),
artifact_sink,
captured_artifact_count,
);
Self {
@ -419,12 +412,6 @@ impl RunLifecycle<WorkflowGraph> for WorkflowLifecycle {
}
async fn on_run_end(&self, outcome: &Outcome, state: &WfRunState) {
if state.cancelled {
self.event.on_run_end(outcome, state).await;
return;
}
self.git.on_run_end(outcome, state).await;
self.event.on_run_end(outcome, state).await;
self.hook.on_run_end(outcome, state).await;
}
}

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

@ -32,7 +32,6 @@ use crate::pipeline::initialize;
use crate::pipeline::types::{InitOptions, LlmSpec, Persisted, SandboxEnvSpec};
use crate::records::RunSpec;
use crate::run_options::{GitCheckpointOptions, LifecycleOptions, RunOptions};
use crate::run_status::{FailureReason, RunStatus};
use crate::test_support::run_graph;
fn local_env() -> Arc<dyn Sandbox> {
@ -789,11 +788,9 @@ 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)));
let status = executed.run_store.state().await.unwrap().status.unwrap();
assert_eq!(status, RunStatus::Failed {
reason: FailureReason::Cancelled,
});
}
#[tokio::test]

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

@ -1,7 +1,7 @@
use std::sync::Arc;
use fabro_hooks::{HookContext, HookEvent, HookRunner};
use fabro_types::BilledTokenCounts;
use fabro_types::{BilledTokenCounts, EventBody};
use super::types::{Concluded, FinalizeOptions, Retroed};
use crate::error::Error;
@ -13,7 +13,7 @@ use crate::run_dump::RunDump;
use crate::run_options::RunOptions;
use crate::run_status::{FailureReason, RunStatus, SuccessReason};
use crate::runtime_store::RunStoreHandle;
use crate::sandbox_git::git_push_host;
use crate::sandbox_git::{git_diff_with_timeout, git_push_host};
fn emit_run_notice(
emitter: &Emitter,
@ -153,9 +153,14 @@ fn build_conclusion_from_parts(
/// Write a finalize projection snapshot commit to the metadata branch.
///
/// This captures the final `run.json` projection state, including conclusion
/// and retro data. Best-effort: errors are logged as warnings.
pub async fn write_finalize_commit(run_options: &RunOptions, run_store: &RunStoreHandle) {
/// `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,
conclusion: &Conclusion,
) {
let (Some(meta_branch), Some(repo_path)) = (
run_options
.git
@ -168,9 +173,12 @@ pub async fn write_finalize_commit(run_options: &RunOptions, run_store: &RunStor
let git_author = run_options.git_author();
let store = MetadataStore::new(repo_path, &git_author);
let Ok(store_state) = run_store.state().await else {
let Ok(mut store_state) = run_store.state().await else {
return;
};
if store_state.conclusion.is_none() {
store_state.conclusion = Some(conclusion.clone());
}
let dump = RunDump::from_projection(&store_state);
if let Err(e) =
dump.write_to_metadata_store(&store, &run_options.run_id.to_string(), "finalize run")
@ -189,6 +197,123 @@ pub async fn write_finalize_commit(run_options: &RunOptions, run_store: &RunStor
.await;
}
/// 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.
async fn compute_final_patch(
run_options: &RunOptions,
sandbox: &dyn fabro_agent::Sandbox,
status: StageStatus,
emitter: &Emitter,
) -> Option<String> {
let base_sha = run_options.git.as_ref().and_then(|g| g.base_sha.clone())?;
let timeout_ms = match status {
StageStatus::Success | StageStatus::PartialSuccess => 30_000,
_ => 10_000,
};
match git_diff_with_timeout(sandbox, &base_sha, timeout_ms).await {
Ok(patch) if !patch.is_empty() => Some(patch),
Ok(_) => None,
Err(err) => {
emit_run_notice(
emitter,
RunNoticeLevel::Warn,
"git_diff_failed",
format!("final diff failed: {err}"),
);
None
}
}
}
/// Build the terminal `WorkflowRunCompleted`/`WorkflowRunFailed` event.
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>,
) -> 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 {
return Event::WorkflowRunFailed {
error: Error::Cancelled,
duration_ms,
reason: FailureReason::Cancelled,
git_commit_sha: final_git_commit_sha,
final_patch,
};
}
if outcome_status == StageStatus::Success || outcome_status == StageStatus::PartialSuccess {
Event::WorkflowRunCompleted {
duration_ms,
artifact_count,
status: outcome_status.to_string(),
reason: match outcome_status {
StageStatus::PartialSuccess => SuccessReason::PartialSuccess,
_ => SuccessReason::Completed,
},
total_usd_micros,
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,
}
}
}
async fn run_hooks(
hook_runner: Option<&HookRunner>,
hook_context: &HookContext,
@ -219,7 +344,14 @@ async fn cleanup_sandbox(
Ok(())
}
/// FINALIZE phase: classify outcome, build conclusion, persist terminal state.
/// 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
/// `on_run_end` lifecycle hook. Observers (CLI attach, daemon SSE) treat the
/// 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
///
@ -240,14 +372,33 @@ pub async fn finalize(retroed: Retroed, options: &FinalizeOptions) -> Result<Con
let (final_status, failure_reason, _run_status) = classify_engine_result(&outcome);
let conclusion = build_conclusion_from_store(
&options.run_store,
final_status,
final_status.clone(),
failure_reason,
duration_ms,
options.last_git_sha.clone(),
)
.await;
write_finalize_commit(&run_options, &options.run_store).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 events = options.run_store.list_events().await.unwrap_or_default();
let artifact_count = events
.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(),
);
emitter.emit(&terminal_event);
if options.preserve_sandbox {
let info = sandbox.sandbox_info();

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

@ -11,7 +11,7 @@ mod validate;
pub use execute::execute;
pub use fabro_types::PullRequestRecord;
pub(crate) use finalize::build_conclusion_from_store;
pub(crate) use finalize::{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;

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

@ -15,10 +15,27 @@ use crate::event::{Emitter, Event, StoreProgressLogger, append_event};
use crate::handler::HandlerRegistry;
use crate::outcome::Outcome;
use crate::pipeline;
use crate::pipeline::types::Initialized;
use crate::pipeline::build_terminal_event;
use crate::pipeline::types::{Executed, Initialized};
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) {
let state = executed.run_store.state().await.ok();
let event = build_terminal_event(
&executed.outcome,
executed.duration_ms,
0,
None,
None,
state.as_ref(),
);
executed.emitter.emit(&event);
}
pub fn test_store_dir(run_dir: &std::path::Path) -> PathBuf {
let mut hasher = std::collections::hash_map::DefaultHasher::new();
std::process::id().hash(&mut hasher);
@ -165,6 +182,7 @@ pub async fn run_graph(
)
.await;
let executed = pipeline::execute(initialized.initialized).await;
emit_test_terminal_event(&executed).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;
@ -192,6 +210,7 @@ pub async fn run_graph_with_state(
)
.await;
let executed = pipeline::execute(initialized.initialized).await;
emit_test_terminal_event(&executed).await;
initialized.store_logger.flush().await;
let outcome = executed.outcome?;
let state = executed
@ -225,6 +244,7 @@ pub async fn run_graph_with_hooks(
)
.await;
let executed = pipeline::execute(initialized.initialized).await;
emit_test_terminal_event(&executed).await;
initialized.store_logger.flush().await;
executed.outcome
}
@ -252,6 +272,7 @@ pub async fn run_graph_with_hooks_and_state(
)
.await;
let executed = pipeline::execute(initialized.initialized).await;
emit_test_terminal_event(&executed).await;
initialized.store_logger.flush().await;
let outcome = executed.outcome?;
let state = executed
@ -284,6 +305,7 @@ pub async fn run_graph_from_checkpoint(
)
.await;
let executed = pipeline::execute(initialized.initialized).await;
emit_test_terminal_event(&executed).await;
initialized.store_logger.flush().await;
executed.outcome
}
@ -310,6 +332,7 @@ pub async fn run_graph_from_checkpoint_with_state(
)
.await;
let executed = pipeline::execute(initialized.initialized).await;
emit_test_terminal_event(&executed).await;
initialized.store_logger.flush().await;
let outcome = executed.outcome?;
let state = executed