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 `