fix(storage, workspace, cli): Report disk-full and write errors clearly

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;

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

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

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.");

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

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,20 +362,38 @@ 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)?;
 
         // 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,22 +612,28 @@ 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,
                 &ctx.mcp_client,
                 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| {