fix(cli): Re-read the conversation index for each plugin read

crates/jp_cli/src/cmd/query.rs

@@ -557,61 +557,20 @@ impl Query {
             }
         }
 
-        // Wait for all MCP servers to finish loading, showing a timer line
-        // when the wait takes long enough to be noticeable.
-        await_mcp_servers(
+        let inputs = TurnInputs::collect(
+            ctx,
+            cfg.clone(),
+            &lock,
+            chat_request,
+            pending_trim,
             mcp_servers_handle,
-            cfg.style.mcp_startup.clone(),
+            // Typed at this terminal, so it renders here.
             ctx.printer.clone(),
-            ctx.term.is_tty,
-            ctx.term.width,
         )
         .await?;
 
-        let forced_tool = cfg.assistant.tool_choice.function_name();
-        let tools =
-            tool_definitions(cfg.conversation.tools.iter(), &ctx.mcp_client, forced_tool).await?;
-
-        let attachment_urls: Vec<_> = cfg
-            .conversation
-            .attachments
-            .iter()
-            .map(jp_config::conversation::attachment::AttachmentConfig::to_url)
-            .collect::<std::result::Result<Vec<_>, _>>()?;
-        let attachments = load_conversation_attachments(ctx, attachment_urls).await?;
-
-        debug!(count = attachments.len(), "Attachments loaded.");
-
-        let thread = build_thread(stream, attachments, &cfg.assistant, !tools.is_empty())?;
-        let root = ctx.workspace.root().to_path_buf();
-        let approvals = Arc::new(load_approval_store(ctx.fs_backend.as_deref()));
-
-        // Sanitize any structural issues (orphaned tool calls, missing
-        // user messages, etc.) before sending the stream to the provider.
-        lock.as_mut().update_events(ConversationStream::sanitize);
-
-        let invocation = InvocationContext {
-            workspace_id: ctx.workspace.id().to_string(),
-            conversation_id: lock.id().to_string(),
-        };
-
-        let turn_result = self
-            .handle_turn(
-                &cfg,
-                &ctx.signals,
-                &ctx.mcp_client,
-                root,
-                ctx.term.is_tty,
-                &thread.attachments,
-                &lock,
-                cfg.assistant.tool_choice.clone(),
-                &tools,
-                ctx.printer.clone(),
-                approvals,
-                chat_request,
-                invocation,
-                pending_trim,
-            )
+        let turn_result = inputs
+            .run(&lock, stream)
             .await
             .map_err(|error| cmd::Error::from(error).with_persistence(true));
 
@@ -897,10 +856,15 @@ impl Query {
         Ok((QuerySource::Editor, editor_provided_config))
     }
 
-    /// Handle a single turn of conversation with the LLM.
+    /// Run one turn of a conversation against the LLM.
+    ///
+    /// Takes no CLI state: everything it needs is an argument, so a caller that
+    /// isn't the `query` command can drive a turn too.
+    ///
+    /// The `lock` is proof the conversation is held for the duration (RFD 020),
+    /// and it owns what it needs, so nothing here borrows the workspace.
     #[expect(clippy::too_many_arguments)]
-    async fn handle_turn(
-        &self,
+    pub(crate) async fn run_turn(
         cfg: &AppConfig,
         signals: &SignalRouter,
         mcp_client: &jp_mcp::Client,
@@ -1101,6 +1065,143 @@ impl Query {
     }
 }
 
+/// Everything a turn needs, owned.
+///
+/// Collecting reads the context; running borrows nothing from it.
+/// That split is the point: a caller can start a turn and then get on with
+/// other work, because the running half holds nothing belonging to whoever
+/// started it.
+pub(crate) struct TurnInputs {
+    config: Arc<AppConfig>,
+
+    /// A handle onto the process router, not a second router, so a handler this
+    /// turn pushes is one the signal task reaches.
+    signals: SignalRouter,
+    mcp_client: jp_mcp::Client,
+    root: Utf8PathBuf,
+    is_tty: bool,
+    width: Option<u16>,
+    attachments: Vec<Attachment>,
+    printer: Arc<Printer>,
+    approvals: Arc<ApprovalStore>,
+    chat_request: ChatRequest,
+    invocation: InvocationContext,
+    pending_trim: PendingStreamTrim,
+
+    /// MCP servers starting in the background, awaited by [`TurnInputs::run`].
+    mcp_servers: StartupSet,
+}
+
+impl TurnInputs {
+    /// Collect what a turn needs from the context.
+    ///
+    /// Deliberately does nothing slow.
+    /// Everything that waits, MCP servers coming up and the provider, happens
+    /// in [`Self::run`], which borrows nothing from the context.
+    /// A caller serving other work off the same task is blocked for exactly as
+    /// long as this takes.
+    ///
+    /// Attachment loading is the one wait that has to stay here, because it
+    /// reads conversations out of the workspace.
+    /// Local reads, not process startup.
+    ///
+    /// `printer` is where the turn's output goes.
+    /// A turn typed at a terminal renders to it; a turn asked for from
+    /// elsewhere has no reader there, and a sink is how it stops writing into a
+    /// session it does not belong to.
+    pub(crate) async fn collect(
+        ctx: &Ctx,
+        config: Arc<AppConfig>,
+        lock: &ConversationLock,
+        chat_request: ChatRequest,
+        pending_trim: PendingStreamTrim,
+        mcp_servers: StartupSet,
+        printer: Arc<Printer>,
+    ) -> Result<Self> {
+        let attachment_urls: Vec<_> = config
+            .conversation
+            .attachments
+            .iter()
+            .map(jp_config::conversation::attachment::AttachmentConfig::to_url)
+            .collect::<std::result::Result<Vec<_>, _>>()?;
+        let attachments = load_conversation_attachments(ctx, attachment_urls).await?;
+
+        debug!(count = attachments.len(), "Attachments loaded.");
+
+        Ok(Self {
+            root: ctx.workspace.root().to_path_buf(),
+            approvals: Arc::new(load_approval_store(ctx.fs_backend.as_deref())),
+            invocation: InvocationContext {
+                workspace_id: ctx.workspace.id().to_string(),
+                conversation_id: lock.id().to_string(),
+            },
+            signals: ctx.signals.clone(),
+            mcp_client: ctx.mcp_client.clone(),
+            printer,
+            is_tty: ctx.term.is_tty,
+            width: ctx.term.width,
+            attachments,
+            mcp_servers,
+            chat_request,
+            pending_trim,
+            config,
+        })
+    }
+
+    /// Finish preparing, then run the turn.
+    ///
+    /// Borrows only the lock, which owns itself, so this can run on a task of
+    /// its own, which is where the waiting belongs.
+    /// `stream` is the snapshot the thread is assembled from.
+    pub(crate) async fn run(
+        self,
+        lock: &ConversationLock,
+        stream: ConversationStream,
+    ) -> Result<()> {
+        let cfg = &self.config;
+
+        // Wait for all MCP servers to finish loading, showing a timer line when the
+        // wait takes long enough to be noticeable. Starting a server can mean
+        // compiling one, so this is not a wait to hold anything else up for.
+        await_mcp_servers(
+            self.mcp_servers,
+            cfg.style.mcp_startup.clone(),
+            self.printer.clone(),
+            self.is_tty,
+            self.width,
+        )
+        .await?;
+
+        let forced_tool = cfg.assistant.tool_choice.function_name();
+        let tools =
+            tool_definitions(cfg.conversation.tools.iter(), &self.mcp_client, forced_tool).await?;
+
+        let thread = build_thread(stream, self.attachments, &cfg.assistant, !tools.is_empty())?;
+
+        // Sanitize any structural issues (orphaned tool calls, missing user
+        // messages, etc.) before sending the stream to the provider.
+        lock.as_mut().update_events(ConversationStream::sanitize);
+
+        Query::run_turn(
+            cfg,
+            &self.signals,
+            &self.mcp_client,
+            self.root,
+            self.is_tty,
+            &thread.attachments,
+            lock,
+            cfg.assistant.tool_choice.clone(),
+            &tools,
+            self.printer,
+            self.approvals,
+            self.chat_request,
+            self.invocation,
+            self.pending_trim,
+        )
+        .await
+    }
+}
+
 /// Wait for background MCP server startups to complete.
 ///
 /// Shows a single aggregate timer line on stderr once the wait exceeds the
@@ -1661,7 +1762,13 @@ fn apply_title_override(lock: &ConversationLock, title: Option<&str>, no_title:
     }
 }
 
-fn get_config_delta_from_cli(
+/// What a conversation would have to record to arrive at `cfg`.
+///
+/// The difference between the configuration the stream already resolves to and
+/// the one a turn is about to run under, which is what `--cfg` amounts to: not
+/// a setting for one turn, but a change from this point on.
+/// `None` when the two agree and there is nothing to record.
+pub(crate) fn get_config_delta_from_cli(
     cfg: &AppConfig,
     lock: &ConversationLock,
 ) -> Result<Option<PartialAppConfig>> {

crates/jp_cli/src/cmd/query/turn_loop.rs

@@ -202,7 +202,13 @@ pub(super) async fn run_turn_loop(
     // thread building, response processing) and receives interrupts the inner
     // streaming/tool handlers decline. Its notifications are consumed at the
     // top of each phase-loop iteration; the guard drops when the turn ends.
-    let (_turn_interrupt_guard, mut turn_interrupt_rx) = signals.push_handler();
+    //
+    // Registered under the conversation, so an interrupt that names one reaches
+    // that turn rather than whichever happens to be topmost. Several turns can be
+    // in flight at once when something other than a terminal is driving them, and
+    // a Ctrl-C's "whatever is in front of me" is the wrong guess for a request
+    // that already said which.
+    let (_turn_interrupt_guard, mut turn_interrupt_rx) = signals.push_handler_for(lock.id());
 
     let mut turn_state = TurnState::default();
     let mut stream_retry = StreamRetryState::new(cfg.assistant.request, is_tty);
@@ -571,6 +577,11 @@ pub(super) async fn run_turn_loop(
 
                             let is_finished = matches!(event, Event::Finished(_));
 
+                            // A `Flush` is the provider saying a content block is
+                            // final, which makes it the point where the stream is
+                            // consistent enough to persist.
+                            let commits_content = matches!(event, Event::Flush { .. });
+
                             // `handle_llm_event` returns turn control plus
                             // any newly committed event that needs immediate
                             // shell handling. We dispatch on the committed
@@ -585,6 +596,26 @@ pub(super) async fn run_turn_loop(
                                     &mut stream_retry,
                                 )
                             });
+
+                            // Persist each completed block and each tool call as it
+                            // lands, rather than once the phase ends.
+                            //
+                            // Durability is one reason: a process that dies mid-turn
+                            // keeps everything the assistant had finished saying,
+                            // instead of losing the whole phase. The other is that
+                            // anything reading the conversation from outside this
+                            // process sees the turn progress, which is what lets a
+                            // second frontend watch a turn rather than only its
+                            // result.
+                            //
+                            // A failed write is not fatal: the phase-end flush will
+                            // try again, and the in-memory stream is still correct.
+                            let commits = commits_content
+                                || matches!(committed, CommittedEvent::ToolCallRequest(_));
+                            if commits && let Err(error) = conv.flush() {
+                                warn!(%error, "Failed to persist mid-turn; will retry.");
+                            }
+
                             match action {
                                 LoopAction::Continue => {}
                                 LoopAction::Break => break,