merge: integrate restored core admission and session-boundary repairs

Preserve Opus7402a66a6/4991ec6d6/9c9c7b9dd/385e03dde histories. Reviewed provider-error batch suppression, stream content cap, final-report admission, request-scoped nudge receipts, edit rollback, session MCP/usage reset, catalog revalidation, human waits, save-before-apply and selected ASR routing. Inherited lane508/0; fixes-off16/16 targeted failures and4collateral matches passing, each independently mapped by second reviewer. Only five exact control files restored from fixedHEAD after preserving script/log/patch; checkoutclean. New integrated/restored proof is pending queued build. Residual C01-09,C02-11/14/15 and supervisor/spill-write-failure subclaims remain root-owned, not silently closed. No hosted/native/release claim.

Signed-off-by: CodeWhale Bot <[email protected]>

crates/tui/src/commands/groups/config/config.rs

@@ -2348,32 +2348,39 @@ pub fn set_config_value(app: &mut App, key: &str, value: &str, persist: bool) ->
             }
             let next_path = PathBuf::from(expand_tilde(value));
             let path_changed = next_path != app.mcp_config_path;
-            app.mcp_config_path = next_path;
-            if path_changed {
-                app.mcp_reload_required = true;
-            }
             let reload_note = if path_changed {
                 "; run /mcp reload to rebuild the live tool pool"
             } else {
                 ""
             };
-            let message = if persist {
+            // Persist before touching live state (C01-08): a failed save
+            // must leave the session exactly as it was, not report failure
+            // over a path and reload flag that already moved.
+            let saved_to = if persist {
                 match persist_root_string_key(app.config_path.as_deref(), "mcp_config_path", value)
                 {
-                    Ok(path) => format!(
-                        "mcp_config_path = {} (saved to {}){}",
-                        app.mcp_config_path.display(),
-                        path.display(),
-                        reload_note
-                    ),
+                    Ok(path) => Some(path),
                     Err(err) => return CommandResult::error(format!("Failed to save: {err}")),
                 }
             } else {
-                format!(
+                None
+            };
+            app.mcp_config_path = next_path;
+            if path_changed {
+                app.mcp_reload_required = true;
+            }
+            let message = match saved_to {
+                Some(path) => format!(
+                    "mcp_config_path = {} (saved to {}){}",
+                    app.mcp_config_path.display(),
+                    path.display(),
+                    reload_note
+                ),
+                None => format!(
                     "mcp_config_path = {} (session only){}",
                     app.mcp_config_path.display(),
                     reload_note
-                )
+                ),
             };
             return CommandResult::message(message);
         }
@@ -2525,7 +2532,6 @@ pub fn set_config_value(app: &mut App, key: &str, value: &str, persist: bool) ->
             } else {
                 raw
             };
-            app.stream_chunk_timeout_secs = resolved;
             let value_label = stream_chunk_timeout_value_label(raw, resolved);
             if persist {
                 match persist_table_value_key(
@@ -2535,6 +2541,9 @@ pub fn set_config_value(app: &mut App, key: &str, value: &str, persist: bool) ->
                     (raw as i64).into(), // validated above: at most 3600
                 ) {
                     Ok(path) => {
+                        // Live state moves only with the engine action and
+                        // only after the save landed (C01-08).
+                        app.stream_chunk_timeout_secs = resolved;
                         return CommandResult::with_message_and_action(
                             format!(
                                 "stream_chunk_timeout_secs = {value_label} (saved to {}; affects subsequent turns in this session)",
@@ -2546,6 +2555,7 @@ pub fn set_config_value(app: &mut App, key: &str, value: &str, persist: bool) ->
                     Err(err) => return CommandResult::error(format!("Failed to save: {err}")),
                 }
             }
+            app.stream_chunk_timeout_secs = resolved;
             return CommandResult::with_message_and_action(
                 format!(
                     "stream_chunk_timeout_secs = {value_label} (session only; affects subsequent turns in this session)"
@@ -2731,14 +2741,8 @@ pub fn set_config_value(app: &mut App, key: &str, value: &str, persist: bool) ->
                 Ok(value) => value,
                 Err(err) => return CommandResult::error(err),
             };
-            match field {
-                "keep_header" => app.mini_window.keep_header = value,
-                "keep_input" => app.mini_window.keep_input = value,
-                "keep_todo" => app.mini_window.keep_todo = value,
-                "keep_sidebar" => app.mini_window.keep_sidebar = value,
-                "keep_footer" => app.mini_window.keep_footer = value,
-                _ => unreachable!("mini_window field matched above"),
-            }
+            // Persist first: a failed save leaves the live layout as it was
+            // (C01-08).
             if persist
                 && let Err(err) = crate::config_persistence::persist_mini_window_bool_key(
                     app.config_path.as_deref(),
@@ -2748,6 +2752,14 @@ pub fn set_config_value(app: &mut App, key: &str, value: &str, persist: bool) ->
             {
                 return CommandResult::error(format!("Failed to persist: {err}"));
             }
+            match field {
+                "keep_header" => app.mini_window.keep_header = value,
+                "keep_input" => app.mini_window.keep_input = value,
+                "keep_todo" => app.mini_window.keep_todo = value,
+                "keep_sidebar" => app.mini_window.keep_sidebar = value,
+                "keep_footer" => app.mini_window.keep_footer = value,
+                _ => unreachable!("mini_window field matched above"),
+            }
             app.needs_redraw = true;
         }
         "composer_density" | "composer" => {
@@ -2798,15 +2810,17 @@ pub fn set_config_value(app: &mut App, key: &str, value: &str, persist: bool) ->
             app.needs_redraw = true;
         }
         "workspace_follow_symlinks" | "follow_symlinks" => {
+            // Persist first: a failed save leaves the live value as it was
+            // (C01-08).
+            if persist && let Err(e) = persist_single_setting(&key, value) {
+                return CommandResult::error(format!("Failed to save: {e}"));
+            }
             app.workspace_follow_symlinks = settings.workspace_follow_symlinks;
             app.composer.mention_completion_cache = None;
             app.composer.mention_discovery.invalidate();
             app.needs_redraw = true;
             // Engine tools use EngineConfig which is fixed at startup
             return CommandResult::message(if persist {
-                if let Err(e) = persist_single_setting(&key, value) {
-                    return CommandResult::error(format!("Failed to save: {e}"));
-                }
                 format!(
                     "workspace_follow_symlinks = {} (saved; restart required for engine tools)",
                     settings.workspace_follow_symlinks
@@ -2904,7 +2918,18 @@ pub fn set_config_value(app: &mut App, key: &str, value: &str, persist: bool) ->
 
     let message = if persist {
         if let Err(e) = persist_single_setting(&key, value) {
-            return CommandResult::error(format!("Failed to save: {e}"));
+            // C01-08: the projection above already changed live state (and
+            // some arms validate only while projecting, so saving first could
+            // persist a value that never applied). Report exactly that partial
+            // effect, and keep the action so the engine matches the UI instead
+            // of silently diverging from it.
+            return CommandResult {
+                message: Some(format!(
+                    "Error: {key} = {display_value} applies to this session only; saving failed: {e}"
+                )),
+                action,
+                is_error: true,
+            };
         }
         format!("{key} = {display_value} (saved)")
     } else {
@@ -5384,6 +5409,35 @@ context_window = 262144
         );
     }
 
+    #[test]
+    fn failed_save_leaves_live_config_state_untouched() {
+        // C01-08: a `--save` that cannot write must not report failure over
+        // live state that already moved.
+        let temp = tempfile::tempdir().unwrap();
+        let _guard = EnvGuard::new(temp.path());
+        let blocker = temp.path().join("not-a-directory");
+        fs::write(&blocker, "a file where the config directory should be").unwrap();
+        let mut app = create_test_app();
+        app.config_path = Some(blocker.join("config.toml"));
+        let mcp_before = app.mcp_config_path.clone();
+        app.mcp_reload_required = false;
+        let timeout_before = app.stream_chunk_timeout_secs;
+
+        let result = set_config_value(&mut app, "mcp_config_path", "/elsewhere/mcp.json", true);
+        assert!(result.is_error, "{:?}", result.message);
+        assert_eq!(app.mcp_config_path, mcp_before);
+        assert!(!app.mcp_reload_required);
+
+        let next_timeout = if timeout_before == 120 { "121" } else { "120" };
+        let result = set_config_value(&mut app, "stream_chunk_timeout_secs", next_timeout, true);
+        assert!(result.is_error, "{:?}", result.message);
+        assert!(
+            result.action.is_none(),
+            "no engine update for an unsaved value"
+        );
+        assert_eq!(app.stream_chunk_timeout_secs, timeout_before);
+    }
+
     #[test]
     fn config_command_stream_chunk_timeout_save_overrides_canonical_value() {
         let nanos = SystemTime::now()

crates/tui/src/commands/groups/core/voice.rs

@@ -127,9 +127,19 @@ pub async fn capture_and_transcribe(
     if !is_available() {
         return Err(tr(locale, MessageId::VoiceErrNoRecorder).to_string());
     }
-    let api_key = config
-        .active_route_api_key()
-        .map_err(|_| tr(locale, MessageId::VoiceErrNoAuth).to_string())?;
+    // C01-10: the credential check follows the ASR backend that will run,
+    // not the chat provider. Local Whisper and Groq need no provider key, so
+    // resolve it lazily — the same contract as `crate::voice::dictate_once`
+    // — and only refuse before recording when provider ASR is the backend.
+    let provider_key = || {
+        config
+            .active_route_api_key()
+            .map_err(|_| tr(locale, MessageId::VoiceErrNoAuth).to_string())
+    };
+    let (asr_kind, _asr_model) = resolve_asr_choice(config);
+    if asr_kind == "provider" {
+        provider_key()?;
+    }
     let base_url = config.active_route_base_url();
     let openrouter_vendor = config
         .openrouter_vendor()
@@ -143,7 +153,6 @@ pub async fn capture_and_transcribe(
     // Streaming interim: poll every 700ms and show partial transcript like Grok Build's
     // VoiceEvent::Interim → VoiceState::Recording{interim}. We re-transcribe the
     // growing buffer (local-whisper is cheap; Groq is ~300ms; provider falls back).
-    let (asr_kind, _asr_model) = resolve_asr_choice(config);
     let interim_enabled = true; // always show partials — feels alive like Spark
 
     // Spawn recorder on blocking thread with a shared buffer for interim polling.
@@ -241,13 +250,30 @@ pub async fn capture_and_transcribe(
     let text = match asr_kind.as_str() {
         "local-whisper" => match transcribe_local_whisper(&samples).await {
             Ok(v) => Ok(v),
-            Err(_) => transcribe(&api_key, &base_url, &samples, openrouter_vendor.as_deref()).await,
+            Err(_) => {
+                transcribe(
+                    &provider_key()?,
+                    &base_url,
+                    &samples,
+                    openrouter_vendor.as_deref(),
+                )
+                .await
+            }
         },
         "groq" => match transcribe_groq(&samples).await {
             Ok(v) => Ok(v),
-            Err(_) => transcribe(&api_key, &base_url, &samples, openrouter_vendor.as_deref()).await,
+            Err(_) => {
+                transcribe(
+                    &provider_key()?,
+                    &base_url,
+                    &samples,
+                    openrouter_vendor.as_deref(),
+                )
+                .await
+            }
         },
         _ => {
+            let api_key = provider_key()?;
             if app.voice_control_enabled {
                 process_voice_control(
                     &api_key,

crates/tui/src/core/engine.rs

@@ -917,6 +917,10 @@ pub struct Engine {
     /// cleanup is conditional on this exact value so an older pass can never
     /// clear a newer receiver.
     mcp_boot_generation: Option<u64>,
+    /// Abort handle for the boot pass that owns `mcp_boot_generation`. A
+    /// session or workspace boundary drops the pool that pass is dialing
+    /// into, so it aborts the pass instead of letting it run on (C02-08).
+    mcp_boot_task: Option<tokio::task::AbortHandle>,
     /// Monotonic generation for engine-authored MCP session snapshots. Boot
     /// task updates retain their spawn generation so later passes can reject
     /// only genuinely stale work.
@@ -2062,6 +2066,7 @@ impl Engine {
             mcp_supervisor_rx: None,
             mcp_boot_done: None,
             mcp_boot_generation: None,
+            mcp_boot_task: None,
             mcp_event_generation: 0,
             plugin_registry,
             extension_host,
@@ -2789,7 +2794,12 @@ impl Engine {
         // Runtime-added MCP servers are conversation capabilities even when
         // both conversations use the same workspace. Configured servers can
         // reconnect lazily after the new session is installed.
-        self.mcp_pool = None;
+        self.drop_mcp_pool();
+        // C02-17: cumulative usage belongs to the conversation that spent
+        // it. The sync carries no prior total for the installed session, so
+        // it starts from zero instead of inheriting the previous session's
+        // tokens into its snapshot and saved `total_tokens`.
+        self.session.total_usage = Default::default();
         self.session.id = next_session_id;
         Some(previous_session_id)
     }
@@ -3645,7 +3655,7 @@ impl Engine {
                             self.config.plugin_registry = Some(Arc::clone(&self.plugin_registry));
                             // A pool may contain plugin servers and authority
                             // receipts from the previous workspace snapshot.
-                            self.mcp_pool = None;
+                            self.drop_mcp_pool();
                             if let Some(attachment) = &self.extension_host {
                                 attachment.set_plugins(Arc::clone(&self.plugin_registry));
                                 attachment.sync_in_background();
@@ -3841,39 +3851,65 @@ impl Engine {
                                 continue;
                             }
                         };
+                        // C02-02: stage the cut. The removed exchange is kept
+                        // until the replacement turn actually starts; a send
+                        // that never starts puts it back (below).
+                        let removed_exchange = self
+                            .session
+                            .messages
+                            .get(idx..)
+                            .map(<[_]>::to_vec)
+                            .unwrap_or_default();
                         self.session.messages.truncate_to(idx);
                         self.session.bump_messages_revision();
                         // Now dispatch the new message as a normal send,
                         // reusing the engine's stored mode/model config.
                         let mode = self.current_mode;
-                        self.handle_send_message(TurnSpec {
-                            content: new_message.clone(),
-                            mode,
-                            route: Box::new(route),
-                            compaction: Box::new(self.config.compaction.clone()),
-                            initial_routed_usage: Box::new(
-                                crate::cost_status::RuntimeUsageBatch::default(),
-                            ),
-                            goal_objective: self.config.goal_objective.clone(),
-                            goal_token_budget: self.config.goal_token_budget,
-                            goal_status: self.config.goal_status,
-                            reasoning_effort: self.session.reasoning_effort.clone(),
-                            reasoning_effort_auto: self.session.reasoning_effort_auto,
-                            auto_model: self.session.auto_model,
-                            allow_shell: self.session.allow_shell,
-                            trust_mode: self.session.trust_mode,
-                            auto_approve: self.session.auto_approve,
-                            approval_mode: self.session.approval_mode,
-                            translation_enabled: self.config.translation_enabled,
-                            allowed_tools: self.config.allowed_tools.clone(),
-                            dynamic_tools: Vec::new(),
-                            hook_executor: self.config.hook_executor.clone(),
-                            verbosity: self.config.verbosity.clone(),
-                            provenance: UserInputProvenance::ExternalUser,
-                            images: Vec::new(),
-                            max_output_tokens: None,
-                        })
-                        .await;
+                        let outcome = self
+                            .handle_send_message(TurnSpec {
+                                content: new_message.clone(),
+                                mode,
+                                route: Box::new(route),
+                                compaction: Box::new(self.config.compaction.clone()),
+                                initial_routed_usage: Box::new(
+                                    crate::cost_status::RuntimeUsageBatch::default(),
+                                ),
+                                goal_objective: self.config.goal_objective.clone(),
+                                goal_token_budget: self.config.goal_token_budget,
+                                goal_status: self.config.goal_status,
+                                reasoning_effort: self.session.reasoning_effort.clone(),
+                                reasoning_effort_auto: self.session.reasoning_effort_auto,
+                                auto_model: self.session.auto_model,
+                                allow_shell: self.session.allow_shell,
+                                trust_mode: self.session.trust_mode,
+                                auto_approve: self.session.auto_approve,
+                                approval_mode: self.session.approval_mode,
+                                translation_enabled: self.config.translation_enabled,
+                                allowed_tools: self.config.allowed_tools.clone(),
+                                dynamic_tools: Vec::new(),
+                                hook_executor: self.config.hook_executor.clone(),
+                                verbosity: self.config.verbosity.clone(),
+                                provenance: UserInputProvenance::ExternalUser,
+                                images: Vec::new(),
+                                max_output_tokens: None,
+                            })
+                            .await;
+                        if matches!(outcome, SendMessageOutcome::NotStarted { .. }) {
+                            // Anything the failed send appended after the cut
+                            // (a drained shell-completion notice) stays, after
+                            // the restored exchange.
+                            let appended = self
+                                .session
+                                .messages
+                                .get(idx..)
+                                .map(<[_]>::to_vec)
+                                .unwrap_or_default();
+                            self.session.messages.truncate_to(idx);
+                            self.session.messages.push_batch(removed_exchange);
+                            self.session.messages.push_batch(appended);
+                            self.session.bump_messages_revision();
+                            self.emit_session_updated().await;
+                        }
                     }
                     Op::SetAdvisorEnabled { enabled } => {
                         self.config.advisor_config.enabled = enabled;
@@ -6711,29 +6747,42 @@ impl Engine {
             self.ensure_mcp_supervisor();
             return Ok(pool);
         }
+        // C02-18: misconfiguration fails loud. A missing file is an ordinary
+        // empty config (`load_config` returns the default); an unreadable or
+        // malformed one still yields an empty, source-aware pool so a fixed
+        // file and `/mcp reload` recover in-process — but the person is told
+        // that the configured servers are gone, not left to find no tools.
+        let mut load_failure = None;
         let mut pool = McpPool::from_config_path_with_workspace_and_plugins(
             &self.session.mcp_config_path,
             &self.session.workspace,
             Arc::clone(&self.plugin_registry),
         )
         .unwrap_or_else(|e| {
-            tracing::debug!(
-                "MCP config unavailable: {}",
-                crate::mcp::format_mcp_error_for_display(&e)
-            );
+            let reason = crate::mcp::format_mcp_error_for_display(&e);
+            tracing::warn!("MCP config unavailable: {reason}");
+            load_failure = Some(reason);
             McpPool::empty_with_workspace_config_sources(
                 &self.session.mcp_config_path,
                 &self.session.workspace,
                 Arc::clone(&self.plugin_registry),
             )
             .unwrap_or_else(|fallback_error| {
-                tracing::debug!(
+                tracing::warn!(
                     "MCP reload source setup failed: {}",
                     crate::mcp::format_mcp_error_for_display(&fallback_error)
                 );
                 McpPool::new(McpConfig::default())
             })
         });
+        if let Some(reason) = load_failure {
+            let _ = self
+                .tx_event
+                .send(Event::status(format!(
+                    "MCP config could not be loaded, so none of its servers are available: {reason}. Fix it, then run /mcp reload."
+                )))
+                .await;
+        }
         pool = pool.with_disallowed_tools(self.config.disallowed_tools.clone().unwrap_or_default());
         if let Some(decider) = self.config.network_policy.as_ref() {
             pool = pool.with_network_policy(decider.clone());
@@ -6886,12 +6935,30 @@ impl Engine {
             return false;
         }
         self.mcp_boot_generation = None;
+        self.mcp_boot_task = None;
         self.mcp_boot_in_flight = false;
         self.mcp_boot_rx = None;
         self.mcp_boot_done = None;
         true
     }
 
+    /// Drop the engine-owned MCP pool at a session or workspace boundary.
+    /// The in-flight boot pass dials into the dropped pool: abort it and
+    /// clear its generation, receiver and diagnoses, so the next pool starts
+    /// clean instead of waiting on — or adopting progress and connection
+    /// errors from — the previous conversation's pass (C02-08).
+    fn drop_mcp_pool(&mut self) {
+        self.mcp_pool = None;
+        if let Some(task) = self.mcp_boot_task.take() {
+            task.abort();
+        }
+        self.mcp_boot_generation = None;
+        self.mcp_boot_in_flight = false;
+        self.mcp_boot_rx = None;
+        self.mcp_boot_done = None;
+        self.mcp_connection_errors.clear();
+    }
+
     async fn emit_mcp_session_boot(&self, generation: u64, finished: bool) {
         let Ok(snapshot) = self.mcp_session_snapshot().await else {
             return;
@@ -7337,7 +7404,7 @@ impl Engine {
         self.emit_mcp_session_boot(generation, false).await;
 
         let pool_for_task = Arc::clone(&pool);
-        spawn_supervised(
+        let boot_task = spawn_supervised(
             "mcp-session-boot",
             std::panic::Location::caller(),
             async move {
@@ -7419,6 +7486,7 @@ impl Engine {
                 let _ = done_tx.send(true);
             },
         );
+        self.mcp_boot_task = Some(boot_task.abort_handle());
 
         Ok(generation)
     }

crates/tui/src/core/engine/approval.rs

@@ -287,14 +287,36 @@ impl Engine {
         tool_id: &str,
         request: UserInputRequest,
     ) -> Result<UserInputResponse, ToolError> {
-        let _ = self
+        // C02-19: a question that never reached a host has nobody to answer
+        // it. Fail now instead of waiting out the timeout — which by default
+        // is no timeout at all.
+        if self
             .tx_event
             .send(Event::UserInputRequired {
                 id: tool_id.to_string(),
                 request,
             })
-            .await;
+            .await
+            .is_err()
+        {
+            return Err(ToolError::execution_failed(
+                "User input request could not reach its host, so nobody was asked. \
+                 Continue without the answer or ask in your reply instead."
+                    .to_string(),
+            ));
+        }
+        // R1, as for tool approval: the per-turn wall-clock budget bounds the
+        // agent's own time, not how long a person takes to answer.
+        self.turn_wall_clock.begin_human_wait();
+        let response = self.await_user_input_decision(tool_id).await;
+        self.turn_wall_clock.end_human_wait();
+        response
+    }
 
+    async fn await_user_input_decision(
+        &mut self,
+        tool_id: &str,
+    ) -> Result<UserInputResponse, ToolError> {
         // #6003: `[tools] user_input_timeout_seconds`. Absent, or an explicit
         // 0, waits until the person answers or cancels. A positive value is
         // one absolute deadline for the whole wait: `select!` drops the

crates/tui/src/core/engine/streaming.rs

@@ -38,7 +38,7 @@ impl ToolUseState {
     }
 }
 
-/// Maximum total bytes of text/thinking content before aborting the stream.
+/// Maximum total bytes of text, reasoning and tool-argument content before aborting the stream.
 pub(super) const STREAM_MAX_CONTENT_BYTES: usize = 10 * 1024 * 1024; // 10 MB
 /// Sanity backstop for total stream wall-clock duration. **Not** a routine
 /// kill switch — the stream chunk idle timeout is the primary stall

crates/tui/src/core/engine/tests.rs

@@ -25109,7 +25109,7 @@ async fn reasoning_only_length_stop_fails_without_retry() {
 /// each carry exactly one extra message over the same baseline.
 #[tokio::test]
 async fn the_reasoning_only_nudge_rides_one_request_and_never_joins_the_session() {
-    let (model, _events) = run_reasoning_only_turn_with_reprompts(usize::MAX, "stop", 3).await;
+    let (model, events) = run_reasoning_only_turn_with_reprompts(usize::MAX, "stop", 3).await;
 
     let requests = model.requests.lock().expect("captured requests").clone();
     assert_eq!(requests.len(), 4, "one initial request plus three retries");
@@ -25144,6 +25144,34 @@ async fn the_reasoning_only_nudge_rides_one_request_and_never_joins_the_session(
         carries_nudge(&requests[2]),
         "nudge present once retrying again"
     );
+
+    // C02-04: model-visible means logged. Every request the nudge rides
+    // leaves a durable (internal) receipt carrying its exact text.
+    let receipts: Vec<&str> = events
+        .iter()
+        .filter_map(|event| match event {
+            Event::Status { message }
+                if message.starts_with(super::turn_loop::REQUEST_NUDGE_RECEIPT_PREFIX) =>
+            {
+                Some(message.as_str())
+            }
+            _ => None,
+        })
+        .collect();
+    assert_eq!(
+        receipts.len(),
+        requests
+            .iter()
+            .filter(|request| carries_nudge(request))
+            .count(),
+        "one receipt per nudged request"
+    );
+    assert!(receipts.iter().all(|receipt| receipt.ends_with(nudge)));
+    assert_eq!(
+        crate::core::events::status_visibility(receipts[0]),
+        crate::core::events::StatusVisibility::Internal,
+        "durable clients keep the receipt, collapsed"
+    );
 }
 
 /// A model that only ever returns reasoning is bounded: it retries up to the
@@ -27443,6 +27471,8 @@ fn engine_adopts_host_owned_session_id_from_config() {
     );
 }
 
+mod admission_gates;
+mod runtime_state;
 mod sse_turn_recovery;
 mod tool_cancellation;
 

crates/tui/src/core/engine/tests/admission_gates.rs

@@ -0,0 +1,382 @@
+//! Admission gates between a provider response and tool execution: a failed
+//! stream (C02-05), the step-budget final report (C02-10), the stream content
+//! cap over tool arguments (C02-13), and bounded tool errors (C02-12). Every
+//! fixture counts handler entries; none runs a shell or reaches a provider.
+
+use super::*;
+use crate::llm_client::mock::{MockLlmClient, canned};
+use crate::tools::spec::{ApprovalRequirement, ToolCapability, ToolSpec};
+use std::sync::atomic::{AtomicUsize, Ordering};
+
+const WRITE_TOOL: &str = "fixture_write";
+
+/// A mutating tool that needs no approval, so a call that is admitted runs.
+struct CountingWriteTool(Arc<AtomicUsize>);
+
+#[async_trait::async_trait]
+impl ToolSpec for CountingWriteTool {
+    fn name(&self) -> &str {
+        WRITE_TOOL
+    }
+    fn description(&self) -> &str {
+        "Count every execution of a mutating fixture."
+    }
+    fn input_schema(&self) -> Value {
+        json!({"type": "object"})
+    }
+    fn capabilities(&self) -> Vec<ToolCapability> {
+        vec![ToolCapability::WritesFiles]
+    }
+    fn approval_requirement(&self) -> ApprovalRequirement {
+        ApprovalRequirement::Auto
+    }
+    async fn execute(&self, _: Value, _: &ToolContext) -> Result<ToolResult, ToolError> {
+        self.0.fetch_add(1, Ordering::SeqCst);
+        Ok(ToolResult::success("wrote"))
+    }
+}
+
+fn write_surface(
+    engine: &Engine,
+    workspace: &Path,
+    executions: Arc<AtomicUsize>,
+) -> ToolSurfacePolicy {
+    let mut registry = crate::tools::ToolRegistry::new(ToolContext::new(workspace));
+    registry.register(Arc::new(CountingWriteTool(executions)));
+    let tools = Some(registry.to_api_tools_with_cache(true));
+    test_tool_surface(engine, registry, tools, AppMode::Agent)
+}
+
+/// Streams a scripted response per call, `Err` items included, which the
+/// queue-driven mock cannot express.
+struct ScriptedStreamClient {
+    calls: AtomicUsize,
+    script: fn(usize) -> Vec<anyhow::Result<StreamEvent>>,
+}
+
+#[async_trait::async_trait]
+impl crate::core::model_client::ModelClient for ScriptedStreamClient {
+    fn provider_name(&self) -> &str {
+        "scripted"
+    }
+    fn model(&self) -> &str {
+        "scripted-model"
+    }
+    async fn create_message(
+        &self,
+        _request: codewhale_models::MessageRequest,
+    ) -> anyhow::Result<codewhale_models::MessageResponse> {
+        anyhow::bail!("scripted fixture only streams")
+    }
+    async fn create_message_stream(
+        &self,
+        _request: codewhale_models::MessageRequest,
+    ) -> anyhow::Result<crate::llm_client::StreamEventBox> {
+        let call = self.calls.fetch_add(1, Ordering::SeqCst);
+        Ok(Box::pin(futures_util::stream::iter((self.script)(call))))
+    }
+    async fn health_check(&self) -> anyhow::Result<bool> {
+        Ok(true)
+    }
+}
+
+async fn drain_events(handle: &EngineHandle) -> Vec<Event> {
+    let mut rx = handle.rx_event.write().await;
+    std::iter::from_fn(|| rx.try_recv().ok()).collect()
+}
+
+fn not_started(result: &Result<ToolResult, ToolError>) -> bool {
+    result.as_ref().is_ok_and(|output| {
+        !output.success
+            && output
+                .metadata
+                .as_ref()
+                .is_some_and(|metadata| metadata["side_effect_status"] == "not_started")
+    })
+}
+
+#[tokio::test]
+async fn terminal_stream_error_after_a_complete_tool_call_never_reaches_the_handler() {
+    // C02-05: a complete mutating call followed by a provider error frame.
+    let workspace = tempdir().unwrap();
+    let mock = Arc::new(MockLlmClient::new(vec![
+        vec![
+            canned::message_start("tool-then-error"),
+            canned::tool_use_block_start(0, "call-write", WRITE_TOOL),
+            canned::tool_input_delta(0, r#"{"path":"target.txt"}"#),
+            canned::block_stop(0),
+            StreamEvent::Error {
+                error: json!({ "message": "Model not exist." }),
+            },
+        ],
+        canned::simple_text_turn("this second request must never be issued"),
+    ]));
+    let (mut engine, handle) = Engine::new_with_model_client(
+        deterministic_engine_config(workspace.path()),
+        &Config::default(),
+        mock.clone(),
+    );
+    engine.session.auto_approve = true;
+    let executions = Arc::new(AtomicUsize::new(0));
+    let surface = write_surface(&engine, workspace.path(), executions.clone());
+    let mut turn = crate::core::turn::TurnContext::new(4);
+
+    let (status, error) = engine.run_turn(&mut turn, surface, None, None).await;
+
+    assert_eq!(
+        executions.load(Ordering::SeqCst),
+        0,
+        "the handler must never run"
+    );
+    assert_eq!(status, TurnOutcomeStatus::Failed);
+    assert!(
+        error
+            .as_deref()
+            .is_some_and(|e| e.contains("Model not exist.")),
+        "{error:?}"
+    );
+    assert_eq!(
+        mock.call_count(),
+        1,
+        "a failed response authorizes no further request"
+    );
+    assert_eq!(
+        turn.stop_diagnostics.last_response_tool_calls_suppressed,
+        Some(1)
+    );
+    let completions: Vec<_> = drain_events(&handle)
+        .await
+        .into_iter()
+        .filter_map(|event| match event {
+            Event::ToolCallComplete { result, .. } => Some(result),
+            _ => None,
+        })
+        .collect();
+    assert_eq!(completions.len(), 1, "the call is settled exactly once");
+    assert!(not_started(&completions[0]), "{:?}", completions[0]);
+    assert!(
+        !engine
+            .session
+            .messages
+            .iter()
+            .flat_map(|message| &message.content)
+            .any(|block| matches!(
+                block,
+                ContentBlock::ToolUse { .. } | ContentBlock::ToolResult { .. }
+            )),
+        "no unpaired or fabricated tool history"
+    );
+}
+
+#[tokio::test]
+async fn exhausted_stream_resume_never_executes_the_failed_batch() {
+    // C02-05, retry path: every attempt streams a complete call and then
+    // drops. Each resume discards its batch; the last, unretried attempt is
+    // refused at admission instead of executing.
+    fn tool_then_drop(_call: usize) -> Vec<anyhow::Result<StreamEvent>> {
+        vec![
+            Ok(canned::message_start("tool-then-drop")),
+            Ok(canned::tool_use_block_start(0, "call-write", WRITE_TOOL)),
+            Ok(canned::tool_input_delta(0, "{}")),
+            Ok(canned::block_stop(0)),
+            Err(anyhow::anyhow!(
+                "Stream read error: error decoding response body"
+            )),
+        ]
+    }
+    let workspace = tempdir().unwrap();
+    let client = Arc::new(ScriptedStreamClient {
+        calls: AtomicUsize::new(0),
+        script: tool_then_drop,
+    });
+    let (mut engine, _handle) = Engine::new_with_model_client(
+        EngineConfig {
+            terminal_chrome_enabled: false,
+            ..deterministic_engine_config(workspace.path())
+        },
+        &Config::default(),
+        client.clone(),
+    );
+    engine.session.auto_approve = true;
+    let resumes = engine.config.stream_retry_limits.max_resumes as usize;
+    let executions = Arc::new(AtomicUsize::new(0));
+    let surface = write_surface(&engine, workspace.path(), executions.clone());
+    let mut turn = crate::core::turn::TurnContext::new(4);
+
+    let (status, error) = engine.run_turn(&mut turn, surface, None, None).await;
+
+    assert_eq!(
+        executions.load(Ordering::SeqCst),
+        0,
+        "no attempt's batch may run"
+    );
+    assert_eq!(client.calls.load(Ordering::SeqCst), 1 + resumes);
+    assert_eq!(status, TurnOutcomeStatus::Failed);
+    assert!(
+        error
+            .as_deref()
+            .is_some_and(|e| e.contains("error decoding response body")),
+        "{error:?}"
+    );
+}
+
+#[tokio::test]
+async fn tool_argument_bytes_count_toward_the_stream_content_cap() {
+    // C02-13: a response made only of tool-argument JSON must still hit the
+    // per-step content cap, and the capped call must not run.
+    let workspace = tempdir().unwrap();
+    let huge_args = format!(r#"{{"blob":"{}"}}"#, "x".repeat(4096));
+    let mock = Arc::new(MockLlmClient::new(vec![
+        canned::tool_call_turn("call-big", WRITE_TOOL, &huge_args),
+        canned::simple_text_turn("this second request must never be issued"),
+    ]));
+    let (mut engine, _handle) = Engine::new_with_model_client(
+        EngineConfig {
+            stream_max_content_bytes: 64,
+            ..deterministic_engine_config(workspace.path())
+        },
+        &Config::default(),
+        mock.clone(),
+    );
+    engine.session.auto_approve = true;
+    let executions = Arc::new(AtomicUsize::new(0));
+    let surface = write_surface(&engine, workspace.path(), executions.clone());
+    let mut turn = crate::core::turn::TurnContext::new(4);
+
+    let (status, _error) = engine.run_turn(&mut turn, surface, None, None).await;
+
+    assert_eq!(executions.load(Ordering::SeqCst), 0);
+    assert_eq!(status, TurnOutcomeStatus::Failed);
+    assert_eq!(mock.call_count(), 1);
+}
+
+#[tokio::test]
+async fn step_budget_final_report_is_report_only() {
+    // C02-10: the one response granted after the step budget is spent asks
+    // for no tools, and a call it returns anyway is refused, not executed.
+    let workspace = tempdir().unwrap();
+    let mock = Arc::new(MockLlmClient::new(vec![
+        canned::tool_call_turn("call-step-1", WRITE_TOOL, "{}"),
+        canned::tool_call_turn("call-final-report", WRITE_TOOL, "{}"),
+    ]));
+    let (mut engine, handle) = Engine::new_with_model_client(
+        deterministic_engine_config(workspace.path()),
+        &Config::default(),
+        mock.clone(),
+    );
+    engine.session.auto_approve = true;
+    let executions = Arc::new(AtomicUsize::new(0));
+    let surface = write_surface(&engine, workspace.path(), executions.clone());
+    let mut turn = crate::core::turn::TurnContext::new(1);
+
+    let (status, error) = engine.run_turn(&mut turn, surface, None, None).await;
+
+    assert_eq!(
+        executions.load(Ordering::SeqCst),
+        1,
+        "only the in-budget step runs"
+    );
+    assert_eq!(status, TurnOutcomeStatus::Failed);
+    assert!(
+        error
+            .as_deref()
+            .is_some_and(|e| e.contains("Maximum model steps")),
+        "{error:?}"
+    );
+    let requests = mock.captured_requests();
+    assert_eq!(requests.len(), 2);
+    assert_ne!(requests[0].tool_choice, Some(json!("none")));
+    assert_eq!(requests[1].tool_choice, Some(json!("none")));
+    let refused = drain_events(&handle).await.into_iter().any(|event| {
+        matches!(
+            event,
+            Event::ToolCallComplete {
+                model_call: Some(ref call),
+                result: Err(ToolError::PermissionDenied { ref message }),
+                ..
+            } if call.provider_id == "call-final-report" && message.contains("final report")
+        )
+    });
+    assert!(refused, "the final-report call is refused with the reason");
+}
+
+#[test]
+fn oversized_tool_errors_are_bounded_before_fanout() {
+    // C02-12: an error is published to the event stream and the session like
+    // a result, so an oversized one gets the same bounded projection.
+    struct HugeErrorTool {
+        as_result: bool,
+    }
+    #[async_trait::async_trait]
+    impl ToolSpec for HugeErrorTool {
+        fn name(&self) -> &str {
+            "fixture_huge_error"
+        }
+        fn description(&self) -> &str {
+            "Fail with an oversized message."
+        }
+        fn input_schema(&self) -> Value {
+            json!({"type": "object"})
+        }
+        fn capabilities(&self) -> Vec<ToolCapability> {
+            vec![ToolCapability::ReadOnly]
+        }
+        async fn execute(&self, _: Value, _: &ToolContext) -> Result<ToolResult, ToolError> {
+            let message = format!("FIRST{}LAST", "e".repeat(300_000));
+            if self.as_result {
+                Ok(ToolResult::error(message))
+            } else {
+                Err(ToolError::execution_failed(message))
+            }
+        }
+    }
+
+    with_artifact_home(|home| {
+        tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .unwrap()
+            .block_on(async {
+                for as_result in [false, true] {
+                    let mock = Arc::new(MockLlmClient::new(vec![
+                        canned::tool_call_turn("call-huge", "fixture_huge_error", "{}"),
+                        canned::simple_text_turn("done"),
+                    ]));
+                    let (mut engine, handle) = Engine::new_with_model_client(
+                        deterministic_engine_config(home),
+                        &Config::default(),
+                        mock.clone(),
+                    );
+                    let mut registry = crate::tools::ToolRegistry::new(ToolContext::new(home));
+                    registry.register(Arc::new(HugeErrorTool { as_result }));
+                    let tools = Some(registry.to_api_tools_with_cache(true));
+                    let surface = test_tool_surface(&engine, registry, tools, AppMode::Agent);
+                    let mut turn = crate::core::turn::TurnContext::new(4);
+                    let (status, error) = engine.run_turn(&mut turn, surface, None, None).await;
+                    assert_eq!(status, TurnOutcomeStatus::Completed, "{error:?}");
+                    let published = drain_events(&handle)
+                        .await
+                        .into_iter()
+                        .find_map(|event| match event {
+                            Event::ToolCallComplete { result, .. } => Some(match result {
+                                Ok(output) => output.content,
+                                Err(error) => error.to_string(),
+                            }),
+                            _ => None,
+                        })
+                        .expect("the call completes");
+                    assert!(
+                        published.len() < crate::tools::truncate::SPILLOVER_THRESHOLD_BYTES,
+                        "as_result={as_result}: {} bytes published",
+                        published.len()
+                    );
+                    assert!(published.contains("FIRST"), "the head survives");
+                    assert!(published.contains("LAST"), "the tail survives");
+                    assert!(
+                        published.contains(crate::tools::truncate::SPILLOVER_RECOVERY_HINT),
+                        "the omitted range stays recoverable"
+                    );
+                }
+            });
+    });
+}

crates/tui/src/core/engine/tests/runtime_state.rs

@@ -0,0 +1,270 @@
+//! Engine state that must stay truthful across a boundary: the live MCP
+//! catalog (C02-07/C02-16), a session switch (C02-08/C02-17), a user-input
+//! wait (C02-19), an `/edit` whose replacement never starts (C02-02), and a
+//! malformed MCP config (C02-18). No fixture reaches a provider or a shell.
+
+use super::*;
+use crate::llm_client::mock::MockLlmClient;
+
+fn quiet_engine(config: EngineConfig) -> (Engine, EngineHandle) {
+    Engine::new_with_model_client(
+        config,
+        &Config::default(),
+        Arc::new(MockLlmClient::new(Vec::new())),
+    )
+}
+
+async fn session_snapshot(handle: &EngineHandle) -> SessionSnapshot {
+    let (tx, rx) = tokio::sync::oneshot::channel();
+    handle
+        .send(Op::GetSessionSnapshot {
+            tx: Arc::new(StdMutex::new(Some(tx))),
+        })
+        .await
+        .unwrap();
+    tokio::time::timeout(model_turn_event_timeout(), rx)
+        .await
+        .expect("snapshot response")
+        .expect("snapshot")
+}
+
+#[test]
+fn live_mcp_refresh_drops_tools_the_pool_no_longer_lists() {
+    // C02-07: the caller's universe is what the pool lists *now*. After a
+    // live 401 that is only the synthetic login tool — never the dead one.
+    let mut catalog = vec![api_tool("mcp_alpha_search"), api_tool("exec_shell")];
+    let mut active: HashSet<String> = ["mcp_alpha_search", "exec_shell"]
+        .into_iter()
+        .map(str::to_string)
+        .collect();
+    let universe: HashSet<String> = ["mcp_alpha_authenticate".to_string()].into();
+    let changed = replace_runtime_mcp_tools(
+        &mut catalog,
+        &mut active,
+        &universe,
+        vec![api_tool("mcp_alpha_authenticate")],
+        AppMode::Agent,
+        &HashSet::new(),
+        crate::model_profile::ToolSurfaceBudget::Standard,
+    );
+    let names: Vec<&str> = catalog.iter().map(|tool| tool.name.as_str()).collect();
+    assert_eq!(names, vec!["exec_shell", "mcp_alpha_authenticate"]);
+    assert!(
+        !active.contains("mcp_alpha_search"),
+        "a dead tool is not callable"
+    );
+    assert!(changed);
+}
+
+#[test]
+fn live_mcp_refresh_reports_a_schema_only_edit() {
+    // C02-16: the same names with an edited description is a change the
+    // prefix check must hear about; an identical refresh is not.
+    let refresh = |catalog: &mut Vec<Tool>, active: &mut HashSet<String>, tool: Tool| {
+        replace_runtime_mcp_tools(
+            catalog,
+            active,
+            &["mcp_alpha_read".to_string()].into(),
+            vec![tool],
+            AppMode::Agent,
+            &HashSet::new(),
+            crate::model_profile::ToolSurfaceBudget::Standard,
+        )
+    };
+    let mut catalog = vec![api_tool("exec_shell")];
+    let mut active = HashSet::new();
+    refresh(&mut catalog, &mut active, api_tool("mcp_alpha_read"));
+    let mut edited = api_tool("mcp_alpha_read");
+    edited.description = "Read a resource (now paginated)".to_string();
+    assert!(refresh(&mut catalog, &mut active, edited.clone()));
+    assert!(!refresh(&mut catalog, &mut active, edited));
+}
+
+async fn engine_mid_mcp_boot_with_usage() -> (Engine, EngineHandle, tokio::task::JoinHandle<()>) {
+    let workspace = tempdir().unwrap();
+    let (mut engine, handle) = quiet_engine(deterministic_engine_config(workspace.path()));
+    engine.session.total_usage.input_tokens = 1_000;
+    engine.session.total_usage.output_tokens = 200;
+    let boot = tokio::spawn(std::future::pending::<()>());
+    engine.mcp_boot_task = Some(boot.abort_handle());
+    engine.mcp_boot_in_flight = true;
+    engine.mcp_boot_generation = Some(7);
+    engine.mcp_connection_errors.insert(
+        "previous-server".to_string(),
+        "previous failure".to_string(),
+    );
+    // A same-conversation re-sync keeps its usage and its boot pass.
+    let same = engine.session.id.clone();
+    assert!(engine.install_synced_session_id(same).is_none());
+    assert_eq!(engine.session_snapshot().total_tokens, 1_200);
+    assert!(engine.mcp_boot_in_flight);
+    (engine, handle, boot)
+}
+
+#[tokio::test]
+async fn a_session_boundary_does_not_inherit_the_previous_usage() {
+    // C02-17: session A's tokens are not session B's.
+    let (mut engine, _handle, _boot) = engine_mid_mcp_boot_with_usage().await;
+    assert!(
+        engine
+            .install_synced_session_id("next-session".to_string())
+            .is_some()
+    );
+    assert_eq!(engine.session_snapshot().total_tokens, 0);
+}
+
+#[tokio::test]
+async fn a_session_boundary_abandons_the_previous_mcp_boot() {
+    // C02-08: the dropped pool's pass is aborted and its state cleared.
+    let (mut engine, _handle, boot) = engine_mid_mcp_boot_with_usage().await;
+    assert!(
+        engine
+            .install_synced_session_id("next-session".to_string())
+            .is_some()
+    );
+    assert!(!engine.mcp_boot_in_flight);
+    assert_eq!(engine.mcp_boot_generation, None);
+    assert!(engine.mcp_boot_rx.is_none());
+    assert!(engine.mcp_connection_errors.is_empty());
+    let joined = tokio::time::timeout(Duration::from_secs(5), boot)
+        .await
+        .expect("the previous boot pass must stop promptly");
+    assert!(joined.unwrap_err().is_cancelled(), "the pass was aborted");
+}
+
+fn empty_user_input_request() -> crate::tools::user_input::UserInputRequest {
+    crate::tools::user_input::UserInputRequest {
+        questions: Vec::new(),
+    }
+}
+
+#[tokio::test]
+async fn an_undeliverable_user_input_request_fails_fast() {
+    // C02-19: the question cannot reach a host (its event channel is closed,
+    // while the answer channel stays open), so nobody can ever answer it.
+    let workspace = tempdir().unwrap();
+    let (mut engine, handle) = quiet_engine(deterministic_engine_config(workspace.path()));
+    handle.rx_event.write().await.close();
+    let result = tokio::time::timeout(
+        Duration::from_secs(5),
+        engine.await_user_input("ask-undeliverable", empty_user_input_request()),
+    )
+    .await
+    .expect("an undeliverable question must not wait for an answer");
+    assert!(
+        matches!(result, Err(ToolError::ExecutionFailed { .. })),
+        "{result:?}"
+    );
+}
+
+#[tokio::test]
+async fn an_open_user_input_question_is_not_charged_to_the_turn() {
+    // C02-19: a delivered question the person leaves open is human time, not
+    // the agent's; the per-turn wall clock does not advance across it.
+    let workspace = tempdir().unwrap();
+    let wait = Duration::from_millis(400);
+    let (mut engine, _handle) = quiet_engine(EngineConfig {
+        user_input_timeout: Some(wait),
+        ..deterministic_engine_config(workspace.path())
+    });
+    let before = engine.turn_wall_clock.spent();
+    let result = engine
+        .await_user_input("ask-unanswered", empty_user_input_request())
+        .await;
+    assert!(
+        matches!(result, Err(ToolError::Timeout { .. })),
+        "{result:?}"
+    );
+    let charged = engine.turn_wall_clock.spent().saturating_sub(before);
+    assert!(
+        charged < wait / 2,
+        "the human wait was charged to the turn budget: {charged:?}"
+    );
+}
+
+#[tokio::test]
+#[allow(clippy::await_holding_lock)]
+async fn edit_last_turn_restores_the_exchange_when_the_replacement_never_starts() {
+    // C02-02: `/edit` cuts the last exchange, then dispatches the new text.
+    // A dispatch that never starts must hand the cut exchange back.
+    let _lock = lock_test_env();
+    let tmp = tempdir().unwrap();
+    let api_config = Config::default().with_legacy_root(
+        Some("test-key".to_string()),
+        Some("http://127.0.0.1:9".to_string()),
+    );
+    let (mut engine, handle) = Engine::new_with_model_client(
+        EngineConfig {
+            workspace: tmp.path().to_path_buf(),
+            model: "deepseek-v4-pro".to_string(),
+            snapshots_enabled: false,
+            subagents_enabled: false,
+            ..Default::default()
+        },
+        &api_config,
+        Arc::new(MockLlmClient::new(Vec::new())),
+    );
+    // No model client: the replacement send returns NotStarted.
+    engine.model_client = None;
+    let run = tokio::spawn(engine.run());
+    let text = |role: Role, text: &str| Message {
+        role,
+        content: vec![ContentBlock::Text {
+            text: text.to_string(),
+            cache_control: None,
+        }],
+    };
+    handle
+        .send(Op::SyncSession {
+            session_id: Some("edit-restore".to_string()),
+            messages: vec![
+                text(Role::User, "original prompt"),
+                text(Role::Assistant, "original answer"),
+            ],
+            system_prompt: None,
+            system_prompt_override: false,
+            model: "deepseek-v4-pro".to_string(),
+            workspace: tmp.path().to_path_buf(),
+            mode: AppMode::Agent,
+        })
+        .await
+        .unwrap();
+    let before = session_snapshot(&handle).await;
+    assert_eq!(before.messages.len(), 2);
+    handle
+        .send(Op::EditLastTurn {
+            new_message: "edited prompt".to_string(),
+        })
+        .await
+        .unwrap();
+    let after = session_snapshot(&handle).await;
+    assert_eq!(
+        after.messages, before.messages,
+        "a replacement that never started must not cost the original exchange"
+    );
+    handle.send(Op::Shutdown).await.unwrap();
+    run.await.unwrap();
+}
+
+#[tokio::test]
+async fn malformed_mcp_config_is_reported_instead_of_silently_empty() {
+    // C02-18: an unreadable config still yields an empty, reloadable pool,
+    // but the person is told the configured servers are gone.
+    let workspace = tempdir().unwrap();
+    let config_path = workspace.path().join("mcp.json");
+    fs::write(&config_path, "{ this is not json").unwrap();
+    let (mut engine, handle) = quiet_engine(deterministic_engine_config(workspace.path()));
+    engine.session.mcp_config_path = config_path;
+    engine
+        .ensure_mcp_pool()
+        .await
+        .expect("a reloadable empty pool still exists");
+    let mut rx = handle.rx_event.write().await;
+    let reported = std::iter::from_fn(|| rx.try_recv().ok()).any(|event| {
+        matches!(
+            event,
+            Event::Status { ref message } if message.contains("MCP config could not be loaded")
+        )
+    });
+    assert!(reported, "a malformed MCP config must not fail silently");
+}

crates/tui/src/core/engine/turn_loop.rs

@@ -112,13 +112,39 @@ async fn preserve_tool_output_before_fanout(
     tool_id: &str,
     tool_name: &str,
 ) -> Result<RichToolResult, ToolError> {
-    let mut rich = result?;
     let model = model.to_owned();
     let session_id = session_id.to_owned();
     let tool_id = tool_id.to_owned();
     let tool_name = tool_name.to_owned();
+    let mut rich = match result {
+        Ok(rich) => rich,
+        // C02-12: an error is fanned out to the event stream and the session
+        // exactly like a result, so an oversized one gets the same bounded
+        // head/tail projection and saved artifact. Ordinary short errors
+        // stay byte-identical (the projection only engages past the
+        // spillover threshold).
+        Err(mut error) => {
+            if tool_error_message_mut(&mut error).is_none_or(|message| {
+                message.len() <= crate::tools::truncate::SPILLOVER_THRESHOLD_BYTES
+            }) {
+                return Err(error);
+            }
+            return tokio::task::spawn_blocking(move || {
+                bound_oversized_tool_error(error, &tool_id, &tool_name, &session_id)
+            })
+            .await
+            .map_err(|join_error| {
+                ToolError::execution_failed(format!(
+                    "Tool output preservation failed: {join_error}"
+                ))
+            })
+            .and_then(Err);
+        }
+    };
     tokio::task::spawn_blocking(move || {
-        if let Some(path) = crate::tools::truncate::apply_spillover_with_artifact(
+        // Failed results are bounded too: the fan-out cost of a huge one is
+        // the same whether or not the tool called it a failure (C02-12).
+        if let Some(path) = crate::tools::truncate::apply_spillover_with_artifact_including_errors(
             &mut rich.result,
             &tool_id,
             &tool_name,
@@ -155,6 +181,42 @@ async fn preserve_tool_output_before_fanout(
     })
 }
 
+/// The free-form text of a tool error, when its variant carries one.
+fn tool_error_message_mut(error: &mut ToolError) -> Option<&mut String> {
+    match error {
+        ToolError::InvalidInput { message }
+        | ToolError::ExecutionFailed { message, .. }
+        | ToolError::Cancelled { message }
+        | ToolError::NotAvailable { message }
+        | ToolError::PermissionDenied { message } => Some(message),
+        _ => None,
+    }
+}
+
+/// Give an oversized tool error message the bounded head/tail projection and
+/// saved artifact a failed result gets (C02-12). Blocking: it may write the
+/// artifact, so callers run it under `spawn_blocking`. A failed artifact
+/// write degrades to the original message, as spillover does for results.
+fn bound_oversized_tool_error(
+    mut error: ToolError,
+    tool_id: &str,
+    tool_name: &str,
+    session_id: &str,
+) -> ToolError {
+    let Some(message) = tool_error_message_mut(&mut error) else {
+        return error;
+    };
+    let mut projected = ToolResult::error(std::mem::take(message));
+    crate::tools::truncate::apply_spillover_with_artifact_including_errors(
+        &mut projected,
+        tool_id,
+        tool_name,
+        session_id,
+    );
+    *message = projected.content;
+    error
+}
+
 fn approval_intent_summary(text: &str) -> Option<String> {
     let trimmed = text.trim();
     if trimmed.is_empty() {
@@ -405,6 +467,28 @@ fn incomplete_tool_result(reason: &str) -> ToolResult {
     }
 }
 
+/// Status receipt carried by the one request a reasoning-only / empty-stop
+/// nudge rides (C02-04). The nudge itself never joins the session.
+pub(super) const REQUEST_NUDGE_RECEIPT_PREFIX: &str =
+    "Continuing — this retry carries a request-scoped nudge (not saved to the conversation): ";
+
+/// The not-executed result for a call collected from a response whose stream
+/// failed before it completed (C02-05). Same shape as
+/// [`incomplete_tool_result`]: nothing started, so nothing needs undoing.
+fn stream_failed_tool_result(error: &str) -> ToolResult {
+    ToolResult {
+        content: format!(
+            "Not executed: the provider stream failed before the model response completed ({error})."
+        ),
+        success: false,
+        metadata: Some(json!({
+            "side_effect_status": "not_started",
+            "error_category": "model_stream_failed",
+            "model_output_incomplete": true,
+        })),
+    }
+}
+
 fn registered_tool_requires_non_bypassable_approval(tool_name: &str) -> bool {
     // `rlm_eval` (and the unified `rlm` tool whose eval action inherits the
     // same Required approval) must never bypass explicit approval (#3866).
@@ -414,9 +498,11 @@ fn registered_tool_requires_non_bypassable_approval(tool_name: &str) -> bool {
 /// Replace the runtime-MCP slice of the tool catalog wholesale. An additive
 /// merge could never remove anything: the synthetic `mcp_<server>_
 /// authenticate` entry would survive its own successful login, and tools
-/// killed by a live 401 would stay callable in name. `universe` is every
-/// name the pool can own; entries inside it are the pool's to manage, and
-/// the refreshed list is the new truth.
+/// killed by a live 401 would stay callable in name. The pool owns `universe`
+/// (every name it can list now) and every MCP name already in the catalog —
+/// the second half is what lets a tool whose server vanished or lost its
+/// authorization leave (C02-07); `universe` alone only names survivors. The
+/// refreshed list is the new truth for all of them.
 ///
 /// The refreshed slice is shaped exactly like the turn's initial catalog —
 /// the same deferral pass, the same surface budget, the same always-load
@@ -425,6 +511,10 @@ fn registered_tool_requires_non_bypassable_approval(tool_name: &str) -> bool {
 /// projection carries `defer_loading = false` on every tool, so pushing it
 /// in unshaped put every MCP tool definition into every remaining request
 /// of the turn (#5939).
+///
+/// Returns whether the catalog or its active set changed in any way — a
+/// schema or description edit included, not only a count change (C02-16) —
+/// so the caller can declare the tool-surface change to the prefix check.
 pub(super) fn replace_runtime_mcp_tools(
     tool_catalog: &mut Vec<Tool>,
     active_tool_names: &mut std::collections::HashSet<String>,
@@ -433,11 +523,12 @@ pub(super) fn replace_runtime_mcp_tools(
     mode: AppMode,
     always_load: &std::collections::HashSet<String>,
     surface_budget: crate::model_profile::ToolSurfaceBudget,
-) -> usize {
-    let before = tool_catalog.len();
+) -> bool {
+    let catalog_before = tool_catalog.clone();
+    let active_before = active_tool_names.clone();
     let mut previously_active = std::collections::HashSet::new();
     tool_catalog.retain(|tool| {
-        let owned = universe.contains(&tool.name);
+        let owned = universe.contains(&tool.name) || McpPool::is_mcp_tool(&tool.name);
         if owned && active_tool_names.remove(&tool.name) {
             previously_active.insert(tool.name.clone());
         }
@@ -455,7 +546,7 @@ pub(super) fn replace_runtime_mcp_tools(
         }
         tool_catalog.push(tool);
     }
-    tool_catalog.len().abs_diff(before)
+    *tool_catalog != catalog_before || *active_tool_names != active_before
 }
 
 impl Engine {
@@ -600,6 +691,18 @@ impl Engine {
         None
     }
 
+    /// R1: the turn-ending error once the per-turn wall-clock budget is spent.
+    /// Checked wherever the loop is about to authorize a provider request.
+    fn turn_wall_clock_exhausted_error(&self) -> Option<String> {
+        self.turn_wall_clock.exhausted().then(|| {
+            format!(
+                "Per-turn wall-clock budget exhausted after {}s (limit: {}s). The turn was stopped before another model request; work already done is in the transcript. Send another message to continue, or raise `[tui].turn_wall_clock_secs`.",
+                self.turn_wall_clock.spent().as_secs(),
+                self.turn_wall_clock.budget().as_secs(),
+            )
+        })
+    }
+
     /// A connection completed during inference must be discoverable in this
     /// turn, without widening its command policy or making every MCP tool eager.
     pub(super) async fn refresh_boot_mcp_catalog(
@@ -615,23 +718,17 @@ impl Engine {
         let Some(pool) = self.mcp_pool.as_ref() else {
             return;
         };
-        let (mut universe, mut refreshed) = {
+        let (universe, mut refreshed) = {
             let pool = pool.lock().await;
             let refreshed = pool.to_api_tools();
             (pool.model_tool_names(&refreshed), refreshed)
         };
         // A config/authority change during handshake can remove a server;
-        // its previous names must also leave this turn's catalog.
-        universe.extend(
-            catalog
-                .iter()
-                .filter(|tool| McpPool::is_mcp_tool(&tool.name))
-                .map(|tool| tool.name.clone()),
-        );
+        // `replace_runtime_mcp_tools` owns every MCP name already in the
+        // catalog, so its previous names leave this turn's catalog too.
         refreshed
             .retain(|tool| policy.passes_allow_list(&tool.name) && !policy.denies_tool(&tool.name));
-        let before = catalog.clone();
-        replace_runtime_mcp_tools(
+        if replace_runtime_mcp_tools(
             catalog,
             active,
             &universe,
@@ -640,8 +737,7 @@ impl Engine {
             &self.config.tools_always_load,
             self.turn_tool_surface_budget
                 .unwrap_or(crate::model_profile::ToolSurfaceBudget::Standard),
-        );
-        if *catalog != before {
+        ) {
             self.session.pending_prefix_change_reason = Some("mcp-session-boot".to_string());
         }
     }
@@ -997,8 +1093,11 @@ impl Engine {
         // retry after that carries a nudge — attached to one outbound request
         // and dropped, never added to the session. Writing it to the session
         // would put a message the user never sent into the transcript, the
-        // exports, and every later turn's context.
-        let mut reasoning_only_nudge: Option<Message> = None;
+        // exports, and every later turn's context. It is still model-visible,
+        // so the request that carries it also emits a durable internal
+        // status receipt with its exact text (C02-04, "model-visible means
+        // logged"): the runtime event log can reconstruct the request.
+        let mut reasoning_only_nudge: Option<String> = None;
         // Outer stream-retry budget: when the chunked-transfer connection
         // dies mid-stream and either nothing useful was streamed (#103
         // Phase 3), the host slept mid-turn (#2990), or a host hit a
@@ -1034,12 +1133,7 @@ impl Engine {
             // result already produced stays in the transcript. Hitting it is
             // never a clean success — the turn ends `Failed` with the limit
             // named, matching how the step ceiling below reports.
-            if self.turn_wall_clock.exhausted() {
-                let error = format!(
-                    "Per-turn wall-clock budget exhausted after {}s (limit: {}s). The turn was stopped before another model request; work already done is in the transcript. Send another message to continue, or raise `[tui].turn_wall_clock_secs`.",
-                    self.turn_wall_clock.spent().as_secs(),
-                    self.turn_wall_clock.budget().as_secs(),
-                );
+            if let Some(error) = self.turn_wall_clock_exhausted_error() {
                 let _ = self.tx_event.send(Event::status(error.clone())).await;
                 return (TurnOutcomeStatus::Failed, Some(error));
             }
@@ -1418,6 +1512,13 @@ impl Engine {
                     }
                 }
                 self.finish_compaction(&compaction_id);
+                // C02-06: a compaction pass has its own bound, not the
+                // turn's. Recheck the wall clock before it can authorize the
+                // provider request below.
+                if let Some(error) = self.turn_wall_clock_exhausted_error() {
+                    let _ = self.tx_event.send(Event::status(error.clone())).await;
+                    return (TurnOutcomeStatus::Failed, Some(error));
+                }
             }
 
             // The guard measures what the compaction gate measures: the honest
@@ -1728,14 +1829,18 @@ impl Engine {
             let fleet_report_response = fleet_denial_guard
                 .as_ref()
                 .is_some_and(FleetDenialGuard::report_only);
+            // `take` is what keeps this request-scoped: the nudge is spent
+            // here and never reaches `self.session.messages`.
+            let request_nudge = reasoning_only_nudge.take();
             let mut request = prepare_primary_turn_request(PrimaryTurnRequest {
                 model: self.session.model.clone(),
                 messages: {
                     let mut messages = self.messages_with_turn_metadata();
-                    // `take` is what keeps this request-scoped: the nudge is
-                    // spent here and never reaches `self.session.messages`.
-                    if let Some(nudge) = reasoning_only_nudge.take() {
-                        messages.push(nudge);
+                    if let Some(nudge) = request_nudge.as_ref() {
+                        messages.push(self.runtime_text_message_with_turn_metadata(
+                            nudge.clone(),
+                            UserInputProvenance::Runtime,
+                        ));
                     }
                     messages
                 },
@@ -1748,10 +1853,11 @@ impl Engine {
                 system: self.session.system_prompt.clone(),
                 tools: active_tools.clone(),
                 tool_choice: if active_tools.is_some() {
-                    if fleet_report_response {
+                    if fleet_report_response || turn.budget_exhausted_final_report {
                         // Keep the pinned tool prefix; only this request's
                         // choice changes. Admission below also enforces this
-                        // if a provider ignores the report-only request.
+                        // if a provider ignores the report-only request
+                        // (C02-10: the step-budget final report included).
                         Some(json!("none"))
                     } else if strict_tool_mode {
                         Some(json!("required"))
@@ -1820,6 +1926,16 @@ impl Engine {
                     snapshot: tool_request_snapshot,
                 })
                 .await;
+            if let Some(nudge) = request_nudge {
+                // The "Continuing — " prefix classifies the receipt as
+                // internal: durable clients keep it, collapsed.
+                let _ = self
+                    .tx_event
+                    .send(Event::status(format!(
+                        "{REQUEST_NUDGE_RECEIPT_PREFIX}{nudge}"
+                    )))
+                    .await;
+            }
             if let Some(mut route) = turn.pending_route.take() {
                 if let Some(billing) = route.billing.as_mut() {
                     // Freeze the exact provider-live row at CodeWhale's
@@ -2015,6 +2131,11 @@ impl Engine {
                 None,
                 Some(super::turn_heartbeat::PREPARING_PHASE_BOUND),
             );
+            // C02-05: a response whose stream failed — a provider error frame,
+            // a transport error, a stall, or a cap — is not a complete
+            // response. Unless the retry below re-issues the request, nothing
+            // it collected may execute or continue the turn.
+            let response_stream_failed = stream_error.is_some();
             turn_error = turn_error.or(stream_error);
             turn.stop_diagnostics
                 .observe_provider_response(stop_reason.as_deref(), tool_uses.len());
@@ -2115,26 +2236,8 @@ impl Engine {
                     output_limit_truncated = Some(reason.to_string());
                     // Fall through to the normal content/tool dispatch below.
                 } else {
-                    for tool in &tool_uses {
-                        let _ = self
-                            .tx_event
-                            .send(Event::ToolCallStarted {
-                                id: tool.execution_id.clone(),
-                                model_call: Some(tool.model_call()),
-                                name: tool.name.clone(),
-                                input: final_tool_input(tool),
-                            })
-                            .await;
-                        let _ = self
-                            .tx_event
-                            .send(Event::ToolCallComplete {
-                                id: tool.execution_id.clone(),
-                                model_call: Some(tool.model_call()),
-                                name: tool.name.clone(),
-                                result: Ok(incomplete_tool_result(reason)),
-                            })
-                            .await;
-                    }
+                    self.settle_unadmitted_tool_calls(&tool_uses, &incomplete_tool_result(reason))
+                        .await;
                     // Do not emit MessageComplete: hosts must retain the visible
                     // fragment as interrupted/failed rather than recording it as
                     // a completed assistant item.
@@ -2334,6 +2437,28 @@ impl Engine {
                 }
             }
 
+            // C02-05: the one admission authority for a failed stream. Calls
+            // collected before the failure are announced with an explicit
+            // not-executed result and never reach planning, approval or a
+            // handler; the visible text is kept as an interrupted fragment
+            // and no tool_use enters history, so the transcript stays paired.
+            // A retry never gets here: it `continue`d above with this batch
+            // dropped, and the re-issued request streams its own calls.
+            if response_stream_failed && !tool_uses.is_empty() {
+                let error = turn_error
+                    .clone()
+                    .unwrap_or_else(|| "provider stream failed".to_string());
+                self.settle_unadmitted_tool_calls(
+                    &tool_uses,
+                    &stream_failed_tool_result(&summarize_text(&error, 200)),
+                )
+                .await;
+                self.add_interrupted_assistant_text(&current_text_visible)
+                    .await;
+                turn.stop_diagnostics.last_response_tool_calls_suppressed = Some(tool_uses.len());
+                return (TurnOutcomeStatus::Failed, Some(error));
+            }
+
             for tool in &tool_uses {
                 let _ = self
                     .tx_event
@@ -2484,6 +2609,15 @@ impl Engine {
                 .await;
             }
 
+            // C02-05: a failed stream that carried no tool call keeps its text
+            // (above) and ends the turn with the stream's error. It never
+            // runs a ```repl fence from the failed text, and never authorizes
+            // another request for an output-limit continuation, a steer, a
+            // sub-agent completion or a goal continuation.
+            if response_stream_failed {
+                break;
+            }
+
             // A truncated response with no tool call cannot continue through
             // tool execution: surface the truncation as a bounded observation
             // and resume the loop so the model can act on it instead of the
@@ -3031,11 +3165,7 @@ impl Engine {
                                 crate::config::DEFAULT_REASONING_ONLY_REPROMPT_MESSAGE.to_string()
                             });
                         if !text.trim().is_empty() {
-                            reasoning_only_nudge =
-                                Some(self.runtime_text_message_with_turn_metadata(
-                                    text,
-                                    UserInputProvenance::Runtime,
-                                ));
+                            reasoning_only_nudge = Some(text);
                         }
                     }
                     let how = if nudged {
@@ -3093,11 +3223,7 @@ impl Engine {
                                         .to_string()
                                 });
                             if !text.trim().is_empty() {
-                                reasoning_only_nudge =
-                                    Some(self.runtime_text_message_with_turn_metadata(
-                                        text,
-                                        UserInputProvenance::Runtime,
-                                    ));
+                                reasoning_only_nudge = Some(text);
                             }
                             "re-requesting the answer with a nudge"
                         }
@@ -3555,6 +3681,18 @@ impl Engine {
                 blocked_error = guard.admission_error(&tool_name, &tool_input);
             }
 
+            // C02-10: the response granted after the step budget ran out is
+            // report-only. A provider that ignores `tool_choice: none` still
+            // gets no execution; the next loop pass ends the turn at the
+            // exhausted budget.
+            if blocked_error.is_none() && turn.budget_exhausted_final_report {
+                blocked_error = Some(ToolError::permission_denied(format!(
+                    "Model-step budget exhausted (limit: {}, {}): this is the final report response, so no tool may execute. Report what you did, what you found, what remains, and the evidence.",
+                    turn.max_steps,
+                    turn.budget_source.key_label(),
+                )));
+            }
+
             if blocked_error.is_none()
                 && let Some(error) = tool.input_parse_error.clone()
             {
@@ -5533,7 +5671,7 @@ impl Engine {
                             self.current_mode,
                             &self.config.tools_always_load,
                             surface_budget,
-                        ) > 0;
+                        );
                     }
                     // Any of the legitimate mid-turn tool-surface changes above
                     // re-pin the header under a declared `change:tool_surface`
@@ -5840,18 +5978,6 @@ impl Engine {
                 break;
             }
 
-            // Guard: max accumulated content bytes
-            if stream_content_bytes > max_content_bytes {
-                let envelope = StreamError::Overflow {
-                    limit_bytes: max_content_bytes,
-                }
-                .into_envelope();
-                crate::logging::warn(&envelope.message);
-                stream_error.get_or_insert(envelope.message.clone());
-                let _ = self.tx_event.send(Event::error(envelope)).await;
-                break;
-            }
-
             let event = match event_result {
                 Ok(e) => {
                     self.turn_heartbeat.stream_progress(
@@ -5895,6 +6021,15 @@ impl Engine {
                             last_progress_mono.elapsed(),
                             wall_elapsed,
                         ));
+                        // Like the network-drop resumes below, keep the real
+                        // error as the prospective outcome: the retry clears
+                        // it, and an exhausted resume budget then fails the
+                        // turn with it instead of admitting the partial
+                        // response as if it had completed.
+                        stream_error.get_or_insert(stream_read_error_user_message(
+                            &message,
+                            any_content_received,
+                        ));
                         pending_resume = Some(StreamResume::AfterSleep);
                         break;
                     }
@@ -6040,6 +6175,24 @@ impl Engine {
                 }
             };
 
+            // Guard: max accumulated content bytes (C02-13). Counted per
+            // event and checked before the event is applied, so the delta
+            // that crosses the cap is never forwarded or accumulated — even
+            // when it is the stream's last — and tool-argument JSON counts
+            // like text and reasoning.
+            stream_content_bytes =
+                stream_content_bytes.saturating_add(stream_event_content_bytes(&event));
+            if stream_content_bytes > max_content_bytes {
+                let envelope = StreamError::Overflow {
+                    limit_bytes: max_content_bytes,
+                }
+                .into_envelope();
+                crate::logging::warn(&envelope.message);
+                stream_error.get_or_insert(envelope.message.clone());
+                let _ = self.tx_event.send(Event::error(envelope)).await;
+                break;
+            }
+
             match event {
                 StreamEvent::ToolProjectionWarning {
                     provider,
@@ -6150,7 +6303,6 @@ impl Engine {
                 },
                 StreamEvent::ContentBlockDelta { index, delta } => match delta {
                     Delta::TextDelta { text } => {
-                        stream_content_bytes = stream_content_bytes.saturating_add(text.len());
                         current_text_raw.push_str(&text);
                         let filtered =
                             filter_tool_call_delta_with_state(&text, &mut tool_call_filter);
@@ -6173,7 +6325,6 @@ impl Engine {
                         }
                     }
                     Delta::ThinkingDelta { thinking } => {
-                        stream_content_bytes = stream_content_bytes.saturating_add(thinking.len());
                         current_thinking.push_str(&thinking);
                         if !thinking.is_empty() {
                             let _ = self
@@ -6338,6 +6489,32 @@ impl Engine {
         }
     }
 
+    /// Announce every call of a response that will not be admitted, each
+    /// paired with its not-executed `result`, so a host never shows a started
+    /// call without a completion. Nothing here plans, approves or executes.
+    async fn settle_unadmitted_tool_calls(&self, tool_uses: &[ToolUseState], result: &ToolResult) {
+        for tool in tool_uses {
+            let _ = self
+                .tx_event
+                .send(Event::ToolCallStarted {
+                    id: tool.execution_id.clone(),
+                    model_call: Some(tool.model_call()),
+                    name: tool.name.clone(),
+                    input: final_tool_input(tool),
+                })
+                .await;
+            let _ = self
+                .tx_event
+                .send(Event::ToolCallComplete {
+                    id: tool.execution_id.clone(),
+                    model_call: Some(tool.model_call()),
+                    name: tool.name.clone(),
+                    result: Ok(result.clone()),
+                })
+                .await;
+        }
+    }
+
     /// Finalize one streamed tool call's input from its accumulated buffer.
     ///
     /// The parse that lands here must be structurally intact: a value that
@@ -7448,6 +7625,37 @@ fn stream_event_has_actionable_content(event: &StreamEvent) -> bool {
     }
 }
 
+/// Bytes an event adds to the response the engine accumulates: text,
+/// reasoning, tool calls (id, name and argument JSON, whether the arguments
+/// arrive whole in the block start or as `InputJsonDelta`s) and replay
+/// signatures. Opaque reasoning state is provider-owned and not counted.
+fn stream_event_content_bytes(event: &StreamEvent) -> usize {
+    fn initial_input_bytes(input: &Value) -> usize {
+        let empty = input.is_null() || input.as_object().is_some_and(serde_json::Map::is_empty);
+        if empty { 0 } else { input.to_string().len() }
+    }
+    match event {
+        StreamEvent::ContentBlockStart { content_block, .. } => match content_block {
+            ContentBlockStart::Text { text } => text.len(),
+            ContentBlockStart::Thinking { thinking } => thinking.len(),
+            ContentBlockStart::ToolUse {
+                id, name, input, ..
+            }
+            | ContentBlockStart::ServerToolUse { id, name, input } => {
+                id.len() + name.len() + initial_input_bytes(input)
+            }
+        },
+        StreamEvent::ContentBlockDelta { delta, .. } => match delta {
+            Delta::TextDelta { text } => text.len(),
+            Delta::ThinkingDelta { thinking } => thinking.len(),
+            Delta::InputJsonDelta { partial_json } => partial_json.len(),
+            Delta::SignatureDelta { signature } => signature.len(),
+            Delta::ReasoningStateDelta { .. } => 0,
+        },
+        _ => 0,
+    }
+}
+
 /// Sentinel reasoning-effort value meaning "let the auto-reasoning system
 /// decide" (#4158).
 pub(super) const REASONING_EFFORT_AUTO: &str = "auto";

crates/tui/src/voice.rs

@@ -421,7 +421,16 @@ pub(crate) async fn transcribe_local_whisper(audio_samples: &[i16]) -> Result<St
 }
 
 fn transcribe_local_whisper_blocking(wav: &[u8]) -> Result<String, String> {
-    let tmp = std::env::temp_dir().join(format!("cw-voice-{}.wav", std::process::id()));
+    // C01-10: a fresh, exclusively created file per call. The old
+    // `cw-voice-<pid>.wav` name was shared by every transcription in the
+    // process (an interim pass, the final pass, a runtime-API dictation) and
+    // predictable in a shared temp dir. Dropping `audio` removes it.
+    let audio = tempfile::Builder::new()
+        .prefix("cw-voice-")
+        .suffix(".wav")
+        .tempfile()
+        .map_err(|e| e.to_string())?;
+    let tmp = audio.path().to_path_buf();
     std::fs::write(&tmp, wav).map_err(|e| e.to_string())?;
     // Try each local binary until one succeeds; whisper.cpp outputs to stdout or file.
     for bin in LOCAL_WHISPER_BINS {
@@ -437,7 +446,6 @@ fn transcribe_local_whisper_blocking(wav: &[u8]) -> Result<String, String> {
             && out.status.success()
         {
             let txt = String::from_utf8_lossy(&out.stdout).trim().to_string();
-            let _ = std::fs::remove_file(&tmp);
             if !txt.is_empty() {
                 return Ok(txt);
             }
@@ -445,14 +453,12 @@ fn transcribe_local_whisper_blocking(wav: &[u8]) -> Result<String, String> {
             let sidecar = tmp.with_extension("txt");
             if let Ok(s) = std::fs::read_to_string(&sidecar) {
                 let _ = std::fs::remove_file(&sidecar);
-                let _ = std::fs::remove_file(&tmp);
                 if !s.trim().is_empty() {
                     return Ok(s.trim().to_string());
                 }
             }
         }
     }
-    let _ = std::fs::remove_file(&tmp);
     Err("local whisper not available".into())
 }
 
@@ -467,28 +473,39 @@ pub(crate) async fn transcribe_groq(audio_samples: &[i16]) -> Result<String, Str
 }
 
 /// Resolve ASR model/provider preference.
-/// Priority: explicit config `voice.asr_model` > env `CODEWHALE_ASR_MODEL` > auto-detect (local-whisper > groq > xiaomi).
+/// Priority: env `CODEWHALE_ASR_MODEL` > auto-detect (local-whisper > groq >
+/// provider). Known limitation (C01-10): there is no config-file ASR setting
+/// yet — `config` is accepted for the day one exists and is not read, so the
+/// env var is the only explicit override.
 pub(crate) fn resolve_asr_choice(_config: &Config) -> (String, String) {
-    // Check explicit env override first (free, cross-platform)
-    if let Ok(m) = std::env::var("CODEWHALE_ASR_MODEL") {
-        let m = m.trim().to_ascii_lowercase();
-        if m.contains("groq") || m.contains("whisper") {
-            return ("groq".into(), GROQ_ASR_MODEL.into());
-        }
-        if m.contains("local") || m.contains("whisper.cpp") {
-            return ("local-whisper".into(), "tiny".into());
-        }
-        if m.contains("mimo") || m.contains("xiaomi") {
-            return ("provider".into(), ASR_MODEL.into());
-        }
+    let explicit = std::env::var("CODEWHALE_ASR_MODEL")
+        .ok()
+        .and_then(|value| explicit_asr_choice(&value));
+    // Auto-detect best free: local whisper (offline, no key) > Groq free
+    // tier > provider ASR (needs key).
+    let (kind, model) = explicit.unwrap_or_else(|| match detect_free_asr() {
+        "local-whisper" => ("local-whisper", "tiny"),
+        "groq" => ("groq", GROQ_ASR_MODEL),
+        _ => ("provider", ASR_MODEL),
+    });
+    (kind.to_string(), model.to_string())
+}
+
+/// Map an explicit `CODEWHALE_ASR_MODEL` value to (kind, model). Local
+/// spellings are tested first: `local-whisper` and `whisper.cpp` both contain
+/// "whisper", which on its own names Groq's hosted Whisper (C01-10).
+fn explicit_asr_choice(value: &str) -> Option<(&'static str, &'static str)> {
+    let m = value.trim().to_ascii_lowercase();
+    if m.contains("local") || m.contains("whisper.cpp") || m.contains("whisper-cpp") {
+        return Some(("local-whisper", "tiny"));
     }
-    // Auto-detect best free: local whisper (offline, no key) > Groq free tier > Xiaomi ASR (needs key)
-    let free = detect_free_asr();
-    match free {
-        "local-whisper" => ("local-whisper".into(), "tiny".into()),
-        "groq" => ("groq".into(), GROQ_ASR_MODEL.into()),
-        _ => ("provider".into(), ASR_MODEL.into()),
+    if m.contains("groq") || m.contains("whisper") {
+        return Some(("groq", GROQ_ASR_MODEL));
     }
+    if m.contains("mimo") || m.contains("xiaomi") {
+        return Some(("provider", ASR_MODEL));
+    }
+    None
 }
 
 // --- Headless capture (HTTP/native-client path) ----------------------------
@@ -717,6 +734,28 @@ mod tests {
         assert!(independent.get("provider").is_none());
     }
 
+    #[test]
+    fn explicit_local_whisper_is_not_routed_to_groq() {
+        // C01-10: every local spelling contains "whisper"; none may reach the
+        // hosted Groq backend (which needs a key and sends audio off-host).
+        for local in ["local-whisper", "whisper.cpp", "whisper-cpp", " LOCAL "] {
+            assert_eq!(
+                explicit_asr_choice(local),
+                Some(("local-whisper", "tiny")),
+                "{local}"
+            );
+        }
+        for hosted in ["groq", "whisper", "whisper-large-v3-turbo"] {
+            assert_eq!(
+                explicit_asr_choice(hosted),
+                Some(("groq", GROQ_ASR_MODEL)),
+                "{hosted}"
+            );
+        }
+        assert_eq!(explicit_asr_choice("mimo"), Some(("provider", ASR_MODEL)));
+        assert_eq!(explicit_asr_choice("something-else"), None);
+    }
+
     #[test]
     fn wav_encoding_produces_valid_header() {
         let samples = vec![0i16; 16000]; // 1 second of silence