From 39d1d98bc5583f41928d95e05cd4827416570569 Mon Sep 17 00:00:00 2001 From: Jean Mertz Date: Thu, 9 Jul 2026 09:00:05 +0200 Subject: [PATCH] fix(cli, conversation): Defer replay trim to turn-start commit MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `jp query --replay` and the bare `--no-edit` replay-last-request path used to remove the trailing request event from the conversation stream while the request was still being built — during editor composition, MCP boot, or attachment loading. An interrupt or an empty query in that window persisted the removal without ever writing the replacement, silently dropping the last turn. The removal is now described by a `PendingStreamTrim` and only applied inside the same lock scope that appends the new turn, so the stream goes from "old turn present" to "old turn replaced" in one write. Building the request is otherwise read-only: `ConversationStream` gains `last_turn_event` to peek at the trailing event without popping it, and `ConversationLock`/`ConversationMut` gain `with_events` for read-only access that never marks the conversation dirty. Inline, piped, and replayed queries now also write their composed request to the conversation's recovery file before the turn starts, matching the existing editor-session behavior: an interrupt during MCP boot or model resolution no longer discards the request, and the next `jp query --edit` seeds its buffer from it. The file is still removed once the turn completes successfully. The `--cfg` config delta is now recorded only after the empty-query check, so a query that turns out empty no longer leaves a stray config event in the stream. Signed-off-by: Jean Mertz --- crates/jp_cli/src/cmd/query.rs | 223 +++++++++++++++--- crates/jp_cli/src/cmd/query/turn_loop.rs | 9 +- .../jp_cli/src/cmd/query/turn_loop_tests.rs | 41 ++++ crates/jp_cli/src/cmd/query_tests.rs | 130 +++++++++- crates/jp_cli/src/lib_tests.rs | 27 +++ crates/jp_conversation/src/stream.rs | 36 ++- crates/jp_conversation/src/stream_tests.rs | 16 ++ crates/jp_workspace/src/conversation_lock.rs | 22 ++ 8 files changed, 445 insertions(+), 59 deletions(-) diff --git a/crates/jp_cli/src/cmd/query.rs b/crates/jp_cli/src/cmd/query.rs index 6cae552be..dd98f9826 100644 --- a/crates/jp_cli/src/cmd/query.rs +++ b/crates/jp_cli/src/cmd/query.rs @@ -53,6 +53,7 @@ mod turn; mod turn_loop; use std::{ + borrow::Cow, collections::HashSet, env, fs, io::{self, BufRead as _, IsTerminal}, @@ -382,11 +383,6 @@ impl Query { warn!(%error, "Failed to record activation."); } - if let Some(delta) = get_config_delta_from_cli(&cfg, &lock)? { - lock.as_mut() - .update_events(|events| events.add_config_delta(delta)); - } - // Fail fast on provider misconfiguration (e.g. a missing API key // environment variable) before any side-effectful work below: // pre-query compaction can run a full summary LLM round-trip, MCP @@ -429,12 +425,21 @@ impl Query { |fs| fs.build_conversation_dir(&cid, conv_title.as_deref(), true), ); - let (query_source, mut editor_provided_config, chat_request) = lock - .as_mut() - .update_events(|stream| self.build_conversation(stream, &cfg, &conversation_path))?; + // Build the request read-only: replay resolution, quote seeding, and + // the editor session all operate on (a view of) the stream without + // mutating it. The destructive replay trim is described by + // `pending_trim` and committed at the turn-start commit point, + // atomically with appending the new request. + let BuiltConversation { + query_source, + mut editor_provided_config, + pending_trim, + chat_request, + } = lock.with_events(|stream| self.build_conversation(stream, &cfg, &conversation_path))?; let Some(mut chat_request) = chat_request else { - // Empty query, early exit. Auto-persist happens on lock drop. + // Empty query, early exit. Nothing was mutated and nothing is + // dirty: the persisted stream is untouched, even for `--replay`. if query_source == QuerySource::Editor { cleanup_query_message_file(ctx.fs_backend.as_deref(), &cid); } @@ -454,6 +459,30 @@ impl Query { chat_request.schema = schema.as_object().cloned(); } + // Preserve the composed request in the conversation's user-local + // scratch file so a pre-turn interrupt (Ctrl-C during MCP boot, + // attachment loading, model resolution, ...) never silently discards + // it: the next `jp q --edit` session seeds its buffer from this file. + // An editor-composed query is already on disk (the editor session + // wrote it and the guard was disarmed); only inline, piped, and + // replayed requests need the explicit write. The file is removed once + // the turn completes successfully. + // + // Only preserve when user-local storage is configured: without it, + // `conversation_path` has fallen back to the workspace conversation + // directory, where `cleanup_query_message_file` (which resolves + // exclusively through the user-local store) could never remove the + // file again — the scratch file must never land in the committed + // workspace tree. + if query_source != QuerySource::Editor + && ctx + .fs_backend + .as_deref() + .is_some_and(|fs| fs.user_storage_path().is_some()) + { + preserve_query_message_file(&conversation_path, &chat_request.content); + } + // Echo the request back through the same role-aware rendering // machinery used by replay and live streaming — a labeled user header // followed by the request body — so the boundary between user input @@ -470,6 +499,15 @@ impl Query { echo.render_user_request(&chat_request); } + // 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 !editor_provided_config.is_empty() { // Resolve any model aliases before storing in the stream so // that per-event configs always contain concrete model IDs. @@ -478,7 +516,16 @@ impl Query { .update_events(|events| events.add_config_delta(editor_provided_config)); } - let stream = lock.events().clone(); + // Snapshot the stream for title generation and thread assembly. The + // snapshot reflects what the durable stream will contain once the turn + // starts: the pending replay trim applied (on the clone — the durable + // stream is only trimmed at the turn-start commit point), the new + // request not yet appended. + let stream = { + let mut stream = lock.events().clone(); + pending_trim.apply(&mut stream); + stream + }; // Set the title for new or empty conversations (including forks). // Skip when `--title` or `--no-title` was provided (the user already @@ -570,6 +617,7 @@ impl Query { approvals, chat_request, invocation, + pending_trim, ) .await .map_err(|error| cmd::Error::from(error).with_persistence(true)); @@ -588,11 +636,12 @@ impl Query { } } - // Clean up the query file, unless we got an error. The conversation + // Clean up the query file, unless we got an error — on failure the + // file is the recovery copy of the request. The conversation // directory may have been renamed mid-turn (e.g. a heading-derived // title), so re-resolve the live directories rather than trusting the // path captured before the turn ran. - if query_source == QuerySource::Editor && turn_result.is_ok() { + if turn_result.is_ok() { cleanup_query_message_file(ctx.fs_backend.as_deref(), &cid); } @@ -610,29 +659,40 @@ impl Query { /// Build the chat request for this query. /// - /// Returns the request's [`QuerySource`], any editor-provided config, and - /// the [`ChatRequest`], if non-empty. - /// The request is **not** added to the stream — that is the responsibility - /// of [`TurnCoordinator::start_turn`]. + /// Returns the request's [`QuerySource`], any editor-provided config, the + /// stream edits deferred to the turn-start commit point, and the + /// [`ChatRequest`], if non-empty. + /// The stream is **not** mutated here: the request is added — and any + /// pending replay trim applied — by [`TurnCoordinator::start_turn`]. /// /// [`TurnCoordinator::start_turn`]: turn::TurnCoordinator::start_turn fn build_conversation( &self, - stream: &mut ConversationStream, + stream: &ConversationStream, config: &AppConfig, conversation_root: &Utf8Path, - ) -> Result<(QuerySource, PartialAppConfig, Option)> { - // If replaying, remove all events up-to-and-including the last - // `ChatRequest` event, which we'll replay. + ) -> Result { + let mut pending_trim = PendingStreamTrim::default(); + + // If replaying, the events up-to-and-including the last `ChatRequest` + // are logically removed and the request is replayed. The removal is + // computed on a clone (`view`) and only committed to the durable + // stream at the turn-start commit point (see [`PendingStreamTrim`]): + // an empty query, an interrupt, or an error before the turn starts + // leaves the persisted stream untouched. // // If not replaying (or replaying but no chat request event exists), we // create a new `ChatRequest` event, to populate with either the // provided query, or the contents of the text editor. - let mut chat_request = self - .replay - .then(|| stream.trim_chat_request()) - .flatten() - .unwrap_or_default(); + let (view, mut chat_request) = if self.replay { + pending_trim.replay_turn = true; + let mut view = stream.clone(); + let chat_request = view.trim_chat_request().unwrap_or_default(); + (Cow::Owned(view), chat_request) + } else { + (Cow::Borrowed(stream), ChatRequest::default()) + }; + let view = view.as_ref(); // If stdin contains data, we prepend it to the chat request. let stdin = io::stdin(); @@ -666,7 +726,7 @@ impl Query { // reply). Missing message (e.g. brand new conversation) degrades to a // warning and the editor opens with whatever else was seeded. if self.quote { - if let Some(message) = last_assistant_message(stream) { + if let Some(message) = last_assistant_message(view) { let quoted = blockquote(message); *chat_request = format!("{quoted}\n\n{chat_request}"); } else { @@ -676,7 +736,8 @@ impl Query { let (query_source, editor_provided_config) = self.edit_message( &mut chat_request, - stream, + view, + &mut pending_trim, !piped.is_empty(), config, conversation_root, @@ -700,11 +761,12 @@ impl Query { *chat_request = tmpl.render(&config.template.values)?; } - Ok(( + Ok(BuiltConversation { query_source, editor_provided_config, - (!chat_request.is_empty()).then_some(chat_request), - )) + pending_trim, + chat_request: (!chat_request.is_empty()).then_some(chat_request), + }) } /// Create a new conversation and return an exclusive lock. @@ -752,7 +814,8 @@ impl Query { fn edit_message( &self, request: &mut ChatRequest, - stream: &mut ConversationStream, + stream: &ConversationStream, + pending_trim: &mut PendingStreamTrim, piped: bool, config: &AppConfig, conversation_root: &Utf8Path, @@ -765,13 +828,18 @@ impl Query { if request.is_empty() && self.force_no_edit() { // If the last event in the stream is a `ChatRequest`, we don't add // anything, and simply "replay" the last message in the - // conversation. + // conversation. The event itself is only removed at the turn-start + // commit point (the request re-enters the stream as part of the + // new turn); until then the persisted stream stays untouched. // // Otherwise we add a default "continue" message. - if let Some(last) = stream.pop_if(ConversationEvent::is_chat_request) - && let Some(req) = last.into_inner().into_chat_request() - { - *request = req; + let last_request = stream + .last_turn_event() + .and_then(ConversationEvent::as_chat_request); + + if let Some(req) = last_request { + *request = req.clone(); + pending_trim.pop_request = true; } else { "continue".clone_into(request); } @@ -840,6 +908,7 @@ impl Query { approvals: Arc, chat_request: ChatRequest, invocation: InvocationContext, + pending_trim: PendingStreamTrim, ) -> Result<()> { let model_id = cfg.assistant.model.id.resolved(); let provider: Arc = Arc::from(provider::get_provider( @@ -881,6 +950,7 @@ impl Query { tool_coordinator, chat_request, invocation, + pending_trim, ) .await } @@ -1887,6 +1957,87 @@ fn load_approval_store( .unwrap_or_default() } +/// Output of [`Query::build_conversation`]. +struct BuiltConversation { + /// Where the request's content came from. + query_source: QuerySource, + /// Config assignments parsed from the editor buffer's front matter. + editor_provided_config: PartialAppConfig, + /// Stream edits deferred to the turn-start commit point. + pending_trim: PendingStreamTrim, + /// The composed request, if non-empty. + chat_request: Option, +} + +/// Destructive stream edits computed while building a request, deferred to the +/// turn-start commit point. +/// +/// `--replay` (and the bare `--no-edit` replay-last-request path) logically +/// removes trailing events before re-sending a request. +/// Applying that removal eagerly would persist a stream missing its last turn +/// while the editor is open and during MCP boot / attachment loading — an +/// interrupt (or an empty query) in that window would make the removal +/// permanent without the replacement request ever landing. +/// Instead the removal is described here and applied inside the same +/// `update_events` scope that appends the new turn +/// ([`TurnCoordinator::start_turn`] via [`run_turn_loop`]), so the persisted +/// stream goes from "old turn present" to "old turn replaced" in a single +/// write. +/// +/// [`TurnCoordinator::start_turn`]: turn::TurnCoordinator::start_turn +#[derive(Debug, Clone, Copy, Default)] +pub(crate) struct PendingStreamTrim { + /// Remove all trailing events up-to-and-including the last [`ChatRequest`] + /// event (`--replay`). + replay_turn: bool, + /// Remove the trailing [`ChatRequest`] event (bare `--no-edit` replaying + /// the last request). + pop_request: bool, +} + +impl PendingStreamTrim { + /// Apply the deferred removals to the stream. + pub(crate) fn apply(self, stream: &mut ConversationStream) { + if self.replay_turn { + drop(stream.trim_chat_request()); + } + if self.pop_request { + drop(stream.pop_if(ConversationEvent::is_chat_request)); + } + + // Both trims strip a turn's request but leave the [`TurnStart`] that + // opened it. Remove the now-empty turn marker so the replacement + // request doesn't open a second turn right behind it — the later + // stream sanitization only collapses duplicate `TurnStart`s before + // the *first* request, so a stale middle marker would survive in + // multi-turn conversations. + // + // Gated on a trim having run so a default (no-op) `PendingStreamTrim` + // never mutates the stream. + // + // [`TurnStart`]: jp_conversation::event::TurnStart + if self.replay_turn || self.pop_request { + stream.trim_trailing_empty_turn(); + } + } +} + +/// Write the pending request text to the conversation's user-local +/// query-message file, so an interrupt before the turn starts never discards +/// the composed request: the next `jp q --edit` session seeds its buffer from +/// this file. +/// +/// Best-effort: failing to preserve a recovery copy must not fail the query +/// itself, so errors are logged and swallowed. +fn preserve_query_message_file(conversation_dir: &Utf8Path, content: &str) { + let path = conversation_dir.join(editor::QUERY_FILENAME); + if let Err(error) = + fs::create_dir_all(conversation_dir).and_then(|()| fs::write(&path, content)) + { + warn!(path = %path, %error, "Failed to preserve query message file."); + } +} + /// Remove the editor's query-message file from a conversation's user-local /// storage directory, tolerating its absence. /// diff --git a/crates/jp_cli/src/cmd/query/turn_loop.rs b/crates/jp_cli/src/cmd/query/turn_loop.rs index 1283e44b2..abc8cd7e5 100644 --- a/crates/jp_cli/src/cmd/query/turn_loop.rs +++ b/crates/jp_cli/src/cmd/query/turn_loop.rs @@ -43,7 +43,7 @@ use tokio_stream::wrappers::ReceiverStream; use tracing::{debug, info, warn}; use super::{ - build_sections, build_thread, + PendingStreamTrim, build_sections, build_thread, interrupt::{ LoopAction, StreamingInterruptResult, handle_llm_event, handle_streaming_interrupt, reply_edit_mode, @@ -191,6 +191,7 @@ pub(super) async fn run_turn_loop( mut tool_coordinator: ToolCoordinator, chat_request: ChatRequest, invocation: InvocationContext, + pending_trim: PendingStreamTrim, ) -> Result<(), Error> { // The turn-level interrupt handler (RFD 045) is the outermost handler // scope within the turn: it owns the gaps between phases (persistence, @@ -270,7 +271,13 @@ pub(super) async fn run_turn_loop( match turn_coordinator.current_phase() { TurnPhase::Idle => { + // The turn-start commit point: any replay trim deferred while + // building the request (see [`PendingStreamTrim`]) is applied + // in the same `update_events` scope that appends the new + // request, so the durable stream never persists the removal + // without its replacement. lock.as_mut().update_events(|stream| { + pending_trim.apply(stream); turn_coordinator.start_turn(stream, chat_request.clone()); }); } diff --git a/crates/jp_cli/src/cmd/query/turn_loop_tests.rs b/crates/jp_cli/src/cmd/query/turn_loop_tests.rs index 471ce7313..e76124d11 100644 --- a/crates/jp_cli/src/cmd/query/turn_loop_tests.rs +++ b/crates/jp_cli/src/cmd/query/turn_loop_tests.rs @@ -337,6 +337,7 @@ async fn test_interrupt_stop_during_streaming_persists_content() { ToolCoordinator::new(config.conversation.tools.clone(), empty_executor_source()), chat_request.clone(), InvocationContext::default(), + PendingStreamTrim::default(), ) .await; @@ -438,6 +439,7 @@ async fn test_streaming_interrupt_menu_cancel_escalates() { ToolCoordinator::new(config.conversation.tools.clone(), empty_executor_source()), chat_request.clone(), InvocationContext::default(), + PendingStreamTrim::default(), ) .await; @@ -520,6 +522,7 @@ async fn test_normal_completion_persists_content() { ToolCoordinator::new(config.conversation.tools.clone(), empty_executor_source()), chat_request.clone(), InvocationContext::default(), + PendingStreamTrim::default(), ) .await .unwrap(); @@ -604,6 +607,7 @@ async fn premature_stream_end_without_finished_returns_error() { ToolCoordinator::new(config.conversation.tools.clone(), empty_executor_source()), ChatRequest::from("hi"), InvocationContext::default(), + PendingStreamTrim::default(), ), ) .await @@ -666,6 +670,7 @@ async fn premature_stream_end_exhausts_retry_budget() { ToolCoordinator::new(config.conversation.tools.clone(), empty_executor_source()), ChatRequest::from("hi"), InvocationContext::default(), + PendingStreamTrim::default(), ), ) .await @@ -760,6 +765,7 @@ async fn orphan_tool_call_is_sanitized_before_provider_request() { ToolCoordinator::new(config.conversation.tools.clone(), empty_executor_source()), ChatRequest::from("new query"), InvocationContext::default(), + PendingStreamTrim::default(), ) .await .unwrap(); @@ -843,6 +849,7 @@ async fn test_tool_call_cycle_completes_with_followup() { ToolCoordinator::new(config.conversation.tools.clone(), empty_executor_source()), chat_request.clone(), InvocationContext::default(), + PendingStreamTrim::default(), ) .await; @@ -1125,6 +1132,7 @@ async fn test_tool_interrupt_menu_cancel_escalates() { ToolCoordinator::new(config.conversation.tools.clone(), Box::new(executor_source)), chat_request.clone(), InvocationContext::default(), + PendingStreamTrim::default(), ) .await; @@ -1273,6 +1281,7 @@ async fn test_tool_stop_on_interrupt_commits_responses_without_follow_up() { .with_interrupt(config.interrupt.tool_call.clone()), chat_request.clone(), InvocationContext::default(), + PendingStreamTrim::default(), ) .await; @@ -1414,6 +1423,7 @@ async fn test_interrupt_during_tool_prompt_completes_turn_early() { ToolCoordinator::new(config.conversation.tools.clone(), Box::new(executor_source)), chat_request.clone(), InvocationContext::default(), + PendingStreamTrim::default(), ) .await; @@ -1524,6 +1534,7 @@ async fn test_multiple_tool_calls_in_sequence() { ToolCoordinator::new(config.conversation.tools.clone(), empty_executor_source()), chat_request.clone(), InvocationContext::default(), + PendingStreamTrim::default(), ) .await; @@ -1613,6 +1624,7 @@ async fn test_empty_tool_response_continues_cycle() { ToolCoordinator::new(config.conversation.tools.clone(), empty_executor_source()), chat_request.clone(), InvocationContext::default(), + PendingStreamTrim::default(), ) .await; @@ -1757,6 +1769,7 @@ async fn test_tool_restart_on_interrupt() { ToolCoordinator::new(config.conversation.tools.clone(), Box::new(executor_source)), chat_request.clone(), InvocationContext::default(), + PendingStreamTrim::default(), ) .await; @@ -1877,6 +1890,7 @@ async fn test_merged_stream_exits_after_tool_response() { ToolCoordinator::new(config.conversation.tools.clone(), empty_executor_source()), chat_request.clone(), InvocationContext::default(), + PendingStreamTrim::default(), ) .await; @@ -2003,6 +2017,7 @@ async fn test_tool_call_with_run_mode_ask_approves() { ToolCoordinator::new(config.conversation.tools.clone(), Box::new(executor_source)), chat_request.clone(), InvocationContext::default(), + PendingStreamTrim::default(), ) .await; @@ -2144,6 +2159,7 @@ async fn test_tool_call_with_run_mode_ask_skips() { ToolCoordinator::new(config.conversation.tools.clone(), Box::new(executor_source)), chat_request.clone(), InvocationContext::default(), + PendingStreamTrim::default(), ) .await; @@ -2289,6 +2305,7 @@ async fn test_tool_call_with_run_mode_unattended() { ToolCoordinator::new(config.conversation.tools.clone(), Box::new(executor_source)), chat_request.clone(), InvocationContext::default(), + PendingStreamTrim::default(), ) .await; @@ -2435,6 +2452,7 @@ async fn test_tool_call_with_run_mode_skip() { ToolCoordinator::new(config.conversation.tools.clone(), Box::new(executor_source)), chat_request.clone(), InvocationContext::default(), + PendingStreamTrim::default(), ) .await; @@ -2637,6 +2655,7 @@ async fn test_multiple_tools_with_different_run_modes() { ToolCoordinator::new(config.conversation.tools.clone(), Box::new(executor_source)), chat_request.clone(), InvocationContext::default(), + PendingStreamTrim::default(), ) .await; @@ -2784,6 +2803,7 @@ async fn test_tool_call_returns_error() { ToolCoordinator::new(config.conversation.tools.clone(), Box::new(executor_source)), chat_request.clone(), InvocationContext::default(), + PendingStreamTrim::default(), ) .await; @@ -3013,6 +3033,7 @@ async fn test_waiting_indicator_shows_during_delay() { ToolCoordinator::new(config.conversation.tools.clone(), empty_executor_source()), chat_request.clone(), InvocationContext::default(), + PendingStreamTrim::default(), ) .await .unwrap(); @@ -3112,6 +3133,7 @@ async fn test_waiting_indicator_survives_keep_alive_and_shows_status() { ToolCoordinator::new(config.conversation.tools.clone(), empty_executor_source()), chat_request.clone(), InvocationContext::default(), + PendingStreamTrim::default(), ) .await .unwrap(); @@ -3225,6 +3247,7 @@ async fn test_waiting_indicator_cleared_before_retry_notice() { ToolCoordinator::new(config.conversation.tools.clone(), empty_executor_source()), chat_request.clone(), InvocationContext::default(), + PendingStreamTrim::default(), ) .await .unwrap(); @@ -3309,6 +3332,7 @@ async fn test_waiting_indicator_not_shown_when_disabled() { ToolCoordinator::new(config.conversation.tools.clone(), empty_executor_source()), chat_request.clone(), InvocationContext::default(), + PendingStreamTrim::default(), ) .await .unwrap(); @@ -3384,6 +3408,7 @@ async fn test_waiting_indicator_not_shown_for_non_tty() { ToolCoordinator::new(config.conversation.tools.clone(), empty_executor_source()), chat_request.clone(), InvocationContext::default(), + PendingStreamTrim::default(), ) .await .unwrap(); @@ -3562,6 +3587,7 @@ async fn test_multi_part_tool_call_shows_preparing_spinner() { ToolCoordinator::new(config.conversation.tools.clone(), empty_executor_source()), chat_request.clone(), InvocationContext::default(), + PendingStreamTrim::default(), ) .await; @@ -3648,6 +3674,7 @@ async fn test_turn_start_event_is_emitted() { ToolCoordinator::new(config.conversation.tools.clone(), empty_executor_source()), chat_request.clone(), InvocationContext::default(), + PendingStreamTrim::default(), ) .await .unwrap(); @@ -3710,6 +3737,7 @@ async fn test_turn_start_index_increments_across_turns() { ToolCoordinator::new(config.conversation.tools.clone(), empty_executor_source()), chat_request.clone(), InvocationContext::default(), + PendingStreamTrim::default(), ) .await .unwrap(); @@ -3744,6 +3772,7 @@ async fn test_turn_start_index_increments_across_turns() { ToolCoordinator::new(config.conversation.tools.clone(), empty_executor_source()), chat_request.clone(), InvocationContext::default(), + PendingStreamTrim::default(), ) .await .unwrap(); @@ -3837,6 +3866,7 @@ async fn test_markdown_flushed_before_tool_header() { ToolCoordinator::new(config.conversation.tools.clone(), empty_executor_source()), chat_request.clone(), InvocationContext::default(), + PendingStreamTrim::default(), ) .await .unwrap(); @@ -4020,6 +4050,7 @@ async fn test_parallel_tool_calls_rendered_atomically() { ToolCoordinator::new(config.conversation.tools.clone(), Box::new(executor_source)), chat_request.clone(), InvocationContext::default(), + PendingStreamTrim::default(), ) .await .unwrap(); @@ -4178,6 +4209,7 @@ async fn test_single_tool_call_rendered_with_args() { ToolCoordinator::new(config.conversation.tools.clone(), Box::new(executor_source)), chat_request.clone(), InvocationContext::default(), + PendingStreamTrim::default(), ) .await .unwrap(); @@ -4434,6 +4466,7 @@ async fn test_tool_with_single_inquiry() { ToolCoordinator::new(config.conversation.tools.clone(), Box::new(executor_source)), chat_request, InvocationContext::default(), + PendingStreamTrim::default(), ) .await; @@ -4571,6 +4604,7 @@ async fn test_tool_with_multiple_inquiries() { ToolCoordinator::new(config.conversation.tools.clone(), Box::new(executor_source)), chat_request, InvocationContext::default(), + PendingStreamTrim::default(), ) .await; @@ -4722,6 +4756,7 @@ async fn test_parallel_tools_one_with_inquiry() { ToolCoordinator::new(config.conversation.tools.clone(), Box::new(executor_source)), chat_request, InvocationContext::default(), + PendingStreamTrim::default(), ) .await; @@ -4859,6 +4894,7 @@ async fn test_parallel_tools_both_with_inquiries() { ToolCoordinator::new(config.conversation.tools.clone(), Box::new(executor_source)), chat_request, InvocationContext::default(), + PendingStreamTrim::default(), ) .await; @@ -5008,6 +5044,7 @@ async fn test_retry_counter_resets_on_successful_event() { ToolCoordinator::new(config.conversation.tools.clone(), empty_executor_source()), chat_request, InvocationContext::default(), + PendingStreamTrim::default(), ) .await; @@ -5146,6 +5183,7 @@ async fn test_unavailable_tool_before_approved_does_not_panic() { ToolCoordinator::new(config.conversation.tools.clone(), Box::new(executor_source)), chat_request, InvocationContext::default(), + PendingStreamTrim::default(), ) .await; @@ -5255,6 +5293,7 @@ async fn test_inquiry_failure_marks_tool_as_error() { ToolCoordinator::new(config.conversation.tools.clone(), Box::new(executor_source)), chat_request, InvocationContext::default(), + PendingStreamTrim::default(), ) .await; @@ -5447,6 +5486,7 @@ async fn test_live_header_uses_configured_model_id_not_provider_returned() { ToolCoordinator::new(config.conversation.tools.clone(), empty_executor_source()), chat_request, InvocationContext::default(), + PendingStreamTrim::default(), ) .await .unwrap(); @@ -5561,6 +5601,7 @@ async fn reasoning_before_a_tool_call_shades_the_tool_chrome() { ToolCoordinator::new(config.conversation.tools.clone(), Box::new(executor_source)), ChatRequest::from("use the tool"), InvocationContext::default(), + PendingStreamTrim::default(), ) .await .unwrap(); diff --git a/crates/jp_cli/src/cmd/query_tests.rs b/crates/jp_cli/src/cmd/query_tests.rs index 93e4b9365..c016c3346 100644 --- a/crates/jp_cli/src/cmd/query_tests.rs +++ b/crates/jp_cli/src/cmd/query_tests.rs @@ -166,6 +166,7 @@ async fn run_mock_turn( tool::ToolCoordinator::new(cfg.conversation.tools.clone(), empty_executor_source()), ChatRequest::from(prompt), InvocationContext::default(), + PendingStreamTrim::default(), ) .await .unwrap(); @@ -1048,27 +1049,44 @@ fn edit_message_synthesizes_when_no_edit_without_query() { // Empty request and empty stream: a default "continue" message is // synthesized, so the caller must echo it. let mut request = ChatRequest::default(); - let mut stream = ConversationStream::new_test(); + let stream = ConversationStream::new_test(); + let mut pending_trim = PendingStreamTrim::default(); let (source, partial) = query - .edit_message(&mut request, &mut stream, false, &config, root) + .edit_message( + &mut request, + &stream, + &mut pending_trim, + false, + &config, + root, + ) .unwrap(); assert_eq!(source, QuerySource::Synthesized); assert_eq!(request.content, "continue"); assert!(partial.is_empty()); // Empty request with the stream's trailing event being a chat request: - // that request is consumed and re-sent verbatim, also synthesized. + // that request is peeked and re-sent verbatim, also synthesized. let mut request = ChatRequest::default(); let mut stream = ConversationStream::new_test(); stream.start_turn("earlier text"); + let mut pending_trim = PendingStreamTrim::default(); let (source, _) = query - .edit_message(&mut request, &mut stream, false, &config, root) + .edit_message( + &mut request, + &stream, + &mut pending_trim, + false, + &config, + root, + ) .unwrap(); assert_eq!(source, QuerySource::Synthesized); assert_eq!(request.content, "earlier text"); - // The trailing request was popped from the stream so the re-sent message - // isn't duplicated. - assert!(stream.pop_if(ConversationEvent::is_chat_request).is_none()); + // The trailing request is not popped here; its removal is deferred to the + // turn-start commit point via `pending_trim`, so the stream is untouched. + assert!(pending_trim.pop_request); + assert!(stream.pop_if(ConversationEvent::is_chat_request).is_some()); } #[test] @@ -1086,11 +1104,13 @@ fn edit_message_quote_without_editor_is_synthesized() { }; let mut request = ChatRequest::from(" > quoted reply"); - let mut stream = ConversationStream::new_test(); + let stream = ConversationStream::new_test(); + let mut pending_trim = PendingStreamTrim::default(); let (source, partial) = query .edit_message( &mut request, - &mut stream, + &stream, + &mut pending_trim, false, &config, Utf8Path::new("/tmp"), @@ -1213,3 +1233,95 @@ fn last_assistant_message_returns_none_when_only_reasoning_present() { assert_eq!(last_assistant_message(&stream), None); } + +/// Count the `TurnStart` events in a stream. +fn turn_start_count(stream: &ConversationStream) -> usize { + stream.iter().filter(|e| e.event.is_turn_start()).count() +} + +/// Assert that no `TurnStart` is immediately followed by another `TurnStart` +/// (an empty middle turn). +fn assert_no_adjacent_turn_starts(stream: &ConversationStream) { + let mut previous_was_turn_start = false; + for e in stream.iter() { + assert!( + !(previous_was_turn_start && e.event.is_turn_start()), + "stream contains adjacent TurnStart events (empty middle turn)" + ); + previous_was_turn_start = e.event.is_turn_start(); + } +} + +#[test] +fn pending_trim_replay_removes_stale_turn_start() { + // Multi-turn conversation: the stale `TurnStart` sits *after* the first + // `ChatRequest`, where `sanitize`'s `normalize_turn_starts` would not + // collapse it. + let mut stream = ConversationStream::new_test(); + stream.start_turn("first question"); + stream + .current_turn_mut() + .add_chat_response(ChatResponse::message("first answer")) + .build() + .unwrap(); + stream.start_turn("second question"); + + let trim = PendingStreamTrim { + replay_turn: true, + pop_request: false, + }; + trim.apply(&mut stream); + + // The replayed request re-enters the stream as a fresh turn. + stream.start_turn("second question, revised"); + + assert_eq!( + turn_start_count(&stream), + 2, + "replay must replace the trimmed turn, not open an extra one" + ); + assert_no_adjacent_turn_starts(&stream); +} + +#[test] +fn pending_trim_pop_request_removes_stale_turn_start() { + // Bare `--no-edit` replay: the last turn holds only its request. + let mut stream = ConversationStream::new_test(); + stream.start_turn("first question"); + stream + .current_turn_mut() + .add_chat_response(ChatResponse::message("first answer")) + .build() + .unwrap(); + stream.start_turn("replayed question"); + + let trim = PendingStreamTrim { + replay_turn: false, + pop_request: true, + }; + trim.apply(&mut stream); + + stream.start_turn("replayed question"); + + assert_eq!( + turn_start_count(&stream), + 2, + "pop_request must replace the trimmed turn, not open an extra one" + ); + assert_no_adjacent_turn_starts(&stream); +} + +#[test] +fn pending_trim_default_is_noop() { + let mut stream = ConversationStream::new_test(); + stream.start_turn("question"); + let before = stream.len(); + + PendingStreamTrim::default().apply(&mut stream); + + assert_eq!( + stream.len(), + before, + "a default PendingStreamTrim must not mutate the stream" + ); +} diff --git a/crates/jp_cli/src/lib_tests.rs b/crates/jp_cli/src/lib_tests.rs index 7a5f96266..f28f79b7b 100644 --- a/crates/jp_cli/src/lib_tests.rs +++ b/crates/jp_cli/src/lib_tests.rs @@ -348,6 +348,7 @@ fn query_model_override_persists_config_delta_through_run_inner() { unsafe { env::set_var("JP_GLOBAL_CONFIG_DIR", global_dir.as_str()) }; unsafe { env::set_var("JP_USER_DATA_DIR", user_data.as_str()) }; + unsafe { env::set_var("JP_TEST_DUMMY_OPENAI_API_KEY", "dummy") }; unsafe { env::remove_var("JP_EDITOR") }; unsafe { env::remove_var("VISUAL") }; unsafe { env::remove_var("EDITOR") }; @@ -379,6 +380,20 @@ fn query_model_override_persists_config_delta_through_run_inner() { &conversation_id.to_string(), "--model", "openai/gpt-4o", + // An inline query so the request builds without an editor and the + // failure happens at the LLM stage: the CLI config delta is only + // recorded once a non-empty request exists, so a query aborted + // before that point (e.g. by the credential preflight) deliberately + // leaves no config event behind. + "hello", + // Pass the credential preflight with a dummy key, then fail at the + // request stage against an unroutable loopback address so the test + // never touches the network, even on machines where a real + // `OPENAI_API_KEY` is set. + "--cfg", + "providers.llm.openai.api_key_env=JP_TEST_DUMMY_OPENAI_API_KEY", + "--cfg", + "providers.llm.openai.base_url=http://127.0.0.1:9", ]); let result = run_inner(cli, OutputFormat::TextPretty); @@ -415,6 +430,7 @@ fn query_model_override_persists_config_delta_through_run_inner() { env::set_current_dir(previous_cwd).unwrap(); unsafe { env::remove_var("JP_GLOBAL_CONFIG_DIR") }; unsafe { env::remove_var("JP_USER_DATA_DIR") }; + unsafe { env::remove_var("JP_TEST_DUMMY_OPENAI_API_KEY") }; match previous_jp_editor { Some(value) => unsafe { env::set_var("JP_EDITOR", value) }, @@ -447,6 +463,7 @@ fn query_model_override_persists_config_delta_through_session_targeting() { unsafe { env::set_var("JP_GLOBAL_CONFIG_DIR", global_dir.as_str()) }; unsafe { env::set_var("JP_USER_DATA_DIR", user_data.as_str()) }; unsafe { env::set_var("JP_SESSION", "jp-cli-test-session") }; + unsafe { env::set_var("JP_TEST_DUMMY_OPENAI_API_KEY", "dummy") }; unsafe { env::remove_var("JP_EDITOR") }; unsafe { env::remove_var("VISUAL") }; unsafe { env::remove_var("EDITOR") }; @@ -493,6 +510,15 @@ fn query_model_override_persists_config_delta_through_session_targeting() { "query", "--model", "openai/gpt-4o", + // See the run_inner variant above: an inline query moves the failure + // past the turn-start commit point so the delta is persisted, and + // the dummy key plus unroutable base URL make the LLM stage fail + // without touching the network. + "hello", + "--cfg", + "providers.llm.openai.api_key_env=JP_TEST_DUMMY_OPENAI_API_KEY", + "--cfg", + "providers.llm.openai.base_url=http://127.0.0.1:9", ]); let result = run_inner(cli, OutputFormat::TextPretty); @@ -529,6 +555,7 @@ fn query_model_override_persists_config_delta_through_session_targeting() { env::set_current_dir(previous_cwd).unwrap(); unsafe { env::remove_var("JP_GLOBAL_CONFIG_DIR") }; unsafe { env::remove_var("JP_USER_DATA_DIR") }; + unsafe { env::remove_var("JP_TEST_DUMMY_OPENAI_API_KEY") }; match previous_jp_session { Some(value) => unsafe { env::set_var("JP_SESSION", value) }, diff --git a/crates/jp_conversation/src/stream.rs b/crates/jp_conversation/src/stream.rs index e168fc441..471c07bf6 100644 --- a/crates/jp_conversation/src/stream.rs +++ b/crates/jp_conversation/src/stream.rs @@ -588,26 +588,36 @@ impl ConversationStream { .map(|event| ConversationEventWithConfig { event, config }) } - /// Similar to [`Self::pop`], but only pops if the predicate returns `true`. - pub fn pop_if( - &mut self, - f: impl Fn(&ConversationEvent) -> bool, - ) -> Option { - let last_turn_event_matches = self - .events + /// Returns the last turn-scoped [`ConversationEvent`] in the stream, + /// skipping trailing `ConfigDelta`/`Compaction` overlays — the event + /// [`Self::pop`] would remove, without removing it. + /// + /// Read-only counterpart of [`Self::pop`] and [`Self::pop_if`]. + /// Callers that intend to replay an event peek here and defer the + /// destructive pop to a later commit point, so an abandoned operation never + /// mutates the stream. + #[must_use] + pub fn last_turn_event(&self) -> Option<&ConversationEvent> { + self.events .iter() .rev() .find_map(|event| match event.scope() { - EventScope::Turn => event.as_event().map(&f), + EventScope::Turn => event.as_event(), EventScope::Global => None, }) - .unwrap_or(false); + } - if !last_turn_event_matches { - return None; + /// Similar to [`Self::pop`], but only pops if the predicate returns `true` + /// for the event [`Self::last_turn_event`] returns. + pub fn pop_if( + &mut self, + f: impl Fn(&ConversationEvent) -> bool, + ) -> Option { + if self.last_turn_event().is_some_and(f) { + self.pop() + } else { + None } - - self.pop() } /// Pop trailing events while `f` returns `true`, returning the removed diff --git a/crates/jp_conversation/src/stream_tests.rs b/crates/jp_conversation/src/stream_tests.rs index 0d6e08777..42b39a447 100644 --- a/crates/jp_conversation/src/stream_tests.rs +++ b/crates/jp_conversation/src/stream_tests.rs @@ -133,6 +133,22 @@ fn test_trim_trailing_empty_turn_removes_lone_turn_start() { assert_eq!(stream.len(), 0); } +#[test] +fn last_turn_event_skips_trailing_overlays() { + let mut stream = ConversationStream::new_test(); + stream.start_turn("hello"); + stream.add_compaction(Compaction::new(0, 0)); + + // The trailing compaction is an overlay; the peek must skip it and find + // the chat request. + assert!( + stream + .last_turn_event() + .is_some_and(ConversationEvent::is_chat_request), + "peek must skip the trailing overlay and find the chat request" + ); +} + #[test] fn pop_if_preserves_trailing_compaction() { let mut stream = ConversationStream::new_test(); diff --git a/crates/jp_workspace/src/conversation_lock.rs b/crates/jp_workspace/src/conversation_lock.rs index c25399bcd..f444b39c2 100644 --- a/crates/jp_workspace/src/conversation_lock.rs +++ b/crates/jp_workspace/src/conversation_lock.rs @@ -156,6 +156,18 @@ impl ConversationLock { self.events.read() } + /// Read the conversation event stream through a callback. + /// + /// Read-only counterpart of [`ConversationMut::update_events`]: the shared + /// guard is scoped to the callback (so it cannot be held across `.await` + /// points) and nothing is marked dirty, so no persist is triggered. + /// Use this instead of `as_mut().update_events(..)` whenever the callback + /// only needs to *read* the stream. + pub fn with_events(&self, f: impl FnOnce(&ConversationStream) -> R) -> R { + let guard = self.events.read(); + f(&guard) + } + /// Create a short-lived mutable scope. /// Persists on drop. /// @@ -252,6 +264,16 @@ impl ConversationMut { self.events.read() } + /// Read the conversation event stream through a callback. + /// + /// Read-only counterpart of [`Self::update_events`]: the shared guard is + /// scoped to the callback and the dirty flag is untouched, so no persist is + /// triggered on drop. + pub fn with_events(&self, f: impl FnOnce(&ConversationStream) -> R) -> R { + let guard = self.events.read(); + f(&guard) + } + /// Mutate conversation metadata through a callback. /// /// The write guard is acquired for the duration of the callback and