fix(runtime): keep a failed turn's tool results, and size guards from the real context window

Five defects found by auditing the agent stack, all of which let the harness act
confidently on a wrong picture of its own state.

1. A turn that errored discarded the record of tools that had already run. The CLI
   hands run_turn a CLONE of the session; the turn appends the user message, each
   assistant reply, and a tool_result per executed tool to that clone alone. The error
   path dropped it while the side effects of those tools remained on disk, and the
   recovery path then replayed the same user input against a session showing no sign
   they had ever run — inviting a second execution of tools that are not idempotent.
   The clone is now adopted before teardown, and recovery resumes the turn instead of
   replaying it (new ConversationRuntime::resume_turn).

2. Every sizing guard was computed against the model NAME's context window. With a
   local server behind a cloud model alias that is 200k against a much smaller
   backend, which does not make the preflights inaccurate — it makes them unreachable.
   CLAW_CONTEXT_WINDOW now declares the real window; model_token_limit honours it and
   returns a limit for unrecognised models instead of None, max_tokens is capped to a
   quarter of it, and the auto-compaction threshold derives from it (70%) rather than
   defaulting to a 200k-window constant.

3. Whether to recover by discarding conversation history was decided by substring
   matching on rendered error prose, including strings that ordinary transport and
   decode failures also produce — so a dropped connection cost the session its
   transcript. Classification now happens in the API layer, where the error still has
   its type and the request is still in hand: typed overflow errors, plus ambiguous
   transport failures only when the request actually filled the window. The two
   copy-pasted detection blocks are gone.

4. The top-level agent loop had no iteration ceiling (usize::MAX) while spawned
   subagents were bounded at 32. It now gets the same bound, overridable with
   CLAW_MAX_TURN_ITERATIONS.

5. Auto-continue nudges were injected as USER turns, so harness-authored text was
   persisted, re-sent on every later request, and folded into compaction summaries as
   something the user had said. They are recorded as system turns now; the wire
   payload is unchanged.

Also replaces the repository's own self-host section, which described a translator
and a serving stack no longer in the request path. That file is folded into the
system prompt verbatim, so it was briefing the model with an architecture that does
not exist.

(cherry picked from commit c8834cb4d1a3372eb70241429f641162a420297e)

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 {

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::{

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<u32> {
+    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<u32> {
+    value
+        .and_then(|raw| raw.trim().parse::<u32>().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<u
 
 #[must_use]
 pub fn model_token_limit(model: &str) -> Option<ModelTokenLimit> {
+    // 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<ModelTokenLimit> {
     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));

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<String>) -> 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<String>) -> 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<String>,
+        prompter: Option<&mut dyn PermissionPrompter>,
+    ) -> Result<TurnSummary, RuntimeError> {
+        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<TurnSummary, RuntimeError> {
+        self.drive_turn(None, prompter)
+    }
+
+    #[allow(clippy::too_many_lines)]
+    fn drive_turn(
+        &mut self,
+        user_input: Option<String>,
         mut prompter: Option<&mut dyn PermissionPrompter>,
     ) -> Result<TurnSummary, RuntimeError> {
-        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<AutoCompactionEvent> {
         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<u32> {
     value
         .and_then(|raw| raw.trim().parse::<u32>().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<Result<Vec<AssistantEvent>, String>>,
+        calls: usize,
+    }
+
+    impl ScriptedApi {
+        fn new(steps: Vec<Result<Vec<AssistantEvent>, String>>) -> Self {
+            Self { steps, calls: 0 }
+        }
+    }
+
+    impl ApiClient for ScriptedApi {
+        fn stream(&mut self, _request: ApiRequest) -> Result<Vec<AssistantEvent>, 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<AssistantEvent> {
+        vec![
+            AssistantEvent::ToolUse {
+                id: id.to_string(),
+                name: "echo".to_string(),
+                input: "payload".to_string(),
+            },
+            AssistantEvent::MessageStop,
+        ]
+    }
+
+    fn scripted_runtime(
+        steps: Vec<Result<Vec<AssistantEvent>, String>>,
+    ) -> ConversationRuntime<ScriptedApi, StaticToolExecutor> {
+        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;

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.

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<String>) -> Self {
+        Self {
+            role: MessageRole::System,
+            blocks: vec![ContentBlock::Text { text: text.into() }],
+            usage: None,
+        }
+    }
+
     #[must_use]
     pub fn assistant(blocks: Vec<ContentBlock>) -> Self {
         Self {

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::<usize>().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)