diff --git a/Cargo.lock b/Cargo.lock index 2e4661a..f22764d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1172,7 +1172,7 @@ checksum = "52560adf09603e58c9a7ee1fe1dcb95a16927b17c127f0ac02d6e768a0e25bc1" [[package]] name = "cursor-byok-desktop" -version = "0.1.6" +version = "0.1.7" dependencies = [ "axum", "cursor-server", diff --git a/apps/desktop/package-lock.json b/apps/desktop/package-lock.json index 322e87a..0ce8b9d 100644 --- a/apps/desktop/package-lock.json +++ b/apps/desktop/package-lock.json @@ -1,12 +1,12 @@ { "name": "cursor-byok-desktop", - "version": "0.1.6", + "version": "0.1.7", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "cursor-byok-desktop", - "version": "0.1.6", + "version": "0.1.7", "license": "MIT", "dependencies": { "@floating-ui/dom": "^1.8.0", diff --git a/apps/desktop/package.json b/apps/desktop/package.json index 6538fae..72b7056 100644 --- a/apps/desktop/package.json +++ b/apps/desktop/package.json @@ -1,6 +1,6 @@ { "name": "cursor-byok-desktop", - "version": "0.1.6", + "version": "0.1.7", "description": "Cursor BYOK desktop management application", "type": "module", "scripts": { diff --git a/apps/desktop/src-tauri/Cargo.toml b/apps/desktop/src-tauri/Cargo.toml index 10c01dd..1cd30c3 100644 --- a/apps/desktop/src-tauri/Cargo.toml +++ b/apps/desktop/src-tauri/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "cursor-byok-desktop" -version = "0.1.6" +version = "0.1.7" edition = "2021" publish = false diff --git a/apps/desktop/src-tauri/tauri.conf.json b/apps/desktop/src-tauri/tauri.conf.json index f7b61cb..846b1d3 100644 --- a/apps/desktop/src-tauri/tauri.conf.json +++ b/apps/desktop/src-tauri/tauri.conf.json @@ -1,7 +1,7 @@ { "$schema": "https://schema.tauri.app/config/2", "productName": "Cursor BYOK", - "version": "0.1.6", + "version": "0.1.7", "identifier": "dev.cursorbyok.desktop", "build": { "beforeDevCommand": "npm run dev", diff --git a/apps/desktop/src/shell/AppLayout.tsx b/apps/desktop/src/shell/AppLayout.tsx index 75f78ff..bba2826 100644 --- a/apps/desktop/src/shell/AppLayout.tsx +++ b/apps/desktop/src/shell/AppLayout.tsx @@ -6,6 +6,7 @@ import cursorIconUrl from "../shared/assets/icons/cursor.svg"; import { api } from "../shared/api"; import { AdMenu } from "./ads/AdMenu"; import { FloatingAd } from "./ads/FloatingAd"; +import { loadCachedAds, saveCachedAds } from "./ads/cache"; import { AdActionType, type AdAction, type AdSlot } from "./ads/types"; import { PageLayout } from "./layout/PageLayout"; import { Card } from "../shared/ui/Card"; @@ -50,7 +51,7 @@ export function AppLayout() { const location = useLocation(); const [leftActionTarget, setLeftActionTarget] = useState(null); const [rightActionTarget, setRightActionTarget] = useState(null); - const [ads, setAds] = useState([]); + const [ads, setAds] = useState(() => loadCachedAds()); const [activeAd, setActiveAd] = useState(null); const [dismissCandidate, setDismissCandidate] = useState(null); const [dismissReason, setDismissReason] = useState(""); @@ -98,6 +99,8 @@ export function AppLayout() { let disposed = false; let pending = false; let lastRequestedAt = 0; + setAds(loadCachedAds()); + setActiveAd(null); const refreshAds = () => { const now = Date.now(); if (pending || now - lastRequestedAt < 10_000) return; @@ -105,6 +108,7 @@ export function AppLayout() { lastRequestedAt = now; void api.ads(dismissedAdIdsRef.current, locale) .then((runtime) => { + saveCachedAds(runtime.slots); if (disposed) return; setAds(runtime.slots); setActiveAd((current) => current @@ -112,9 +116,7 @@ export function AppLayout() { : null); }) .catch(() => { - if (disposed) return; - setAds([]); - setActiveAd(null); + // Keep the current ads and local cache available when refreshing fails. }) .finally(() => { pending = false; }); }; diff --git a/apps/desktop/src/shell/ads/cache.ts b/apps/desktop/src/shell/ads/cache.ts new file mode 100644 index 0000000..94828db --- /dev/null +++ b/apps/desktop/src/shell/ads/cache.ts @@ -0,0 +1,21 @@ +import type { AdSlot } from "./types"; + +const adCacheStorageKey = "cursor-byok:ad-runtime"; + +export function loadCachedAds(): AdSlot[] { + try { + const value: unknown = JSON.parse(localStorage.getItem(adCacheStorageKey) ?? "null"); + if (!value || typeof value !== "object" || !Array.isArray((value as { slots?: unknown }).slots)) return []; + return (value as { slots: AdSlot[] }).slots; + } catch { + return []; + } +} + +export function saveCachedAds(slots: AdSlot[]): void { + try { + localStorage.setItem(adCacheStorageKey, JSON.stringify({ slots })); + } catch { + // The current session still uses the fetched ads when storage is unavailable. + } +} diff --git a/server/src/cursor/tools/tool_call_result/gate.rs b/server/src/cursor/tools/tool_call_result/gate.rs index 3c72a4b..b12283a 100644 --- a/server/src/cursor/tools/tool_call_result/gate.rs +++ b/server/src/cursor/tools/tool_call_result/gate.rs @@ -517,11 +517,9 @@ fn gate_mcp_resources(tool: &mut pb::ListMcpResourcesToolCall) { .push(pb::list_mcp_resources_exec_result::McpResource { uri: "truncated:list-mcp-resources".into(), name: Some("truncated".into()), - description: Some(truncation_notice( - "ListMcpResources", - MCP_TEXT_LIMIT, - success.resources.len(), - original, + description: Some(format!( + "[truncated: ListMcpResources result exceeded {MCP_RESOURCE_LIMIT} resources; showing {} of {original} resources]", + success.resources.len() )), ..Default::default() }); @@ -861,4 +859,39 @@ mod tests { let mut content = String::new(); tool_completion("Grep", &mut tool, &mut content); } + + #[test] + fn list_mcp_resources_reports_the_cap_it_actually_applied() { + let resources = (0..MCP_RESOURCE_LIMIT + 50) + .map(|index| pb::list_mcp_resources_exec_result::McpResource { + uri: format!("mcp://resource/{index}"), + ..Default::default() + }) + .collect(); + let mut tool = pb::ListMcpResourcesToolCall { + args: None, + result: Some(pb::ListMcpResourcesExecResult { + result: Some(pb::list_mcp_resources_exec_result::Result::Success( + pb::ListMcpResourcesSuccess { resources }, + )), + }), + }; + + gate_mcp_resources(&mut tool); + + let pb::list_mcp_resources_exec_result::Result::Success(success) = + tool.result.unwrap().result.unwrap() + else { + panic!("expected a successful result"); + }; + let notice = success.resources.last().unwrap(); + assert_eq!(notice.uri, "truncated:list-mcp-resources"); + assert_eq!( + notice.description.as_deref(), + Some( + "[truncated: ListMcpResources result exceeded 200 resources; \ + showing 200 of 250 resources]" + ) + ); + } } diff --git a/server/src/local_app/mod.rs b/server/src/local_app/mod.rs index b569403..a12db94 100644 --- a/server/src/local_app/mod.rs +++ b/server/src/local_app/mod.rs @@ -153,19 +153,7 @@ impl CursorHarness { } pub async fn set_enabled(&self, enabled: bool) -> Result { - let settings_applied = { - let proxy = self.inner.proxy.lock().await; - proxy - .url() - .as_deref() - .map(settings::settings_match) - .transpose()? - .unwrap_or(false) - }; if enabled { - if !settings_applied { - process::terminate_cursor().await?; - } self.inner.store.set_cursor_takeover_enabled(true).await?; self.enable().await?; } else { @@ -193,6 +181,15 @@ impl CursorHarness { .read() .ok_or_else(|| Error::Config("desktop management server is not ready".into()))?; let mut proxy = self.inner.proxy.lock().await; + let settings_applied = proxy + .url() + .as_deref() + .map(settings::settings_match) + .transpose()? + .unwrap_or(false); + if !settings_applied { + process::terminate_cursor().await?; + } if proxy.running() { if let Some(url) = proxy.url() { apply_cursor_configuration(&url).await?; 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 `