opencrabs 0.3.18

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;

/// Strip ANSI escape codes from raw tool output before persisting to DB.
/// Prevents garbled artifacts in session history.
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() {
            self.service
                .swap_provider_for_session(self.session_id, original);
        }
    }
}

/// 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>,
        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());

        // 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,
            )
            .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.
    #[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>,
    ) -> Result<AgentResponse> {
        // 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 mut model_name = model.or_else(|| session.model.clone()).unwrap_or_else(|| {
            self.provider_for_session(session_id)
                .default_model()
                .to_string()
        });
        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);

        // 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);
                }
            }
        }

        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 {
            context.token_count += AgentContext::estimate_tokens(brain);
            context.system_brain = Some(brain.clone());
        }

        // 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(&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 — detect and attach images from paths/URLs
        let user_msg = Self::build_user_message(&user_message).await;
        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,
                        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,
                        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 mut cont_text =
                "[SYSTEM: Context was auto-compacted. The summary above includes a snapshot \
                 of recent messages before compaction.\n\
                 POST-COMPACTION PROTOCOL (follow in order):\n\
                 1. Read the compaction summary and the recent message snapshot to understand \
                 the current task, tools in use, and what you were doing.\n\
                 2. If the summary references older context you don't have in the snapshot, use \
                 `session_search` with specific keywords to find those messages. Example: if the \
                 summary mentions \"vision fallback investigation\", run session_search with that \
                 query to recover the details.\n\
                 3. If you need specific brain context, selectively load ONLY the relevant \
                 brain file (e.g. TOOLS.md, SOUL.md, USER.md). NEVER use name=\"all\".\n\
                 4. Continue the task immediately. Do NOT repeat completed work. \
                 Do NOT ask the user for instructions — you have everything you need.]"
                    .to_string();
            if !self.auto_approve_tools {
                cont_text.push_str("\n\nCRITICAL: Tool approval is REQUIRED. You MUST wait for user approval before EVERY tool execution. Do NOT batch tool calls without approval.");
            }
            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 execution loop
        let mut iteration = 0;
        let mut total_input_tokens = 0u32;
        let mut total_output_tokens = 0u32;
        let mut total_cache_creation = 0u32;
        let mut total_cache_read = 0u32;
        // 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 = 2; // Retry up to 2 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.
        // Three retries per turn is enough to break the loop without
        // letting pathological cases chew quota forever.
        let mut phantom_retries_used: u32 = 0;
        const MAX_PHANTOM_RETRIES: u32 = 3;
        // 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 the model once to actually
        // produce the answer; repeated loops are unlikely to recover so
        // one retry is enough.
        let mut empty_reasoning_retry_used: bool = false;
        // 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;
        // 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 mut cont_text =
                    "[SYSTEM: Context was auto-compacted mid-loop. The summary above includes \
                     a snapshot of recent messages. POST-COMPACTION PROTOCOL:\n\
                     1. Review the summary and snapshot to understand current task state.\n\
                     2. Use `session_search` with keywords from the summary if you need older \
                     context not in the snapshot.\n\
                     3. Continue the task immediately. Do NOT repeat completed work. \
                     Do NOT ask for instructions.]"
                        .to_string();
                if !self.auto_approve_tools {
                    cont_text.push_str("\n\nCRITICAL: Tool approval is REQUIRED. You MUST wait for user approval before EVERY tool execution. Do NOT batch tool calls without approval.");
                }
                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_registry.get_tool_definitions();
                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) => 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 mut cont_text =
                                "[SYSTEM: Emergency compaction — provider rejected the prompt as \
                                 too large. Context has been compacted. POST-COMPACTION PROTOCOL:\n\
                                 1. Review the summary to understand where you left off.\n\
                                 2. Use `session_search` with keywords if you need older context.\n\
                                 3. Briefly acknowledge the compaction with a fun/cheeky remark, \
                                 then resume the task. Do NOT repeat completed work.]"
                                    .to_string();
                            if !self.auto_approve_tools {
                                cont_text.push_str("\n\nCRITICAL: Tool approval is REQUIRED. You MUST wait for user approval before EVERY tool execution. Do NOT batch tool calls without approval.");
                            }
                            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_registry.get_tool_definitions());
                    }
                    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();
                    let provider_model_dim = format!("{}/{}", primary_from_name, model_name);

                    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 '{}/{}'", 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_registry.get_tool_definitions());
                        }

                        // 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());
                        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 provider_model_dim = format!("{}/{}", primary_from_name, model_name);
                    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_registry.get_tool_definitions());
                        }

                        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_registry.get_tool_definitions());
                            }

                            // 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());
                            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);
                                    stream_succeeded = Some(resp);
                                    // Emit ProviderSwitched so the TUI persists
                                    // the swap to the session DB and updates the
                                    // footer (matches rate-limit sticky path).
                                    if let Some(ref cb) = progress_callback {
                                        let primary_from_name =
                                            self.provider_name_for_session(session_id);
                                        cb(
                                            session_id,
                                            ProgressEvent::SelfHealingAlert {
                                                message: format!(
                                                    "Stream error → switched to {}/{}",
                                                    fb_name, fb_model
                                                ),
                                            },
                                        );
                                        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
                                    // independently of the progress callback
                                    // (channel cbs historically dropped this
                                    // event — see persist_sticky_pair docs).
                                    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 provider_model_dim = format!("{}/{}", primary_from_name, model_name);
                    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_registry.get_tool_definitions());
                        }

                        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_registry.get_tool_definitions());
                            }

                            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);
                                    // Sticky swap so subsequent iterations use this provider
                                    self.swap_provider_for_session(session_id, (*fallback).clone());
                                    // Emit ProviderSwitched so the TUI persists
                                    // the swap to the session DB and updates the footer
                                    if let Some(ref cb) = progress_callback {
                                        let primary_from_name =
                                            self.provider_name_for_session(session_id);
                                        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(),
                                            },
                                        );
                                    }
                                    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) => return Err(AgentError::Provider(e)),
            };

            // 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 {
                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;
            // 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 {
                        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 and continue immediately.]"
                        .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 text from response blocks
                    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);
                            let _ = message_service
                                .append_content(assistant_db_msg.id, &format!("{}\n\n", text))
                                .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
                        if input.as_object().map(|o| o.is_empty()).unwrap_or(true) {
                            tracing::error!(
                                "[TOOL_EXEC] ⚠️ Tool '{}' received empty input — tool call will fail",
                                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: separate writes (tool execution happens between iterations)
                // NOTE: reasoning_text is intentionally NOT persisted to DB *content*.
                // Writing <!-- reasoning --> markers teaches models (esp. qwen3.6-plus)
                // to echo them back in the content field, causing reasoning leaks.
                // Instead, reasoning is persisted to the separate `thinking` column
                // so it survives restart without polluting the content sent back to
                // the model or leaking to Telegram/Slack channels.

                if let Some(ref reasoning) = reasoning_text
                    && !reasoning.trim().is_empty()
                {
                    let _ = message_service
                        .set_thinking(assistant_db_msg.id, reasoning)
                        .await;
                }

                if !iteration_text.is_empty() {
                    if !accumulated_text.is_empty() {
                        accumulated_text.push_str("\n\n");
                    }
                    accumulated_text.push_str(&iteration_text);

                    let _ = message_service
                        .append_content(assistant_db_msg.id, &format!("{}\n\n", iteration_text))
                        .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 ──────────────────────────
                // The model narrated actions but never executed any tool
                // calls this iteration. We're already inside
                // `if tool_uses.is_empty()`, so the relaxed detector
                // `has_phantom_tool_intent_no_tools` is the right gate:
                // it looks for intent phrases ("Let me check…", "I'll
                // run…", "Now I…") that indicate the model BELIEVED it
                // was calling a tool but didn't. Up to MAX_PHANTOM_RETRIES
                // corrections per turn.
                //
                // Earlier we broadened this for local providers to fire on
                // ANY short response that produced zero tool_calls. That
                // backfired: legitimate answers (tables, status rundowns,
                // one-liner replies) from local models like MLX got
                // flagged as phantom and replaced with a confused retry
                // cycle — seen in logs 03:30:52 where a full status table
                // triggered `intent_match=false, local=true` retry.
                //
                // Since the text-based tool-call extractor now recovers
                // every Qwen/Kimi/DeepSeek leak format, if the model
                // really tried to call a tool it's already been promoted
                // to a real tool_use. Zero tool_uses + no intent phrases
                // = a legitimate text answer. Keep the gate narrow.
                if phantom_retries_used < MAX_PHANTOM_RETRIES
                    && !is_cli_provider
                    && 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(),
                            },
                        );
                    }
                    // Add the narration as assistant context so the model sees
                    // what it said, then inject a system correction. Local
                    // models respond better to Unsloth's blunter wording;
                    // cloud models get our existing, more-specific nudge.
                    context.add_message(Message::assistant(iteration_text));
                    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.]"
                    } 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.]"
                    };
                    context.add_message(Message::user(nudge.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 local reasoning runtimes (notably MLX Qwen3) finish
                // a turn with only `reasoning_content` chunks — zero visible
                // text, zero tool calls — after an earlier tool iteration.
                // The user sees the tool card and then nothing, which reads
                // as a dropped request with no self-heal. Seen in logs at
                // 2026-04-17 04:06:54 where finish_reason=stop fired right
                // after reasoning wrapped.
                //
                // One-shot retry: ask the model to actually produce the
                // answer. Bounded to avoid loops on models that will never
                // emit content in this turn shape.
                let has_meaningful_reasoning = reasoning_text
                    .as_deref()
                    .map(|r| r.trim().len() >= 40)
                    .unwrap_or(false);
                if !empty_reasoning_retry_used
                    && iteration > 0
                    && !is_cli_provider
                    && iteration_text.trim().is_empty()
                    && has_meaningful_reasoning
                {
                    empty_reasoning_retry_used = true;
                    tracing::warn!(
                        "Model ended turn with reasoning but no visible response \
                         (reasoning_len={}, iteration={}) — nudging for the actual answer.",
                        reasoning_text.as_deref().map(|r| r.len()).unwrap_or(0),
                        iteration,
                    );
                    if let Some(ref cb) = progress_callback {
                        cb(
                            session_id,
                            ProgressEvent::SelfHealingAlert {
                                message: "Model reasoned without answering — nudging for response"
                                    .into(),
                            },
                        );
                    }
                    // Preserve the reasoning on the (empty) assistant turn,
                    // then inject a user nudge so the next iteration has to
                    // produce the actual answer.
                    context.add_message(Message::assistant(String::new()));
                    context.add_message(Message::user(
                        "[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.]"
                            .to_string(),
                    ));
                    continue;
                }

                // ── 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 {
                    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;
                }
            }

            // 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,
                                        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(),
                                };

                                // 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 = if result.success {
                                            result.output
                                        } else {
                                            result.error.unwrap_or_else(|| {
                                                "Tool execution failed".to_string()
                                            })
                                        };

                                        // GRANULAR LOG: Tool execution result
                                        if success {
                                            tracing::info!(
                                                "[TOOL_EXEC] ✅ Tool '{}' executed successfully, output_len={}",
                                                tool_name,
                                                content.len()
                                            );
                                            // 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,
                                            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,
                                            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 = if result.success {
                            result.output
                        } else {
                            result
                                .error
                                .unwrap_or_else(|| "Tool execution failed".to_string())
                        };

                        // GRANULAR LOG: Direct tool execution result
                        if success {
                            tracing::info!(
                                "[TOOL_EXEC] ✅ Tool '{}' executed successfully, output_len={}",
                                tool_name,
                                content.len()
                            );
                            // 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,
                            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, 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 mut cont_text =
                    "[SYSTEM: Mid-loop context compaction complete. The summary above has \
                     full context of everything done so far. POST-COMPACTION PROTOCOL:\n\
                     1. Review the summary to understand current task state.\n\
                     2. Use `session_search` with keywords if you need older context.\n\
                     Briefly acknowledge the compaction to the user with a fun/cheeky remark (be \
                     creative, surprise them — cursing allowed), then pick up where you left off. \
                     Do NOT re-do completed work.]"
                        .to_string();
                if !self.auto_approve_tools {
                    cont_text.push_str("\n\nCRITICAL: Tool approval is REQUIRED. You MUST wait for user approval before EVERY tool execution. Do NOT batch tool calls without approval.");
                }
                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),
                }
            }
            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),
            )
            .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,
            ));
        }

        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,
            cost,
            model: response.model,
            provider_name: self.provider_name_for_session(session_id),
        })
    }
}