diff --git a/crates/jp_cli/src/cmd/query.rs b/crates/jp_cli/src/cmd/query.rs index 6cae552b..dd98f982 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 1283e44b..abc8cd7e 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 471ce731..e76124d1 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 93e4b936..c016c334 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 7a5f9626..f28f79b7 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 e168fc44..471c07bf 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 0d6e0877..42b39a44 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 c25399bc..f444b39c 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