Merge pull request #779 from fabro-sh/codex/fail-closed-run-event-persistence

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/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<fabro_validate::ValidationError> for Error {
     }
 }
 
+impl From<RunEventPersistenceError> for Error {
+    fn from(err: RunEventPersistenceError) -> Self {
+        Self::engine_with_source("run event persistence failed", err)
+    }
+}
+
 impl From<fabro_checkpoint::MetadataError> for Error {
     fn from(err: fabro_checkpoint::MetadataError) -> Self {
         match err {

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};
@@ -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<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;
+            // 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);
+    }
 }

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

@@ -43,8 +43,7 @@ pub async fn resume(run_dir: &Path, services: StartServices) -> Result<Started,
         &services.run_id,
         &Event::RunSubmitted { definition_blob },
     )
-    .await
-    .map_err(|err| Error::engine(err.to_string()))?;
+    .await?;
 
     Box::pin(execute_persisted_run(run_dir, Some(resume_state), services)).await
 }

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

@@ -1,4 +1,5 @@
 use std::collections::HashSet;
+use std::future::Future;
 use std::path::Path;
 use std::sync::{Arc, Mutex};
 use std::time::{Duration, Instant};
@@ -21,6 +22,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 +33,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,8 +161,7 @@ pub async fn start(run_dir: &Path, services: StartServices) -> Result<Started, E
                 actor:  None,
             },
         )
-        .await
-        .map_err(|err| Error::engine(err.to_string()))?;
+        .await?;
         append_event_to_sink(
             &services.event_sink,
             &services.run_id,
@@ -168,8 +170,7 @@ pub async fn start(run_dir: &Path, services: StartServices) -> Result<Started, E
                 actor:  None,
             },
         )
-        .await
-        .map_err(|err| Error::engine(err.to_string()))?;
+        .await?;
     }
 
     Box::pin(execute_persisted_run(run_dir, None, services)).await
@@ -199,7 +200,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::from(err);
         let _ = persist_detached_failure(
             run_id,
             &run_store,
@@ -319,7 +320,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).join(": ");
+        tracing::error!(
+            run_id = %run_id,
+            event = "run.failed",
+            error = %rendered_error,
+            "Failed to append run.failed event",
+        );
     }
 }
 
@@ -348,6 +355,40 @@ async fn persist_terminal_engine_failure(
     .await;
 }
 
+fn stop_for_run_event_persistence_failure(
+    cancel_token: &CancellationToken,
+    error: RunEventPersistenceError,
+) -> 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<T>(
+    logger: &RunEventLogger,
+    cancel_token: &CancellationToken,
+    step: impl Future<Output = T>,
+) -> Result<T, Error> {
+    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<Self, Error> {
         let record = persisted.run_spec();
@@ -694,6 +735,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 {
@@ -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<Outcome, Error> {
+            std::future::pending().await
+        }
+    }
+
     fn memory_store() -> Arc<Database> {
         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();

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

@@ -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<String>,
+        fork_source_ref: Option<ForkSourceRef>,
+    ) {
+        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!(

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