opencrabs 0.3.38

use super::builder::AgentService;
use super::types::*;
use crate::brain::agent::context::AgentContext;
use crate::brain::agent::error::{AgentError, Result};
use crate::brain::provider::{ContentBlock, LLMRequest, LLMResponse, Message};
use crate::brain::tools::ToolExecutionContext;
use crate::services::{MessageService, SessionService};
use serde_json::Value;
use std::sync::Arc;
use tokio_util::sync::CancellationToken;
use uuid::Uuid;

/// How many consecutive primary-provider failures (each rescued by
/// a successful fallback) before the fallback gets persisted into
/// the session as the new active provider. Below this, every primary
/// failure triggers a one-shot rescue and the primary is restored for
/// the next request — so a brief outage doesn't permanently demote
/// the primary. User-stated intent (2026-05-30): "if fallback 3
/// times consecutively successfully, the 4th it sticks".
const STICKY_FALLBACK_THRESHOLD: u32 = 4;

/// True when a provider's reported `input_tokens` is implausibly larger than
/// the real content size (system + messages + tool schemas) — the signature of
/// an OVER-REPORTING endpoint. The zhipu "coding" endpoint was observed adding
/// a flat ~20k to every call's reported input regardless of content size (a
/// fixed additive overhead, not a tokenizer ratio): an 8.4k request came back
/// as 28.8k. tiktoken and any real model tokenizer agree within ~2× for normal
/// text, so beyond that we don't trust the provider's number for calibrating
/// the ctx counter (and the billed-cost display). `local_estimate` is the
/// pre-calibration `context.token_count` (system + messages); `tool_tokens` is
/// the tool-schema size the provider also receives but the local estimate omits.
pub(crate) fn is_implausible_token_report(
    local_estimate: usize,
    tool_tokens: usize,
    reported: usize,
) -> bool {
    let expected = local_estimate + tool_tokens;
    expected >= 1000 && reported > expected.saturating_mul(2)
}

/// Minimum summed active-streaming time (seconds) below which a tok/s
/// reading is not credible. Burst-delivering providers (e.g. glm-5.1)
/// can dump an entire short response in a single sub-second SSE chunk,
/// making the active window ~8ms and the computed rate physically
/// impossible (37203 tok/s observed 2026-06-06). Below this floor the
/// timing is too coarse to represent a generation rate.
const MIN_ACTIVE_SECS_FOR_TOK_S: f64 = 0.3;

/// Upper bound on a believable streaming tok/s. Frontier models stream
/// to end users at tens to low-hundreds tok/s; the fastest specialized
/// inference (Groq/Cerebras on small models) tops out around 1-2k. A
/// computed rate above this is a network-burst measurement artifact,
/// not a real generation rate, so we show nothing rather than a fantasy
/// number.
const MAX_PLAUSIBLE_TOK_S: f64 = 2000.0;

/// Compute the streaming tokens-per-second for a turn, returning `None`
/// when the measurement isn't credible.
///
/// `total_output_tokens` is the turn's summed output tokens; `active_secs`
/// is the summed per-iteration active-streaming windows (idle gaps and
/// tool/approval time already excluded by `helpers::stream_complete`).
///
/// Returns `None` when:
/// - there are no output tokens, OR
/// - the active window is below `MIN_ACTIVE_SECS_FOR_TOK_S` (burst
///   delivery — timing too coarse to be a rate), OR
/// - the resulting rate exceeds `MAX_PLAUSIBLE_TOK_S` (multi-burst
///   artifact that still clears the floor).
///
/// Pure + free-function so the channel-footer rate can be unit-tested
/// without spinning the whole tool loop.
pub(crate) fn compute_streaming_tok_per_sec(
    total_output_tokens: u32,
    active_secs: f64,
) -> Option<f64> {
    if total_output_tokens == 0 || active_secs < MIN_ACTIVE_SECS_FOR_TOK_S {
        return None;
    }
    let rate = total_output_tokens as f64 / active_secs;
    if rate.is_finite() && rate <= MAX_PLAUSIBLE_TOK_S {
        Some(rate)
    } else {
        None
    }
}

/// Cross-provider model leak guard. Returns the model the next LLM call
/// should ship with, plus `Some(stale)` when the resolved pin had to be
/// substituted (so the caller can log the swap once with rich context).
///
/// Logic: if the active provider's `supported_models()` list is non-empty
/// AND the pinned model isn't in it, substitute the provider's own default
/// and report the original as stale. Empty `supported_models()` means the
/// provider hasn't declared a catalogue (no `/v1/models` impl, manual config
/// without a `models = [...]` array) — accept the pin in that case so those
/// providers still work.
///
/// Why a free function: the caller is an async method on `AgentService` with
/// DB / provider lookups, but the substitution itself is pure. Factoring it
/// out keeps the guard unit-testable from a synchronous test file under
/// `src/tests/` without spinning a runtime or mocking SessionService.
pub(crate) fn guard_cross_provider_model_leak(
    resolved: String,
    provider_default: &str,
    supported: &[String],
) -> (String, Option<String>) {
    if supported.is_empty() || supported.iter().any(|m| m == &resolved) {
        (resolved, None)
    } else {
        (provider_default.to_string(), Some(resolved))
    }
}

/// Strip ANSI escape codes from raw tool output before persisting to DB.
/// Prevents garbled artifacts in session history.
/// Build the content string sent to the LLM for a tool result.
///
/// On success, returns the raw output. On failure, includes the error
/// message plus captured output (ANSI-stripped, size-capped to 8000 chars).
pub(crate) fn build_tool_result_content(
    success: bool,
    error: Option<String>,
    output: &str,
) -> String {
    if success {
        output.to_string()
    } else {
        let mut msg =
            strip_ansi_output(&error.unwrap_or_else(|| "Tool execution failed".to_string()));
        if !output.is_empty() {
            let captured: String = strip_ansi_output(output).chars().take(8000).collect();
            msg.push_str("\n\n-- output captured before error --\n");
            msg.push_str(&captured);
            if output.len() > 8000 {
                msg.push_str("\n... (output truncated)");
            }
        }
        msg
    }
}

pub(crate) fn strip_ansi_output(raw: &str) -> String {
    strip_ansi::strip_ansi(raw)
}

/// Pull the file path the agent just touched out of a successful tool
/// call, ready for the persistent recent-paths store. Returns `None`
/// for tools that don't address a single file (`bash`, `glob`, …),
/// for inputs that don't carry a path field, or for empty strings.
///
/// The 2026-04-25/26 logs showed that the agent's wrong-path failures
/// concentrate on `read_file`, `edit_file`, `grep`, `ls` — those are
/// the tools whose successful inputs are worth re-surfacing later.
/// `write_file` is included for symmetry: if the agent just wrote a
/// file, it'll likely want to read or edit it again next turn.
///
/// The returned path is resolved against `working_directory` so we
/// always store an absolute, then-collapsed `~/...` form.
pub(crate) fn extract_path_for_recent_buffer(
    tool_name: &str,
    input: &Value,
    working_directory: &std::path::Path,
) -> Option<std::path::PathBuf> {
    let raw_path = match tool_name {
        "read_file" | "edit_file" | "write_file" | "ls" | "grep" => {
            input.get("path").and_then(|v| v.as_str())?
        }
        _ => return None,
    };
    if raw_path.trim().is_empty() {
        return None;
    }
    Some(crate::brain::tools::error::resolve_tool_path(
        raw_path,
        working_directory,
    ))
}

/// RAII guard that restores a session's provider on drop.
///
/// The fallback arms in `run_tool_loop_inner` swap the session's
/// provider to a fallback, await the fallback's stream, then restore
/// the original. That pattern was cancellation-unsafe: when the user
/// sent a new message mid-fallback, the containing future was
/// dropped, the line after `.await` never ran, and
/// `session_providers[session_id]` stayed on the fallback. The next
/// turn then built a request with the session's saved model (primary)
/// but sent it to the fallback provider — producing the
/// 2026-04-18 18:14 "400 Unknown Model, please check the model code"
/// from zhipu after it received `model=qwen3.6-plus`.
///
/// Drop runs whether the future completes, errors, or is cancelled.
struct FallbackProviderGuard<'a> {
    service: &'a AgentService,
    session_id: Uuid,
    original: Option<Arc<dyn crate::brain::provider::Provider>>,
}

impl Drop for FallbackProviderGuard<'_> {
    fn drop(&mut self) {
        if let Some(original) = self.original.take() {
            // Restore the original provider with its own paired model — the
            // guard is a transient temp-swap, so put the pair back as it was.
            let model = original
                .active_subprovider_model()
                .unwrap_or_else(|| original.default_model().to_string());
            self.service
                .swap_provider_for_session(self.session_id, original, model);
        }
    }
}

/// Detect whether a user message is a correction or negative feedback.
///
/// Public for testing — used internally by the tool loop.
///
/// Looks for patterns like "no", "wrong", "that's not what I meant", "try again",
/// "you broke", etc. Only checks the first 300 chars and requires the message
/// to be short-ish (under 500 chars) — long messages are usually new instructions,
/// not corrections.
pub fn is_user_correction(msg: &str) -> bool {
    let len = msg.len();
    // Long messages are usually new prompts, not corrections
    if !(2..=500).contains(&len) {
        return false;
    }
    let lower: String = msg.chars().take(300).collect::<String>().to_lowercase();
    // Strong negative signals — short phrases that clearly indicate correction
    const PATTERNS: &[&str] = &[
        "no,",
        "no.",
        "no!",
        "no that",
        "no not",
        "nope",
        "wrong",
        "that's not",
        "thats not",
        "not what i",
        "try again",
        "redo",
        "revert",
        "undo",
        "you broke",
        "broke it",
        "doesn't work",
        "doesnt work",
        "didn't work",
        "didnt work",
        "not working",
        "stop",
        "don't do",
        "dont do",
        "i said",
        "i asked",
        "that's wrong",
        "thats wrong",
        "not correct",
        "fix it",
        "fix this",
    ];
    PATTERNS.iter().any(|p| lower.contains(p))
}

impl AgentService {
    /// Core tool-execution loop — called by all public shims.
    /// `override_approval_callback` and `override_progress_callback` take
    /// precedence over the service-level callbacks (used by Telegram, etc.)
    ///
    /// `display_text_override`: when `Some`, channels can supply a human-
    /// readable user message for DB persistence/TUI display while the LLM
    /// still receives the full `user_message` (typically wrapped with
    /// channel/sender/reply metadata for context).
    #[allow(clippy::too_many_arguments)]
    pub(super) async fn run_tool_loop(
        &self,
        session_id: Uuid,
        user_message: String,
        display_text_override: Option<String>,
        model: Option<String>,
        cancel_token: Option<CancellationToken>,
        override_approval_callback: Option<ApprovalCallback>,
        override_progress_callback: Option<ProgressCallback>,
        override_question_callback: Option<QuestionCallback>,
        channel: &str,
        channel_chat_id: Option<&str>,
    ) -> Result<AgentResponse> {
        // Track this request for restart recovery
        let pending_repo = crate::db::PendingRequestRepository::new(self.context.pool());
        let request_id = Uuid::new_v4();
        if let Err(e) = pending_repo
            .insert(
                request_id,
                session_id,
                &user_message,
                channel,
                channel_chat_id,
            )
            .await
        {
            tracing::warn!("Failed to track pending request: {}", e);
        }

        // Per-call effective callbacks (override wins over service-level).
        // Track whether an explicit per-call override was provided so we can honour
        // channel approval callbacks even when the factory set auto_approve_tools=true.
        let has_override_approval = override_approval_callback.is_some();
        let approval_callback: Option<ApprovalCallback> =
            override_approval_callback.or_else(|| self.approval_callback.clone());
        let has_progress_override = override_progress_callback.is_some();
        let progress_callback: Option<ProgressCallback> =
            override_progress_callback.or_else(|| self.progress_callback.clone());
        // Effective question callback: per-call override wins over the
        // service-level fallback. Channels with native button surfaces
        // pass their own callback per message; everyone else passes
        // None and the `follow_up_question` tool returns a graceful
        // "no interactive surface" error.
        let question_callback: Option<QuestionCallback> =
            override_question_callback.or_else(|| self.question_callback.clone());

        // Notify TUI when a remote channel starts/finishes processing so it can
        // block concurrent sends on the same session and avoid garbled display.
        if has_progress_override && let Some(ref tx) = self.session_updated_tx {
            let _ = tx.send(crate::brain::agent::ChannelSessionEvent::ProcessingStarted(
                session_id,
            ));
        }

        // Run the actual loop
        let result = self
            .run_tool_loop_inner(
                session_id,
                user_message,
                display_text_override,
                model,
                cancel_token,
                has_override_approval,
                approval_callback,
                has_progress_override,
                progress_callback,
                question_callback,
            )
            .await;

        if has_progress_override && let Some(ref tx) = self.session_updated_tx {
            let _ =
                tx.send(crate::brain::agent::ChannelSessionEvent::ProcessingFinished(session_id));
        }

        // Request finished — delete the tracking row. Only PROCESSING rows
        // survive (meaning the process crashed/restarted mid-request).
        if let Err(e) = pending_repo.delete(request_id).await {
            tracing::warn!("Failed to clean up pending request: {}", e);
        }

        result
    }

    /// Inner tool loop — separated so `run_tool_loop` can wrap with request tracking.
    ///
    /// `display_text_override`: optional clean message for DB persistence/TUI.
    /// The LLM context still gets `user_message` (full agent input including
    /// channel-injected sender metadata, reply context, group history, etc.).
    /// When `None`, behaves identically to before — `user_message` is used
    /// for both context and DB.
    /// Honor a mid-turn manual provider/model switch AFTER a turn finishes.
    /// If the user switched while this turn was running (`start_epoch` no
    /// longer matches), re-install their pinned provider+model pair in memory
    /// and persist it to the session DB row — so the next turn (and channel
    /// restore) use the user's pick, not the fallback the turn happened to
    /// take. Called only after the response is fully built, so it cannot drop
    /// or change the current turn's result. No-op when nothing changed.
    async fn finalize_manual_switch(
        &self,
        session_id: Uuid,
        start_epoch: u64,
        session_service: &SessionService,
    ) {
        let Some(model) = self.restore_manual_switch_if_changed(session_id, start_epoch) else {
            return;
        };
        let provider_name = self.provider_name_for_session(session_id);
        if let Ok(Some(mut s)) = session_service.get_session(session_id).await {
            s.provider_name = Some(provider_name.clone());
            s.model = Some(model.clone());
            if let Err(e) = session_service.update_session(&s).await {
                tracing::warn!("finalize_manual_switch: DB persist failed for {session_id}: {e}");
            }
        }
        tracing::info!(
            "Restored user's mid-turn switch for session {session_id}: {provider_name}/{model}"
        );
    }

    #[allow(clippy::too_many_arguments)]
    async fn run_tool_loop_inner(
        &self,
        session_id: Uuid,
        user_message: String,
        display_text_override: Option<String>,
        model: Option<String>,
        cancel_token: Option<CancellationToken>,
        has_override_approval: bool,
        approval_callback: Option<ApprovalCallback>,
        has_progress_override: bool,
        progress_callback: Option<ProgressCallback>,
        question_callback: Option<QuestionCallback>,
    ) -> Result<AgentResponse> {
        // Snapshot the manual-switch epoch at turn start. If the user
        // switches provider/model while this turn is in flight, an automatic
        // fallback the turn takes could otherwise stick over their pick. We
        // detect the change AFTER the turn completes (see the restore near
        // the final return) and re-apply the user's pick then — off the
        // completion path, so it can never drop or contaminate the request.
        let start_switch_epoch = self.manual_switch_epoch(session_id);

        // Get or create session
        let session_service = SessionService::new(self.context.clone());
        let session = session_service
            .get_session(session_id)
            .await
            .map_err(|e| AgentError::Database(e.to_string()))?
            .ok_or(AgentError::SessionNotFound(session_id))?;

        // Load conversation context with budget-aware message trimming
        let message_service = MessageService::new(self.context.clone());
        let all_db_messages = message_service
            .list_messages_for_session(session_id)
            .await
            .map_err(|e| AgentError::Database(e.to_string()))?;

        // Resolve model name: explicit caller arg > session's saved model >
        // current provider's default. The session.model fallback is critical
        // for restart-recovery: when the resume task races with TUI session
        // restore, reading provider.default_model() can capture the wrong
        // (pre-swap) provider's model. session.model is read from DB and is
        // always provider-correct.
        // Mutable so the sticky-fallback path can rebind this to the
        // successful fallback's model — otherwise subsequent tool-loop
        // iterations in the same turn rebuild requests with the primary
        // model name pointed at the fallback provider → 400 unknown model.
        let session_provider = self.provider_for_session(session_id);
        let resolved_model = model
            .or_else(|| session.model.clone())
            .unwrap_or_else(|| session_provider.default_model().to_string());
        let provider_default = session_provider.default_model().to_string();
        let supported = session_provider.supported_models();
        let (mut model_name, leaked) =
            guard_cross_provider_model_leak(resolved_model, &provider_default, &supported);
        if let Some(stale) = leaked {
            tracing::warn!(
                "Stale model pin '{}' for session {} is not in active provider '{}' catalogue ({} entries) — \
                 substituting provider default '{}'. \
                 This usually means an earlier sticky-fallback or provider switch left a cross-provider model pinned in session.model.",
                stale,
                session_id,
                session_provider.name(),
                supported.len(),
                provider_default
            );
        }
        let context_window = self.context_limit_for_session(session_id);

        // Load from last compaction point — find the last CONTEXT COMPACTION marker
        // and only load messages from there forward. No arbitrary trimming.
        let mut db_messages = Self::messages_from_last_compaction(all_db_messages);

        // Auto-title: fire a one-shot background LLM call using the current
        // user_message as the seed. Works on ALL channels (TUI, Telegram,
        // Discord, Slack, etc). Issue #118 + #120.
        //
        // Why no `db_message_count >= 1` guard: the previous version only
        // fired from the SECOND user message onward (because db_message_count
        // is taken before the current message is stored). The reporter's
        // sessions all had exactly 1 turn each (one /new, one message, done),
        // so auto-title never ran and every session sat with the default
        // `Telegram: DM <name> (<id>) [chat:<id>]` title forever — looking
        // identical in `/sessions`. We already have `user_message` in scope,
        // so fire on the first turn directly.
        //
        // Why a reset-on-failure path: `mark_auto_title_attempted` runs
        // BEFORE the LLM call to prevent race conditions if the user sends
        // a second message while the title generation is still in flight.
        // But if the LLM call fails (provider down, 5xx, timeout), the flag
        // stays true forever and the session is stuck. The Err arm now
        // resets the flag so the next message retries.
        if !user_message.trim().is_empty()
            && !session.auto_title_attempted
            && session
                .title
                .as_deref()
                .map(|t| t.is_empty() || Self::is_default_channel_title(t))
                .unwrap_or(true)
        {
            let title_provider = self.provider_for_session(session_id);
            let title_model = model_name.clone();
            let title_msg = user_message.chars().take(500).collect::<String>();
            let session_svc = SessionService::new(self.context.clone());
            // Capture the channel BEFORE spawn so the new title can fan
            // out to the TUI/footer the moment it lands in DB. Without
            // this, the footer kept showing "New Chat" after Ctrl+N
            // until the user switched sessions and load_session re-read
            // the row from DB.
            let title_update_tx = self.session_updated_tx.clone();
            // Mark auto_title_attempted BEFORE spawning to prevent race
            // conditions where the next message arrives before the
            // background task completes. The Err arm resets it.
            let _ = session_svc.mark_auto_title_attempted(session_id).await;
            // Capture the old title to preserve channel prefix
            let old_title = session.title.clone().unwrap_or_default();
            tokio::spawn(async move {
                let title_request = LLMRequest::new(
                    title_model,
                    vec![Message::user(format!(
                        "Generate a concise session title (3-7 words) based on this user message. \
                         Return ONLY the title text, nothing else. No quotes, no punctuation at the end.\n\n\
                         Message: {}",
                        title_msg
                    ))],
                );
                match title_provider.complete(title_request).await {
                    Ok(response) => {
                        // Use the thinking-aware extractor — reasoning
                        // models sometimes return ONLY a Thinking block
                        // for short prompts and never produce a Text
                        // block. The old `extract_text_from_response`
                        // returned "" in that case, sessions stayed
                        // stuck on default titles forever (issue #121).
                        let clean_title = Self::extract_title_candidate(&response);
                        if !clean_title.is_empty() {
                            // Preserve channel prefix if it existed
                            let prefix = Self::extract_channel_prefix(&old_title);
                            // Preserve [chat:ID] suffix — critical for session resolution
                            // via find_session_by_title_suffix (issue #115)
                            let chat_suffix = Self::extract_chat_id_suffix(&old_title);
                            let final_title = if prefix.is_empty() {
                                if chat_suffix.is_empty() {
                                    clean_title
                                } else {
                                    format!("{} {}", clean_title, chat_suffix)
                                }
                            } else if chat_suffix.is_empty() {
                                format!("{}{}", prefix, clean_title)
                            } else {
                                format!("{}{} {}", prefix, clean_title, chat_suffix)
                            };
                            match session_svc
                                .update_session_title(session_id, Some(final_title.clone()))
                                .await
                            {
                                Ok(()) => {
                                    if let Some(tx) = title_update_tx.as_ref()
                                        && let Err(e) = tx.send(
                                            crate::brain::agent::ChannelSessionEvent::TitleUpdated(
                                                session_id,
                                                final_title,
                                            ),
                                        )
                                    {
                                        tracing::warn!(
                                            "Auto-title: title written to DB but TUI notify channel \
                                             closed, footer will lag until next session switch: {}",
                                            e
                                        );
                                    }
                                }
                                Err(e) => {
                                    tracing::warn!(
                                        "Auto-title: update_session_title failed for {}: {}",
                                        session_id,
                                        e
                                    );
                                }
                            }
                        } else {
                            // Empty/unusable title — allow the next message
                            // to retry. Same recovery path as the Err arm.
                            let _ = session_svc.reset_auto_title_attempted(session_id).await;
                        }
                    }
                    Err(e) => {
                        tracing::warn!(
                            "Auto-title generation failed for session {}: {} — resetting flag so the next message retries",
                            session_id,
                            e,
                        );
                        let _ = session_svc.reset_auto_title_attempted(session_id).await;
                    }
                }
            });
        }

        // Detect CLI + local provider once (neither changes during the loop).
        //
        // `is_cli_provider` controls TWO unrelated behaviors:
        //   1. Skip local tool execution (CLI runs tools internally)
        //   2. Skip OpenCrabs-side context compaction (CLI persists session)
        //
        // `is_local_provider` relaxes the phantom-tool-call detector so
        // local llama.cpp/MLX models that answer in prose when they should
        // have called a tool get re-prompted, matching what Unsloth Studio
        // does out of the box.
        let (is_cli_provider, cli_owns_context, is_dialagram, is_local_provider) = {
            let p = self.provider_for_session(session_id);
            let base = p.base_url();
            // Detect dialagram by base_url — users add it as a custom
            // provider under any name they choose (typos included),
            // but the proxy URL is always https://www.dialagram.me/...
            let is_dialagram = base
                .map(|u| u.to_lowercase().contains("dialagram.me"))
                .unwrap_or(false);
            let is_local = base
                .map(crate::brain::provider::factory::is_local_base_url)
                .unwrap_or(false);
            (
                p.cli_handles_tools(),
                p.cli_manages_context(),
                is_dialagram,
                is_local,
            )
        };

        // For API providers ONLY: strip persisted `<!-- tools-v2: ... -->` and
        // `<!-- reasoning -->` markers from DB content before loading into the
        // LLM context. These markers exist for TUI replay/cancel-persist, but
        // feeding them back to the model teaches it to echo the JSON tool-result
        // format verbatim in its responses (a closed feedback loop that produces
        // raw JSON dumps and dropped streaming responses for API providers like
        // qwen3.6-plus on OpenRouter).
        //
        // CLI providers MUST keep markers — their DB content drives session
        // resume/replay and the CLI subprocess never sees this content.
        if !is_cli_provider {
            for msg in db_messages.iter_mut() {
                if msg.content.contains("<!--") {
                    msg.content = crate::utils::sanitize::strip_llm_artifacts(&msg.content);
                }
                Self::strip_compaction_banner(&mut msg.content);
            }
        }

        let mut context =
            AgentContext::from_db_messages(session_id, db_messages, context_window as usize);

        // Add system brain if available (count its tokens so context.token_count
        // reflects the full API input from the start — prevents gross undercount
        // that causes the TUI context counter to jump wildly on first calibration)
        if let Some(brain) = &self.default_system_brain {
            // mimo narrates/text-emits tool calls instead of using the
            // structured field; remind it up front (the self-heal + the
            // <tool_call_list> parser are the after-the-fact safety nets).
            let brain = if super::helpers::is_mimo_model(&model_name) {
                format!("{brain}\n\n{}", super::helpers::MIMO_TOOL_CALL_HINT)
            } else {
                brain.clone()
            };
            context.token_count += AgentContext::estimate_tokens(&brain);
            context.system_brain = Some(brain);
        }

        // Emit token count immediately after DB reload so the TUI reflects the
        // real post-compaction value. Without this, the TUI shows the stale
        // pre-compaction count from the previous request until the API responds.
        if let Some(ref cb) = progress_callback {
            cb(session_id, ProgressEvent::TokenCount(context.token_count));
        }

        // Detect user corrections / negative feedback and record automatically.
        // Only fires on real user messages (not system continuations).
        if !user_message.starts_with("[System:")
            && !user_message.starts_with("[SYSTEM:")
            && is_user_correction(&user_message)
        {
            self.record_provider_feedback(
                session_id,
                "user_correction",
                "user_message",
                Some(
                    &display_text_override
                        .as_deref()
                        .unwrap_or(&user_message)
                        .chars()
                        .take(200)
                        .collect::<String>(),
                ),
            );
        }

        // Check for manual /compact before user_message is consumed
        let is_manual_compact = user_message.contains("[SYSTEM: Compact context now.");

        // Build user message — `<<IMG:path>>` markers become text hints; the
        // agent views images via analyze_image (no inline image_url content).
        let user_msg = Self::build_user_message(&user_message);
        context.add_message(user_msg);

        // Save user message to database (text only — images are ephemeral).
        // Skip DB persistence for internal system continuations (restart recovery)
        // — they go to context for the LLM but never appear in chat history.
        // Redact secrets so Bearer tokens, API keys etc. from cron prompts
        // never persist to DB or appear in TUI chat history.
        //
        // When a channel handler supplies `display_text_override`, the DB row
        // (and therefore the TUI chat history) shows that clean text instead
        // of the LLM-context-augmented `user_message`. This keeps Telegram /
        // Discord / Slack / WhatsApp / Trello sessions readable in OpenCrabs
        // — no sender brackets, no reply context, no recent-history dump.
        let is_system_continuation = user_message.starts_with("[System:");
        if !is_system_continuation {
            let raw_for_db = display_text_override.as_deref().unwrap_or(&user_message);
            let safe_message = crate::utils::sanitize::redact_secrets(raw_for_db);
            let _user_db_msg = message_service
                .create_message(session_id, "user".to_string(), safe_message)
                .await
                .map_err(|e| AgentError::Database(e.to_string()))?;
        }

        // Create assistant message placeholder NOW for real-time persistence.
        // We'll append content as we go and update with final tokens at the end.
        let mut assistant_db_msg = message_service
            .create_message(session_id, "assistant".to_string(), String::new())
            .await
            .map_err(|e| AgentError::Database(e.to_string()))?;

        // Manual /compact: force compaction and return summary directly — no second LLM call.
        // The summary already contains next steps and follow-ups, so it IS the response.
        if is_manual_compact {
            match self
                .compact_context(session_id, &mut context, &model_name, None)
                .await
            {
                Ok(summary) => {
                    // Persist compaction marker to DB so restarts load from this point
                    let compaction_marker = format!(
                        "[CONTEXT COMPACTION — The conversation was automatically compacted. \
                         Below is a structured summary of everything before this point.]\n\n{}",
                        summary
                    );
                    message_service
                        .create_message(session_id, "user".to_string(), compaction_marker)
                        .await
                        .map_err(|e| AgentError::Database(e.to_string()))?;

                    // Persist summary as the assistant response
                    message_service
                        .append_content(assistant_db_msg.id, &summary)
                        .await
                        .map_err(|e| AgentError::Database(e.to_string()))?;

                    if let Some(ref cb) = progress_callback {
                        cb(session_id, ProgressEvent::TokenCount(context.token_count));
                    }

                    return Ok(AgentResponse {
                        message_id: assistant_db_msg.id,
                        content: summary,
                        stop_reason: Some(crate::brain::provider::StopReason::EndTurn),
                        usage: crate::brain::provider::TokenUsage {
                            input_tokens: 0,
                            output_tokens: 0,
                            ..Default::default()
                        },
                        context_tokens: context.token_count as u32,
                        tokens_per_second: None,
                        cost: 0.0,
                        model: model_name,
                        provider_name: self.provider_name_for_session(session_id),
                    });
                }
                Err(e) => {
                    tracing::error!("Manual compaction failed: {}", e);
                    let error_msg = format!(
                        "Compaction failed: {}\n\nThis can happen if:\n\
                         - The session has too few messages to summarize\n\
                         - The AI provider returned an error\n\
                         - The database is locked or inaccessible\n\n\
                         Try again, or continue the conversation normally — \
                         auto-compaction will trigger at 65% context usage.",
                        e
                    );
                    message_service
                        .append_content(assistant_db_msg.id, &error_msg)
                        .await
                        .map_err(|e2| AgentError::Database(e2.to_string()))?;

                    return Ok(AgentResponse {
                        message_id: assistant_db_msg.id,
                        content: error_msg,
                        stop_reason: Some(crate::brain::provider::StopReason::EndTurn),
                        usage: crate::brain::provider::TokenUsage {
                            input_tokens: 0,
                            output_tokens: 0,
                            ..Default::default()
                        },
                        context_tokens: context.token_count as u32,
                        tokens_per_second: None,
                        cost: 0.0,
                        model: model_name,
                        provider_name: self.provider_name_for_session(session_id),
                    });
                }
            }
        }

        // CLI providers manage their own context window internally.
        // Keep our tiktoken estimate from DB messages as-is — it's a reasonable
        // approximation. Don't reset to 0, because CLI cache tokens (used for
        // calibration below) are cumulative across internal tool rounds, not
        // the actual context window size.

        // Auto-compact: triggers at >65% usage.
        // Skip ONLY when the CLI manages its own session (qwen-code with
        // --resume). Claude CLI handles tools internally but DOES NOT
        // manage the context we feed it via stdin — we send the full
        // conversation each turn, so we MUST compact for Claude CLI too.
        // The other 3 call sites in this file already use cli_owns_context;
        // this one was using is_cli_provider, which let Claude CLI
        // bypass compaction entirely and inflated the ctx counter past
        // 200% (user report 2026-05-04: 484k/200k = 242%).
        let compaction_result = if cli_owns_context {
            None
        } else {
            self.enforce_context_budget(
                session_id,
                &mut context,
                &model_name,
                cancel_token.as_ref(),
                &progress_callback,
            )
            .await
        };

        if let Some(ref summary) = compaction_result {
            // Persist compaction marker to DB so restarts load from this point
            let compaction_marker = format!(
                "[CONTEXT COMPACTION — The conversation was automatically compacted. \
                 Below is a structured summary of everything before this point.]\n\n{}",
                summary
            );
            if let Err(e) = message_service
                .create_message(session_id, "user".to_string(), compaction_marker)
                .await
            {
                tracing::error!("Failed to persist compaction marker to DB: {}", e);
            }

            let cont_text = super::compaction_prompts::build_continuation(
                super::compaction_prompts::CompactionKind::Regular,
                self.silent_compaction,
                self.auto_approve_tools,
            );
            context.add_message(Message::user(cont_text));
        }

        // Create tool execution context
        let mut tool_context = ToolExecutionContext::new(session_id)
            .with_auto_approve(self.auto_approve_tools)
            .with_working_directory(
                self.working_directory
                    .read()
                    .expect("working_directory lock poisoned")
                    .clone(),
            );
        tool_context.sudo_callback = self.sudo_callback.clone();
        tool_context.ssh_callback = self.ssh_callback.clone();
        tool_context.shared_working_directory = Some(Arc::clone(&self.working_directory));
        tool_context.service_context = Some(self.context.clone());
        tool_context.question_callback = question_callback.clone();

        // Tool execution loop
        let mut iteration = 0;
        // Number of tools that completed SUCCESSFULLY in this turn so
        // far. Drives the post-success exemption in the phantom-tool-call
        // detector below: once the turn has produced at least one real
        // tool result, a subsequent text-only iteration is a completion
        // acknowledgement ("Done.", "Pushed.", "Committed."), not
        // phantom intent. Without this counter the detector mistook
        // every successful turn's wrap-up text for "model narrated
        // without executing", forced retries, eventually rolled the
        // self-heal budget, and produced minute-long loops on already-
        // completed work (logged in user reports as "phantom detected"
        // x8+ after a clean commit+push).
        let mut tool_calls_completed_this_turn: usize = 0;
        // One-shot nudge budget for the empty-analysis case: model ran
        // tool calls (e.g. `gh pr view`) on a user request whose verb
        // signals analysis ("audit the PR") but ended with
        // `finish_reason: stop` and zero text. The user expected an
        // analytical answer that uses the fetched data; the FINISHING
        // A TURN directive earlier in the brain prompt explains both
        // task shapes (side-effect vs. analysis). Once is enough — if
        // the model still won't write analysis after the nudge, the
        // text-completion path takes over and a one-line "Done." is
        // the user-visible behaviour.
        let mut analysis_nudge_used: bool = false;
        let mut total_input_tokens = 0u32;
        let mut total_output_tokens = 0u32;
        let mut total_cache_creation = 0u32;
        let mut total_cache_read = 0u32;
        // Sum of per-iteration active-streaming time, used as the tok/s
        // denominator. Replaces the previous `turn_start.elapsed()`
        // wall-clock which silently halved the displayed rate on every
        // tool-heavy turn by including bash exec / approval waits / DB
        // persistence in the denominator. Per-iteration values come
        // from `LLMResponse.streaming_active_secs` populated by
        // `stream_complete` in helpers.rs.
        let mut total_streaming_active_secs: f64 = 0.0;
        // Last iteration's prompt size, used for the "current context
        // usage" indicator. Distinct from `total_input_tokens` which
        // sums across every iteration for cost/billing. The UI ctx
        // meter must show the LAST call's prompt — summing all
        // iterations inflates by a factor of N and showed 150K for a
        // turn whose final prompt was 22K (2026-04-17 05:55 logs).
        let mut last_iter_input_tokens = 0u32;
        let mut final_response: Option<LLMResponse> = None;
        let mut accumulated_text = String::new(); // Collect text from all iterations (not just final)
        let mut recent_tool_calls: Vec<String> = Vec::new(); // Track tool calls to detect loops
        let mut stream_retry_count = 0u32; // Track consecutive stream drop retries
        const MAX_STREAM_RETRIES: u32 = 3; // Retry up to 3 times on dropped streams
        // Phantom-retry budget per turn. Single-shot proved insufficient —
        // when the model is stuck in a "Let me check…" narration loop, one
        // correction nudges it for one iteration and it drifts right back.
        // Five retries per turn gives the model room to recover from a
        // bumpy start while still capping pathological cases before they
        // chew the whole quota; once the cap is hit we force a sticky
        // fallback to a different provider rather than giving up.
        let mut phantom_retries_used: u32 = 0;
        const MAX_PHANTOM_RETRIES: u32 = 5;
        // Set to true after we have forced a sticky fallback because
        // phantom retries exhausted. Guarantees we only swap once per
        // turn even if the fallback provider is also phantom-prone.
        let mut phantom_sticky_swap_done: bool = false;
        // Bounded retry for the "reasoning-only, no answer" failure mode:
        // MLX Qwen models periodically emit finish_reason=stop after only
        // reasoning_content chunks — zero text, zero tool calls — so the
        // user sees a dropped request. We nudge up to 5 times, escalating
        // the system instruction each round, and if the model STILL refuses
        // to emit visible text we walk the fallback chain (sticky swap) so
        // the turn never silently disappears.
        let mut empty_reasoning_retries: u32 = 0;
        const EMPTY_REASONING_MAX_NUDGES: u32 = 5;
        // Local reasoning models (notably Qwen3.6-35B on MLX) periodically
        // emit an EOS token mid-sentence — the response looks complete from
        // a protocol standpoint (proper finish_reason=stop + usage chunk)
        // but the visible text ends mid-word ("Standard Get I"). One-shot
        // nudge to continue from where they left off.
        let mut truncated_mid_sentence_retry_used: bool = false;
        // One-shot nudge for the browser screenshot-spam pattern detected
        // by the semantic-loop check below the per-iteration tool dispatch.
        // Reset per turn; fires at most once.
        let mut browser_screenshot_loop_nudged: bool = false;
        // Tracks whether the CURRENT iteration is a same-provider continuation
        // requested after a truncated-mid-sentence detection. Reset at the top
        // of every iteration; set true just before `continue;` from the
        // truncation-continue branch. When true, the stream-error fallback
        // path skips cross-provider fallback — switching providers mid-table
        // (e.g. qwen vertical-label format → glm pipe-table format) produces
        // garbled output stitched at the seam. Better to abort the continue
        // and leave the visibly truncated response than fabricate a Frankenstein
        // continuation in a different format.
        let mut current_iter_is_truncation_continue: bool = false;
        let mut rotation_retry_used = false; // Single retry when Qwen rotation yields 0 tools

        // Ordered content segments for CLI providers — tracks text and tool markers
        // in the exact order they stream, so DB persistence preserves interleaving.
        #[derive(Clone)]
        enum CliSegment {
            Text(String),
            Tool(serde_json::Value),
        }
        let cli_segments: std::sync::Arc<std::sync::Mutex<Vec<CliSegment>>> =
            std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));

        // Wrap progress_callback for CLI providers to intercept IntermediateText
        // and ToolCompleted events, preserving their streaming order.
        let progress_callback: Option<ProgressCallback> = if is_cli_provider {
            if let Some(ref original_cb) = progress_callback {
                let orig = original_cb.clone();
                let segs = cli_segments.clone();
                Some(std::sync::Arc::new(
                    move |sid: Uuid, event: ProgressEvent| {
                        match event {
                            ProgressEvent::IntermediateText { ref text, .. }
                                if !text.is_empty() =>
                            {
                                if let Ok(mut acc) = segs.lock() {
                                    acc.push(CliSegment::Text(text.clone()));
                                }
                            }
                            ProgressEvent::ToolCompleted {
                                ref tool_name,
                                ref tool_input,
                                success,
                                ref summary,
                                ..
                            } => {
                                let desc = AgentService::format_tool_summary(
                                    &tool_name.to_lowercase(),
                                    tool_input,
                                );
                                let entry = if summary.is_empty() {
                                    serde_json::json!({"d": desc, "s": success, "i": tool_input})
                                } else {
                                    serde_json::json!({"d": desc, "s": success, "o": summary, "i": tool_input})
                                };
                                if let Ok(mut acc) = segs.lock() {
                                    acc.push(CliSegment::Tool(entry));
                                }
                            }
                            _ => {}
                        }
                        orig(sid, event);
                    },
                ))
            } else {
                None
            }
        } else {
            progress_callback
        };

        loop {
            // Snapshot + reset the truncation-continue marker at the top of
            // every iteration. The branch that sets it does so just before
            // `continue;`, so it's always one-shot — true for exactly the
            // iteration that follows a truncated response, false otherwise.
            let iter_is_truncation_continue = current_iter_is_truncation_continue;
            current_iter_is_truncation_continue = false;
            // Safety: warn every 50 iterations but never hard-stop
            // Loop detection (below) is the real safety net
            if self.max_tool_iterations > 0 && iteration >= self.max_tool_iterations {
                tracing::warn!(
                    "Tool iteration {} exceeded configured max of {} — continuing (loop detection is active)",
                    iteration,
                    self.max_tool_iterations
                );
            }
            // Check for cancellation
            if let Some(ref token) = cancel_token
                && token.is_cancelled()
            {
                tracing::warn!(
                    "🛑 Tool loop cancelled at iteration {} (cancel_token fired). \
                     Accumulated text: {} chars, tool iterations so far: {}",
                    iteration,
                    accumulated_text.len(),
                    iteration,
                );
                break;
            }

            iteration += 1;

            // Emit thinking progress
            if let Some(ref cb) = progress_callback {
                cb(session_id, ProgressEvent::Thinking);
            }

            // Enforce 65% budget before every API call. Skip ONLY when the
            // CLI manages its own session (claude-cli with --resume). Qwen
            // is spawned cold every turn so we MUST compact for it.
            if let Some(ref summary) = if cli_owns_context {
                None
            } else {
                self.enforce_context_budget(
                    session_id,
                    &mut context,
                    &model_name,
                    cancel_token.as_ref(),
                    &progress_callback,
                )
                .await
            } {
                // Persist compaction marker to DB so restarts load from this point
                let compaction_marker = format!(
                    "[CONTEXT COMPACTION — The conversation was automatically compacted. \
                     Below is a structured summary of everything before this point.]\n\n{}",
                    summary
                );
                if let Err(e) = message_service
                    .create_message(session_id, "user".to_string(), compaction_marker)
                    .await
                {
                    tracing::error!("Failed to persist mid-loop compaction marker to DB: {}", e);
                }

                let cont_text = super::compaction_prompts::build_continuation(
                    super::compaction_prompts::CompactionKind::MidLoop,
                    self.silent_compaction,
                    self.auto_approve_tools,
                );
                context.add_message(Message::user(cont_text));
            }

            // Build LLM request with tools if available
            let mut request = LLMRequest::new(model_name.clone(), context.messages.clone())
                .with_max_tokens(self.max_tokens);
            request.working_directory =
                Some(self.get_working_directory().to_string_lossy().to_string());
            request.session_id = Some(session_id);

            if let Some(system) = &context.system_brain {
                request = request.with_system(system.clone());
            }

            // Add tools if registry has any
            let tool_count = self.tool_registry.count();
            tracing::debug!("Tool registry contains {} tools", tool_count);
            if tool_count > 0 {
                let tool_defs = self.tool_schemas_for_session(session_id);
                tracing::debug!("Adding {} tool definitions to request", tool_defs.len());
                request = request.with_tools(tool_defs);
            } else {
                tracing::warn!("No tools registered in tool registry!");
            }

            // CLI providers: pass queue callback so stream_complete can check
            // for queued user messages at tool boundaries mid-stream.
            let queued_buf = tokio::sync::Mutex::new(None);

            // Send to provider via streaming — retry once after emergency compaction if prompt is too long
            let (mut response, reasoning_text): (LLMResponse, Option<String>) = match self
                .stream_complete(
                    session_id,
                    request,
                    cancel_token.as_ref(),
                    progress_callback.as_ref(),
                    if is_cli_provider {
                        self.message_queue_callback.as_ref()
                    } else {
                        None
                    },
                    if is_cli_provider {
                        Some(&queued_buf)
                    } else {
                        None
                    },
                    false,
                )
                .await
            {
                Ok(resp) => {
                    // Primary succeeded on first try (no retry / no
                    // fallback rescue needed). Reset the consecutive-
                    // failure streak so a future hiccup starts fresh
                    // at 1 instead of inheriting a count from an
                    // unrelated earlier outage. Without this, a
                    // primary that hit 3 transient failures days ago
                    // would stick the fallback on the NEXT failure
                    // even though it's been working flawlessly since.
                    self.reset_primary_failure_streak(session_id);
                    resp
                }
                Err(ref e)
                    if e.to_string().contains("prompt is too long")
                        || e.to_string().contains("too many tokens")
                        || e.to_string().contains("Argument list too long")
                        || matches!(
                            e,
                            crate::brain::provider::ProviderError::ContextLengthExceeded(_)
                        ) =>
                {
                    tracing::warn!("Prompt too long for provider — emergency compaction");
                    self.record_provider_feedback(
                        session_id,
                        "context_compaction",
                        &model_name,
                        Some(&format!("tokens={}", context.token_count)),
                    );

                    // Pre-truncate to 85% of max so compact_context() can actually run.
                    // For 200k models: ~170k. For custom providers: scales proportionally.
                    const PRE_TRUNCATE_PCT: f64 = 0.85;
                    let pre_truncate_target =
                        (context.max_tokens as f64 * PRE_TRUNCATE_PCT).max(16_000.0) as usize;
                    if context.token_count > pre_truncate_target {
                        tracing::warn!(
                            "Context too large for compaction ({} tokens) — pre-truncating to {}K",
                            context.token_count,
                            pre_truncate_target / 1000
                        );
                        context.hard_truncate_to(pre_truncate_target);
                        tracing::info!(
                            "Pre-truncated to {} messages ({} tokens) — now attempting compaction",
                            context.messages.len(),
                            context.token_count
                        );
                    }

                    match self
                        .compact_context(
                            session_id,
                            &mut context,
                            &model_name,
                            cancel_token.as_ref(),
                        )
                        .await
                    {
                        Ok(summary) => {
                            // Persist compaction marker to DB so restarts load from this point
                            let compaction_marker = format!(
                                "[CONTEXT COMPACTION — The conversation was automatically compacted. \
                                 Below is a structured summary of everything before this point.]\n\n{}",
                                summary
                            );
                            if let Err(e) = message_service
                                .create_message(session_id, "user".to_string(), compaction_marker)
                                .await
                            {
                                tracing::error!(
                                    "Failed to persist emergency compaction marker to DB: {}",
                                    e
                                );
                            }

                            let cont_text = super::compaction_prompts::build_continuation(
                                super::compaction_prompts::CompactionKind::Emergency,
                                self.silent_compaction,
                                self.auto_approve_tools,
                            );
                            context.add_message(Message::user(cont_text));

                            // Notify user about emergency compaction
                            if let Some(ref cb) = progress_callback {
                                cb(
                                    session_id,
                                    ProgressEvent::SelfHealingAlert {
                                        message: "Emergency compaction: context was too large for the provider. Conversation has been compacted automatically.".to_string(),
                                    },
                                );
                            }
                        }
                        Err(compact_err) => {
                            tracing::error!(
                                "Emergency compaction also failed: {} — falling back to hard truncation",
                                compact_err
                            );

                            // Hard truncate: keep last 12 message pairs (24 messages).
                            // Full conversation is in the DB — agent can search_session for older context.
                            const KEEP_MESSAGES: usize = 24;
                            let total = context.messages.len();
                            if total > KEEP_MESSAGES {
                                let dropped = total - KEEP_MESSAGES;
                                context.messages = context.messages.split_off(dropped);
                                tracing::warn!(
                                    "Hard truncated context: dropped {} messages, kept {}",
                                    dropped,
                                    context.messages.len()
                                );
                            }

                            // Insert truncation marker so the agent knows context was lost
                            let truncation_marker = format!(
                                "[CONTEXT TRUNCATION — The conversation was too large for the provider \
                                 and compaction failed. The {} oldest messages were dropped. \
                                 The full conversation history is still in the database — use the \
                                 search_session tool if you need to recall earlier context. \
                                 Continue from where you left off.]",
                                total.saturating_sub(KEEP_MESSAGES)
                            );
                            context
                                .messages
                                .insert(0, Message::user(truncation_marker.clone()));

                            // Persist truncation marker to DB
                            if let Err(e) = message_service
                                .create_message(session_id, "user".to_string(), truncation_marker)
                                .await
                            {
                                tracing::error!("Failed to persist truncation marker: {}", e);
                            }

                            // Notify user about hard truncation
                            if let Some(ref cb) = progress_callback {
                                cb(
                                    session_id,
                                    ProgressEvent::SelfHealingAlert {
                                        message: format!(
                                            "Hard truncation: compaction failed, {} oldest messages were dropped. Full history is still in the database.",
                                            total.saturating_sub(KEEP_MESSAGES)
                                        ),
                                    },
                                );
                            }

                            // Re-estimate token count after truncation
                            context.token_count = context
                                .messages
                                .iter()
                                .map(|m| {
                                    m.content
                                        .iter()
                                        .map(|b| match b {
                                            ContentBlock::Text { text } => {
                                                crate::brain::tokenizer::count_tokens(text)
                                            }
                                            ContentBlock::ToolUse { input, .. } => {
                                                crate::brain::tokenizer::count_tokens(
                                                    &input.to_string(),
                                                )
                                            }
                                            ContentBlock::ToolResult { content, .. } => {
                                                crate::brain::tokenizer::count_tokens(content)
                                            }
                                            ContentBlock::Thinking { thinking, .. } => {
                                                crate::brain::tokenizer::count_tokens(thinking)
                                            }
                                            ContentBlock::Image { .. } => 1000,
                                        })
                                        .sum::<usize>()
                                })
                                .sum();
                        }
                    }

                    // Rebuild request with compacted context
                    let mut retry_req =
                        LLMRequest::new(model_name.clone(), context.messages.clone())
                            .with_max_tokens(self.max_tokens);
                    retry_req.working_directory =
                        Some(self.get_working_directory().to_string_lossy().to_string());
                    retry_req.session_id = Some(session_id);
                    if let Some(system) = &context.system_brain {
                        retry_req = retry_req.with_system(system.clone());
                    }
                    if self.tool_registry.count() > 0 {
                        retry_req = retry_req.with_tools(self.tool_schemas_for_session(session_id));
                    }
                    self.stream_complete(
                        session_id,
                        retry_req,
                        cancel_token.as_ref(),
                        progress_callback.as_ref(),
                        if is_cli_provider {
                            self.message_queue_callback.as_ref()
                        } else {
                            None
                        },
                        if is_cli_provider {
                            Some(&queued_buf)
                        } else {
                            None
                        },
                        false,
                    )
                    .await
                    .map_err(AgentError::Provider)?
                }
                Err(e)
                    if matches!(
                        &e,
                        crate::brain::provider::ProviderError::RateLimitExceeded(_)
                    ) || matches!(
                        &e,
                        crate::brain::provider::ProviderError::StreamError(s) if s.contains("rate limit") || s.contains("hit your limit")
                    ) || matches!(
                        &e,
                        crate::brain::provider::ProviderError::ApiError { status, .. }
                            if *status == 401 || *status == 403 || *status == 402
                    ) || matches!(&e, crate::brain::provider::ProviderError::InvalidApiKey) =>
                {
                    // 401/403 auth failures and missing-key errors are
                    // unrecoverable on the current provider (retry with
                    // the same bad key is pointless) but perfectly
                    // fallback-able: the next provider in the chain has
                    // its own key and may work fine. The 2026-04-18
                    // 13:52 log caught opencodeiolo returning 401
                    // "Missing API key" and falling straight through
                    // to the terminal AgentError — no retry, no
                    // fallback, user saw a raw error string in
                    // Telegram. Treat it like a rate-limit: skip
                    // in-place retry, walk the fallback chain.
                    // Distinguish three failure flavours that all arrive here:
                    //   • genuine auth: 401/403 with an auth-ish error_type or
                    //     InvalidApiKey. Key is bad — walk fallback.
                    //   • model rejection disguised as 401: some proxies
                    //     (opencode.ai/zen) return 401 with
                    //     `error_type: "ModelError"` when the key is valid
                    //     but the requested model isn't in their allowlist.
                    //     Reporting "Auth error" here misled the user into
                    //     thinking keys were wrong.
                    //   • rate/account limits: 429 / RateLimitExceeded.
                    let is_model_mismatch = e.is_model_unsupported();
                    let is_payment_required = matches!(
                        &e,
                        crate::brain::provider::ProviderError::ApiError { status, .. }
                            if *status == 402
                    );
                    let (is_auth, reason) = if is_model_mismatch {
                        (false, "model_unsupported")
                    } else if is_payment_required {
                        // 402 means the upstream account has run out of
                        // credit / hit a hard billing cap. Same fallback
                        // path as auth/rate, but reported distinctly so
                        // the user knows it's a quota issue, not a key
                        // misconfiguration or temporary throttle.
                        (false, "payment_required")
                    } else if matches!(
                        &e,
                        crate::brain::provider::ProviderError::ApiError { status, .. }
                            if *status == 401 || *status == 403
                    ) || matches!(
                        &e,
                        crate::brain::provider::ProviderError::InvalidApiKey
                    ) {
                        (true, "auth_error")
                    } else {
                        (false, "rate_limit_exceeded")
                    };
                    let flavour_label = if is_model_mismatch {
                        "Model not supported by provider"
                    } else if is_payment_required {
                        "Quota/payment limit"
                    } else if is_auth {
                        "Auth error"
                    } else {
                        "Rate/account limit"
                    };
                    tracing::warn!(
                        "{} hit ({}) — checking for fallback provider",
                        flavour_label,
                        e
                    );

                    // Resolve the session's CURRENT primary provider name
                    // for the alert AND feedback dimension — never just the model.
                    // A user can have opencode, opencode2, opencode3 … all routing to
                    // the same underlying model name. "Rate limit on
                    // 'claude-sonnet-4-6'" hides WHICH subscription got
                    // rate-limited. The session's provider is the truth
                    // source (global `self.provider` may differ after
                    // per-session swaps).
                    let primary_from_name = self.provider_name_for_session(session_id);
                    let primary_from_model = model_name.clone();
                    // Record the ACTUAL provider/model pair that will be sent
                    // (not the requested one). helpers.rs remaps mismatched
                    // pairs silently — RSI must reflect what actually hit the
                    // wire so entries like "dialagram/zhipu" (where "zhipu"
                    // is a provider name that leaked into the model slot from
                    // a reversed cron config) never appear in feedback.
                    let actual_model = {
                        let p = self.provider_for_session(session_id);
                        let supported = p.supported_models();
                        if !supported.is_empty() && !supported.iter().any(|m| m == &model_name) {
                            p.default_model().to_string()
                        } else {
                            model_name.clone()
                        }
                    };
                    let provider_model_dim = format!("{}/{}", primary_from_name, actual_model);

                    self.record_provider_feedback(
                        session_id,
                        "provider_error",
                        &provider_model_dim,
                        Some(reason),
                    );

                    if let Some(ref cb) = progress_callback {
                        let prefix = if is_model_mismatch {
                            format!(
                                "Model '{}' not supported by '{}'",
                                model_name, primary_from_name
                            )
                        } else if is_payment_required {
                            format!(
                                "Quota/payment limit on '{}/{}'",
                                primary_from_name, model_name
                            )
                        } else if is_auth {
                            format!("Auth error on '{}/{}'", primary_from_name, model_name)
                        } else {
                            format!(
                                "Rate limit on '{}/{}' (retried 3x in-place)",
                                primary_from_name, model_name
                            )
                        };
                        let message = if self.fallback_providers.is_empty() {
                            format!("{} — no fallback providers configured.", prefix)
                        } else {
                            format!("{} — walking fallback chain...", prefix)
                        };
                        cb(session_id, ProgressEvent::SelfHealingAlert { message });
                    }

                    // Walk the entire fallback chain, skipping the SESSION's
                    // active provider (not the global default — a per-session
                    // swap may already have this session on opencode2 while
                    // `self.provider` still holds opencode).
                    let active_name = self.provider_name_for_session(session_id);
                    let candidates: Vec<_> = self
                        .fallback_providers
                        .iter()
                        .filter(|p| p.name() != active_name)
                        .collect();

                    if candidates.is_empty() {
                        return Err(AgentError::Provider(e));
                    }

                    let mut last_err = e;
                    // stream_complete returns (LLMResponse, Option<String>);
                    // we also need fb_name / fb_model alongside for the
                    // ProviderSwitched event emitted once on success.
                    let mut succeeded: Option<(
                        (crate::brain::provider::LLMResponse, Option<String>),
                        String,
                        String,
                    )> = None;
                    for fallback in &candidates {
                        let fb_name = fallback.name().to_string();
                        let fb_model = fallback.default_model().to_string();
                        tracing::info!(
                            "Trying fallback provider '{}' (model '{}')",
                            fb_name,
                            fb_model
                        );
                        if let Some(ref cb) = progress_callback {
                            cb(
                                session_id,
                                ProgressEvent::SelfHealingAlert {
                                    message: format!(
                                        "Trying fallback '{}/{}'...",
                                        fb_name, fb_model
                                    ),
                                },
                            );
                        }

                        let mut fb_req =
                            LLMRequest::new(fb_model.clone(), context.messages.clone())
                                .with_max_tokens(self.max_tokens);
                        fb_req.working_directory =
                            Some(self.get_working_directory().to_string_lossy().to_string());
                        fb_req.session_id = Some(session_id);
                        if let Some(system) = &context.system_brain {
                            fb_req = fb_req.with_system(system.clone());
                        }
                        if self.tool_registry.count() > 0 {
                            fb_req = fb_req.with_tools(self.tool_schemas_for_session(session_id));
                        }

                        // STICKY FALLBACK (rate-limit / auth path): swap
                        // session provider to the fallback, and on success
                        // DON'T restore. Rate limits from subscription
                        // quotas can last hours; without stickiness every
                        // subsequent turn hits the same 429, walks the
                        // chain again, and bounces back — the user sees a
                        // warning every turn and never settles on the
                        // working provider.
                        //
                        // Guard pattern: if the await errors OR is
                        // cancelled (future dropped mid-stream), Drop fires
                        // and restores the original. Avoids the nightmare
                        // where session_providers points at fallback but
                        // session.model in DB is still primary → 400
                        // "unknown model" on next turn. On success we
                        // disable the guard (set original=None) so the
                        // swap STICKS, then emit ProviderSwitched to
                        // persist the pairing to DB via state.rs:2205.
                        let original_provider = self.provider_for_session(session_id);
                        self.swap_provider_for_session(
                            session_id,
                            (*fallback).clone(),
                            (*fallback)
                                .active_subprovider_model()
                                .unwrap_or_else(|| (*fallback).default_model().to_string()),
                        );
                        let mut restore_guard = FallbackProviderGuard {
                            service: self,
                            session_id,
                            original: Some(original_provider),
                        };
                        let fb_result = self
                            .stream_complete(
                                session_id,
                                fb_req,
                                cancel_token.as_ref(),
                                progress_callback.as_ref(),
                                None,
                                None,
                                false,
                            )
                            .await;
                        match fb_result {
                            Ok(resp) => {
                                // Disable restore — the swap must stick.
                                restore_guard.original = None;
                                drop(restore_guard);
                                succeeded = Some((resp, fb_name, fb_model));
                                break;
                            }
                            Err(fb_err) => {
                                // Guard's Drop restores the original so
                                // the next candidate iteration starts
                                // clean.
                                drop(restore_guard);
                                tracing::warn!(
                                    "Fallback '{}' failed: {} — trying next",
                                    fallback.name(),
                                    fb_err
                                );
                                last_err = fb_err;
                            }
                        }
                    }
                    match succeeded {
                        Some((resp, fb_name, fb_model)) => {
                            // Emit ProviderSwitched so the TUI persists the
                            // swap to the session DB (session.provider_name
                            // and session.model). state.rs:2205 picks this
                            // up and calls session_service.update_session
                            // so the NEXT turn resolves model_name from
                            // the fallback, not the rate-limited primary.
                            if let Some(ref cb) = progress_callback {
                                let reason = if is_model_mismatch {
                                    "model_unsupported".to_string()
                                } else if is_auth {
                                    "auth_error".to_string()
                                } else {
                                    "rate_limit".to_string()
                                };
                                cb(
                                    session_id,
                                    ProgressEvent::SelfHealingAlert {
                                        message: format!(
                                            "Sticky fallback → '{}/{}' (was '{}/{}', {}). \
                                             Pinned until you change via /models.",
                                            fb_name,
                                            fb_model,
                                            primary_from_name,
                                            primary_from_model,
                                            reason
                                        ),
                                    },
                                );
                                cb(
                                    session_id,
                                    ProgressEvent::ProviderSwitched {
                                        from_name: primary_from_name.clone(),
                                        from_model: primary_from_model.clone(),
                                        to_name: fb_name.clone(),
                                        to_model: fb_model.clone(),
                                        reason,
                                    },
                                );
                            }
                            // Persist the locked {provider, model} pair to
                            // DB independently of the progress callback —
                            // channel handlers (Slack/Telegram/Discord/
                            // WhatsApp) historically dropped ProviderSwitched
                            // on the floor, leaving DB stale while
                            // session_providers[sid] was already swapped to
                            // the fallback. Stale DB → next turn's
                            // sync_provider_for_session "restored" memory
                            // from the wrong row → cross-pair leak.
                            self.persist_sticky_pair(session_id, fb_name.clone(), fb_model.clone());
                            // Update the local model_name binding so any
                            // further iterations in THIS turn build
                            // requests with the fallback's model
                            // (otherwise the next tool-loop iteration
                            // would send the primary's model name to the
                            // fallback provider → 400).
                            model_name = fb_model;
                            resp
                        }
                        None => {
                            tracing::error!(
                                "All {} fallback providers exhausted",
                                candidates.len()
                            );
                            return Err(AgentError::Provider(last_err));
                        }
                    }
                }
                Err(e)
                    if matches!(
                        &e,
                        crate::brain::provider::ProviderError::StreamError(_)
                            | crate::brain::provider::ProviderError::Timeout(_)
                    ) && !e.to_string().contains("rate limit")
                        && !e.to_string().contains("hit your limit") =>
                {
                    // Timeout covers the new handshake-timeout path: a wedged
                    // local server that accepts TCP but never emits headers.
                    // Funnel it into the same 3-retry + fallback chain as
                    // mid-stream StreamError so the user sees recovery
                    // activity instead of a dead turn.
                    let err_msg = e.to_string();
                    tracing::warn!("Mid-stream error: {} — retrying up to 3 times", err_msg);
                    let primary_from_name = self.provider_name_for_session(session_id);
                    let actual_model = {
                        let p = self.provider_for_session(session_id);
                        let supported = p.supported_models();
                        if !supported.is_empty() && !supported.iter().any(|m| m == &model_name) {
                            p.default_model().to_string()
                        } else {
                            model_name.clone()
                        }
                    };
                    let provider_model_dim = format!("{}/{}", primary_from_name, actual_model);
                    self.record_provider_feedback(
                        session_id,
                        "provider_error",
                        &provider_model_dim,
                        Some(&err_msg),
                    );

                    let mut last_err = e;
                    let mut succeeded = None;

                    for attempt in 1..=3 {
                        tracing::info!("Stream retry attempt {}/3 after: {}", attempt, last_err);

                        // Brief backoff: 500ms, 1s, 2s
                        tokio::time::sleep(tokio::time::Duration::from_millis(
                            500 * (1 << (attempt - 1)),
                        ))
                        .await;

                        // Rebuild request
                        let mut retry_req =
                            LLMRequest::new(model_name.clone(), context.messages.clone())
                                .with_max_tokens(self.max_tokens);
                        retry_req.working_directory =
                            Some(self.get_working_directory().to_string_lossy().to_string());
                        retry_req.session_id = Some(session_id);
                        if let Some(system) = &context.system_brain {
                            retry_req = retry_req.with_system(system.clone());
                        }
                        if self.tool_registry.count() > 0 {
                            retry_req =
                                retry_req.with_tools(self.tool_schemas_for_session(session_id));
                        }

                        match self
                            .stream_complete(
                                session_id,
                                retry_req,
                                cancel_token.as_ref(),
                                progress_callback.as_ref(),
                                if is_cli_provider {
                                    self.message_queue_callback.as_ref()
                                } else {
                                    None
                                },
                                if is_cli_provider {
                                    Some(&queued_buf)
                                } else {
                                    None
                                },
                                false,
                            )
                            .await
                        {
                            Ok(resp) => {
                                tracing::info!("Stream retry {}/3 succeeded", attempt);
                                succeeded = Some(resp);
                                break;
                            }
                            Err(retry_err) => {
                                tracing::warn!("Stream retry {}/3 failed: {}", attempt, retry_err);
                                last_err = retry_err;
                            }
                        }
                    }

                    if let Some(resp) = succeeded {
                        resp
                    } else if iter_is_truncation_continue {
                        // This iteration is the same-provider continuation we
                        // asked for after a truncated-mid-sentence response.
                        // Falling back to a different provider here produces
                        // garbled output: providers don't share format style,
                        // so the continuation gets stitched in a different
                        // shape (e.g. qwen vertical-label labels → glm pipe-
                        // table syntax) and the result is unreadable. Better
                        // to abort the continue and leave the visibly cut-off
                        // response than fabricate a Frankenstein answer.
                        tracing::warn!(
                            "All 3 stream retries failed during a truncation-continue — \
                             aborting continuation rather than falling back to a different \
                             provider (would cause format drift)."
                        );
                        if let Some(ref cb) = progress_callback {
                            let active_name = self.provider_name_for_session(session_id);
                            let err_snippet: String =
                                last_err.to_string().chars().take(120).collect();
                            cb(
                                session_id,
                                ProgressEvent::SelfHealingAlert {
                                    message: format!(
                                        "Continuation request to '{}/{}' failed after 3 retries: {}. \
                                         Leaving the previous response truncated.",
                                        active_name, model_name, err_snippet,
                                    ),
                                },
                            );
                        }
                        return Err(AgentError::Provider(last_err));
                    } else {
                        // All retries failed — try fallback provider
                        tracing::warn!(
                            "All 3 stream retries failed — checking for fallback provider"
                        );

                        if let Some(ref cb) = progress_callback {
                            let active_name = self.provider_name_for_session(session_id);
                            // Surface the underlying error so users (and the
                            // maintainer when users report) can tell whether
                            // it was a timeout, TLS issue, 5xx, etc., instead
                            // of a generic "Stream error" banner.
                            let err_snippet: String =
                                last_err.to_string().chars().take(120).collect();
                            cb(
                                session_id,
                                ProgressEvent::SelfHealingAlert {
                                    message: format!(
                                        "Stream error on '{}/{}' after 3 retries: {}. {}",
                                        active_name,
                                        model_name,
                                        err_snippet,
                                        if self.has_fallback_provider() {
                                            "Switching to fallback provider..."
                                        } else {
                                            "No fallback provider configured."
                                        }
                                    ),
                                },
                            );
                        }

                        // Walk the entire fallback chain, skipping the active provider.
                        let stream_active_name = self
                            .provider
                            .read()
                            .ok()
                            .map(|p| p.name().to_string())
                            .unwrap_or_default();
                        let stream_candidates: Vec<_> = self
                            .fallback_providers
                            .iter()
                            .filter(|p| p.name() != stream_active_name)
                            .collect();

                        if stream_candidates.is_empty() {
                            return Err(AgentError::Provider(last_err));
                        }

                        let mut stream_succeeded = None;
                        for fallback in &stream_candidates {
                            let fb_name = fallback.name().to_string();
                            let fb_model = fallback.default_model().to_string();
                            tracing::info!(
                                "Stream fallback trying '{}' (model '{}')",
                                fb_name,
                                fb_model
                            );
                            // Tell the user which fallback we're attempting —
                            // the earlier "Switching to fallback provider..."
                            // banner named the origin but not the destination,
                            // so after 3 retries users saw a provider swap
                            // with no hint what they're now talking to.
                            if let Some(ref cb) = progress_callback {
                                cb(
                                    session_id,
                                    ProgressEvent::SelfHealingAlert {
                                        message: format!(
                                            "Trying fallback '{}/{}'...",
                                            fb_name, fb_model
                                        ),
                                    },
                                );
                            }

                            let mut fb_req =
                                LLMRequest::new(fb_model.clone(), context.messages.clone())
                                    .with_max_tokens(self.max_tokens);
                            fb_req.working_directory =
                                Some(self.get_working_directory().to_string_lossy().to_string());
                            fb_req.session_id = Some(session_id);
                            if let Some(system) = &context.system_brain {
                                fb_req = fb_req.with_system(system.clone());
                            }
                            if self.tool_registry.count() > 0 {
                                fb_req =
                                    fb_req.with_tools(self.tool_schemas_for_session(session_id));
                            }

                            // Swap only this session's provider for the
                            // stream-fallback attempt; guard ensures restore
                            // runs even if the outer future is cancelled
                            // mid-await (see FallbackProviderGuard doc).
                            let original_provider = self.provider_for_session(session_id);
                            self.swap_provider_for_session(
                                session_id,
                                (*fallback).clone(),
                                (*fallback)
                                    .active_subprovider_model()
                                    .unwrap_or_else(|| (*fallback).default_model().to_string()),
                            );
                            let mut restore_guard = FallbackProviderGuard {
                                service: self,
                                session_id,
                                original: Some(original_provider),
                            };
                            let fb_result = self
                                .stream_complete(
                                    session_id,
                                    fb_req,
                                    cancel_token.as_ref(),
                                    progress_callback.as_ref(),
                                    None,
                                    None,
                                    false,
                                )
                                .await;
                            match fb_result {
                                Ok(resp) => {
                                    // Streak gate: only stick the fallback as
                                    // the session's persistent provider after
                                    // STICKY_FALLBACK_THRESHOLD consecutive
                                    // rescues. Most primary outages are
                                    // transient (network blip, model warm-up,
                                    // brief 5xx), so making the first rescue
                                    // sticky meant a 5-second hiccup
                                    // permanently demoted the primary until
                                    // the user noticed and reset via /models.
                                    // 4-rescues-in-a-row matches the user's
                                    // intent: "if fallback rescues 3 times
                                    // consecutively successfully, the 4th it
                                    // sticks".
                                    let streak = self.bump_primary_failure_streak(session_id);
                                    let sticky = streak >= STICKY_FALLBACK_THRESHOLD;
                                    if sticky {
                                        // Disable restore — the swap stays.
                                        restore_guard.original = None;
                                    }
                                    // Either way: dropping the guard either
                                    // restores the primary (non-sticky) or is
                                    // a no-op (sticky). Done either way before
                                    // emitting events so the post-event state
                                    // matches what callers see.
                                    drop(restore_guard);
                                    stream_succeeded = Some(resp);
                                    if let Some(ref cb) = progress_callback {
                                        let primary_from_name =
                                            self.provider_name_for_session(session_id);
                                        cb(
                                            session_id,
                                            ProgressEvent::SelfHealingAlert {
                                                message: if sticky {
                                                    format!(
                                                        "Stream error → switched to {}/{} (sticky after {} consecutive rescues)",
                                                        fb_name, fb_model, streak
                                                    )
                                                } else {
                                                    format!(
                                                        "Stream error → rescued by {}/{} ({}/{} consecutive; primary will be tried again next turn)",
                                                        fb_name,
                                                        fb_model,
                                                        streak,
                                                        STICKY_FALLBACK_THRESHOLD
                                                    )
                                                },
                                            },
                                        );
                                        if sticky {
                                            cb(
                                                session_id,
                                                ProgressEvent::ProviderSwitched {
                                                    from_name: primary_from_name,
                                                    from_model: self
                                                        .provider_model_for_session(session_id),
                                                    to_name: fb_name.to_string(),
                                                    to_model: fb_model.to_string(),
                                                    reason: "stream_error".to_string(),
                                                },
                                            );
                                        }
                                    }
                                    // Persist the locked pair to DB only on
                                    // sticky — a transient rescue should not
                                    // mutate persistent session state.
                                    if sticky {
                                        self.persist_sticky_pair(
                                            session_id,
                                            fb_name.to_string(),
                                            fb_model.to_string(),
                                        );
                                    }
                                    break;
                                }
                                Err(fb_err) => {
                                    // Guard's Drop restores the original so the
                                    // next candidate iteration starts clean.
                                    drop(restore_guard);
                                    tracing::warn!(
                                        "Stream fallback '{}' failed: {} — trying next",
                                        fb_name,
                                        fb_err
                                    );
                                    last_err = fb_err;
                                }
                            }
                        }
                        match stream_succeeded {
                            Some(resp) => resp,
                            None => {
                                tracing::error!(
                                    "All {} stream fallback providers exhausted",
                                    stream_candidates.len()
                                );
                                return Err(AgentError::Provider(last_err));
                            }
                        }
                    }
                }
                Err(e) if matches!(&e, crate::brain::provider::ProviderError::ApiError { status, .. } if *status >= 500 && *status < 600) =>
                {
                    // 5xx upstream errors (500/502/503/504) are transient — retry
                    // up to 3 times with backoff before falling back, same as
                    // StreamError/Timeout. Without this the user sees a hard
                    // failure on every blip from the provider.
                    let err_msg = e.to_string();
                    tracing::warn!("Upstream 5xx error: {} — retrying up to 3 times", err_msg);
                    let primary_from_name = self.provider_name_for_session(session_id);
                    let actual_model = {
                        let p = self.provider_for_session(session_id);
                        let supported = p.supported_models();
                        if !supported.is_empty() && !supported.iter().any(|m| m == &model_name) {
                            p.default_model().to_string()
                        } else {
                            model_name.clone()
                        }
                    };
                    let provider_model_dim = format!("{}/{}", primary_from_name, actual_model);
                    self.record_provider_feedback(
                        session_id,
                        "provider_error",
                        &provider_model_dim,
                        Some(&err_msg),
                    );

                    let mut last_err = e;
                    let mut succeeded = None;

                    for attempt in 1..=3 {
                        tracing::info!("5xx retry attempt {}/3 after: {}", attempt, last_err);

                        tokio::time::sleep(tokio::time::Duration::from_millis(
                            500 * (1 << (attempt - 1)),
                        ))
                        .await;

                        let mut retry_req =
                            LLMRequest::new(model_name.clone(), context.messages.clone())
                                .with_max_tokens(self.max_tokens);
                        retry_req.working_directory =
                            Some(self.get_working_directory().to_string_lossy().to_string());
                        retry_req.session_id = Some(session_id);
                        if let Some(system) = &context.system_brain {
                            retry_req = retry_req.with_system(system.clone());
                        }
                        if self.tool_registry.count() > 0 {
                            retry_req =
                                retry_req.with_tools(self.tool_schemas_for_session(session_id));
                        }

                        match self
                            .stream_complete(
                                session_id,
                                retry_req,
                                cancel_token.as_ref(),
                                progress_callback.as_ref(),
                                if is_cli_provider {
                                    self.message_queue_callback.as_ref()
                                } else {
                                    None
                                },
                                if is_cli_provider {
                                    Some(&queued_buf)
                                } else {
                                    None
                                },
                                false,
                            )
                            .await
                        {
                            Ok(resp) => {
                                tracing::info!("5xx retry {}/3 succeeded", attempt);
                                succeeded = Some(resp);
                                break;
                            }
                            Err(retry_err) => {
                                tracing::warn!("5xx retry {}/3 failed: {}", attempt, retry_err);
                                last_err = retry_err;
                            }
                        }
                    }

                    if let Some(resp) = succeeded {
                        resp
                    } else {
                        tracing::warn!("All 3 5xx retries failed — checking for fallback provider");

                        if let Some(ref cb) = progress_callback {
                            let active_name = self.provider_name_for_session(session_id);
                            cb(
                                session_id,
                                ProgressEvent::SelfHealingAlert {
                                    message: format!(
                                        "5xx error on '{}/{}' after 3 retries. {}",
                                        active_name,
                                        model_name,
                                        if self.has_fallback_provider() {
                                            "Switching to fallback provider..."
                                        } else {
                                            "No fallback provider configured."
                                        }
                                    ),
                                },
                            );
                        }

                        let stream_active_name = self
                            .provider
                            .read()
                            .ok()
                            .map(|p| p.name().to_string())
                            .unwrap_or_default();
                        let stream_candidates: Vec<_> = self
                            .fallback_providers
                            .iter()
                            .filter(|p| p.name() != stream_active_name)
                            .collect();

                        if stream_candidates.is_empty() {
                            return Err(AgentError::Provider(last_err));
                        }

                        let mut stream_succeeded = None;
                        for fallback in &stream_candidates {
                            let fb_name = fallback.name().to_string();
                            let fb_model = fallback.default_model().to_string();
                            tracing::info!("5xx fallback trying '{}/{}'", fb_name, fb_model);
                            if let Some(ref cb) = progress_callback {
                                cb(
                                    session_id,
                                    ProgressEvent::SelfHealingAlert {
                                        message: format!(
                                            "Trying fallback '{}/{}'...",
                                            fb_name, fb_model
                                        ),
                                    },
                                );
                            }

                            let mut fb_req =
                                LLMRequest::new(fb_model.clone(), context.messages.clone())
                                    .with_max_tokens(self.max_tokens);
                            fb_req.working_directory =
                                Some(self.get_working_directory().to_string_lossy().to_string());
                            fb_req.session_id = Some(session_id);
                            if let Some(system) = &context.system_brain {
                                fb_req = fb_req.with_system(system.clone());
                            }
                            if self.tool_registry.count() > 0 {
                                fb_req =
                                    fb_req.with_tools(self.tool_schemas_for_session(session_id));
                            }

                            match self
                                .stream_complete(
                                    session_id,
                                    fb_req,
                                    cancel_token.as_ref(),
                                    progress_callback.as_ref(),
                                    if is_cli_provider {
                                        self.message_queue_callback.as_ref()
                                    } else {
                                        None
                                    },
                                    if is_cli_provider {
                                        Some(&queued_buf)
                                    } else {
                                        None
                                    },
                                    false,
                                )
                                .await
                            {
                                Ok(resp) => {
                                    tracing::info!(
                                        "5xx fallback succeeded with '{}/{}'",
                                        fb_name,
                                        fb_model
                                    );
                                    stream_succeeded = Some(resp);
                                    // Same streak gate as the stream-error
                                    // path: only swap the session's provider
                                    // permanently after the threshold.
                                    let streak = self.bump_primary_failure_streak(session_id);
                                    let sticky = streak >= STICKY_FALLBACK_THRESHOLD;
                                    if let Some(ref cb) = progress_callback {
                                        let primary_from_name =
                                            self.provider_name_for_session(session_id);
                                        cb(
                                            session_id,
                                            ProgressEvent::SelfHealingAlert {
                                                message: if sticky {
                                                    format!(
                                                        "5xx error → switched to {}/{} (sticky after {} consecutive rescues)",
                                                        fb_name, fb_model, streak
                                                    )
                                                } else {
                                                    format!(
                                                        "5xx error → rescued by {}/{} ({}/{} consecutive; primary will be tried again next turn)",
                                                        fb_name,
                                                        fb_model,
                                                        streak,
                                                        STICKY_FALLBACK_THRESHOLD
                                                    )
                                                },
                                            },
                                        );
                                        if sticky {
                                            cb(
                                                session_id,
                                                ProgressEvent::ProviderSwitched {
                                                    from_name: primary_from_name,
                                                    from_model: self
                                                        .provider_model_for_session(session_id),
                                                    to_name: fb_name.clone(),
                                                    to_model: fb_model.clone(),
                                                    reason: "5xx_error".to_string(),
                                                },
                                            );
                                        }
                                    }
                                    if sticky {
                                        // Sticky: swap session provider AND
                                        // persist so subsequent iterations +
                                        // future turns use the fallback.
                                        self.swap_provider_for_session(
                                            session_id,
                                            (*fallback).clone(),
                                            (*fallback).active_subprovider_model().unwrap_or_else(
                                                || (*fallback).default_model().to_string(),
                                            ),
                                        );
                                        self.persist_sticky_pair(
                                            session_id,
                                            fb_name.clone(),
                                            fb_model.clone(),
                                        );
                                    }
                                    break;
                                }
                                Err(fb_err) => {
                                    tracing::warn!(
                                        "5xx fallback '{}/{}' failed: {}",
                                        fb_name,
                                        fb_model,
                                        fb_err
                                    );
                                }
                            }
                        }

                        if let Some(resp) = stream_succeeded {
                            resp
                        } else {
                            tracing::error!(
                                "All {} 5xx fallback providers exhausted",
                                stream_candidates.len()
                            );
                            return Err(AgentError::Provider(last_err));
                        }
                    }
                }
                Err(e) => {
                    // Any non-5xx provider error (405, 404, 400, etc.) —
                    // walk the entire fallback chain before giving up.
                    let err_msg = e.to_string();
                    let active_name = self.provider_name_for_session(session_id);
                    tracing::warn!(
                        "Provider error from {}: {} — walking fallback chain",
                        active_name,
                        err_msg
                    );
                    self.record_provider_feedback(
                        session_id,
                        "provider_error",
                        &format!("{}/{}", active_name, model_name),
                        Some(&err_msg),
                    );

                    let fallback_candidates: Vec<_> = self
                        .fallback_providers
                        .iter()
                        .filter(|p| p.name() != active_name)
                        .collect();

                    if fallback_candidates.is_empty() {
                        tracing::warn!(
                            "No fallback providers configured for {} error — \
                             user should configure a fallback chain",
                            active_name
                        );
                        if let Some(ref cb) = progress_callback {
                            cb(
                                session_id,
                                ProgressEvent::SelfHealingAlert {
                                    message: "No fallback provider available. \
                                         Configure one with /onboard:provider"
                                        .to_string(),
                                },
                            );
                        }
                        return Err(AgentError::Provider(e));
                    }

                    let mut last_err = e;
                    let mut succeeded = None;

                    for fallback in &fallback_candidates {
                        let fb_name = fallback.name().to_string();
                        let fb_model = fallback.default_model().to_string();
                        tracing::info!(
                            "Fallback trying '{}/{}' for provider error",
                            fb_name,
                            fb_model
                        );
                        if let Some(ref cb) = progress_callback {
                            cb(
                                session_id,
                                ProgressEvent::SelfHealingAlert {
                                    message: format!(
                                        "Trying fallback '{}/{}'...",
                                        fb_name, fb_model
                                    ),
                                },
                            );
                        }

                        let mut fb_req =
                            LLMRequest::new(fb_model.clone(), context.messages.clone())
                                .with_max_tokens(self.max_tokens);
                        fb_req.working_directory =
                            Some(self.get_working_directory().to_string_lossy().to_string());
                        fb_req.session_id = Some(session_id);
                        if let Some(system) = &context.system_brain {
                            fb_req = fb_req.with_system(system.clone());
                        }
                        if self.tool_registry.count() > 0 {
                            fb_req = fb_req.with_tools(self.tool_schemas_for_session(session_id));
                        }

                        // Swap provider for this session so stream_complete
                        // uses the fallback
                        self.swap_provider_for_session(
                            session_id,
                            (*fallback).clone(),
                            (*fallback)
                                .active_subprovider_model()
                                .unwrap_or_else(|| (*fallback).default_model().to_string()),
                        );

                        match self
                            .stream_complete(
                                session_id,
                                fb_req,
                                cancel_token.as_ref(),
                                progress_callback.as_ref(),
                                if is_cli_provider {
                                    self.message_queue_callback.as_ref()
                                } else {
                                    None
                                },
                                if is_cli_provider {
                                    Some(&queued_buf)
                                } else {
                                    None
                                },
                                false,
                            )
                            .await
                        {
                            Ok(resp) => {
                                tracing::info!("Fallback '{}/{}' succeeded", fb_name, fb_model);
                                succeeded = Some(resp);
                                break;
                            }
                            Err(fb_err) => {
                                tracing::warn!(
                                    "Fallback '{}/{}' also failed: {}",
                                    fb_name,
                                    fb_model,
                                    fb_err
                                );
                                last_err = fb_err;
                            }
                        }
                    }

                    if let Some(resp) = succeeded {
                        resp
                    } else {
                        tracing::error!(
                            "All {} fallback providers exhausted",
                            fallback_candidates.len()
                        );
                        return Err(AgentError::Provider(last_err));
                    }
                }
            };

            // Surface any in-place retries the provider performed (connection
            // blip, 5xx, rate limit) so the user SEES the resilience working
            // instead of an apparent instant jump to fallback. Drained once
            // per iteration; the FallbackProvider aggregates retries from the
            // primary and every fallback tried this turn.
            if let Some(ref cb) = progress_callback {
                for (attempt, max, reason) in
                    self.provider_for_session(session_id).take_retry_notices()
                {
                    cb(
                        session_id,
                        ProgressEvent::RetryAttempt {
                            attempt,
                            max,
                            reason,
                        },
                    );
                }
            }

            // Surface any sticky-fallback swap that the FallbackProvider
            // performed during this turn so the user sees which provider/model
            // is now active. Fires at most once per swap.
            let rotated_this_iteration = self.provider_for_session(session_id).take_swap_event();
            if let Some(ref swap) = rotated_this_iteration {
                let reason = if swap.reason.is_empty() {
                    "unavailable".to_string()
                } else {
                    swap.reason.clone()
                };
                if let Some(ref cb) = progress_callback {
                    cb(
                        session_id,
                        ProgressEvent::SelfHealingAlert {
                            message: format!(
                                "Switched to {}/{} — {}/{} {}",
                                swap.to_name,
                                swap.to_model,
                                swap.from_name,
                                swap.from_model,
                                reason
                            ),
                        },
                    );
                    // Structured follow-up so UIs can update the session footer
                    // without parsing the alert text above.
                    cb(
                        session_id,
                        ProgressEvent::ProviderSwitched {
                            from_name: swap.from_name.clone(),
                            from_model: swap.from_model.clone(),
                            to_name: swap.to_name.clone(),
                            to_model: swap.to_model.clone(),
                            reason,
                        },
                    );
                }
                // Persist the locked pair to DB even when no progress
                // callback is wired (e.g. a2a, RSI, subagent paths) and
                // independently of whether the consuming UI handles
                // ProviderSwitched.
                self.persist_sticky_pair(session_id, swap.to_name.clone(), swap.to_model.clone());
            }

            // CLI providers return "Prompt is too long" as a successful response
            // with is_error=true in the content — detect and re-route to the
            // same emergency compaction path used for Err cases above.
            let is_cli_too_long = is_cli_provider
                && response.content.iter().any(|b| {
                    if let ContentBlock::Text { text } = b {
                        text.trim().starts_with("Prompt is too long")
                            || text.contains("prompt is too long")
                    } else {
                        false
                    }
                });

            if is_cli_too_long {
                tracing::warn!(
                    "CLI returned 'Prompt is too long' as content — triggering emergency compaction"
                );
                // Emergency pre-truncate: 85% of max (scales with custom providers)
                let too_long_pre_truncate =
                    (context.max_tokens as f64 * 0.85).max(16_000.0) as usize;
                if context.token_count > too_long_pre_truncate {
                    context.hard_truncate_to(too_long_pre_truncate);
                }
                match self
                    .compact_context(session_id, &mut context, &model_name, cancel_token.as_ref())
                    .await
                {
                    Ok(summary) => {
                        let compaction_marker = format!(
                            "[CONTEXT COMPACTION — The conversation was automatically compacted. \
                             Below is a structured summary of everything before this point.]\n\n{}",
                            summary
                        );
                        let _ = message_service
                            .create_message(session_id, "user".to_string(), compaction_marker)
                            .await;
                    }
                    Err(e) => {
                        tracing::error!(
                            "Emergency compaction also failed: {} — hard truncating",
                            e
                        );
                        const KEEP_MESSAGES: usize = 24;
                        let total = context.messages.len();
                        if total > KEEP_MESSAGES {
                            context.messages.drain(..total - KEEP_MESSAGES);
                        }
                    }
                }
                // Emit updated token count so TUI reflects post-compaction value.
                if let Some(ref cb) = progress_callback {
                    cb(session_id, ProgressEvent::TokenCount(context.token_count));
                }
                // Re-run the loop iteration with the compacted context
                continue;
            }

            // Track token usage — fall back to tiktoken estimate when provider
            // doesn't report usage (e.g. MiniMax streaming ignores include_usage,
            // some MLX streaming paths drop the final usage chunk).
            //
            // The fallback must match the server's `prompt_tokens` semantic:
            // messages + system prompt + tool schemas. `base_context_tokens()`
            // already sums system + tool schemas, so we add
            // `context.token_count` (messages) on top. Previously we only
            // added tool tokens and dropped the 20k system prompt baseline,
            // producing a ~20k undercount that made the UI ctx counter
            // display 7k when the real prompt was 23k+ (post-compaction).
            let call_input_tokens = if response.usage.input_tokens > 0 {
                // Real-time data only: use whatever the provider
                // reported. No local-tokenizer calibration, no learned
                // ratio. The ctx footer reads `response.context_tokens`
                // downstream and shows the user the exact same number
                // the API just told us about.
                response.usage.input_tokens
            } else {
                let baseline = self.base_context_tokens();
                let estimate = context.token_count as u32 + baseline;
                tracing::debug!(
                    "Provider reported 0 input tokens, using tiktoken estimate: {} ({} msg + {} baseline (system + tool schemas))",
                    estimate,
                    context.token_count,
                    baseline
                );
                estimate
            };
            total_input_tokens += call_input_tokens;
            last_iter_input_tokens = call_input_tokens;
            total_output_tokens += response.usage.output_tokens;
            if let Some(secs) = response.streaming_active_secs {
                total_streaming_active_secs += secs;
            }
            // Use billing fields (cumulative across CLI rounds) when available
            total_cache_creation += if response.usage.billing_cache_creation > 0 {
                response.usage.billing_cache_creation
            } else {
                response.usage.cache_creation_tokens
            };
            total_cache_read += if response.usage.billing_cache_read > 0 {
                response.usage.billing_cache_read
            } else {
                response.usage.cache_read_tokens
            };

            // Calibrate context token count from the provider's reported usage.
            //
            // Claude CLI handles caching internally — its reported cache_read /
            // cache_creation tokens reflect Claude's own cached system prompt +
            // tool schemas + accumulated session state, NOT the conversation
            // OpenCrabs sent. Adding those to context_input() inflates the
            // counter past the model's window (e.g. 484k/200k = 242%) on every
            // turn, which then triggers spurious auto-compaction that drops
            // the in-flight request. We manage the context we send; Claude
            // manages its own cache. Trust the local tiktoken estimate that
            // already reflects what we sent in `request.messages`.
            //
            // Other CLI providers (qwen-code) re-spawn cold each turn — their
            // reported context_input() IS what we sent and is calibration-worthy.
            let is_claude_cli = self.provider_for_session(session_id).name() == "claude-cli";
            if is_cli_provider && !is_claude_cli {
                let cli_context = response.usage.context_input() as usize;
                if cli_context > 0 {
                    // Sanity guard: if the CLI's reported context is more
                    // than 10× the local tiktoken estimate AND the estimate
                    // is non-trivial, the CLI is almost certainly reporting
                    // a cumulative-across-rounds figure rather than the
                    // last round's prompt size. Don't trust it — keep the
                    // local estimate so the ctx % display stays sane.
                    let estimate = context.token_count;
                    if estimate >= 1000 && cli_context > estimate.saturating_mul(10) {
                        tracing::warn!(
                            "CLI context calibration REJECTED: {} → {} ({}× estimate, \
                             likely cumulative/inflated; keeping local estimate). \
                             Provider: {}, model: {}",
                            estimate,
                            cli_context,
                            cli_context / estimate.max(1),
                            self.provider_for_session(session_id).name(),
                            model_name,
                        );
                    } else {
                        tracing::info!(
                            "CLI context calibration: {} → {} (from per-call cache tokens)",
                            context.token_count,
                            cli_context,
                        );
                        context.token_count = cli_context;
                    }
                }
            } else if is_claude_cli {
                // Local estimate stays authoritative for Claude CLI — already
                // computed from `request.messages`, no API calibration needed.
                tracing::debug!(
                    "Claude CLI: keeping local estimate {} (reported context_input={} \
                     ignored — represents Claude's internal cache, not our sent context)",
                    context.token_count,
                    response.usage.context_input(),
                );
            } else {
                let api_input = response.usage.input_tokens as usize;
                // API input_tokens includes system prompt + tool schemas + messages.
                // Subtract both to get the real message-only token count.
                let overhead = self.base_context_tokens() as usize;
                let real_message_tokens = api_input.saturating_sub(overhead);
                let min_sane = 100;
                let max_drop_ratio = 0.2;
                let min_after_drop = (context.token_count as f64 * max_drop_ratio) as usize;
                if real_message_tokens >= min_sane && real_message_tokens >= min_after_drop {
                    let drift = (context.token_count as f64 - real_message_tokens as f64).abs();
                    if drift > 5000.0 {
                        // Sanity guard against a provider/proxy that OVER-REPORTS usage.
                        // tiktoken and any real model tokenizer agree within ~2× for normal
                        // text, so a reported input more than 2× the real content size
                        // (system + tool schemas + messages) is the endpoint inflating the
                        // count — observed as a flat ~20k additive overhead on EVERY call
                        // from one fallback endpoint (zhipu "coding"). Trusting it blows up
                        // the ctx counter and the billed-cost display. Keep the local
                        // estimate instead, and make the anomaly visible. Mirrors the
                        // cumulative-inflation guard already on the CLI calibration path.
                        let tool_tokens = self.actual_tool_schema_tokens();
                        let expected = context.token_count + tool_tokens;
                        if is_implausible_token_report(
                            context.token_count,
                            tool_tokens,
                            real_message_tokens,
                        ) {
                            tracing::warn!(
                                "Token usage REJECTED: provider '{}' reported {} input tokens, but \
                                 the real content is ~{} ({} system+messages + {} tool schemas) — \
                                 {}× over. Endpoint is over-reporting; keeping local estimate {} so \
                                 the ctx counter and cost stay accurate.",
                                self.provider_for_session(session_id).name(),
                                api_input,
                                expected,
                                context.token_count,
                                tool_tokens,
                                real_message_tokens / expected.max(1),
                                context.token_count,
                            );
                        } else {
                            tracing::info!(
                                "Token calibration: estimated {} → API actual {} (drift: {:.0})",
                                context.token_count,
                                real_message_tokens,
                                drift,
                            );
                            context.token_count = real_message_tokens;
                        }
                    }
                } else if real_message_tokens > 0 && real_message_tokens < min_sane {
                    tracing::warn!(
                        "Token calibration skipped: api_input={}, overhead={}, result={} (below sanity threshold)",
                        api_input,
                        overhead,
                        real_message_tokens,
                    );
                }
            }
            // Fire real-time token count update after every API response
            if let Some(ref cb) = progress_callback {
                cb(session_id, ProgressEvent::TokenCount(context.token_count));
            }
            // When a channel override is active, also fire to the service-level callback
            // so the TUI ctx display stays in sync with channel interactions.
            if has_progress_override && let Some(ref cb) = self.progress_callback {
                cb(session_id, ProgressEvent::TokenCount(context.token_count));
            }

            // Post-calibration compaction check. Skip ONLY when the CLI
            // owns its session (claude-cli with --resume). Qwen is spawned
            // cold every turn so we MUST compact for it.
            if let Some(ref summary) = if cli_owns_context {
                None
            } else {
                self.enforce_context_budget(
                    session_id,
                    &mut context,
                    &model_name,
                    cancel_token.as_ref(),
                    &progress_callback,
                )
                .await
            } {
                let compaction_marker = format!(
                    "[CONTEXT COMPACTION — The conversation was automatically compacted \
                     after token calibration revealed high context usage.]\n\n{}",
                    summary
                );
                if let Err(e) = message_service
                    .create_message(session_id, "user".to_string(), compaction_marker)
                    .await
                {
                    tracing::error!(
                        "Failed to persist post-calibration compaction marker: {}",
                        e
                    );
                }
                context.add_message(Message::user(
                    "[SYSTEM: Context was auto-compacted after calibration. \
                     Review the summary above. The \"IMMEDIATE TASK\" section tells you \
                     exactly what to do next. Continue that task immediately. \
                     Do NOT start a new topic or deviate to unrelated work.]"
                        .to_string(),
                ));
            }

            // --- CANCEL CHECK BEFORE STREAM DROP RETRY ---
            // If the user cancelled during streaming, don't retry — save partial text and break.
            if response.stop_reason.is_none()
                && let Some(ref token) = cancel_token
                && token.is_cancelled()
            {
                if is_cli_provider {
                    // CLI providers: persist interleaved text + tool markers from
                    // streaming events. These were accumulated by the wrapped callback.
                    let mut cancel_content = String::new();

                    // Reasoning
                    if let Some(ref reasoning) = reasoning_text
                        && !reasoning.trim().is_empty()
                    {
                        cancel_content.push_str(&format!(
                            "<!-- reasoning -->\n{}\n<!-- /reasoning -->\n\n",
                            reasoning
                        ));
                    }

                    // Build interleaved content from ordered segments
                    let segments: Vec<CliSegment> = cli_segments
                        .lock()
                        .map(|mut s| s.drain(..).collect())
                        .unwrap_or_default();
                    let mut pending_tools: Vec<serde_json::Value> = Vec::new();
                    for seg in segments {
                        match seg {
                            CliSegment::Text(text) => {
                                // Flush pending tools before text
                                if !pending_tools.is_empty() {
                                    let marker = format!(
                                        "\n<!-- tools-v2: {} -->\n",
                                        serde_json::to_string(&pending_tools).unwrap_or_default()
                                    );
                                    cancel_content.push_str(&marker);
                                    accumulated_text.push_str(&marker);
                                    pending_tools.clear();
                                }
                                cancel_content.push_str(&format!("{}\n\n", text));
                                if !accumulated_text.is_empty() {
                                    accumulated_text.push_str("\n\n");
                                }
                                accumulated_text.push_str(&text);
                            }
                            CliSegment::Tool(entry) => {
                                pending_tools.push(entry);
                            }
                        }
                    }
                    // Flush trailing tools
                    if !pending_tools.is_empty() {
                        let marker = format!(
                            "\n<!-- tools-v2: {} -->\n",
                            serde_json::to_string(&pending_tools).unwrap_or_default()
                        );
                        cancel_content.push_str(&marker);
                        accumulated_text.push_str(&marker);
                    }

                    // Also extract any text from the partial response not yet
                    // emitted as IntermediateText (trailing text after last tool)
                    for block in &response.content {
                        if let ContentBlock::Text { text } = block
                            && !text.trim().is_empty()
                        {
                            // Only append if not already covered by segments
                            if cancel_content.is_empty() || !cancel_content.contains(text.trim()) {
                                cancel_content.push_str(&format!("{}\n\n", text));
                                if !accumulated_text.is_empty() {
                                    accumulated_text.push_str("\n\n");
                                }
                                accumulated_text.push_str(text);
                            }
                        }
                    }

                    // Single atomic write
                    if !cancel_content.is_empty() {
                        let _ = message_service
                            .append_content(assistant_db_msg.id, &cancel_content)
                            .await;
                    }
                } else {
                    // Non-CLI: persist partial reasoning + text from response blocks
                    // as a single append so the `<!-- reasoning -->` marker stays
                    // attached to its iteration's text (same chronological-layout
                    // contract as the regular per-iteration persist below).
                    let mut cancel_content = String::new();
                    if let Some(ref reasoning) = reasoning_text
                        && !reasoning.trim().is_empty()
                    {
                        cancel_content.push_str(&format!(
                            "<!-- reasoning -->\n{}\n<!-- /reasoning -->\n\n",
                            reasoning
                        ));
                    }
                    for block in &response.content {
                        if let ContentBlock::Text { text } = block
                            && !text.trim().is_empty()
                        {
                            if !accumulated_text.is_empty() {
                                accumulated_text.push_str("\n\n");
                            }
                            accumulated_text.push_str(text);
                            cancel_content.push_str(&format!("{}\n\n", text));
                        }
                    }
                    if !cancel_content.is_empty() {
                        let _ = message_service
                            .append_content(assistant_db_msg.id, &cancel_content)
                            .await;
                    }
                }
                tracing::info!(
                    "Stream cancelled by user — saving partial text ({} chars)",
                    accumulated_text.len()
                );
                break;
            }

            // --- STREAM DROP DETECTION ---
            // If stop_reason is None, the stream ended without [DONE]/MessageStop.
            // This means a network interruption, provider timeout, or dropped connection.
            // The response may contain partial/corrupt data. Retry instead of proceeding
            // with garbage that silently drops the task.
            if response.stop_reason.is_none() {
                if stream_retry_count < MAX_STREAM_RETRIES {
                    stream_retry_count += 1;
                    tracing::warn!(
                        "🔄 Stream dropped without completion (no stop_reason) at iteration {}. \
                         Retrying ({}/{}) — partial content discarded.",
                        iteration,
                        stream_retry_count,
                        MAX_STREAM_RETRIES,
                    );
                    // Emit transient retry notification
                    if let Some(ref cb) = progress_callback {
                        cb(
                            session_id,
                            ProgressEvent::RetryAttempt {
                                attempt: stream_retry_count,
                                max: MAX_STREAM_RETRIES,
                                reason: "stream dropped".to_string(),
                            },
                        );
                    }
                    // Subtract the tokens we just counted — they'll be re-counted on retry
                    total_input_tokens -= response.usage.input_tokens;
                    total_output_tokens -= response.usage.output_tokens;
                    total_cache_creation =
                        total_cache_creation.saturating_sub(response.usage.cache_creation_tokens);
                    total_cache_read =
                        total_cache_read.saturating_sub(response.usage.cache_read_tokens);
                    // Don't increment iteration — this is a retry, not a new turn
                    iteration -= 1;
                    continue;
                } else {
                    let drop_msg = format!(
                        "Provider stream dropped {} times consecutively. \
                         The request could not be completed. \
                         Check logs at ~/.opencrabs/logs/ for details.",
                        MAX_STREAM_RETRIES,
                    );
                    tracing::error!(
                        "🚨 {} Content blocks: {}, stop_reason: None",
                        drop_msg,
                        response.content.len(),
                    );

                    // Record as feedback for RSI analysis
                    self.record_provider_feedback(
                        session_id,
                        "stream_drop",
                        &model_name,
                        Some(&format!(
                            "retries={}, content_blocks={}, provider={}",
                            MAX_STREAM_RETRIES,
                            response.content.len(),
                            self.provider_for_session(session_id).name(),
                        )),
                    );

                    // Try to fallback to next provider before giving up
                    let fallback_reason =
                        format!("stream dropped {} times with 0 content", MAX_STREAM_RETRIES,);
                    if self
                        .provider_for_session(session_id)
                        .force_next_fallback(&fallback_reason)
                    {
                        tracing::info!(
                            "🔄 Fallback triggered after stream drops — retrying with next provider"
                        );
                        // Emit self-heal alert so user sees the fallback in TUI
                        if let Some(ref cb) = progress_callback {
                            cb(
                                session_id,
                                ProgressEvent::SelfHealingAlert {
                                    message: format!(
                                        "Stream dropped {} times — switching to fallback provider",
                                        MAX_STREAM_RETRIES,
                                    ),
                                },
                            );
                        }
                        // Reset and retry with the new provider
                        stream_retry_count = 0;
                        total_input_tokens -= response.usage.input_tokens;
                        total_output_tokens -= response.usage.output_tokens;
                        total_cache_creation = total_cache_creation
                            .saturating_sub(response.usage.cache_creation_tokens);
                        total_cache_read =
                            total_cache_read.saturating_sub(response.usage.cache_read_tokens);
                        iteration -= 1;
                        continue;
                    }

                    // No fallback available — inject error and accept partial
                    if response.content.iter().all(
                        |b| !matches!(b, ContentBlock::Text { text } if !text.trim().is_empty()),
                    ) {
                        response.content.push(ContentBlock::Text {
                            text: format!("⚠️ {}", drop_msg),
                        });
                    }
                    stream_retry_count = 0;
                }
            } else {
                // Successful stream completion — reset retry counter
                stream_retry_count = 0;
            }

            // Separate text blocks and tool use blocks from the response
            tracing::debug!("Response has {} content blocks", response.content.len());

            // ── Gaslighting refusal strip ───────────────────────────────
            // Some providers (notably dialagram qwen-thinking) emit canned
            // "I can't analyze this image / tool isn't available" refusals
            // even though the tools ARE available globally. The detector
            // is narrow enough (first-person refusal opening + image
            // context, OR exact phrase from known quirks) that we can
            // strip unconditionally without false-positive risk.
            if is_dialagram {
                let mut stripped_bytes = 0usize;
                let mut stripped_preview = String::new();
                response.content.retain_mut(|b| match b {
                    ContentBlock::Text { text } => {
                        // First try stripping just a leading preamble so
                        // we keep any legitimate draft that follows the
                        // gaslighting opener in the same block.
                        if let Some(remainder) =
                            super::gaslighting::strip_gaslighting_preamble(text)
                        {
                            let removed = text.len().saturating_sub(remainder.len());
                            stripped_bytes += removed;
                            if stripped_preview.is_empty() {
                                stripped_preview = text.chars().take(80).collect::<String>();
                            }
                            if remainder.trim().is_empty() {
                                return false;
                            }
                            *text = remainder;
                            return true;
                        }
                        // Fallback: whole-block match (small pure refusals)
                        if super::gaslighting::is_gaslighting_preamble(text) {
                            stripped_bytes += text.len();
                            if stripped_preview.is_empty() {
                                stripped_preview = text.chars().take(80).collect::<String>();
                            }
                            return false;
                        }
                        true
                    }
                    _ => true,
                });
                if stripped_bytes > 0 {
                    let had_tool_use = response
                        .content
                        .iter()
                        .any(|b| matches!(b, ContentBlock::ToolUse { .. }));
                    tracing::warn!(
                        "[GASLIGHT_STRIP] dropped {} bytes of refusal (had_tool_use={}) — preview: {:?}",
                        stripped_bytes,
                        had_tool_use,
                        stripped_preview
                    );
                    // Wipe the TUI's in-progress streaming buffer so the
                    // lie doesn't stay on screen.
                    if let Some(ref cb) = progress_callback {
                        cb(
                            session_id,
                            ProgressEvent::StripStreamedContent {
                                bytes: stripped_bytes,
                                reason: format!(
                                    "gaslighting refusal ({} bytes) stripped (had_tool_use={})",
                                    stripped_bytes, had_tool_use
                                ),
                            },
                        );
                    }
                }
            }

            let mut iteration_text = String::new();
            let mut tool_uses: Vec<(String, String, Value)> = Vec::new();

            for (i, block) in response.content.iter().enumerate() {
                match block {
                    ContentBlock::Text { text } => {
                        tracing::debug!(
                            "Block {}: Text ({}...)",
                            i,
                            &text.chars().take(50).collect::<String>()
                        );
                        if !text.trim().is_empty() {
                            if !iteration_text.is_empty() {
                                iteration_text.push_str("\n\n");
                            }
                            iteration_text.push_str(text);
                        }
                    }
                    ContentBlock::ToolUse { id, name, input } => {
                        // GRANULAR LOG: Tool call received from provider
                        let input_keys: Vec<_> = input
                            .as_object()
                            .map(|o| o.keys().cloned().collect())
                            .unwrap_or_default();
                        tracing::info!(
                            "[TOOL_EXEC] 📥 Tool call received: name={}, id={}, input_keys={:?}",
                            name,
                            id,
                            input_keys
                        );

                        // Check for empty/Invalid input — only warn when the
                        // tool actually has required parameters. Tools like
                        // browser_screenshot accept zero args (selector is
                        // optional) and call validly with `{}`; logging an
                        // ERROR there is pure noise and showed up in logs as
                        // 4+ false positives per browser session.
                        if input.as_object().map(|o| o.is_empty()).unwrap_or(true) {
                            let has_required = self
                                .tool_registry
                                .get(name.as_str())
                                .map(|t| {
                                    t.input_schema()
                                        .get("required")
                                        .and_then(|r| r.as_array())
                                        .map(|a| !a.is_empty())
                                        .unwrap_or(false)
                                })
                                .unwrap_or(true);
                            if has_required {
                                tracing::error!(
                                    "[TOOL_EXEC] ⚠️ Tool '{}' received empty input — tool call will fail",
                                    name
                                );
                            } else {
                                tracing::debug!(
                                    "[TOOL_EXEC] Tool '{}' called with empty input (schema has no required fields, this is fine)",
                                    name
                                );
                            }
                        }

                        // Normalize hallucinated tool names: some providers send
                        // "Plan: complete_task" instead of tool="plan" + operation="complete_task".
                        let (norm_name, norm_input) =
                            Self::normalize_tool_call(name.clone(), input.clone());

                        tool_uses.push((id.clone(), norm_name, norm_input));
                    }
                    _ => {
                        tracing::debug!("Block {}: Other content block", i);
                    }
                }
            }

            // ── Strip echoed markup ──────────────────────────────────────
            // The LLM echoes or invents HTML comment markers from context:
            // <!-- tools-v2: ... -->, <!-- lens -->, <!-- /tools-v2>, etc.
            // Strip ALL HTML comments from iteration text to prevent any
            // from leaking into Telegram/channel output or the TUI.
            if iteration_text.contains("<!--") {
                iteration_text = Self::strip_html_comments(&iteration_text);
            }

            // ── XML tool-call recovery ──────────────────────────────────
            // MiniMax (and some other providers) sometimes emit tool calls as
            // XML in the content instead of using the API's tool_calls field.
            // Parse them into real tool_uses AND inject into response.content
            // so the context has matching ToolUse blocks for ToolResult messages.
            //
            // CRITICAL: Only strip XML blocks that were SUCCESSFULLY parsed as
            // valid tool calls. If the model is just talking ABOUT XML tags in
            // prose (e.g. release notes), parsing finds no valid JSON inside
            // the tags and we leave the text untouched.
            if Self::has_xml_tool_block(&iteration_text) {
                let parsed = Self::parse_xml_tool_calls(&iteration_text);
                if !parsed.is_empty() {
                    tracing::info!(
                        "Recovered {} XML tool call(s) from content text",
                        parsed.len()
                    );
                    for (name, input) in parsed {
                        let synthetic_id = format!("xml-{}", uuid::Uuid::new_v4().simple());
                        tool_uses.push((synthetic_id.clone(), name.clone(), input.clone()));
                        response.content.push(ContentBlock::ToolUse {
                            id: synthetic_id,
                            name,
                            input,
                        });
                    }
                    // Only strip after successful parse — prose mentions are left alone
                    iteration_text = Self::strip_xml_tool_calls(&iteration_text);
                }
            }

            // ── DB persistence ──────────────────────────────────────────
            // CLI providers: build interleaved content from ordered segments
            // (text + tool markers in streaming order) for a single atomic write.
            // This preserves the text→tools→text sequence seen during live streaming
            // and survives Esc×2 cancel + restart.
            if is_cli_provider {
                let mut cli_content = String::new();

                // CLI providers (opencode, claude, qwen-cli) maintain their own
                // conversation history server-side via session IDs, so writing
                // reasoning markers into our DB content doesn't feed back into
                // the model's context — no leak risk like the non-CLI path.
                if let Some(ref reasoning) = reasoning_text
                    && !reasoning.trim().is_empty()
                {
                    cli_content.push_str(&format!(
                        "<!-- reasoning -->\n{}\n<!-- /reasoning -->\n\n",
                        reasoning
                    ));
                }

                // Interleaved text + tool markers from streaming events
                let segments: Vec<CliSegment> = cli_segments
                    .lock()
                    .map(|mut s| s.drain(..).collect())
                    .unwrap_or_default();
                let mut pending_tools: Vec<serde_json::Value> = Vec::new();
                for seg in segments {
                    match seg {
                        CliSegment::Text(text) => {
                            // Flush pending tools before text
                            if !pending_tools.is_empty() {
                                let marker = format!(
                                    "\n<!-- tools-v2: {} -->\n",
                                    serde_json::to_string(&pending_tools).unwrap_or_default()
                                );
                                cli_content.push_str(&marker);
                                accumulated_text.push_str(&marker);
                                pending_tools.clear();
                            }
                            cli_content.push_str(&format!("{}\n\n", text));
                            if !accumulated_text.is_empty() {
                                accumulated_text.push_str("\n\n");
                            }
                            accumulated_text.push_str(&text);
                        }
                        CliSegment::Tool(entry) => {
                            pending_tools.push(entry);
                        }
                    }
                }
                // Flush trailing tools
                if !pending_tools.is_empty() {
                    let marker = format!(
                        "\n<!-- tools-v2: {} -->\n",
                        serde_json::to_string(&pending_tools).unwrap_or_default()
                    );
                    cli_content.push_str(&marker);
                    accumulated_text.push_str(&marker);
                }

                // Single atomic write — no partial state visible to load_session
                if !cli_content.is_empty() {
                    let _ = message_service
                        .append_content(assistant_db_msg.id, &cli_content)
                        .await;
                }
            } else {
                // Non-CLI: per-iteration write of `<!-- reasoning -->` marker +
                // iteration text into the SAME `content` column the CLI path
                // uses. Markers are stripped before the next turn's LLM
                // context is built (see top of run_tool_loop:
                // `strip_llm_artifacts` on db_messages when !is_cli_provider),
                // so the model never sees them in its history and can't echo
                // them back. Persisting per-iteration keeps the chronological
                // layout (think → text → tools → think → text → …) intact on
                // session reload, matching the live streamed view exactly.
                //
                // EXCEPTION: phantom iterations (narrated actions, zero
                // tool_use blocks) are operational scaffolding for the
                // self-heal retry loop, not turn history. Persisting them
                // pollutes the DB row that gets reloaded as assistant
                // context on the next turn (and on session reconnect —
                // matches the 34-entry Telegram session reported in
                // discussion #86 / gist 85cfdc26), so the model sees its
                // own past phantoms and repeats the pattern. Skip the
                // append on phantom iterations; the eventual successful
                // iteration (where a real tool runs after the self-heal
                // nudge or after a sticky-fallback swap) gets persisted
                // normally because by then the phantom signature has
                // been replaced.
                let iteration_is_phantom = !iteration_text.is_empty()
                    && tool_uses.is_empty()
                    && super::phantom::has_phantom_tool_intent_no_tools(&iteration_text);

                if iteration_is_phantom {
                    tracing::debug!(
                        "[phantom] Skipping DB persist for phantom iteration \
                         (text_len={}, has_reasoning={})",
                        iteration_text.len(),
                        reasoning_text
                            .as_deref()
                            .map(|r| !r.trim().is_empty())
                            .unwrap_or(false),
                    );
                } else {
                    let mut iter_content = String::new();
                    if let Some(ref reasoning) = reasoning_text
                        && !reasoning.trim().is_empty()
                    {
                        iter_content.push_str(&format!(
                            "<!-- reasoning -->\n{}\n<!-- /reasoning -->\n\n",
                            reasoning
                        ));
                    }
                    if !iteration_text.is_empty() {
                        if !accumulated_text.is_empty() {
                            accumulated_text.push_str("\n\n");
                        }
                        accumulated_text.push_str(&iteration_text);
                        iter_content.push_str(&format!("{}\n\n", iteration_text));
                    }
                    if !iter_content.is_empty() {
                        let _ = message_service
                            .append_content(assistant_db_msg.id, &iter_content)
                            .await;
                    }
                }
            }

            tracing::debug!("Found {} tool uses to execute", tool_uses.len());

            // CLI providers handle tools internally — emit progress events for
            // TUI display (expandable tool groups) but don't execute them.
            // Break immediately after — the CLI already completed its full run.
            if is_cli_provider && !tool_uses.is_empty() {
                // Text/tool interleaving and ToolStarted/ToolCompleted events
                // are already emitted during streaming by helpers.rs
                // (cli_unflushed_text flushes at tool boundaries + stream end).
                // Tool markers already persisted atomically above via cli_segments.
                //
                // Do NOT re-emit IntermediateText here — helpers.rs already sent
                // all text blocks during streaming. Emitting again causes the
                // entire conversation text to appear duplicated in the TUI.
                iteration_text.clear();
                tool_uses.clear();
            }

            if tool_uses.is_empty() {
                // Check queued messages — stream_complete may have consumed
                // one mid-stream (stored in queued_buf), or check the queue now.
                let (queued_msg, from_buf) = {
                    let buffered = queued_buf.lock().await.take();
                    if buffered.is_some() {
                        (buffered, true)
                    } else if let Some(ref queue_cb) = self.message_queue_callback {
                        (queue_cb(session_id).await, false)
                    } else {
                        (None, false)
                    }
                };
                if let Some(queued_msg) = queued_msg {
                    tracing::info!("Injecting queued user message (from_buf={})", from_buf);
                    // Emit assistant's intermediate text FIRST so it appears
                    // before the queued user message in the TUI
                    if !iteration_text.is_empty()
                        && let Some(ref cb) = progress_callback
                    {
                        cb(
                            session_id,
                            ProgressEvent::IntermediateText {
                                text: iteration_text,
                                reasoning: reasoning_text,
                            },
                        );
                    }
                    // Emit QueuedUserMessage — always here, never in stream_complete
                    if let Some(ref cb) = progress_callback {
                        cb(
                            session_id,
                            ProgressEvent::QueuedUserMessage {
                                text: queued_msg.clone(),
                            },
                        );
                    }
                    // Add assistant response + queued user message to context
                    let assistant_text = response
                        .content
                        .iter()
                        .filter_map(|b| match b {
                            ContentBlock::Text { text } => Some(text.as_str()),
                            _ => None,
                        })
                        .collect::<Vec<_>>()
                        .join("\n");
                    context.add_message(Message::assistant(assistant_text));
                    let injected = Message::user(queued_msg.clone());
                    context.add_message(injected);
                    let _ = message_service
                        .create_message(session_id, "user".to_string(), queued_msg)
                        .await;
                    // Create a NEW assistant placeholder so the next response
                    // gets a sequence number AFTER the queued user message.
                    // Without this, the next LLM response appends to the old
                    // placeholder (created before the user message), causing
                    // the reply to appear ABOVE the user's message in the DB.
                    assistant_db_msg = message_service
                        .create_message(session_id, "assistant".to_string(), String::new())
                        .await
                        .map_err(|e| AgentError::Database(e.to_string()))?;
                    continue;
                }

                // ── Phantom tool call detection ──────────────────────────
                // We're inside `if tool_uses.is_empty()`. The narrow
                // intent-phrase detector is the gate: extracting real
                // tool calls from text-shaped leak formats already runs
                // upstream, so zero tool_uses + no intent phrases is a
                // legitimate text answer that must pass through.
                //
                // POST-SUCCESS EXEMPTION: if at least one tool already
                // succeeded in this turn, the text-only iteration is a
                // completion acknowledgement ("Done.", "Pushed.",
                // "Committed.") — not phantom intent. Without this
                // guard the detector mistook every successful turn's
                // wrap-up for "described actions without executing",
                // forced phantom retries, eventually rolled the
                // self-heal budget, and switched providers — all on
                // already-completed work. Symptom: 8+ "Phantom tool
                // calls detected" alerts after a clean commit+push,
                // 293s and 4683 tokens wasted finalising nothing.
                // POST-SUCCESS EXEMPTION (refined). Phantom-eligibility gate. Two regimes:
                //
                //   1. No tool call completed this turn: standard
                //      phantom check — the iteration's text must not
                //      narrate an action without a tool call.
                //   2. Tool call(s) ALREADY completed: phantom stays
                //      exempt for pure completion acks (`Done.` /
                //      `Pushed.` / `On main.`) BUT re-engages when
                //      the text carries a FORWARD-looking intent
                //      phrase (`Let me dig into …`, `I'll check the
                //      …`). Forward intent after a tool call means
                //      the model promised more work and dropped it.
                //      Logs 2026-06-03 captured this regression:
                //      "Good, on main. Let me dig into the delete
                //      invitation endpoint, the email send path, and
                //      the invite flow to find the bugs." silently
                //      closed with three promised investigations un-
                //      dispatched because the original exemption
                //      disabled phantom for the whole post-tool
                //      portion of the turn.
                let phantom_eligible = !is_cli_provider
                    && (tool_calls_completed_this_turn == 0
                        || super::phantom::has_forward_intent_post_success(&iteration_text));
                if !phantom_eligible && tool_calls_completed_this_turn > 0 {
                    tracing::info!(
                        target: "phantom",
                        tools_completed = tool_calls_completed_this_turn,
                        text_len = iteration_text.len(),
                        "phantom detection skipped: turn already produced successful tool calls \
                         and the text-only iteration is a pure completion acknowledgement \
                         (no forward-looking intent phrase)"
                    );
                }
                let stuck_loop_now =
                    phantom_eligible && super::phantom::is_stuck_in_intent_loop(&iteration_text);
                if stuck_loop_now {
                    let reps = super::phantom::count_intent_line_starts(&iteration_text);
                    tracing::warn!(
                        "Phantom intent-loop detected ({} line-start repetitions) — escalating \
                         self-heal (nudge + fast-escalate to sticky fallback if budget half-burned).",
                        reps
                    );
                    self.record_provider_feedback(
                        session_id,
                        "phantom_intent_loop",
                        "self_heal",
                        Some(&format!(
                            "{} line-start repetitions in a single iteration",
                            reps
                        )),
                    );
                }

                // Fast-escalate to sticky fallback when the budget is
                // exhausted, or when the stuck-loop signal fires after
                // we've already burned at least half the budget. Either
                // condition means the current provider can't reach its
                // tool-call channel for this prompt and another nudge
                // won't help.
                let should_force_fallback = phantom_eligible
                    && !phantom_sticky_swap_done
                    && super::phantom::has_phantom_tool_intent_no_tools(&iteration_text)
                    && (phantom_retries_used >= MAX_PHANTOM_RETRIES
                        || (stuck_loop_now && phantom_retries_used >= MAX_PHANTOM_RETRIES / 2));
                if should_force_fallback {
                    let fb_provider = self.provider_for_session(session_id);
                    if fb_provider.force_next_fallback("phantom_intent_loop_or_exhausted") {
                        phantom_sticky_swap_done = true;
                        phantom_retries_used = 0;
                        let new_name = fb_provider
                            .active_subprovider_name()
                            .unwrap_or_else(|| fb_provider.name().to_string());
                        let new_model = fb_provider
                            .active_subprovider_model()
                            .unwrap_or_else(|| fb_provider.default_model().to_string());
                        tracing::warn!(
                            "Self-heal escalation: swapping from '{}' to '{}/{}' (stuck={}, retries={}).",
                            self.provider_name_for_session(session_id),
                            new_name,
                            new_model,
                            stuck_loop_now,
                            phantom_retries_used
                        );
                        self.record_provider_feedback(
                            session_id,
                            "phantom_sticky_swap",
                            "self_heal",
                            Some(&format!("→ {}/{}", new_name, new_model)),
                        );
                        if let Some(ref cb) = progress_callback {
                            cb(
                                session_id,
                                ProgressEvent::SelfHealingAlert {
                                    message: format!(
                                        "Self-heal switching to {}/{} (current provider can't reach \
                                         its tool channel)",
                                        new_name, new_model
                                    ),
                                },
                            );
                        }
                        self.persist_sticky_pair(session_id, new_name.clone(), new_model.clone());
                        model_name = new_model;
                        context.add_message(Message::user(
                            "[System: A different provider is now handling this turn. Invoke the \
                             correct tool through the structured tool-call API now. Do not \
                             narrate.]"
                                .to_string(),
                        ));
                        continue;
                    }
                }
                if phantom_retries_used < MAX_PHANTOM_RETRIES
                    && phantom_eligible
                    && super::phantom::has_phantom_tool_intent_no_tools(&iteration_text)
                {
                    phantom_retries_used += 1;
                    tracing::warn!(
                        "Phantom tool call detected (local={}) — model described \
                         actions without executing tools. Injecting retry prompt.",
                        is_local_provider
                    );
                    self.record_provider_feedback(
                        session_id,
                        "phantom_tool_call",
                        "self_heal",
                        Some(&iteration_text.chars().take(300).collect::<String>()),
                    );
                    if let Some(ref cb) = progress_callback {
                        cb(
                            session_id,
                            ProgressEvent::SelfHealingAlert {
                                message: "Phantom tool calls detected — retrying with enforcement"
                                    .into(),
                            },
                        );
                    }
                    // Inject a system correction nudge. Local
                    // models respond better to Unsloth's blunter wording;
                    // cloud models get our existing, more-specific nudge.
                    // Do NOT add the phantom text as assistant message — it
                    // pollutes context and causes the model to hallucinate
                    // new responses from the correction feedback itself.
                    let nudge = if is_local_provider {
                        // Local models (Qwen/Kimi/DeepSeek) over-index on
                        // "STOP" and interpret it as "wait for further
                        // instruction" — they reply with acknowledgements
                        // ("Under the STOP rule I'll wait") instead of
                        // calling a tool. Also emphasise the STRUCTURED
                        // API: when the prompt is ambiguous the model
                        // writes JSON text like `{"tool_call":{...}}`
                        // thinking that IS the invocation.
                        "[System: Your last response produced ZERO tool_use blocks — the tool \
                         was NOT executed. Do not write JSON, do not write markdown code blocks, \
                         do not describe what you would do. Invoke the tool through the \
                         provider's structured tool-call API (the same channel the function \
                         schemas were registered on). Pick the correct tool and call it now. \
                         If the task is already completed and you've reported the results, \
                         respond with a short confirmation (e.g., 'Done.', 'Fixed.', 'Committed.') \
                         and stop — do not run additional tool calls to verify work you already did.]"
                    } else {
                        "[System: You described changes to files but did not execute any tool \
                         calls. Your response contained action language and file paths but zero \
                         tool_use blocks. Execute the actual tool calls NOW. Do not narrate — \
                         call the tools. If the task is already completed and you've reported \
                         the results, respond with a short confirmation (e.g., 'Done.', 'Fixed.', \
                         'Committed.') and stop — do not run additional tool calls to verify work \
                         you already did.]"
                    };
                    context.add_message(Message::user(nudge.to_string()));
                    continue;
                }

                // Cap hit and the fast-escalate block above couldn't
                // swap (no fallback left, or already swapped once).
                // Reset the counter and keep nudging the active provider
                // — user presses Stop if they want out.
                if phantom_eligible
                    && phantom_retries_used >= MAX_PHANTOM_RETRIES
                    && super::phantom::has_phantom_tool_intent_no_tools(&iteration_text)
                {
                    tracing::warn!(
                        "Phantom retry cap rolling ({} retries used, sticky_swapped={}) — \
                         resetting counter and re-nudging the active provider.",
                        phantom_retries_used,
                        phantom_sticky_swap_done
                    );
                    self.record_provider_feedback(
                        session_id,
                        "phantom_retry_rolling",
                        "self_heal",
                        Some(&iteration_text.chars().take(300).collect::<String>()),
                    );
                    if let Some(ref cb) = progress_callback {
                        cb(
                            session_id,
                            ProgressEvent::SelfHealingAlert {
                                message: "Self-heal retry budget rolled — forcing another retry"
                                    .to_string(),
                            },
                        );
                    }
                    phantom_retries_used = 0;
                    context.add_message(Message::user(
                        "[System: You have repeatedly described actions without invoking any \
                         tool. STOP narrating. Pick the correct tool and call it now through the \
                         structured tool-call API. No JSON, no markdown code blocks, only a real \
                         tool_use block. If the task is already completed and you've reported the \
                         results, respond with a short confirmation (e.g., 'Done.', 'Fixed.', \
                         'Committed.') and stop — do not run additional tool calls to verify work \
                         you already did.]"
                            .to_string(),
                    ));
                    continue;
                }

                // ── Rotation continuation ──────────────────────────────
                // When Qwen OAuth rotation happens mid-task, the new account
                // gets the same request but may respond with text-only (0 tools)
                // because it's a cold start on a fresh account. Inject a
                // continuation prompt so it picks up where the previous account
                // left off. Only retry once to avoid infinite loops.
                if !rotation_retry_used && rotated_this_iteration.is_some() && iteration > 1 {
                    rotation_retry_used = true;
                    tracing::warn!(
                        "Rotation yielded 0 tool calls after {} iterations — injecting continuation prompt",
                        iteration
                    );
                    if let Some(ref cb) = progress_callback {
                        cb(
                            session_id,
                            ProgressEvent::SelfHealingAlert {
                                message:
                                    "Account rotation mid-task — retrying with continuation context"
                                        .into(),
                            },
                        );
                    }
                    // Add the text-only response as assistant context, then nudge
                    context.add_message(Message::assistant(iteration_text));
                    context.add_message(Message::user(
                        "[System: Provider account rotation just occurred mid-task. You were \
                         actively executing tools in previous iterations but your last response \
                         contained zero tool calls. This is a continuation — review the conversation \
                         above and resume executing tools from where you left off. Do NOT summarize \
                         or re-explain. Execute the next tool call immediately.]"
                            .to_string(),
                    ));
                    continue;
                }

                // ── Empty-response + reasoning retry ─────────────────────
                // Some reasoning runtimes (local MLX Qwen3, and cloud
                // thinking models like alibaba-qwen `qwen-latest-series-
                // invite-beta-v34`) finish a turn with only
                // `reasoning_content` chunks — zero visible text, zero
                // tool calls. The user sees a tool card / their own
                // message and then nothing, which reads as a dropped
                // request with no self-heal.
                //
                // Escalation:
                //   1. Nudge up to EMPTY_REASONING_MAX_NUDGES (5) times,
                //      sharpening the system instruction each round.
                //   2. If still empty after the budget, walk the fallback
                //      chain (sticky swap + persist) so the next turn
                //      runs on a model that will actually answer.
                //   3. If no fallback succeeds, emit a final visible
                //      SelfHealingAlert so the user knows to switch
                //      providers manually — never a silent drop.
                let has_meaningful_reasoning = reasoning_text
                    .as_deref()
                    .map(|r| r.trim().len() >= 40)
                    .unwrap_or(false);
                if iteration > 0
                    && !is_cli_provider
                    && iteration_text.trim().is_empty()
                    && has_meaningful_reasoning
                {
                    if empty_reasoning_retries < EMPTY_REASONING_MAX_NUDGES {
                        empty_reasoning_retries += 1;
                        let attempt = empty_reasoning_retries;
                        tracing::warn!(
                            "Model ended turn with reasoning but no visible response \
                             (reasoning_len={}, iteration={}, nudge {}/{}) — nudging \
                             for the actual answer.",
                            reasoning_text.as_deref().map(|r| r.len()).unwrap_or(0),
                            iteration,
                            attempt,
                            EMPTY_REASONING_MAX_NUDGES,
                        );
                        if let Some(ref cb) = progress_callback {
                            cb(
                                session_id,
                                ProgressEvent::SelfHealingAlert {
                                    message: format!(
                                        "Model reasoned without answering — nudge {}/{}",
                                        attempt, EMPTY_REASONING_MAX_NUDGES,
                                    ),
                                },
                            );
                        }
                        // Each round gets a sharper system message so the
                        // model can't keep replying with more silent
                        // thinking. The last two attempts are explicit
                        // commands to stop reasoning entirely.
                        let nudge = match attempt {
                            1 => {
                                "[System: Your previous turn produced only internal reasoning \
                                  and no visible reply. The tool results above are sufficient — \
                                  write the answer now as plain text (tables, prose, or whatever \
                                  the user asked for). Do not re-reason, do not call more tools \
                                  unless strictly necessary.]"
                            }
                            2 => {
                                "[System: Second nudge — you again produced only reasoning. \
                                  Output the answer as plain text on this turn. No reasoning \
                                  block. No tool calls. Just the answer the user asked for.]"
                            }
                            3 => {
                                "[System: Third nudge. Stop reasoning. Reply now in plain \
                                  prose, one or two short paragraphs. No <thinking>, no \
                                  reasoning_content, no internal monologue.]"
                            }
                            4 => {
                                "[System: Fourth nudge — final warning before fallback. \
                                  Emit a visible text reply NOW. If you produce another \
                                  reasoning-only turn the conversation will switch to a \
                                  different provider automatically.]"
                            }
                            _ => {
                                "[System: Fifth and last nudge. Reply in plain text on this \
                                  turn or the system will hand the conversation to a fallback \
                                  provider on the next turn.]"
                            }
                        };
                        // Preserve the reasoning on the (empty) assistant
                        // turn, then inject a sharpening user nudge so
                        // the next iteration has to produce the actual
                        // answer.
                        context.add_message(Message::assistant(String::new()));
                        context.add_message(Message::user(nudge.to_string()));
                        continue;
                    }

                    // Budget exhausted — walk the fallback chain. Sticky
                    // swap so the next user turn also lands on the new
                    // provider; the original is gone for this session.
                    tracing::warn!(
                        "Empty-reasoning nudge budget exhausted ({}/{}) — walking \
                         fallback chain",
                        empty_reasoning_retries,
                        EMPTY_REASONING_MAX_NUDGES,
                    );
                    let active_name = self.provider_name_for_session(session_id);
                    let candidates: Vec<_> = self
                        .fallback_providers
                        .iter()
                        .filter(|p| p.name() != active_name)
                        .collect();

                    if candidates.is_empty() {
                        // No escape hatch configured. Surface a visible
                        // alert so the user knows to swap manually — do
                        // NOT exit silently.
                        if let Some(ref cb) = progress_callback {
                            cb(
                                session_id,
                                ProgressEvent::SelfHealingAlert {
                                    message: format!(
                                        "Model '{}/{}' refused to answer after {} nudges \
                                         and no fallback provider is configured. Use \
                                         /models to switch.",
                                        active_name, model_name, EMPTY_REASONING_MAX_NUDGES,
                                    ),
                                },
                            );
                        }
                        // Fall through: final_response = Some(response); break
                        // happens below (existing path). The user sees the
                        // alert instead of silence.
                    } else {
                        let mut fb_succeeded = None;
                        for fallback in &candidates {
                            let fb_name = fallback.name().to_string();
                            let fb_model = fallback.default_model().to_string();
                            if let Some(ref cb) = progress_callback {
                                cb(
                                    session_id,
                                    ProgressEvent::SelfHealingAlert {
                                        message: format!(
                                            "Trying fallback '{}/{}' for empty-reasoning \
                                             recovery...",
                                            fb_name, fb_model,
                                        ),
                                    },
                                );
                            }

                            let mut fb_req =
                                LLMRequest::new(fb_model.clone(), context.messages.clone())
                                    .with_max_tokens(self.max_tokens);
                            fb_req.working_directory =
                                Some(self.get_working_directory().to_string_lossy().to_string());
                            fb_req.session_id = Some(session_id);
                            if let Some(system) = &context.system_brain {
                                fb_req = fb_req.with_system(system.clone());
                            }
                            if self.tool_registry.count() > 0 {
                                fb_req =
                                    fb_req.with_tools(self.tool_schemas_for_session(session_id));
                            }

                            let original_provider = self.provider_for_session(session_id);
                            self.swap_provider_for_session(
                                session_id,
                                (*fallback).clone(),
                                (*fallback)
                                    .active_subprovider_model()
                                    .unwrap_or_else(|| (*fallback).default_model().to_string()),
                            );
                            let mut restore_guard = FallbackProviderGuard {
                                service: self,
                                session_id,
                                original: Some(original_provider),
                            };
                            let fb_result = self
                                .stream_complete(
                                    session_id,
                                    fb_req,
                                    cancel_token.as_ref(),
                                    progress_callback.as_ref(),
                                    None,
                                    None,
                                    false,
                                )
                                .await;
                            match fb_result {
                                Ok((fb_resp, _fb_reasoning)) => {
                                    let has_visible_text = fb_resp.content.iter().any(|b| {
                                        matches!(
                                            b,
                                            crate::brain::provider::ContentBlock::Text {
                                                text,
                                            } if !text.trim().is_empty()
                                        )
                                    });
                                    if has_visible_text {
                                        // Swap sticks for the rest of the session.
                                        restore_guard.original = None;
                                        drop(restore_guard);
                                        if let Some(ref cb) = progress_callback {
                                            let from_name =
                                                self.provider_name_for_session(session_id);
                                            cb(
                                                session_id,
                                                ProgressEvent::SelfHealingAlert {
                                                    message: format!(
                                                        "Empty-reasoning recovery → switched \
                                                         to {}/{}",
                                                        fb_name, fb_model,
                                                    ),
                                                },
                                            );
                                            cb(
                                                session_id,
                                                ProgressEvent::ProviderSwitched {
                                                    from_name,
                                                    from_model: self
                                                        .provider_model_for_session(session_id),
                                                    to_name: fb_name.clone(),
                                                    to_model: fb_model.clone(),
                                                    reason: "empty_reasoning".to_string(),
                                                },
                                            );
                                        }
                                        self.persist_sticky_pair(
                                            session_id,
                                            fb_name.clone(),
                                            fb_model.clone(),
                                        );
                                        fb_succeeded = Some(fb_resp);
                                        break;
                                    } else {
                                        // Also empty — restore and try next candidate.
                                        drop(restore_guard);
                                        tracing::warn!(
                                            "Empty-reasoning fallback '{}' also returned \
                                             empty — trying next",
                                            fb_name,
                                        );
                                    }
                                }
                                Err(fb_err) => {
                                    drop(restore_guard);
                                    tracing::warn!(
                                        "Empty-reasoning fallback '{}' failed: {} — \
                                         trying next",
                                        fb_name,
                                        fb_err,
                                    );
                                }
                            }
                        }

                        if let Some(fb_resp) = fb_succeeded {
                            // Replace the empty-reasoning response with the
                            // fallback's response and let the outer loop
                            // tail handle final_response + IntermediateText.
                            response = fb_resp;
                            // Re-derive iteration_text from the new response
                            // so the IntermediateText emit below has the
                            // actual reply text.
                            iteration_text = response
                                .content
                                .iter()
                                .filter_map(|b| match b {
                                    crate::brain::provider::ContentBlock::Text { text } => {
                                        Some(text.as_str())
                                    }
                                    _ => None,
                                })
                                .collect::<Vec<_>>()
                                .join("");
                        } else if let Some(ref cb) = progress_callback {
                            cb(
                                session_id,
                                ProgressEvent::SelfHealingAlert {
                                    message: format!(
                                        "All {} fallback providers also returned empty \
                                         reasoning. Use /models to pick a different model.",
                                        candidates.len(),
                                    ),
                                },
                            );
                        }
                    }
                }

                // ── Mid-sentence truncation retry ────────────────────────
                // Local reasoning models sometimes hit an internal EOS mid-
                // sentence. The response stream closes cleanly (finish_reason
                // =stop + usage chunk), but the visible text ends in the
                // middle of a word or clause: "Standard Get I", "Changelog
                // automation, duplicate CI fix, 1,890", etc. Detect the
                // truncation by looking at the last non-whitespace character
                // — if it's not a terminal token (punctuation, close-tag,
                // table pipe, code fence) we ask the model to continue once.
                if !truncated_mid_sentence_retry_used
                    && iteration > 0
                    && !is_cli_provider
                    && matches!(
                        response.stop_reason,
                        Some(crate::brain::provider::StopReason::EndTurn)
                    )
                    && super::truncation::try_emit_truncation_continue(
                        &iteration_text,
                        reasoning_text.as_ref(),
                        &mut context,
                        session_id,
                        &progress_callback,
                    )
                {
                    truncated_mid_sentence_retry_used = true;
                    // Mark the next iteration so the stream-error path skips
                    // cross-provider fallback for the continuation request.
                    current_iter_is_truncation_continue = true;
                    continue;
                }

                if iteration > 0 {
                    // Empty-analysis nudge: the model ran successful
                    // tool calls but produced ZERO text on the final
                    // iteration. For side-effect tasks (commit / push /
                    // edit) this is the intended outcome of the
                    // FINISHING A TURN directive — the tool result IS
                    // the deliverable. For analysis tasks ("audit X",
                    // "compare A and B", "what does Y do") the fetched
                    // data was meant to be INPUT to a text answer, and
                    // empty text means the user got nothing. One-shot
                    // nudge wakes the model up. If even after the
                    // nudge it still emits nothing, fall through and
                    // let the empty-text close stand — better than
                    // looping. Detection uses the clean user message
                    // when a channel handler supplied one (the
                    // `[Channel: ...]` prefix would otherwise pin
                    // every match to the wrapper, not the body).
                    let user_text_for_intent =
                        display_text_override.as_deref().unwrap_or(&user_message);
                    if !analysis_nudge_used
                        && iteration_text.trim().is_empty()
                        && tool_calls_completed_this_turn > 0
                        && super::phantom::is_analysis_intent(user_text_for_intent)
                    {
                        analysis_nudge_used = true;
                        tracing::warn!(
                            target: "analysis_empty_close",
                            tools_completed = tool_calls_completed_this_turn,
                            iteration,
                            "model ended turn with zero text after analysis-intent request — \
                             nudging once to produce the answer the user asked for"
                        );
                        self.record_provider_feedback(
                            session_id,
                            "analysis_empty_close",
                            "self_heal",
                            Some(&format!(
                                "iteration={iteration}, tools_completed={tool_calls_completed_this_turn}"
                            )),
                        );
                        if let Some(ref cb) = progress_callback {
                            cb(
                                session_id,
                                ProgressEvent::SelfHealingAlert {
                                    message: "Empty answer after data fetch — nudging the model to write the analysis"
                                        .to_string(),
                                },
                            );
                        }
                        context.add_message(Message::user(
                            "[System: You fetched data via tool calls but ended the turn with NO \
                             text response. The user's request was an analysis task (audit / \
                             review / explain / compare / summarise) where the tool result is \
                             INPUT to your answer, not the answer itself. Write the actual \
                             analysis now — cite specific fields, line numbers, or values from \
                             what you fetched. Do NOT run more tool calls; you already have the \
                             data. Do NOT reply with 'Done.' or 'Got it.' — those are for \
                             side-effect tasks, this is data interpretation. End once the \
                             analysis is written.]"
                                .to_string(),
                        ));
                        continue;
                    }

                    tracing::info!("Agent completed after {} tool iterations", iteration);
                    // Emit final text so TUI persists it as a permanent message.
                    // CLI providers: helpers.rs already flushed cli_unflushed_text
                    // as IntermediateText at stream end — skip to avoid duplication.
                    if !is_cli_provider
                        && !iteration_text.is_empty()
                        && let Some(ref cb) = progress_callback
                    {
                        cb(
                            session_id,
                            ProgressEvent::IntermediateText {
                                text: iteration_text,
                                reasoning: reasoning_text,
                            },
                        );
                    }
                } else {
                    tracing::info!("Agent responded with text only (no tool calls)");
                }
                final_response = Some(response);
                break;
            }

            // Emit intermediate text to TUI so it appears before the tool calls.
            //
            // Also emit when the iteration produced ONLY reasoning (no visible
            // text) but is about to execute tool calls. Without this, a local
            // reasoning model like MLX Qwen that emits
            // `reasoning_content` + structured `tool_calls` never persists its
            // per-iteration thinking — everything accumulates in
            // `streaming_reasoning` until the FINAL turn bundles all four
            // iterations' thoughts into one giant Thinking block at the
            // bottom of the chat (screenshot 2026-04-17 04:17). Firing per
            // iteration splits the thinking into its proper chronological
            // slots: iter-1-thinking → tools → iter-2-thinking → tools …
            let has_reasoning_to_persist = reasoning_text
                .as_deref()
                .map(|r| !r.trim().is_empty())
                .unwrap_or(false);
            if (!iteration_text.is_empty() || has_reasoning_to_persist)
                && let Some(ref cb) = progress_callback
            {
                cb(
                    session_id,
                    ProgressEvent::IntermediateText {
                        text: iteration_text,
                        // Clone: reasoning_text is still needed downstream to
                        // seed the assistant message's ContentBlock::Thinking
                        // so follow-up turns (notably kimi/Moonshot) have the
                        // required `reasoning_content` to echo back.
                        reasoning: reasoning_text.clone(),
                    },
                );
            }

            // Detect tool loops: hash the full input for every tool.
            // Different arguments = different hash = no false loop detection.
            let current_call_signature = tool_uses
                .iter()
                .map(|(_, name, input)| {
                    let input_str = serde_json::to_string(input).unwrap_or_default();
                    let hash: u64 = input_str
                        .bytes()
                        .fold(0u64, |acc, b| acc.wrapping_mul(31).wrapping_add(b as u64));
                    format!("{}:{:x}", name, hash)
                })
                .collect::<Vec<_>>()
                .join(",");

            recent_tool_calls.push(current_call_signature.clone());

            // Keep last 50 iterations for loop detection.
            // Modern agents legitimately make dozens of tool calls with different args.
            // Signatures include arguments, so only truly identical calls match.
            if recent_tool_calls.len() > 50 {
                recent_tool_calls.remove(0);
            }

            // Check for repeated patterns with tool-specific thresholds.
            // Only triggers for truly identical calls (same tool + same arguments).

            let is_modification_tool = current_call_signature.starts_with("write:")
                || current_call_signature.starts_with("edit:")
                || current_call_signature.starts_with("bash:");

            // Modification tools get a lower threshold (dangerous if looping).
            // Everything else gets a generous threshold since signatures
            // already distinguish different arguments.
            let loop_threshold = if is_modification_tool {
                4 // Same exact write/edit/bash command 4 times = stuck
            } else {
                8 // Same exact call with same exact args 8 times = stuck
            };

            // Check if we have enough calls to detect a loop
            if recent_tool_calls.len() >= loop_threshold {
                let last_n = &recent_tool_calls[recent_tool_calls.len() - loop_threshold..];
                if last_n.iter().all(|call| call == &current_call_signature) {
                    tracing::warn!(
                        "⚠️ Detected tool loop: '{}' called {} times in a row. Breaking loop.",
                        current_call_signature,
                        loop_threshold
                    );

                    if is_modification_tool {
                        tracing::warn!(
                            "⚠️ Modification tool loop detected. \
                             Same command repeated {} times with identical arguments.",
                            loop_threshold
                        );
                    }

                    // Force a final response by breaking the loop
                    final_response = Some(response);
                    break;
                }
            }

            // Semantic-loop detection for browser navigation cycles.
            //
            // Exact-loop detection above only fires when the SAME tool +
            // SAME args repeat. The browser navigate→wait→screenshot
            // rotation uses three different tools with varying args, so
            // it slips through every check despite producing zero progress.
            // 2026-05-23 09:04 logs show 32+ iterations of this rotation
            // before the user gave up.
            //
            // Heuristic: if the last 8 iterations contain `browser_screenshot`
            // 4+ times AND zero progress-making interactions (click/type),
            // inject a nudge once per turn telling the agent to interact
            // instead of screenshot again.
            if !browser_screenshot_loop_nudged && recent_tool_calls.len() >= 8 {
                let last8 = &recent_tool_calls[recent_tool_calls.len() - 8..];
                let screenshot_count = last8
                    .iter()
                    .filter(|sig| sig.starts_with("browser_screenshot:"))
                    .count();
                let progress_count = last8
                    .iter()
                    .filter(|sig| {
                        sig.starts_with("browser_click:")
                            || sig.starts_with("browser_type:")
                            || sig.starts_with("browser_eval:")
                    })
                    .count();
                if screenshot_count >= 4 && progress_count == 0 {
                    tracing::warn!(
                        "Browser semantic loop: {} screenshots in last 8 iterations, \
                         0 interactions — injecting nudge.",
                        screenshot_count,
                    );
                    if let Some(ref cb) = progress_callback {
                        cb(
                            session_id,
                            ProgressEvent::SelfHealingAlert {
                                message: format!(
                                    "Stuck in screenshot loop ({}/8 iterations) — \
                                     nudging agent to interact with the page",
                                    screenshot_count,
                                ),
                            },
                        );
                    }
                    context.add_message(Message::user(
                        "[System: You have taken multiple screenshots of the same page \
                         without interacting. The screenshots already show what's on \
                         screen — STOP screenshotting. To move forward you must either: \
                         (1) `browser_click` an element (use `text=Label`, `xpath=...`, \
                         or a CSS selector), (2) `browser_type` into an input field, or \
                         (3) `browser_navigate` to a new URL. If you can't find the \
                         element you need, call `browser_find` with mode=\"text\" or \
                         mode=\"aria\" to enumerate candidates and get back stable \
                         `[data-opencrabs-match=\"N\"]` selectors. Do not screenshot \
                         again on this turn.]"
                            .to_string(),
                    ));
                    browser_screenshot_loop_nudged = true;
                    continue;
                }
            }

            // Execute tools and build response message
            let mut tool_results = Vec::new();
            let mut tool_descriptions: Vec<String> = Vec::new(); // For DB persistence
            let mut tool_outputs: Vec<(bool, String)> = Vec::new(); // (success, output) parallel to descriptions

            for (tool_id, tool_name, tool_input) in tool_uses {
                // Check for cancellation before each tool
                if let Some(ref token) = cancel_token
                    && token.is_cancelled()
                {
                    tracing::warn!(
                        "🛑 Tool execution cancelled before '{}' at iteration {}",
                        tool_name,
                        iteration,
                    );
                    break;
                }

                tracing::info!("Executing tool '{}' (iteration {})", tool_name, iteration,);

                // Save tool input for progress reporting (before it's moved to execute)
                let tool_input_for_progress = tool_input.clone();

                // Build short description for DB persistence
                tool_descriptions.push(Self::format_tool_summary(&tool_name, &tool_input));

                // Emit tool started progress
                if let Some(ref cb) = progress_callback {
                    cb(
                        session_id,
                        ProgressEvent::ToolStarted {
                            tool_name: tool_name.clone(),
                            tool_input: tool_input_for_progress.clone(),
                        },
                    );
                }

                // Check if approval is needed.
                // Each channel's make_approval_callback() already checks
                // check_approval_policy() from config — the tool loop only
                // respects the auto_approve_tools flag and tool-level policy.
                let needs_approval = if let Some(tool) = self.tool_registry.get(&tool_name) {
                    tool.requires_approval_for_input(&tool_input)
                        && (!self.auto_approve_tools || has_override_approval)
                        && !tool_context.auto_approve
                } else {
                    false
                };

                // Request approval if needed
                if needs_approval {
                    if let Some(ref approval_cb) = approval_callback {
                        // Get tool details for approval request
                        let tool_info = if let Some(tool) = self.tool_registry.get(&tool_name) {
                            ToolApprovalInfo {
                                session_id,
                                tool_name: tool_name.clone(),
                                tool_description: tool.description().to_string(),
                                tool_input: tool_input.clone(),
                                capabilities: tool
                                    .capabilities()
                                    .iter()
                                    .map(|c| format!("{:?}", c))
                                    .collect(),
                            }
                        } else {
                            // Tool not found, skip approval
                            let err = format!("Tool not found: {}", tool_name);
                            tool_outputs.push((false, err.clone()));
                            tool_results.push(ContentBlock::ToolResult {
                                tool_use_id: tool_id,
                                content: err,
                                is_error: Some(true),
                            });
                            continue;
                        };

                        // Call approval callback
                        tracing::info!("Requesting user approval for tool '{}'", tool_name);
                        match approval_cb(tool_info).await {
                            Ok((approved, always_approve)) => {
                                if !approved {
                                    tracing::warn!("User denied approval for tool '{}'", tool_name);
                                    self.record_tool_feedback(
                                        session_id,
                                        &tool_name,
                                        Some(&tool_input_for_progress),
                                        false,
                                        Some("user_denied_approval"),
                                    );
                                    tool_outputs
                                        .push((false, "User denied permission".to_string()));
                                    tool_results.push(ContentBlock::ToolResult {
                                        tool_use_id: tool_id,
                                        content: "User denied permission to execute this tool"
                                            .to_string(),
                                        is_error: Some(true),
                                    });
                                    continue;
                                }
                                // Propagate "always approve" to skip callbacks for remaining tools
                                if always_approve {
                                    tool_context.auto_approve = true;
                                    tracing::info!(
                                        "User selected 'Always' — auto-approving remaining tools in this loop"
                                    );
                                }
                                tracing::info!("User approved tool '{}'", tool_name);
                                // Create approved context for this tool execution
                                let approved_tool_context = ToolExecutionContext {
                                    session_id: tool_context.session_id,
                                    working_directory: tool_context.working_directory.clone(),
                                    env_vars: tool_context.env_vars.clone(),
                                    auto_approve: true, // User approved this execution
                                    timeout_secs: tool_context.timeout_secs,
                                    sudo_callback: tool_context.sudo_callback.clone(),
                                    ssh_callback: tool_context.ssh_callback.clone(),
                                    shared_working_directory: tool_context
                                        .shared_working_directory
                                        .clone(),
                                    service_context: tool_context.service_context.clone(),
                                    question_callback: tool_context.question_callback.clone(),
                                };

                                // Execute the tool with approved context, racing against cancel
                                let exec_result = tokio::select! {
                                    biased;
                                    _ = async {
                                        if let Some(ref t) = cancel_token { t.cancelled().await } else { std::future::pending().await }
                                    } => {
                                        tracing::warn!("🛑 Tool '{}' cancelled mid-execution", tool_name);
                                        break;
                                    }
                                    r = self.tool_registry.execute(&tool_name, tool_input, &approved_tool_context) => r,
                                };
                                match exec_result {
                                    Ok(result) => {
                                        let success = result.success;
                                        let images = result.images;
                                        let content = build_tool_result_content(
                                            result.success,
                                            result.error,
                                            &result.output,
                                        );

                                        // GRANULAR LOG: Tool execution result
                                        if success {
                                            tracing::info!(
                                                "[TOOL_EXEC] ✅ Tool '{}' executed successfully, output_len={}",
                                                tool_name,
                                                content.len()
                                            );
                                            // Mirror of the non-approval branch:
                                            // a successful tool run wipes the
                                            // phantom retry counter so a later
                                            // isolated phantom burst is judged
                                            // on its own merits, not on debt
                                            // accumulated earlier in the turn.
                                            phantom_retries_used = 0;
                                            tool_calls_completed_this_turn += 1;
                                            // Persist the touched path so a later
                                            // session on this project can re-anchor
                                            // on real paths instead of guessing.
                                            if let Some(p) = extract_path_for_recent_buffer(
                                                &tool_name,
                                                &tool_input_for_progress,
                                                &approved_tool_context.working_directory,
                                            ) {
                                                self.record_recent_path(
                                                    &approved_tool_context.working_directory,
                                                    &p,
                                                );
                                            }
                                        } else {
                                            tracing::error!(
                                                "[TOOL_EXEC] ❌ Tool '{}' failed: {}",
                                                tool_name,
                                                content.chars().take(200).collect::<String>()
                                            );
                                        }

                                        // Auto-record to feedback ledger (fire-and-forget)
                                        self.record_tool_feedback(
                                            session_id,
                                            &tool_name,
                                            Some(&tool_input_for_progress),
                                            success,
                                            if success { None } else { Some(&content) },
                                        );

                                        // Record tool execution for usage dashboard
                                        if let Some(pool) = crate::db::global_pool() {
                                            let tool_repo =
                                                crate::db::repository::ToolExecutionRepository::new(
                                                    pool.clone(),
                                                );
                                            let exec_id = uuid::Uuid::new_v4().to_string();
                                            let mid = assistant_db_msg.id.to_string();
                                            let sid = session_id.to_string();
                                            let tname = tool_name.clone();
                                            let status = if success { "success" } else { "error" };
                                            tokio::spawn(async move {
                                                if let Err(e) = tool_repo
                                                    .record(&exec_id, &mid, &sid, &tname, status)
                                                    .await
                                                {
                                                    tracing::error!(
                                                        "[TOOL_EXEC] Failed to record tool execution: {}",
                                                        e
                                                    );
                                                }
                                            });
                                        }

                                        let output_summary: String = strip_ansi_output(&content)
                                            .chars()
                                            .take(2000)
                                            .collect();
                                        tool_outputs.push((success, output_summary.clone()));
                                        if let Some(ref cb) = progress_callback {
                                            cb(
                                                session_id,
                                                ProgressEvent::ToolCompleted {
                                                    tool_name: tool_name.clone(),
                                                    tool_input: tool_input_for_progress.clone(),
                                                    success,
                                                    summary: output_summary,
                                                },
                                            );
                                        }
                                        tool_results.push(ContentBlock::ToolResult {
                                            tool_use_id: tool_id,
                                            content,
                                            is_error: Some(!success),
                                        });
                                        // Append images (e.g. browser auto-screenshots) so the model sees them
                                        for (media_type, data) in images {
                                            tool_results.push(ContentBlock::Image {
                                                source:
                                                    crate::brain::provider::ImageSource::Base64 {
                                                        media_type,
                                                        data,
                                                    },
                                            });
                                        }
                                    }
                                    Err(e) => {
                                        let err_msg = format!("Tool execution error: {}", e);
                                        // GRANULAR LOG: Tool execution error
                                        tracing::error!(
                                            "[TOOL_EXEC] 💥 Tool '{}' error: {}",
                                            tool_name,
                                            err_msg
                                        );
                                        self.record_tool_feedback(
                                            session_id,
                                            &tool_name,
                                            Some(&tool_input_for_progress),
                                            false,
                                            Some(&err_msg),
                                        );
                                        // Record tool execution for usage dashboard
                                        if let Some(pool) = crate::db::global_pool() {
                                            let tool_repo =
                                                crate::db::repository::ToolExecutionRepository::new(
                                                    pool.clone(),
                                                );
                                            let exec_id = uuid::Uuid::new_v4().to_string();
                                            let mid = assistant_db_msg.id.to_string();
                                            let sid = session_id.to_string();
                                            let tname = tool_name.clone();
                                            tokio::spawn(async move {
                                                if let Err(e) = tool_repo
                                                    .record(&exec_id, &mid, &sid, &tname, "error")
                                                    .await
                                                {
                                                    tracing::error!(
                                                        "[TOOL_EXEC] Failed to record tool execution: {}",
                                                        e
                                                    );
                                                }
                                            });
                                        }
                                        let output_summary: String = strip_ansi_output(&err_msg)
                                            .chars()
                                            .take(2000)
                                            .collect();
                                        tool_outputs.push((false, output_summary.clone()));
                                        if let Some(ref cb) = progress_callback {
                                            cb(
                                                session_id,
                                                ProgressEvent::ToolCompleted {
                                                    tool_name: tool_name.clone(),
                                                    tool_input: tool_input_for_progress.clone(),
                                                    success: false,
                                                    summary: output_summary,
                                                },
                                            );
                                        }
                                        tool_results.push(ContentBlock::ToolResult {
                                            tool_use_id: tool_id,
                                            content: err_msg,
                                            is_error: Some(true),
                                        });
                                    }
                                }
                                continue; // Skip the normal execution path below
                            }
                            Err(e) => {
                                tracing::error!("Approval callback error: {}", e);
                                tool_outputs.push((false, format!("Approval failed: {}", e)));
                                tool_results.push(ContentBlock::ToolResult {
                                    tool_use_id: tool_id,
                                    content: format!("Approval request failed: {}", e),
                                    is_error: Some(true),
                                });
                                continue;
                            }
                        }
                    } else {
                        // No approval callback configured, deny execution
                        tracing::warn!(
                            "Tool '{}' requires approval but no approval callback configured",
                            tool_name
                        );
                        tool_outputs.push((false, "No approval mechanism configured".to_string()));
                        tool_results.push(ContentBlock::ToolResult {
                            tool_use_id: tool_id,
                            content: "Tool requires approval but no approval mechanism configured"
                                .to_string(),
                            is_error: Some(true),
                        });
                        continue;
                    }
                }

                // Execute the tool (no approval needed — mark context as approved
                // so the registry's own approval check doesn't block it)
                let mut approved_context = tool_context.clone();
                approved_context.auto_approve = true;
                let exec_result = tokio::select! {
                    biased;
                    _ = async {
                        if let Some(ref t) = cancel_token { t.cancelled().await } else { std::future::pending().await }
                    } => {
                        tracing::warn!("🛑 Tool '{}' cancelled mid-execution", tool_name);
                        break;
                    }
                    r = self.tool_registry.execute(&tool_name, tool_input, &approved_context) => r,
                };
                match exec_result {
                    Ok(result) => {
                        let success = result.success;
                        let images = result.images;
                        let content =
                            build_tool_result_content(result.success, result.error, &result.output);

                        // GRANULAR LOG: Direct tool execution result
                        if success {
                            tracing::info!(
                                "[TOOL_EXEC] ✅ Tool '{}' executed successfully, output_len={}",
                                tool_name,
                                content.len()
                            );
                            // Reset the phantom retry counter so accumulated
                            // debt from earlier in the turn doesn't push a
                            // later, isolated phantom burst over the cap. The
                            // counter is now "consecutive phantoms since the
                            // last real tool execution" rather than "phantom
                            // count across the entire turn".
                            phantom_retries_used = 0;
                            // Mark the turn as having produced real work so
                            // the subsequent text-only wrap-up iteration is
                            // exempt from phantom-tool-call detection. See
                            // the comment on the `tool_calls_completed_this_turn`
                            // declaration at the top of `run_tool_loop_inner`.
                            tool_calls_completed_this_turn += 1;
                            // Persist the touched path (same rationale as the
                            // approval-path branch above).
                            if let Some(p) = extract_path_for_recent_buffer(
                                &tool_name,
                                &tool_input_for_progress,
                                &approved_context.working_directory,
                            ) {
                                self.record_recent_path(&approved_context.working_directory, &p);
                            }
                        } else {
                            tracing::error!(
                                "[TOOL_EXEC] ❌ Tool '{}' failed: {}",
                                tool_name,
                                content.chars().take(200).collect::<String>()
                            );
                        }

                        // Auto-record to feedback ledger (fire-and-forget)
                        self.record_tool_feedback(
                            session_id,
                            &tool_name,
                            Some(&tool_input_for_progress),
                            success,
                            if success { None } else { Some(&content) },
                        );

                        // Record tool execution for usage dashboard
                        if let Some(pool) = crate::db::global_pool() {
                            let tool_repo =
                                crate::db::repository::ToolExecutionRepository::new(pool.clone());
                            let exec_id = uuid::Uuid::new_v4().to_string();
                            let mid = assistant_db_msg.id.to_string();
                            let sid = session_id.to_string();
                            let tname = tool_name.clone();
                            let status = if success { "success" } else { "error" };
                            tokio::spawn(async move {
                                if let Err(e) =
                                    tool_repo.record(&exec_id, &mid, &sid, &tname, status).await
                                {
                                    tracing::error!(
                                        "[TOOL_EXEC] Failed to record tool execution: {}",
                                        e
                                    );
                                }
                            });
                        }

                        let output_summary: String =
                            strip_ansi_output(&content).chars().take(2000).collect();
                        tool_outputs.push((success, output_summary.clone()));
                        if let Some(ref cb) = progress_callback {
                            cb(
                                session_id,
                                ProgressEvent::ToolCompleted {
                                    tool_name: tool_name.clone(),
                                    tool_input: tool_input_for_progress.clone(),
                                    success,
                                    summary: output_summary,
                                },
                            );
                        }
                        tool_results.push(ContentBlock::ToolResult {
                            tool_use_id: tool_id,
                            content,
                            is_error: Some(!success),
                        });
                        // Append images (e.g. browser auto-screenshots) so the model sees them
                        for (media_type, data) in images {
                            tool_results.push(ContentBlock::Image {
                                source: crate::brain::provider::ImageSource::Base64 {
                                    media_type,
                                    data,
                                },
                            });
                        }
                    }
                    Err(e) => {
                        let err_msg = format!("Tool execution error: {}", e);
                        // GRANULAR LOG: Direct tool execution error
                        tracing::error!("[TOOL_EXEC] 💥 Tool '{}' error: {}", tool_name, err_msg);
                        self.record_tool_feedback(
                            session_id,
                            &tool_name,
                            Some(&tool_input_for_progress),
                            false,
                            Some(&err_msg),
                        );
                        // Record tool execution for usage dashboard
                        if let Some(pool) = crate::db::global_pool() {
                            let tool_repo =
                                crate::db::repository::ToolExecutionRepository::new(pool.clone());
                            let exec_id = uuid::Uuid::new_v4().to_string();
                            let mid = assistant_db_msg.id.to_string();
                            let sid = session_id.to_string();
                            let tname = tool_name.clone();
                            tokio::spawn(async move {
                                if let Err(e) = tool_repo
                                    .record(&exec_id, &mid, &sid, &tname, "error")
                                    .await
                                {
                                    tracing::error!(
                                        "[TOOL_EXEC] Failed to record tool execution: {}",
                                        e
                                    );
                                }
                            });
                        }
                        let output_summary: String = err_msg.chars().take(2000).collect();
                        tool_outputs.push((false, output_summary.clone()));
                        if let Some(ref cb) = progress_callback {
                            cb(
                                session_id,
                                ProgressEvent::ToolCompleted {
                                    tool_name: tool_name.clone(),
                                    tool_input: tool_input_for_progress.clone(),
                                    success: false,
                                    summary: output_summary,
                                },
                            );
                        }
                        tool_results.push(ContentBlock::ToolResult {
                            tool_use_id: tool_id,
                            content: err_msg,
                            is_error: Some(true),
                        });
                    }
                }
            }

            // Append tool call data to accumulated text for DB persistence.
            // v2 format: <!-- tools-v2: [{"d":"desc","s":true,"o":"output..."}] -->
            // Includes tool output so Ctrl+O expansion works after session reload.
            if !tool_descriptions.is_empty() {
                if !accumulated_text.is_empty() {
                    accumulated_text.push('\n');
                }
                let entries: Vec<serde_json::Value> = tool_descriptions.iter()
                    .zip(tool_outputs.iter())
                    .map(|(desc, (success, output))| {
                        serde_json::json!({"d": desc, "s": success, "o": output})
                    })
                    .collect();
                accumulated_text.push_str(&format!(
                    "<!-- tools-v2: {} -->",
                    serde_json::to_string(&entries).unwrap_or_default()
                ));

                // REAL-TIME PERSISTENCE: Save tool results to DB immediately
                let tool_block = format!(
                    "\n<!-- tools-v2: {} -->\n",
                    serde_json::to_string(&entries).unwrap_or_default()
                );
                let _ = message_service
                    .append_content(assistant_db_msg.id, &tool_block)
                    .await;

                // Notify TUI after each tool iteration so it refreshes in real-time,
                // even during long-running channel sessions (Telegram, WhatsApp, etc.)
                if let Some(ref tx) = self.session_updated_tx {
                    let _ = tx.send(crate::brain::agent::ChannelSessionEvent::Updated(
                        session_id,
                    ));
                }

                tool_descriptions.clear();
                tool_outputs.clear();
            }

            // Add assistant message with tool use to context (filter empty text blocks).
            // Preserve the live reasoning text as a ContentBlock::Thinking so
            // downstream providers that require it — notably Moonshot kimi
            // via opencode.ai/zen/go, which rejects any follow-up turn whose
            // assistant tool_call messages omit `reasoning_content` — can
            // echo the real reasoning instead of a placeholder. Other
            // providers either use it natively (Anthropic) or ignore
            // unknown blocks (OpenAI, Zhipu, Minimax).
            let mut clean_content: Vec<ContentBlock> = response
                .content
                .iter()
                .filter(|b| !matches!(b, ContentBlock::Text { text } if text.is_empty()))
                .cloned()
                .collect();
            if let Some(ref reasoning) = reasoning_text
                && !reasoning.trim().is_empty()
                && !clean_content
                    .iter()
                    .any(|b| matches!(b, ContentBlock::Thinking { .. }))
            {
                clean_content.insert(
                    0,
                    ContentBlock::Thinking {
                        thinking: reasoning.clone(),
                        signature: None,
                    },
                );
            }
            let assistant_msg = Message {
                role: crate::brain::provider::Role::Assistant,
                content: clean_content,
            };
            context.add_message(assistant_msg);

            // Cap oversized tool_result bodies BEFORE they enter context.
            // A single 1 MB read_file output (e.g., an HTML file with an
            // embedded base64 PNG) dumps ~256k tokens into context in one
            // push — exceeding the model's window AND the compaction
            // summarizer's window, triggering a hard-truncate-to-zero
            // cascade observed today on session 5ed9ff25 (read of
            // opencrabs-retro-release.html, 1,025,562 bytes → ctx jumps
            // 8k → 738k → 0 messages after truncate). Truncate generously
            // (50 KB chars ≈ 12k tokens, ~6% of a 200k window) and
            // instruct the agent to re-call with offsets / grep / line
            // ranges for the part it actually needs.
            const MAX_TOOL_RESULT_CHARS: usize = 50_000;
            for block in tool_results.iter_mut() {
                if let ContentBlock::ToolResult { content, .. } = block
                    && content.len() > MAX_TOOL_RESULT_CHARS
                {
                    let original_len = content.len();
                    let mut cut = MAX_TOOL_RESULT_CHARS;
                    while cut > 0 && !content.is_char_boundary(cut) {
                        cut -= 1;
                    }
                    content.truncate(cut);
                    content.push_str(&format!(
                        "\n\n[Output truncated: {} → {} bytes. Re-call with \
                         start_line/line_count (read_file), head/tail, \
                         grep --max-count, or similar to fetch specific \
                         portions instead of the whole blob.]",
                        original_len, cut,
                    ));
                    tracing::warn!(
                        "Tool result content capped: {} → {} bytes (max {} chars)",
                        original_len,
                        cut,
                        MAX_TOOL_RESULT_CHARS,
                    );
                }
            }

            // Add user message with tool results to context
            let tool_result_msg = Message {
                role: crate::brain::provider::Role::User,
                content: tool_results,
            };
            context.add_message(tool_result_msg);

            // Fire token count update after tool results are added — keeps TUI in sync.
            if let Some(ref cb) = progress_callback {
                cb(session_id, ProgressEvent::TokenCount(context.token_count));
            }
            if has_progress_override && let Some(ref cb) = self.progress_callback {
                cb(session_id, ProgressEvent::TokenCount(context.token_count));
            }

            // Enforce 65% budget after tool results. Skip ONLY when the CLI
            // owns its session (claude-cli with --resume). Qwen is spawned
            // cold every turn so we MUST compact for it.
            if let Some(ref summary) = if cli_owns_context {
                None
            } else {
                self.enforce_context_budget(
                    session_id,
                    &mut context,
                    &model_name,
                    cancel_token.as_ref(),
                    &progress_callback,
                )
                .await
            } {
                // Persist compaction marker to DB so restarts load from this point
                let compaction_marker = format!(
                    "[CONTEXT COMPACTION — The conversation was automatically compacted. \
                     Below is a structured summary of everything before this point.]\n\n{}",
                    summary
                );
                if let Err(e) = message_service
                    .create_message(session_id, "user".to_string(), compaction_marker)
                    .await
                {
                    tracing::error!("Failed to persist post-tool compaction marker to DB: {}", e);
                }

                let cont_text = super::compaction_prompts::build_continuation(
                    super::compaction_prompts::CompactionKind::PostTool,
                    self.silent_compaction,
                    self.auto_approve_tools,
                );
                context.add_message(Message::user(cont_text));
            }

            // Check for queued user messages to inject between tool iterations.
            // This lets the user provide follow-up feedback mid-execution (like Claude Code).
            if let Some(ref queue_cb) = self.message_queue_callback
                && let Some(queued_msg) = queue_cb(session_id).await
            {
                tracing::info!("Injecting queued user message between tool iterations");

                // Notify TUI so the user message appears inline in the chat flow
                if let Some(ref cb) = progress_callback {
                    cb(
                        session_id,
                        ProgressEvent::QueuedUserMessage {
                            text: queued_msg.clone(),
                        },
                    );
                }

                let injected = Message::user(queued_msg.clone());
                context.add_message(injected);

                // Save to database so conversation history stays consistent
                let _ = message_service
                    .create_message(session_id, "user".to_string(), queued_msg)
                    .await;
                // Create a NEW assistant placeholder so the next response
                // gets a sequence number AFTER the queued user message.
                assistant_db_msg = message_service
                    .create_message(session_id, "assistant".to_string(), String::new())
                    .await
                    .map_err(|e| AgentError::Database(e.to_string()))?;
            }
        }

        // === GRACEFUL SAVE ON CANCEL/LOOP-BREAK ===
        // If we broke out of the loop without a final_response (cancellation, error, etc.)
        // but we have accumulated text/tool results, they're already in the DB from real-time persistence.
        // Usage update is handled below in the unified path after response synthesis —
        // doing it here too would double-count because the synthesized response (line below)
        // still flows through the final update_session_usage call.
        if final_response.is_none() && !accumulated_text.is_empty() {
            tracing::info!(
                "Loop broken without final response but accumulated text ({} chars) already persisted in real-time",
                accumulated_text.len()
            );
        }

        // If the loop broke without a final_response but we have accumulated text,
        // synthesize a partial response instead of erroring — the user already saw the
        // text streamed in real-time, so returning it keeps the TUI consistent.
        let response = match final_response {
            Some(resp) => resp,
            None if !accumulated_text.is_empty() => {
                tracing::warn!(
                    "Synthesizing partial response from {} chars of accumulated text \
                     (loop broke without final LLM response)",
                    accumulated_text.len()
                );
                LLMResponse {
                    id: String::new(),
                    content: vec![ContentBlock::Text {
                        text: accumulated_text.clone(),
                    }],
                    model: model_name.clone(),
                    usage: crate::brain::provider::TokenUsage {
                        input_tokens: total_input_tokens,
                        output_tokens: total_output_tokens,
                        cache_creation_tokens: total_cache_creation,
                        cache_read_tokens: total_cache_read,
                        ..Default::default()
                    },
                    stop_reason: Some(crate::brain::provider::StopReason::EndTurn),
                    // Synthesised from accumulated state — the real
                    // streaming windows already contributed via the
                    // per-iteration LLMResponses; this top-level
                    // synthesis is for content/usage handoff only.
                    streaming_active_secs: None,
                }
            }
            None => {
                // If the cancel token is set and was triggered, this is a user-initiated
                // cancellation — return Cancelled instead of a noisy Internal error.
                if let Some(ref token) = cancel_token
                    && token.is_cancelled()
                {
                    return Err(AgentError::Cancelled);
                }
                return Err(AgentError::Internal(
                    "Tool loop ended without final response".to_string(),
                ));
            }
        };

        // Extract text from the final response only (for TUI display).
        // Intermediate text was already shown in real-time via IntermediateText events.
        let final_text = Self::extract_text_from_response(&response);

        // The assistant message was already created and updated in real-time.
        // Now update with final token usage.

        // Calculate total cost with full cache breakdown for accurate pricing.
        // input_tokens = non-cached, cache_creation/read tracked separately.
        let billable_input = total_input_tokens + total_cache_creation + total_cache_read;
        let total_tokens = billable_input + total_output_tokens;
        let cost = self
            .provider_for_session(session_id)
            .calculate_cost_with_cache(
                &response.model,
                total_input_tokens,
                total_output_tokens,
                total_cache_creation,
                total_cache_read,
            );

        // Update message with usage info. The stashed prompt-token count
        // drives the UI ctx meter, which must show the LAST iteration's
        // prompt size — i.e. the actual context window the model just
        // saw. Summing across iterations (`billable_input`) inflates by
        // factor N: a turn with 5 tool rounds against a 22K final prompt
        // displayed as ~150K (2026-04-17 05:55 logs). Cost calculation
        // still uses the cumulative billing fields above; only the
        // displayed ctx number uses the last-iter value here.
        let stored_input_tokens: i32 = if last_iter_input_tokens > 0 {
            last_iter_input_tokens as i32
        } else {
            let overhead = self.base_context_tokens();
            (context.token_count.saturating_add(overhead as usize)) as i32
        };
        message_service
            .update_message_usage(
                assistant_db_msg.id,
                total_tokens as i32,
                cost,
                Some(stored_input_tokens),
                Some(total_cache_creation as i32),
                Some(total_cache_read as i32),
            )
            .await
            .map_err(|e| AgentError::Database(e.to_string()))?;

        // Update session token usage
        session_service
            .update_session_usage(session_id, total_tokens as i32, cost)
            .await
            .map_err(|e| AgentError::Database(e.to_string()))?;

        // Notify the TUI that this session was updated (enables live refresh when
        // a remote channel — Telegram, WhatsApp, Discord, Slack — processes a message).
        if let Some(ref tx) = self.session_updated_tx {
            let _ = tx.send(crate::brain::agent::ChannelSessionEvent::Updated(
                session_id,
            ));
        }

        // Calculate tokens per second for channel footer display.
        //
        // Numerator: provider-reported output_tokens summed across
        // every iteration of the tool loop. This is the authoritative
        // count from the provider's billing pipeline — never the
        // tiktoken approximation the TUI uses during live streaming
        // (cl100k_base over-counts Qwen/Kimi/GLM bytes by ~1.5-2×,
        // which combined with sub-second burst windows showed users
        // 700-800 tok/s for providers that genuinely run ~100 tok/s).
        //
        // Denominator: sum of per-iteration active-streaming windows
        // populated by `stream_complete` in helpers.rs. Each window
        // is the wall time between the first and last content delta
        // of that iteration, with idle gaps >1s excluded — so tool
        // execution, approval prompts, DB persistence, and the gap
        // between iterations are all left out. The result is the
        // model's actual sustained generation rate during streaming,
        // not output-divided-by-full-turn-wall-clock (which silently
        // halved the rate on tool-heavy turns).
        //
        // Guarded against burst-delivery artifacts: a near-zero active
        // window (provider dumped the response in one sub-second chunk)
        // or an implausibly high rate yields None instead of a fantasy
        // number like the 37203 tok/s observed on a glm-5.1 short reply.
        let tokens_per_second =
            compute_streaming_tok_per_sec(total_output_tokens, total_streaming_active_secs);

        // The turn is complete and the response is built — NOW (never before)
        // honor a manual provider/model switch the user made mid-turn, so an
        // automatic fallback this turn took doesn't stick over their pick.
        self.finalize_manual_switch(session_id, start_switch_epoch, &session_service)
            .await;

        Ok(AgentResponse {
            message_id: assistant_db_msg.id,
            content: final_text,
            stop_reason: response.stop_reason,
            usage: crate::brain::provider::TokenUsage {
                input_tokens: total_input_tokens,
                output_tokens: total_output_tokens,
                cache_creation_tokens: total_cache_creation,
                cache_read_tokens: total_cache_read,
                ..Default::default()
            },
            context_tokens: context.token_count as u32,
            tokens_per_second,
            cost,
            model: response.model,
            provider_name: self.provider_name_for_session(session_id),
        })
    }
}