diff --git a/crates/jp_cli/src/cmd/query/stream.rs b/crates/jp_cli/src/cmd/query/stream.rs index 857d4edf..5f6aefd5 100644 --- a/crates/jp_cli/src/cmd/query/stream.rs +++ b/crates/jp_cli/src/cmd/query/stream.rs @@ -6,8 +6,8 @@ pub(crate) mod retry; pub(crate) use retry::{ - RebuildRefusal, StreamErrorOutcome, StreamRetryState, commit_partial_response, - handle_stream_error, + RebuildRefusal, ResponseBoundary, StreamErrorOutcome, StreamRetryState, + commit_partial_response, handle_stream_error, }; pub(crate) use crate::render::TurnView; diff --git a/crates/jp_cli/src/cmd/query/stream/retry.rs b/crates/jp_cli/src/cmd/query/stream/retry.rs index ff6746dc..fd44ad38 100644 --- a/crates/jp_cli/src/cmd/query/stream/retry.rs +++ b/crates/jp_cli/src/cmd/query/stream/retry.rs @@ -255,6 +255,17 @@ impl StreamRetryState { } } +/// Whether the assistant's response continues past a flush boundary. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ResponseBoundary { + /// The request is resent and the same response continues on screen, with + /// nothing persistent rendered in between. + Continuation, + + /// The response ends here: nothing more of it will be rendered. + Final, +} + /// Flush buffered output and commit any unflushed partial assistant content to /// the conversation stream. /// @@ -263,12 +274,22 @@ impl StreamRetryState { /// user already saw on screen. /// Call this before ending a turn on any path that bypasses the coordinator's /// own terminal handling. +/// +/// `boundary` decides how the renderer treats an open reasoning region. +/// A continuation keeps the gap the region owes pending for the resent +/// request's output to resolve (the retry notification is a transient line that +/// leaves nothing behind); a final boundary closes the region with an unshaded +/// gap. pub fn commit_partial_response( turn_coordinator: &mut TurnCoordinator, conv: &ConversationMut, printer: &Arc, + boundary: ResponseBoundary, ) { - turn_coordinator.flush_renderer(); + match boundary { + ResponseBoundary::Continuation => turn_coordinator.flush_renderer_for_continuation(), + ResponseBoundary::Final => turn_coordinator.flush_renderer(), + } printer.flush_instant(); let partial = turn_coordinator.peek_partial_events(); @@ -320,9 +341,17 @@ pub async fn handle_stream_error( // to the stream BEFORE deciding whether to retry or abort. Streamed text // the user already saw must never be dropped just because the error turned // out to be fatal. - commit_partial_response(turn_coordinator, conv, printer); + // The retry decision does feed the flush: an aborted response ends its + // reasoning region here, while a retry hands the region to the resent + // request. + let boundary = if retry_state.can_retry(&error) { + ResponseBoundary::Continuation + } else { + ResponseBoundary::Final + }; + commit_partial_response(turn_coordinator, conv, printer, boundary); - if !retry_state.can_retry(&error) { + if boundary == ResponseBoundary::Final { // Clear the temp line before printing the final error so it doesn't // linger on screen. retry_state.clear_line(printer); @@ -337,7 +366,7 @@ pub async fn handle_stream_error( // Reset the coordinator for the next streaming cycle. The committed partial // response becomes continuation context in the rebuilt Thread; the Provider // decides how to encode it for the target model. - turn_coordinator.prepare_continuation(); + turn_coordinator.prepare_retry_continuation(); // Notify the user. let attempt = retry_state.consecutive_failures; diff --git a/crates/jp_cli/src/cmd/query/stream/retry_tests.rs b/crates/jp_cli/src/cmd/query/stream/retry_tests.rs index a0fbbfa9..28f015a2 100644 --- a/crates/jp_cli/src/cmd/query/stream/retry_tests.rs +++ b/crates/jp_cli/src/cmd/query/stream/retry_tests.rs @@ -9,7 +9,7 @@ use jp_conversation::{ event::{ChatRequest, ChatResponse}, }; use jp_llm::{StreamError, event::Event}; -use jp_printer::{OutputFormat, Printer}; +use jp_printer::{OutputFormat, Printer, SharedBuffer}; use jp_workspace::{ConversationLock, Workspace}; use super::*; @@ -38,6 +38,22 @@ fn make_turn_coordinator() -> TurnCoordinator { ) } +/// Create a coordinator that shares its printer with the caller, so a test can +/// read what the renderer wrote. +fn make_turn_coordinator_with_output() -> (TurnCoordinator, Arc, SharedBuffer) { + let (printer, out, _err) = Printer::memory(OutputFormat::TextPretty); + let printer = Arc::new(printer); + let coordinator = TurnCoordinator::new( + Arc::clone(&printer), + AppConfig::new_test().style, + None, + None, + None, + ); + + (coordinator, printer, out) +} + /// Create a workspace with a single conversation and return a test lock. fn make_test_lock() -> (Workspace, ConversationLock) { let config = Arc::new(AppConfig::new_test()); @@ -332,6 +348,45 @@ async fn retry_without_partial_content_still_works() { ); } +/// A fatal error ends the response, so the gap the interrupted reasoning block +/// owes closes its region: the reasoning background must stop before it, not +/// leave a shaded strip under the error message. +#[tokio::test] +async fn fatal_error_after_reasoning_leaves_the_gap_unshaded() { + let (mut turn_coordinator, printer, out) = make_turn_coordinator_with_output(); + let mut retry_state = make_retry_state(3); + let (_ws, lock) = make_test_lock(); + let conv = lock.as_mut(); + conv.update_events(|stream| { + turn_coordinator.start_turn(stream, ChatRequest::from("test")); + }); + + conv.update_events(|stream| { + turn_coordinator.handle_event(stream, Event::reasoning(0, "Thinking.\n\n")); + }); + + let router = detached_router(); + let result = handle_stream_error( + StreamError::other("auth failure"), + &mut retry_state, + &mut turn_coordinator, + &conv, + &printer, + &router, + ) + .await; + + assert!(matches!(result, StreamErrorOutcome::Fatal(_))); + + printer.flush(); + assert_eq!( + out.lock().clone(), + "\n── \u{1b}[1mjp\u{1b}[0m \ + ──────────────────────────────────────────────────────────────────────────\n\n\u{1b}[48;\ + 5;236mThinking.\u{1b}[48;5;236m\u{1b}[K\u{1b}[0m\n\n" + ); +} + #[tokio::test] async fn interrupt_during_backoff_cuts_wait_short() { let (printer, _out, _err) = Printer::memory(OutputFormat::TextPretty); diff --git a/crates/jp_cli/src/cmd/query/turn/coordinator.rs b/crates/jp_cli/src/cmd/query/turn/coordinator.rs index 67b4539e..58b8a2f3 100644 --- a/crates/jp_cli/src/cmd/query/turn/coordinator.rs +++ b/crates/jp_cli/src/cmd/query/turn/coordinator.rs @@ -422,9 +422,27 @@ impl TurnCoordinator { /// The Provider decides whether the target model accepts native assistant /// prefill or needs another supported wire representation. pub fn prepare_continuation(&mut self) { + self.prepare_next_cycle(); + self.view.reset_for_continuation(); + } + + /// Reset per-request state before resending a request a stream error cut + /// short. + /// + /// Like [`prepare_continuation`], but keeps the renderer's content region + /// open: a retry puts nothing persistent on the terminal, so the reasoning + /// region and the separator it owes span the boundary. + /// + /// [`prepare_continuation`]: Self::prepare_continuation + pub fn prepare_retry_continuation(&mut self) { + self.prepare_next_cycle(); + self.view.reset_for_stream_retry(); + } + + /// Drop the per-request event buffer and re-enter the streaming phase. + fn prepare_next_cycle(&mut self) { // The committed partial response replaces these per-request buffers. self.event_builder = EventBuilder::new(); - self.view.reset_for_continuation(); self.state = TurnPhase::Streaming; } @@ -437,6 +455,19 @@ impl TurnCoordinator { self.view.flush(); } + /// Flush the renderer at a streaming-cycle boundary the same response + /// continues across. + /// + /// Commits buffered content like [`flush_renderer`], but leaves a separator + /// owed by reasoning pending: the continuation resends the request with no + /// persistent content rendered in between, so its first content decides the + /// gap's shading. + /// + /// [`flush_renderer`]: Self::flush_renderer + pub fn flush_renderer_for_continuation(&mut self) { + self.view.flush_for_continuation(); + } + /// Resolve the live tool-call boundary, returning the background the tool's /// chrome should be filled with to keep a reasoning region continuous. /// diff --git a/crates/jp_cli/src/cmd/query/turn_loop.rs b/crates/jp_cli/src/cmd/query/turn_loop.rs index a751b030..3d3183b9 100644 --- a/crates/jp_cli/src/cmd/query/turn_loop.rs +++ b/crates/jp_cli/src/cmd/query/turn_loop.rs @@ -49,7 +49,10 @@ use super::{ LoopAction, StreamingInterruptResult, handle_llm_event, handle_streaming_interrupt, reply_edit_mode, }, - stream::{StreamErrorOutcome, StreamRetryState, commit_partial_response, handle_stream_error}, + stream::{ + ResponseBoundary, StreamErrorOutcome, StreamRetryState, commit_partial_response, + handle_stream_error, + }, tool::{ PendingEntry, PendingTools, ToolCallDecision, ToolCallState, ToolCoordinator, ToolPrompter, ToolRenderer, build_execution_plan, @@ -595,7 +598,12 @@ pub(super) async fn run_turn_loop( // buffered output, which would otherwise land // after the parked cursor. stream_retry.clear_line(&printer); - commit_partial_response(&mut turn_coordinator, &conv, &printer); + commit_partial_response( + &mut turn_coordinator, + &conv, + &printer, + ResponseBoundary::Final, + ); if let Err(err) = conv.flush() { warn!("Failed to persist before abort: {err}"); } diff --git a/crates/jp_cli/src/render/chat.rs b/crates/jp_cli/src/render/chat.rs index ace3c643..074ee28b 100644 --- a/crates/jp_cli/src/render/chat.rs +++ b/crates/jp_cli/src/render/chat.rs @@ -664,6 +664,23 @@ impl ChatRenderer { self.emit_pending_separator(false); } + /// Commit buffered content at a streaming-cycle boundary the same response + /// continues across, leaving the deferred separator unresolved. + /// + /// A stream error commits what was streamed and resends the request, with + /// nothing persistent rendered in between. + /// A separator owed by reasoning therefore stays pending and the + /// continuation's first content decides its shading, exactly as the next + /// block would inside one streaming cycle. + /// Pair with [`reset_preserving_region`], which carries the pending + /// separator across the reset the continuation performs. + /// + /// [`reset_preserving_region`]: Self::reset_preserving_region + pub fn flush_for_continuation(&mut self) { + self.cancel_reasoning_timer(); + self.drain_buffer(); + } + /// Drain the buffer's end-of-region events to the printer, committing /// buffered blocks and closing any open code block. /// @@ -833,6 +850,24 @@ impl ChatRenderer { self.para_source.clear(); self.para_emitted = 0; } + + /// Reset the renderer state, keeping the content region open. + /// + /// Used at a streaming-cycle boundary the same response continues across: + /// the deferred separator survives, along with the content kinds that + /// decide its shading, so the continuation's first content resolves the gap + /// the interrupted block owed. + pub fn reset_preserving_region(&mut self) { + let last_content_kind = self.last_content_kind; + let last_response_kind = self.last_response_kind; + let pending_separator = self.pending_separator; + + self.reset(); + + self.last_content_kind = last_content_kind; + self.last_response_kind = last_response_kind; + self.pending_separator = pending_separator; + } } /// Build a labeled horizontal rule used as a role-boundary marker. diff --git a/crates/jp_cli/src/render/chat_tests.rs b/crates/jp_cli/src/render/chat_tests.rs index 8e2da0c7..e33adf37 100644 --- a/crates/jp_cli/src/render/chat_tests.rs +++ b/crates/jp_cli/src/render/chat_tests.rs @@ -573,6 +573,67 @@ fn test_reasoning_block_gap_is_shaded() { ); } +/// A stream error mid-reasoning commits what was streamed and resets the +/// renderer before the continuation request goes out. +/// The reasoning region spans that boundary, so the gap the interrupted block +/// owes stays inside the region and keeps its background. +#[test] +fn test_reasoning_gap_across_a_continuation_is_shaded() { + let mut config = AppConfig::new_test(); + config.style.reasoning.display = ReasoningDisplayConfig::Full; + config.style.reasoning.background = Some(Color::Ansi256(236)); + let (mut renderer, out, _err) = create_renderer_with_config(config); + + renderer.render_response(&ChatResponse::Reasoning { + reasoning: "First section.\n\n".into(), + }); + renderer.flush_for_continuation(); + renderer.reset_preserving_region(); + renderer.render_response(&ChatResponse::Reasoning { + reasoning: "Second section.\n\n".into(), + }); + renderer.flush(); + renderer.printer.flush(); + + let output = out.lock().clone(); + assert_eq!( + output, + "\u{1b}[48;5;236mFirst \ + section.\u{1b}[48;5;236m\u{1b}[K\u{1b}[0m\n\u{1b}[48;5;236m\u{1b}[K\u{1b}[49m\n\u{1b}[48;\ + 5;236mSecond section.\u{1b}[48;5;236m\u{1b}[K\u{1b}[0m\n\n" + ); +} + +/// The provider is free to resume an interrupted reasoning block with the +/// answer instead of more reasoning. +/// The gap then leaves the reasoning region, so it is unshaded — the same +/// output the reasoning-to-answer transition produces without a retry in +/// between. +#[test] +fn test_reasoning_gap_across_a_continuation_into_a_message_is_unshaded() { + let mut config = AppConfig::new_test(); + config.style.reasoning.display = ReasoningDisplayConfig::Full; + config.style.reasoning.background = Some(Color::Ansi256(236)); + let (mut renderer, out, _err) = create_renderer_with_config(config); + + renderer.render_response(&ChatResponse::Reasoning { + reasoning: "First section.\n\n".into(), + }); + renderer.flush_for_continuation(); + renderer.reset_preserving_region(); + renderer.render_response(&ChatResponse::Message { + message: "Answer.\n\n".into(), + }); + renderer.flush(); + renderer.printer.flush(); + + let output = out.lock().clone(); + assert_eq!( + output, + "\u{1b}[48;5;236mFirst section.\u{1b}[48;5;236m\u{1b}[K\u{1b}[0m\n\nAnswer.\n\n" + ); +} + #[test] fn test_message_buffer_flushed_on_explicit_flush() { let (mut renderer, out, _err) = create_renderer(); diff --git a/crates/jp_cli/src/render/turn_view.rs b/crates/jp_cli/src/render/turn_view.rs index 876ef5fa..86531f56 100644 --- a/crates/jp_cli/src/render/turn_view.rs +++ b/crates/jp_cli/src/render/turn_view.rs @@ -248,6 +248,17 @@ impl TurnView { self.structured.flush(); } + /// Flush pending output at a streaming-cycle boundary the same response + /// continues across. + /// + /// The chat renderer keeps its content region open, leaving a separator + /// owed by reasoning pending: nothing persistent renders between the two + /// cycles, so the continuation's first content decides the gap's shading. + pub fn flush_for_continuation(&mut self) { + self.chat.flush_for_continuation(); + self.structured.flush(); + } + /// Signal to the printer that the current streaming cycle has ended. /// /// Forwards to the chat renderer, which switches the printer's @@ -272,6 +283,17 @@ impl TurnView { self.structured.reset(); } + /// Reset internal renderer state at a stream-retry boundary, keeping the + /// chat renderer's content region open. + /// + /// A retry resends the request without rendering anything persistent in + /// between, so the reasoning region and the separator it owes span the + /// boundary and the continuation's first content resolves the gap. + pub fn reset_for_stream_retry(&mut self) { + self.chat.reset_preserving_region(); + self.structured.reset(); + } + /// Replace the underlying renderers and identity. /// /// Used by replay's per-turn config rebuild when the conversation's