diff --git a/crates/jp_cli/src/cmd.rs b/crates/jp_cli/src/cmd.rs
index abdef848..71148ecc 100644
--- a/crates/jp_cli/src/cmd.rs
+++ b/crates/jp_cli/src/cmd.rs
@@ -178,6 +178,25 @@ impl IntoPartialAppConfig for Commands {
 
 pub(crate) type Output = std::result::Result<(), Error>;
 
+/// Fold a drained persist failure into a command's result.
+///
+/// With no failure the result passes through.
+/// A failure on an otherwise successful run becomes the error, so an unsaved
+/// conversation cannot exit zero.
+/// Alongside an existing error the failure is attached as metadata: the two can
+/// be independent (a provider error and a full disk), and the primary error is
+/// the more specific diagnostic.
+pub(crate) fn fold_persist_failure(result: Output, persist: Option<jp_workspace::Error>) -> Output {
+    match (result, persist) {
+        (result, None) => result,
+        (Ok(()), Some(persist)) => Err(Error::from(crate::error::Error::Workspace(persist))),
+        (Err(mut error), Some(persist)) => {
+            error.push_metadata("persist_failure", persist.to_string());
+            Err(error)
+        }
+    }
+}
+
 #[derive(Debug, thiserror::Error)]
 pub(crate) struct Error {
     /// The error code.
@@ -221,6 +240,14 @@ impl Error {
             ..self
         }
     }
+
+    /// Append a metadata entry, rendered under the error's message.
+    ///
+    /// Used to attach a secondary fact to an error that already has a more
+    /// specific message of its own.
+    pub(super) fn push_metadata(&mut self, key: &str, value: impl Into<Value>) {
+        self.metadata.push((key.to_owned(), value.into()));
+    }
 }
 
 impl fmt::Display for Error {
@@ -738,6 +765,24 @@ impl From<jp_storage::Error> for Error {
                 ("path", path.to_string().into()),
             ]
             .into(),
+            Error::OutOfSpace { path, source } => [
+                ("message", "No space left on device".into()),
+                ("path", path.to_string().into()),
+                ("error", source.to_string().into()),
+                (
+                    "suggestion",
+                    "Free up disk space and re-run. Anything after the last successful write was \
+                     not saved."
+                        .into(),
+                ),
+            ]
+            .into(),
+            Error::WriteFailed { path, source } => [
+                ("message", "Failed to write file".into()),
+                ("path", path.to_string().into()),
+                ("error", source.to_string().into()),
+            ]
+            .into(),
             Error::ConversationNotFound(id) => [
                 ("message", "Conversation not found.".into()),
                 ("id", id.to_string().into()),
@@ -794,3 +839,7 @@ impl From<jp_id::Error> for Error {
         Self::from(metadata)
     }
 }
+
+#[cfg(test)]
+#[path = "cmd_tests.rs"]
+mod tests;
diff --git a/crates/jp_cli/src/cmd/config/set.rs b/crates/jp_cli/src/cmd/config/set.rs
index aca90843..a3bacac5 100644
--- a/crates/jp_cli/src/cmd/config/set.rs
+++ b/crates/jp_cli/src/cmd/config/set.rs
@@ -53,9 +53,12 @@ impl Set {
                 LockOutcome::ForkConversation(_) => unreachable!("fork not allowed"),
             };
 
-            let conv = lock.into_mut();
+            let mut conv = lock.into_mut();
             let id = conv.id();
             conv.update_events(|events| events.add_config_delta(config_delta.clone()));
+            // Write before reporting success, so a failed write is an error
+            // rather than a confirmation the user cannot trust.
+            conv.flush()?;
             ctx.printer
                 .println(format!("Set configuration in conversation {id}"));
         }
diff --git a/crates/jp_cli/src/cmd/conversation/compact.rs b/crates/jp_cli/src/cmd/conversation/compact.rs
index 160c4dd6..dd90e090 100644
--- a/crates/jp_cli/src/cmd/conversation/compact.rs
+++ b/crates/jp_cli/src/cmd/conversation/compact.rs
@@ -602,7 +602,7 @@ impl Compact {
         };
 
         let cfg = ctx.config();
-        let conv = lock.into_mut();
+        let mut conv = lock.into_mut();
         let events_snapshot = conv.events().clone();
 
         if self.reset {
@@ -617,6 +617,9 @@ impl Compact {
                 }
             } else {
                 let removed = conv.update_events(ConversationStream::remove_compactions);
+                // Write before reporting, so a failed write is an error rather
+                // than a confirmation the user cannot trust.
+                conv.flush()?;
                 if removed > 0 {
                     ctx.printer
                         .println(format!("Removed {removed} compaction event(s)."));
@@ -681,6 +684,9 @@ impl Compact {
             &conv.id().to_string(),
         ));
         apply_compactions(&conv, compactions);
+        // Write before reporting the timeline, so a failed write is an error
+        // rather than a confirmation the user cannot trust.
+        conv.flush()?;
         for line in timeline_lines(&segments, last_turn, false) {
             ctx.printer.println(line);
         }
diff --git a/crates/jp_cli/src/cmd/conversation/edit.rs b/crates/jp_cli/src/cmd/conversation/edit.rs
index 994d5e9f..ba0b8d45 100644
--- a/crates/jp_cli/src/cmd/conversation/edit.rs
+++ b/crates/jp_cli/src/cmd/conversation/edit.rs
@@ -301,6 +301,10 @@ impl Edit {
             } else if self.no_expires_at {
                 conv.update_metadata(|m| m.expires_at = None);
             }
+
+            // Write before reporting success, so a failed write is an error
+            // rather than a confirmation the user cannot trust.
+            conv.flush()?;
         }
 
         ctx.printer.println("Conversation(s) updated.");
diff --git a/crates/jp_cli/src/cmd/conversation/fork.rs b/crates/jp_cli/src/cmd/conversation/fork.rs
index e242009d..98626b76 100644
--- a/crates/jp_cli/src/cmd/conversation/fork.rs
+++ b/crates/jp_cli/src/cmd/conversation/fork.rs
@@ -124,9 +124,13 @@ impl Fork {
                 }
             })?;
 
+            // One mutable scope for both post-fork mutations, so they share a
+            // single write at the closing flush.
+            let mut conv = lock.as_mut();
+
             if self.compact.should_compact() {
                 let cfg = ctx.config();
-                let events_snapshot = lock.events().clone();
+                let events_snapshot = conv.events().clone();
                 let rules = self
                     .compact
                     .effective_rules(&cfg.conversation.compaction.rules)
@@ -143,17 +147,21 @@ impl Fork {
                 )
                 .await?;
                 for compaction in compactions {
-                    lock.as_mut()
-                        .update_events(|events| events.add_compaction(compaction));
+                    conv.update_events(|events| events.add_compaction(compaction));
                 }
             }
 
             if let Some(title) = &self.title {
-                lock.as_mut().update_metadata(|m| {
+                conv.update_metadata(|m| {
                     m.title = Some(title.clone());
                 });
             }
 
+            // Write before reporting success, so a failed write is an error
+            // rather than a confirmation the user cannot trust.
+            conv.flush()?;
+            drop(conv);
+
             if self.activate
                 && let Some(session) = &ctx.session
                 && let Err(error) =
diff --git a/crates/jp_cli/src/cmd/query.rs b/crates/jp_cli/src/cmd/query.rs
index 83e92bd8..f434af7f 100644
--- a/crates/jp_cli/src/cmd/query.rs
+++ b/crates/jp_cli/src/cmd/query.rs
@@ -344,7 +344,6 @@ pub(crate) struct Query {
 }
 
 impl Query {
-    #[expect(clippy::too_many_lines)]
     pub(crate) async fn run(
         self,
         ctx: &mut Ctx,
@@ -353,8 +352,6 @@ impl Query {
     ) -> Output {
         debug!("Running `query` command.");
         trace!(args = ?self, "Received arguments.");
-        let now = ctx.now();
-        let cfg = ctx.config();
 
         // Resolve the target conversation and acquire an exclusive lock.
         //
@@ -365,6 +362,24 @@ impl Query {
         // 4. Lock contention: user picks "new" or "fork" from the prompt.
         let lock = self.acquire_lock(ctx, handle, start_new).await?;
 
+        let result = self.run_locked(ctx, &lock).await;
+
+        // Every exit from the locked region lands here, which is what makes
+        // this the one reliable drain point: a mutation scope that dropped
+        // while dirty recorded its persist failure on the lock, and any `?` in
+        // the region above returns past every other candidate site.
+        cmd::fold_persist_failure(result, lock.take_persist_failure())
+    }
+
+    /// Run the query against an already-locked conversation.
+    ///
+    /// Errors propagate freely: the caller drains any persist failure the
+    /// unwinding left behind.
+    #[expect(clippy::too_many_lines)]
+    async fn run_locked(self, ctx: &mut Ctx, lock: &ConversationLock) -> Output {
+        let now = ctx.now();
+        let cfg = ctx.config();
+
         // Create symlinks and seed approvals for any `--mount` flags before the
         // turn runs, so tools can reach the mounted paths.
         create_mount_effects(&self.mount, &ctx.workspace, ctx.fs_backend.as_deref(), now)?;
@@ -372,13 +387,13 @@ impl Query {
         // The two flags are mutually exclusive (enforced by clap), and the
         // resolved conversation may be new, freshly forked (which clones the
         // source's metadata, including any title), or resumed.
-        apply_title_override(&lock, self.title.as_deref(), self.no_title);
+        apply_title_override(lock, self.title.as_deref(), self.no_title);
 
         // Record this conversation as the session's active conversation.
         if let Some(session) = &ctx.session
             && let Err(error) = ctx
                 .workspace
-                .activate_session_conversation(&lock, session, now)
+                .activate_session_conversation(lock, session, now)
         {
             warn!(%error, "Failed to record activation.");
         }
@@ -400,7 +415,7 @@ impl Query {
 
         // Compact the conversation before querying, if requested.
         if self.compact.should_compact() {
-            self.apply_pre_query_compaction(&lock, &cfg).await?;
+            self.apply_pre_query_compaction(lock, &cfg).await?;
         }
 
         let mut mcp_servers_handle = ctx.configure_active_mcp_servers().await?;
@@ -499,21 +514,24 @@ impl Query {
             echo.render_user_request(&chat_request);
         }
 
+        // One mutable scope for the whole pre-turn setup. Every mutation below
+        // shares its single write at the closing `flush`, instead of each
+        // statement persisting the entire conversation on its own drop.
+        let mut setup = lock.as_mut();
+
         // Record the CLI-provided config delta (`--cfg`) now that the query is
         // known to be non-empty. Recording it before the empty-query check
         // would leave a config event behind for a query that was ultimately
         // ignored.
-        if let Some(delta) = get_config_delta_from_cli(&cfg, &lock)? {
-            lock.as_mut()
-                .update_events(|events| events.add_config_delta(delta));
+        if let Some(delta) = get_config_delta_from_cli(&cfg, lock)? {
+            setup.update_events(|events| events.add_config_delta(delta));
         }
 
         if !editor_provided_config.is_empty() {
             // Resolve any model aliases before storing in the stream so
             // that per-event configs always contain concrete model IDs.
             editor_provided_config.resolve_model_aliases(&cfg.providers.llm.aliases);
-            lock.as_mut()
-                .update_events(|events| events.add_config_delta(editor_provided_config));
+            setup.update_events(|events| events.add_config_delta(editor_provided_config));
         }
 
         // Snapshot the stream for title generation and thread assembly. The
@@ -522,7 +540,7 @@ impl Query {
         // stream is only trimmed at the turn-start commit point), the new
         // request not yet appended.
         let stream = {
-            let mut stream = lock.events().clone();
+            let mut stream = setup.events().clone();
             pending_trim.apply(&mut stream);
             stream
         };
@@ -546,8 +564,7 @@ impl Query {
             ) {
                 NewTitle::FromHeading(title) => {
                     debug!("Using leading markdown heading as conversation title");
-                    lock.as_mut()
-                        .update_metadata(|m| m.title = Some(title.clone()));
+                    setup.update_metadata(|m| m.title = Some(title.clone()));
                     if ctx.term.is_tty {
                         jp_term::osc::set_title(format!("{cid}: {title}"));
                     }
@@ -595,14 +612,20 @@ impl Query {
 
         // Sanitize any structural issues (orphaned tool calls, missing
         // user messages, etc.) before sending the stream to the provider.
-        lock.as_mut().update_events(ConversationStream::sanitize);
+        setup.update_events(ConversationStream::sanitize);
+
+        // Commit the setup phase before the turn starts. Errors propagate here
+        // rather than being swallowed by a drop, and the turn loop's own
+        // checkpoints take over from this point.
+        setup.flush()?;
+        drop(setup);
 
         let invocation = InvocationContext {
             workspace_id: ctx.workspace.id().to_string(),
             conversation_id: lock.id().to_string(),
         };
 
-        let turn_result = self
+        let mut turn_result = self
             .handle_turn(
                 &cfg,
                 &ctx.signals,
@@ -610,7 +633,7 @@ impl Query {
                 root,
                 ctx.term.is_tty,
                 &thread.attachments,
-                &lock,
+                lock,
                 cfg.assistant.tool_choice.clone(),
                 &tools,
                 ctx.printer.clone(),
@@ -622,6 +645,19 @@ impl Query {
             .await
             .map_err(|error| cmd::Error::from(error).with_persistence(true));
 
+        // Fold a drop-time persist failure into the turn's result before
+        // anything downstream treats the turn as a success. Both gates below
+        // key off `is_ok`, and the scratch file one of them deletes is the
+        // user's only recovery copy of a request that was never saved.
+        //
+        // `run` drains again for the exits that never reach this point; the
+        // failure is only ever handed out once.
+        if turn_result.is_ok()
+            && let Some(error) = lock.take_persist_failure()
+        {
+            turn_result = Err(cmd::Error::from(Error::Workspace(error)));
+        }
+
         // Extract structured data from the conversation after the turn.
         if self.schema.is_some() && turn_result.is_ok() {
             let data = lock.events().iter().rev().find_map(|e| {
diff --git a/crates/jp_cli/src/cmd_tests.rs b/crates/jp_cli/src/cmd_tests.rs
new file mode 100644
index 00000000..9f7cf842
--- /dev/null
+++ b/crates/jp_cli/src/cmd_tests.rs
@@ -0,0 +1,142 @@
+use std::io;
+
+use camino::Utf8Path;
+use test_log::test;
+
+use super::*;
+
+/// `ENOSPC` on every Unix platform this crate targets.
+///
+/// These tests assert the exact text the user sees, which includes the OS's own
+/// message, so they build the error from the raw code the kernel delivers
+/// rather than naming [`io::ErrorKind::StorageFull`] directly.
+#[cfg(unix)]
+const ENOSPC: i32 = 28;
+
+/// Render a CLI error the way the terminal output does: the message, then each
+/// metadata entry as `key=value`.
+fn rendered(error: &Error) -> String {
+    let mut lines = vec![error.message.clone().unwrap_or_default()];
+    for (key, value) in &error.metadata {
+        lines.push(format!("{key}={}", value.as_str().unwrap_or_default()));
+    }
+    lines.join("\n")
+}
+
+/// A drop-time persist failure as the workspace records one.
+fn persist_failure() -> jp_workspace::Error {
+    jp_workspace::Error::Storage(jp_storage::Error::write_failed(
+        Utf8Path::new("/data/conv/events.json"),
+        io::Error::from(io::ErrorKind::StorageFull),
+    ))
+}
+
+#[test]
+fn a_clean_run_passes_through_unchanged() {
+    assert!(fold_persist_failure(Ok(()), None).is_ok());
+}
+
+#[test]
+fn a_persist_failure_fails_an_otherwise_successful_run() {
+    let error =
+        fold_persist_failure(Ok(()), Some(persist_failure())).expect_err("must not exit zero");
+
+    assert_eq!(error.message.as_deref(), Some("No space left on device"));
+    assert_eq!(error.code.get(), 1);
+}
+
+#[test]
+fn a_persist_failure_rides_along_with_an_existing_error() {
+    // The two can be independent — a provider error and a full disk — so the
+    // primary error stays the headline and the persist failure is attached
+    // rather than dropped.
+    let error = fold_persist_failure(Err(Error::from("Rate limited")), Some(persist_failure()))
+        .expect_err("the primary error is preserved");
+
+    assert_eq!(error.message.as_deref(), Some("Rate limited"));
+    assert_eq!(
+        error
+            .metadata
+            .iter()
+            .find(|(key, _)| key == "persist_failure")
+            .map(|(_, value)| value.as_str().unwrap_or_default()),
+        Some("Storage error: no space left on device while writing /data/conv/events.json")
+    );
+}
+
+#[test]
+fn an_existing_error_survives_when_nothing_was_recorded() {
+    let error = fold_persist_failure(Err(Error::from("Rate limited")), None)
+        .expect_err("the primary error is preserved");
+
+    assert_eq!(error.message.as_deref(), Some("Rate limited"));
+    assert!(error.metadata.is_empty());
+}
+
+#[cfg(unix)]
+#[test]
+fn out_of_space_renders_path_cause_and_action() {
+    let error = Error::from(jp_storage::Error::write_failed(
+        Utf8Path::new("/data/conv/events.json"),
+        io::Error::from_raw_os_error(ENOSPC),
+    ));
+
+    assert_eq!(
+        rendered(&error),
+        "No space left on device\npath=/data/conv/events.json\nerror=No space left on device (os \
+         error 28)\nsuggestion=Free up disk space and re-run. Anything after the last successful \
+         write was not saved."
+    );
+    assert_eq!(error.code.get(), 1);
+}
+
+#[test]
+fn write_failure_renders_the_path_and_the_os_error() {
+    let error = Error::from(jp_storage::Error::write_failed(
+        Utf8Path::new("/data/conv/events.json"),
+        io::Error::from(io::ErrorKind::PermissionDenied),
+    ));
+
+    assert_eq!(
+        rendered(&error),
+        "Failed to write file\npath=/data/conv/events.json\nerror=permission denied"
+    );
+}
+
+#[cfg(unix)]
+#[test]
+fn a_workspace_persist_failure_keeps_the_full_cause_chain() {
+    // The path a failed drop-time persist takes to the terminal: storage error
+    // wrapped by the workspace, wrapped by the CLI. Each layer has to keep the
+    // one below it renderable, or the user is told "IO error" and nothing else.
+    let error = Error::from(crate::error::Error::Workspace(
+        jp_workspace::Error::Storage(jp_storage::Error::write_failed(
+            Utf8Path::new("/data/conv/events.json"),
+            io::Error::from_raw_os_error(ENOSPC),
+        )),
+    ));
+
+    assert_eq!(
+        rendered(&error),
+        "No space left on device\npath=/data/conv/events.json\nerror=No space left on device (os \
+         error 28)\nsuggestion=Free up disk space and re-run. Anything after the last successful \
+         write was not saved."
+    );
+}
+
+#[cfg(unix)]
+#[test]
+fn a_bare_io_error_renders_without_a_path() {
+    // `Error::Io` has no path to name, so it renders through the generic
+    // cause-chain walk: an "IO error" headline plus the OS message. Pinned to
+    // show what a caller gives up by not routing through
+    // `jp_storage::Error::write_failed`.
+    let error = Error::from(jp_storage::Error::from(io::Error::from_raw_os_error(
+        ENOSPC,
+    )));
+
+    assert_eq!(
+        rendered(&error),
+        "IO error\n=No space left on device (os error 28)"
+    );
+}
diff --git a/crates/jp_cli/src/lib.rs b/crates/jp_cli/src/lib.rs
index e94db9ca..3ff7f145 100644
--- a/crates/jp_cli/src/lib.rs
+++ b/crates/jp_cli/src/lib.rs
@@ -466,6 +466,15 @@ fn run_inner(cli: Cli, format: OutputFormat) -> Result<()> {
         }
     });
 
+    // The shutdown arm above drops the command future, so a conversation scope
+    // that was dirty persists from its `Drop` and records any failure on the
+    // workspace — with the command's own drain gone along with the future.
+    // Draining here, before the `disable_persistence` check below, is what lets
+    // an interrupted unsaved run still say so.
+    // Commands that reported already left nothing behind: the record yields
+    // each failure once.
+    let output = cmd::fold_persist_failure(output, ctx.workspace.take_persist_failure());
+
     if let Err(error) = output.as_ref()
         && error.disable_persistence
     {
@@ -491,8 +500,14 @@ fn run_inner(cli: Cli, format: OutputFormat) -> Result<()> {
     // shutdown request (Ctrl-C, an interrupt earlier in the run, or SIGTERM)
     // switches to a 2s cancellation countdown; any further Ctrl-C exits the
     // process immediately via the signal router's escalation ladder.
-    rt.block_on(drain_background_tasks(&mut ctx))
-        .map_err(Error::Task)?;
+    let drained = rt.block_on(drain_background_tasks(&mut ctx));
+
+    // Task sync takes conversation locks of its own — the title generator
+    // persists its result — so this is the run's last write, after the drain
+    // above. Held separately from the drain's own error so a failing task and an
+    // unsaved conversation are not reported as the same thing.
+    let output = cmd::fold_persist_failure(output, ctx.workspace.take_persist_failure());
+    drained.map_err(Error::Task)?;
 
     // Remove ephemeral conversations that are no longer needed, but protect
     // any conversation that is active in a terminal session.
diff --git a/crates/jp_cli/src/lib_tests.rs b/crates/jp_cli/src/lib_tests.rs
index e715b031..a8499e6d 100644
--- a/crates/jp_cli/src/lib_tests.rs
+++ b/crates/jp_cli/src/lib_tests.rs
@@ -64,6 +64,143 @@ fn build_cfg(
     pipeline.partial_without_conversation()
 }
 
+/// Persistence backend whose every write fails with a full disk.
+#[derive(Debug)]
+struct AlwaysFullBackend;
+
+impl jp_storage::backend::PersistBackend for AlwaysFullBackend {
+    fn write(
+        &self,
+        _id: &ConversationId,
+        _metadata: &Conversation,
+        _events: &jp_conversation::ConversationStream,
+        _projection: jp_storage::backend::Projection,
+    ) -> std::result::Result<(), jp_storage::Error> {
+        Err(jp_storage::Error::write_failed(
+            camino::Utf8Path::new("/data/conv/events.json"),
+            std::io::Error::from(std::io::ErrorKind::StorageFull),
+        ))
+    }
+
+    fn remove(&self, _id: &ConversationId) -> std::result::Result<(), jp_storage::Error> {
+        Ok(())
+    }
+
+    fn archive(&self, _id: &ConversationId) -> std::result::Result<(), jp_storage::Error> {
+        Ok(())
+    }
+
+    fn unarchive(&self, _id: &ConversationId) -> std::result::Result<(), jp_storage::Error> {
+        Ok(())
+    }
+}
+
+#[test]
+fn a_cancelled_command_future_still_reports_its_persist_failure() {
+    // Mirrors `run_inner`'s shutdown arm: on Ctrl-C the command future is
+    // dropped mid-flight, so the drain inside it never runs. The command arm
+    // here parks forever after dirtying the conversation, which holds the
+    // window open — cancellation always lands while the scope is still alive,
+    // rather than racing a sleep.
+    let rt = tokio::runtime::Builder::new_current_thread()
+        .build()
+        .unwrap();
+    let mut workspace = Workspace::new("root").with_persist(Arc::new(AlwaysFullBackend));
+    let lock = workspace
+        .create_and_lock_conversation(
+            Conversation::default(),
+            Arc::new(AppConfig::new_test()),
+            None,
+        )
+        .unwrap();
+
+    let output: cmd::Output = rt.block_on(async {
+        tokio::select! {
+            biased;
+            () = async {
+                let conv = lock.as_mut();
+                conv.update_metadata(|m| m.title = Some("unsaved".into()));
+                std::future::pending::<()>().await;
+            } => unreachable!("the command arm never completes"),
+            () = std::future::ready(()) => Err(cmd::Error::interrupted()),
+        }
+    });
+    drop(lock);
+
+    let error = cmd::fold_persist_failure(output, workspace.take_persist_failure())
+        .expect_err("the run was interrupted");
+
+    // The interrupt stays the headline and keeps its exit code; the fact that
+    // nothing was saved rides along instead of being lost to the log file.
+    assert_eq!(error.message.as_deref(), Some("Interrupted"));
+    assert_eq!(error.code.get(), 130);
+    assert_eq!(
+        error
+            .metadata
+            .iter()
+            .find(|(key, _)| key == "persist_failure")
+            .map(|(_, value)| value.as_str().unwrap_or_default()),
+        Some("Storage error: no space left on device while writing /data/conv/events.json")
+    );
+}
+
+#[test]
+fn a_background_task_persist_failure_is_recorded_after_the_command_finished() {
+    // `TitleGeneratorTask::sync` takes a conversation lock of its own and
+    // swallows a failed `flush()` into a `warn!`. The still-dirty scope's drop
+    // then records on the workspace, which happens while background tasks are
+    // draining — after the command's own drain has already run. That is why
+    // `run_inner` folds a second time once the drain returns.
+    let rt = tokio::runtime::Builder::new_current_thread()
+        .build()
+        .unwrap();
+    let mut workspace = Workspace::new("root").with_persist(Arc::new(AlwaysFullBackend));
+
+    let lock = workspace
+        .create_and_lock_conversation(
+            Conversation::default(),
+            Arc::new(AppConfig::new_test()),
+            None,
+        )
+        .unwrap();
+    let id = lock.id();
+    // `sync` acquires the lock itself and skips when it is already held, which
+    // would make every assertion below pass vacuously.
+    drop(lock);
+    assert!(
+        workspace.take_persist_failure().is_none(),
+        "setup must record nothing, or the failure asserted below proves nothing"
+    );
+
+    let cfg = AppConfig::new_test();
+    let task = Box::new(jp_task::task::TitleGeneratorTask {
+        conversation_id: id,
+        model_id: cfg.assistant.model.id.resolved().clone(),
+        providers: cfg.providers.llm.clone(),
+        events: jp_conversation::ConversationStream::new_test(),
+        title: Some("generated".into()),
+        is_tty: false,
+    });
+
+    // `sync` reports success: it only logs the write failure.
+    rt.block_on(jp_task::Task::sync(task, &mut workspace))
+        .expect("sync swallows the write failure");
+
+    let recorded = workspace
+        .take_persist_failure()
+        .expect("the title task's failed write is recorded on the workspace");
+    assert_eq!(
+        recorded.to_string(),
+        "Storage error: no space left on device while writing /data/conv/events.json"
+    );
+
+    // What the post-drain fold makes of it: a run that would otherwise have
+    // exited zero now carries the failure.
+    let error = cmd::fold_persist_failure(Ok(()), Some(recorded))
+        .expect_err("an unsaved conversation must not exit zero");
+    assert_eq!(error.message.as_deref(), Some("No space left on device"));
+}
+
 #[test]
 fn tracing_guard_persist_returns_explicit_log_file_path() {
     // `--log-file <path>`: the file lives wherever the caller put it; persist
diff --git a/crates/jp_storage/src/error.rs b/crates/jp_storage/src/error.rs
index 007ec986..2e2316f7 100644
--- a/crates/jp_storage/src/error.rs
+++ b/crates/jp_storage/src/error.rs
@@ -1,3 +1,5 @@
+use std::io;
+
 use camino::Utf8PathBuf;
 use jp_conversation::ConversationId;
 
@@ -17,9 +19,32 @@ pub enum Error {
     #[error("configuration error")]
     Config(#[from] jp_config::error::Error),
 
-    #[error("IO error")]
+    #[error(transparent)]
     Io(#[from] std::io::Error),
 
+    /// A write to `path` failed.
+    ///
+    /// `std::io::Error` carries no path, so a bare I/O failure cannot tell the
+    /// user which file it was.
+    #[error("failed to write {path}")]
+    WriteFailed {
+        path: Utf8PathBuf,
+        #[source]
+        source: io::Error,
+    },
+
+    /// The filesystem holding `path` has no room left.
+    ///
+    /// Separate from [`Error::WriteFailed`] because it is the one write failure
+    /// with an obvious user action, and because no retry can succeed until
+    /// space is freed.
+    #[error("no space left on device while writing {path}")]
+    OutOfSpace {
+        path: Utf8PathBuf,
+        #[source]
+        source: io::Error,
+    },
+
     #[error("invalid JSON data")]
     Json(#[from] serde_json::Error),
 
@@ -30,6 +55,31 @@ pub enum Error {
     ConversationNotFound(ConversationId),
 }
 
+impl Error {
+    /// Classify a failed write to `path`.
+    ///
+    /// Yields [`Error::OutOfSpace`] when the operating system reported a full
+    /// filesystem, and [`Error::WriteFailed`] otherwise.
+    #[must_use]
+    pub fn write_failed(path: impl Into<Utf8PathBuf>, source: io::Error) -> Self {
+        let path = path.into();
+        if is_storage_full(&source) {
+            Self::OutOfSpace { path, source }
+        } else {
+            Self::WriteFailed { path, source }
+        }
+    }
+}
+
+/// Whether an OS error reports a full filesystem.
+///
+/// Recognises both a raw OS error (`ENOSPC`, `ERROR_DISK_FULL`) and an error
+/// constructed directly from the kind, since `std` maps the platform codes onto
+/// [`io::ErrorKind::StorageFull`].
+fn is_storage_full(error: &io::Error) -> bool {
+    error.kind() == io::ErrorKind::StorageFull
+}
+
 #[cfg(test)]
 impl PartialEq for Error {
     fn eq(&self, other: &Self) -> bool {
@@ -41,3 +91,7 @@ impl PartialEq for Error {
         format!("{self:?}") == format!("{other:?}")
     }
 }
+
+#[cfg(test)]
+#[path = "error_tests.rs"]
+mod tests;
diff --git a/crates/jp_storage/src/error_tests.rs b/crates/jp_storage/src/error_tests.rs
new file mode 100644
index 00000000..daf8fbf8
--- /dev/null
+++ b/crates/jp_storage/src/error_tests.rs
@@ -0,0 +1,65 @@
+use camino::Utf8Path;
+
+use super::*;
+
+/// `ENOSPC` on every Unix platform this crate targets.
+///
+/// Building the error from the raw code keeps the classification tests honest:
+/// they run against what the kernel actually delivers, not against an error
+/// synthesized from the very [`io::ErrorKind`] the classifier reads back.
+#[cfg(unix)]
+const ENOSPC: i32 = 28;
+
+/// An out-of-space error as the operating system reports it.
+#[cfg(unix)]
+fn os_disk_full() -> io::Error {
+    io::Error::from_raw_os_error(ENOSPC)
+}
+
+#[cfg(not(unix))]
+fn os_disk_full() -> io::Error {
+    io::Error::from(io::ErrorKind::StorageFull)
+}
+
+#[test]
+fn write_failed_classifies_full_disk_as_out_of_space() {
+    let error = Error::write_failed(Utf8Path::new("/data/conv/events.json"), os_disk_full());
+
+    assert!(matches!(error, Error::OutOfSpace { .. }));
+    assert_eq!(
+        error.to_string(),
+        "no space left on device while writing /data/conv/events.json"
+    );
+}
+
+#[test]
+fn write_failed_classifies_other_errors_as_write_failed() {
+    let error = Error::write_failed(
+        Utf8Path::new("/data/conv/events.json"),
+        io::Error::from(io::ErrorKind::PermissionDenied),
+    );
+
+    assert!(matches!(error, Error::WriteFailed { .. }));
+    assert_eq!(error.to_string(), "failed to write /data/conv/events.json");
+}
+
+#[test]
+fn write_failed_keeps_the_os_error_as_the_source() {
+    let error = Error::write_failed(Utf8Path::new("/data/conv/events.json"), os_disk_full());
+
+    let source = std::error::Error::source(&error)
+        .and_then(|source| source.downcast_ref::<io::Error>())
+        .expect("the os error is retained as the source, unwrapped");
+    assert_eq!(source.kind(), io::ErrorKind::StorageFull);
+}
+
+#[cfg(unix)]
+#[test]
+fn io_variant_renders_the_os_error_verbatim() {
+    // `Error::Io` is transparent: a bare I/O failure must not be flattened into
+    // an opaque "IO error", or the cause is lost wherever only `Display` is
+    // rendered.
+    let error = Error::from(os_disk_full());
+
+    assert_eq!(error.to_string(), "No space left on device (os error 28)");
+}
diff --git a/crates/jp_storage/src/lib.rs b/crates/jp_storage/src/lib.rs
index 9c93ba78..a2ae350f 100644
--- a/crates/jp_storage/src/lib.rs
+++ b/crates/jp_storage/src/lib.rs
@@ -27,8 +27,39 @@ pub(crate) const METADATA_FILE: &str = "metadata.json";
 const EVENTS_FILE: &str = "events.json";
 const BASE_CONFIG_FILE: &str = "base_config.json";
 pub(crate) const CONVERSATIONS_DIR: &str = "conversations";
+
+/// Name prefix for a conversation directory being staged by an import.
+///
+/// Dot-prefixed so a leftover is invisible to conversation lookups and index
+/// scans, both of which match on the bare conversation dirname.
+const IMPORT_STAGING_PREFIX: &str = ".import-";
 pub(crate) const ARCHIVE_DIR: &str = ".archive";
 
+/// Build the staging directory name an import copies into.
+///
+/// The name is unique per attempt, so a copy can never land in a tree left by
+/// an earlier one.
+/// That matters because a merge would publish entries the source no longer has:
+/// `copy_dir_all` overwrites the files it copies but removes nothing, so
+/// anything surviving from an older generation would be renamed into place
+/// alongside the fresh copy.
+///
+/// The uniqueness suffix goes *after* the conversation dirname so the leading
+/// timestamp segment stays intact.
+/// The sanitize sweep identifies a leftover by parsing that segment, and can
+/// only take the conversation's lock (and so only reap the directory) while it
+/// stays parseable.
+fn import_staging_dirname(dirname: &str) -> String {
+    let nanos = SystemTime::now()
+        .duration_since(SystemTime::UNIX_EPOCH)
+        .map_or(0, |d| d.as_nanos());
+
+    format!(
+        "{IMPORT_STAGING_PREFIX}{dirname}-{}-{nanos}",
+        std::process::id()
+    )
+}
+
 #[derive(Debug, Clone)]
 struct Storage {
     /// The path to the original storage directory.
@@ -237,7 +268,7 @@ impl Storage {
         // Bring any existing copy to the current directory name (e.g. after a
         // title change) and drop other stale copies for this id.
         reconcile_conversation_dir(id, conversations_dir, &conv_dir)?;
-        fs::create_dir_all(&conv_dir)?;
+        fs::create_dir_all(&conv_dir).map_err(|error| Error::write_failed(&conv_dir, error))?;
 
         let (base_config, events_json) = events
             .to_parts()
@@ -821,6 +852,10 @@ fn remove_conversation_dirs(id: &ConversationId, conversations_dir: &Utf8Path) -
 /// name the upcoming write will use.
 /// A conversation already present in user-local, or one with no workspace copy,
 /// is left untouched.
+///
+/// The copy lands in a staging directory and is renamed into place, so the
+/// import is all-or-nothing: a failed or killed run leaves nothing that later
+/// runs can mistake for an imported conversation.
 fn import_external_copy(
     id: &ConversationId,
     title: Option<&str>,
@@ -836,16 +871,46 @@ fn import_external_copy(
         return Ok(());
     };
 
-    fs::create_dir_all(user_conversations)?;
-    copy_dir_all(
-        &workspace_conv,
-        &user_conversations.join(id.to_dirname(title)),
-    )
+    fs::create_dir_all(user_conversations)
+        .map_err(|error| Error::write_failed(user_conversations, error))?;
+
+    // Copying straight to the final name would leave a partial directory behind
+    // on failure. The next run then finds it, treats the import as already done
+    // and skips it forever, while its fresh mtime makes the loader prefer it
+    // over the intact workspace copy. Staging plus a rename makes the visible
+    // result atomic.
+    //
+    // The staging name is unique per attempt, so this copy cannot land in a tree
+    // an earlier attempt left behind. Reusing one name and clearing it first
+    // would make correctness depend on that removal succeeding: a partial
+    // failure leaves entries the source no longer has, and the rename publishes
+    // them.
+    let dirname = id.to_dirname(title);
+    let staging = user_conversations.join(import_staging_dirname(&dirname));
+    let target = user_conversations.join(&dirname);
+
+    // Both cleanups are best-effort: a leftover under a unique name is inert.
+    // It is never copied into, `find_normal_conversation_dir_path` cannot match
+    // it, and the sanitize sweep reaps it under the conversation's lock.
+    if let Err(error) = copy_dir_all(&workspace_conv, &staging) {
+        let _err = fs::remove_dir_all(&staging);
+        return Err(error);
+    }
+
+    if let Err(error) = fs::rename(&staging, &target) {
+        let _err = fs::remove_dir_all(&staging);
+        return Err(Error::write_failed(&target, error));
+    }
+
+    Ok(())
 }
 
 /// Recursively copy directory `src` into `dst`.
+///
+/// A failure names the destination path it was writing, and reports a full
+/// filesystem as [`Error::OutOfSpace`].
 fn copy_dir_all(src: &Utf8Path, dst: &Utf8Path) -> Result<()> {
-    fs::create_dir_all(dst)?;
+    fs::create_dir_all(dst).map_err(|error| Error::write_failed(dst, error))?;
     for entry in dir_entries(src) {
         let to = dst.join(entry.file_name());
         let is_dir = entry.file_type().is_ok_and(|ty| ty.is_dir());
@@ -853,7 +918,7 @@ fn copy_dir_all(src: &Utf8Path, dst: &Utf8Path) -> Result<()> {
         if is_dir {
             copy_dir_all(&from, &to)?;
         } else {
-            fs::copy(&from, &to)?;
+            fs::copy(&from, &to).map_err(|error| Error::write_failed(&to, error))?;
         }
     }
     Ok(())
@@ -939,7 +1004,7 @@ fn reconcile_conversation_dir(
             .into_iter()
             .find(|dir| dir != target)
     {
-        fs::rename(&src, target)?;
+        fs::rename(&src, target).map_err(|error| Error::write_failed(target, error))?;
     }
 
     for dir in conversation_dirs_for_id(conversations_dir, &prefix) {
diff --git a/crates/jp_storage/src/lib_tests.rs b/crates/jp_storage/src/lib_tests.rs
index 375f73e6..e47b791d 100644
--- a/crates/jp_storage/src/lib_tests.rs
+++ b/crates/jp_storage/src/lib_tests.rs
@@ -36,6 +36,109 @@ fn test_storage_new_errors_on_source_file() {
     }
 }
 
+#[cfg(unix)]
+#[test]
+fn a_failed_import_leaves_nothing_that_looks_like_a_conversation() {
+    let tmp = tempdir().unwrap();
+    let workspace = tmp.path().join("ws").join("conversations");
+    let user = tmp.path().join("user").join("conversations");
+    let id = ConversationId::from_str("jp-c17457886043-otvo8").unwrap();
+    let prefix = id.to_dirname(None);
+
+    let src = workspace.join(&prefix);
+    fs::create_dir_all(&src).unwrap();
+    fs::write(src.join("events.json"), "[]").unwrap();
+    fs::write(src.join("notes.md"), "keep me").unwrap();
+    // A dangling symlink makes exactly one `fs::copy` fail mid-directory, the
+    // way a disk filling up would, without having to fill a disk.
+    std::os::unix::fs::symlink("nowhere", src.join("dangling")).unwrap();
+
+    let error = import_external_copy(&id, Some("title"), &workspace, &user)
+        .expect_err("the copy fails on the dangling entry");
+    assert!(matches!(error, Error::WriteFailed { .. }));
+
+    // The poisoning this guards against: a partial directory under the real
+    // name makes every later run skip the import, and its fresh mtime makes the
+    // loader prefer it over the intact workspace copy.
+    assert!(
+        find_normal_conversation_dir_path(&user, &prefix).is_none(),
+        "a failed import must not leave a conversation directory"
+    );
+    assert_eq!(
+        dir_entries(&user).count(),
+        0,
+        "the staging directory is cleaned up too"
+    );
+
+    // Clear the obstruction: the retry now completes, non-managed file included.
+    fs::remove_file(src.join("dangling")).unwrap();
+    import_external_copy(&id, Some("title"), &workspace, &user).expect("the retry succeeds");
+
+    let imported =
+        find_normal_conversation_dir_path(&user, &prefix).expect("the conversation is imported");
+    assert_eq!(
+        fs::read_to_string(imported.join("notes.md")).unwrap(),
+        "keep me"
+    );
+}
+
+#[test]
+fn an_import_never_merges_into_a_leftover_staging_tree() {
+    // Debris from a crashed import can survive: the error-path cleanup is
+    // best-effort, and `remove_dir_all` can partially fail (a read-only file on
+    // Windows is enough). `copy_dir_all` overwrites the files it copies but
+    // removes nothing, so a merge would publish entries the workspace copy no
+    // longer has. Staging under a fresh name per attempt is what rules that out.
+    let tmp = tempdir().unwrap();
+    let workspace = tmp.path().join("ws").join("conversations");
+    let user = tmp.path().join("user").join("conversations");
+    let id = ConversationId::from_str("jp-c17457886043-otvo8").unwrap();
+    let prefix = id.to_dirname(None);
+
+    let src = workspace.join(&prefix);
+    fs::create_dir_all(&src).unwrap();
+    fs::write(src.join("events.json"), "[]").unwrap();
+
+    // A stale tree holding a file the source does not have, under the name a
+    // single fixed staging directory would have used.
+    let stale = user.join(format!(".import-{}", id.to_dirname(Some("title"))));
+    fs::create_dir_all(&stale).unwrap();
+    fs::write(stale.join("stale.md"), "from an older generation").unwrap();
+
+    import_external_copy(&id, Some("title"), &workspace, &user).expect("the import succeeds");
+
+    let imported =
+        find_normal_conversation_dir_path(&user, &prefix).expect("the conversation is imported");
+    assert!(
+        !imported.join("stale.md").exists(),
+        "the published conversation must hold only what the source has"
+    );
+    assert!(imported.join("events.json").is_file());
+}
+
+#[test]
+fn copy_dir_all_failure_names_the_destination_file() {
+    // The import leg of a persist copies whole conversation directories, so a
+    // full disk can fail here rather than in `write_json`. Without the path the
+    // user is told only "IO error".
+    let tmp = tempdir().unwrap();
+    let src = tmp.path().join("src");
+    let dst = tmp.path().join("dst");
+    fs::create_dir_all(&src).unwrap();
+    fs::write(src.join("events.json"), "[]").unwrap();
+
+    // A directory where the copied file must land fails on all platforms.
+    let blocker = dst.join("events.json");
+    fs::create_dir_all(&blocker).unwrap();
+
+    let error = copy_dir_all(&src, &dst).expect_err("copying onto a directory should fail");
+
+    match error {
+        Error::WriteFailed { path, .. } => assert_eq!(path, blocker),
+        other => panic!("expected WriteFailed, got {other:?}"),
+    }
+}
+
 #[test]
 fn test_conversation_dir_name_generation() {
     let id = ConversationId::from_str("jp-c17457886043-otvo8").unwrap();
diff --git a/crates/jp_storage/src/validate.rs b/crates/jp_storage/src/validate.rs
index 03edac8b..47364dd1 100644
--- a/crates/jp_storage/src/validate.rs
+++ b/crates/jp_storage/src/validate.rs
@@ -6,8 +6,8 @@ use rayon::prelude::*;
 use tracing::{debug, trace};
 
 use crate::{
-    BASE_CONFIG_FILE, CONVERSATIONS_DIR, EVENTS_FILE, METADATA_FILE, Storage, dir_entries,
-    trash::trash_conversation, value::cleanup_tmp_files,
+    BASE_CONFIG_FILE, CONVERSATIONS_DIR, EVENTS_FILE, IMPORT_STAGING_PREFIX, METADATA_FILE,
+    Storage, dir_entries, trash::trash_conversation, value::cleanup_tmp_files,
 };
 
 /// Result of validating all conversation directories across storage roots.
@@ -128,6 +128,7 @@ impl Storage {
                 continue;
             }
 
+            self.cleanup_import_staging(&conversations_dir);
             validate_root(&conversations_dir, &mut result);
         }
 
@@ -139,6 +140,49 @@ impl Storage {
 
         result
     }
+
+    /// Remove leftover import staging directories from one `conversations/`
+    /// directory.
+    ///
+    /// A staging directory only survives a crash between an import's copy and
+    /// its rename, and holds nothing the workspace copy does not still have, so
+    /// removing one is pure disk hygiene: each import stages under a fresh name
+    /// and never reads a leftover.
+    ///
+    /// Each removal is made while holding the conversation's own lock.
+    /// Without it, this would race an import running under that lock in another
+    /// process: `copy_dir_all` recreates directories as it recurses, so a
+    /// mid-copy removal would leave the importer to rename an incomplete tree
+    /// into place.
+    /// A conversation whose lock is held is skipped entirely, and because the
+    /// sweep is only reclaiming disk space, skipping costs nothing but a delay
+    /// until the next run.
+    fn cleanup_import_staging(&self, conversations_dir: &Utf8Path) {
+        for entry in dir_entries(conversations_dir) {
+            let Some(dirname) = entry.file_name().strip_prefix(IMPORT_STAGING_PREFIX) else {
+                continue;
+            };
+            if !entry.path().is_dir() {
+                continue;
+            }
+
+            // An unparseable name cannot be tied to a lock, so leave it be
+            // rather than remove a directory whose owner is unknown.
+            let Ok(id) = ConversationId::try_from_dirname(dirname) else {
+                trace!(path = %entry.path(), "Skipping unparseable staging directory.");
+                continue;
+            };
+
+            let Ok(Some(_guard)) = self.try_lock_conversation(&id.to_string(), None) else {
+                trace!(%id, "Skipping staging directory: conversation is locked.");
+                continue;
+            };
+
+            let path = entry.into_path();
+            trace!(%path, "Removing leftover import staging directory.");
+            let _err = std::fs::remove_dir_all(path);
+        }
+    }
 }
 
 /// Trash a conversation that failed validation.
diff --git a/crates/jp_storage/src/validate_tests.rs b/crates/jp_storage/src/validate_tests.rs
index 71ee2a38..d7d0bddf 100644
--- a/crates/jp_storage/src/validate_tests.rs
+++ b/crates/jp_storage/src/validate_tests.rs
@@ -20,6 +20,75 @@ fn write_valid(storage: &camino::Utf8Path, id: &ConversationId) {
     write_json(&dir.join(EVENTS_FILE), &events).unwrap();
 }
 
+#[test]
+fn staging_dirs_are_swept_only_when_the_conversation_is_unlocked() {
+    // The sweep runs at startup in every process, while another process may be
+    // importing that same conversation under its lock. Reaping mid-copy would
+    // let the importer rename an incomplete tree into place, so a held lock has
+    // to make the sweep skip. Driven by taking the lock in-process rather than
+    // racing two imports, so the outcome is deterministic.
+    let tmp = tempdir().unwrap();
+    let storage = Storage::new(tmp.path()).unwrap();
+    let conversations = tmp.path().join(CONVERSATIONS_DIR);
+
+    let id = ConversationId::try_from_deciseconds_str("17636257526").unwrap();
+    // Built through the production helper: its per-attempt uniqueness suffix is
+    // exactly what the sweep has to still recognise and be able to lock.
+    let staging = conversations.join(crate::import_staging_dirname(&id.to_dirname(Some("title"))));
+    fs::create_dir_all(staging.join("partial")).unwrap();
+
+    let guard = storage
+        .try_lock_conversation(&id.to_string(), None)
+        .unwrap()
+        .expect("the lock is free in this test");
+
+    drop(storage.validate_conversations());
+    assert!(
+        staging.is_dir(),
+        "an in-flight import's staging directory must survive the sweep"
+    );
+
+    drop(guard);
+
+    drop(storage.validate_conversations());
+    assert!(
+        !staging.exists(),
+        "an abandoned staging directory is reaped once nobody holds the lock"
+    );
+}
+
+#[test]
+fn the_sweep_leaves_real_conversations_and_dot_dirs_alone() {
+    let tmp = tempdir().unwrap();
+    let storage = Storage::new(tmp.path()).unwrap();
+    let conversations = tmp.path().join(CONVERSATIONS_DIR);
+
+    let id = ConversationId::try_from_deciseconds_str("17636257526").unwrap();
+    write_valid(tmp.path(), &id);
+    fs::create_dir_all(conversations.join(".archive")).unwrap();
+
+    drop(storage.validate_conversations());
+
+    assert!(conversations.join(id.to_dirname(None)).is_dir());
+    assert!(conversations.join(".archive").is_dir());
+}
+
+#[test]
+fn a_staging_dir_with_an_unparseable_name_is_left_in_place() {
+    // Without a parseable id there is no lock to check, so removing it could
+    // race an owner this code cannot identify.
+    let tmp = tempdir().unwrap();
+    let storage = Storage::new(tmp.path()).unwrap();
+    let conversations = tmp.path().join(CONVERSATIONS_DIR);
+
+    let staging = conversations.join(".import-not-an-id");
+    fs::create_dir_all(&staging).unwrap();
+
+    drop(storage.validate_conversations());
+
+    assert!(staging.is_dir());
+}
+
 #[test]
 fn test_valid_conversations_are_collected() {
     let tmp = tempdir().unwrap();
diff --git a/crates/jp_storage/src/value.rs b/crates/jp_storage/src/value.rs
index 7519fb33..a6608afc 100644
--- a/crates/jp_storage/src/value.rs
+++ b/crates/jp_storage/src/value.rs
@@ -4,7 +4,7 @@ use camino::{Utf8Path, Utf8PathBuf};
 use serde::{Serialize, de::DeserializeOwned};
 use serde_json::Value;
 
-use crate::error::Result;
+use crate::error::{Error, Result};
 
 /// The temp-file suffix used by [`write_json`] for atomic writes.
 pub const TMP_SUFFIX: &str = ".tmp";
@@ -57,11 +57,16 @@ pub fn read_json<T: DeserializeOwned>(path: &Utf8Path) -> Result<T> {
 /// target.
 /// If anything fails before the rename, the original file is left untouched and
 /// the temp file is cleaned up on a best-effort basis.
+/// The temp file lives on the same filesystem as `path`, so the write needs
+/// room for a full second copy of the content.
 ///
 /// The write is skipped when `path` already holds the exact same bytes, so an
 /// unchanged file keeps its modification time.
 /// Use [`write_json_force`] when a refreshed mtime is required regardless of
 /// content.
+///
+/// A failure names the path it was writing, and reports a full filesystem as
+/// [`Error::OutOfSpace`].
 pub fn write_json<T: Serialize>(path: &Utf8Path, value: &T) -> Result<()> {
     write_json_impl(path, value, false)
 }
@@ -84,7 +89,7 @@ fn write_json_impl<T: Serialize>(path: &Utf8Path, value: &T, force: bool) -> Res
 /// is skipped so the file's modification time is preserved.
 fn write_bytes(path: &Utf8Path, bytes: &[u8], force: bool) -> Result<()> {
     if let Some(parent) = path.parent() {
-        fs::create_dir_all(parent)?;
+        fs::create_dir_all(parent).map_err(|error| Error::write_failed(parent, error))?;
     }
 
     if !force
@@ -97,12 +102,12 @@ fn write_bytes(path: &Utf8Path, bytes: &[u8], force: bool) -> Result<()> {
     let tmp_path = tmp_path_for(path);
     if let Err(error) = fs::write(&tmp_path, bytes) {
         let _err = fs::remove_file(&tmp_path);
-        return Err(error.into());
+        return Err(Error::write_failed(&tmp_path, error));
     }
 
     if let Err(error) = fs::rename(&tmp_path, path) {
         let _err = fs::remove_file(&tmp_path);
-        return Err(error.into());
+        return Err(Error::write_failed(path, error));
     }
 
     Ok(())
diff --git a/crates/jp_storage/src/value_tests.rs b/crates/jp_storage/src/value_tests.rs
index a40a0b38..2732ea4b 100644
--- a/crates/jp_storage/src/value_tests.rs
+++ b/crates/jp_storage/src/value_tests.rs
@@ -54,6 +54,25 @@ fn write_json_preserves_original_on_write_failure() {
     assert_eq!(content, json!({"original": true}));
 }
 
+#[test]
+fn write_json_failure_names_the_path_it_was_writing() {
+    let tmp = tempdir().unwrap();
+    let path = tmp.path().join("out.json");
+
+    // A directory at the .tmp path makes the temp write fail on all platforms.
+    let blocker = Utf8PathBuf::from(format!("{path}{TMP_SUFFIX}"));
+    fs::create_dir(&blocker).unwrap();
+
+    let error = write_json(&path, &json!({"v": 1})).expect_err("temp write should fail");
+
+    // The failing path is the temp sibling, not the target: naming the target
+    // would point the user at a file that was never touched.
+    match error {
+        Error::WriteFailed { path: failed, .. } => assert_eq!(failed, blocker),
+        other => panic!("expected WriteFailed, got {other:?}"),
+    }
+}
+
 #[test]
 fn write_json_overwrites_existing_file() {
     let tmp = tempdir().unwrap();
diff --git a/crates/jp_workspace/src/conversation_lock.rs b/crates/jp_workspace/src/conversation_lock.rs
index f444b39c..b02a0075 100644
--- a/crates/jp_workspace/src/conversation_lock.rs
+++ b/crates/jp_workspace/src/conversation_lock.rs
@@ -61,10 +61,32 @@
 //! - **`Drop`** — safety net.
 //!   If the `ConversationMut` drops while dirty (e.g., due to `?` unwinding),
 //!   `Drop` persists the data.
-//!   Errors are logged but cannot be propagated from `Drop`.
 //!
 //! Long-running loops should call `flush()` at each checkpoint so disk errors
 //! halt immediately rather than letting the loop continue with unsaved data.
+//!
+//! # Reporting Drop-Time Failures
+//!
+//! `Drop` cannot propagate, so a failed drop-time persist is recorded on state
+//! shared with the originating lock instead of being written to the terminal.
+//! The next `flush()` returns it, and
+//! [`take_persist_failure`][ConversationLock::take_persist_failure] drains it
+//! for a caller that wants to report at teardown.
+//! Only the first failure is kept, so one failing disk yields one diagnostic
+//! rather than one per mutation scope, and the reporting happens where the
+//! output channel and the exit code live.
+//!
+//! Every later scope still attempts its own write.
+//! A persist can span two filesystems (the durable user-local copy and the
+//! workspace projection), so a failure against one of them says nothing about
+//! the other: skipping subsequent writes would strand new events that the
+//! healthy root could still accept.
+//!
+//! The record is owned by the `Workspace`, not by the lock, so a failure
+//! recorded while a dropped future unwinds is still there once the lock is
+//! gone.
+//! `Workspace::take_persist_failure` is the drain of last resort for a run
+//! whose command future was cancelled before it could report.
 
 use std::sync::{
     Arc,
@@ -73,10 +95,36 @@ use std::sync::{
 
 use jp_conversation::{Conversation, ConversationId, ConversationStream};
 use jp_storage::backend::{ConversationLockGuard, PersistBackend, Projection};
-use parking_lot::{RwLock, RwLockReadGuard};
-use tracing::info;
+use parking_lot::{Mutex, RwLock, RwLockReadGuard};
+use tracing::{info, warn};
 
-use crate::handle::ConversationHandle;
+use crate::{error::Error, handle::ConversationHandle};
+
+/// Shared record of drop-time persistence outcomes.
+///
+/// See the module docs for how a recorded failure reaches the user.
+#[derive(Debug, Default)]
+pub(crate) struct PersistState {
+    /// The first drop-time failure no caller has been told about yet.
+    ///
+    /// The first is kept rather than the last: later failures are consequences
+    /// of the same condition, while the first names the write that actually
+    /// broke.
+    failure: Option<Error>,
+}
+
+impl PersistState {
+    /// Take the recorded failure, leaving the slot empty.
+    pub(crate) fn take(&mut self) -> Option<Error> {
+        self.failure.take()
+    }
+}
+
+/// Shared handle to the record of drop-time persistence failures.
+///
+/// Held by the workspace and by every lock and scope derived from it, so a
+/// failure recorded while a future unwinds outlives the scope that recorded it.
+pub(crate) type PersistFailures = Arc<Mutex<PersistState>>;
 
 /// Result of attempting to acquire a conversation lock.
 #[derive(Debug)]
@@ -105,6 +153,7 @@ pub struct ConversationLock {
     writer: Arc<dyn PersistBackend>,
     lock_guard: Arc<Box<dyn ConversationLockGuard>>,
     projection: Projection,
+    persist: PersistFailures,
 }
 
 impl ConversationLock {
@@ -120,6 +169,7 @@ impl ConversationLock {
         writer: Arc<dyn PersistBackend>,
         lock_guard: Box<dyn ConversationLockGuard>,
         projection: Projection,
+        persist: PersistFailures,
     ) -> Self {
         Self {
             id: handle.into_inner(),
@@ -128,9 +178,21 @@ impl ConversationLock {
             writer,
             lock_guard: Arc::new(lock_guard),
             projection,
+            persist,
         }
     }
 
+    /// Take the drop-time persist failure recorded since the last call.
+    ///
+    /// A caller reports it once, through whatever output channel and exit code
+    /// it owns.
+    /// Failures already surfaced through [`ConversationMut::flush`] are not
+    /// recorded here and are never returned twice.
+    #[must_use]
+    pub fn take_persist_failure(&self) -> Option<Error> {
+        self.persist.lock().take()
+    }
+
     /// The write projection this lock persists with.
     ///
     /// Resolved from storage presence at acquisition (or the creation flags for
@@ -184,6 +246,7 @@ impl ConversationLock {
             dirty: AtomicBool::new(false),
             writer: Arc::clone(&self.writer),
             projection: self.projection,
+            persist: Arc::clone(&self.persist),
             _lock_guard: Arc::clone(&self.lock_guard),
         }
     }
@@ -202,6 +265,7 @@ impl ConversationLock {
             dirty: AtomicBool::new(false),
             writer: self.writer,
             projection: self.projection,
+            persist: self.persist,
             _lock_guard: self.lock_guard,
         }
     }
@@ -235,6 +299,10 @@ pub struct ConversationMut {
     writer: Arc<dyn PersistBackend>,
     projection: Projection,
 
+    // Shared with the workspace, the originating lock, and every other scope
+    // derived from it, so a failure recorded here survives this scope's drop.
+    persist: PersistFailures,
+
     // Holds the lock guard alive. Released when the last Arc drops.
     _lock_guard: Arc<Box<dyn ConversationLockGuard>>,
 }
@@ -331,8 +399,10 @@ impl ConversationMut {
     /// Long-running loops **must** call this at each checkpoint (e.g., after
     /// each turn in the LLM loop) so that I/O errors propagate immediately via
     /// `?`.
-    /// The `Drop` implementation is a safety net for `?` unwinding — it
-    /// persists partial state but swallows errors.
+    ///
+    /// Returns a persist failure recorded by an earlier drop-time write before
+    /// attempting its own, so a failure that could not propagate from `Drop`
+    /// still reaches a caller that can act on it.
     ///
     /// Takes `&mut self` to prevent calling while a write guard from
     /// `update_events()` or `update_metadata()` is held (which would deadlock).
@@ -341,12 +411,19 @@ impl ConversationMut {
     ///
     /// After a successful flush, the dirty flag is cleared.
     pub fn flush(&mut self) -> crate::error::Result<()> {
+        if let Some(error) = self.persist.lock().take() {
+            return Err(error);
+        }
+
         if !self.dirty.load(Ordering::Relaxed) {
             return Ok(());
         }
 
         let meta = self.metadata.read();
         let evts = self.events.read();
+        // Not recorded on the shared state: this error is returned, so the
+        // caller owns reporting it. The scope stays dirty, so if the caller
+        // swallows it, the drop below retries and records that failure.
         self.writer.write(&self.id, &meta, &evts, self.projection)?;
         self.dirty.store(false, Ordering::Relaxed);
 
@@ -354,6 +431,15 @@ impl ConversationMut {
         Ok(())
     }
 
+    /// Take the drop-time persist failure recorded since the last call.
+    ///
+    /// Counterpart of [`ConversationLock::take_persist_failure`] for a scope
+    /// that owns the lock, where no `ConversationLock` remains to drain.
+    #[must_use]
+    pub fn take_persist_failure(&self) -> Option<Error> {
+        self.persist.lock().take()
+    }
+
     /// The write projection this scope persists with.
     #[must_use]
     pub fn projection(&self) -> Projection {
@@ -402,9 +488,14 @@ impl Drop for ConversationMut {
         let meta = self.metadata.read();
         let evts = self.events.read();
 
-        #[expect(clippy::print_stderr)]
-        if let Err(e) = self.writer.write(&self.id, &meta, &evts, self.projection) {
-            eprintln!("Failed to persist conversation {}: {e}", self.id);
+        if let Err(error) = self.writer.write(&self.id, &meta, &evts, self.projection) {
+            let error = Error::from(error);
+            warn!(id = %self.id, %error, "Failed to persist conversation.");
+
+            let mut state = self.persist.lock();
+            if state.failure.is_none() {
+                state.failure = Some(error);
+            }
         }
     }
 }
diff --git a/crates/jp_workspace/src/conversation_lock_tests.rs b/crates/jp_workspace/src/conversation_lock_tests.rs
index cda218ee..8fab1bf7 100644
--- a/crates/jp_workspace/src/conversation_lock_tests.rs
+++ b/crates/jp_workspace/src/conversation_lock_tests.rs
@@ -1,5 +1,12 @@
-use std::sync::{Arc, Mutex};
-
+use std::{
+    io,
+    sync::{
+        Arc, Mutex,
+        atomic::{AtomicUsize, Ordering as AtomicOrdering},
+    },
+};
+
+use camino::Utf8Path;
 use jp_conversation::{Conversation, ConversationId, ConversationStream};
 use jp_storage::backend::{NoopLockGuard, NullPersistBackend, PersistBackend, Projection};
 use parking_lot::RwLock;
@@ -53,6 +60,79 @@ impl PersistBackend for MockPersistBackend {
     }
 }
 
+/// Persistence backend whose every write fails with a full disk, counting the
+/// attempts.
+///
+/// The count is what makes the retry claims testable: a backend that only
+/// records the error cannot distinguish "the write was skipped" from "the write
+/// was retried and failed again".
+#[derive(Debug, Default)]
+struct FailingPersistBackend {
+    attempts: AtomicUsize,
+}
+
+impl FailingPersistBackend {
+    fn attempts(&self) -> usize {
+        self.attempts.load(AtomicOrdering::Relaxed)
+    }
+}
+
+impl PersistBackend for FailingPersistBackend {
+    fn write(
+        &self,
+        _id: &ConversationId,
+        _metadata: &Conversation,
+        _events: &ConversationStream,
+        _projection: Projection,
+    ) -> Result<(), jp_storage::Error> {
+        self.attempts.fetch_add(1, AtomicOrdering::Relaxed);
+
+        Err(jp_storage::Error::write_failed(
+            Utf8Path::new("/data/conv/events.json"),
+            io::Error::from(io::ErrorKind::StorageFull),
+        ))
+    }
+
+    fn remove(&self, _id: &ConversationId) -> Result<(), jp_storage::Error> {
+        Ok(())
+    }
+
+    fn archive(&self, _id: &ConversationId) -> Result<(), jp_storage::Error> {
+        Ok(())
+    }
+
+    fn unarchive(&self, _id: &ConversationId) -> Result<(), jp_storage::Error> {
+        Ok(())
+    }
+}
+
+fn test_lock_with_failing_backend() -> (ConversationLock, Arc<FailingPersistBackend>) {
+    let (lock, backend, _) = test_lock_sharing_persist_state();
+    (lock, backend)
+}
+
+/// A failing-backend lock plus the persist record the workspace would own.
+///
+/// Handed out separately so a test can drain after the lock itself is gone.
+fn test_lock_sharing_persist_state() -> (
+    ConversationLock,
+    Arc<FailingPersistBackend>,
+    PersistFailures,
+) {
+    let backend = Arc::new(FailingPersistBackend::default());
+    let failures = PersistFailures::default();
+    let lock = ConversationLock::new(
+        test_handle(),
+        Arc::new(RwLock::new(Conversation::default())),
+        Arc::new(RwLock::new(ConversationStream::new_test())),
+        Arc::clone(&backend) as _,
+        Box::new(NoopLockGuard),
+        Projection::Projected,
+        Arc::clone(&failures),
+    );
+    (lock, backend, failures)
+}
+
 fn test_id() -> ConversationId {
     ConversationId::try_from(chrono::DateTime::<chrono::Utc>::UNIX_EPOCH).unwrap()
 }
@@ -70,6 +150,7 @@ fn test_lock_with_mock() -> (ConversationLock, Arc<MockPersistBackend>) {
         Arc::clone(&mock) as _,
         Box::new(NoopLockGuard),
         Projection::Projected,
+        PersistFailures::default(),
     );
     (lock, mock)
 }
@@ -82,6 +163,7 @@ fn test_lock_no_writer() -> ConversationLock {
         Arc::new(NullPersistBackend),
         Box::new(NoopLockGuard),
         Projection::Projected,
+        PersistFailures::default(),
     )
 }
 
@@ -291,6 +373,190 @@ fn as_mut_mutations_visible_through_lock() {
     assert_eq!(lock.metadata().title.as_deref(), Some("visible"));
 }
 
+#[test]
+fn no_persist_failure_recorded_on_a_healthy_backend() {
+    let (lock, _mock) = test_lock_with_mock();
+    {
+        let conv = lock.as_mut();
+        conv.update_metadata(|m| m.title = Some("fine".into()));
+    }
+    assert!(lock.take_persist_failure().is_none());
+}
+
+#[test]
+fn drop_records_persist_failure_on_the_lock() {
+    let (lock, backend) = test_lock_with_failing_backend();
+
+    {
+        let conv = lock.as_mut();
+        conv.update_metadata(|m| m.title = Some("unsaved".into()));
+    }
+
+    assert_eq!(backend.attempts(), 1);
+    let error = lock
+        .take_persist_failure()
+        .expect("the failed drop-time write is recorded on the lock");
+    assert_eq!(
+        error.to_string(),
+        "Storage error: no space left on device while writing /data/conv/events.json"
+    );
+}
+
+#[test]
+fn recorded_persist_failure_is_returned_only_once() {
+    let (lock, _backend) = test_lock_with_failing_backend();
+
+    {
+        let conv = lock.as_mut();
+        conv.update_metadata(|_| {});
+    }
+
+    assert!(lock.take_persist_failure().is_some());
+    assert!(
+        lock.take_persist_failure().is_none(),
+        "draining the failure must not report it a second time"
+    );
+}
+
+#[test]
+fn every_scope_attempts_its_own_write_after_a_failure() {
+    // A persist spans two roots that can be separate filesystems. A failure
+    // against one says nothing about the other, so a scope must never skip its
+    // write on the strength of an earlier failure: doing so strands new events
+    // that the healthy root would have accepted.
+    let (lock, backend) = test_lock_with_failing_backend();
+
+    for _ in 0..3 {
+        let conv = lock.as_mut();
+        conv.update_metadata(|_| {});
+    }
+
+    assert_eq!(backend.attempts(), 3);
+}
+
+#[test]
+fn out_of_space_records_only_the_first_failure() {
+    let (lock, _backend) = test_lock_with_failing_backend();
+
+    {
+        let conv = lock.as_mut();
+        conv.update_metadata(|m| m.title = Some("first".into()));
+    }
+    {
+        let conv = lock.as_mut();
+        conv.update_metadata(|m| m.title = Some("second".into()));
+    }
+
+    assert!(lock.take_persist_failure().is_some());
+    assert!(lock.take_persist_failure().is_none());
+}
+
+#[test]
+fn flush_returns_a_failure_recorded_by_an_earlier_drop() {
+    let (lock, backend) = test_lock_with_failing_backend();
+
+    {
+        let conv = lock.as_mut();
+        conv.update_metadata(|_| {});
+    }
+    assert_eq!(backend.attempts(), 1);
+
+    let mut conv = lock.as_mut();
+    let error = conv
+        .flush()
+        .expect_err("flush surfaces the earlier drop-time failure");
+
+    assert_eq!(
+        error.to_string(),
+        "Storage error: no space left on device while writing /data/conv/events.json"
+    );
+    assert_eq!(
+        backend.attempts(),
+        1,
+        "the recorded failure is returned without a further write of its own"
+    );
+    assert!(
+        lock.take_persist_failure().is_none(),
+        "a failure surfaced through flush is not reported a second time"
+    );
+}
+
+#[test]
+fn a_swallowed_flush_failure_still_reaches_a_drain() {
+    // `flush` records nothing of its own: it propagates, so the caller owns
+    // reporting. A caller that logs and moves on (the turn loop does this when
+    // aborting on a fatal stream error) would otherwise lose the fact that
+    // nothing was saved. The scope stays dirty, so its drop retries and
+    // records that failure for a teardown drain.
+    let (lock, backend) = test_lock_with_failing_backend();
+
+    let mut conv = lock.as_mut();
+    conv.update_metadata(|_| {});
+    assert!(conv.flush().is_err());
+    assert!(
+        conv.is_dirty(),
+        "a failed flush must not clear the dirty flag"
+    );
+    drop(conv);
+
+    assert_eq!(
+        backend.attempts(),
+        2,
+        "the drop safety net retries the failed flush"
+    );
+    assert!(
+        lock.take_persist_failure().is_some(),
+        "the retry's failure is recorded, so a swallowed flush is not silent"
+    );
+}
+
+#[test]
+fn a_recorded_failure_outlives_the_lock_that_produced_it() {
+    // What a cancelled command future leaves behind: the scope and the lock are
+    // both dropped during the unwind, so the only reachable drain is the record
+    // the workspace owns. If the record died with the lock, an interrupted
+    // out-of-space run would report nothing but `Interrupted`.
+    let (lock, backend, failures) = test_lock_sharing_persist_state();
+
+    {
+        let conv = lock.as_mut();
+        conv.update_metadata(|m| m.title = Some("unsaved".into()));
+    }
+    drop(lock);
+
+    assert_eq!(backend.attempts(), 1);
+    let error = failures
+        .lock()
+        .take()
+        .expect("the failure is still reachable after the lock is gone");
+    assert_eq!(
+        error.to_string(),
+        "Storage error: no space left on device while writing /data/conv/events.json"
+    );
+}
+
+#[test]
+fn a_scope_drains_a_failure_recorded_before_it_existed() {
+    // The lock-owning scope (`into_mut`) is the only drain point once the lock
+    // has been consumed, so it must see failures recorded by earlier scopes.
+    let (lock, _backend) = test_lock_with_failing_backend();
+
+    {
+        let conv = lock.as_mut();
+        conv.update_metadata(|_| {});
+    }
+
+    let conv = lock.into_mut();
+    let error = conv
+        .take_persist_failure()
+        .expect("the owning scope sees the earlier failure");
+    assert_eq!(
+        error.to_string(),
+        "Storage error: no space left on device while writing /data/conv/events.json"
+    );
+    assert!(conv.take_persist_failure().is_none());
+}
+
 #[test]
 fn multiple_as_mut_each_persist_independently() {
     let (lock, mock) = test_lock_with_mock();
diff --git a/crates/jp_workspace/src/lib.rs b/crates/jp_workspace/src/lib.rs
index 05800061..d06f69db 100644
--- a/crates/jp_workspace/src/lib.rs
+++ b/crates/jp_workspace/src/lib.rs
@@ -19,6 +19,7 @@ use std::{
 };
 
 use camino::{FromPathBufError, Utf8Path, Utf8PathBuf};
+use conversation_lock::PersistFailures;
 pub use conversation_lock::{ConversationLock, ConversationMut, LockResult};
 pub use error::Error;
 use error::Result;
@@ -65,6 +66,13 @@ pub struct Workspace {
 
     /// The in-memory state of the workspace.
     state: State,
+
+    /// Drop-time persistence failures recorded by conversation scopes.
+    ///
+    /// Lives here rather than on a [`ConversationLock`] so a failure recorded
+    /// while a cancelled future unwinds is still reachable after the lock is
+    /// gone.
+    persist_failures: PersistFailures,
 }
 
 impl Workspace {
@@ -116,6 +124,7 @@ impl Workspace {
             locker: backend.clone(),
             sessions: backend,
             state: State::default(),
+            persist_failures: PersistFailures::default(),
         }
     }
 
@@ -125,6 +134,18 @@ impl Workspace {
         &self.root
     }
 
+    /// Take the drop-time persist failure recorded by any conversation scope.
+    ///
+    /// The drain of last resort: a scope that dropped while a cancelled future
+    /// unwound recorded its failure here, and the lock it came from is already
+    /// gone.
+    /// Yields each failure once, so draining here after a
+    /// [`ConversationLock::take_persist_failure`] does not report it twice.
+    #[must_use]
+    pub fn take_persist_failure(&self) -> Option<Error> {
+        self.persist_failures.lock().take()
+    }
+
     /// Set the persist backend.
     #[must_use]
     pub fn with_persist(mut self, persist: Arc<dyn PersistBackend>) -> Self {
@@ -453,6 +474,7 @@ impl Workspace {
             Arc::clone(&self.persist),
             lock_guard,
             projection,
+            Arc::clone(&self.persist_failures),
         ))
     }
 
@@ -580,6 +602,7 @@ impl Workspace {
             Arc::clone(&self.persist),
             lock_guard,
             projection,
+            Arc::clone(&self.persist_failures),
         )))
     }
 
@@ -776,6 +799,7 @@ impl Workspace {
             Arc::clone(&self.persist),
             Box::new(NoopLockGuard),
             projection,
+            Arc::clone(&self.persist_failures),
         )
     }
 }
diff --git a/crates/jp_workspace/src/lib_tests.rs b/crates/jp_workspace/src/lib_tests.rs
index 44bf3085..ddb89178 100644
--- a/crates/jp_workspace/src/lib_tests.rs
+++ b/crates/jp_workspace/src/lib_tests.rs
@@ -24,6 +24,67 @@ fn workspace_with_fs(root: impl Into<Utf8PathBuf>, fs: &FsStorageBackend) -> Wor
     Workspace::new(root).with_backend(Arc::new(fs.clone()))
 }
 
+/// Persistence backend whose every write fails with a full disk.
+#[derive(Debug)]
+struct AlwaysFullBackend;
+
+impl jp_storage::backend::PersistBackend for AlwaysFullBackend {
+    fn write(
+        &self,
+        _id: &ConversationId,
+        _metadata: &Conversation,
+        _events: &ConversationStream,
+        _projection: jp_storage::backend::Projection,
+    ) -> std::result::Result<(), jp_storage::Error> {
+        Err(jp_storage::Error::write_failed(
+            Utf8Path::new("/data/conv/events.json"),
+            std::io::Error::from(std::io::ErrorKind::StorageFull),
+        ))
+    }
+
+    fn remove(&self, _id: &ConversationId) -> std::result::Result<(), jp_storage::Error> {
+        Ok(())
+    }
+
+    fn archive(&self, _id: &ConversationId) -> std::result::Result<(), jp_storage::Error> {
+        Ok(())
+    }
+
+    fn unarchive(&self, _id: &ConversationId) -> std::result::Result<(), jp_storage::Error> {
+        Ok(())
+    }
+}
+
+#[test]
+fn workspace_drains_a_persist_failure_left_by_a_dropped_lock() {
+    // The teardown path for a cancelled command future: everything the run held
+    // is dropped without any drain running, so the workspace is the only place
+    // left that can still report the conversation was not saved.
+    let mut workspace = Workspace::new("root").with_persist(Arc::new(AlwaysFullBackend));
+    let config = Arc::new(AppConfig::new_test());
+
+    let lock = workspace
+        .create_and_lock_conversation(Conversation::default(), config, None)
+        .unwrap();
+    {
+        let conv = lock.as_mut();
+        conv.update_metadata(|m| m.title = Some("unsaved".into()));
+    }
+    drop(lock);
+
+    let error = workspace
+        .take_persist_failure()
+        .expect("the workspace still holds the failure");
+    assert_eq!(
+        error.to_string(),
+        "Storage error: no space left on device while writing /data/conv/events.json"
+    );
+    assert!(
+        workspace.take_persist_failure().is_none(),
+        "the failure is reported once"
+    );
+}
+
 #[test]
 fn conversation_presence_reflects_creation_intent() {
     let mut ws = Workspace::new("root");
diff --git a/docs/.vitepress/rfd-summaries.json b/docs/.vitepress/rfd-summaries.json
index 731e64e9..ce484119 100644
--- a/docs/.vitepress/rfd-summaries.json
+++ b/docs/.vitepress/rfd-summaries.json
@@ -268,7 +268,7 @@
     "summary": "Automatically retry forced tool calls with reasoning disabled if the model ignores soft-force directives."
   },
   "069-guard-scoped-persistence-for-conversations.md": {
-    "hash": "19eae2427dbd5a7952b03406e1bf51b985e315786297e64fd659c252e3e5038e",
+    "hash": "15ef78674300a43b4ba7eeac9ab29415314b560933e8b5d84949e8b7a4c49871",
     "summary": "Guard-scoped persistence: ConversationMut auto-persists on drop while holding cross-process file lock, with callback-based writes and explicit flush for checkpoints."
   },
   "072-command-plugin-system.md": {
diff --git a/docs/rfd/069-guard-scoped-persistence-for-conversations.md b/docs/rfd/069-guard-scoped-persistence-for-conversations.md
index 89e49d19..f2bf7aab 100644
--- a/docs/rfd/069-guard-scoped-persistence-for-conversations.md
+++ b/docs/rfd/069-guard-scoped-persistence-for-conversations.md
@@ -166,6 +166,9 @@ impl ConversationMut {
     /// Long-running loops must call this at each checkpoint so I/O
     /// errors propagate via `?`. Drop is the safety net for unwinding.
     ///
+    /// Returns a failure recorded by an earlier drop-time write before
+    /// attempting its own.
+    ///
     /// Takes `&mut self` to prevent calling while a write guard from
     /// update_events() is held (which would deadlock).
     pub fn flush(&mut self) -> Result<()>;
@@ -174,17 +177,44 @@ impl ConversationMut {
 impl Drop for ConversationMut {
     fn drop(&mut self) {
         if !self.dirty.load(Ordering::Relaxed) { return; }
-        if let Some(writer) = &self.writer {
-            let meta = self.metadata.read();
-            let evts = self.events.read();
-            if let Err(e) = writer.write(&self.id, &meta, &evts) {
-                eprintln!("Failed to persist conversation {}: {e}", self.id);
-            }
+
+        let meta = self.metadata.read();
+        let evts = self.events.read();
+        if let Err(error) = self.writer.write(&self.id, &meta, &evts, self.projection) {
+            warn!(id = %self.id, %error, "Failed to persist conversation.");
+            let mut state = self.persist.lock();
+            if state.failure.is_none() { state.failure = Some(error); }
         }
     }
 }
 ```
 
+`Drop` cannot propagate, so it records the failure on state shared with the
+originating lock rather than writing to the terminal.
+`ConversationLock::take_persist_failure` (and its `ConversationMut` counterpart)
+drains it, so the shell reports once, through the printer, with a non-zero exit
+code.
+Only the first failure is kept, so one failing disk yields one diagnostic
+instead of one per mutation scope.
+
+Every later scope still attempts its own write.
+A persist spans two roots that can be separate filesystems (the durable
+user-local copy and the workspace projection), so a failure against one says
+nothing about the other; skipping subsequent writes would strand new events that
+the healthy root would have accepted.
+
+`flush` records nothing of its own, since it propagates and the caller owns
+reporting.
+A caller that swallows that error leaves the scope dirty, so the drop retries
+and records the outcome — a swallowed flush failure still reaches a drain.
+
+The record is owned by the `Workspace`, not by the lock.
+A cancelled command future (Ctrl-C, SIGTERM) is dropped mid-flight: the scope
+persists from its `Drop` and the lock goes with it, taking any drain inside the
+future along.
+`Workspace::take_persist_failure` runs after the command future has resolved or
+been dropped, and is the drain of last resort for that path.
+
 `AtomicBool` is used for the dirty flag instead of `Cell<bool>`.
 `Cell<bool>` is `!Sync`, which would make `ConversationMut` `!Sync` and cause
 async futures holding `&ConversationMut` across `.await` points to become
@@ -352,8 +382,12 @@ This is slightly more verbose but structurally prevents
 `.await`-across-lock-guard bugs.
 `?` composes naturally since the callback's return type is forwarded.
 
-**Errors in `Drop` are swallowed.** If persist fails during `ConversationMut`'s
-drop, the error is logged to stderr but cannot be propagated.
+**Errors in `Drop` cannot be propagated.** A persist failure during
+`ConversationMut`'s drop is recorded on the workspace-owned persist state and
+surfaced by the next `flush()`, by a `take_persist_failure()` drain in the
+command, or by the CLI's teardown drain, rather than returned from `drop`.
+The record outliving both the scope and the lock is what makes the teardown
+drain reachable after a cancelled future.
 Long-running loops must call `flush()?` at checkpoints so that I/O failures halt
 immediately.