From fdae9c41c7776018ac25fc19bd7a51fa85505051 Mon Sep 17 00:00:00 2001 From: The Gru <163427431+dgruhin-hrizn@users.noreply.github.com> Date: Sun, 6 Sep 2026 08:33:03 -0400 Subject: [PATCH] fix(run): make automatic compaction recover an over-limit conversation (#426) Automatic compaction cannot do its job once a conversation crosses the context window, so the conversation stays there permanently. Observed against a 1M-token Anthropic window: 1. The summarize call replays the full history. It runs precisely because that history is too large, so the request is itself over the limit ("prompt is too long"), or it ends with an assistant/tool message that Anthropic refuses as a prefill. Either way the run falls back to the 12K truncated JSON summary, which discards the context. In one trace the summarizer received 771 messages (2.78 MB) and returned a single token. 2. The compaction check uses a 10K fixed reserve. The estimate trails the provider's own count by the request context and provider-side overhead that the message-tail estimate does not model; a 948K estimate passed the check and Anthropic counted 1,017,628. 3. When the provider does refuse the prompt, the run retries the same prompt eight times at 5s intervals and then fails. Nothing compacts. Fixes, all in server/src/run: - compaction_history trims the summarizer input to the context budget at user-turn boundaries (never splitting a tool call from its results) and guarantees it ends with a user message. - context_budget keeps 10% of the window free instead of a fixed 10K, so the reserve scales with the model and absorbs the drift. - A provider refusal matching is_context_overflow compacts once and retries the turn instead of failing it. - 4xx responses other than 408/425/429 are terminal. A rejected request fails identically every time, so retrying only delays the error. Separately, Cursor can resume a finished turn whose checkpoint already ends with the assistant, which Anthropic also rejects as a prefill. run/history.rs appends a transient user tail to every provider request that would otherwise end with the assistant. The tail is never persisted, so committed checkpoints stay an exact prefix of the next turn and the usage anchor still counts persisted messages only. --- server/src/run/compaction.rs | 275 +++++++++++++++++++++++++++++++--- server/src/run/engine.rs | 85 ++++++++++- server/src/run/history.rs | 60 ++++++++ server/src/run/mod.rs | 1 + server/src/run/model_cycle.rs | 73 ++++++++- server/tests/compaction.rs | 192 +++++++++++++++++++++++- 6 files changed, 663 insertions(+), 23 deletions(-) create mode 100644 server/src/run/history.rs diff --git a/server/src/run/compaction.rs b/server/src/run/compaction.rs index 9cb45b6..a42238f 100644 --- a/server/src/run/compaction.rs +++ b/server/src/run/compaction.rs @@ -5,22 +5,118 @@ use std::collections::HashSet; use crate::{ model::{ estimate_context_tokens, estimate_projected_messages_tokens, CanonicalMessage, PreparedRun, - ProjectedMessage, + ProjectedContent, ProjectedMessage, Role, }, store::ContextUsageAnchor, }; const FALLBACK_CHARS: usize = 12_000; -pub(super) const RESERVE_TOKENS: u64 = 10_000; +/// Fraction of the context window kept free, as a divisor: 10 = 10%. +/// +/// The estimate runs behind the provider: the anchor is what the provider +/// charged for the *previous* call, and the next request re-sends request +/// context and carries provider-side overhead the message-tail estimate does +/// not model. A real conversation measured 948K estimated against 1,017,628 +/// actual, a 7% shortfall that landed it over a 1M window while the check said +/// there was room. A proportional reserve absorbs that drift and scales with +/// the model: a 200K window keeps 20K free and a 1M window keeps 100K. +const CONTEXT_RESERVE_DIVISOR: u64 = 10; pub(super) const OUTPUT_TOKENS: u64 = 4_096; pub(super) const INSTRUCTIONS: &str = "Summarize the conversation for the next model turn. Preserve goals, constraints, decisions, files, commands, errors, results, and unfinished work. Do not call tools. Return only the concise durable summary."; +/// Usable prompt budget: the window minus the proportional reserve. +pub(super) fn context_budget(context_window: u64) -> u64 { + context_window.saturating_sub(context_window / CONTEXT_RESERVE_DIVISOR) +} + pub(super) fn input_budget(prepared: &PreparedRun) -> Option { - prepared - .model - .context_window_tokens - .map(|window| window.saturating_sub(RESERVE_TOKENS)) + prepared.model.context_window_tokens.map(context_budget) +} + +/// Whether a provider failure means the prompt did not fit. +/// +/// Providers report this as a plain 400 with prose, so there is nothing +/// structured to match on. Anthropic says "prompt is too long"; OpenAI-style +/// gateways use `context_length_exceeded` or "maximum context length". +pub(super) fn is_context_overflow(message: &str) -> bool { + let lowered = message.to_ascii_lowercase(); + lowered.contains("prompt is too long") + || lowered.contains("context window exceeded") + || lowered.contains("model_context_window_exceeded") + || lowered.contains("context_length_exceeded") + || (lowered.contains("maximum context length") && lowered.contains("token")) +} + +/// Builds the history for a compaction call. +/// +/// Two properties matter, and replaying the raw history guarantees neither: +/// +/// 1. The history must end with a user message. Otherwise providers read the +/// request as an assistant prefill and refuse it outright: Anthropic answers +/// "This model does not support assistant message prefill. The conversation +/// must end with a user message." +/// 2. The history must fit the context window. Compaction runs precisely +/// because the conversation is too large, so replaying all of it asks the +/// summarizer to accept a prompt that is already over the limit and the +/// call fails with "prompt is too long". +/// +/// Either failure falls back to the truncated summary, which is usually still +/// too large, so the conversation stays over its window and cannot recover. +/// +/// Trimming keeps the most recent turns and only ever cuts at a user-message +/// boundary, so an assistant tool call is never separated from its results. +pub(super) fn compaction_history( + history: Vec, + context_window: Option, +) -> Vec { + super::history::user_terminated( + trim_to_context(history, context_window), + "compaction:instruction", + INSTRUCTIONS, + ) +} + +fn trim_to_context( + mut history: Vec, + context_window: Option, +) -> Vec { + let Some(budget) = context_window + .filter(|window| *window > 0) + .map(context_budget) + .map(|budget| budget.saturating_sub(OUTPUT_TOKENS)) + .filter(|budget| *budget > 0) + else { + return history; + }; + if estimate_projected_messages_tokens(&history) <= budget { + return history; + } + // Walk back from the newest turn, keeping whole user-delimited turns. + let mut kept = 0; + let mut newest_turn = None; + for (index, message) in history.iter().enumerate().rev() { + if !is_turn_boundary(message) { + continue; + } + newest_turn.get_or_insert(index); + if estimate_projected_messages_tokens(&history[index..]) > budget { + break; + } + kept = history.len() - index; + } + if kept == 0 { + // Not even the newest turn fits. Keep it anyway rather than sending an + // empty history: an empty summarize call returns a summary of nothing + // that would then replace the whole conversation. + let start = newest_turn.unwrap_or(0); + return history.split_off(start); + } + history.split_off(history.len() - kept) +} + +fn is_turn_boundary(message: &ProjectedMessage) -> bool { + message.role == Role::User && matches!(message.content, ProjectedContent::Parts(_)) } pub(super) fn estimated_tokens( @@ -115,8 +211,8 @@ pub(super) fn fallback_summary(messages: &[CanonicalMessage]) -> String { mod tests { use super::*; use crate::model::{ - project_messages, CheckpointId, ConversationId, ModelSpec, Origin, PromptSpec, Role, - RunAction, RunId, RunKind, + project_messages, CheckpointId, ContentPart, ConversationId, ModelSpec, Origin, PromptSpec, + Role, RunAction, RunId, RunKind, }; fn prepared(context_window_tokens: u64) -> PreparedRun { @@ -139,7 +235,7 @@ mod tests { } #[test] - fn automatic_compaction_uses_fixed_reserve_for_every_action() { + fn automatic_compaction_uses_proportional_reserve_for_every_action() { let messages = vec![CanonicalMessage::text( "user", Role::User, @@ -148,10 +244,14 @@ mod tests { )]; let projected = project_messages(&messages).unwrap(); let estimated = estimate_context_tokens(&prepared(1).prompt, &projected); - let mut prepared = prepared(estimated + RESERVE_TOKENS); + // Smallest multiple-of-ten window whose 90% budget covers the estimate. + let window = estimated.div_ceil(9) * 10; + let mut prepared = prepared(window); + assert!(context_budget(window) >= estimated); + assert!(context_budget(window - 10) < estimated); assert!(!should_compact(&prepared, &projected, None)); - prepared.model.context_window_tokens = Some(estimated + RESERVE_TOKENS - 1); + prepared.model.context_window_tokens = Some(window - 10); assert!(should_compact(&prepared, &projected, None)); prepared.action = RunAction::Resume { @@ -160,6 +260,146 @@ mod tests { assert!(should_compact(&prepared, &projected, None)); } + #[test] + fn the_reserve_leaves_room_for_the_estimate_to_run_behind() { + // Reproduces a conversation that wedged itself against a 1M window. + // The anchor said 947,797 tokens, so a fixed 10K reserve found room + // and let the request through. Anthropic counted 1,017,628 and + // refused it, and every retry repeated the same arithmetic. + let messages = vec![CanonicalMessage::text( + "user", + Role::User, + Origin::Runtime, + "hello", + )]; + let projected = project_messages(&messages).unwrap(); + let anchor = ContextUsageAnchor { + context_input_tokens: 947_797, + message_count: 1, + }; + assert!(should_compact( + &prepared(1_000_000), + &projected, + Some(anchor) + )); + } + + #[test] + fn the_reserve_scales_with_the_window() { + assert_eq!(context_budget(200_000), 180_000); + assert_eq!(context_budget(1_000_000), 900_000); + assert_eq!(context_budget(0), 0); + assert_eq!(context_budget(1), 1); + } + + #[test] + fn provider_refusals_that_mean_the_prompt_did_not_fit_are_recognized() { + assert!(is_context_overflow( + "provider error: Anthropic 400 Bad Request: {\"type\":\"error\",\"error\":\ + {\"type\":\"invalid_request_error\",\"message\":\"prompt is too long: \ + 1017628 tokens > 1000000 maximum\"}}" + )); + assert!(is_context_overflow("model_context_window_exceeded")); + assert!(is_context_overflow("context_length_exceeded")); + assert!(is_context_overflow( + "This model's maximum context length is 128000 tokens" + )); + + // Unrelated failures must not trigger a compaction, which would + // destroy history to fix something compaction cannot fix. + assert!(!is_context_overflow("401 Unauthorized: invalid api key")); + assert!(!is_context_overflow("429 Too Many Requests")); + assert!(!is_context_overflow( + "This model does not support assistant message prefill" + )); + } + + fn user(id: &str, text: &str) -> ProjectedMessage { + ProjectedMessage { + message_id: id.into(), + role: Role::User, + content: ProjectedContent::Parts(vec![ContentPart::Text { text: text.into() }]), + } + } + + fn assistant(id: &str, text: &str) -> ProjectedMessage { + ProjectedMessage { + message_id: id.into(), + role: Role::Assistant, + content: ProjectedContent::Assistant { + text: text.into(), + thinking: String::new(), + replay_state: None, + calls: Vec::new(), + }, + } + } + + #[test] + fn compaction_history_always_ends_with_a_user_message() { + // Providers reject an assistant-terminated history as a prefill, which + // made every automatic compaction fall back to the truncated summary. + let history = vec![user("u1", "question"), assistant("a1", "answer")]; + let prepared = compaction_history(history, Some(200_000)); + assert_eq!(prepared.last().unwrap().role, Role::User); + assert_eq!( + prepared.last().unwrap().message_id, + "compaction:instruction" + ); + + // An already user-terminated history is left alone. + let history = vec![assistant("a1", "answer"), user("u2", "next")]; + let prepared = compaction_history(history.clone(), Some(200_000)); + assert_eq!(prepared, history); + } + + #[test] + fn compaction_history_is_trimmed_to_fit_the_context_window() { + // Compaction runs because the conversation is too large, so the + // summarize call must not replay a prompt that is over the window. + let big = "x".repeat(400_000); + let history = vec![ + user("u1", &big), + assistant("a1", &big), + user("u2", &big), + assistant("a2", "recent answer"), + ]; + let window = 200_000; + let prepared = compaction_history(history, Some(window)); + + let budget = context_budget(window) - OUTPUT_TOKENS; + assert!(estimate_projected_messages_tokens(&prepared) <= budget); + assert_eq!(prepared.last().unwrap().role, Role::User); + // The newest turn survives the trim and is never split. + assert!(prepared.iter().any(|message| message.message_id == "u2")); + assert!(prepared.iter().any(|message| message.message_id == "a2")); + assert!(!prepared.iter().any(|message| message.message_id == "u1")); + } + + #[test] + fn compaction_history_keeps_the_newest_turn_even_when_it_is_over_budget() { + // A history whose newest turn alone exceeds the budget is still sent + // rather than trimmed to nothing: a summary of nothing would replace + // the whole conversation. + let history = vec![ + user("u1", "old"), + assistant("a1", "old answer"), + user("u2", &"x".repeat(400_000)), + assistant("a2", "answer"), + ]; + let prepared = compaction_history(history, Some(50_000)); + assert_eq!(prepared[0].message_id, "u2"); + assert_eq!(prepared.last().unwrap().role, Role::User); + } + + #[test] + fn compaction_history_without_a_context_window_is_untouched_apart_from_termination() { + let history = vec![user("u1", "question"), assistant("a1", "answer")]; + let prepared = compaction_history(history.clone(), None); + assert_eq!(prepared[..2], history[..]); + assert_eq!(prepared.last().unwrap().role, Role::User); + } + #[test] fn provider_usage_anchor_only_estimates_messages_added_after_last_request() { let messages = vec![ @@ -253,15 +493,16 @@ mod tests { )]; let projected = project_messages(&messages).unwrap(); let estimated = estimate_context_tokens(&prepared(1).prompt, &projected); + let window = estimated.div_ceil(9) * 10; + assert!(context_budget(window) >= estimated); + assert!(context_budget(window - 10) < estimated); assert_eq!( - validate_compacted(&prepared(estimated + RESERVE_TOKENS), &projected), + validate_compacted(&prepared(window), &projected), Ok(estimated) ); - assert!( - validate_compacted(&prepared(estimated + RESERVE_TOKENS - 1), &projected) - .unwrap_err() - .contains("context overflow after compaction") - ); + assert!(validate_compacted(&prepared(window - 10), &projected) + .unwrap_err() + .contains("context overflow after compaction")); } } diff --git a/server/src/run/engine.rs b/server/src/run/engine.rs index 4694f36..fdd0393 100644 --- a/server/src/run/engine.rs +++ b/server/src/run/engine.rs @@ -25,6 +25,12 @@ pub struct RunEngine { provider: Arc, } +/// Provider-visible tail appended when the committed history ends with the +/// assistant, which happens when Cursor resumes a turn that had already +/// finished (for example after a cancelled follow-up). +const CONTINUE_MESSAGE_ID: &str = "runtime:continue"; +const CONTINUE_INSTRUCTION: &str = "Continue from where you left off."; + impl RunEngine { pub fn new(store: Store, provider: Arc) -> Self { Self { store, provider } @@ -171,6 +177,10 @@ impl RunEngine { }; } + // A provider refusal for an over-limit prompt triggers one compaction + // per run. A second refusal after compacting means the current input + // itself does not fit, and compacting again would only destroy history. + let mut overflow_compacted = false; 'model: loop { if cancellation.is_cancelled() { return (RunOutcome::Cancelled, usage); @@ -226,6 +236,18 @@ impl RunEngine { if let Err(error) = hydrate_tool_images(&self.store, &mut history).await { return (RunOutcome::Failed(error.into()), usage); } + // The usage anchor counts persisted messages only; a transient + // tail is provider-visible but never committed. + let anchored_messages = history.len(); + let history = if prepared.action == RunAction::Compact { + super::history::user_terminated( + history, + "compaction:instruction", + super::compaction::INSTRUCTIONS, + ) + } else { + super::history::user_terminated(history, CONTINUE_MESSAGE_ID, CONTINUE_INSTRUCTION) + }; let request = crate::model::ModelRequest { prompt: prepared.prompt.clone(), model: prepared.model.clone(), @@ -296,7 +318,7 @@ impl RunEngine { update_context_usage_anchor( &mut context_usage_anchor, cycle_usage, - request.history.len(), + anchored_messages, ); accumulate_usage(&mut usage, cycle_usage); } @@ -306,7 +328,7 @@ impl RunEngine { update_context_usage_anchor( &mut context_usage_anchor, cycle_usage, - request.history.len(), + anchored_messages, ); accumulate_usage(&mut usage, cycle_usage); } @@ -353,7 +375,7 @@ impl RunEngine { update_context_usage_anchor( &mut context_usage_anchor, cycle_usage, - request.history.len(), + anchored_messages, ); accumulate_usage(&mut usage, cycle_usage); } @@ -361,6 +383,58 @@ impl RunEngine { let _ = emit(client, RunEvent::CycleInterrupted).await; return (RunOutcome::Cancelled, usage); } + // The estimate that cleared the compaction check can + // still land over the real limit: it trails the + // provider's own count by whatever the request adds + // after the anchor was taken. The provider is the + // authority, so treat its refusal as the trigger the + // estimate missed. Without this a conversation that + // crosses the line is wedged: every retry rebuilds the + // same prompt and gets the same refusal. + if !overflow_compacted + && prepared.action != RunAction::Compact + && matches!(&cycle_failure.failure, RunFailure::Provider(message) + if super::compaction::is_context_overflow(message)) + { + tracing::warn!( + provider_call_index, + checkpoint_id = checkpoint.0, + "provider rejected the prompt as over-limit; compacting and retrying" + ); + overflow_compacted = true; + checkpoint = match super::messages::append_batches( + &self.store, + prepared, + client, + cancellation, + checkpoint, + std::mem::take(&mut pending_insertions), + ) + .await + { + Ok((checkpoint, _)) => checkpoint, + Err(outcome) => return (outcome, usage), + }; + let messages = + match self.store.load_checkpoint_messages(checkpoint).await { + Ok(messages) => messages, + Err(error) => return (RunOutcome::Failed(error.into()), usage), + }; + match self + .auto_compact(prepared, checkpoint, &messages, client, cancellation) + .await + { + Ok((next_checkpoint, compaction_usage)) => { + checkpoint = next_checkpoint; + context_usage_anchor = None; + if let Some(compaction_usage) = compaction_usage { + accumulate_usage(&mut usage, compaction_usage); + } + continue 'model; + } + Err(outcome) => return (outcome, usage), + } + } if !should_retry(&cycle_failure, retries) { return (RunOutcome::Failed(cycle_failure.failure), usage); } @@ -459,7 +533,7 @@ impl RunEngine { update_context_usage_anchor( &mut context_usage_anchor, cycle_usage, - request.history.len(), + anchored_messages, ); accumulate_usage(&mut usage, cycle_usage); } @@ -721,6 +795,9 @@ impl RunEngine { .await .map_err(|error| RunOutcome::Failed(error.into()))?; let history = crate::model::project_messages(&compactable) + .map(|history| { + super::compaction::compaction_history(history, prepared.model.context_window_tokens) + }) .map_err(|error| RunOutcome::Failed(error.into()))?; let mut model = prepared.model.clone(); model.max_output_tokens = Some(super::compaction::OUTPUT_TOKENS); diff --git a/server/src/run/history.rs b/server/src/run/history.rs new file mode 100644 index 0000000..ab508cc --- /dev/null +++ b/server/src/run/history.rs @@ -0,0 +1,60 @@ +//! Guarantees provider history ends with a user message before dispatch. + +use crate::model::{ContentPart, ProjectedContent, ProjectedMessage, Role}; + +/// Appends a transient user message when the history ends with the assistant. +/// +/// Providers read an assistant-terminated history as a prefill request, and +/// Anthropic refuses it outright: "This model does not support assistant +/// message prefill. The conversation must end with a user message." The +/// appended message is provider-visible only; it is never persisted, so the +/// committed checkpoint stays an exact prefix of the next turn. +pub(super) fn user_terminated( + mut history: Vec, + message_id: &str, + text: &str, +) -> Vec { + if history + .last() + .is_none_or(|message| message.role != Role::Assistant) + { + return history; + } + history.push(ProjectedMessage { + message_id: message_id.into(), + role: Role::User, + content: ProjectedContent::Parts(vec![ContentPart::Text { text: text.into() }]), + }); + history +} + +#[cfg(test)] +mod tests { + use super::*; + + fn message(id: &str, role: Role) -> ProjectedMessage { + ProjectedMessage { + message_id: id.into(), + role, + content: ProjectedContent::Parts(vec![ContentPart::Text { text: id.into() }]), + } + } + + #[test] + fn assistant_terminated_history_gains_a_user_tail() { + let history = vec![message("u1", Role::User), message("a1", Role::Assistant)]; + let terminated = user_terminated(history.clone(), "tail", "continue"); + assert_eq!(terminated[..2], history[..]); + assert_eq!(terminated.last().unwrap().role, Role::User); + assert_eq!(terminated.last().unwrap().message_id, "tail"); + } + + #[test] + fn user_and_tool_terminated_histories_are_untouched() { + let user = vec![message("a1", Role::Assistant), message("u2", Role::User)]; + assert_eq!(user_terminated(user.clone(), "tail", "continue"), user); + let tool = vec![message("a1", Role::Assistant), message("t1", Role::Tool)]; + assert_eq!(user_terminated(tool.clone(), "tail", "continue"), tool); + assert!(user_terminated(Vec::new(), "tail", "continue").is_empty()); + } +} diff --git a/server/src/run/mod.rs b/server/src/run/mod.rs index d9462db..7cf0157 100644 --- a/server/src/run/mod.rs +++ b/server/src/run/mod.rs @@ -5,6 +5,7 @@ mod compaction; mod engine; mod event; mod handle; +mod history; mod messages; mod model_cycle; mod model_retry; diff --git a/server/src/run/model_cycle.rs b/server/src/run/model_cycle.rs index 23824a7..469dc5f 100644 --- a/server/src/run/model_cycle.rs +++ b/server/src/run/model_cycle.rs @@ -373,7 +373,12 @@ fn failure( partial_reasoning: String, usage: Option, ) -> Box { - let retryable = matches!(failure, RunFailure::Protocol(_) | RunFailure::Provider(_)); + let retryable = match &failure { + RunFailure::Protocol(_) => true, + RunFailure::Provider(message) if is_rejected_request(message) => false, + RunFailure::Provider(_) => true, + RunFailure::Store(_) | RunFailure::Client(_) => false, + }; Box::new(ModelCycleFailure { failure, partial_text, @@ -383,6 +388,21 @@ fn failure( }) } +/// A rejected request (wrong key, unknown model, malformed or oversized body) +/// fails identically every time, so retrying it only delays the error the user +/// needs to see. Provider failures are formatted as `