From d64acacfb04d6868a6648419bfebc10c01023860 Mon Sep 17 00:00:00 2001 From: Jean Mertz Date: Fri, 31 Jul 2026 12:03:59 +0200 Subject: [PATCH 1/4] fix(llm, cli, anthropic): Bound provider rebuild loop and heal turns Anthropic rejects a thinking block whenever any thinking or redacted-thinking block in the latest assistant message was rewritten, not only when a signature goes stale. The repair now recognizes both rejection messages and, when the offending block sits in the latest assistant turn, downgrades every thinking block in that message in one patch set instead of one block per round. Anthropic requires that message's thinking blocks to come back unchanged as a whole, so patching one block at a time left the message half-rewritten and the next request failed the same way. A conversation already left in that half-rewritten state is now healed by the same patch. Earlier turns keep their narrower, single-block repair since they carry no such constraint. A provider that keeps asking to rebuild the request without actually fixing anything could previously loop forever, resending the full conversation each time. `jp query` now refuses a rebuild unless the preceding patch changed the stream in a way guaranteed to make progress (`PatchAction::shrinks_stream`), and caps consecutive rebuilds within a turn at 10. Hitting either limit aborts the turn with an error instead of looping, after flushing whatever was already streamed so no partial response is lost. The counter resets once a turn streams a successful event, so later, unrelated repairs are not penalized by an earlier one. Signed-off-by: Jean Mertz --- .../jp_cli/src/cmd/conversation/summarize.rs | 2 +- .../jp_cli/src/cmd/query/interrupt/signals.rs | 35 +- .../src/cmd/query/interrupt/signals_tests.rs | 280 +++++++++++++- crates/jp_cli/src/cmd/query/stream.rs | 2 +- crates/jp_cli/src/cmd/query/stream/retry.rs | 102 ++++- crates/jp_cli/src/cmd/query/turn_loop.rs | 22 +- crates/jp_llm/src/event.rs | 21 ++ crates/jp_llm/src/provider/anthropic.rs | 156 ++++++-- crates/jp_llm/src/provider/anthropic_tests.rs | 357 +++++++++++++++++- 9 files changed, 932 insertions(+), 45 deletions(-) diff --git a/crates/jp_cli/src/cmd/conversation/summarize.rs b/crates/jp_cli/src/cmd/conversation/summarize.rs index 80a2d0fa..e5b0385f 100644 --- a/crates/jp_cli/src/cmd/conversation/summarize.rs +++ b/crates/jp_cli/src/cmd/conversation/summarize.rs @@ -105,7 +105,7 @@ pub async fn generate_summary( /// A provider that answers with [`FinishReason::Retry`] supplies patches that /// make the request acceptable; each round applies them to `stream` and /// resends. -/// Providers degrade one bad event per round (see Anthropic's +/// Providers degrade a bounded subset of bad events per round (see Anthropic's /// `build_thinking_patches` and Google's `build_thought_signature_patch`), so a /// stream carrying several bad events legitimately needs several rounds. /// diff --git a/crates/jp_cli/src/cmd/query/interrupt/signals.rs b/crates/jp_cli/src/cmd/query/interrupt/signals.rs index e5dc5b54..f5920466 100644 --- a/crates/jp_cli/src/cmd/query/interrupt/signals.rs +++ b/crates/jp_cli/src/cmd/query/interrupt/signals.rs @@ -20,7 +20,10 @@ use tokio_util::sync::CancellationToken; use tracing::{info, trace}; use super::handler::{InterruptAction, InterruptHandler}; -use crate::cmd::query::turn::{Action, CommittedEvent, TurnCoordinator, TurnPhase}; +use crate::cmd::query::{ + stream::{RebuildRefusal, StreamRetryState}, + turn::{Action, CommittedEvent, TurnCoordinator, TurnPhase}, +}; /// Action to take in a select loop. /// @@ -32,6 +35,10 @@ pub enum LoopAction { /// Break the inner loop. Break, + + /// The provider asked to rebuild the request and was refused. + /// The turn cannot make progress and must abort. + RebuildRefused(RebuildRefusal), } /// Result of handling an interrupt during LLM streaming. @@ -126,15 +133,36 @@ pub fn handle_llm_event( event: Event, turn_coordinator: &mut TurnCoordinator, conversation_stream: &mut ConversationStream, + retry_state: &mut StreamRetryState, ) -> (LoopAction, CommittedEvent) { // `Patch` is a side-channel instruction from the provider to fix bad events // in the conversation stream. This can be handled directly instead of // passing through the turn coordinator. if let Event::Patch(patches) = event { let count = apply_patches(conversation_stream, &patches); + let shrinks = patches.iter().all(|patch| patch.action.shrinks_stream()); + retry_state.record_patch(count, shrinks); + + if !shrinks { + // A patch set that can grow the stream gives no guarantee the repair + // loop ends, so the rebuild below is refused whatever it changed. + tracing::warn!( + patches = patches.len(), + "History patches include an action that may not shrink the conversation stream." + ); + } + if count > 0 { tracing::debug!(count, "Applied history patches to conversation stream."); + } else { + // The rebuilt request would be byte-identical, so the rebuild that + // follows is refused below rather than resent. + tracing::warn!( + patches = patches.len(), + "History patches matched no events in the conversation stream." + ); } + return (LoopAction::Continue, CommittedEvent::None); } @@ -142,7 +170,10 @@ pub fn handle_llm_event( // Break the inner streaming loop while keeping the phase as `Streaming` so // the outer turn loop re-enters with a fresh request. if matches!(event, Event::Finished(FinishReason::Retry)) { - return (LoopAction::Break, CommittedEvent::None); + return match retry_state.authorize_rebuild() { + Ok(()) => (LoopAction::Break, CommittedEvent::None), + Err(refusal) => (LoopAction::RebuildRefused(refusal), CommittedEvent::None), + }; } let outcome = turn_coordinator.handle_event(conversation_stream, event); diff --git a/crates/jp_cli/src/cmd/query/interrupt/signals_tests.rs b/crates/jp_cli/src/cmd/query/interrupt/signals_tests.rs index 32b7cb0b..6ba5ae81 100644 --- a/crates/jp_cli/src/cmd/query/interrupt/signals_tests.rs +++ b/crates/jp_cli/src/cmd/query/interrupt/signals_tests.rs @@ -1,12 +1,20 @@ use std::sync::Arc; use assert_matches::assert_matches; -use jp_config::AppConfig; -use jp_conversation::{ConversationStream, event::ChatRequest}; +use jp_config::{ + AppConfig, + assistant::request::{CachePolicy, RequestConfig}, +}; +use jp_conversation::{ + ConversationEvent, ConversationStream, + event::{ChatRequest, ChatResponse}, +}; use jp_inquire::{ReplyEditMode, ReplyOutcome, prompt::MockPromptBackend}; +use jp_llm::event::{EventMatcher, EventPatch, FinishReason, PatchAction}; use jp_printer::{OutputFormat, Printer}; use super::*; +use crate::cmd::query::stream::retry::MAX_CONSECUTIVE_REBUILDS; fn make_printer() -> Printer { let (printer, _out, _err) = Printer::memory(OutputFormat::TextPretty); @@ -32,6 +40,274 @@ fn make_turn_coordinator() -> TurnCoordinator { ) } +fn make_retry_state(max_retries: u32) -> StreamRetryState { + let config = RequestConfig { + max_retries, + base_backoff_ms: 1, + max_backoff_secs: 1, + stream_idle_timeout_secs: 120, + cache: CachePolicy::default(), + }; + StreamRetryState::new(config, false) +} + +/// The metadata a provider patch targets in these tests. +const PATCH_KEY: &str = "provider_signature"; +const PATCH_VALUE: &str = "stale"; + +/// A stream holding one reasoning event that carries patchable metadata. +fn stream_with_patchable_event() -> ConversationStream { + let mut stream = ConversationStream::new_test(); + let mut event = ConversationEvent::now(ChatResponse::reasoning("thinking")); + event.metadata.insert(PATCH_KEY.into(), PATCH_VALUE.into()); + + stream + .current_turn_mut() + .add_event(event) + .build() + .expect("valid stream"); + + stream +} + +fn stale_signature_patch() -> Event { + Event::Patch(vec![EventPatch { + matcher: EventMatcher::MetadataValue { + key: PATCH_KEY.into(), + value: PATCH_VALUE.into(), + }, + action: PatchAction::RemoveMetadata(PATCH_KEY.into()), + }]) +} + +/// A patch that changed the stream earns the rebuild that follows it: the +/// request the provider gets next is genuinely different. +#[test] +fn rebuild_allowed_after_a_patch_changed_the_stream() { + let mut turn_coordinator = make_turn_coordinator(); + let mut stream = stream_with_patchable_event(); + let mut retry_state = make_retry_state(2); + + let (patch_action, _) = handle_llm_event( + stale_signature_patch(), + &mut turn_coordinator, + &mut stream, + &mut retry_state, + ); + assert_matches!(patch_action, LoopAction::Continue); + + let (action, _) = handle_llm_event( + Event::Finished(FinishReason::Retry), + &mut turn_coordinator, + &mut stream, + &mut retry_state, + ); + + assert_matches!(action, LoopAction::Break); +} + +/// A patch that matched nothing leaves the request byte-identical, so resending +/// it can only fail the same way. +#[test] +fn rebuild_refused_when_the_patch_matched_nothing() { + let mut turn_coordinator = make_turn_coordinator(); + // No event carries the targeted metadata, so the patch applies to nothing. + let mut stream = ConversationStream::new_test(); + let mut retry_state = make_retry_state(2); + + handle_llm_event( + stale_signature_patch(), + &mut turn_coordinator, + &mut stream, + &mut retry_state, + ); + + let (action, _) = handle_llm_event( + Event::Finished(FinishReason::Retry), + &mut turn_coordinator, + &mut stream, + &mut retry_state, + ); + + assert_matches!( + action, + LoopAction::RebuildRefused(RebuildRefusal::NoProgress) + ); +} + +/// A rebuild request with no patch at all cannot change anything either. +#[test] +fn rebuild_refused_without_a_preceding_patch() { + let mut turn_coordinator = make_turn_coordinator(); + let mut stream = stream_with_patchable_event(); + let mut retry_state = make_retry_state(2); + + let (action, _) = handle_llm_event( + Event::Finished(FinishReason::Retry), + &mut turn_coordinator, + &mut stream, + &mut retry_state, + ); + + assert_matches!( + action, + LoopAction::RebuildRefused(RebuildRefusal::NoProgress) + ); +} + +/// This is what bounds the loop for a misbehaving provider: one applied patch +/// authorizes exactly one rebuild, so a provider that keeps asking without +/// patching again is stopped on its second request. +#[test] +fn one_patch_authorizes_only_one_rebuild() { + let mut turn_coordinator = make_turn_coordinator(); + let mut stream = stream_with_patchable_event(); + let mut retry_state = make_retry_state(2); + + handle_llm_event( + stale_signature_patch(), + &mut turn_coordinator, + &mut stream, + &mut retry_state, + ); + + let retry = || Event::Finished(FinishReason::Retry); + let (first, _) = handle_llm_event( + retry(), + &mut turn_coordinator, + &mut stream, + &mut retry_state, + ); + let (second, _) = handle_llm_event( + retry(), + &mut turn_coordinator, + &mut stream, + &mut retry_state, + ); + + assert_matches!(first, LoopAction::Break); + assert_matches!( + second, + LoopAction::RebuildRefused(RebuildRefusal::NoProgress) + ); +} + +/// A repair that keeps making progress is still capped, so a provider whose +/// rejection carries no position (Google strips one signature per round) cannot +/// walk a long conversation at the cost of a full request per step. +#[test] +fn rebuild_refused_past_the_consecutive_limit() { + let mut turn_coordinator = make_turn_coordinator(); + let mut retry_state = make_retry_state(2); + + // Every round patches a fresh stream, so every round makes progress and the + // only thing that can stop the loop is the cap. + let mut actions = vec![]; + for _ in 0..=MAX_CONSECUTIVE_REBUILDS { + let mut stream = stream_with_patchable_event(); + + handle_llm_event( + stale_signature_patch(), + &mut turn_coordinator, + &mut stream, + &mut retry_state, + ); + + let (action, _) = handle_llm_event( + Event::Finished(FinishReason::Retry), + &mut turn_coordinator, + &mut stream, + &mut retry_state, + ); + actions.push(action); + } + + let (last, allowed) = actions.split_last().expect("at least one round"); + + assert_eq!(allowed.len(), MAX_CONSECUTIVE_REBUILDS as usize); + for (round, action) in allowed.iter().enumerate() { + assert_matches!(action, LoopAction::Break, "round {round} is allowed"); + } + + assert_matches!( + last, + LoopAction::RebuildRefused(RebuildRefusal::LimitReached { + limit: MAX_CONSECUTIVE_REBUILDS + }) + ); +} + +/// Streaming content again clears the cap, so a turn that repairs, answers, and +/// later needs an unrelated repair is not punished for the first one. +#[test] +fn successful_cycle_restores_the_rebuild_allowance() { + let mut turn_coordinator = make_turn_coordinator(); + let mut retry_state = make_retry_state(2); + + for _ in 0..MAX_CONSECUTIVE_REBUILDS { + let mut stream = stream_with_patchable_event(); + handle_llm_event( + stale_signature_patch(), + &mut turn_coordinator, + &mut stream, + &mut retry_state, + ); + handle_llm_event( + Event::Finished(FinishReason::Retry), + &mut turn_coordinator, + &mut stream, + &mut retry_state, + ); + } + + retry_state.reset(); + + let mut stream = stream_with_patchable_event(); + handle_llm_event( + stale_signature_patch(), + &mut turn_coordinator, + &mut stream, + &mut retry_state, + ); + let (action, _) = handle_llm_event( + Event::Finished(FinishReason::Retry), + &mut turn_coordinator, + &mut stream, + &mut retry_state, + ); + + assert_matches!(action, LoopAction::Break); +} + +/// A cycle that streams content again drops any pending repair, so a later +/// unrelated rebuild request cannot ride on a patch from before it. +#[test] +fn successful_cycle_drops_the_pending_patch() { + let mut turn_coordinator = make_turn_coordinator(); + let mut stream = stream_with_patchable_event(); + let mut retry_state = make_retry_state(2); + + handle_llm_event( + stale_signature_patch(), + &mut turn_coordinator, + &mut stream, + &mut retry_state, + ); + retry_state.reset(); + + let (action, _) = handle_llm_event( + Event::Finished(FinishReason::Retry), + &mut turn_coordinator, + &mut stream, + &mut retry_state, + ); + + assert_matches!( + action, + LoopAction::RebuildRefused(RebuildRefusal::NoProgress) + ); +} + /// Regression: when the user picks `Continue` (`'c'`) while the LLM stream is /// still alive, the action is `Resume` — "keep waiting for the current /// stream." diff --git a/crates/jp_cli/src/cmd/query/stream.rs b/crates/jp_cli/src/cmd/query/stream.rs index ee7badce..bebedfc6 100644 --- a/crates/jp_cli/src/cmd/query/stream.rs +++ b/crates/jp_cli/src/cmd/query/stream.rs @@ -5,6 +5,6 @@ pub(crate) mod retry; -pub(crate) use retry::{StreamErrorOutcome, StreamRetryState, handle_stream_error}; +pub(crate) use retry::{RebuildRefusal, StreamErrorOutcome, StreamRetryState, 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 84260e53..886ce6c0 100644 --- a/crates/jp_cli/src/cmd/query/stream/retry.rs +++ b/crates/jp_cli/src/cmd/query/stream/retry.rs @@ -25,7 +25,7 @@ //! //! [`StreamError::is_retryable`]: jp_llm::StreamError::is_retryable -use std::{fmt::Write as _, sync::Arc}; +use std::{fmt, fmt::Write as _, mem, sync::Arc}; use jp_config::assistant::request::RequestConfig; use jp_llm::{StreamError, exponential_backoff}; @@ -35,6 +35,53 @@ use tracing::{error, warn}; use crate::{cmd::query::turn::TurnCoordinator, error::Error, signals::SignalRouter}; +/// How many provider-requested rebuilds a turn may attempt in a row. +/// +/// Every rebuild re-sends the whole conversation, so a repair costs the round +/// count times the conversation size. +/// A repair needing more rounds than this is not worth paying for; the turn +/// reports the dead end instead. +/// +/// Anthropic repairs a wedged turn in one round, and one round per affected +/// message after that. +/// Google's rejection carries no position, so its repair walks one signature +/// per round and a long conversation can exceed this, surfacing as an error +/// rather than a large bill. +/// +/// The count resets once a cycle produces its first successful event, so a turn +/// that repairs, streams, and later repairs again gets a fresh allowance. +pub(crate) const MAX_CONSECUTIVE_REBUILDS: u32 = 10; + +/// Why a provider-requested rebuild was refused. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum RebuildRefusal { + /// The preceding patch changed nothing, so the rebuilt request would be + /// identical to the one the provider just rejected. + NoProgress, + + /// The turn has rebuilt as many times in a row as it is allowed to. + LimitReached { + /// The allowance that was reached. + limit: u32, + }, +} + +impl fmt::Display for RebuildRefusal { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::NoProgress => f.write_str( + "the provider asked to rebuild the request but sent no fix that changed it, so \ + resending would fail the same way", + ), + Self::LimitReached { limit } => write!( + f, + "the provider asked to rebuild the request {limit} times in a row without \ + producing a usable response" + ), + } + } +} + /// Tracks retry state for stream errors within a single turn. /// /// Counts consecutive stream failures and enforces retry limits from @@ -48,6 +95,14 @@ pub struct StreamRetryState { /// Number of consecutive stream failures without a successful cycle. consecutive_failures: u32, + /// Whether the most recent provider patch set changed the conversation + /// stream in a way that makes a rebuild worth sending. + patch_changed_stream: bool, + + /// Number of consecutive provider-requested rebuilds without a successful + /// cycle. + consecutive_rebuilds: u32, + /// Whether a temporary retry notification line is currently displayed. /// /// When `true`, the next retry or successful event should overwrite the @@ -64,6 +119,8 @@ impl StreamRetryState { Self { config, consecutive_failures: 0, + patch_changed_stream: false, + consecutive_rebuilds: 0, line_active: false, is_tty, } @@ -77,6 +134,49 @@ impl StreamRetryState { /// mid-response) don't permanently consume the retry budget. pub fn reset(&mut self) { self.consecutive_failures = 0; + self.patch_changed_stream = false; + self.consecutive_rebuilds = 0; + } + + /// Record the outcome of a provider patch set. + /// + /// `applied` is how many events it changed, and `shrinks` whether every + /// action in the set strictly shrinks the stream (see + /// [`PatchAction::shrinks_stream`]). + /// A rebuild is only worth sending when both hold: an unchanged stream + /// rebuilds to an identical request, and a set that can grow the stream + /// gives no guarantee the loop ends. + /// + /// [`PatchAction::shrinks_stream`]: jp_llm::event::PatchAction::shrinks_stream + pub fn record_patch(&mut self, applied: usize, shrinks: bool) { + self.patch_changed_stream = applied > 0 && shrinks; + } + + /// Authorize a provider-requested rebuild, or explain why it is refused. + /// + /// A provider that patches the conversation stream and asks for a rebuild + /// reports no error, so these attempts never reach [`handle_stream_error`] + /// and consume no part of the stream-error budget. + /// They are bounded two ways: each rebuild must follow a patch that made + /// progress, which is what guarantees the loop ends, and the number in a + /// row is capped, which keeps a terminating-but-expensive repair from + /// running up a bill. + /// + /// Consumes the record of the preceding patch, so one patch cannot + /// authorize two rebuilds. + pub fn authorize_rebuild(&mut self) -> Result<(), RebuildRefusal> { + if !mem::take(&mut self.patch_changed_stream) { + return Err(RebuildRefusal::NoProgress); + } + + self.consecutive_rebuilds += 1; + if self.consecutive_rebuilds > MAX_CONSECUTIVE_REBUILDS { + return Err(RebuildRefusal::LimitReached { + limit: MAX_CONSECUTIVE_REBUILDS, + }); + } + + Ok(()) } /// Clear the retry notification line if one is currently displayed. diff --git a/crates/jp_cli/src/cmd/query/turn_loop.rs b/crates/jp_cli/src/cmd/query/turn_loop.rs index 2f429f08..30ba5347 100644 --- a/crates/jp_cli/src/cmd/query/turn_loop.rs +++ b/crates/jp_cli/src/cmd/query/turn_loop.rs @@ -27,7 +27,7 @@ use jp_conversation::{ }; use jp_inquire::prompt::PromptBackend; use jp_llm::{ - Provider, + Error as LlmError, Provider, error::StreamError, event::{Event, EventPart, ToolCallPart}, model::ModelDetails, @@ -547,11 +547,29 @@ pub(super) async fn run_turn_loop( // from a misbehaving provider commits nothing and // so cannot drive a double dispatch. let (action, committed) = conv.update_events(|stream| { - handle_llm_event(event, &mut turn_coordinator, stream) + handle_llm_event( + event, + &mut turn_coordinator, + stream, + &mut stream_retry, + ) }); match action { LoopAction::Continue => {} LoopAction::Break => break, + // The repair cannot or should not continue. + // Persist what was streamed and surface the dead + // end, which the refusal describes. + LoopAction::RebuildRefused(refusal) => { + if let Err(err) = conv.flush() { + warn!("Failed to persist before abort: {err}"); + } + + return Err(LlmError::Stream(StreamError::other( + refusal.to_string(), + )) + .into()); + } } // On a flushed tool-call request: clear the temp diff --git a/crates/jp_llm/src/event.rs b/crates/jp_llm/src/event.rs index d58f47dc..bf301a2d 100644 --- a/crates/jp_llm/src/event.rs +++ b/crates/jp_llm/src/event.rs @@ -181,6 +181,27 @@ pub enum PatchAction { RemoveMetadata(String), } +impl PatchAction { + /// Whether applying this action strictly shrinks the conversation stream. + /// + /// A provider-driven repair loop rebuilds the request after every applied + /// patch, so it terminates only if each round shrinks a finite measure. + /// Producing a change is not enough: an action that sets a fresh value or + /// appends content can change an event on every round forever. + /// + /// An action that cannot guarantee shrinkage returns `false`, and the + /// repair loop refuses to rebuild rather than risk an unbounded number of + /// requests. + /// Answer this deliberately when adding a variant; `false` is the safe + /// answer. + #[must_use] + pub const fn shrinks_stream(&self) -> bool { + match self { + Self::RemoveMetadata(_) => true, + } + } +} + /// Apply provider-issued patches to the events already in `stream`, returning /// how many events it changed. /// diff --git a/crates/jp_llm/src/provider/anthropic.rs b/crates/jp_llm/src/provider/anthropic.rs index e849a0bc..cbfbb3a7 100644 --- a/crates/jp_llm/src/provider/anthropic.rs +++ b/crates/jp_llm/src/provider/anthropic.rs @@ -1,4 +1,4 @@ -use std::{env, mem, time::Duration}; +use std::{env, mem, ops::RangeInclusive, time::Duration}; use async_anthropic::{ Client, @@ -262,9 +262,9 @@ impl ForcedToolFallback { /// issued with thinking disabled and the original forced `tool_choice` /// restored. /// -/// If the API rejects the request due to an invalid thinking-block signature, -/// emits [`Event::Patch`] instructions to fix the conversation stream and -/// finishes with [`FinishReason::Retry`] so the caller can rebuild and retry. +/// If the API rejects a thinking block in the request, emits [`Event::Patch`] +/// instructions to fix the conversation stream and finishes with +/// [`FinishReason::Retry`] so the caller can rebuild and retry. fn call( client: Client, request: types::CreateMessagesRequest, @@ -306,14 +306,17 @@ fn call( pin_mut!(stream); while let Some(result) = stream.next().await { - // Anthropic rejects requests with stale thinking signatures as a - // 400 `invalid_request_error`. Emit a patch to strip the offending - // metadata and ask the caller to retry. + // Anthropic rejects requests carrying thinking blocks it won't + // accept as a 400 `invalid_request_error`. Emit a patch to strip the + // offending metadata and ask the caller to retry. if let Err(ref err) = result - && is_invalid_thinking_signature(err) + && is_thinking_block_rejection(err) && let Some(patches) = build_thinking_patches(&request, err) { - warn!("Invalid thinking signature, requesting retry."); + warn!( + patches = patches.len(), + "Thinking block rejected by the API, patching history and retrying: {err}" + ); yield Event::Patch(patches); yield Event::Finished(FinishReason::Retry); return; @@ -686,13 +689,21 @@ fn soft_force_retry( })) } -/// Returns `true` if the error is an Anthropic `invalid_request_error` -/// complaining about an invalid `signature` in a `thinking` block. -fn is_invalid_thinking_signature(error: &StreamError) -> bool { +/// Returns `true` if the error is an Anthropic `invalid_request_error` about a +/// `thinking` block the API refuses to accept. +/// +/// Two rejections fall in this class: an invalid `signature` on a thinking +/// block, and thinking blocks in the latest assistant message having been +/// modified. +/// Both are repaired the same way, by replaying the reasoning as `` text +/// instead of a native thinking block. +fn is_thinking_block_rejection(error: &StreamError) -> bool { + let message = error.message(); + error.kind == StreamErrorKind::Other - && error.message().contains("invalid_request_error") - && error.message().contains("signature") - && error.message().contains("thinking") + && message.contains("invalid_request_error") + && message.contains("thinking") + && (message.contains("signature") || message.contains("cannot be modified")) } /// Extract the `(turn_index, flat_content_index)` from an Anthropic error @@ -780,14 +791,22 @@ fn is_tool_result_message(msg: &types::Message) -> bool { .all(|c| matches!(c, types::MessageContent::ToolResult(_))) } -/// Build [`EventPatch`] instructions for the thinking block identified by the -/// API error. -/// The caller yields these as [`Event::Patch`] so the CLI can remove the stale -/// signature metadata from the persisted conversation stream. +/// Build [`EventPatch`] instructions for the thinking blocks the API refused. +/// The caller yields these as [`Event::Patch`] so the CLI can remove the +/// provider metadata from the persisted conversation stream. /// On the next request rebuild, [`convert_event`] will fall through to the /// `` tag path for events that no longer carry a signature. /// -/// Returns `None` if the offending block cannot be located. +/// A block anywhere in the final assistant turn downgrades every thinking and +/// redacted-thinking block in the message holding it. +/// Anthropic treats that whole turn as one assistant message and requires its +/// thinking blocks back exactly as generated, so rewriting a subset is rejected +/// with `thinking blocks ... cannot be modified`; rewriting all of a message's +/// blocks leaves text and tool use, which the API accepts. +/// Blocks in earlier turns are downgraded individually, keeping the rest of +/// their reasoning native. +/// +/// Returns `None` if no thinking block can be located. fn build_thinking_patches( request: &types::CreateMessagesRequest, error: &StreamError, @@ -797,19 +816,97 @@ fn build_thinking_patches( .and_then(|(turn, flat)| resolve_turn_position(&request.messages, turn, flat)) .or_else(|| find_oldest_thinking_block(request))?; - let (key, value) = identify_thinking_block(request, msg_idx, content_idx).or_else(|| { - // Resolved position wasn't a thinking block. Fall back to oldest. - let (msg, idx) = find_oldest_thinking_block(request)?; - identify_thinking_block(request, msg, idx) - })?; + debug!( + ?api_position, + msg_idx, content_idx, "Located the thinking block rejected by the API." + ); + + let in_final_turn = + last_assistant_turn(&request.messages).is_some_and(|turn| turn.contains(&msg_idx)); + + let content_indices = if in_final_turn { + thinking_block_indices(request, msg_idx) + } else { + vec![content_idx] + }; + + let patches: Vec = content_indices + .into_iter() + .filter_map(|idx| identify_thinking_block(request, msg_idx, idx)) + .map(|(key, value)| thinking_patch(key, value)) + .collect(); + + if !patches.is_empty() { + return Some(patches); + } - Some(vec![EventPatch { + // The located position held no thinking block. Fall back to the oldest one. + let (msg_idx, content_idx) = find_oldest_thinking_block(request)?; + let (key, value) = identify_thinking_block(request, msg_idx, content_idx)?; + + Some(vec![thinking_patch(key, value)]) +} + +/// Build the patch that strips a thinking block's provider metadata, so the +/// event is replayed as `` text rather than a native thinking block. +fn thinking_patch(key: &'static str, value: String) -> EventPatch { + EventPatch { matcher: EventMatcher::MetadataValue { key: key.to_owned(), value, }, action: PatchAction::RemoveMetadata(key.to_owned()), - }]) + } +} + +/// Returns `true` for a native `thinking` or `redacted_thinking` content block. +fn is_thinking_block(content: &types::MessageContent) -> bool { + matches!( + content, + types::MessageContent::Thinking(_) | types::MessageContent::RedactedThinking { .. } + ) +} + +/// Messages that make up the final assistant turn. +/// +/// Anthropic groups an assistant turn from its first assistant message through +/// every following tool-result and assistant message, and treats that whole +/// span as one "latest assistant message". +/// The span ends at the last assistant message and begins right after the most +/// recent user message before it that is not purely tool results. +/// A user prompt appended after the turn opens the next turn without shortening +/// this one, so a wedged turn is still repairable on the first request of the +/// following turn. +/// +/// Returns `None` when the request holds no assistant message. +fn last_assistant_turn(messages: &[types::Message]) -> Option> { + let end = messages + .iter() + .rposition(|msg| msg.role == types::MessageRole::Assistant)?; + + let start = messages[..end] + .iter() + .rposition(|msg| msg.role == types::MessageRole::User && !is_tool_result_message(msg)) + .map_or(0, |idx| idx + 1); + + Some(start..=end) +} + +/// Content indices of every thinking and redacted-thinking block in one +/// message, in order. +fn thinking_block_indices(request: &types::CreateMessagesRequest, msg_idx: usize) -> Vec { + let Some(message) = request.messages.get(msg_idx) else { + return vec![]; + }; + + message + .content + .0 + .iter() + .enumerate() + .filter(|(_, content)| is_thinking_block(content)) + .map(|(idx, _)| idx) + .collect() } /// Find the first (oldest) `Thinking` or `RedactedThinking` content block in @@ -819,10 +916,7 @@ fn build_thinking_patches( fn find_oldest_thinking_block(request: &types::CreateMessagesRequest) -> Option<(usize, usize)> { for (i, msg) in request.messages.iter().enumerate() { for (j, content) in msg.content.0.iter().enumerate() { - if matches!( - content, - types::MessageContent::Thinking(_) | types::MessageContent::RedactedThinking { .. } - ) { + if is_thinking_block(content) { return Some((i, j)); } } diff --git a/crates/jp_llm/src/provider/anthropic_tests.rs b/crates/jp_llm/src/provider/anthropic_tests.rs index 370d3ac4..9440c180 100644 --- a/crates/jp_llm/src/provider/anthropic_tests.rs +++ b/crates/jp_llm/src/provider/anthropic_tests.rs @@ -2240,13 +2240,24 @@ mod thinking_signature_recovery { use crate::{ error::StreamError, - event::{EventMatcher, PatchAction}, + event::{EventMatcher, EventPatch, PatchAction}, provider::anthropic::{ build_thinking_patches, find_oldest_thinking_block, identify_thinking_block, - is_invalid_thinking_signature, parse_signature_error_position, resolve_turn_position, + is_thinking_block_rejection, parse_signature_error_position, resolve_turn_position, }, }; + /// The `(metadata_key, metadata_value)` each patch targets, in order. + fn patch_targets(patches: &[EventPatch]) -> Vec<(&str, &str)> { + patches + .iter() + .map(|patch| { + let EventMatcher::MetadataValue { key, value } = &patch.matcher; + (key.as_str(), value.as_str()) + }) + .collect() + } + fn make_thinking(text: &str, sig: &str) -> types::MessageContent { types::MessageContent::Thinking(types::Thinking { thinking: text.to_owned(), @@ -2333,19 +2344,29 @@ mod thinking_signature_recovery { "api error: invalid_request_error: messages.1.content.0: Invalid `signature` in \ `thinking` block", ); - assert!(is_invalid_thinking_signature(&error)); + assert!(is_thinking_block_rejection(&error)); + } + + #[test] + fn detects_unmodifiable_thinking_error() { + let error = StreamError::other( + "api error: invalid_request_error: messages.9.content.89: `thinking` or \ + `redacted_thinking` blocks in the latest assistant message cannot be modified. These \ + blocks must remain as they were in the original response.", + ); + assert!(is_thinking_block_rejection(&error)); } #[test] fn ignores_unrelated_errors() { let error = StreamError::other("api error: rate_limit_error: too many requests"); - assert!(!is_invalid_thinking_signature(&error)); + assert!(!is_thinking_block_rejection(&error)); } #[test] fn ignores_retryable_errors() { let error = StreamError::transient("server error with signature and thinking"); - assert!(!is_invalid_thinking_signature(&error)); + assert!(!is_thinking_block_rejection(&error)); } #[test] @@ -2501,6 +2522,190 @@ mod thinking_signature_recovery { assert!(build_thinking_patches(&request, &error).is_none()); } + /// A tool-use loop whose newest assistant message interleaves redacted and + /// signed thinking blocks before its tool call, and which ends with the + /// matching tool result. + /// + /// Turn 1 spans messages 1 through 4, flattening to: `[thinking(0), + /// tool_use(1), tool_result(2), r0(3), r1(4), sig_3(5), r2(6), sig_5(7), + /// tool_use(8), tool_result(9)]` + fn interleaved_thinking_conversation() -> Vec { + vec![ + msg(MessageRole::User, vec![make_text("hello")]), + msg(MessageRole::Assistant, vec![ + make_thinking("early", "sig_early"), + make_tool_use("tu1"), + ]), + msg(MessageRole::User, vec![make_tool_result("tu1")]), + msg(MessageRole::Assistant, vec![ + make_redacted("r0"), + make_redacted("r1"), + make_thinking("t3", "sig_3"), + make_redacted("r2"), + make_thinking("t5", "sig_5"), + make_tool_use("tu2"), + ]), + msg(MessageRole::User, vec![make_tool_result("tu2")]), + ] + } + + /// Anthropic requires the latest assistant message to carry its thinking + /// blocks back exactly as generated, so downgrading only the block named in + /// the error is rejected on the next request with `cannot be modified`. + #[test] + fn latest_assistant_message_downgrades_every_thinking_block() { + let request = request(interleaved_thinking_conversation()); + + // Flat index 5 is the first signed block in the newest assistant + // message. + let error = StreamError::other( + "api error: invalid_request_error: messages.1.content.5: Invalid `signature` in \ + `thinking` block", + ); + + let patches = build_thinking_patches(&request, &error).unwrap(); + + assert_eq!(patch_targets(&patches), [ + ("anthropic_redacted_thinking", "r0"), + ("anthropic_redacted_thinking", "r1"), + ("anthropic_thinking_signature", "sig_3"), + ("anthropic_redacted_thinking", "r2"), + ("anthropic_thinking_signature", "sig_5"), + ]); + } + + /// The rejected block often sits in the middle of the final turn rather + /// than in its last assistant message: Anthropic reports a position in the + /// turn, and the trailing assistant message has already been rewritten as + /// `` text by [`downgrade_trailing_thinking`]. + /// Every thinking block in the named message still has to go. + /// + /// [`downgrade_trailing_thinking`]: super::super::downgrade_trailing_thinking + #[test] + fn mid_turn_message_downgrades_every_thinking_block() { + let request = request(vec![ + msg(MessageRole::User, vec![make_text("hello")]), + msg(MessageRole::Assistant, vec![ + make_redacted("r0"), + make_thinking("t1", "sig_1"), + make_redacted("r2"), + make_tool_use("tu1"), + ]), + msg(MessageRole::User, vec![make_tool_result("tu1")]), + // The trailing assistant message carries no native thinking. + msg(MessageRole::Assistant, vec![make_text("answer")]), + ]); + + // Flat index 1 is the signed block in messages[1], which is in the final + // turn but is not its last assistant message. + let error = StreamError::other( + "api error: invalid_request_error: messages.1.content.1: `thinking` or \ + `redacted_thinking` blocks in the latest assistant message cannot be modified.", + ); + + let patches = build_thinking_patches(&request, &error).unwrap(); + + assert_eq!(patch_targets(&patches), [ + ("anthropic_redacted_thinking", "r0"), + ("anthropic_thinking_signature", "sig_1"), + ("anthropic_redacted_thinking", "r2"), + ]); + } + + /// A fresh user prompt opens a new turn without an assistant reply yet, but + /// Anthropic still calls the preceding assistant turn "the latest assistant + /// message". + /// A wedged turn must therefore be repaired on the first request of the + /// next turn, before any tool call has run. + #[test] + fn new_prompt_after_wedged_turn_downgrades_every_thinking_block() { + let request = request(vec![ + msg(MessageRole::User, vec![make_text("first prompt")]), + msg(MessageRole::Assistant, vec![ + make_redacted("r0"), + make_thinking("t1", "sig_1"), + make_tool_use("tu1"), + ]), + msg(MessageRole::User, vec![make_tool_result("tu1")]), + msg(MessageRole::Assistant, vec![make_text("answer")]), + msg(MessageRole::User, vec![make_text("second prompt")]), + ]); + + // Turn 1 spans messages 1 through 3; flat index 1 is the signed block in + // messages[1]. + let error = StreamError::other( + "api error: invalid_request_error: messages.1.content.1: `thinking` or \ + `redacted_thinking` blocks in the latest assistant message cannot be modified.", + ); + + let patches = build_thinking_patches(&request, &error).unwrap(); + + assert_eq!(patch_targets(&patches), [ + ("anthropic_redacted_thinking", "r0"), + ("anthropic_thinking_signature", "sig_1"), + ]); + } + + /// Turns before the final one carry no immutability constraint, so a stale + /// signature there is repaired in place, leaving that turn's other + /// reasoning native. + #[test] + fn earlier_turn_downgrades_only_the_named_block() { + let request = request(vec![ + msg(MessageRole::User, vec![make_text("first")]), + msg(MessageRole::Assistant, vec![ + make_thinking("t0", "sig_0"), + make_thinking("t1", "sig_1"), + make_text("answer"), + ]), + msg(MessageRole::User, vec![make_text("second")]), + msg(MessageRole::Assistant, vec![make_thinking("t2", "sig_2")]), + ]); + + // Turn 1 is messages[1], two turns back from the final one. + let error = StreamError::other( + "api error: invalid_request_error: messages.1.content.1: Invalid `signature` in \ + `thinking` block", + ); + + let patches = build_thinking_patches(&request, &error).unwrap(); + + assert_eq!(patch_targets(&patches), [( + "anthropic_thinking_signature", + "sig_1" + )]); + } + + /// A conversation left half-downgraded by an earlier one-block-at-a-time + /// repair is unsendable: the message holds native thinking blocks next to + /// the `` text that replaced its siblings. + /// The remaining native blocks all have to go. + #[test] + fn unmodifiable_error_heals_a_half_downgraded_message() { + let request = request(vec![ + msg(MessageRole::User, vec![make_text("hello")]), + msg(MessageRole::Assistant, vec![ + make_redacted("r0"), + make_text("\nt3\n\n\n"), + make_redacted("r1"), + make_tool_use("tu1"), + ]), + msg(MessageRole::User, vec![make_tool_result("tu1")]), + ]); + + let error = StreamError::other( + "api error: invalid_request_error: messages.1.content.0: `thinking` or \ + `redacted_thinking` blocks in the latest assistant message cannot be modified.", + ); + + let patches = build_thinking_patches(&request, &error).unwrap(); + + assert_eq!(patch_targets(&patches), [ + ("anthropic_redacted_thinking", "r0"), + ("anthropic_redacted_thinking", "r1"), + ]); + } + /// Reproduce the exact message structure from the user's failing request: /// two assistant turns with tool-use loops, separated by a user message. fn tool_use_conversation() -> Vec { @@ -2594,6 +2799,148 @@ mod thinking_signature_recovery { assert_eq!(resolve_turn_position(&msgs, 1, 999), None); } + /// Message shape of the request that failed in the field, as `(kind, + /// block_count)` per message: `p` a user prompt, `u` a user tool result, + /// `a` an assistant message. + /// + /// Only roles and block counts drive [`resolve_turn_position`], so blocks + /// are placeholders everywhere except the final assistant message, which + /// carries the real interleaving of redacted and signed thinking. + /// The five prompts at messages 0, 26, 30, 34 and 36 are what make the + /// closing tool-use loop Anthropic's turn 9. + #[rustfmt::skip] + const INCIDENT_SHAPE: &[(char, usize)] = &[ + ('p',11), ('a',4), ('u',2), ('a',3), ('u',2), ('a',3), // 0 + ('u',2), ('a',2), ('u',1), ('a',2), ('u',1), ('a',1), // 6 + ('u',1), ('a',3), ('u',2), ('a',2), ('u',1), ('a',2), // 12 + ('u',1), ('a',3), ('u',2), ('a',2), ('u',1), ('a',2), // 18 + ('u',1), ('a',2), ('p',1), ('a',4), ('u',2), ('a',2), // 24 + ('p',1), ('a',4), ('u',2), ('a',1), ('p',1), ('a',1), // 30 + ('p',1), ('a',4), ('u',2), ('a',2), ('u',1), ('a',1), // 36 + ('u',1), ('a',3), ('u',2), ('a',2), ('u',1), ('a',2), // 42 + ('u',1), ('a',3), ('u',1), ('a',2), ('u',1), ('a',1), // 48 + ('u',1), ('a',2), ('u',1), ('a',2), ('u',1), ('a',2), // 54 + ('u',1), ('a',1), ('u',1), ('a',1), ('u',1), ('a',2), // 60 + ('u',1), ('a',2), ('u',1), ('a',1), ('u',1), ('a',2), // 66 + ('u',1), ('a',1), ('u',1), ('a',2), ('u',1), ('a',1), // 72 + ('u',1), ('a',2), ('u',1), ('a',1), ('u',1), ('a',1), // 78 + ('u',1), ('a',1), ('u',1), ('a',1), ('u',1), ('a',2), // 84 + ('u',1), ('a',1), ('u',1), ('a',3), ('u',1), ('a',2), // 90 + ('u',1), ('a',9), ('u',1), // 96 + ]; + + /// The newest assistant message of [`INCIDENT_SHAPE`], at index 97: three + /// redacted blocks, then signed thinking alternating with redacted, then + /// the tool call. + fn incident_newest_message() -> types::Message { + msg(MessageRole::Assistant, vec![ + make_redacted("r0"), + make_redacted("r1"), + make_redacted("r2"), + make_thinking("t3", "sig_3"), + make_redacted("r4"), + make_thinking("t5", "sig_5"), + make_redacted("r6"), + make_thinking("t7", "sig_7"), + make_tool_use("tu"), + ]) + } + + fn incident_messages() -> Vec { + INCIDENT_SHAPE + .iter() + .enumerate() + .map(|(idx, &(kind, count))| { + if idx == 97 { + return incident_newest_message(); + } + + let blocks = (0..count) + .map(|_| match kind { + 'u' => make_tool_result("tu"), + _ => make_text("filler"), + }) + .collect(); + + let role = if kind == 'a' { + MessageRole::Assistant + } else { + MessageRole::User + }; + + msg(role, blocks) + }) + .collect() + } + + /// Only a user message that isn't purely tool results opens a new turn, so + /// the five prompts in the recorded request produce ten turns rather than + /// one turn per request/response pair. + #[test] + fn incident_turn_boundaries_follow_user_prompts() { + let msgs = incident_messages(); + assert_eq!(msgs.len(), 99); + + assert_eq!(resolve_turn_position(&msgs, 0, 0), Some((0, 0))); + assert_eq!(resolve_turn_position(&msgs, 2, 0), Some((26, 0))); + assert_eq!(resolve_turn_position(&msgs, 4, 0), Some((30, 0))); + assert_eq!(resolve_turn_position(&msgs, 6, 0), Some((34, 0))); + assert_eq!(resolve_turn_position(&msgs, 8, 0), Some((36, 0))); + assert_eq!(resolve_turn_position(&msgs, 9, 0), Some((37, 0))); + } + + /// Every position Anthropic named across the four failed requests, mapped + /// against the recorded message shape. + /// + /// Turn 9 spans messages 37 to 98, and messages 37 to 96 contribute 85 + /// blocks, so the newest assistant message begins at flat index 85. + #[test] + fn incident_positions_resolve_to_the_blocks_the_api_named() { + let msgs = incident_messages(); + + // The three signed thinking blocks, rejected one per request. + assert_eq!(resolve_turn_position(&msgs, 9, 88), Some((97, 3))); + assert_eq!(resolve_turn_position(&msgs, 9, 90), Some((97, 5))); + assert_eq!(resolve_turn_position(&msgs, 9, 92), Some((97, 7))); + + // `messages.9.content.89` from the final rejection is a redacted block: + // by then every signed block had already been rewritten as text. + assert_eq!(resolve_turn_position(&msgs, 9, 89), Some((97, 4))); + + // Message boundaries either side of the thinking blocks. + assert_eq!(resolve_turn_position(&msgs, 9, 85), Some((97, 0))); + assert_eq!(resolve_turn_position(&msgs, 9, 93), Some((97, 8))); + assert_eq!(resolve_turn_position(&msgs, 9, 94), Some((98, 0))); + } + + /// End to end on the recorded request: one round strips all eight thinking + /// blocks from the newest assistant message, leaving text and tool use. + /// + /// The field failure took three rounds, one signed block each, and each + /// round left the message in the half-rewritten state the API rejects. + #[test] + fn incident_request_downgrades_the_whole_newest_message() { + let request = request(incident_messages()); + + let error = StreamError::other( + "api error: invalid_request_error: messages.9.content.88: Invalid `signature` in \ + `thinking` block", + ); + + let patches = build_thinking_patches(&request, &error).unwrap(); + + assert_eq!(patch_targets(&patches), [ + ("anthropic_redacted_thinking", "r0"), + ("anthropic_redacted_thinking", "r1"), + ("anthropic_redacted_thinking", "r2"), + ("anthropic_thinking_signature", "sig_3"), + ("anthropic_redacted_thinking", "r4"), + ("anthropic_thinking_signature", "sig_5"), + ("anthropic_redacted_thinking", "r6"), + ("anthropic_thinking_signature", "sig_7"), + ]); + } + #[test] fn resolve_non_thinking_blocks() { let msgs = tool_use_conversation(); From 6b04b436a426d81f15960b9381aed28464943c65 Mon Sep 17 00:00:00 2001 From: Jean Mertz Date: Fri, 31 Jul 2026 13:39:50 +0200 Subject: [PATCH 2/4] review feedback Signed-off-by: Jean Mertz --- .../src/cmd/query/interrupt/signals_tests.rs | 44 ++++ crates/jp_cli/src/cmd/query/stream.rs | 5 +- crates/jp_cli/src/cmd/query/stream/retry.rs | 78 +++++-- crates/jp_cli/src/cmd/query/turn_loop.rs | 23 +- .../jp_cli/src/cmd/query/turn_loop_tests.rs | 199 +++++++++++++++++- 5 files changed, 319 insertions(+), 30 deletions(-) diff --git a/crates/jp_cli/src/cmd/query/interrupt/signals_tests.rs b/crates/jp_cli/src/cmd/query/interrupt/signals_tests.rs index 6ba5ae81..054bfca8 100644 --- a/crates/jp_cli/src/cmd/query/interrupt/signals_tests.rs +++ b/crates/jp_cli/src/cmd/query/interrupt/signals_tests.rs @@ -80,6 +80,50 @@ fn stale_signature_patch() -> Event { }]) } +/// A patch set targeting metadata no event carries, so applying it changes +/// nothing. +fn no_op_patch() -> Event { + Event::Patch(vec![EventPatch { + matcher: EventMatcher::MetadataValue { + key: "absent_key".into(), + value: "absent_value".into(), + }, + action: PatchAction::RemoveMetadata("absent_key".into()), + }]) +} + +/// The event contract allows several patch sets before the rebuild request, so +/// their outcomes accumulate: one set that changed the stream is enough to make +/// the rebuilt request differ, whatever the sets around it did. +#[test] +fn a_later_no_op_patch_does_not_erase_an_earlier_effective_one() { + let mut turn_coordinator = make_turn_coordinator(); + let mut stream = stream_with_patchable_event(); + let mut retry_state = make_retry_state(2); + + handle_llm_event( + stale_signature_patch(), + &mut turn_coordinator, + &mut stream, + &mut retry_state, + ); + handle_llm_event( + no_op_patch(), + &mut turn_coordinator, + &mut stream, + &mut retry_state, + ); + + let (action, _) = handle_llm_event( + Event::Finished(FinishReason::Retry), + &mut turn_coordinator, + &mut stream, + &mut retry_state, + ); + + assert_matches!(action, LoopAction::Break); +} + /// A patch that changed the stream earns the rebuild that follows it: the /// request the provider gets next is genuinely different. #[test] diff --git a/crates/jp_cli/src/cmd/query/stream.rs b/crates/jp_cli/src/cmd/query/stream.rs index bebedfc6..857d4edf 100644 --- a/crates/jp_cli/src/cmd/query/stream.rs +++ b/crates/jp_cli/src/cmd/query/stream.rs @@ -5,6 +5,9 @@ pub(crate) mod retry; -pub(crate) use retry::{RebuildRefusal, StreamErrorOutcome, StreamRetryState, handle_stream_error}; +pub(crate) use retry::{ + RebuildRefusal, 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 886ce6c0..ff6746dc 100644 --- a/crates/jp_cli/src/cmd/query/stream/retry.rs +++ b/crates/jp_cli/src/cmd/query/stream/retry.rs @@ -95,10 +95,17 @@ pub struct StreamRetryState { /// Number of consecutive stream failures without a successful cycle. consecutive_failures: u32, - /// Whether the most recent provider patch set changed the conversation - /// stream in a way that makes a rebuild worth sending. + /// Whether any provider patch set in this cycle changed the conversation + /// stream. patch_changed_stream: bool, + /// Whether every provider patch set in this cycle strictly shrinks the + /// stream. + /// + /// Accumulated rather than replaced, so a later set that shrinks cannot + /// erase an earlier one that may not. + patch_sets_shrink: bool, + /// Number of consecutive provider-requested rebuilds without a successful /// cycle. consecutive_rebuilds: u32, @@ -120,6 +127,7 @@ impl StreamRetryState { config, consecutive_failures: 0, patch_changed_stream: false, + patch_sets_shrink: true, consecutive_rebuilds: 0, line_active: false, is_tty, @@ -135,21 +143,25 @@ impl StreamRetryState { pub fn reset(&mut self) { self.consecutive_failures = 0; self.patch_changed_stream = false; + self.patch_sets_shrink = true; self.consecutive_rebuilds = 0; } - /// Record the outcome of a provider patch set. + /// Record the outcome of one provider patch set. /// /// `applied` is how many events it changed, and `shrinks` whether every /// action in the set strictly shrinks the stream (see /// [`PatchAction::shrinks_stream`]). - /// A rebuild is only worth sending when both hold: an unchanged stream - /// rebuilds to an identical request, and a set that can grow the stream - /// gives no guarantee the loop ends. + /// + /// Outcomes accumulate until a rebuild consumes them, because a provider + /// may send several patch sets before asking for the rebuild: the rebuilt + /// request differs if any set changed the stream, and the loop only + /// terminates if every set shrinks it. /// /// [`PatchAction::shrinks_stream`]: jp_llm::event::PatchAction::shrinks_stream pub fn record_patch(&mut self, applied: usize, shrinks: bool) { - self.patch_changed_stream = applied > 0 && shrinks; + self.patch_changed_stream |= applied > 0; + self.patch_sets_shrink &= shrinks; } /// Authorize a provider-requested rebuild, or explain why it is refused. @@ -162,10 +174,13 @@ impl StreamRetryState { /// row is capped, which keeps a terminating-but-expensive repair from /// running up a bill. /// - /// Consumes the record of the preceding patch, so one patch cannot - /// authorize two rebuilds. + /// Consumes the accumulated patch record, so one patch cannot authorize two + /// rebuilds. pub fn authorize_rebuild(&mut self) -> Result<(), RebuildRefusal> { - if !mem::take(&mut self.patch_changed_stream) { + let changed = mem::take(&mut self.patch_changed_stream); + let shrinks = mem::replace(&mut self.patch_sets_shrink, true); + + if !(changed && shrinks) { return Err(RebuildRefusal::NoProgress); } @@ -240,6 +255,36 @@ impl StreamRetryState { } } +/// Flush buffered output and commit any unflushed partial assistant content to +/// the conversation stream. +/// +/// Content sits in the coordinator's event builder until a flush or terminal +/// event reaches it, so persisting the conversation alone would drop text the +/// user already saw on screen. +/// Call this before ending a turn on any path that bypasses the coordinator's +/// own terminal handling. +pub fn commit_partial_response( + turn_coordinator: &mut TurnCoordinator, + conv: &ConversationMut, + printer: &Arc, +) { + turn_coordinator.flush_renderer(); + printer.flush_instant(); + + let partial = turn_coordinator.peek_partial_events(); + if partial.is_empty() { + return; + } + + conv.update_events(|stream| { + let mut turn = stream.current_turn_mut(); + for response in partial { + turn = turn.add_chat_response(response); + } + turn.build().expect("Invalid ConversationStream state"); + }); +} + /// Outcome of [`handle_stream_error`]. #[derive(Debug)] pub enum StreamErrorOutcome { @@ -275,18 +320,7 @@ 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. - turn_coordinator.flush_renderer(); - printer.flush_instant(); - let partial = turn_coordinator.peek_partial_events(); - if !partial.is_empty() { - conv.update_events(|stream| { - let mut turn = stream.current_turn_mut(); - for response in partial { - turn = turn.add_chat_response(response); - } - turn.build().expect("Invalid ConversationStream state"); - }); - } + commit_partial_response(turn_coordinator, conv, printer); if !retry_state.can_retry(&error) { // Clear the temp line before printing the final error so it doesn't diff --git a/crates/jp_cli/src/cmd/query/turn_loop.rs b/crates/jp_cli/src/cmd/query/turn_loop.rs index 30ba5347..9c3ea56b 100644 --- a/crates/jp_cli/src/cmd/query/turn_loop.rs +++ b/crates/jp_cli/src/cmd/query/turn_loop.rs @@ -29,7 +29,7 @@ use jp_inquire::prompt::PromptBackend; use jp_llm::{ Error as LlmError, Provider, error::StreamError, - event::{Event, EventPart, ToolCallPart}, + event::{Event, EventPart, FinishReason, ToolCallPart}, model::ModelDetails, provider::get_provider, query::ChatQuery, @@ -48,7 +48,7 @@ use super::{ LoopAction, StreamingInterruptResult, handle_llm_event, handle_streaming_interrupt, reply_edit_mode, }, - stream::{StreamErrorOutcome, StreamRetryState, handle_stream_error}, + stream::{StreamErrorOutcome, StreamRetryState, commit_partial_response, handle_stream_error}, tool::{ PendingEntry, PendingTools, ToolCallDecision, ToolCallState, ToolCoordinator, ToolPrompter, ToolRenderer, build_execution_plan, @@ -502,11 +502,25 @@ pub(super) async fn run_turn_loop( } }; - // Reset the retry counter on the first successful + // Reset the retry counters on the first successful // event in this cycle. This ensures that partially // successful streams (rate-limited mid-response) // don't permanently consume the retry budget. - if !received_provider_event { + // + // Patches, keep-alives and rebuild requests are + // excluded: a repair cycle consists only of those, so + // counting them would clear the rebuild budget on + // every rebuilt request and its cap could never be + // reached. `reset` also drops the pending patch + // record, which a rebuild request needs intact to be + // authorized at all. + let advances_cycle = !matches!( + event, + Event::Patch(_) + | Event::KeepAlive + | Event::Finished(FinishReason::Retry) + ); + if !received_provider_event && advances_cycle { received_provider_event = true; stream_retry.clear_line(&printer); stream_retry.reset(); @@ -561,6 +575,7 @@ pub(super) async fn run_turn_loop( // Persist what was streamed and surface the dead // end, which the refusal describes. LoopAction::RebuildRefused(refusal) => { + commit_partial_response(&mut turn_coordinator, &conv, &printer); if let Err(err) = conv.flush() { warn!("Failed to persist before abort: {err}"); } 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 e76124d1..502155b4 100644 --- a/crates/jp_cli/src/cmd/query/turn_loop_tests.rs +++ b/crates/jp_cli/src/cmd/query/turn_loop_tests.rs @@ -25,7 +25,7 @@ use jp_config::{ model::id::{self, ProviderId}, }; use jp_conversation::{ - Conversation, + Conversation, ConversationEvent, event::{ChatRequest, ChatResponse, InquirySource, ToolCallRequest, TurnStart}, }; use jp_inquire::{ @@ -35,7 +35,7 @@ use jp_inquire::{ use jp_llm::{ Error as LlmError, EventStream, Provider, error::StreamError, - event::{Event, FinishReason}, + event::{Event, EventMatcher, EventPatch, FinishReason, PatchAction}, model::ModelDetails, provider::mock::MockProvider, query::ChatQuery, @@ -58,7 +58,10 @@ use tokio_util::sync::CancellationToken; use super::*; use crate::{ - cmd::query::tool::{ToolCoordinator, executor::TerminalExecutorSource}, + cmd::query::{ + stream::retry::MAX_CONSECUTIVE_REBUILDS, + tool::{ToolCoordinator, executor::TerminalExecutorSource}, + }, signals::testing::{detached_router, test_router}, }; @@ -5619,3 +5622,193 @@ async fn reasoning_before_a_tool_call_shades_the_tool_chrome() { "the shaded header text should still be present.\nChrome:\n{chrome:?}" ); } + +/// Metadata key a provider repair patch targets in the rebuild tests. +fn rebuild_patch_key(round: u32) -> String { + format!("stale_signature_{round}") +} + +const REBUILD_PATCH_VALUE: &str = "stale"; + +/// One repair cycle: strip the metadata for `round`, then ask for a rebuild. +/// +/// This is the shape a provider repair actually takes. +/// The rejection is a request-validation error, so nothing streams before it. +fn repair_cycle(round: u32) -> Vec { + vec![ + Event::Patch(vec![EventPatch { + matcher: EventMatcher::MetadataValue { + key: rebuild_patch_key(round), + value: REBUILD_PATCH_VALUE.to_owned(), + }, + action: PatchAction::RemoveMetadata(rebuild_patch_key(round)), + }]), + Event::Finished(FinishReason::Retry), + ] +} + +/// Seed one assistant event carrying a distinct patchable key per round, so +/// every repair cycle has something of its own to remove and therefore makes +/// real progress. +fn seed_patchable_metadata(lock: &jp_workspace::ConversationLock, rounds: u32) { + lock.as_mut().update_events(|stream| { + // A complete prior turn: an incomplete trailing turn would be trimmed + // when the loop starts its own. + stream.start_turn(ChatRequest::from("earlier question")); + + let mut event = ConversationEvent::now(ChatResponse::reasoning("thinking")); + for round in 0..rounds { + event + .metadata + .insert(rebuild_patch_key(round), REBUILD_PATCH_VALUE.into()); + } + + stream + .current_turn_mut() + .add_event(event) + .build() + .expect("valid stream"); + }); +} + +/// A provider that keeps patching and asking for a rebuild is stopped by the +/// consecutive-rebuild cap. +/// +/// Every cycle here makes genuine progress, so the progress guard never fires +/// and the cap is the only thing that can end the turn. +/// The mock is scripted with exactly one cycle more than the cap allows: if the +/// cap regresses, the loop asks for a further batch and the mock panics rather +/// than looping silently. +#[tokio::test] +async fn test_rebuild_cap_stops_a_provider_that_keeps_requesting_rebuilds() { + let tmp = tempdir().unwrap(); + let root = tmp.path(); + let storage = root.join(".jp"); + + let config = AppConfig::new_test(); + let fs = Arc::new(FsStorageBackend::new(&storage).expect("failed to create backend")); + let mut workspace = Workspace::new(root).with_backend(fs.clone()); + + let lock = workspace + .create_and_lock_conversation(Conversation::default(), config.clone().into(), None) + .unwrap(); + + let rounds = MAX_CONSECUTIVE_REBUILDS + 1; + seed_patchable_metadata(&lock, rounds); + + let batches = (0..rounds).map(repair_cycle).collect(); + let provider: Arc = Arc::new(MockProvider::with_batches(batches)); + let model = provider + .model_details(&"test-model".parse().unwrap()) + .await + .unwrap(); + + let (printer, _out, _err) = Printer::memory(OutputFormat::TextPretty); + let printer = Arc::new(printer); + let mcp_client = jp_mcp::Client::default(); + let (router, _signals) = test_router(); + let router = Arc::new(router); + + let result = run_turn_loop( + Arc::clone(&provider), + &model, + &config, + &router, + &mcp_client, + root, + false, + &[], + &lock, + ToolChoice::Auto, + &[], + printer.clone(), + Arc::new(MockPromptBackend::new()), + ToolCoordinator::new(config.conversation.tools.clone(), empty_executor_source()), + ChatRequest::from("repair this"), + InvocationContext::default(), + PendingStreamTrim::default(), + ) + .await; + + let error = result.expect_err("the turn must abort once the rebuild cap is reached"); + // The outer error only renders "LLM error"; the refusal is in the source. + let message = format!("{error:?}"); + assert!( + message.contains("times in a row"), + "the abort should name the rebuild cap.\nError:\n{message}" + ); +} + +/// Content streamed before a refused rebuild survives the abort. +/// +/// A part sits in the coordinator's event builder until a flush or terminal +/// event reaches it, and the rebuild request is intercepted before the +/// coordinator sees it, so persisting the conversation alone would drop text +/// the user already saw. +#[tokio::test] +async fn test_refused_rebuild_persists_streamed_content() { + let tmp = tempdir().unwrap(); + let root = tmp.path(); + let storage = root.join(".jp"); + + let config = AppConfig::new_test(); + let fs = Arc::new(FsStorageBackend::new(&storage).expect("failed to create backend")); + let mut workspace = Workspace::new(root).with_backend(fs.clone()); + + let lock = workspace + .create_and_lock_conversation(Conversation::default(), config.clone().into(), None) + .unwrap(); + let conv_id = lock.id(); + + // No patch precedes the rebuild request, so it is refused for lack of + // progress while a streamed part is still unflushed. + let provider: Arc = Arc::new(MockProvider::with_batches(vec![vec![ + Event::message(0, "partial answer"), + Event::Finished(FinishReason::Retry), + ]])); + let model = provider + .model_details(&"test-model".parse().unwrap()) + .await + .unwrap(); + + let (printer, _out, _err) = Printer::memory(OutputFormat::TextPretty); + let printer = Arc::new(printer); + let mcp_client = jp_mcp::Client::default(); + let (router, _signals) = test_router(); + let router = Arc::new(router); + + let result = run_turn_loop( + Arc::clone(&provider), + &model, + &config, + &router, + &mcp_client, + root, + false, + &[], + &lock, + ToolChoice::Auto, + &[], + printer.clone(), + Arc::new(MockPromptBackend::new()), + ToolCoordinator::new(config.conversation.tools.clone(), empty_executor_source()), + ChatRequest::from("answer this"), + InvocationContext::default(), + PendingStreamTrim::default(), + ) + .await; + + assert!( + result.is_err(), + "a rebuild without a preceding patch must abort the turn" + ); + + let content = fs + .read_test_events_raw(&conv_id) + .expect("events should be persisted"); + + assert!( + content.contains("partial answer"), + "streamed content must survive the abort.\nFile contents:\n{content}" + ); +} From f702381f84c9bf3489c7b5f719b4d83599a8b16f Mon Sep 17 00:00:00 2001 From: Jean Mertz Date: Fri, 31 Jul 2026 14:03:43 +0200 Subject: [PATCH 3/4] review feedback Signed-off-by: Jean Mertz --- crates/jp_cli/src/cmd/query/turn_loop.rs | 30 ++-- .../jp_cli/src/cmd/query/turn_loop_tests.rs | 6 + crates/jp_llm/src/provider/anthropic.rs | 101 ++++++++++--- crates/jp_llm/src/provider/anthropic_tests.rs | 139 ++++++++++++++++-- 4 files changed, 231 insertions(+), 45 deletions(-) diff --git a/crates/jp_cli/src/cmd/query/turn_loop.rs b/crates/jp_cli/src/cmd/query/turn_loop.rs index 9c3ea56b..059b638a 100644 --- a/crates/jp_cli/src/cmd/query/turn_loop.rs +++ b/crates/jp_cli/src/cmd/query/turn_loop.rs @@ -507,19 +507,23 @@ pub(super) async fn run_turn_loop( // successful streams (rate-limited mid-response) // don't permanently consume the retry budget. // - // Patches, keep-alives and rebuild requests are - // excluded: a repair cycle consists only of those, so - // counting them would clear the rebuild budget on - // every rebuilt request and its cap could never be - // reached. `reset` also drops the pending patch - // record, which a rebuild request needs intact to be - // authorized at all. - let advances_cycle = !matches!( - event, - Event::Patch(_) - | Event::KeepAlive - | Event::Finished(FinishReason::Retry) - ); + // Only a content part or a non-repair terminal event + // counts. A repair cycle is made of patches, + // keep-alives, part-less flushes and the rebuild + // request itself; counting any of them clears the + // rebuild budget on every rebuilt request, so its cap + // could never be reached. `reset` also drops the + // pending patch record, which a rebuild request needs + // intact to be authorized at all. + // + // A content-bearing flush is always preceded by a + // `Part` for the same index, which already reset, so + // excluding `Flush` here loses nothing. + let advances_cycle = match &event { + Event::Part { .. } => true, + Event::Finished(reason) => *reason != FinishReason::Retry, + Event::Flush { .. } | Event::Patch(_) | Event::KeepAlive => false, + }; if !received_provider_event && advances_cycle { received_provider_event = true; stream_retry.clear_line(&printer); 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 502155b4..ddf5d931 100644 --- a/crates/jp_cli/src/cmd/query/turn_loop_tests.rs +++ b/crates/jp_cli/src/cmd/query/turn_loop_tests.rs @@ -5634,8 +5634,14 @@ const REBUILD_PATCH_VALUE: &str = "stale"; /// /// This is the shape a provider repair actually takes. /// The rejection is a request-validation error, so nothing streams before it. +/// +/// The leading flush has no buffered part behind it, so it commits nothing. +/// It is here because a cycle that renders nothing must not count as progress: +/// if it did, the rebuild budget would reset on every rebuilt request and the +/// cap could never be reached. fn repair_cycle(round: u32) -> Vec { vec![ + Event::flush(0), Event::Patch(vec![EventPatch { matcher: EventMatcher::MetadataValue { key: rebuild_patch_key(round), diff --git a/crates/jp_llm/src/provider/anthropic.rs b/crates/jp_llm/src/provider/anthropic.rs index cbfbb3a7..18d77eaf 100644 --- a/crates/jp_llm/src/provider/anthropic.rs +++ b/crates/jp_llm/src/provider/anthropic.rs @@ -310,10 +310,11 @@ fn call( // accept as a 400 `invalid_request_error`. Emit a patch to strip the // offending metadata and ask the caller to retry. if let Err(ref err) = result - && is_thinking_block_rejection(err) - && let Some(patches) = build_thinking_patches(&request, err) + && let Some(rejection) = classify_thinking_rejection(err) + && let Some(patches) = build_thinking_patches(&request, err, rejection) { warn!( + ?rejection, patches = patches.len(), "Thinking block rejected by the API, patching history and retrying: {err}" ); @@ -689,21 +690,50 @@ fn soft_force_retry( })) } -/// Returns `true` if the error is an Anthropic `invalid_request_error` about a -/// `thinking` block the API refuses to accept. +/// Why the API refused a thinking block. /// -/// Two rejections fall in this class: an invalid `signature` on a thinking -/// block, and thinking blocks in the latest assistant message having been -/// modified. -/// Both are repaired the same way, by replaying the reasoning as `` text -/// instead of a native thinking block. -fn is_thinking_block_rejection(error: &StreamError) -> bool { +/// Both kinds are repaired the same way, by replaying the reasoning as +/// `` text instead of a native thinking block. +/// They differ in where the offending block can be, which is what decides where +/// to look when the error carries no usable position. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum ThinkingRejection { + /// An invalid `signature` on a thinking block. + /// + /// Any block in the request can carry a stale signature. + InvalidSignature, + + /// Thinking blocks in the latest assistant message were modified. + /// + /// This always concerns the final assistant turn, whatever position the + /// message names. + UnmodifiableTurn, +} + +/// Classify an Anthropic `invalid_request_error` about a `thinking` block the +/// API refuses to accept. +/// +/// Returns `None` for every other error. +fn classify_thinking_rejection(error: &StreamError) -> Option { + if error.kind != StreamErrorKind::Other { + return None; + } + let message = error.message(); + if !(message.contains("invalid_request_error") && message.contains("thinking")) { + return None; + } + + // Tested before the signature wording: the unmodifiable-blocks message names + // `thinking` blocks without mentioning a signature, and when a request draws + // both complaints the turn-scoped one is the more specific. + if message.contains("cannot be modified") { + return Some(ThinkingRejection::UnmodifiableTurn); + } - error.kind == StreamErrorKind::Other - && message.contains("invalid_request_error") - && message.contains("thinking") - && (message.contains("signature") || message.contains("cannot be modified")) + message + .contains("signature") + .then_some(ThinkingRejection::InvalidSignature) } /// Extract the `(turn_index, flat_content_index)` from an Anthropic error @@ -806,15 +836,19 @@ fn is_tool_result_message(msg: &types::Message) -> bool { /// Blocks in earlier turns are downgraded individually, keeping the rest of /// their reasoning native. /// +/// When the error carries no usable position, [`fallback_thinking_block`] picks +/// where to look based on `rejection`. +/// /// Returns `None` if no thinking block can be located. fn build_thinking_patches( request: &types::CreateMessagesRequest, error: &StreamError, + rejection: ThinkingRejection, ) -> Option> { let api_position = parse_signature_error_position(error.message()); let (msg_idx, content_idx) = api_position .and_then(|(turn, flat)| resolve_turn_position(&request.messages, turn, flat)) - .or_else(|| find_oldest_thinking_block(request))?; + .or_else(|| fallback_thinking_block(request, rejection))?; debug!( ?api_position, @@ -840,13 +874,46 @@ fn build_thinking_patches( return Some(patches); } - // The located position held no thinking block. Fall back to the oldest one. - let (msg_idx, content_idx) = find_oldest_thinking_block(request)?; + // The located position held no thinking block. + let (msg_idx, content_idx) = fallback_thinking_block(request, rejection)?; let (key, value) = identify_thinking_block(request, msg_idx, content_idx)?; Some(vec![thinking_patch(key, value)]) } +/// Where to look for the offending thinking block when the error carries no +/// position, or names one that does not resolve. +/// +/// A turn-scoped rejection identifies the final assistant turn even without a +/// position, so the search stays inside it: stripping a block from an earlier +/// turn cannot satisfy that complaint, and costs a valid signature per attempt. +/// A stale signature can sit anywhere, so that search starts at the oldest +/// block and later rounds walk forward. +fn fallback_thinking_block( + request: &types::CreateMessagesRequest, + rejection: ThinkingRejection, +) -> Option<(usize, usize)> { + match rejection { + ThinkingRejection::UnmodifiableTurn => last_turn_thinking_block(request), + ThinkingRejection::InvalidSignature => find_oldest_thinking_block(request), + } +} + +/// The newest thinking-bearing message in the final assistant turn, with the +/// content index of its first thinking block. +/// +/// Returns `None` when that turn carries no native thinking block, which leaves +/// the caller reporting the error rather than patching something unrelated. +fn last_turn_thinking_block(request: &types::CreateMessagesRequest) -> Option<(usize, usize)> { + let turn = last_assistant_turn(&request.messages)?; + + turn.rev().find_map(|msg_idx| { + thinking_block_indices(request, msg_idx) + .first() + .map(|&content_idx| (msg_idx, content_idx)) + }) +} + /// Build the patch that strips a thinking block's provider metadata, so the /// event is replayed as `` text rather than a native thinking block. fn thinking_patch(key: &'static str, value: String) -> EventPatch { diff --git a/crates/jp_llm/src/provider/anthropic_tests.rs b/crates/jp_llm/src/provider/anthropic_tests.rs index 9440c180..cdd8eb40 100644 --- a/crates/jp_llm/src/provider/anthropic_tests.rs +++ b/crates/jp_llm/src/provider/anthropic_tests.rs @@ -2242,11 +2242,25 @@ mod thinking_signature_recovery { error::StreamError, event::{EventMatcher, EventPatch, PatchAction}, provider::anthropic::{ - build_thinking_patches, find_oldest_thinking_block, identify_thinking_block, - is_thinking_block_rejection, parse_signature_error_position, resolve_turn_position, + ThinkingRejection, build_thinking_patches, classify_thinking_rejection, + find_oldest_thinking_block, identify_thinking_block, parse_signature_error_position, + resolve_turn_position, }, }; + /// Build patches the way the provider does: classify the error, then + /// repair. + /// + /// Going through the classifier keeps these expectations honest about which + /// rejection kind each error text produces. + fn patches_for( + request: &types::CreateMessagesRequest, + error: &StreamError, + ) -> Option> { + let rejection = classify_thinking_rejection(error).expect("a thinking-block rejection"); + build_thinking_patches(request, error, rejection) + } + /// The `(metadata_key, metadata_value)` each patch targets, in order. fn patch_targets(patches: &[EventPatch]) -> Vec<(&str, &str)> { patches @@ -2344,7 +2358,10 @@ mod thinking_signature_recovery { "api error: invalid_request_error: messages.1.content.0: Invalid `signature` in \ `thinking` block", ); - assert!(is_thinking_block_rejection(&error)); + assert_eq!( + classify_thinking_rejection(&error), + Some(ThinkingRejection::InvalidSignature) + ); } #[test] @@ -2354,19 +2371,22 @@ mod thinking_signature_recovery { `redacted_thinking` blocks in the latest assistant message cannot be modified. These \ blocks must remain as they were in the original response.", ); - assert!(is_thinking_block_rejection(&error)); + assert_eq!( + classify_thinking_rejection(&error), + Some(ThinkingRejection::UnmodifiableTurn) + ); } #[test] fn ignores_unrelated_errors() { let error = StreamError::other("api error: rate_limit_error: too many requests"); - assert!(!is_thinking_block_rejection(&error)); + assert_eq!(classify_thinking_rejection(&error), None); } #[test] fn ignores_retryable_errors() { let error = StreamError::transient("server error with signature and thinking"); - assert!(!is_thinking_block_rejection(&error)); + assert_eq!(classify_thinking_rejection(&error), None); } #[test] @@ -2472,7 +2492,7 @@ mod thinking_signature_recovery { `thinking` block", ); - let patches = build_thinking_patches(&request, &error).unwrap(); + let patches = patches_for(&request, &error).unwrap(); assert_eq!(patches.len(), 1); assert_eq!(patches[0].matcher, EventMatcher::MetadataValue { key: "anthropic_thinking_signature".to_owned(), @@ -2499,7 +2519,7 @@ mod thinking_signature_recovery { "api error: invalid_request_error: Invalid `signature` in `thinking` block", ); - let patches = build_thinking_patches(&request, &error).unwrap(); + let patches = patches_for(&request, &error).unwrap(); assert_eq!(patches.len(), 1); assert_eq!(patches[0].matcher, EventMatcher::MetadataValue { key: "anthropic_thinking_signature".to_owned(), @@ -2519,7 +2539,96 @@ mod thinking_signature_recovery { `thinking` block", ); - assert!(build_thinking_patches(&request, &error).is_none()); + assert!(patches_for(&request, &error).is_none()); + } + + /// Anthropic's unmodifiable-blocks message concerns the latest assistant + /// turn by definition, so a rejection carrying no `messages.N.content.M` + /// still identifies where to repair. + /// + /// Reaching past that turn cannot satisfy the complaint and costs a valid + /// signature per attempt. + #[test] + fn unmodifiable_error_without_a_position_downgrades_the_final_turn() { + let request = request(vec![ + msg(MessageRole::User, vec![make_text("first")]), + msg(MessageRole::Assistant, vec![ + make_thinking("early", "sig_early"), + make_text("answer"), + ]), + msg(MessageRole::User, vec![make_text("second")]), + msg(MessageRole::Assistant, vec![ + make_redacted("r0"), + make_thinking("t1", "sig_1"), + make_tool_use("tu1"), + ]), + msg(MessageRole::User, vec![make_tool_result("tu1")]), + ]); + + let error = StreamError::other( + "api error: invalid_request_error: `thinking` or `redacted_thinking` blocks in the \ + latest assistant message cannot be modified.", + ); + + let patches = patches_for(&request, &error).unwrap(); + + assert_eq!(patch_targets(&patches), [ + ("anthropic_redacted_thinking", "r0"), + ("anthropic_thinking_signature", "sig_1"), + ]); + } + + /// A position that parses but does not resolve is no better than none, so + /// it takes the same turn-scoped path. + #[test] + fn unmodifiable_error_with_an_unresolvable_position_downgrades_the_final_turn() { + let request = request(vec![ + msg(MessageRole::User, vec![make_text("first")]), + msg(MessageRole::Assistant, vec![make_thinking( + "early", + "sig_early", + )]), + msg(MessageRole::User, vec![make_text("second")]), + msg(MessageRole::Assistant, vec![make_thinking("t1", "sig_1")]), + ]); + + let error = StreamError::other( + "api error: invalid_request_error: messages.1.content.999: `thinking` or \ + `redacted_thinking` blocks in the latest assistant message cannot be modified.", + ); + + let patches = patches_for(&request, &error).unwrap(); + + assert_eq!(patch_targets(&patches), [( + "anthropic_thinking_signature", + "sig_1" + )]); + } + + /// A stale signature can sit anywhere, so a position-less rejection of that + /// kind still starts at the oldest block and walks forward over rounds. + #[test] + fn signature_error_without_a_position_falls_back_to_the_oldest_block() { + let request = request(vec![ + msg(MessageRole::User, vec![make_text("first")]), + msg(MessageRole::Assistant, vec![make_thinking( + "early", + "sig_early", + )]), + msg(MessageRole::User, vec![make_text("second")]), + msg(MessageRole::Assistant, vec![make_thinking("t1", "sig_1")]), + ]); + + let error = StreamError::other( + "api error: invalid_request_error: Invalid `signature` in `thinking` block", + ); + + let patches = patches_for(&request, &error).unwrap(); + + assert_eq!(patch_targets(&patches), [( + "anthropic_thinking_signature", + "sig_early" + )]); } /// A tool-use loop whose newest assistant message interleaves redacted and @@ -2563,7 +2672,7 @@ mod thinking_signature_recovery { `thinking` block", ); - let patches = build_thinking_patches(&request, &error).unwrap(); + let patches = patches_for(&request, &error).unwrap(); assert_eq!(patch_targets(&patches), [ ("anthropic_redacted_thinking", "r0"), @@ -2603,7 +2712,7 @@ mod thinking_signature_recovery { `redacted_thinking` blocks in the latest assistant message cannot be modified.", ); - let patches = build_thinking_patches(&request, &error).unwrap(); + let patches = patches_for(&request, &error).unwrap(); assert_eq!(patch_targets(&patches), [ ("anthropic_redacted_thinking", "r0"), @@ -2638,7 +2747,7 @@ mod thinking_signature_recovery { `redacted_thinking` blocks in the latest assistant message cannot be modified.", ); - let patches = build_thinking_patches(&request, &error).unwrap(); + let patches = patches_for(&request, &error).unwrap(); assert_eq!(patch_targets(&patches), [ ("anthropic_redacted_thinking", "r0"), @@ -2668,7 +2777,7 @@ mod thinking_signature_recovery { `thinking` block", ); - let patches = build_thinking_patches(&request, &error).unwrap(); + let patches = patches_for(&request, &error).unwrap(); assert_eq!(patch_targets(&patches), [( "anthropic_thinking_signature", @@ -2698,7 +2807,7 @@ mod thinking_signature_recovery { `redacted_thinking` blocks in the latest assistant message cannot be modified.", ); - let patches = build_thinking_patches(&request, &error).unwrap(); + let patches = patches_for(&request, &error).unwrap(); assert_eq!(patch_targets(&patches), [ ("anthropic_redacted_thinking", "r0"), @@ -2927,7 +3036,7 @@ mod thinking_signature_recovery { `thinking` block", ); - let patches = build_thinking_patches(&request, &error).unwrap(); + let patches = patches_for(&request, &error).unwrap(); assert_eq!(patch_targets(&patches), [ ("anthropic_redacted_thinking", "r0"), From a71b356667bbd1dde6eb0c9e70faf4c815fdf1a9 Mon Sep 17 00:00:00 2001 From: Jean Mertz Date: Fri, 31 Jul 2026 14:31:31 +0200 Subject: [PATCH 4/4] review feedback Signed-off-by: Jean Mertz --- crates/jp_cli/src/cmd/query/turn_loop.rs | 6 + .../jp_cli/src/cmd/query/turn_loop_tests.rs | 105 ++++++++++++++++++ 2 files changed, 111 insertions(+) diff --git a/crates/jp_cli/src/cmd/query/turn_loop.rs b/crates/jp_cli/src/cmd/query/turn_loop.rs index 059b638a..6565e1e9 100644 --- a/crates/jp_cli/src/cmd/query/turn_loop.rs +++ b/crates/jp_cli/src/cmd/query/turn_loop.rs @@ -579,6 +579,12 @@ pub(super) async fn run_turn_loop( // Persist what was streamed and surface the dead // end, which the refusal describes. LoopAction::RebuildRefused(refusal) => { + // A repair cycle renders nothing, so no event + // reached the clear above. Retire any retry + // line before the commit below flushes + // buffered output, which would otherwise land + // after the parked cursor. + stream_retry.clear_line(&printer); commit_partial_response(&mut turn_coordinator, &conv, &printer); if let Err(err) = conv.flush() { warn!("Failed to persist before abort: {err}"); 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 ddf5d931..1e857b2d 100644 --- a/crates/jp_cli/src/cmd/query/turn_loop_tests.rs +++ b/crates/jp_cli/src/cmd/query/turn_loop_tests.rs @@ -5745,6 +5745,111 @@ async fn test_rebuild_cap_stops_a_provider_that_keeps_requesting_rebuilds() { ); } +/// A refused rebuild ends the turn, so a retry line left by an earlier cycle +/// has to be retired first. +/// +/// A repair cycle renders nothing, so no event reaches the clear on the success +/// path, and the final error would otherwise be written onto the notification. +#[tokio::test(flavor = "multi_thread")] +async fn test_refused_rebuild_clears_the_retry_line() { + let test_result = Box::pin(timeout(Duration::from_secs(10), async { + let tmp = tempdir().unwrap(); + let root = tmp.path(); + let storage = root.join(".jp"); + + let mut config = AppConfig::new_test(); + // The waiting indicator writes its own erase sequences, which would + // blur what this test asserts. + config.style.streaming.progress.show = false; + // Keep the retry backoff out of the test's runtime. + config.assistant.request.base_backoff_ms = 1; + + let fs = Arc::new(FsStorageBackend::new(&storage).expect("failed to create backend")); + let mut workspace = Workspace::new(root).with_backend(fs.clone()); + + let lock = workspace + .create_and_lock_conversation(Conversation::default(), Arc::new(config.clone()), None) + .unwrap(); + + // First cycle: a retryable error, which writes the retry notification and + // leaves the cursor parked at the end of it. + // Second cycle: a patch matching no event, so the rebuild that follows is + // refused for lack of progress and the turn aborts. + let provider: Arc = Arc::new(PacedMockProvider::new(Duration::ZERO, vec![ + vec![( + Duration::ZERO, + Err(StreamError::transient("simulated hiccup")), + )], + vec![ + ( + Duration::ZERO, + Ok(Event::Patch(vec![EventPatch { + matcher: EventMatcher::MetadataValue { + key: "absent_key".into(), + value: "absent_value".into(), + }, + action: PatchAction::RemoveMetadata("absent_key".into()), + }])), + ), + (Duration::ZERO, Ok(Event::Finished(FinishReason::Retry))), + ], + ])); + let model = provider + .model_details(&"test-model".parse().unwrap()) + .await + .unwrap(); + + let (printer, _out, err) = Printer::memory(OutputFormat::TextPretty); + let printer = Arc::new(printer); + let mcp_client = jp_mcp::Client::default(); + let router = detached_router(); + + let result = run_turn_loop( + Arc::clone(&provider), + &model, + &config, + &router, + &mcp_client, + root, + true, // is_tty + &[], + &lock, + ToolChoice::Auto, + &[], + printer.clone(), + Arc::new(MockPromptBackend::new()), + ToolCoordinator::new(config.conversation.tools.clone(), empty_executor_source()), + ChatRequest::from("answer this"), + InvocationContext::default(), + PendingStreamTrim::default(), + ) + .await; + + assert!( + result.is_err(), + "a rebuild without progress must abort the turn" + ); + + printer.flush(); + + let chrome = err.lock(); + assert!( + chrome.contains("retrying (1/"), + "the first cycle should have written a retry notice.\nChrome:\n{chrome}" + ); + // The notice writes its own `\r\x1b[K` prefix, so occurrences cannot be + // counted. What distinguishes a retired line is that the erase is the + // last thing written, leaving the terminal clean for the final error. + assert!( + chrome.ends_with("\r\x1b[K"), + "the retry line must be retired before the turn aborts.\nChrome:\n{chrome:?}" + ); + })) + .await; + + assert!(test_result.is_ok(), "Test timed out"); +} + /// Content streamed before a refused rebuild survives the abort. /// /// A part sits in the coordinator's event builder until a flush or terminal