diff --git a/crates/jp_cli/src/cmd/conversation/fork.rs b/crates/jp_cli/src/cmd/conversation/fork.rs index 24d55ff6d..1d29986ae 100644 --- a/crates/jp_cli/src/cmd/conversation/fork.rs +++ b/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 { +) -> 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,8 +283,9 @@ 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(), @@ -285,7 +293,7 @@ pub(crate) async fn fork_conversation( "Forked conversation." ); - Ok(lock) + Ok((lock, staged)) } #[cfg(test)] diff --git a/crates/jp_cli/src/cmd/conversation/fork_tests.rs b/crates/jp_cli/src/cmd/conversation/fork_tests.rs index 77c143bcc..864c8c28b 100644 --- a/crates/jp_cli/src/cmd/conversation/fork_tests.rs +++ b/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() { diff --git a/crates/jp_cli/src/cmd/query.rs b/crates/jp_cli/src/cmd/query.rs index 1c59dca42..5dbf11367 100644 --- a/crates/jp_cli/src/cmd/query.rs +++ b/crates/jp_cli/src/cmd/query.rs @@ -112,7 +112,9 @@ use jp_printer::{LineSink, PrintableExt as _, Printer, RegionStyle, StatusRegion use jp_storage::backend::{FsStorageBackend, Projection}; use jp_task::task::TitleGeneratorTask; use jp_term::width::{display_width, truncate_to_width}; -use jp_workspace::{ConversationHandle, ConversationLock, Id as WorkspaceId, Workspace}; +use jp_workspace::{ + ConversationHandle, ConversationLock, ConversationMut, Id as WorkspaceId, Workspace, +}; use minijinja::{Environment, UndefinedBehavior}; use strip_ansi_escapes::strip_str; use tokio::sync::broadcast::error::RecvError; @@ -192,7 +194,6 @@ pub(crate) struct Query { #[arg( long = "fork", num_args = 0..=1, - default_missing_value = "", value_parser = parse_fork_turns, conflicts_with = "new", )] @@ -394,9 +395,23 @@ impl Query { // 2. picker "start new": `start_new` is set, create a fresh conversation. // 3. --fork/--id/session: resolve an existing conversation, lock it. // 4. Lock contention: user picks "new" or "fork" from the prompt. - let (lock, fresh) = self.acquire_lock(ctx, handle, start_new).await?; + let AcquiredConversation { + lock, + fresh, + staged, + } = self.acquire_lock(ctx, handle, start_new).await?; - let result = self.run_locked(ctx, &lock, query, fresh).await; + let result = self.run_locked(ctx, &lock, query, fresh, staged).await; + + // A run that never started a turn wrote nothing, so a directory the + // editor created to compose in is all that is left of the conversation. + // An editor that failed to open has already had its directory reclaimed + // by the draft's revert guard; one that closed successfully on an empty + // buffer has not. + // + // Done here, while the lock is still held, so no other process is + // mid-compose in the same directory. + remove_empty_conversation_dir(ctx.fs_backend.as_deref(), &lock.id()); // Every exit from the locked region lands here, which is what makes // this the one reliable drain point: a mutation scope that dropped @@ -409,6 +424,9 @@ impl Query { /// /// `fresh` is `true` when this run created the conversation, so no config /// state predates its base config. + /// `staged` carries mutations acquiring the conversation already made + /// without writing them, which this run either commits or discards along + /// with its own. /// /// Errors propagate freely: the caller drains any persist failure the /// unwinding left behind. @@ -419,26 +437,36 @@ impl Query { lock: &ConversationLock, query: Option, fresh: bool, + staged: Option, ) -> Output { let now = ctx.now(); let cfg = ctx.config(); + // One scope for everything this run changes about the conversation + // before the turn starts — whatever `--tmp`, `--title`, `--fork`, + // `--compact`, or `--cfg` put on it. + // + // It discards on drop, so the `flush` below, once the request is known + // to be non-empty, is the only thing that writes. A query the user walks + // away from, and any failure before the turn starts, leave the stored + // conversation exactly as they found it. + let mut setup = staged.unwrap_or_else(|| lock.as_mut()); + setup.discard_on_drop(); + // 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); + // Stamp the expiry a fresh conversation was created with. An existing + // conversation keeps the expiry it has: `--tmp` describes a conversation + // this run starts, not one it continues, and clap rejects it without + // `--new`. + if fresh && let Some(duration) = self.expires_in_duration() { + let expires_at = chrono::Duration::from_std(duration) + .ok() + .and_then(|v| lock.id().timestamp().checked_add_signed(v)); - // 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) - { - warn!(%error, "Failed to record activation."); + setup.update_metadata(|m| m.expires_at = expires_at); } // Fail fast on provider misconfiguration (e.g. a missing API key @@ -456,9 +484,11 @@ impl Query { ) .map_err(Error::from)?; - // Compact the conversation before querying, if requested. + // Compact the conversation before querying, if requested. Staged on + // `setup`, so the composed request and the editor's history preview see + // the compacted stream while nothing is written yet. if self.compact.should_compact() { - self.apply_pre_query_compaction(lock, &cfg).await?; + self.apply_pre_query_compaction(&setup, &cfg).await?; } // `-u`/`-U` never enter the config, so the turn's choice is resolved @@ -471,7 +501,13 @@ impl Query { .configure_active_mcp_servers(forced_tool, McpServerScope::Exclusive) .await?; - let conv_title = lock.metadata().title.clone(); + // The title this run ends with, resolved here because the terminal + // title is named after it. Writing it into metadata waits until the + // request is known to be non-empty, so an abandoned query leaves the + // conversation as it found it. + let stored_title = lock.metadata().title.clone(); + let conv_title = + resolve_title_override(stored_title.clone(), self.title.as_deref(), self.no_title); // Show conversation identity in the terminal title. if ctx.term.is_tty { @@ -479,16 +515,29 @@ impl Query { } let cid = lock.id(); + + // Where the editor composes the draft. + // + // Named for where the conversation lives now, not for the title it will + // end the run with: a write reconciles the id to a single directory, + // renaming the live one into the new name and deleting every other + // copy. Composing under the new name would put the draft in the copy + // that gets deleted, while the rename carries it across for free. let conversation_path = ctx.fs_backend.as_deref().map_or_else( || { ctx.workspace .root() - .join(cid.to_dirname(conv_title.as_deref())) + .join(cid.to_dirname(stored_title.as_deref())) }, // The query draft is a transient editor scratch file, so it is // written to durable user-local storage (`user = true`) and never // projected into the committed workspace tree. - |fs| fs.build_conversation_dir(&cid, conv_title.as_deref(), true), + |fs| { + fs.find_user_local_conversation_dir(&cid) + .unwrap_or_else(|| { + fs.build_conversation_dir(&cid, stored_title.as_deref(), true) + }) + }, ); let piped = read_piped_stdin()?; @@ -515,8 +564,11 @@ impl Query { })?; let Some(mut chat_request) = chat_request else { - // Empty query, early exit. Nothing was mutated and nothing is - // dirty: the persisted stream is untouched, even for `--replay`. + // Empty query, early exit. `setup` drops unflushed, discarding + // everything staged on it, and the request was composed against a + // view of the stream rather than the stream itself — so the stored + // conversation is untouched, even for `--replay`, and even when it + // was this run that created it. if query_source == QuerySource::Editor { cleanup_query_message_file(ctx.fs_backend.as_deref(), &cid, DraftRemoval::Any); } @@ -524,6 +576,20 @@ impl Query { return Ok(()); }; + // Record this conversation as the session's active conversation. + // + // Deferred until the request is known to be non-empty: the metadata + // bump this performs is the first write a fresh conversation gets, so + // running it earlier would leave a stored, session-active conversation + // behind for a query that was ultimately ignored. + if let Some(session) = &ctx.session + && let Err(error) = ctx + .workspace + .activate_session_conversation(lock, session, now) + { + warn!(%error, "Failed to record activation."); + } + // Stamp the request with the configured user name so transcripts // attribute each turn correctly even when teammates with different // local configs continue the conversation. `None` falls back to a @@ -587,10 +653,12 @@ 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(); + // Store the title resolved before the draft was composed. A run that + // names the title the conversation already carries changes nothing, and + // leaves the scope clean rather than rewriting it. + if conv_title != stored_title { + setup.update_metadata(|m| m.title.clone_from(&conv_title)); + } // Persist config state changes into the conversation stream, now that // the query is known to be non-empty. Recording them before the @@ -891,6 +959,9 @@ impl Query { } /// Create a new conversation and return an exclusive lock. + /// + /// The conversation exists in memory only: nothing about it reaches storage + /// until a mutation scope on the returned lock is flushed. async fn create_new_conversation(&self, ctx: &mut Ctx) -> Result { let cfg = ctx.config(); @@ -923,16 +994,6 @@ impl Query { )?; let id = lock.id(); - if let Some(duration) = self.expires_in_duration() { - let mut conv = lock.as_mut(); - conv.update_metadata(|m| { - m.expires_at = chrono::Duration::from_std(duration) - .ok() - .and_then(|v| id.timestamp().checked_add_signed(v)); - }); - conv.flush()?; - } - debug!( id = id.to_string(), local = self.is_local(&cfg.conversation), @@ -1185,13 +1246,14 @@ impl Query { /// Apply compaction before the query turn starts. /// /// Applies all compaction rules from the resolved config and appends the - /// compaction events to the conversation. + /// compaction events to `conv`, which decides when — or whether — they + /// are written. async fn apply_pre_query_compaction( &self, - lock: &ConversationLock, + conv: &ConversationMut, cfg: &AppConfig, ) -> Result<()> { - let events = lock.events().clone(); + let events = conv.events().clone(); // The inline DSL plan never enters the config; assemble the effective // rules from the resolved config rules plus any `-k SPEC` here. @@ -1213,27 +1275,23 @@ impl Query { ) .await?; - super::conversation::compact::apply_compactions(&lock.as_mut(), compactions); + super::conversation::compact::apply_compactions(conv, compactions); Ok(()) } - /// Resolve the target conversation and return its exclusive lock. - /// - /// The second element is `true` when the conversation was freshly created - /// by this call: its base config is this invocation's resolved config, so - /// no config state predates it. - /// Forks return `false` — a fork copies the source's base config and - /// events, and therefore carries config state from before this invocation. + /// Resolve the target conversation and acquire its exclusive lock. async fn acquire_lock( &self, ctx: &mut Ctx, handle: Option, start_new: bool, - ) -> Result<(ConversationLock, bool)> { + ) -> Result { // Handle --new: create a fresh conversation. if self.is_new() { - return Ok((self.create_new_conversation(ctx).await?, true)); + return Ok(AcquiredConversation::created( + self.create_new_conversation(ctx).await?, + )); } // Handle the picker's "start a new conversation" choice. It carries no @@ -1243,7 +1301,9 @@ impl Query { if !self.allows_new_from_picker() { return Err(Error::NewConflictsWithTarget); } - return Ok((self.create_new_conversation(ctx).await?, true)); + return Ok(AcquiredConversation::created( + self.create_new_conversation(ctx).await?, + )); } // `--new` is only worth suggesting when it wouldn't conflict with a @@ -1253,7 +1313,9 @@ impl Query { // Handle --fork: fork the conversation before locking. if let Some(fork_turns) = &self.fork { - return Ok((fork_conversation(ctx, &handle, *fork_turns).await?, false)); + return Ok(AcquiredConversation::forked( + fork_conversation(ctx, &handle, *fork_turns).await?, + )); } let req = LockRequest::from_ctx(handle, ctx) @@ -1261,11 +1323,62 @@ impl Query { .allow_fork(true); match acquire_lock(req).await? { - LockOutcome::Acquired(lock) => Ok((lock, false)), - LockOutcome::NewConversation => Ok((self.create_new_conversation(ctx).await?, true)), - LockOutcome::ForkConversation(handle) => { - Ok((fork_conversation(ctx, &handle, None).await?, false)) - } + LockOutcome::Acquired(lock) => Ok(AcquiredConversation::resumed(lock)), + LockOutcome::NewConversation => Ok(AcquiredConversation::created( + self.create_new_conversation(ctx).await?, + )), + LockOutcome::ForkConversation(handle) => Ok(AcquiredConversation::forked( + fork_conversation(ctx, &handle, None).await?, + )), + } + } +} + +/// The conversation a run targets, exclusively locked, and how it got there. +struct AcquiredConversation { + lock: ConversationLock, + + /// Whether this run created the conversation, so no config state predates + /// its base config. + /// + /// A fork is `false`: it copies the source's base config and events, and + /// therefore carries config state from before this invocation. + fresh: bool, + + /// Mutations made while acquiring the conversation, held unwritten. + /// + /// A fork arrives with its inherited stream staged here. + /// The scope discards on drop, so whoever takes it decides whether the + /// conversation is ever written by flushing it. + staged: Option, +} + +impl AcquiredConversation { + /// A conversation that already existed, with nothing staged on it. + fn resumed(lock: ConversationLock) -> Self { + Self { + lock, + fresh: false, + staged: None, + } + } + + /// A conversation this run created, whose base config is this invocation's + /// resolved config. + fn created(lock: ConversationLock) -> Self { + Self { + lock, + fresh: true, + staged: None, + } + } + + /// A fork, whose inherited stream is staged but not yet written. + fn forked((lock, staged): (ConversationLock, ConversationMut)) -> Self { + Self { + lock, + fresh: false, + staged: Some(staged), } } } @@ -2194,7 +2307,7 @@ async fn fork_conversation( ctx: &mut Ctx, source: &ConversationHandle, fork_turns: Option, -) -> Result { +) -> Result<(ConversationLock, ConversationMut)> { fork::fork_conversation(ctx, source, |events| { if let Some(n) = fork_turns { events.retain_last_turns(n); @@ -2271,25 +2384,29 @@ fn resolve_new_title(from_heading: bool, generate_auto: bool, content: &str) -> NewTitle::Skip } -/// Apply `--title` / `--no-title` to the resolved conversation. +/// The title the conversation ends the run with, given the title it carries now +/// and the `--title` / `--no-title` flags. /// -/// Both flags act on `metadata.title` directly so the run ends with the title -/// the user asked for, regardless of whether the conversation is new, freshly -/// forked (which inherits the source's title), or resumed: +/// The two flags are mutually exclusive (enforced by clap): /// -/// - `--title T` sets the title to `Some(T)`. -/// - `--no-title` clears any existing title. -/// - Neither flag is a no-op. -fn apply_title_override(lock: &ConversationLock, title: Option<&str>, no_title: bool) { +/// - `--title T` names the title. +/// - `--no-title` clears it. +/// - Neither keeps `current`, whether that came from a fork inheriting the +/// source's title, a resumed conversation, or a new one with none. +fn resolve_title_override( + current: Option, + title: Option<&str>, + no_title: bool, +) -> Option { if let Some(title) = title { - lock.as_mut().update_metadata(|m| { - m.title = Some(title.to_owned()); - }); - } else if no_title { - lock.as_mut().update_metadata(|m| { - m.title = None; - }); + return Some(title.to_owned()); + } + + if no_title { + return None; } + + current } /// Append a `--cfg` reset keyword's events to a conversation stream. @@ -3166,6 +3283,26 @@ fn cleanup_query_message_file( } } +/// Remove a conversation's user-local directory when it holds nothing. +/// +/// The editor composes into a directory named after the conversation, which it +/// creates before anything about that conversation has been written. +/// A run that ends without starting a turn writes nothing, so the directory is +/// all that is left of it — and a directory without the managed files beside +/// it is indexed as a conversation and then trashed as corrupt by the next run. +/// +/// [`fs::remove_dir`] refuses a directory that is not empty, so a conversation +/// with stored files, or a draft deliberately kept for recovery, is left alone. +fn remove_empty_conversation_dir(fs_backend: Option<&FsStorageBackend>, id: &ConversationId) { + let Some(dir) = fs_backend.and_then(|fs| fs.find_user_local_conversation_dir(id)) else { + return; + }; + + if fs::remove_dir(&dir).is_ok() { + debug!(path = %dir, "Removed an empty conversation directory."); + } +} + fn current_dir_utf8() -> BoxedResult { let cwd = env::current_dir()?; Utf8PathBuf::from_path_buf(cwd) @@ -3280,14 +3417,13 @@ fn parse_schema(s: String) -> Result { .map_err(Into::into) } -/// Parse the `--fork` value. -/// Empty string means "all turns", a number means "keep last N turns". -fn parse_fork_turns(s: &str) -> std::result::Result, String> { - if s.is_empty() { - return Ok(None); - } - s.parse::() - .map(Some) +/// Parse the `--fork` value: how many trailing turns the fork keeps. +/// +/// Only reached when a value was written. +/// A bare `--fork` keeps every turn and never lands here: clap reads the flag's +/// absent value as `None` for the inner `Option`. +fn parse_fork_turns(s: &str) -> std::result::Result { + s.parse() .map_err(|_| format!("expected a positive integer, got '{s}'")) } diff --git a/crates/jp_cli/src/cmd/query_tests.rs b/crates/jp_cli/src/cmd/query_tests.rs index 71cee3596..89b750f77 100644 --- a/crates/jp_cli/src/cmd/query_tests.rs +++ b/crates/jp_cli/src/cmd/query_tests.rs @@ -13,6 +13,7 @@ use jp_config::{ }, model::id::{ModelIdConfig, PartialModelIdConfig, ProviderId}, style::stderr_rows::{RowCount, StderrRows}, + types::command::CommandConfigOrString, util::build, }; use jp_conversation::{ @@ -27,7 +28,10 @@ use jp_llm::{ }; use jp_mcp::{Startup, StderrLine}; use jp_printer::{OutputFormat, Printer, SharedBuffer, TerminalCapability}; -use jp_storage::backend::FsStorageBackend; +use jp_storage::{ + backend::{ConversationFilter, FsStorageBackend, LoadBackend}, + load::projected_conversation_ids, +}; use jp_term::width::display_width; use jp_workspace::{ ConversationHandle, Workspace, @@ -2024,20 +2028,6 @@ fn a_deferred_slot_keeps_its_whole_group_together() { assert_eq!(sources, ["mcp://a", "mcp://b", "file://last"]); } -fn lock_with_title( - workspace: &mut Workspace, - id: ConversationId, - title: Option<&str>, -) -> jp_workspace::ConversationLock { - let conversation = Conversation { - title: title.map(str::to_owned), - ..Default::default() - }; - workspace.create_conversation_with_id(id, conversation, Arc::new(AppConfig::new_test())); - let handle = workspace.acquire_conversation(&id).unwrap(); - workspace.test_lock(handle) -} - #[test] fn resolve_new_title_uses_leading_heading() { assert_eq!( @@ -2082,51 +2072,44 @@ fn resolve_new_title_skips_when_both_disabled() { } #[test] -fn apply_title_override_no_title_clears_existing_title() { +fn resolve_title_override_no_title_clears_inherited_title() { // `--no-title` should clear an inherited title (the // `--fork --no-title` case from PR #600 review): a forked // conversation inherits the source's title via // `fork_conversation`, and `--no-title` is supposed to leave // the run with no title at all. - let mut workspace = Workspace::in_memory("/tmp/test"); - let lock = lock_with_title(&mut workspace, make_id(1000), Some("inherited")); - - apply_title_override(&lock, None, true); - - assert_eq!(lock.metadata().title, None); + assert_eq!( + resolve_title_override(Some("inherited".to_owned()), None, true), + None + ); } #[test] -fn apply_title_override_no_title_clears_resumed_title() { - // `--no-title` is symmetric with `--title T`: both write the - // user's intent into `metadata.title`, regardless of whether - // the conversation is new, forked, or resumed. - let mut workspace = Workspace::in_memory("/tmp/test"); - let lock = lock_with_title(&mut workspace, make_id(1001), Some("existing")); - - apply_title_override(&lock, None, true); - - assert_eq!(lock.metadata().title, None); +fn resolve_title_override_no_title_clears_resumed_title() { + // `--no-title` is symmetric with `--title T`: both name what + // `metadata.title` ends up as, regardless of whether the + // conversation is new, forked, or resumed. + assert_eq!( + resolve_title_override(Some("existing".to_owned()), None, true), + None + ); } #[test] -fn apply_title_override_title_overwrites_existing_title() { - let mut workspace = Workspace::in_memory("/tmp/test"); - let lock = lock_with_title(&mut workspace, make_id(1002), Some("old")); - - apply_title_override(&lock, Some("new"), false); - - assert_eq!(lock.metadata().title.as_deref(), Some("new")); +fn resolve_title_override_title_overwrites_existing_title() { + assert_eq!( + resolve_title_override(Some("old".to_owned()), Some("new"), false), + Some("new".to_owned()) + ); } #[test] -fn apply_title_override_neither_flag_is_noop() { - let mut workspace = Workspace::in_memory("/tmp/test"); - let lock = lock_with_title(&mut workspace, make_id(1003), Some("keep")); - - apply_title_override(&lock, None, false); - - assert_eq!(lock.metadata().title.as_deref(), Some("keep")); +fn resolve_title_override_neither_flag_keeps_the_stored_title() { + assert_eq!( + resolve_title_override(Some("keep".to_owned()), None, false), + Some("keep".to_owned()) + ); + assert_eq!(resolve_title_override(None, None, false), None); } #[test] @@ -3462,6 +3445,463 @@ fn run_missing_at_path_query_leaves_conversation_and_session_untouched() { assert_eq!(ctx.workspace.session_active_conversation(&session), None); } +/// A context whose editor leaves the seeded draft exactly as JP wrote it, which +/// is what quitting without typing looks like to the query parser. +/// +/// The temp dir comes back so the caller keeps it alive. +fn empty_editor_ctx(session: &Session) -> (Ctx, SharedBuffer, SharedBuffer, Utf8TempDir) { + editor_ctx(session, "true") +} + +/// A context that runs `editor_cmd` as the user's editor. +/// +/// The temp dir comes back so the caller keeps it alive. +fn editor_ctx( + session: &Session, + editor_cmd: &str, +) -> (Ctx, SharedBuffer, SharedBuffer, Utf8TempDir) { + let tmp = camino_tempfile::tempdir().unwrap(); + let root = tmp.path(); + // User-local storage keeps the editor's draft out of the workspace tree, + // which is where the real CLI puts it. + let fs = Arc::new( + FsStorageBackend::new(&root.join(".jp")) + .unwrap() + .with_user_storage(&root.join("user"), None, "abc") + .unwrap(), + ); + + let (printer, out, err) = Printer::memory(OutputFormat::TextPretty); + let workspace = Workspace::in_memory(root).with_backend(Arc::clone(&fs)); + + let mut config = config_with_model(ProviderId::Test, "mock"); + config.editor.cmd = Some(CommandConfigOrString::String(editor_cmd.to_owned())); + // The test provider streams nothing, so a turn that starts fails. Retrying + // it five times with backoff is time these tests would only spend waiting. + config.assistant.request.max_retries = 0; + + let ctx = Ctx::new( + crate::bootstrap::ExecutionContext::for_workspace(&workspace), + workspace, + Some(fs), + Runtime::new().unwrap(), + Globals::default(), + config, + Some(session.clone()), + printer, + ); + + (ctx, out, err, tmp) +} + +/// A backend over the same two roots [`editor_ctx`] configures. +fn test_backend(root: &Utf8Path) -> FsStorageBackend { + FsStorageBackend::new(&root.join(".jp")) + .unwrap() + .with_user_storage(&root.join("user"), None, "abc") + .unwrap() +} + +/// Every conversation the backend can see, across both storage roots. +/// +/// [`projected_conversation_ids`] scans the workspace root alone, so it cannot +/// see the durable user-local store the editor's draft directory lives in. +fn stored_conversation_ids(root: &Utf8Path) -> Vec { + test_backend(root).load_conversation_ids(ConversationFilter::default()) +} + +/// Assert the store holds nothing for the next run's sanitization to repair. +/// +/// A directory left behind without its managed files is indexed as a +/// conversation and then trashed as corrupt, which an id assertion alone does +/// not catch: the id is there either way. +fn assert_store_needs_no_repair(root: &Utf8Path) { + let report = test_backend(root) + .sanitize() + .expect("sanitization succeeds"); + + assert!( + !report.has_repairs(), + "sanitization repaired: {:?}", + report.trashed + ); +} + +/// Run `jp query ` to completion, without `parse_query`'s `--no-edit`: +/// opening the editor is the path these tests exercise. +fn run_query(ctx: &mut Ctx, args: &[&str]) -> crate::cmd::Output { + let argv = ["query"].into_iter().chain(args.iter().copied()); + let query = QueryArgs::try_parse_from(argv).unwrap().query; + + let result = Runtime::new() + .unwrap() + .block_on(query.run(ctx, None, false)); + ctx.printer.flush(); + result +} + +// Quitting the editor without typing anything leaves no trace: the `--new` +// conversation is never written to storage, and the session is left pointing +// wherever it pointed before. Recording the activation is what writes it: the +// `last_activated_at` bump is the first mutation a fresh conversation gets, so +// this pins that no activation is recorded for a query that was ignored. +#[test] +fn run_empty_editor_query_stores_no_conversation() { + let session = Session { + id: SessionId::new("jp-cli-empty-query-test").unwrap(), + source: SessionSource::env("JP_SESSION"), + }; + let (mut ctx, out, err, tmp) = empty_editor_ctx(&session); + + run_query(&mut ctx, &["--new"]).unwrap(); + + assert_eq!(out.lock().as_str(), "Query is empty, ignoring.\n"); + assert_eq!(err.lock().as_str(), ""); + assert_eq!(stored_conversation_ids(tmp.path()), []); + assert_store_needs_no_repair(tmp.path()); + assert_eq!(ctx.workspace.session_active_conversation(&session), None); +} + +// `--title` names the title the run ends with, and the draft's directory is +// named after it, so it is resolved before the request is composed. Storing it +// is what must wait: an abandoned query leaves no titled conversation behind. +#[test] +fn run_empty_titled_query_stores_no_conversation() { + let session = Session { + id: SessionId::new("jp-cli-empty-title-test").unwrap(), + source: SessionSource::env("JP_SESSION"), + }; + let (mut ctx, out, _err, tmp) = empty_editor_ctx(&session); + + run_query(&mut ctx, &["--new", "--title", "a title"]).unwrap(); + + assert_eq!(out.lock().as_str(), "Query is empty, ignoring.\n"); + assert_eq!(stored_conversation_ids(tmp.path()), []); + assert_store_needs_no_repair(tmp.path()); +} + +// A bare `--fork` keeps every turn, and a value keeps that many trailing turns. +// +// The reported failure: the flag declared a `default_missing_value` fed through +// a parser returning `Option`, so clap registered the value as one type +// and the derived reader asked for another. Every `jp query --fork` panicked +// while parsing its own arguments. +#[test] +fn fork_flag_parses_with_and_without_a_turn_count() { + let bare = QueryArgs::try_parse_from(["query", "--fork"]) + .unwrap() + .query; + assert_eq!(bare.fork, Some(None)); + + let counted = QueryArgs::try_parse_from(["query", "--fork=2"]) + .unwrap() + .query; + assert_eq!(counted.fork, Some(Some(2))); +} + +#[test] +fn fork_flag_rejects_a_non_numeric_turn_count() { + let Err(error) = QueryArgs::try_parse_from(["query", "--fork=x"]) else { + panic!("a non-numeric turn count must be rejected"); + }; + + assert!( + error.to_string().contains("expected a positive integer"), + "unexpected error: {error}" + ); +} + +/// Seed a written conversation holding one completed turn, and return a handle +/// to it. +fn seed_conversation(ctx: &mut Ctx) -> ConversationHandle { + let id = make_id(1_700_000_000); + ctx.workspace.create_conversation_with_id( + id, + Conversation::default().with_last_activated_at(ctx.now()), + ctx.config(), + ); + + let handle = ctx.workspace.acquire_conversation(&id).unwrap(); + let lock = ctx.workspace.test_lock(handle); + lock.as_mut().update_events(|events| { + events.start_turn(ChatRequest::from("seeded question")); + events.extend([ConversationEvent::now(ChatResponse::message( + "seeded answer", + ))]); + }); + drop(lock); + + ctx.workspace.acquire_conversation(&id).unwrap() +} + +// `--expires-in` used to be stamped and flushed while the conversation was +// being created, which wrote it before there was any request to send. +#[test] +fn run_empty_expiring_query_stores_no_conversation() { + let session = Session { + id: SessionId::new("jp-cli-empty-expiry-test").unwrap(), + source: SessionSource::env("JP_SESSION"), + }; + let (mut ctx, out, _err, tmp) = empty_editor_ctx(&session); + + run_query(&mut ctx, &["--new", "--tmp=1h"]).unwrap(); + + assert_eq!(out.lock().as_str(), "Query is empty, ignoring.\n"); + assert_eq!(stored_conversation_ids(tmp.path()), []); + assert_store_needs_no_repair(tmp.path()); +} + +// A `--fork` query builds the fork before the request is composed, because the +// editor's history preview and `--replay` both read the fork's inherited +// stream. Abandoning the query must leave the source alone on disk. +#[test] +fn run_empty_forking_query_leaves_only_the_source() { + let session = Session { + id: SessionId::new("jp-cli-empty-fork-test").unwrap(), + source: SessionSource::env("JP_SESSION"), + }; + let (mut ctx, out, _err, tmp) = empty_editor_ctx(&session); + let source = seed_conversation(&mut ctx); + let source_id = source.id(); + + let query = QueryArgs::try_parse_from(["query", "--fork"]) + .unwrap() + .query; + Runtime::new() + .unwrap() + .block_on(query.run(&mut ctx, Some(source), false)) + .unwrap(); + ctx.printer.flush(); + + assert_eq!(out.lock().as_str(), "Query is empty, ignoring.\n"); + assert_eq!(stored_conversation_ids(tmp.path()), [source_id]); + assert_store_needs_no_repair(tmp.path()); +} + +// `--compact` is staged against the stream the request is composed against, so +// the editor's history preview shows the compacted conversation. Abandoning the +// query must leave the stored stream as it was. +#[test] +fn run_empty_compacting_query_stores_no_compaction() { + let session = Session { + id: SessionId::new("jp-cli-empty-compact-test").unwrap(), + source: SessionSource::env("JP_SESSION"), + }; + let (mut ctx, out, _err, tmp) = empty_editor_ctx(&session); + let source = seed_conversation(&mut ctx); + let source_id = source.id(); + + let query = QueryArgs::try_parse_from(["query", "--compact=r:.."]) + .unwrap() + .query; + Runtime::new() + .unwrap() + .block_on(query.run(&mut ctx, Some(source), false)) + .unwrap(); + ctx.printer.flush(); + + assert_eq!(out.lock().as_str(), "Query is empty, ignoring.\n"); + + // The compaction was staged, so the run really did exercise the path: what + // the assertion below pins is that staging it never reached storage. + let handle = ctx.workspace.acquire_conversation(&source_id).unwrap(); + assert_eq!( + ctx.workspace.events(&handle).unwrap().compactions().count(), + 1, + "the compaction must have been staged in memory" + ); + + let storage = FsStorageBackend::new(&tmp.path().join(".jp")).unwrap(); + let stored = storage + .load_conversation_stream(&source_id, &PartialAppConfig::empty()) + .unwrap(); + assert_eq!(stored.compactions().count(), 0); + + assert_store_needs_no_repair(tmp.path()); +} + +// Not only the empty query: any failure before the turn starts leaves the +// staged conversation unwritten. Here the editor cannot be spawned, which +// aborts the run after `--tmp` and the fresh conversation are already staged. +#[test] +fn run_failing_before_the_turn_stores_no_conversation() { + let session = Session { + id: SessionId::new("jp-cli-failed-setup-test").unwrap(), + source: SessionSource::env("JP_SESSION"), + }; + let (mut ctx, _out, _err, tmp) = editor_ctx(&session, "jp-no-such-editor-binary"); + + let Err(error) = run_query(&mut ctx, &["--new", "--tmp=1h"]) else { + panic!("a query whose editor cannot be spawned must fail"); + }; + assert_eq!(error.message.as_deref(), Some("Editor error")); + + assert_eq!(stored_conversation_ids(tmp.path()), []); + assert_store_needs_no_repair(tmp.path()); +} + +// A run that renames the conversation composes its draft in the directory the +// conversation lives in now, not the one its new title will name. Naming the +// draft's directory after the uncommitted title put it beside the real one, +// where the next write deleted it as a stale copy of the same id — taking the +// request the user had just typed. +#[test] +fn run_renaming_query_keeps_the_composed_draft() { + let session = Session { + id: SessionId::new("jp-cli-rename-draft-test").unwrap(), + source: SessionSource::env("JP_SESSION"), + }; + // An editor that types for the user, so the request is non-empty and the + // draft it leaves behind is the only copy of what was composed. + let (mut ctx, _out, _err, tmp) = editor_ctx( + &session, + r#"sh -c 'printf "typed request\n" > "$1"' jp-editor"#, + ); + + let id = make_id(1_700_000_000); + ctx.workspace.create_conversation_with_id( + id, + Conversation { + title: Some("Old".to_owned()), + ..Conversation::default().with_last_activated_at(ctx.now()) + }, + ctx.config(), + ); + let handle = ctx.workspace.acquire_conversation(&id).unwrap(); + let lock = ctx.workspace.test_lock(handle); + lock.as_mut() + .update_events(|events| events.start_turn(ChatRequest::from("seeded question"))); + drop(lock); + + let source = ctx.workspace.acquire_conversation(&id).unwrap(); + let query = QueryArgs::try_parse_from(["query", "--edit", "--title", "New"]) + .unwrap() + .query; + let Err(error) = Runtime::new() + .unwrap() + .block_on(query.run(&mut ctx, Some(source), false)) + else { + panic!("the test provider streams nothing, so the turn must fail"); + }; + assert_eq!(error.message.as_deref(), Some("Stream error")); + + let dir = ctx + .fs_backend + .as_deref() + .unwrap() + .find_user_local_conversation_dir(&id) + .expect("the conversation still has a user-local directory"); + let draft = std::fs::read_to_string(dir.join(editor::QUERY_FILENAME)) + .expect("the composed request is still recoverable"); + assert_eq!(draft, "typed request\n"); + + // The rename did happen, so the draft survived it rather than the title + // having been quietly dropped. + let storage = FsStorageBackend::new(&tmp.path().join(".jp")).unwrap(); + let metadata = storage.load_conversation_metadata(&id).unwrap(); + assert_eq!(metadata.title.as_deref(), Some("New")); +} + +// The other side of the deferral for `--tmp`: a query that does go ahead stores +// the expiry, stamped from the conversation's own creation time. +#[test] +fn run_expiring_query_stores_the_expiry_before_the_turn() { + let session = Session { + id: SessionId::new("jp-cli-stored-expiry-test").unwrap(), + source: SessionSource::env("JP_SESSION"), + }; + let (mut ctx, _out, _err, tmp) = empty_editor_ctx(&session); + + // An inline query skips the editor, so the request is non-empty. + let error = run_query(&mut ctx, &["--new", "--tmp=1h", "hello"]).unwrap_err(); + assert_eq!(error.message.as_deref(), Some("Stream error")); + + let storage_dir = tmp.path().join(".jp"); + let ids = projected_conversation_ids(&storage_dir); + assert_eq!(ids.len(), 1, "expected one stored conversation: {ids:?}"); + + let storage = FsStorageBackend::new(&storage_dir).unwrap(); + let metadata = storage.load_conversation_metadata(&ids[0]).unwrap(); + assert_eq!( + metadata.expires_at, + Some(ids[0].timestamp() + chrono::Duration::hours(1)) + ); +} + +// The other side of the deferral for `--fork`: a query that does go ahead +// writes the fork, with the source's turns inherited. +#[test] +fn run_forking_query_stores_the_fork_before_the_turn() { + let session = Session { + id: SessionId::new("jp-cli-stored-fork-test").unwrap(), + source: SessionSource::env("JP_SESSION"), + }; + let (mut ctx, _out, _err, tmp) = empty_editor_ctx(&session); + let source = seed_conversation(&mut ctx); + let source_id = source.id(); + + // An inline query skips the editor, so the request is non-empty. It goes + // ahead of `--fork`, which would otherwise swallow it as its turn count. + let query = QueryArgs::try_parse_from(["query", "hello", "--fork"]) + .unwrap() + .query; + let Err(error) = Runtime::new() + .unwrap() + .block_on(query.run(&mut ctx, Some(source), false)) + else { + panic!("the test provider streams nothing, so the turn must fail"); + }; + assert_eq!(error.message.as_deref(), Some("Stream error")); + + let storage_dir = tmp.path().join(".jp"); + let ids = projected_conversation_ids(&storage_dir); + assert_eq!(ids.len(), 2, "source and fork are both on disk: {ids:?}"); + + let fork_id = ids + .into_iter() + .find(|id| *id != source_id) + .expect("the fork is one of the two"); + let storage = FsStorageBackend::new(&storage_dir).unwrap(); + let stored = storage + .load_conversation_stream(&fork_id, &PartialAppConfig::empty()) + .unwrap(); + + assert_eq!( + stored.turn_count(), + 2, + "the fork inherited the source's turn and started its own" + ); +} + +// The other side of the deferral: a query that does go ahead ends with the +// title `--title` named, written to storage before the turn starts. The turn +// then fails here, because the test provider streams nothing. +#[test] +fn run_titled_query_stores_the_title_before_the_turn() { + let session = Session { + id: SessionId::new("jp-cli-stored-title-test").unwrap(), + source: SessionSource::env("JP_SESSION"), + }; + let (mut ctx, _out, _err, tmp) = empty_editor_ctx(&session); + + // An inline query skips the editor, so the request is non-empty. + let error = run_query(&mut ctx, &["--new", "--title", "a title", "hello"]).unwrap_err(); + assert_eq!(error.message.as_deref(), Some("Stream error")); + + let storage_dir = tmp.path().join(".jp"); + let ids = projected_conversation_ids(&storage_dir); + assert_eq!(ids.len(), 1, "expected one stored conversation: {ids:?}"); + + let storage = FsStorageBackend::new(&storage_dir).unwrap(); + let metadata = storage.load_conversation_metadata(&ids[0]).unwrap(); + assert_eq!(metadata.title.as_deref(), Some("a title")); + + assert_eq!( + ctx.workspace.session_active_conversation(&session), + Some(ids[0]) + ); +} + #[test] fn resolve_query_missing_at_path_errors() { let dir = camino_tempfile::tempdir().unwrap(); diff --git a/crates/jp_workspace/src/conversation_lock.rs b/crates/jp_workspace/src/conversation_lock.rs index d13d2c232..61f6cb992 100644 --- a/crates/jp_workspace/src/conversation_lock.rs +++ b/crates/jp_workspace/src/conversation_lock.rs @@ -256,6 +256,7 @@ impl ConversationLock { dirty: AtomicBool::new(false), writer: Arc::clone(&self.writer), projection: self.projection, + discard_on_drop: false, persist: Arc::clone(&self.persist), written: Arc::clone(&self.written), _lock_guard: Arc::clone(&self.lock_guard), @@ -276,6 +277,7 @@ impl ConversationLock { dirty: AtomicBool::new(false), writer: self.writer, projection: self.projection, + discard_on_drop: false, persist: self.persist, written: self.written, _lock_guard: self.lock_guard, @@ -300,8 +302,11 @@ impl std::fmt::Debug for ConversationLock { /// /// When dropped, if any mutation occurred (the dirty flag is set), the /// conversation data is persisted to disk while the flock is still held. +/// [`discard_on_drop()`] turns that off for a scope whose mutations are only +/// provisional. /// /// [`as_mut()`]: ConversationLock::as_mut +/// [`discard_on_drop()`]: Self::discard_on_drop /// [`into_mut()`]: ConversationLock::into_mut pub struct ConversationMut { id: ConversationId, @@ -311,6 +316,10 @@ pub struct ConversationMut { writer: Arc, projection: Projection, + /// Whether dropping the scope throws its mutations away instead of writing + /// them. + discard_on_drop: bool, + // 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, @@ -530,6 +539,25 @@ impl ConversationMut { pub(crate) fn clear_dirty(&self) { self.dirty.store(false, Ordering::Relaxed); } + + /// Discard staged mutations when the scope drops, instead of writing them. + /// + /// Inverts the default for a scope that stages work before knowing whether + /// the work will happen: [`flush`] becomes the only thing that writes, and + /// every other way out — an early return, a `?`, an unwind — leaves + /// storage untouched. + /// Without it the default holds, and a caller who never flushes still has + /// their changes written. + /// + /// Discarded mutations remain in the shared in-memory conversation, which + /// is where every scope on the same lock writes them. + /// Any *other* scope that persists afterwards therefore writes them too: + /// this is a statement about one scope, not a rollback. + /// + /// [`flush`]: Self::flush + pub fn discard_on_drop(&mut self) { + self.discard_on_drop = true; + } } // Static assertion: ConversationMut must be Send + Sync so it can be @@ -542,7 +570,7 @@ const _: () = { impl Drop for ConversationMut { fn drop(&mut self) { - if !self.dirty.load(Ordering::Relaxed) { + if self.discard_on_drop || !self.dirty.load(Ordering::Relaxed) { return; } diff --git a/crates/jp_workspace/src/conversation_lock_tests.rs b/crates/jp_workspace/src/conversation_lock_tests.rs index 5345da3a4..7c8e35b43 100644 --- a/crates/jp_workspace/src/conversation_lock_tests.rs +++ b/crates/jp_workspace/src/conversation_lock_tests.rs @@ -341,6 +341,32 @@ fn drop_skips_after_clear_dirty() { assert_eq!(mock.writes().len(), 0); } +#[test] +fn drop_discards_staged_mutations_when_asked() { + let (lock, mock) = test_lock_with_mock(); + let mut conv = lock.into_mut(); + conv.discard_on_drop(); + conv.update_metadata(|m| m.title = Some("staged".into())); + + drop(conv); + + assert_eq!(mock.writes().len(), 0); +} + +#[test] +fn flush_writes_even_when_dropping_would_discard() { + let (lock, mock) = test_lock_with_mock(); + let mut conv = lock.into_mut(); + conv.discard_on_drop(); + conv.update_metadata(|m| m.title = Some("committed".into())); + + conv.flush().unwrap(); + drop(conv); + + assert_eq!(mock.writes().len(), 1); + assert_eq!(mock.writes()[0].1.title.as_deref(), Some("committed")); +} + #[test] fn drop_skips_without_writer() { let lock = test_lock_no_writer();