diff --git a/docs/internal/events-strategy.md b/docs/internal/events-strategy.md index 8059687c3..0d62bcaf6 100644 --- a/docs/internal/events-strategy.md +++ b/docs/internal/events-strategy.md @@ -186,4 +186,8 @@ Do not rebuild or mutate the `RunEvent` in downstream listeners. Any JSONL sink, the run store, and SSE should reflect the same canonical envelope bytes after redaction. +An active workflow treats any run-event sink write failure as fatal. It cancels execution and +attempts to persist `run.failed` through the direct sink path. Persistence-error logs must include +the full source chain so an HTTP status or transport failure remains visible. + `status.json` remains the authoritative completion signal for detached runs. Terminal run status should only be written after all post-run work is finished. diff --git a/lib/components/fabro-workflow/src/error.rs b/lib/components/fabro-workflow/src/error.rs index 290aacf91..7bedbda70 100644 --- a/lib/components/fabro-workflow/src/error.rs +++ b/lib/components/fabro-workflow/src/error.rs @@ -14,6 +14,7 @@ use fabro_validate::Diagnostic; use regex::Regex; use thiserror::Error as ThisError; +use crate::event::RunEventPersistenceError; use crate::outcome::{FailureDetail, Outcome, StageOutcome}; /// Classify an LLM error into a `FailureCategory` based on its structure. @@ -721,6 +722,12 @@ impl From for Error { } } +impl From for Error { + fn from(err: RunEventPersistenceError) -> Self { + Self::engine_with_source("run event persistence failed", err) + } +} + impl From for Error { fn from(err: fabro_checkpoint::MetadataError) -> Self { match err { diff --git a/lib/components/fabro-workflow/src/event.rs b/lib/components/fabro-workflow/src/event.rs index a5c1f583e..656b69922 100644 --- a/lib/components/fabro-workflow/src/event.rs +++ b/lib/components/fabro-workflow/src/event.rs @@ -18,7 +18,7 @@ pub use self::redaction::{ build_redacted_event_payload, event_payload_from_redacted_json, redacted_event_json, }; pub use self::sink::{ - RunEventLogger, RunEventSink, StoreProgressLogger, append_event, append_event_if, - append_event_to_sink, + RunEventLogger, RunEventPersistenceError, RunEventSink, StoreProgressLogger, append_event, + append_event_if, append_event_to_sink, }; pub use crate::stage_scope::StageScope; diff --git a/lib/components/fabro-workflow/src/event/sink.rs b/lib/components/fabro-workflow/src/event/sink.rs index 9588a3ded..a82f2e20f 100644 --- a/lib/components/fabro-workflow/src/event/sink.rs +++ b/lib/components/fabro-workflow/src/event/sink.rs @@ -5,8 +5,9 @@ use std::sync::Arc; use ::fabro_types::{RunEvent, RunId, RunProjection}; use anyhow::Result; use fabro_store::RunDatabase; +use fabro_util::error::{SharedError, collect_chain}; use tokio::io::{AsyncWrite, AsyncWriteExt}; -use tokio::sync::{Mutex as AsyncMutex, mpsc, oneshot}; +use tokio::sync::{Mutex as AsyncMutex, mpsc, oneshot, watch}; use super::emitter::Emitter; use super::redaction::{build_redacted_event_payload, redacted_event_json}; @@ -42,9 +43,15 @@ pub async fn append_event_to_sink( sink: &RunEventSink, run_id: &RunId, event: &Event, -) -> Result<()> { +) -> Result<(), RunEventPersistenceError> { let stored = to_run_event(run_id, event); - sink.write_run_event(&stored).await + sink.write_run_event(&stored) + .await + .map_err(|err| RunEventPersistenceError::Write { + run_id: *run_id, + event: stored.body.event_name().to_string(), + source: SharedError::new(err), + }) } #[derive(Clone)] @@ -149,86 +156,100 @@ impl RunEventSink { )] enum RunEventCommand { Event(RunEvent), - Flush(oneshot::Sender<()>), + Flush(oneshot::Sender>), +} + +#[derive(Clone, Debug, thiserror::Error)] +pub enum RunEventPersistenceError { + #[error("failed to persist run event {event} for run {run_id}")] + Write { + run_id: RunId, + event: String, + #[source] + source: SharedError, + }, + #[error("run event persistence task stopped")] + TaskStopped, } #[derive(Clone)] pub struct RunEventLogger { - tx: mpsc::UnboundedSender, + tx: mpsc::UnboundedSender, + failure_rx: watch::Receiver>, } impl RunEventLogger { #[must_use] pub fn new(sink: RunEventSink) -> Self { let (tx, mut rx) = mpsc::unbounded_channel(); + let (failure_tx, failure_rx) = watch::channel(None); tokio::spawn(async move { - // A dropped run event is unrecoverable history loss, so the first - // one is an ERROR worth investigating. A broken sink fails for - // every event that follows, so report the rest as a count at flush - // instead of one ERROR per event. Flush runs per stage and per - // agent turn, so only losses since the last summary are reported. - let mut write_failures: u64 = 0; - let mut summarized_failures: u64 = 0; + // The watch channel is the single record of the latched failure: + // the worker is its only writer, so borrowing it here cannot race. while let Some(command) = rx.recv().await { match command { RunEventCommand::Event(event) => { + if failure_tx.borrow().is_some() { + continue; + } if let Err(err) = sink.write_run_event(&event).await { - write_failures += 1; - if write_failures == 1 { - tracing::error!( - run_id = %event.run_id, - event = %event.body.event_name(), - error = %err, - "Failed to write run event", - ); - } else { - tracing::debug!( - run_id = %event.run_id, - event = %event.body.event_name(), - failures = write_failures, - error = %err, - "Failed to write run event", - ); - } + let rendered_error = collect_chain(err.as_ref()).join(": "); + tracing::error!( + run_id = %event.run_id, + event = %event.body.event_name(), + error = %rendered_error, + "Failed to persist run event; stopping workflow", + ); + failure_tx.send_replace(Some(RunEventPersistenceError::Write { + run_id: event.run_id, + event: event.body.event_name().to_string(), + source: SharedError::new(err), + })); } } RunEventCommand::Flush(tx) => { - if write_failures > summarized_failures { - tracing::error!( - lost = write_failures - summarized_failures, - total = write_failures, - "Run events were lost to write failures", - ); - summarized_failures = write_failures; - } - let _ = tx.send(()); + let result = failure_tx.borrow().clone().map_or(Ok(()), Err); + let _ = tx.send(result); } } } }); - Self { tx } + Self { tx, failure_rx } } pub fn register(&self, emitter: &Emitter) { let tx = self.tx.clone(); emitter.on_event(move |event| { if tx.send(RunEventCommand::Event(event.clone())).is_err() { - tracing::warn!("Run event logger channel closed while forwarding event"); + tracing::error!( + run_id = %event.run_id, + event = %event.body.event_name(), + "Run event persistence task stopped while forwarding event", + ); } }); } - pub async fn flush(&self) { + pub async fn wait_for_failure(&self) -> RunEventPersistenceError { + let mut failure_rx = self.failure_rx.clone(); + let failure = failure_rx.wait_for(Option::is_some).await; + match failure { + Ok(failure) => failure + .clone() + .expect("wait_for only returns values matching the predicate"), + Err(_) => RunEventPersistenceError::TaskStopped, + } + } + + pub async fn flush(&self) -> Result<(), RunEventPersistenceError> { let (tx, rx) = oneshot::channel(); if self.tx.send(RunEventCommand::Flush(tx)).is_err() { - tracing::warn!("Run event logger channel closed before flush"); - return; - } - if rx.await.is_err() { - tracing::warn!("Run event logger flush dropped before completion"); + return Err(RunEventPersistenceError::TaskStopped); } + rx.await + .unwrap_or(Err(RunEventPersistenceError::TaskStopped)) } } @@ -249,14 +270,15 @@ impl StoreProgressLogger { self.inner.register(emitter); } - pub async fn flush(&self) { - self.inner.flush().await; + pub async fn flush(&self) -> Result<(), RunEventPersistenceError> { + self.inner.flush().await } } #[cfg(test)] mod tests { use std::sync::Arc; + use std::sync::atomic::{AtomicUsize, Ordering}; use ::fabro_types::{Graph, RunNoticeLevel, WorkflowSettings, fixtures}; use fabro_types::test_support; @@ -439,7 +461,7 @@ mod tests { logger.register(&emitter); emitter.emit(&Event::RunPaused); - logger.flush().await; + logger.flush().await.unwrap(); let mut reader = BufReader::new(reader); let mut line = String::new(); @@ -448,4 +470,38 @@ mod tests { let payload = event_payload_from_redacted_json(line.trim_end(), &fixtures::RUN_8).unwrap(); assert_eq!(payload.as_value()["event"], "run.paused"); } + + #[tokio::test] + async fn run_event_logger_latches_write_failure_and_preserves_cause_chain() { + let writes = Arc::new(AtomicUsize::new(0)); + let writes_for_sink = Arc::clone(&writes); + let sink = RunEventSink::callback(move |_| { + writes_for_sink.fetch_add(1, Ordering::SeqCst); + async { + Err( + anyhow::anyhow!("request failed with status 413 Payload Too Large") + .context("worker lost canonical run store during append run event"), + ) + } + }); + let logger = RunEventLogger::new(sink); + let emitter = Emitter::new(fixtures::RUN_8); + logger.register(&emitter); + + emitter.emit(&Event::RunPaused); + + let failure = logger.wait_for_failure().await; + let rendered = collect_chain(&failure).join(": "); + assert!(rendered.contains("run.paused"), "{rendered}"); + assert!( + rendered.contains("worker lost canonical run store"), + "{rendered}" + ); + assert!(rendered.contains("413 Payload Too Large"), "{rendered}"); + + emitter.emit(&Event::RunUnpaused); + let flush_failure = logger.flush().await.unwrap_err(); + assert_eq!(collect_chain(&flush_failure), collect_chain(&failure)); + assert_eq!(writes.load(Ordering::SeqCst), 1); + } } diff --git a/lib/components/fabro-workflow/src/handler/agent.rs b/lib/components/fabro-workflow/src/handler/agent.rs index b3c67612b..b262cb976 100644 --- a/lib/components/fabro-workflow/src/handler/agent.rs +++ b/lib/components/fabro-workflow/src/handler/agent.rs @@ -630,7 +630,7 @@ mod tests { .execute(&node, &context, &graph, tmp.path(), &services) .await .unwrap(); - logger.flush().await; + logger.flush().await.unwrap(); let state = run_store.state().await.unwrap(); let node_state = state.stage(&StageId::new("plan", 1)).unwrap(); @@ -657,7 +657,7 @@ mod tests { .execute(&node, &context, &graph, tmp.path(), &services) .await .unwrap(); - logger.flush().await; + logger.flush().await.unwrap(); let state = run_store.state().await.unwrap(); let node_state = state.stage(&StageId::new("work", 1)).unwrap(); @@ -1151,7 +1151,7 @@ All checks passed. .execute(&node, &context, &graph, tmp.path(), &services) .await .unwrap(); - logger.flush().await; + logger.flush().await.unwrap(); let state = run_store.state().await.unwrap(); let node_state = state.stage(&StageId::new("step", 1)).unwrap(); @@ -1555,7 +1555,7 @@ Some text in between. .execute(&node, &context, &graph, tmp.path(), &services) .await .unwrap(); - logger.flush().await; + logger.flush().await.unwrap(); let state = run_store.state().await.unwrap(); let node_state = state.stage(&StageId::new("report", 1)).unwrap(); diff --git a/lib/components/fabro-workflow/src/handler/command.rs b/lib/components/fabro-workflow/src/handler/command.rs index c24353ccd..49c43230c 100644 --- a/lib/components/fabro-workflow/src/handler/command.rs +++ b/lib/components/fabro-workflow/src/handler/command.rs @@ -1077,7 +1077,7 @@ mod tests { .execute(&node, &context, &graph, run_dir.path(), &services) .await .unwrap(); - logger.flush().await; + logger.flush().await.unwrap(); let snapshot = run_store.state().await.unwrap(); let node_state = snapshot.stage(&StageId::new("script_node", 1)).unwrap(); @@ -1108,7 +1108,7 @@ mod tests { .execute(&node, &context, &graph, run_dir.path(), &services) .await .unwrap(); - logger.flush().await; + logger.flush().await.unwrap(); let snapshot = run_store.state().await.unwrap(); let node_state = snapshot.stage(&StageId::new("script_node", 1)).unwrap(); @@ -1135,7 +1135,7 @@ mod tests { .execute(&node, &context, &graph, run_dir.path(), &services) .await .unwrap(); - logger.flush().await; + logger.flush().await.unwrap(); let snapshot = run_store.state().await.unwrap(); let node_state = snapshot.stage(&StageId::new("script_node", 1)).unwrap(); @@ -1162,7 +1162,7 @@ mod tests { .execute(&node, &context, &graph, run_dir.path(), &services) .await .unwrap(); - logger.flush().await; + logger.flush().await.unwrap(); let snapshot = run_store.state().await.unwrap(); let node_state = snapshot.stage(&StageId::new("script_node", 1)).unwrap(); @@ -1187,7 +1187,7 @@ mod tests { .execute(&node, &context, &graph, run_dir.path(), &services) .await .unwrap(); - logger.flush().await; + logger.flush().await.unwrap(); let snapshot = run_store.state().await.unwrap(); let node_state = snapshot.stage(&StageId::new("script_node", 1)).unwrap(); @@ -1212,7 +1212,7 @@ mod tests { .execute(&node, &context, &graph, run_dir.path(), &services) .await .unwrap(); - logger.flush().await; + logger.flush().await.unwrap(); let snapshot = run_store.state().await.unwrap(); let node_state = snapshot.stage(&StageId::new("script_node", 1)).unwrap(); @@ -1242,7 +1242,7 @@ mod tests { .execute(&node, &context, &graph, run_dir.path(), &services) .await .unwrap_err(); - logger.flush().await; + logger.flush().await.unwrap(); let snapshot = run_store.state().await.unwrap(); let node_state = snapshot.stage(&StageId::new("script_node", 1)).unwrap(); @@ -1269,7 +1269,7 @@ mod tests { .execute(&node, &context, &graph, run_dir.path(), &services) .await .unwrap(); - logger.flush().await; + logger.flush().await.unwrap(); let snapshot = run_store.state().await.unwrap(); let node = snapshot diff --git a/lib/components/fabro-workflow/src/handler/parallel.rs b/lib/components/fabro-workflow/src/handler/parallel.rs index 1d245659f..fed0fd03e 100644 --- a/lib/components/fabro-workflow/src/handler/parallel.rs +++ b/lib/components/fabro-workflow/src/handler/parallel.rs @@ -1405,7 +1405,7 @@ mod tests { .execute(&node, &context, &graph, Path::new("/tmp/test"), &services) .await .unwrap(); - logger.flush().await; + logger.flush().await.unwrap(); assert_eq!(outcome.status, StageOutcome::Succeeded); let results: Vec = diff --git a/lib/components/fabro-workflow/src/handler/prompt.rs b/lib/components/fabro-workflow/src/handler/prompt.rs index e61a2809c..f2d3df535 100644 --- a/lib/components/fabro-workflow/src/handler/prompt.rs +++ b/lib/components/fabro-workflow/src/handler/prompt.rs @@ -570,7 +570,7 @@ mod tests { .execute(&node, &context, &graph, tmp.path(), &services) .await .unwrap(); - logger.flush().await; + logger.flush().await.unwrap(); let state = run_store.state().await.unwrap(); let node_state = state.stage(&StageId::new("classify", 1)).unwrap(); diff --git a/lib/components/fabro-workflow/src/operations/resume.rs b/lib/components/fabro-workflow/src/operations/resume.rs index 433947716..466ae5e0c 100644 --- a/lib/components/fabro-workflow/src/operations/resume.rs +++ b/lib/components/fabro-workflow/src/operations/resume.rs @@ -43,8 +43,7 @@ pub async fn resume(run_dir: &Path, services: StartServices) -> Result Result Result Error { + cancel_token.cancel(); + error.into() +} + +/// Race a pipeline step against the first latched run-event persistence +/// failure. When the failure wins, the step future is dropped mid-flight and +/// the run token is cancelled. +async fn race_persistence( + logger: &RunEventLogger, + cancel_token: &CancellationToken, + step: impl Future, +) -> Result { + tokio::select! { + result = step => Ok(result), + failure = logger.wait_for_failure() => { + Err(stop_for_run_event_persistence_failure(cancel_token, failure)) + } + } +} + +async fn flush_or_stop( + logger: &RunEventLogger, + cancel_token: &CancellationToken, +) -> Result<(), Error> { + logger + .flush() + .await + .map_err(|failure| stop_for_run_event_persistence_failure(cancel_token, failure)) +} + impl RunSession { async fn new(persisted: &Persisted, services: StartServices) -> Result { let record = persisted.run_spec(); @@ -694,6 +735,7 @@ impl RunSession { resume: Option, ) -> Result { let on_node = self.on_node.clone(); + let run_cancel_token = self.cancel_token.clone(); let record = persisted.run_spec(); let run_options = RunOptions { @@ -782,7 +824,19 @@ impl RunSession { seed_context: self.seed_context, fabro_run_tools: self.fabro_run_tools, }; - let mut initialized = Box::pin(pipeline::initialize(persisted, init_options)).await?; + let mut initialized = match race_persistence( + &store_progress_logger, + &run_cancel_token, + Box::pin(pipeline::initialize(persisted, init_options)), + ) + .await? + { + Ok(initialized) => initialized, + Err(err) => { + flush_or_stop(&store_progress_logger, &run_cancel_token).await?; + return Err(err); + } + }; initialized.on_node = on_node; let sandbox_for_cleanup = Arc::clone(&initialized.engine.run.sandbox); @@ -806,8 +860,15 @@ impl RunSession { steering_hub_for_drain.drain_pending_at_run_end(); }); - let executed = pipeline::execute(initialized).await; - store_progress_logger.flush().await; + flush_or_stop(&store_progress_logger, &run_cancel_token).await?; + + let executed = race_persistence( + &store_progress_logger, + &run_cancel_token, + Box::pin(pipeline::execute(initialized)), + ) + .await?; + flush_or_stop(&store_progress_logger, &run_cancel_token).await?; let final_context = Some(executed.final_context.clone()); let finalize_opts = FinalizeOptions { @@ -827,16 +888,21 @@ impl RunSession { model: self.pr_model, }; - let concluding = async { - let concluded = Box::pin(pipeline::conclude(executed, &finalize_opts)).await?; - let published = Box::pin(pipeline::publish(concluded, &publish_opts)).await; - Box::pin(pipeline::finalize(published, &finalize_opts)).await - }; - let finalized = match concluding.await { + let concluding = race_persistence( + &store_progress_logger, + &run_cancel_token, + Box::pin(async { + let concluded = Box::pin(pipeline::conclude(executed, &finalize_opts)).await?; + let published = Box::pin(pipeline::publish(concluded, &publish_opts)).await; + Box::pin(pipeline::finalize(published, &finalize_opts)).await + }), + ) + .await?; + let finalized = match concluding { Ok(finalized) => finalized, Err(err) => { self.steering_hub.drain_pending_at_run_end(); - store_progress_logger.flush().await; + flush_or_stop(&store_progress_logger, &run_cancel_token).await?; return Err(err); } }; @@ -845,7 +911,7 @@ impl RunSession { // scopeguard above re-runs as a no-op (drain is idempotent on an // already-empty buffer) on the way out of scope. self.steering_hub.drain_pending_at_run_end(); - store_progress_logger.flush().await; + flush_or_stop(&store_progress_logger, &run_cancel_token).await?; scopeguard::ScopeGuard::into_inner(cleanup_guard); @@ -1005,13 +1071,20 @@ impl Drop for DetachedRunCompletionGuard { 0, ) .await; - let _ = append_event_to_sink(&event_sink, &run_id, &Event::RunNotice { + if let Err(err) = append_event_to_sink(&event_sink, &run_id, &Event::RunNotice { level: RunNoticeLevel::Error, code: code.to_string(), message: message.to_string(), exec_output_tail: None, }) - .await; + .await + { + let rendered_error = collect_chain(&err).join(": "); + tracing::warn!( + error = %rendered_error, + "Failed to append detached completion notice", + ); + } }); } } @@ -1035,7 +1108,11 @@ async fn persist_detached_failure( exec_output_tail: None, }; if let Err(err) = append_event_to_sink(event_sink, &run_id, &event).await { - tracing::warn!(error = %err, "Failed to append detached failure notice"); + let rendered_error = collect_chain(&err).join(": "); + tracing::warn!( + error = %rendered_error, + "Failed to append detached failure notice", + ); } Ok(()) @@ -1094,8 +1171,19 @@ mod tests { work -> exit }"#; + const BLOCKING_DOT: &str = r#"digraph Test { + graph [goal="Wait forever"] + start [shape=Mdiamond] + block [type="blocking"] + exit [shape=Msquare] + start -> block + block -> exit + }"#; + struct TimedOutcomeHandler; + struct BlockingHandler; + fn timed_success_outcome() -> Outcome { let mut outcome = Outcome::success(); outcome.timing = Some(StageTiming::new(0, 100, 50)); @@ -1127,6 +1215,20 @@ mod tests { } } + #[async_trait::async_trait] + impl Handler for BlockingHandler { + async fn execute( + &self, + _node: &fabro_graphviz::graph::Node, + _context: &Context, + _graph: &fabro_graphviz::graph::Graph, + _run_dir: &Path, + _services: &EngineServices, + ) -> Result { + std::future::pending().await + } + } + fn memory_store() -> Arc { Arc::new(Database::new( Arc::new(InMemory::new()), @@ -2130,6 +2232,74 @@ reasoning = false assert!(run_store.state().await.unwrap().conclusion.is_some()); } + #[tokio::test] + async fn event_persistence_failure_stops_execution_and_fails_run() { + let temp = tempfile::tempdir().unwrap(); + let (storage_root, run_dir) = storage_root_and_run_dir(&temp); + let emitter = Arc::new(Emitter::new(fixtures::RUN_1)); + let mut registry = test_registry(); + registry.register("blocking", Box::new(BlockingHandler)); + let (_persisted, store) = persisted_workflow(BLOCKING_DOT, &storage_root).await; + let run_store = store.open_run(&fixtures::RUN_1).await.unwrap(); + let canonical_sink = RunEventSink::store(run_store.clone()); + let mut services = test_start_services(&store, &run_dir, emitter, Arc::new(registry)).await; + let cancel_token = services.cancel_token.clone(); + services.event_sink = RunEventSink::callback(move |event| { + let canonical_sink = canonical_sink.clone(); + async move { + if matches!(&event.body, EventBody::StageStarted(_)) + && event.node_id.as_deref() == Some("block") + { + return Err(anyhow::anyhow!( + "request failed with status 413 Payload Too Large" + ) + .context("worker lost canonical run store during append run event")); + } + canonical_sink.write_run_event(&event).await + } + }); + + let result = tokio::time::timeout(Duration::from_secs(2), start(&run_dir, services)) + .await + .expect("event persistence failure should stop the blocking stage"); + let Err(error) = result else { + panic!("event persistence failure should fail the run"); + }; + + assert!(cancel_token.is_cancelled()); + let rendered = error.display_with_causes(); + assert!( + rendered.contains("run event persistence failed"), + "{rendered}" + ); + assert!(rendered.contains("stage.started"), "{rendered}"); + assert!(rendered.contains("413 Payload Too Large"), "{rendered}"); + + let projection = run_store.state().await.unwrap(); + assert!(matches!(projection.status, RunStatus::Failed { .. })); + let events = run_store.list_events().await.unwrap(); + let run_failed = events + .iter() + .find_map(|event| match &event.event.body { + EventBody::RunFailed(properties) => Some(properties), + _ => None, + }) + .expect("persistence failure should emit run.failed"); + assert!( + run_failed + .failure + .detail + .causes + .iter() + .any(|cause| cause.contains("413 Payload Too Large")) + ); + assert!( + events + .iter() + .all(|event| !matches!(&event.event.body, EventBody::RunCompleted(_))) + ); + } + #[tokio::test] async fn start_can_run_bundle_backed_child_workflow_without_workflow_bundle_json() { let temp = tempfile::tempdir().unwrap(); diff --git a/lib/components/fabro-workflow/src/pipeline/execute/tests.rs b/lib/components/fabro-workflow/src/pipeline/execute/tests.rs index af83d7cfe..6a2f63078 100644 --- a/lib/components/fabro-workflow/src/pipeline/execute/tests.rs +++ b/lib/components/fabro-workflow/src/pipeline/execute/tests.rs @@ -304,7 +304,7 @@ async fn execute_test_run_with_options( .unwrap(); let executed = execute(initialized).await; - store_logger.flush().await; + store_logger.flush().await.unwrap(); executed } diff --git a/lib/components/fabro-workflow/src/pipeline/finalize.rs b/lib/components/fabro-workflow/src/pipeline/finalize.rs index 18b7ea15b..76dde359f 100644 --- a/lib/components/fabro-workflow/src/pipeline/finalize.rs +++ b/lib/components/fabro-workflow/src/pipeline/finalize.rs @@ -1114,8 +1114,8 @@ mod tests { let temp = tempfile::tempdir().unwrap(); let run_dir = temp.path().join("run"); std::fs::create_dir_all(&run_dir).unwrap(); - let inner_store = test_store().create_run(&test_run_id()).await.unwrap(); - let run_store = inner_store; + let run_store = seeded_run_store().await; + crate::test_support::mark_run_running(&run_store, &test_run_id()).await; let emitter = Arc::new(Emitter::new(test_run_id())); let store_logger = StoreProgressLogger::new(run_store.clone()); store_logger.register(&emitter); @@ -1158,7 +1158,7 @@ mod tests { }) .await .unwrap(); - store_logger.flush().await; + store_logger.flush().await.unwrap(); assert_eq!(concluded.conclusion.status, StageOutcome::Succeeded); } diff --git a/lib/components/fabro-workflow/src/pipeline/initialize.rs b/lib/components/fabro-workflow/src/pipeline/initialize.rs index 286e8bb77..ebc73568c 100644 --- a/lib/components/fabro-workflow/src/pipeline/initialize.rs +++ b/lib/components/fabro-workflow/src/pipeline/initialize.rs @@ -751,7 +751,7 @@ mod tests { use fabro_graphviz::graph::{AttrValue, Edge, Graph, Node}; use fabro_interview::AutoApproveInterviewer; use fabro_sandbox::SandboxSpec; - use fabro_store::Database; + use fabro_store::{Database, RunDatabase}; use fabro_types::settings::run::RunModelControls; use fabro_types::{ EventBody, ForkSourceRef, RunEvent, RunId, WorkflowSettings, fixtures, test_support, @@ -796,6 +796,37 @@ mod tests { )) } + async fn seed_run_created( + run_store: &RunDatabase, + settings: serde_json::Value, + graph: serde_json::Value, + source_directory: Option, + fork_source_ref: Option, + ) { + crate::event::append_event(run_store, &test_run_id(), &Event::RunCreated { + run_id: test_run_id(), + title: None, + settings, + graph, + workflow_source: None, + labels: BTreeMap::new(), + source_directory, + workflow_slug: Some("test".to_string()), + workflow_version_id: None, + automation: None, + provenance: test_support::test_run_provenance(), + manifest_blob: None, + spec_blob: None, + git: None, + fork_source_ref, + retried_from: None, + parent_id: None, + web_url: None, + }) + .await + .unwrap(); + } + fn simple_graph() -> (Graph, String) { let source = r"digraph test { start [shape=Mdiamond]; @@ -1122,28 +1153,14 @@ mod tests { let mut run_options = test_settings(&run_dir); run_options.settings = settings; run_options.fork_source_ref = fork_source_ref; - crate::event::append_event(&run_store, &test_run_id(), &Event::RunCreated { - run_id: test_run_id(), - title: None, - settings: serde_json::to_value(&run_options.settings).unwrap(), - graph: serde_json::to_value(&graph).unwrap(), - workflow_source: None, - labels: BTreeMap::new(), - source_directory: Some(workspace.display().to_string()), - workflow_slug: Some("test".to_string()), - workflow_version_id: None, - automation: None, - provenance: test_support::test_run_provenance(), - manifest_blob: None, - spec_blob: None, - git: None, - fork_source_ref: run_options.fork_source_ref.clone(), - retried_from: None, - parent_id: None, - web_url: None, - }) - .await - .unwrap(); + seed_run_created( + &run_store, + serde_json::to_value(&run_options.settings).unwrap(), + serde_json::to_value(&graph).unwrap(), + Some(workspace.display().to_string()), + run_options.fork_source_ref.clone(), + ) + .await; initialize(persisted, InitOptions { resume: Some(ResumeState::for_test( @@ -1409,10 +1426,18 @@ mod tests { let run_dir = temp.path().join("run"); std::fs::create_dir_all(&run_dir).unwrap(); let (graph, source) = simple_graph(); - let persisted = test_persisted(graph, source, &run_dir); + let persisted = test_persisted(graph.clone(), source, &run_dir); let emitter = Arc::new(crate::event::Emitter::new(test_run_id())); let store = memory_store(); let run_store = store.create_run(&test_run_id()).await.unwrap(); + seed_run_created( + &run_store, + serde_json::to_value(WorkflowSettings::default()).unwrap(), + serde_json::to_value(graph).unwrap(), + None, + None, + ) + .await; let store_logger = StoreProgressLogger::new(run_store.clone()); let seen = Arc::new(std::sync::Mutex::new(Vec::new())); emitter.on_event({ @@ -1463,7 +1488,7 @@ mod tests { }) .await .unwrap(); - store_logger.flush().await; + store_logger.flush().await.unwrap(); assert_eq!(initialized.run_options.run_dir, run_dir); assert!( diff --git a/lib/components/fabro-workflow/src/test_support.rs b/lib/components/fabro-workflow/src/test_support.rs index 55f7f9cd4..6c9f55b99 100644 --- a/lib/components/fabro-workflow/src/test_support.rs +++ b/lib/components/fabro-workflow/src/test_support.rs @@ -52,7 +52,11 @@ pub(crate) fn test_configured_provider_ids( /// persisted before tests reopen the run store. async fn execute_and_emit_terminal(initialized: InitializedState) -> Executed { let executed = Box::pin(pipeline::execute(initialized.initialized)).await; - initialized.store_logger.flush().await; + initialized + .store_logger + .flush() + .await + .expect("test run events should persist"); let state = executed.engine.run.run_store.state().await.ok(); let billing = state.as_ref().and_then(billing_from_projection); let event = build_terminal_event( @@ -65,7 +69,11 @@ async fn execute_and_emit_terminal(initialized: InitializedState) -> Executed { billing, ); executed.engine.run.emitter.emit(&event); - initialized.store_logger.flush().await; + initialized + .store_logger + .flush() + .await + .expect("test run events should persist"); executed } @@ -481,7 +489,11 @@ pub async fn run_graph_with_state_and_llm_source( ) .await; let executed = pipeline::execute(initialized.initialized).await; - initialized.store_logger.flush().await; + initialized + .store_logger + .flush() + .await + .expect("test run events should persist"); let outcome = executed.outcome?; let state = executed .engine