fix(cli, workspace): Store a conversation only when a turn starts

crates/jp_cli/src/cmd/conversation/fork.rs

@@ -4,7 +4,7 @@ use jp_conversation::{ConversationId, ConversationStream, Error as ConversationE
 use jp_inquire::prompt::TerminalPromptBackend;
 use jp_printer::Printer;
 use jp_storage::backend::Projection;
-use jp_workspace::{ConversationHandle, ConversationLock};
+use jp_workspace::{ConversationHandle, ConversationLock, ConversationMut};
 use serde_json::Value;
 use tracing::debug;
 
@@ -71,10 +71,11 @@ impl Fork {
     pub(crate) async fn run(self, ctx: &mut Ctx, handles: &[ConversationHandle]) -> Output {
         let mut forked = Vec::with_capacity(handles.len());
 
-        // A fork is persisted as soon as it is created, so work that fails
-        // after that point leaves it behind. Reporting the IDs either way keeps
-        // the created conversations addressable; the error still propagates, so
-        // the exit status says the run did not finish.
+        // A fork is reported once it is written, and a source whose fork fails
+        // to write leaves nothing behind. With several sources the earlier forks
+        // are already on disk, so reporting them keeps them addressable; the
+        // error still propagates, so the exit status says the run did not
+        // finish.
         let result = self.fork_each(ctx, handles, &mut forked).await;
         print_forked(&ctx.printer, &forked);
 
@@ -121,7 +122,7 @@ impl Fork {
                 None
             };
 
-            let lock = fork_conversation(ctx, source, |events| {
+            let (lock, mut conv) = fork_conversation(ctx, source, |events| {
                 if let Some(config) = &collapsed {
                     // Discard every turn; the merged config becomes the new
                     // base, making this fork identical to a conversation
@@ -144,10 +145,9 @@ impl Fork {
             })
             .await?;
 
-            // One mutable scope for both post-fork mutations, so they share a
-            // single write at the closing flush.
-            let mut conv = lock.as_mut();
-
+            // The fork's own stream and both post-fork mutations share one
+            // scope, so the whole fork lands in a single write at the closing
+            // flush.
             if self.compact.should_compact() {
                 let cfg = ctx.config();
                 let events_snapshot = conv.events().clone();
@@ -178,8 +178,9 @@ impl Fork {
                 });
             }
 
-            // Write before reporting success, so a failed write is an error
-            // rather than a confirmation the user cannot trust.
+            // The fork exists only in memory until here, so this is the write
+            // that creates it. Before reporting success, so a failed write is an
+            // error rather than a confirmation the user cannot trust.
             conv.flush()?;
             drop(conv);
 
@@ -216,7 +217,12 @@ fn print_forked(printer: &Printer, ids: &[ConversationId]) {
     }
 }
 
-/// Fork a conversation and return the new conversation's lock.
+/// Fork a conversation, returning its lock and the scope holding its inherited
+/// stream.
+///
+/// The fork exists in memory only, and the returned scope discards on drop:
+/// flushing it is what writes the fork, and every other way out leaves no trace
+/// of it.
 ///
 /// The fork inherits the source's labels, then every `apply_on.fork` rule is
 /// re-resolved over them.
@@ -227,10 +233,11 @@ pub(crate) async fn fork_conversation(
     ctx: &mut Ctx,
     source: &ConversationHandle,
     mut filter: impl FnMut(&mut ConversationStream),
-) -> crate::Result<ConversationLock> {
+) -> crate::Result<(ConversationLock, ConversationMut)> {
     let now = ctx.now();
 
-    // Resolved up front: everything below this line writes to disk.
+    // Resolved up front: a rule that cannot be confirmed must fail before the
+    // fork exists at all, even in memory.
     let config = ctx.config();
     let prompts = TerminalPromptBackend;
     let resolved = Resolver::new(
@@ -276,16 +283,17 @@ pub(crate) async fn fork_conversation(
     // fork see more history than the source, but dropping a patch overlay
     // replays metadata the provider already rejected, wedging the fork on its
     // first query.
-    lock.as_mut()
-        .update_events(|events| events.append_stream(new_events));
+    let mut staged = lock.as_mut();
+    staged.discard_on_drop();
+    staged.update_events(|events| events.append_stream(new_events));
 
     debug!(
         source = source.id().to_string(),
         fork = lock.id().to_string(),
         "Forked conversation."
     );
 
-    Ok(lock)
+    Ok((lock, staged))
 }
 
 #[cfg(test)]

crates/jp_cli/src/cmd/conversation/fork_tests.rs

@@ -13,7 +13,10 @@ use jp_conversation::{
     event::{ChatRequest, ChatResponse, TurnStart},
 };
 use jp_printer::{OutputFormat, Printer};
-use jp_storage::backend::{FsStorageBackend, Projection};
+use jp_storage::{
+    backend::{FsStorageBackend, LoadBackend, Projection},
+    load::projected_conversation_ids,
+};
 use jp_workspace::Workspace;
 use tokio::runtime::Runtime;
 
@@ -1104,11 +1107,12 @@ fn fork_reresolves_apply_on_fork_rules() {
     ctx.set_now(ctx.now() + Duration::from_secs(1));
 
     let source_handle = ctx.workspace.acquire_conversation(&source_id).unwrap();
-    let fork_lock = Runtime::new()
+    let (fork_lock, fork_staged) = Runtime::new()
         .unwrap()
         .block_on(fork_conversation(&mut ctx, &source_handle, |_| {}))
         .unwrap();
     let fork_id = fork_lock.id();
+    drop(fork_staged);
     drop(fork_lock);
 
     let fork_handle = ctx.workspace.acquire_conversation(&fork_id).unwrap();
@@ -1553,10 +1557,11 @@ fn fork_inherits_local_only_projection() {
     );
 
     let source = ctx.workspace.acquire_conversation(&id).unwrap();
-    let lock = Runtime::new()
+    let (lock, staged) = Runtime::new()
         .unwrap()
         .block_on(fork_conversation(&mut ctx, &source, |_| {}))
         .unwrap();
+    drop(staged);
 
     assert_eq!(
         lock.projection(),
@@ -1615,10 +1620,11 @@ fn fork_inherits_patch_overlays() {
     ctx.set_now(ctx.now() + Duration::from_secs(1));
 
     let source = ctx.workspace.acquire_conversation(&id).unwrap();
-    let fork_lock = Runtime::new()
+    let (fork_lock, staged) = Runtime::new()
         .unwrap()
         .block_on(fork_conversation(&mut ctx, &source, |_| {}))
         .unwrap();
+    drop(staged);
 
     assert_eq!(
         fork_lock.events().overlays().count(),
@@ -1627,6 +1633,79 @@ fn fork_inherits_patch_overlays() {
     );
 }
 
+/// `jp c fork` writes the fork, inherited stream and all.
+///
+/// The fork is built in memory and written by the flush at the end of the
+/// command, so nothing but that flush puts it on disk.
+#[test]
+fn fork_writes_the_new_conversation() {
+    let tmp = tempdir().unwrap();
+    let (printer, _out, _) = Printer::memory(OutputFormat::TextPretty);
+    let storage = tmp.path().join(".jp");
+    let fs = Arc::new(FsStorageBackend::new(&storage).unwrap());
+    let workspace = Workspace::in_memory(tmp.path()).with_backend(Arc::clone(&fs));
+    let mut ctx = Ctx::new(
+        crate::bootstrap::ExecutionContext::for_workspace(&workspace),
+        workspace,
+        Some(Arc::clone(&fs)),
+        Runtime::new().unwrap(),
+        Globals::default(),
+        AppConfig::new_test(),
+        None,
+        printer,
+    );
+
+    let source_id = ConversationId::try_from(ctx.now()).unwrap();
+    ctx.workspace.create_conversation_with_id(
+        source_id,
+        Conversation::default().with_last_activated_at(ctx.now()),
+        ctx.config(),
+    );
+    let handle = ctx.workspace.acquire_conversation(&source_id).unwrap();
+    let lock = ctx.workspace.test_lock(handle);
+    lock.as_mut()
+        .update_events(|e| e.start_turn(ChatRequest::from("inherited")));
+    drop(lock);
+
+    ctx.set_now(ctx.now() + Duration::from_secs(1));
+
+    let fork = Fork {
+        target: PositionalIds::default(),
+        activate: false,
+        range: TurnSelection::default(),
+        title: None,
+        compact: CompactFlag::default(),
+        no_turns: false,
+    };
+    let source = ctx.workspace.acquire_conversation(&source_id).unwrap();
+    Runtime::new()
+        .unwrap()
+        .block_on(fork.run(&mut ctx, &[source]))
+        .unwrap();
+
+    let mut stored = projected_conversation_ids(&storage);
+    stored.sort_unstable();
+    assert_eq!(
+        stored.len(),
+        2,
+        "source and fork are both on disk: {stored:?}"
+    );
+
+    let fork_id = stored
+        .into_iter()
+        .find(|id| *id != source_id)
+        .expect("the fork is one of the two");
+    let events = fs
+        .load_conversation_stream(&fork_id, &PartialAppConfig::empty())
+        .expect("the fork's stream is readable");
+
+    assert_eq!(
+        events.turn_count(),
+        1,
+        "the fork inherited the source's turn"
+    );
+}
+
 /// A machine reader gets a top-level array of IDs, not lines to split.
 #[test]
 fn fork_prints_a_json_array_of_ids() {