diff --git a/rust/crates/api/src/error.rs b/rust/crates/api/src/error.rs index f52810af..d4dc0daa 100644 --- a/rust/crates/api/src/error.rs +++ b/rust/crates/api/src/error.rs @@ -255,6 +255,25 @@ impl ApiError { } } + /// True for failures whose own text cannot say what went wrong: transport, decode, and frame + /// errors. + /// + /// These matter because a server that answers an oversized request with a non-SSE HTTP 500 + /// body lands here rather than in [`Self::is_context_window_failure`] — reqwest reports + /// "error decoding response body" and the real cause is lost. Callers that know how large the + /// request was can combine the two signals; callers that do not must treat this as unknown, + /// never as an overflow. + #[must_use] + pub fn is_ambiguous_transport_failure(&self) -> bool { + match self { + Self::Http(_) | Self::Json { .. } | Self::InvalidSseFrame(_) => true, + Self::RetriesExhausted { last_error, .. } => { + last_error.is_ambiguous_transport_failure() + } + _ => false, + } + } + #[must_use] pub fn is_context_window_failure(&self) -> bool { match self { diff --git a/rust/crates/api/src/lib.rs b/rust/crates/api/src/lib.rs index d27bab28..0a6c3cd5 100644 --- a/rust/crates/api/src/lib.rs +++ b/rust/crates/api/src/lib.rs @@ -27,9 +27,10 @@ pub use providers::openai_compat::{ OpenAiCompatClient, OpenAiCompatConfig, }; pub use providers::{ - detect_provider_kind, max_tokens_for_model, max_tokens_for_model_with_override, - model_family_identity_for, model_family_identity_for_kind, provider_diagnostics_for_model, - resolve_model_alias, ProviderDiagnostics, ProviderKind, + context_window_override, detect_provider_kind, estimate_message_request_input_tokens, + max_tokens_for_model, max_tokens_for_model_with_override, model_family_identity_for, + model_family_identity_for_kind, model_token_limit, provider_diagnostics_for_model, + resolve_model_alias, ModelTokenLimit, ProviderDiagnostics, ProviderKind, }; pub use sse::{parse_frame, SseParser}; pub use types::{ diff --git a/rust/crates/api/src/providers/mod.rs b/rust/crates/api/src/providers/mod.rs index 993a4e7d..9c567a06 100644 --- a/rust/crates/api/src/providers/mod.rs +++ b/rust/crates/api/src/providers/mod.rs @@ -619,16 +619,43 @@ fn web_passthrough_diagnostic( } } +/// Overrides the context window every sizing guard is computed against. +/// +/// The model *name* is not a reliable statement of the window when the endpoint is a local +/// server: the harness sends its compiled-in default name (a Claude alias) while the backend +/// serves whatever `-c` it was started with. Every guard downstream of [`model_token_limit`] +/// then sizes itself against the name's window instead of the real one, which makes the +/// preflights unable to fire rather than merely inaccurate. This variable is the operator's +/// statement of the truth, and it wins over the table. +const CONTEXT_WINDOW_ENV_VAR: &str = "CLAW_CONTEXT_WINDOW"; + +/// Fraction of the window a single response may be allowed to claim when the window is +/// declared. Output and input share one budget on a local server, so reserving a quarter for +/// the reply leaves three quarters for prompt plus transcript. +const MAX_OUTPUT_SHARE_OF_WINDOW: u32 = 4; + +/// The operator-declared context window, when set and parseable as a positive token count. +#[must_use] +pub fn context_window_override() -> Option { + parse_context_window_override(std::env::var(CONTEXT_WINDOW_ENV_VAR).ok().as_deref()) +} + +#[must_use] +fn parse_context_window_override(value: Option<&str>) -> Option { + value + .and_then(|raw| raw.trim().parse::().ok()) + .filter(|window| *window > 0) +} + #[must_use] pub fn max_tokens_for_model(model: &str) -> u32 { - let canonical = resolve_model_alias(model); - let heuristic = if canonical.contains("opus") { - 32_000 - } else { - 64_000 - }; - - model_token_limit(model).map_or(heuristic, |limit| heuristic.min(limit.max_output_tokens)) + // A declared window caps the reply too. Asking a 65k server for 32k of output reserves half + // the window for a single message, which the server then silently clamps — leaving the + // client's own accounting wrong by the difference. + model_token_limit(model).map_or_else( + || max_tokens_for_model_by_name(model), + |limit| max_tokens_for_model_by_name(model).min(limit.max_output_tokens), + ) } /// Returns the effective max output tokens for a model, preferring a plugin @@ -641,6 +668,33 @@ pub fn max_tokens_for_model_with_override(model: &str, plugin_override: Option Option { + // A declared window replaces the table's, and supplies one for models the table has never + // heard of. Returning `Some` for an unknown model is the point: `None` makes both preflights + // return `Ok` without looking at anything, so an unrecognised local model gets no guard at all. + if let Some(context_window_tokens) = context_window_override() { + return Some(ModelTokenLimit { + max_output_tokens: max_tokens_for_model_by_name(model) + .min(context_window_tokens / MAX_OUTPUT_SHARE_OF_WINDOW), + context_window_tokens, + }); + } + model_token_limit_by_name(model) +} + +#[must_use] +fn max_tokens_for_model_by_name(model: &str) -> u32 { + let canonical = resolve_model_alias(model); + let heuristic = if canonical.contains("opus") { + 32_000 + } else { + 64_000 + }; + model_token_limit_by_name(model) + .map_or(heuristic, |limit| heuristic.min(limit.max_output_tokens)) +} + +#[must_use] +fn model_token_limit_by_name(model: &str) -> Option { let canonical = resolve_model_alias(model); let base_model = canonical.rsplit('/').next().unwrap_or(canonical.as_str()); match base_model { @@ -723,7 +777,7 @@ pub fn preflight_message_request(request: &MessageRequest) -> Result<(), ApiErro Ok(()) } -fn estimate_message_request_input_tokens(request: &MessageRequest) -> u32 { +pub fn estimate_message_request_input_tokens(request: &MessageRequest) -> u32 { let mut estimate = estimate_serialized_tokens(&request.messages); estimate = estimate.saturating_add(estimate_serialized_tokens(&request.system)); estimate = estimate.saturating_add(estimate_serialized_tokens(&request.tools)); diff --git a/rust/crates/runtime/src/conversation.rs b/rust/crates/runtime/src/conversation.rs index e79415c6..851149b9 100644 --- a/rust/crates/runtime/src/conversation.rs +++ b/rust/crates/runtime/src/conversation.rs @@ -20,6 +20,15 @@ const AUTO_COMPACTION_THRESHOLD_ENV_VAR: &str = "CLAUDE_CODE_AUTO_COMPACT_INPUT_ const DEFAULT_MAX_STALL_NUDGES: usize = 3; const MAX_STALL_NUDGES_ENV_VAR: &str = "CLAW_MAX_STALL_NUDGES"; +/// Placeholder recorded in the trace when a turn is resumed rather than started, so the trace +/// does not attribute a second user message to a turn that only had one. +const RESUMED_TURN_MARKER: &str = "[resumed turn]"; +/// Operator-declared context window. Kept as a literal rather than imported because `runtime` +/// sits below `api` in the dependency graph; `api::context_window_override` reads the same name. +const CONTEXT_WINDOW_ENV_VAR: &str = "CLAW_CONTEXT_WINDOW"; +/// Share of a declared context window at which compaction is triggered, leaving the remainder +/// as headroom for the reply plus whatever the next tool result carries. +const AUTO_COMPACTION_WINDOW_PERCENT: u32 = 70; /// Injected when the model narrates its next step (or asks for a go-ahead mid-task) /// instead of emitting the tool call that would perform it. @@ -122,6 +131,7 @@ impl std::error::Error for ToolError {} #[derive(Debug, Clone, PartialEq, Eq)] pub struct RuntimeError { message: String, + context_window: bool, } impl RuntimeError { @@ -129,8 +139,28 @@ impl RuntimeError { pub fn new(message: impl Into) -> Self { Self { message: message.into(), + context_window: false, } } + + /// A failure the caller may be able to clear by shrinking the session. + /// + /// This is a decision the API layer makes and hands up, not one the caller reconstructs by + /// matching substrings against the rendered message. Recovering from context exhaustion means + /// discarding conversation history, so a transport hiccup misread as an overflow destroys the + /// session for no reason — the classification has to come from whoever saw the real error. + #[must_use] + pub fn context_window(message: impl Into) -> Self { + Self { + message: message.into(), + context_window: true, + } + } + + #[must_use] + pub const fn is_context_window_failure(&self) -> bool { + self.context_window + } } impl Display for RuntimeError { @@ -367,14 +397,33 @@ where } } - #[allow(clippy::too_many_lines)] pub fn run_turn( &mut self, user_input: impl Into, + prompter: Option<&mut dyn PermissionPrompter>, + ) -> Result { + self.drive_turn(Some(user_input.into()), prompter) + } + + /// Continues the turn already recorded in the session, without adding a user message. + /// + /// Recovery paths retry a turn that failed part-way through. Calling [`Self::run_turn`] again + /// with the original input would append that input a second time and invite the model to redo + /// tool calls whose side effects have already landed. Resuming reads the transcript as it + /// stands — including the tool results that did complete — and carries on from there. + pub fn resume_turn( + &mut self, + prompter: Option<&mut dyn PermissionPrompter>, + ) -> Result { + self.drive_turn(None, prompter) + } + + #[allow(clippy::too_many_lines)] + fn drive_turn( + &mut self, + user_input: Option, mut prompter: Option<&mut dyn PermissionPrompter>, ) -> Result { - let user_input = user_input.into(); - // ROADMAP #38: Session-health canary - probe if context was compacted if self.session.compaction.is_some() { if let Err(error) = self.run_session_health_probe() { @@ -386,10 +435,15 @@ where } } - self.record_turn_started(&user_input); - self.session - .push_user_text(user_input) - .map_err(|error| RuntimeError::new(error.to_string()))?; + match user_input { + Some(user_input) => { + self.record_turn_started(&user_input); + self.session + .push_user_text(user_input) + .map_err(|error| RuntimeError::new(error.to_string()))?; + } + None => self.record_turn_started(RESUMED_TURN_MARKER), + } let mut assistant_messages = Vec::new(); let mut tool_results = Vec::new(); @@ -442,6 +496,7 @@ where match build_assistant_message(events) { Ok(result) => result, Err(error) => { + let error = self.classify_empty_reply(error); self.record_turn_failed(iterations, &error); return Err(error); } @@ -494,8 +549,13 @@ where ); if stalled { stall_nudges += 1; + // Pushed as a system turn, not a user one. The model needs to see it, but the + // transcript must not claim the user asked for it: this text is persisted, + // re-sent on every later request, and folded into the next compaction summary, + // where a user-labelled nudge becomes indistinguishable from something the + // user actually said. self.session - .push_user_text(STALL_NUDGE_PROMPT) + .push_message(ConversationMessage::system_text(STALL_NUDGE_PROMPT)) .map_err(|error| RuntimeError::new(error.to_string()))?; continue; } @@ -662,6 +722,25 @@ where /// This is the guard that actually keeps a session inside the context window: it sees growth /// that usage counters cannot, because it runs after tool results and long generations have /// been appended but before the request that would carry them is built. + /// Decides whether an empty or unreadable reply was an overflow, by size rather than by text. + /// + /// A stream that produces no content says nothing about why. Some backends answer an + /// oversized request that way instead of with a typed error, so the failure is worth + /// recovering from — but only when the session was already large enough for that to be the + /// plausible cause. Below the compaction threshold it is a blip, and discarding the + /// transcript would cost more than the retry could win. + fn classify_empty_reply(&self, error: RuntimeError) -> RuntimeError { + if error.is_context_window_failure() { + return error; + } + if estimate_session_tokens(&self.session) + >= self.auto_compaction_input_tokens_threshold as usize + { + return RuntimeError::context_window(error.to_string()); + } + error + } + fn maybe_auto_compact_before_request(&mut self) -> Option { let estimated = estimate_session_tokens(&self.session); if estimated < self.auto_compaction_input_tokens_threshold as usize { @@ -827,19 +906,40 @@ where /// Reads the automatic compaction threshold from the environment. #[must_use] pub fn auto_compaction_threshold_from_env() -> u32 { - parse_auto_compaction_threshold( + resolve_auto_compaction_threshold( std::env::var(AUTO_COMPACTION_THRESHOLD_ENV_VAR) .ok() .as_deref(), + std::env::var(CONTEXT_WINDOW_ENV_VAR).ok().as_deref(), + ) +} + +/// Resolves the threshold from an explicit setting, else from the declared context window. +/// +/// `DEFAULT_AUTO_COMPACTION_INPUT_TOKENS_THRESHOLD` is a 200k-window number. Against a smaller +/// server it sits above the window entirely, so the pre-request guard can never fire and the +/// first sign of trouble is the backend rejecting the request. Deriving from the declared window +/// keeps the guard meaningful without every caller having to compute a second number by hand. +#[must_use] +fn resolve_auto_compaction_threshold(explicit: Option<&str>, context_window: Option<&str>) -> u32 { + if let Some(threshold) = parse_positive_u32(explicit) { + return threshold; + } + parse_positive_u32(context_window).map_or( + DEFAULT_AUTO_COMPACTION_INPUT_TOKENS_THRESHOLD, + |window| { + (u64::from(window) * u64::from(AUTO_COMPACTION_WINDOW_PERCENT) / 100) + .try_into() + .unwrap_or(DEFAULT_AUTO_COMPACTION_INPUT_TOKENS_THRESHOLD) + }, ) } #[must_use] -fn parse_auto_compaction_threshold(value: Option<&str>) -> u32 { +fn parse_positive_u32(value: Option<&str>) -> Option { value .and_then(|raw| raw.trim().parse::().ok()) - .filter(|threshold| *threshold > 0) - .unwrap_or(DEFAULT_AUTO_COMPACTION_INPUT_TOKENS_THRESHOLD) + .filter(|parsed| *parsed > 0) } /// How many auto-continue nudges a single turn may inject. `0` disables the behaviour. @@ -1019,10 +1119,11 @@ impl ToolExecutor for StaticToolExecutor { #[cfg(test)] mod tests { use super::{ - build_assistant_message, is_stalled_reply, parse_auto_compaction_threshold, - parse_max_stall_nudges, ApiClient, ApiRequest, AssistantEvent, AutoCompactionEvent, - ConversationRuntime, PromptCacheEvent, RuntimeError, StaticToolExecutor, ToolExecutor, - DEFAULT_AUTO_COMPACTION_INPUT_TOKENS_THRESHOLD, DEFAULT_MAX_STALL_NUDGES, + assistant_text, build_assistant_message, is_stalled_reply, parse_max_stall_nudges, + resolve_auto_compaction_threshold, ApiClient, ApiRequest, AssistantEvent, + AutoCompactionEvent, ConversationRuntime, PromptCacheEvent, RuntimeError, + StaticToolExecutor, ToolExecutor, DEFAULT_AUTO_COMPACTION_INPUT_TOKENS_THRESHOLD, + DEFAULT_MAX_STALL_NUDGES, }; use crate::compact::{estimate_session_tokens, CompactionConfig}; use crate::config::{RuntimeFeatureConfig, RuntimeHookConfig}; @@ -2073,16 +2174,41 @@ mod tests { #[test] fn auto_compaction_threshold_defaults_and_parses_values() { assert_eq!( - parse_auto_compaction_threshold(None), + resolve_auto_compaction_threshold(None, None), DEFAULT_AUTO_COMPACTION_INPUT_TOKENS_THRESHOLD ); - assert_eq!(parse_auto_compaction_threshold(Some("4321")), 4321); + assert_eq!(resolve_auto_compaction_threshold(Some("4321"), None), 4321); assert_eq!( - parse_auto_compaction_threshold(Some("0")), + resolve_auto_compaction_threshold(Some("0"), None), DEFAULT_AUTO_COMPACTION_INPUT_TOKENS_THRESHOLD ); assert_eq!( - parse_auto_compaction_threshold(Some("not-a-number")), + resolve_auto_compaction_threshold(Some("not-a-number"), None), + DEFAULT_AUTO_COMPACTION_INPUT_TOKENS_THRESHOLD + ); + } + + #[test] + fn auto_compaction_threshold_derives_from_a_declared_context_window() { + // 70% of 65536, so the guard sits inside the window instead of 34k above it. + assert_eq!( + resolve_auto_compaction_threshold(None, Some("65536")), + 45875 + ); + } + + #[test] + fn an_explicit_threshold_still_wins_over_the_declared_window() { + assert_eq!( + resolve_auto_compaction_threshold(Some("48000"), Some("65536")), + 48000 + ); + } + + #[test] + fn an_unparseable_context_window_falls_back_to_the_default_threshold() { + assert_eq!( + resolve_auto_compaction_threshold(None, Some("plenty")), DEFAULT_AUTO_COMPACTION_INPUT_TOKENS_THRESHOLD ); } @@ -2254,6 +2380,146 @@ mod tests { assert_eq!(error.to_string(), "unknown tool: missing"); } + /// Replays a fixed script of stream outcomes, one per call. + struct ScriptedApi { + steps: Vec, String>>, + calls: usize, + } + + impl ScriptedApi { + fn new(steps: Vec, String>>) -> Self { + Self { steps, calls: 0 } + } + } + + impl ApiClient for ScriptedApi { + fn stream(&mut self, _request: ApiRequest) -> Result, RuntimeError> { + let step = self + .steps + .get(self.calls) + .cloned() + .unwrap_or_else(|| Err("script exhausted".to_string())); + self.calls += 1; + step.map_err(RuntimeError::new) + } + } + + fn tool_call_events(id: &str) -> Vec { + vec![ + AssistantEvent::ToolUse { + id: id.to_string(), + name: "echo".to_string(), + input: "payload".to_string(), + }, + AssistantEvent::MessageStop, + ] + } + + fn scripted_runtime( + steps: Vec, String>>, + ) -> ConversationRuntime { + ConversationRuntime::new( + Session::new(), + ScriptedApi::new(steps), + StaticToolExecutor::new().register("echo", |input| Ok(input.to_string())), + PermissionPolicy::new(PermissionMode::DangerFullAccess), + vec!["system".to_string()], + ) + } + + #[test] + fn a_failed_turn_keeps_the_results_of_tools_that_already_ran() { + // given: a turn whose first iteration runs a tool, and whose second call dies + let mut runtime = scripted_runtime(vec![ + Ok(tool_call_events("tool-1")), + Err("upstream failed".to_string()), + ]); + + // when + runtime + .run_turn("do the thing", None) + .expect_err("the second call fails"); + + // then: the executed tool's result is still in the transcript. The side effect happened; + // a caller that discards this record will let the model run the same tool again. + let tool_results = runtime + .session() + .messages + .iter() + .filter(|message| { + message + .blocks + .iter() + .any(|block| matches!(block, ContentBlock::ToolResult { .. })) + }) + .count(); + assert_eq!(tool_results, 1); + } + + #[test] + fn resume_turn_continues_without_appending_another_user_message() { + // given: a session that already carries a user turn and a completed tool result + let mut runtime = scripted_runtime(vec![ + Ok(tool_call_events("tool-1")), + Err("upstream failed".to_string()), + Ok(vec![ + AssistantEvent::TextDelta("done".to_string()), + AssistantEvent::MessageStop, + ]), + ]); + runtime.run_turn("do the thing", None).expect_err("fails"); + let user_messages_before = runtime + .session() + .messages + .iter() + .filter(|message| message.role == MessageRole::User) + .count(); + + // when + runtime + .resume_turn(None) + .expect("the resumed turn succeeds"); + + // then: the prompt was not repeated + let user_messages_after = runtime + .session() + .messages + .iter() + .filter(|message| message.role == MessageRole::User) + .count(); + assert_eq!(user_messages_before, user_messages_after); + } + + #[test] + fn a_stall_nudge_is_recorded_as_a_system_turn_not_a_user_one() { + // given: a reply that announces work and stops, so the nudge fires once + let mut runtime = nudge_runtime( + vec![ + text_only("Let me check the logs."), + text_only("The logs were clean."), + ], + 1, + ); + runtime + .run_turn("check the logs", None) + .expect("the nudged turn completes"); + + // then: exactly one user message — the real one + let user_messages = runtime + .session() + .messages + .iter() + .filter(|message| message.role == MessageRole::User) + .count(); + assert_eq!(user_messages, 1); + assert!(runtime + .session() + .messages + .iter() + .any(|message| message.role == MessageRole::System + && assistant_text(message).contains("[auto-continue]"))); + } + #[test] fn run_turn_errors_when_max_iterations_is_exceeded() { struct LoopingApi; diff --git a/rust/crates/runtime/src/file_ops.rs b/rust/crates/runtime/src/file_ops.rs index d13b2668..6e034f29 100644 --- a/rust/crates/runtime/src/file_ops.rs +++ b/rust/crates/runtime/src/file_ops.rs @@ -881,15 +881,11 @@ mod tests { use std::path::PathBuf; use std::time::{SystemTime, UNIX_EPOCH}; - use super::{ - component_contains_glob, derive_glob_walk_root, edit_file, expand_braces, glob_search, use super::{ component_contains_glob, derive_glob_walk_root, edit_file, expand_braces, glob_search, grep_search, is_symlink_escape, max_read_chars, normalize_for_comparison, read_file, read_file_in_workspace, write_file, write_file_in_workspace, GrepSearchInput, - DEFAULT_MAX_READ_CHARS, DEFAULT_READ_LINES, MAX_READ_CHARS, MAX_READ_CHARS_ENV_VAR, - MAX_WRITE_SIZE, - }; + DEFAULT_MAX_READ_CHARS, DEFAULT_READ_LINES, MAX_READ_CHARS_ENV_VAR, MAX_WRITE_SIZE, }; /// Serializes the tests that mutate `CLAW_MAX_READ_CHARS`; env vars are process-global. diff --git a/rust/crates/runtime/src/session.rs b/rust/crates/runtime/src/session.rs index 2ecfd97d..61cca8c8 100644 --- a/rust/crates/runtime/src/session.rs +++ b/rust/crates/runtime/src/session.rs @@ -718,6 +718,20 @@ impl ConversationMessage { } } + /// Text the harness injected on its own initiative, not something the user typed. + /// + /// Reaches the model as an ordinary turn — the wire protocol has nowhere else to put it — but + /// stays labelled in the transcript, so a replay, an export, or a compaction summary does not + /// present the harness's own prompting back as the user's words. + #[must_use] + pub fn system_text(text: impl Into) -> Self { + Self { + role: MessageRole::System, + blocks: vec![ContentBlock::Text { text: text.into() }], + usage: None, + } + } + #[must_use] pub fn assistant(blocks: Vec) -> Self { Self { diff --git a/rust/crates/rusty-claude-cli/src/main.rs b/rust/crates/rusty-claude-cli/src/main.rs index 9f0dd044..8a307ccd 100644 --- a/rust/crates/rusty-claude-cli/src/main.rs +++ b/rust/crates/rusty-claude-cli/src/main.rs @@ -74,6 +74,22 @@ use tools::{ const DEFAULT_MODEL: &str = "anthropic/claude-opus-4-7"; +/// Ceiling on model/tool round trips within one turn. +/// +/// `ConversationRuntime` defaults to `usize::MAX`, which for the interactive and `-p` paths means +/// no ceiling at all: a model that keeps re-reading the same file loops until the user kills it. +/// Spawned subagents have always been bounded; this gives the top-level loop the same treatment. +const DEFAULT_MAX_TURN_ITERATIONS: usize = 32; +const MAX_TURN_ITERATIONS_ENV_VAR: &str = "CLAW_MAX_TURN_ITERATIONS"; + +fn max_turn_iterations_from_env() -> usize { + std::env::var(MAX_TURN_ITERATIONS_ENV_VAR) + .ok() + .and_then(|raw| raw.trim().parse::().ok()) + .filter(|iterations| *iterations > 0) + .unwrap_or(DEFAULT_MAX_TURN_ITERATIONS) +} + /// #148: Model provenance for `claw status` JSON/text output. Records where /// the resolved model string came from so claws don't have to re-read argv /// to audit whether their `--model` flag was honored vs falling back to env @@ -7854,6 +7870,15 @@ impl LiveCli { Ok(()) } Err(error) => { + // Salvage the transcript before the failed runtime is torn down. + // + // `prepare_turn_runtime` handed `run_turn` a CLONE of the session, and the turn + // mutated only that clone — appending the user message, each assistant reply, and + // a tool_result for every tool that actually executed. Dropping the clone throws + // all of that away while the side effects of those tools remain on disk, so the + // retry below would re-issue them against a session that shows no sign they ever + // ran. Adopting the clone keeps the record and the reality in step. + *self.runtime.session_mut() = runtime.session().clone(); runtime.shutdown_plugins()?; // Same ownership rule as the success arm: `fail` clears the current // line, so it may only do so when the emitter stopped on a line @@ -7880,37 +7905,17 @@ impl LiveCli { // This eliminates the need for users to manually run /compact when they // hit context limits - the recovery happens automatically. // - // Detection: We look for "context_window" or "Context window" in the error - // message, which covers error types like: - // - "context_window_blocked" - // - "Context window blocked" - // - "This model's maximum context length is X tokens..." + // Detection: the API layer classified this when it still had the typed error and + // the outgoing request in hand — see `AnthropicRuntimeClient::runtime_error_from_api`. + // It is deliberately NOT re-derived from the rendered message here: compaction + // discards conversation history, and the substrings that used to gate it + // ("error decoding response body", "Failed to parse input at pos") are emitted by + // ordinary transport failures as well as by overflow, so a dropped connection cost + // the session its transcript. // ============================================================================ let error_str = error.to_string(); - // Detect context window overflow. Some providers (e.g. OpenAI-compat backends) - // return 400 with "no parseable body" instead of a proper context_length_exceeded - // error when the request is too large to even parse — treat that as context overflow too. - // Also detect model-specific context error markers (e.g. llama.cpp returns - // "Context size has been exceeded." / "exceed_context_size_error" / "exceeds the available context size"). - let is_context_window = error_str.contains("context_window") - || error_str.contains("Context window") - || error_str.contains("no parseable body") - || error_str.contains("exceed_context_size") - || error_str.contains("exceeds the available context size") - || error_str - .to_ascii_lowercase() - .contains("context size has been exceeded"); - - // Also treat "assistant stream produced no content" and reqwest decode failures - // as recoverable errors that may benefit from auto-compaction. Some backends (e.g. - // llama.cpp) return a non-SSE HTTP 500 body when context overflows, causing - // reqwest to fail with "error decoding response body" — treat that as context overflow too. - let is_no_content = error_str.contains("assistant stream produced no content") - || error_str.contains("Failed to parse input at pos") - || error_str.contains("error decoding response body"); - - if is_context_window || is_no_content { + if error.is_context_window_failure() { // If the error tells us the server's actual context window, adapt our // auto-compaction threshold so future auto-compact-trigger checks are accurate. if let Some(window) = extract_context_window_tokens_from_error(&error_str) { @@ -7977,7 +7982,11 @@ impl LiveCli { drop(hook_abort_monitor); let mut rp = CliPermissionPrompter::new(self.permission_mode); - match new_runtime.run_turn(input, Some(&mut rp)) { + // Resume, do not replay. The salvaged session already carries this turn's + // user message and every tool result that completed before the failure; + // re-running `run_turn(input, ..)` would duplicate the prompt and invite a + // second execution of tools that are not idempotent. + match new_runtime.resume_turn(Some(&mut rp)) { Ok(summary) => { self.replace_runtime(new_runtime)?; if !terminal_tail.at_line_start() { @@ -8004,20 +8013,7 @@ impl LiveCli { } Err(retry_error) => { let retry_str = retry_error.to_string(); - let still_context_window = retry_str.contains("context_window") - || retry_str.contains("Context window") - || retry_str.contains("no parseable body") - || retry_str.contains("exceed_context_size") - || retry_str.contains("exceeds the available context size") - || retry_str - .to_ascii_lowercase() - .contains("context size has been exceeded"); - let still_no_content = retry_str - .contains("assistant stream produced no content") - || retry_str.contains("Failed to parse input at pos") - || retry_str.contains("error decoding response body"); - - if (still_context_window || still_no_content) + if retry_error.is_context_window_failure() && round + 1 < max_compact_rounds { // If the retry error reveals the context window, adapt threshold. @@ -12613,7 +12609,8 @@ fn build_runtime_with_plugin_state( policy, system_prompt, &feature_config, - ); + ) + .with_max_iterations(max_turn_iterations_from_env()); if emit_output { runtime = runtime.with_hook_progress_reporter(Box::new(CliHookProgressReporter)); } @@ -12934,7 +12931,49 @@ impl ApiClient for AnthropicRuntimeClient { } } +/// How full a request must be, as a percentage of the declared context window, before an +/// unreadable failure is taken to mean the window overflowed. +const AMBIGUOUS_FAILURE_WINDOW_PERCENT: u64 = 85; + +/// True when a request was close enough to the window that an otherwise unreadable failure is +/// most likely an overflow. +/// +/// This is the piece that lets transport errors be classified without matching on their prose. +/// A backend can answer an oversized request with a non-SSE 500 that reqwest reports as +/// "error decoding response body"; the same string also appears when the connection simply +/// dropped. What separates them is not the text, it is whether the request we just sent had any +/// room left. Requires a declared window — with none, every request looks small and this +/// correctly returns false. +fn request_fills_context_window(request: &MessageRequest) -> bool { + let Some(limit) = api::model_token_limit(&request.model) else { + return false; + }; + let estimated = + api::estimate_message_request_input_tokens(request).saturating_add(request.max_tokens); + u64::from(estimated) * 100 + >= u64::from(limit.context_window_tokens) * AMBIGUOUS_FAILURE_WINDOW_PERCENT +} + impl AnthropicRuntimeClient { + /// Converts an API failure into a runtime error, carrying the overflow verdict with it. + /// + /// The verdict is made here, where the error still has its type and the request is still in + /// hand, rather than reconstructed downstream from the rendered message. + fn runtime_error_from_api( + &self, + error: &api::ApiError, + request: &MessageRequest, + ) -> RuntimeError { + let message = format_user_visible_api_error(&self.session_id, error); + if error.is_context_window_failure() + || (error.is_ambiguous_transport_failure() && request_fills_context_window(request)) + { + RuntimeError::context_window(message) + } else { + RuntimeError::new(message) + } + } + /// Consume a single streaming response, optionally applying a stall /// timeout on the first event for post-tool continuations. #[allow(clippy::too_many_lines)] @@ -12947,9 +12986,7 @@ impl AnthropicRuntimeClient { .client .stream_message(message_request) .await - .map_err(|error| { - RuntimeError::new(format_user_visible_api_error(&self.session_id, &error)) - })?; + .map_err(|error| self.runtime_error_from_api(&error, message_request))?; let mut stdout = io::stdout(); let mut sink = io::sink(); let mut tracked = TailTrackingWriter::new(&mut stdout, self.terminal_tail.clone()); @@ -12971,9 +13008,8 @@ impl AnthropicRuntimeClient { loop { let next = if apply_stall_timeout && !received_any_event { match tokio::time::timeout(POST_TOOL_STALL_TIMEOUT, stream.next_event()).await { - Ok(inner) => inner.map_err(|error| { - RuntimeError::new(format_user_visible_api_error(&self.session_id, &error)) - })?, + Ok(inner) => inner + .map_err(|error| self.runtime_error_from_api(&error, message_request))?, Err(_elapsed) => { return Err(RuntimeError::new( "post-tool stall: model did not respond within timeout", @@ -12981,9 +13017,10 @@ impl AnthropicRuntimeClient { } } } else { - stream.next_event().await.map_err(|error| { - RuntimeError::new(format_user_visible_api_error(&self.session_id, &error)) - })? + stream + .next_event() + .await + .map_err(|error| self.runtime_error_from_api(&error, message_request))? }; let Some(event) = next else { @@ -13123,9 +13160,7 @@ impl AnthropicRuntimeClient { ..message_request.clone() }) .await - .map_err(|error| { - RuntimeError::new(format_user_visible_api_error(&self.session_id, &error)) - })?; + .map_err(|error| self.runtime_error_from_api(&error, message_request))?; let mut events = response_to_events(response, out)?; push_prompt_cache_record(&self.client, &mut events); Ok(events)