fix: fail runs when event persistence is lost

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.

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;

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};
@@ -149,86 +150,102 @@ impl RunEventSink {
 )]
 enum RunEventCommand {
     Event(RunEvent),
-    Flush(oneshot::Sender<()>),
+    Flush(oneshot::Sender<Result<(), RunEventPersistenceError>>),
+}
+
+#[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<RunEventCommand>,
+    tx:         mpsc::UnboundedSender<RunEventCommand>,
+    failure_rx: watch::Receiver<Option<RunEventPersistenceError>>,
 }
 
 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;
+            let mut persistence_failure = None;
             while let Some(command) = rx.recv().await {
                 match command {
                     RunEventCommand::Event(event) => {
+                        if persistence_failure.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",
+                            );
+                            let failure = RunEventPersistenceError::Write {
+                                run_id: event.run_id,
+                                event:  event.body.event_name().to_string(),
+                                source: SharedError::new(err),
+                            };
+                            persistence_failure = Some(failure.clone());
+                            failure_tx.send_replace(Some(failure));
                         }
                     }
                     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 = persistence_failure.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();
+        loop {
+            if let Some(failure) = failure_rx.borrow_and_update().clone() {
+                return failure;
+            }
+            if failure_rx.changed().await.is_err() {
+                return 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
+            .map_err(|_| RunEventPersistenceError::TaskStopped)?
     }
 }
 
@@ -249,14 +266,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 +457,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 +466,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);
+    }
 }

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();

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

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<ParallelBranchResult> =

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();

lib/components/fabro-workflow/src/operations/resume.rs

@@ -44,7 +44,7 @@ pub async fn resume(run_dir: &Path, services: StartServices) -> Result<Started,
         &Event::RunSubmitted { definition_blob },
     )
     .await
-    .map_err(|err| Error::engine(err.to_string()))?;
+    .map_err(|err| Error::engine_with_source("failed to persist run.submitted event", err))?;
 
     Box::pin(execute_persisted_run(run_dir, Some(resume_state), services)).await
 }

lib/components/fabro-workflow/src/operations/start.rs

@@ -21,6 +21,7 @@ use fabro_types::settings::run::{
     RunPrepareSettings as ResolvedRunPrepareSettings,
 };
 use fabro_types::{ManifestPath, RunId, RunRunnableSource, SandboxProviderKind};
+use fabro_util::error::collect_chain;
 use fabro_vault::Vault;
 use tokio::runtime::Handle;
 use tokio::sync::RwLock as AsyncRwLock;
@@ -31,7 +32,8 @@ use crate::artifact_upload::ArtifactSink;
 use crate::context::Context;
 use crate::error::{self, Error};
 use crate::event::{
-    Emitter, Event, EventBody, RunEventLogger, RunEventSink, RunNoticeLevel, append_event_to_sink,
+    Emitter, Event, EventBody, RunEventLogger, RunEventPersistenceError, RunEventSink,
+    RunNoticeLevel, append_event_to_sink,
 };
 use crate::handler::HandlerRegistry;
 use crate::model_fallback::{ModelFallbackNotice, ResolvedModelFallbacks, resolve_model_fallbacks};
@@ -158,7 +160,9 @@ pub async fn start(run_dir: &Path, services: StartServices) -> Result<Started, E
             },
         )
         .await
-        .map_err(|err| Error::engine(err.to_string()))?;
+        .map_err(|err| {
+            Error::engine_with_source("failed to persist run.start_requested event", err)
+        })?;
         append_event_to_sink(
             &services.event_sink,
             &services.run_id,
@@ -168,7 +172,7 @@ pub async fn start(run_dir: &Path, services: StartServices) -> Result<Started, E
             },
         )
         .await
-        .map_err(|err| Error::engine(err.to_string()))?;
+        .map_err(|err| Error::engine_with_source("failed to persist run.runnable event", err))?;
     }
 
     Box::pin(execute_persisted_run(run_dir, None, services)).await
@@ -198,7 +202,7 @@ pub(super) async fn execute_persisted_run(
         return Err(error);
     }
     if let Err(err) = append_event_to_sink(&event_sink, &run_id, &Event::RunStarting).await {
-        let error = Error::engine(err.to_string());
+        let error = Error::engine_with_source("failed to persist run.starting event", err);
         let _ = persist_detached_failure(
             run_id,
             &run_store,
@@ -318,7 +322,13 @@ async fn emit_workflow_run_failed(
         conclusion.billing,
     );
     if let Err(err) = append_event_to_sink(event_sink, &run_id, &failure_event).await {
-        tracing::warn!(error = %err, "Failed to append run.failed event");
+        let rendered_error = collect_chain(err.as_ref()).join(": ");
+        tracing::error!(
+            run_id = %run_id,
+            event = "run.failed",
+            error = %rendered_error,
+            "Failed to append run.failed event",
+        );
     }
 }
 
@@ -347,6 +357,14 @@ async fn persist_terminal_engine_failure(
     .await;
 }
 
+fn stop_for_run_event_persistence_failure(
+    cancel_token: &CancellationToken,
+    error: RunEventPersistenceError,
+) -> Error {
+    cancel_token.cancel();
+    Error::engine_with_source("run event persistence failed", error)
+}
+
 impl RunSession {
     async fn new(persisted: &Persisted, services: StartServices) -> Result<Self, Error> {
         let record = persisted.run_spec();
@@ -691,6 +709,7 @@ impl RunSession {
         resume: Option<ResumeState>,
     ) -> Result<Started, Error> {
         let on_node = self.on_node.clone();
+        let run_cancel_token = self.cancel_token.clone();
 
         let record = persisted.run_spec();
         let run_options = RunOptions {
@@ -779,7 +798,28 @@ 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 initializing = Box::pin(pipeline::initialize(persisted, init_options));
+        let initialized = tokio::select! {
+            result = &mut initializing => result,
+            failure = store_progress_logger.wait_for_failure() => {
+                return Err(stop_for_run_event_persistence_failure(
+                    &run_cancel_token,
+                    failure,
+                ));
+            }
+        };
+        let mut initialized = match initialized {
+            Ok(initialized) => initialized,
+            Err(err) => {
+                if let Err(failure) = store_progress_logger.flush().await {
+                    return Err(stop_for_run_event_persistence_failure(
+                        &run_cancel_token,
+                        failure,
+                    ));
+                }
+                return Err(err);
+            }
+        };
         initialized.on_node = on_node;
 
         let sandbox_for_cleanup = Arc::clone(&initialized.engine.run.sandbox);
@@ -803,8 +843,23 @@ impl RunSession {
             steering_hub_for_drain.drain_pending_at_run_end();
         });
 
-        let executed = pipeline::execute(initialized).await;
-        store_progress_logger.flush().await;
+        store_progress_logger.flush().await.map_err(|failure| {
+            stop_for_run_event_persistence_failure(&run_cancel_token, failure)
+        })?;
+
+        let mut executing = Box::pin(pipeline::execute(initialized));
+        let executed = tokio::select! {
+            executed = &mut executing => executed,
+            failure = store_progress_logger.wait_for_failure() => {
+                return Err(stop_for_run_event_persistence_failure(
+                    &run_cancel_token,
+                    failure,
+                ));
+            }
+        };
+        store_progress_logger.flush().await.map_err(|failure| {
+            stop_for_run_event_persistence_failure(&run_cancel_token, failure)
+        })?;
         let final_context = Some(executed.final_context.clone());
 
         let finalize_opts = FinalizeOptions {
@@ -824,16 +879,30 @@ impl RunSession {
             model:      self.pr_model,
         };
 
-        let concluding = async {
+        let mut concluding = 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
+        });
+        let concluding = tokio::select! {
+            result = &mut concluding => result,
+            failure = store_progress_logger.wait_for_failure() => {
+                return Err(stop_for_run_event_persistence_failure(
+                    &run_cancel_token,
+                    failure,
+                ));
+            }
         };
-        let finalized = match concluding.await {
+        let finalized = match concluding {
             Ok(finalized) => finalized,
             Err(err) => {
                 self.steering_hub.drain_pending_at_run_end();
-                store_progress_logger.flush().await;
+                if let Err(failure) = store_progress_logger.flush().await {
+                    return Err(stop_for_run_event_persistence_failure(
+                        &run_cancel_token,
+                        failure,
+                    ));
+                }
                 return Err(err);
             }
         };
@@ -842,7 +911,9 @@ 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;
+        store_progress_logger.flush().await.map_err(|failure| {
+            stop_for_run_event_persistence_failure(&run_cancel_token, failure)
+        })?;
 
         scopeguard::ScopeGuard::into_inner(cleanup_guard);
 
@@ -1002,13 +1073,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.as_ref()).join(": ");
+                    tracing::warn!(
+                        error = %rendered_error,
+                        "Failed to append detached completion notice",
+                    );
+                }
             });
         }
     }
@@ -1032,7 +1110,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.as_ref()).join(": ");
+        tracing::warn!(
+            error = %rendered_error,
+            "Failed to append detached failure notice",
+        );
     }
 
     Ok(())
@@ -1091,8 +1173,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));
@@ -1124,6 +1217,31 @@ 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<Outcome, Error> {
+            std::future::pending().await
+        }
+
+        async fn simulate(
+            &self,
+            _node: &fabro_graphviz::graph::Node,
+            _context: &Context,
+            _graph: &fabro_graphviz::graph::Graph,
+            _run_dir: &Path,
+            _services: &EngineServices,
+        ) -> Result<Outcome, Error> {
+            std::future::pending().await
+        }
+    }
+
     fn memory_store() -> Arc<Database> {
         Arc::new(Database::new(
             Arc::new(InMemory::new()),
@@ -2087,6 +2205,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();

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
 }
 

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);
     }

lib/components/fabro-workflow/src/pipeline/initialize.rs

@@ -1326,10 +1326,32 @@ 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();
+        crate::event::append_event(&run_store, &test_run_id(), &Event::RunCreated {
+            run_id:              test_run_id(),
+            title:               None,
+            settings:            serde_json::to_value(WorkflowSettings::default()).unwrap(),
+            graph:               serde_json::to_value(graph).unwrap(),
+            workflow_source:     None,
+            labels:              BTreeMap::new(),
+            source_directory:    None,
+            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:     None,
+            retried_from:        None,
+            parent_id:           None,
+            web_url:             None,
+        })
+        .await
+        .unwrap();
         let store_logger = StoreProgressLogger::new(run_store.clone());
         let seen = Arc::new(std::sync::Mutex::new(Vec::new()));
         emitter.on_event({
@@ -1380,7 +1402,7 @@ mod tests {
         })
         .await
         .unwrap();
-        store_logger.flush().await;
+        store_logger.flush().await.unwrap();
 
         assert_eq!(initialized.run_options.run_dir, run_dir);
         assert!(

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