opencrabs 0.3.13

//! Telegram Message Handler
//!
//! Processes incoming messages: text, voice (STT/TTS), photos, image documents, allowlist enforcement.
//! Supports live streaming (edit-based) and Telegram-native approval inline keyboards.

use super::TelegramState;
use crate::brain::agent::{AgentService, ProgressCallback, ProgressEvent};
use crate::config::{Config, RespondTo};
use crate::db::ChannelMessageRepository;
use crate::db::models::ChannelMessage as DbChannelMessage;
use crate::services::SessionService;
use crate::utils::sanitize::redact_secrets;
use crate::utils::truncate_str;
use std::collections::HashSet;
use std::sync::Arc;
use teloxide::prelude::*;
use teloxide::types::{
    ChatAction, ChatKind, InlineKeyboardButton, InlineKeyboardMarkup, InputFile, MessageId,
    ParseMode,
};
use tokio::sync::Mutex;
use tokio_util::sync::CancellationToken;
use uuid::Uuid;

/// Guard that cancels a CancellationToken on drop (used for typing loop).
struct TypingGuard(CancellationToken);
impl Drop for TypingGuard {
    fn drop(&mut self) {
        self.0.cancel();
    }
}

/// Individual tool call — each gets its own Telegram message.
struct ToolMsg {
    msg_id: Option<MessageId>,
    name: String,
    context: String,
    /// None = running, Some(true) = success, Some(false) = failed
    completed: Option<bool>,
    dirty: bool,
}

/// Fun rotating status quips shown during long tool execution.
const TOOL_STATUS_QUIPS: &[&str] = &[
    "☕ Grab a coffee — my sub-agents are on fire right now",
    "🦀 My crabs are working their claws off — hang tight",
    "🔥 Still cooking... deep in the code",
    "⚡ Sub-agents going brrr — almost there",
    "🧠 Thinking hard so you don't have to",
    "🏗️ Building something beautiful — one sec",
    "🎯 Locked in — the crabs are laser-focused",
    "🚀 Full speed ahead — engines at max",
    "💪 Crunching through the code like a boss",
    "🌊 Riding the wave — results incoming",
    "🎪 The circus is in town — all crabs performing",
    "🔧 Wrenching away at it — precision work",
    "🏎️ Pedal to the metal — no brakes",
    "🧪 Experimenting... for science!",
    "🎵 Working to the rhythm — almost done",
];

/// Per-message streaming state shared between the progress callback and the edit loop.
/// Each tool call gets its own message above; response streams in a separate message below.
/// Ordered display event — preserves chronological ordering of tools and intermediate texts.
#[derive(Clone)]
enum DisplayItem {
    /// New tool at this index in tool_msgs (needs send_message)
    NewTool(usize),
    /// Intermediate text between tool rounds
    Intermediate(String),
}

struct StreamingState {
    /// Response/thinking message (always at bottom)
    msg_id: Option<MessageId>,
    /// Reasoning/thinking text — streamed live, cleared before tool calls or response
    thinking: String,
    /// Each tool call = its own individual message
    tool_msgs: Vec<ToolMsg>,
    /// Ordered queue of new display items (tools + intermediates in chronological order)
    display_queue: Vec<DisplayItem>,
    /// Response text from streaming chunks — own message at bottom
    response: String,
    dirty: bool,
    /// When true, the edit loop deletes the response message and creates a fresh one
    /// at the bottom of the chat (so it appears below tool/approval messages).
    recreate: bool,
    /// Rolling status message shown during long tool execution (single message, edited in-place)
    status_msg_id: Option<MessageId>,
    /// Number of tool rounds completed (for display)
    tool_round_count: usize,
    /// When tool execution started (for elapsed time)
    tools_started_at: Option<std::time::Instant>,
    /// Index into TOOL_STATUS_QUIPS for rotation
    quip_index: usize,
    /// When the current status quip was shown (for show/vanish timing)
    status_shown_at: Option<std::time::Instant>,
    /// Intermediate texts already sent — used to dedup final response
    sent_intermediates: Vec<String>,
    /// Message IDs of every intermediate chunk delivered to Telegram, so a
    /// cancelled in-flight call can clean up after itself. Without this, a
    /// cancelled old call leaves its intermediate visible and the new call
    /// re-sends the same text — the exact-match duplicate the user reported.
    intermediate_msg_ids: Vec<MessageId>,
    /// Message IDs of every voice note delivered to Telegram via `send_voice`
    /// (TTS responses to voice-input turns). This field exists purely as a
    /// load-bearing invariant: voice-reply IDs live here and MUST NEVER be
    /// iterated for deletion by any cleanup/cancellation/rebuild path. If a
    /// future contributor adds a bulk cleanup over message IDs they have to
    /// consciously skip this field. The user's TTS voice note is the most
    /// expensive artefact to reproduce — it's a real synthesis call, not a
    /// cheap text render — so losing it to a sweep that "looked reasonable
    /// at the time" is a regression we've deliberately made hard to introduce.
    voice_msg_ids: Vec<MessageId>,
    /// True from start until first response text arrives — enables rolling messages for CLI providers
    /// where tools complete instantly (ToolStarted+ToolCompleted back-to-back)
    processing: bool,
}

impl StreamingState {
    /// Render response message: thinking + response only (tools are separate messages).
    fn render(&self) -> String {
        let mut parts = Vec::new();
        if !self.thinking.is_empty() {
            let t = if self.thinking.len() > 800 {
                let start = self.thinking.ceil_char_boundary(self.thinking.len() - 800);
                &self.thinking[start..]
            } else {
                &self.thinking
            };
            let t = crate::utils::sanitize::strip_llm_artifacts(t.trim());
            // Collapse repeated whitespace/newlines, then add line breaks after
            // sentence-ending punctuation so thinking text reads well in Telegram.
            let t = t.split_whitespace().collect::<Vec<_>>().join(" ");
            let t = t
                .replace(". ", ".\n")
                .replace("? ", "?\n")
                .replace("! ", "!\n");
            parts.push(format!("💭 _{}_", redact_secrets(&t)));
        }
        if !self.response.is_empty() {
            let resp = crate::utils::sanitize::strip_llm_artifacts(&self.response);
            parts.push(redact_secrets(&resp));
        }
        if parts.is_empty() {
            String::new()
        } else {
            parts.join("\n\n")
        }
    }
}

#[allow(clippy::too_many_arguments)]
pub(crate) async fn handle_message(
    bot: Bot,
    msg: Message,
    agent: Arc<AgentService>,
    session_svc: SessionService,
    bot_token: Arc<String>,
    shared_session: Arc<Mutex<Option<Uuid>>>,
    telegram_state: Arc<TelegramState>,
    config_rx: tokio::sync::watch::Receiver<Config>,
    channel_msg_repo: ChannelMessageRepository,
) -> ResponseResult<()> {
    let user = match msg.from {
        Some(ref u) => u,
        None => return Ok(()),
    };

    let user_id = user.id.0 as i64;

    // /start command -- always respond with user ID (for allowlist setup)
    if let Some(text) = msg.text()
        && text.starts_with("/start")
    {
        let reply = format!(
            "OpenCrabs Telegram Bot\n\nYour user ID: {}\n\nAdd this ID to your config.toml under [channels.telegram] allowed_users to get started.",
            user_id
        );
        bot.send_message(msg.chat.id, reply).await?;
        tracing::info!(
            "Telegram: /start from user {} ({})",
            user_id,
            user.first_name
        );
        return Ok(());
    }

    // Read latest config from watch channel — single source of truth
    let cfg = config_rx.borrow().clone();
    let tg_cfg = &cfg.channels.telegram;
    let allowed: HashSet<i64> = tg_cfg
        .allowed_users
        .iter()
        .filter_map(|s| s.parse().ok())
        .collect();
    let respond_to = &tg_cfg.respond_to;
    let allowed_channels: HashSet<String> = tg_cfg.allowed_channels.iter().cloned().collect();
    let idle_timeout_hours = tg_cfg.session_idle_hours;
    let voice_config = cfg.voice_config();

    // Allowlist check — read from config (hot-reloaded via watch channel)
    if !allowed.is_empty() && !allowed.contains(&user_id) {
        tracing::debug!(
            "Telegram: ignoring message from non-allowed user {}",
            user_id
        );
        bot.send_message(
            msg.chat.id,
            "You are not authorized. Send /start to get your user ID.",
        )
        .await?;
        return Ok(());
    }

    // respond_to / allowed_channels filtering — private chats always pass
    let is_dm = matches!(msg.chat.kind, ChatKind::Private { .. });
    let chat_title = msg
        .chat
        .title()
        .unwrap_or(if is_dm { "DM" } else { "unknown" });
    let chat_kind = match &msg.chat.kind {
        ChatKind::Private { .. } => "private",
        ChatKind::Public(public) => match &public.kind {
            teloxide::types::PublicChatKind::Group { .. } => "group",
            teloxide::types::PublicChatKind::Supergroup { .. } => "supergroup",
            teloxide::types::PublicChatKind::Channel { .. } => "channel",
        },
    };

    tracing::info!(
        "Telegram: incoming msg in {} \"{}\" (chat_id={}) from {} ({}) — kind={}, text={}",
        chat_kind,
        chat_title,
        msg.chat.id.0,
        user.first_name,
        user_id,
        if msg.text().is_some() {
            "text"
        } else if msg.voice().is_some() {
            "voice"
        } else if msg.photo().is_some() {
            "photo"
        } else if msg.document().is_some() {
            "document"
        } else {
            "other"
        },
        truncate_str(msg.text().or(msg.caption()).unwrap_or(""), 60),
    );

    // Helper: passively capture a group message for channel history
    let store_channel_msg = |text: String| {
        let repo = channel_msg_repo.clone();
        let channel_chat_id = msg.chat.id.0.to_string();
        let chat_name = chat_title.to_string();
        let sender_id = user.id.0.to_string();
        let sender_name = user.first_name.clone();
        let msg_id = msg.id.0.to_string();
        async move {
            if text.is_empty() {
                return;
            }
            let cm = DbChannelMessage::new(
                "telegram".into(),
                channel_chat_id,
                Some(chat_name),
                sender_id,
                sender_name,
                text,
                "text".into(),
                Some(msg_id),
            );
            if let Err(e) = repo.insert(&cm).await {
                tracing::warn!("Failed to store channel message: {e}");
            }
        }
    };

    if !is_dm {
        let chat_id_str = msg.chat.id.0.to_string();

        // Check allowed_channels (empty = all channels allowed)
        if !allowed_channels.is_empty() && !allowed_channels.contains(&chat_id_str) {
            tracing::debug!(
                "Telegram: dropping — chat {} not in allowed_channels",
                chat_id_str
            );
            store_channel_msg(msg.text().or(msg.caption()).unwrap_or("").to_string()).await;
            return Ok(());
        }

        match respond_to {
            RespondTo::DmOnly => {
                tracing::debug!(
                    "Telegram: dropping — respond_to=dm_only, {} \"{}\"",
                    chat_kind,
                    chat_title
                );
                store_channel_msg(msg.text().or(msg.caption()).unwrap_or("").to_string()).await;
                return Ok(());
            }
            RespondTo::Mention => {
                // Check if bot is @mentioned in text or message is a reply to the bot
                let bot_username = telegram_state.bot_username().await;
                let text_content = msg.text().or(msg.caption()).unwrap_or("");

                let mentioned_by_username = bot_username
                    .as_ref()
                    .is_some_and(|uname| text_content.contains(&format!("@{}", uname)));

                let replied_to_bot = msg
                    .reply_to_message()
                    .is_some_and(|reply| reply.from.as_ref().is_some_and(|u| u.is_bot));

                tracing::info!(
                    "Telegram: group mention check — mentioned={}, replied_to_bot={}, bot_username={:?}",
                    mentioned_by_username,
                    replied_to_bot,
                    bot_username,
                );

                if !mentioned_by_username && !replied_to_bot {
                    tracing::info!(
                        "Telegram: group msg not directed at bot — {} in \"{}\" said: {}",
                        user.first_name,
                        chat_title,
                        truncate_str(text_content, 80),
                    );
                    store_channel_msg(text_content.to_string()).await;
                    return Ok(());
                }
                tracing::info!(
                    "Telegram: bot mentioned/replied in \"{}\" by {} — processing",
                    chat_title,
                    user.first_name,
                );
            }
            RespondTo::All => {
                tracing::debug!(
                    "Telegram: respond_to=all, processing {} \"{}\"",
                    chat_kind,
                    chat_title
                );
            }
        }
    }

    // Also store directed group messages for complete history
    if !is_dm {
        store_channel_msg(msg.text().or(msg.caption()).unwrap_or("").to_string()).await;
    }

    // Extract text from either text message or voice note (via STT)
    let (text, is_voice) = if let Some(t) = msg.text() {
        if t.is_empty() {
            return Ok(());
        }
        (t.to_string(), false)
    } else if let Some(voice) = msg.voice() {
        // Voice note -- transcribe via STT provider
        if !voice_config.stt_enabled {
            bot.send_message(msg.chat.id, "Voice notes are not enabled.")
                .await?;
            return Ok(());
        }

        tracing::info!(
            "Telegram: voice note from user {} ({}) — {}s",
            user_id,
            user.first_name,
            voice.duration,
        );

        // Show typing immediately so user knows we're processing
        let _ = bot
            .send_chat_action(msg.chat.id, teloxide::types::ChatAction::Typing)
            .await;

        // Download the voice file from Telegram
        let file = bot.get_file(&voice.file.id).await?;
        let download_url = format!(
            "https://api.telegram.org/file/bot{}/{}",
            bot_token.as_str(),
            file.path
        );

        let audio_bytes = match reqwest::get(&download_url).await {
            Ok(resp) => match resp.bytes().await {
                Ok(b) => b.to_vec(),
                Err(e) => {
                    tracing::error!("Telegram: failed to read voice file bytes: {}", e);
                    bot.send_message(msg.chat.id, "Failed to download voice note.")
                        .await?;
                    return Ok(());
                }
            },
            Err(e) => {
                tracing::error!("Telegram: failed to download voice file: {}", e);
                bot.send_message(msg.chat.id, "Failed to download voice note.")
                    .await?;
                return Ok(());
            }
        };

        // Transcribe with STT dispatch (API or Local based on config)
        match crate::channels::voice::transcribe(audio_bytes, &voice_config).await {
            Ok(transcript) => {
                tracing::info!(
                    "Telegram: transcribed voice: {}",
                    truncate_str(&transcript, 80)
                );
                (transcript, true)
            }
            Err(e) => {
                tracing::error!("Telegram: STT error: {}", e);
                bot.send_message(msg.chat.id, format!("Transcription error: {}", e))
                    .await?;
                return Ok(());
            }
        }
    } else if let Some(photos) = msg.photo() {
        // Photo -- download and send to agent as image attachment
        let Some(photo) = photos.last() else {
            return Ok(());
        };
        tracing::info!(
            "Telegram: photo from user {} ({}) — {}x{}",
            user_id,
            user.first_name,
            photo.width,
            photo.height,
        );

        let file = bot.get_file(&photo.file.id).await?;
        let download_url = format!(
            "https://api.telegram.org/file/bot{}/{}",
            bot_token.as_str(),
            file.path
        );

        let photo_bytes = match reqwest::get(&download_url).await {
            Ok(resp) => match resp.bytes().await {
                Ok(b) => b.to_vec(),
                Err(e) => {
                    tracing::error!("Telegram: failed to read photo bytes: {}", e);
                    bot.send_message(msg.chat.id, "Failed to download photo.")
                        .await?;
                    return Ok(());
                }
            },
            Err(e) => {
                tracing::error!("Telegram: failed to download photo: {}", e);
                bot.send_message(msg.chat.id, "Failed to download photo.")
                    .await?;
                return Ok(());
            }
        };

        // Save to temp file so the agent's <<IMG:path>> pipeline can handle it
        let tmp_path = std::env::temp_dir().join(format!("tg_photo_{}.jpg", Uuid::new_v4()));
        if let Err(e) = tokio::fs::write(&tmp_path, &photo_bytes).await {
            tracing::error!("Telegram: failed to write temp photo: {}", e);
            bot.send_message(msg.chat.id, "Failed to process photo.")
                .await?;
            return Ok(());
        }

        // Use caption if provided, otherwise generic prompt
        let caption = msg.caption().unwrap_or("Analyze this image");
        let text_with_img = format!("<<IMG:{}>> {}", tmp_path.display(), caption);

        // Clean up temp file after 24h — the agent and follow-up tool calls may
        // need to re-read the file long after the initial message was processed.
        // The OS temp reaper handles orphaned files if the daemon is killed.
        let cleanup_path = tmp_path.clone();
        tokio::spawn(async move {
            tokio::time::sleep(std::time::Duration::from_secs(24 * 3600)).await;
            let _ = tokio::fs::remove_file(cleanup_path).await;
        });

        (text_with_img, false)
    } else if let Some(doc) = msg.document() {
        let fname = doc.file_name.as_deref().unwrap_or("file");
        let mime = doc.mime_type.as_ref().map(|m| m.as_ref()).unwrap_or("");
        let caption = msg.caption().unwrap_or("");

        tracing::info!(
            "Telegram: document from user {} — name={} mime={}",
            user_id,
            fname,
            mime
        );

        let file = bot.get_file(&doc.file.id).await?;
        let download_url = format!(
            "https://api.telegram.org/file/bot{}/{}",
            bot_token.as_str(),
            file.path
        );

        let bytes = match reqwest::get(&download_url).await {
            Ok(resp) => match resp.bytes().await {
                Ok(b) => b.to_vec(),
                Err(e) => {
                    tracing::error!("Telegram: failed to read document bytes: {}", e);
                    bot.send_message(msg.chat.id, "Failed to download file.")
                        .await?;
                    return Ok(());
                }
            },
            Err(e) => {
                tracing::error!("Telegram: failed to download document: {}", e);
                bot.send_message(msg.chat.id, "Failed to download file.")
                    .await?;
                return Ok(());
            }
        };

        use crate::utils::{inject_file_content, process_file_with_vision};
        let content = process_file_with_vision(&bytes, mime, fname, &cfg);
        let result = inject_file_content(&content).0;
        let result = if caption.is_empty() || result.contains("<<IMG:") {
            result
        } else {
            format!("{caption}\n\n{result}")
        };
        (result, false)
    } else {
        // Non-text, non-voice, non-photo message -- ignore
        return Ok(());
    };

    // Log ALL processed messages (voice transcripts, photo captions, doc text) for group context.
    // Text-only messages in groups were already logged above during respond_to filtering;
    // this catches voice, photo, and document messages that bypassed the early return paths.
    if !is_dm {
        let log_content = if is_voice {
            format!("[voice] {}", truncate_str(&text, 500))
        } else if msg.photo().is_some() {
            format!("[photo] {}", msg.caption().unwrap_or(""))
        } else if msg.document().is_some() {
            format!("[document] {}", msg.caption().unwrap_or(""))
        } else {
            String::new() // text was already logged above
        };
        if !log_content.is_empty() {
            store_channel_msg(log_content).await;
        }
    }

    // Strip @bot_username suffix from ALL text (Telegram appends it in menus, even in DMs).
    // Without this, /stop@opencrabsbot won't match /stop in handle_command.
    let text = if let Some(ref uname) = telegram_state.bot_username().await {
        text.replace(&format!("@{}", uname), "").trim().to_string()
    } else {
        text
    };

    tracing::info!(
        "Telegram: {} from user {} ({}): {}",
        if is_voice { "voice" } else { "text" },
        user_id,
        user.first_name,
        truncate_str(&text, 50)
    );

    // Start typing indicator loop — cancelled via guard on all return paths
    let typing_cancel = CancellationToken::new();
    let _typing_guard = TypingGuard(typing_cancel.clone());
    tokio::spawn({
        let bot = bot.clone();
        let chat = msg.chat.id;
        let cancel = typing_cancel.clone();
        async move {
            loop {
                let _ = bot.send_chat_action(chat, ChatAction::Typing).await;
                tokio::select! {
                    _ = cancel.cancelled() => break,
                    _ = tokio::time::sleep(std::time::Duration::from_secs(4)) => {}
                }
            }
        }
    });

    // Resolve session: owner shares the TUI session, other users get their own.
    // Owner = first user in the config's allowed_users list (Vec order, not HashSet).
    let owner_id = tg_cfg
        .allowed_users
        .first()
        .and_then(|s| s.parse::<i64>().ok());
    let is_owner = allowed.is_empty() || owner_id == Some(user_id);

    tracing::info!(
        "Telegram: session resolve — is_owner={}, is_dm={}, chat=\"{}\" ({}), user={} ({})",
        is_owner,
        is_dm,
        chat_title,
        msg.chat.id.0,
        user.first_name,
        user_id,
    );

    // Track owner's chat ID for proactive messaging
    if is_owner {
        telegram_state.set_owner_chat_id(msg.chat.id.0).await;
    }

    let session_id = if is_owner && is_dm {
        // Owner DM shares the TUI's current session (or daemon's persisted session)
        let shared = shared_session.lock().await;
        match *shared {
            Some(id) => id,
            None => {
                drop(shared); // release lock before async calls
                // Try to resume the most recent active session from DB (survives daemon restarts)
                let restored = match session_svc.get_most_recent_session().await {
                    Ok(Some(session)) => {
                        tracing::info!(
                            "Telegram: restored most recent session {} for owner",
                            session.id
                        );
                        Some(session.id)
                    }
                    _ => None,
                };
                let id = match restored {
                    Some(id) => id,
                    None => {
                        tracing::info!("Telegram: no existing session, creating one for owner");
                        match crate::channels::session_init::create_channel_session(
                            &session_svc,
                            Some("Chat".to_string()),
                        )
                        .await
                        {
                            Ok(session) => session.id,
                            Err(e) => {
                                tracing::error!("Telegram: failed to create session: {}", e);
                                bot.send_message(msg.chat.id, "Internal error creating session.")
                                    .await?;
                                return Ok(());
                            }
                        }
                    }
                };
                *shared_session.lock().await = Some(id);
                id
            }
        }
    } else {
        // Non-DM-owner sessions: persisted in DB by title — survives restarts.
        let session_title = if is_dm {
            format!("Telegram: {}", user.first_name)
        } else {
            format!("Telegram: {}", chat_title)
        };

        let existing = session_svc
            .find_session_by_title(&session_title)
            .await
            .ok()
            .flatten();

        if let Some(session) = existing {
            if idle_timeout_hours.is_some_and(|h| {
                let elapsed = (chrono::Utc::now() - session.updated_at).num_seconds();
                elapsed > (h * 3600.0) as i64
            }) {
                if let Err(e) = session_svc.archive_session(session.id).await {
                    tracing::error!("Telegram: failed to archive session {}: {}", session.id, e);
                }
                match crate::channels::session_init::create_channel_session(
                    &session_svc,
                    Some(session_title),
                )
                .await
                {
                    Ok(new_session) => new_session.id,
                    Err(e) => {
                        tracing::error!("Telegram: failed to create session: {}", e);
                        bot.send_message(msg.chat.id, "Internal error creating session.")
                            .await?;
                        return Ok(());
                    }
                }
            } else {
                session.id
            }
        } else {
            match crate::channels::session_init::create_channel_session(
                &session_svc,
                Some(session_title),
            )
            .await
            {
                Ok(session) => {
                    tracing::info!(
                        "Telegram: created new channel session {} for {}",
                        session.id,
                        chat_title
                    );
                    session.id
                }
                Err(e) => {
                    tracing::error!("Telegram: failed to create session: {}", e);
                    bot.send_message(msg.chat.id, "Internal error creating session.")
                        .await?;
                    return Ok(());
                }
            }
        }
    };

    tracing::info!(
        "Telegram: resolved session={} for {} in {} \"{}\" (chat_id={})",
        session_id,
        user.first_name,
        chat_kind,
        chat_title,
        msg.chat.id.0,
    );

    // Register session → chat for approval routing
    telegram_state
        .register_session_chat(session_id, msg.chat.id.0)
        .await;

    // Restore session's own provider (each session keeps its provider independently)
    let session_meta = session_svc.get_session(session_id).await.ok().flatten();
    crate::channels::commands::sync_provider_for_session(
        &agent,
        session_meta
            .as_ref()
            .and_then(|s| s.provider_name.as_deref()),
        session_meta.as_ref().and_then(|s| s.model.as_deref()),
    )
    .await;

    // ── Channel commands (/help, /usage, /models) ──────────────────────────
    let mut text = text;
    if !is_voice {
        use crate::channels::commands::{self, ChannelCommand};
        let cmd = commands::handle_command(&text, session_id, &agent, &session_svc).await;

        // Handle simple text-response commands (Help, Usage, Evolve, Doctor, etc.)
        if let Some(reply) = commands::try_execute_text_command(&cmd).await {
            bot.send_message(msg.chat.id, md_to_html(&reply))
                .parse_mode(ParseMode::Html)
                .await?;
            return Ok(());
        }

        match cmd {
            ChannelCommand::Models(resp) => {
                let rows: Vec<Vec<InlineKeyboardButton>> = resp
                    .providers
                    .iter()
                    .map(|(name, label)| {
                        let display = if *name == resp.current_provider {
                            format!("✓ {}", label)
                        } else {
                            label.clone()
                        };
                        vec![InlineKeyboardButton::callback(
                            display,
                            format!("provider:{}", name),
                        )]
                    })
                    .collect();
                let keyboard = InlineKeyboardMarkup::new(rows);
                bot.send_message(msg.chat.id, md_to_html(&resp.text))
                    .parse_mode(ParseMode::Html)
                    .reply_markup(keyboard)
                    .await?;
                return Ok(());
            }
            ChannelCommand::NewSession => {
                let session_title = if is_dm {
                    format!("Telegram: {}", user.first_name)
                } else {
                    format!("Telegram: {}", chat_title)
                };
                if !is_owner
                    && let Ok(Some(old)) = session_svc.find_session_by_title(&session_title).await
                    && let Err(e) = session_svc.archive_session(old.id).await
                {
                    tracing::error!("Telegram: failed to archive old session {}: {}", old.id, e);
                }
                match crate::channels::session_init::create_channel_session(
                    &session_svc,
                    Some(session_title),
                )
                .await
                {
                    Ok(new_session) => {
                        if is_owner {
                            *shared_session.lock().await = Some(new_session.id);
                        }
                        telegram_state
                            .register_session_chat(new_session.id, msg.chat.id.0)
                            .await;
                        bot.send_message(msg.chat.id, "✅ New session started.")
                            .await?;
                    }
                    Err(e) => {
                        tracing::error!("Telegram: failed to create session: {}", e);
                        bot.send_message(msg.chat.id, "Failed to create session.")
                            .await?;
                    }
                }
                return Ok(());
            }
            ChannelCommand::Sessions(resp) => {
                let rows: Vec<Vec<InlineKeyboardButton>> = resp
                    .sessions
                    .iter()
                    .map(|(id, label)| {
                        let display = if *id == resp.current_session_id {
                            format!("✓ {}", label)
                        } else {
                            label.clone()
                        };
                        vec![InlineKeyboardButton::callback(
                            display,
                            format!("session:{}", id),
                        )]
                    })
                    .collect();
                let keyboard = InlineKeyboardMarkup::new(rows);
                bot.send_message(msg.chat.id, md_to_html(&resp.text))
                    .parse_mode(ParseMode::Html)
                    .reply_markup(keyboard)
                    .await?;
                return Ok(());
            }
            ChannelCommand::Stop => {
                let cancelled = telegram_state.cancel_session(session_id).await;
                let reply = if cancelled {
                    "Operation cancelled."
                } else {
                    "No operation in progress."
                };
                bot.send_message(msg.chat.id, reply).await?;
                return Ok(());
            }
            ChannelCommand::Compact => {
                bot.send_message(msg.chat.id, "⏳ Compacting context...")
                    .await?;
                text = "[SYSTEM: Compact context now. Summarize this conversation for continuity.]"
                    .to_string();
                // fall through to agent
            }
            ChannelCommand::UserPrompt(prompt) => {
                text = prompt;
                // fall through to agent with the prompt as the message
            }
            ChannelCommand::NotACommand => {} // fall through to agent
            // Help, Usage, Evolve, Doctor, UserSystem handled by try_execute_text_command above
            _ => {}
        }
    }

    // Extract replied-to message context so the agent knows what the user is referencing.
    let reply_context = msg.reply_to_message().and_then(|reply| {
        let reply_text = reply.text().or(reply.caption()).unwrap_or("").trim();
        if reply_text.is_empty() {
            return None;
        }
        let reply_sender = reply
            .from
            .as_ref()
            .map(|u| {
                if u.is_bot {
                    "assistant".to_string()
                } else {
                    u.first_name.clone()
                }
            })
            .unwrap_or_else(|| "unknown".to_string());
        Some(format!("[Replying to {reply_sender}: \"{reply_text}\"]"))
    });

    // Prepend sender identity and group context so the agent knows who and where.
    let agent_input = {
        let mut name = user.first_name.clone();
        if let Some(ref last) = user.last_name {
            name.push(' ');
            name.push_str(last);
        }
        let handle = user
            .username
            .as_ref()
            .map(|u| format!(" (@{})", u))
            .unwrap_or_default();
        if is_dm {
            if is_owner {
                text.clone()
            } else {
                format!("[Telegram DM from {name}{handle}, ID {user_id}]\n{text}")
            }
        } else {
            // Always include group context — even for the owner — so the agent
            // knows it's in a group and who is speaking.
            format!(
                "[Telegram group \"{}\" — {} from {name}{handle}]\n{text}",
                chat_title,
                if is_owner { "owner" } else { "user" },
            )
        }
    };

    // Prepend reply context if the user is replying to a specific message.
    let agent_input = if let Some(ref ctx) = reply_context {
        format!("{ctx}\n{agent_input}")
    } else {
        agent_input
    };

    // Inject recent group history so the agent has full conversation context.
    let agent_input = if !is_dm {
        let chat_id_str = msg.chat.id.0.to_string();
        match channel_msg_repo
            .recent(Some("telegram"), &chat_id_str, 30)
            .await
        {
            Ok(messages) if !messages.is_empty() => {
                let history: Vec<String> = messages
                    .iter()
                    .rev() // oldest first
                    .map(|m| {
                        let ts = m.created_at.format("%H:%M");
                        format!("[{}] {}: {}", ts, m.sender_name, m.content)
                    })
                    .collect();
                format!(
                    "[Recent group history ({} messages):\n{}\n--- end history ---]\n{}",
                    history.len(),
                    history.join("\n"),
                    agent_input
                )
            }
            _ => agent_input,
        }
    } else {
        agent_input
    };

    // Tell the LLM its text response is automatically delivered to the chat,
    // so it should NOT use telegram_send for simple text replies.
    let agent_input = format!(
        "[Channel: Telegram — your text response is automatically sent to this chat. \
         Do NOT call telegram_send to deliver your answer. Only use telegram_send for: \
         sending to a different chat_id, media, polls, buttons, reactions, or moderation.]\n{agent_input}"
    );

    // ── Streaming setup ───────────────────────────────────────────────────────
    let streaming = Arc::new(std::sync::Mutex::new(StreamingState {
        msg_id: None,
        thinking: String::new(),
        tool_msgs: Vec::new(),
        display_queue: Vec::new(),
        response: String::new(),
        dirty: false,
        recreate: false,
        status_msg_id: None,
        tool_round_count: 0,
        tools_started_at: Some(std::time::Instant::now()),
        quip_index: 0,
        status_shown_at: None,
        sent_intermediates: Vec::new(),
        intermediate_msg_ids: Vec::new(),
        voice_msg_ids: Vec::new(),
        processing: true,
    }));

    let edit_cancel = CancellationToken::new();

    // Edit loop: sends individual tool messages + streams response at bottom
    tokio::spawn({
        let bot = bot.clone();
        let chat = msg.chat.id;
        let st = streaming.clone();
        let cancel = edit_cancel.clone();
        async move {
            loop {
                tokio::select! {
                    _ = cancel.cancelled() => break,
                    _ = tokio::time::sleep(std::time::Duration::from_millis(1500)) => {
                        // ── Snapshot state under lock, then release immediately ──
                        struct Snapshot {
                            dirty: bool,
                            recreate: bool,
                            response_text: String,
                            msg_id: Option<MessageId>,
                            status_msg_id: Option<MessageId>,
                            tool_round_count: usize,
                            tools_started_at: Option<std::time::Instant>,
                            quip_index: usize,
                            status_shown_at: Option<std::time::Instant>,
                            /// Ordered display items (tools + intermediates in chronological order)
                            display_items: Vec<DisplayItem>,
                            /// Dirty tools that already have messages (need editing, not new sends)
                            tool_edits: Vec<(usize, String, Option<bool>, MessageId)>,
                            has_active_tools: bool,
                            has_intermediates: bool,
                            processing: bool,
                        }

                        let snap = {
                            let mut s = st.lock().unwrap_or_else(|e| e.into_inner());
                            let has_display = !s.display_queue.is_empty();
                            let any_tools_dirty = s.tool_msgs.iter().any(|t| t.dirty);
                            let has_active_tools = s.tool_msgs.iter().any(|t| t.completed.is_none());

                            let processing = s.processing;

                            if !s.dirty && !s.recreate && !any_tools_dirty && !has_display && !has_active_tools && !processing { continue; }

                            // Drain the ordered display queue
                            let display_items: Vec<DisplayItem> = s.display_queue.drain(..).collect();
                            let has_intermediates = display_items.iter().any(|d| matches!(d, DisplayItem::Intermediate(_)));

                            // Collect dirty tools that already have messages (for editing)
                            let tool_edits: Vec<_> = s.tool_msgs.iter().enumerate()
                                .filter(|(_, t)| t.dirty && t.msg_id.is_some())
                                .map(|(i, t)| {
                                    let label = format!("**{}**{}", t.name, t.context);
                                    (i, label, t.completed, t.msg_id.unwrap())
                                })
                                .collect();

                            // Mark tools as not dirty
                            for t in s.tool_msgs.iter_mut().filter(|t| t.dirty) {
                                t.dirty = false;
                            }

                            // Snapshot response
                            let response_text = if s.dirty || s.recreate {
                                s.render()
                            } else {
                                String::new()
                            };

                            let snap = Snapshot {
                                dirty: s.dirty,
                                recreate: s.recreate,
                                response_text,
                                msg_id: s.msg_id,
                                status_msg_id: s.status_msg_id,
                                tool_round_count: s.tool_round_count,
                                tools_started_at: s.tools_started_at,
                                quip_index: s.quip_index,
                                status_shown_at: s.status_shown_at,
                                display_items,
                                tool_edits,
                                has_active_tools,
                                has_intermediates,
                                processing,
                            };

                            // Pre-clear state that will be handled
                            if s.recreate {
                                s.recreate = false;
                            }
                            if s.dirty {
                                s.dirty = false;
                            }
                            // Clear status tracking if content arriving
                            if snap.has_intermediates || (snap.dirty && !snap.response_text.is_empty()) {
                                s.status_msg_id = None;
                                s.tools_started_at = None;
                                s.tool_round_count = 0;
                            }

                            snap
                        };
                        // Lock is now released

                        // ── Ordered display: tools and intermediates in chronological order ──
                        for item in &snap.display_items {
                            match item {
                                DisplayItem::NewTool(idx) => {
                                    let tool_info = {
                                        let s = st.lock().unwrap_or_else(|e| e.into_inner());
                                        s.tool_msgs.get(*idx).map(|t| {
                                            let label = format!("**{}**{}", t.name, t.context);
                                            (label, t.completed, t.msg_id)
                                        })
                                    };
                                    if let Some((label, completed, existing_mid)) = tool_info {
                                        let text = match completed {
                                            None => format!("⚙️ {}", label),
                                            Some(true) => format!("✅ {}", label),
                                            Some(false) => format!("❌ {}", label),
                                        };
                                        let html = markdown_to_telegram_html(&text);
                                        if existing_mid.is_none()
                                            && let Ok(m) = bot
                                                .send_message(chat, &html)
                                                .parse_mode(ParseMode::Html)
                                                .await
                                        {
                                            let mut s = st.lock().unwrap_or_else(|e| e.into_inner());
                                            if let Some(tool) = s.tool_msgs.get_mut(*idx) {
                                                tool.msg_id = Some(m.id);
                                            }
                                        }
                                    }
                                }
                                DisplayItem::Intermediate(text) => {
                                    // Apply the same sanitization chain as the
                                    // final response path: strip LLM artifacts
                                    // AND redact secrets. Without the redact
                                    // step here, an intermediate carrying a
                                    // Drive URL `…/file/d/<id>/view` goes out
                                    // with the raw id, while the final-response
                                    // edit of the same text gets redacted to
                                    // `[REDACTED_TOKEN]`. Two different strings
                                    // → dedup's substring replace fails → both
                                    // shown verbatim as back-to-back duplicate
                                    // messages (2026-04-18 20:57 + 21:21 TG
                                    // screenshots). Redact here so both sides
                                    // match.
                                    let text =
                                        crate::utils::sanitize::strip_llm_artifacts(text);
                                    let text = redact_secrets(&text);

                                    // Pre-send dedup: if this exact text was
                                    // already delivered as an intermediate in
                                    // this turn, skip. Downstream dedup only
                                    // strips the final-placeholder edit — it
                                    // cannot un-send intermediates already in
                                    // the chat. Twin intermediates from a
                                    // retry loop (e.g. truncation-retry
                                    // firing on a URL-terminated response
                                    // and producing the same text in
                                    // iteration N+1) would otherwise land
                                    // verbatim twice (2026-04-18 23:12/23:13).
                                    {
                                        let s = st.lock().unwrap_or_else(|e| e.into_inner());
                                        if s.sent_intermediates.iter().any(|prev| prev == &text) {
                                            tracing::info!(
                                                "Telegram: suppressing duplicate intermediate (len={})",
                                                text.len()
                                            );
                                            continue;
                                        }
                                    }

                                    let html = markdown_to_telegram_html(&text);
                                    if !html.is_empty() {
                                        // Chunk to 4096 and only record as delivered if every
                                        // chunk succeeded — if we record on failure, dedup later
                                        // strips text the user never saw.
                                        let chunks: Vec<String> = split_message(&html, 4096)
                                            .into_iter()
                                            .map(|s| s.to_string())
                                            .collect();
                                        let mut sent_ids: Vec<MessageId> = Vec::new();
                                        let mut all_ok = true;
                                        for chunk in &chunks {
                                            match send_html_or_plain(&bot, chat, chunk).await {
                                                Ok(id) => sent_ids.push(id),
                                                Err(e) => {
                                                    tracing::warn!(
                                                        "Telegram edit-loop intermediate send failed ({e}) — NOT marking as delivered; final response will carry it",
                                                    );
                                                    all_ok = false;
                                                    break;
                                                }
                                            }
                                        }
                                        if all_ok {
                                            let mut s = st.lock().unwrap_or_else(|e| e.into_inner());
                                            s.sent_intermediates.push(text.clone());
                                            s.intermediate_msg_ids.extend(sent_ids);
                                        }
                                    }
                                }
                            }
                        }

                        // ── Edit existing tool messages (status updates) ──
                        for (idx, label, completed, mid) in &snap.tool_edits {
                            let _ = idx; // used for identification only
                            let text = match completed {
                                None => format!("⚙️ {}", label),
                                Some(true) => format!("✅ {}", label),
                                Some(false) => format!("❌ {}", label),
                            };
                            let html = markdown_to_telegram_html(&text);
                            let _ = bot
                                .edit_message_text(chat, *mid, &html)
                                .parse_mode(ParseMode::Html)
                                .await;
                        }

                        // ── Rolling status quips during processing ──
                        // Show quips when: tools are active (non-CLI), OR tools ran but no
                        // response yet (CLI inter-tool), OR still processing (CLI initial wait).
                        let show_quips = snap.has_active_tools
                            || (snap.tool_round_count > 0 && snap.response_text.is_empty())
                            || snap.processing;
                        if show_quips {
                            let now = std::time::Instant::now();
                            let shown_elapsed = snap.status_shown_at
                                .map(|t| now.duration_since(t).as_secs())
                                .unwrap_or(999);

                            if snap.status_msg_id.is_some() && shown_elapsed >= 5 {
                                if let Some(mid) = snap.status_msg_id {
                                    let _ = bot.delete_message(chat, mid).await;
                                }
                                let mut s = st.lock().unwrap_or_else(|e| e.into_inner());
                                s.status_msg_id = None;
                                s.status_shown_at = Some(now);
                            } else if snap.status_msg_id.is_none() && shown_elapsed >= 2 {
                                let elapsed_total = snap.tools_started_at
                                    .map(|t| t.elapsed().as_secs())
                                    .unwrap_or(0);
                                let quip = TOOL_STATUS_QUIPS[snap.quip_index % TOOL_STATUS_QUIPS.len()];

                                let mut status = if snap.tool_round_count > 0 {
                                    format!("{} ({} tools", quip, snap.tool_round_count)
                                } else {
                                    format!("{} (thinking", quip)
                                };
                                if elapsed_total >= 5 {
                                    let mins = elapsed_total / 60;
                                    let secs = elapsed_total % 60;
                                    if mins > 0 {
                                        status.push_str(&format!(", {}m {}s", mins, secs));
                                    } else {
                                        status.push_str(&format!(", {}s", secs));
                                    }
                                }
                                status.push(')');

                                if let Ok(m) = bot.send_message(chat, &status).await {
                                    let mut s = st.lock().unwrap_or_else(|e| e.into_inner());
                                    s.status_msg_id = Some(m.id);
                                    s.status_shown_at = Some(now);
                                    s.quip_index += 1;
                                }
                            }
                        }

                        // ── Delete status when real content arrives ──
                        if (snap.has_intermediates || (snap.dirty && !snap.response_text.is_empty()))
                            && let Some(mid) = snap.status_msg_id
                        {
                            let _ = bot.delete_message(chat, mid).await;
                        }

                        // ── Response message (thinking + response, always at bottom) ──
                        if snap.dirty || snap.recreate {
                            if snap.recreate
                                && let Some(old_mid) = snap.msg_id
                            {
                                let _ = bot.delete_message(chat, old_mid).await;
                                let mut s = st.lock().unwrap_or_else(|e| e.into_inner());
                                s.msg_id = None;
                            }
                            if !snap.response_text.is_empty() {
                                // Delete status msg if still present
                                if let Some(mid) = snap.status_msg_id {
                                    let _ = bot.delete_message(chat, mid).await;
                                    let mut s = st.lock().unwrap_or_else(|e| e.into_inner());
                                    s.status_msg_id = None;
                                }
                                let current_msg_id = {
                                    let s = st.lock().unwrap_or_else(|e| e.into_inner());
                                    s.msg_id
                                };
                                if current_msg_id.is_none()
                                    && let Ok(m) = bot.send_message(chat, "\u{258b}").await
                                {
                                    let mut s = st.lock().unwrap_or_else(|e| e.into_inner());
                                    s.msg_id = Some(m.id);
                                }
                                let msg_id = {
                                    let s = st.lock().unwrap_or_else(|e| e.into_inner());
                                    s.msg_id
                                };
                                if let Some(mid) = msg_id {
                                    let html = markdown_to_telegram_html(&snap.response_text);
                                    let display = format!("{}\u{258b}", html); // ▋ cursor
                                    let _ = bot
                                        .edit_message_text(chat, mid, display)
                                        .parse_mode(ParseMode::Html)
                                        .await;
                                }
                            }
                        }

                        // Re-send typing indicator after any bot message
                        let _ = bot.send_chat_action(chat, ChatAction::Typing).await;
                    }
                }
            }
        }
    });

    // Progress callback: accumulates streaming chunks + tool status into shared state
    let progress_cb: ProgressCallback = {
        let st = streaming.clone();
        Arc::new(move |_sid, event| {
            match event {
                ProgressEvent::ReasoningChunk { text } => {
                    if let Ok(mut s) = st.lock() {
                        s.thinking.push_str(&text);
                        s.dirty = true;
                    }
                }
                ProgressEvent::StreamingChunk { text } => {
                    if let Ok(mut s) = st.lock() {
                        if !s.thinking.is_empty() {
                            s.thinking.clear();
                        }
                        s.response.push_str(&text);
                        s.dirty = true;
                        s.processing = false; // first real text = stop rolling messages
                    }
                }
                ProgressEvent::ToolStarted {
                    tool_name,
                    tool_input,
                } => {
                    if let Ok(mut s) = st.lock() {
                        s.thinking.clear();
                        if s.tools_started_at.is_none() {
                            s.tools_started_at = Some(std::time::Instant::now());
                        }
                        let ctx = tool_context(&tool_name, &tool_input);
                        let idx = s.tool_msgs.len();
                        s.tool_msgs.push(ToolMsg {
                            msg_id: None,
                            name: tool_name,
                            context: ctx,
                            completed: None,
                            dirty: true,
                        });
                        s.display_queue.push(DisplayItem::NewTool(idx));
                    }
                }
                ProgressEvent::ToolCompleted {
                    tool_name, success, ..
                } => {
                    if let Ok(mut s) = st.lock() {
                        s.tool_round_count += 1;
                        if let Some(tool) = s
                            .tool_msgs
                            .iter_mut()
                            .rev()
                            .find(|t| t.name == tool_name && t.completed.is_none())
                        {
                            tool.completed = Some(success);
                            tool.dirty = true;
                        }
                        // Push response to bottom so it stays below tool/approval messages
                        if s.msg_id.is_some() {
                            s.recreate = true;
                        }
                    }
                }
                ProgressEvent::IntermediateText { text, reasoning } => {
                    if let Ok(mut s) = st.lock() {
                        s.thinking.clear();
                        // Clear accumulated streaming response — it's now captured
                        // as an intermediate message. Without this, text from
                        // consecutive tool rounds gets concatenated without spacing.
                        s.response.clear();
                        // Delete the streaming message so stale text doesn't linger
                        if s.msg_id.is_some() {
                            s.recreate = true;
                        }
                        // Use reasoning as fallback when model produces no text
                        // blocks between tool rounds (only thinking + tool_use).
                        let content = if text.is_empty() {
                            reasoning.unwrap_or_default()
                        } else {
                            text
                        };
                        if !content.is_empty() {
                            s.display_queue.push(DisplayItem::Intermediate(content));
                        }
                    }
                }
                ProgressEvent::SelfHealingAlert { message } => {
                    if let Ok(mut s) = st.lock() {
                        s.display_queue
                            .push(DisplayItem::Intermediate(format!("🔧 {}", message)));
                    }
                }
                _ => {}
            }
        })
    };

    // Build Telegram-native approval callback for this session
    let approval_cb = make_approval_callback(telegram_state.clone());

    // ── Agent call ────────────────────────────────────────────────────────────
    let cancel_token = tokio_util::sync::CancellationToken::new();
    telegram_state
        .store_cancel_token(session_id, cancel_token.clone())
        .await;

    let chat_id_str = msg.chat.id.0.to_string();
    let result = agent
        .send_message_with_tools_and_callback(
            session_id,
            agent_input.clone(),
            None,
            Some(cancel_token.clone()),
            Some(approval_cb),
            Some(progress_cb.clone()),
            "telegram",
            Some(&chat_id_str),
        )
        .await;

    // If session lookup failed (DB contention on restart), create a fresh session and retry once
    let result = if let Err(ref e) = result {
        let es = e.to_string();
        if es.contains("Failed to get session") || es.contains("Session not found") {
            tracing::warn!(
                "Telegram: session {} lookup failed ({}), creating fresh session and retrying",
                session_id,
                es
            );
            match crate::channels::session_init::create_channel_session(
                &session_svc,
                Some("Chat".to_string()),
            )
            .await
            {
                Ok(new_session) => {
                    let new_id = new_session.id;
                    if is_owner {
                        *shared_session.lock().await = Some(new_id);
                    }
                    telegram_state
                        .register_session_chat(new_id, msg.chat.id.0)
                        .await;
                    let approval_cb2 = make_approval_callback(telegram_state.clone());
                    let cancel_token2 = tokio_util::sync::CancellationToken::new();
                    telegram_state
                        .store_cancel_token(new_id, cancel_token2.clone())
                        .await;
                    let retry_result = agent
                        .send_message_with_tools_and_callback(
                            new_id,
                            agent_input,
                            None,
                            Some(cancel_token2),
                            Some(approval_cb2),
                            Some(progress_cb),
                            "telegram",
                            Some(&chat_id_str),
                        )
                        .await;
                    telegram_state.remove_cancel_token(new_id).await;
                    retry_result
                }
                Err(e2) => {
                    tracing::error!("Telegram: failed to create fallback session: {}", e2);
                    result
                }
            }
        } else {
            result
        }
    } else {
        result
    };

    // Clean up cancel token
    telegram_state.remove_cancel_token(session_id).await;

    // Stop edit loop — final content will be written below
    edit_cancel.cancel();
    // _typing_guard drop cancels typing loop

    // Grab streaming message id and clean up status message
    let (streaming_msg_id, status_msg_id, remaining_display) = {
        let mut s = streaming.lock().unwrap_or_else(|e| e.into_inner());
        let display: Vec<DisplayItem> = s.display_queue.drain(..).collect();
        (s.msg_id, s.status_msg_id, display)
    };
    // Delete rolling status message if still present
    if let Some(mid) = status_msg_id {
        let _ = bot.delete_message(msg.chat.id, mid).await;
    }

    // Guard against stale delivery BEFORE sending remaining display items:
    // if a newer message cancelled this call, any queued tool/intermediate
    // messages are stale and must not be sent — otherwise they duplicate
    // alongside the newer call's messages.
    if cancel_token.is_cancelled() {
        tracing::info!(
            "Telegram: agent call for session {} finished after cancellation — suppressing stale delivery",
            session_id
        );
        // Voice-input + TTS case: the TTS block later in handle_message
        // (line ~1727) only fires on the Ok arm of the agent result, so a
        // cancelled voice-input turn silently drops the TTS reply. That
        // looks to the user like "my voice reply disappeared" — log it
        // specifically so the drop is traceable in logs instead of being
        // indistinguishable from a send_voice failure.
        if is_voice && voice_config.tts_enabled {
            tracing::warn!(
                "Telegram: voice-input turn cancelled before TTS synthesis for session {} \
                 — user sent a new message while this turn was in-flight, so no voice reply \
                 will be synthesized for this request (text intermediates already delivered are kept).",
                session_id
            );
        }
        // Only delete the streaming placeholder (the typing
        // indicator). Keep the intermediate content and tool-call
        // bubbles that were already posted — those are chat history
        // the user wants to see. Previous behavior (dd9eedf Apr 17)
        // deleted both to prevent duplicate intermediates on the
        // replacement turn, but the pre-send dedup in the edit-loop
        // now blocks duplicates in-turn and cross-turn restating is
        // rare enough to tolerate. User explicitly asked 2026-04-18
        // not to remove prior chat on follow-up messages.
        if let Some(mid) = streaming_msg_id {
            let _ = bot.delete_message(msg.chat.id, mid).await;
        }
        return Ok(());
    }

    // Send any remaining display items that weren't flushed by the edit loop
    for item in remaining_display {
        match item {
            DisplayItem::NewTool(idx) => {
                let tool_info = {
                    let s = streaming.lock().unwrap_or_else(|e| e.into_inner());
                    s.tool_msgs.get(idx).map(|t| {
                        let label = format!("**{}**{}", t.name, t.context);
                        (label, t.completed, t.msg_id)
                    })
                };
                if let Some((label, completed, existing_mid)) = tool_info {
                    let text = match completed {
                        None => format!("⚙️ {}", label),
                        Some(true) => format!("✅ {}", label),
                        Some(false) => format!("❌ {}", label),
                    };
                    let html = markdown_to_telegram_html(&text);
                    if existing_mid.is_none()
                        && let Ok(m) = bot
                            .send_message(msg.chat.id, &html)
                            .parse_mode(ParseMode::Html)
                            .await
                    {
                        let mut s = streaming.lock().unwrap_or_else(|e| e.into_inner());
                        if let Some(tool) = s.tool_msgs.get_mut(idx) {
                            tool.msg_id = Some(m.id);
                        }
                    }
                }
            }
            DisplayItem::Intermediate(text) => {
                let text = crate::utils::sanitize::strip_llm_artifacts(&text);
                let text = redact_secrets(&text);
                // Pre-send dedup — see matching block in edit-loop above.
                {
                    let s = streaming.lock().unwrap_or_else(|e| e.into_inner());
                    if s.sent_intermediates.iter().any(|prev| prev == &text) {
                        tracing::info!(
                            "Telegram: suppressing duplicate intermediate (len={})",
                            text.len()
                        );
                        continue;
                    }
                }
                let html = markdown_to_telegram_html(&text);
                if !html.is_empty() {
                    // Chunk to Telegram's 4096-char limit and send each chunk.
                    // Only record as "sent" if every chunk succeeded — otherwise
                    // the dedup pass on the final response would strip a message
                    // the user never actually saw, leaving them with no reply.
                    let chunks: Vec<String> = split_message(&html, 4096)
                        .into_iter()
                        .map(|s| s.to_string())
                        .collect();
                    let mut sent_ids: Vec<MessageId> = Vec::new();
                    let mut all_ok = true;
                    for chunk in &chunks {
                        match send_html_or_plain(&bot, msg.chat.id, chunk).await {
                            Ok(id) => sent_ids.push(id),
                            Err(e) => {
                                tracing::warn!(
                                    "Telegram intermediate send failed ({e}) — NOT marking as delivered; final response will carry it",
                                );
                                all_ok = false;
                                break;
                            }
                        }
                    }
                    if all_ok {
                        let mut s = streaming.lock().unwrap_or_else(|e| e.into_inner());
                        s.sent_intermediates.push(text.clone());
                        s.intermediate_msg_ids.extend(sent_ids);
                    }
                }
            }
        }
    }

    tracing::info!(
        "Telegram: agent call completed for session {} — delivering final response",
        session_id
    );

    // ── Final response ────────────────────────────────────────────────────────
    match result {
        Ok(response) => {
            // Extract <<IMG:path>> markers — send each as a Telegram photo.
            let (text_only, img_paths) = crate::utils::extract_img_markers(&response.content);
            // Strip LLM-hallucinated artifacts (<!-- tools-v2 -->, XML tool blocks)
            let text_only = crate::utils::sanitize::strip_llm_artifacts(&text_only);
            let text_only = redact_secrets(&text_only);

            // Dedup: strip text that was already sent as intermediate messages
            // to avoid duplicating content on Telegram. An intermediate chunk
            // that already carries the final answer (e.g. "Done. Uploaded to
            // Drive: https://…") will otherwise be repeated when the
            // streaming placeholder is edited with the final response.
            // Intermediates stay visible as-is; only the streaming
            // placeholder's final text is pruned.
            let sent = {
                let s = streaming.lock().unwrap_or_else(|e| e.into_inner());
                s.sent_intermediates.clone()
            };
            tracing::info!(
                "Telegram dedup: response.content len={}, sent_intermediates count={}",
                text_only.len(),
                sent.len(),
            );
            let text_only = if !sent.is_empty() {
                let mut remaining = text_only.clone();
                for intermediate in &sent {
                    remaining = remaining.replace(intermediate.as_str(), "");
                }
                let result = remaining.trim().to_string();
                if result != text_only {
                    tracing::info!(
                        "Telegram dedup: stripped {} chars, remaining len={}",
                        text_only.len() - result.len(),
                        result.len()
                    );
                }
                result
            } else {
                text_only
            };

            for img_path in img_paths {
                match tokio::fs::read(&img_path).await {
                    Ok(bytes) => {
                        if let Err(e) = bot.send_photo(msg.chat.id, InputFile::memory(bytes)).await
                        {
                            tracing::error!("Telegram: failed to send generated image: {}", e);
                        }
                    }
                    Err(e) => {
                        tracing::error!("Telegram: failed to read image {}: {}", img_path, e);
                    }
                }
            }

            // Deliver final response — prefer editing the streaming message in-place
            // to avoid the delete+send race that causes duplicates.
            let html = markdown_to_telegram_html(&text_only);
            if !html.is_empty() {
                let chunks: Vec<String> = split_message(&html, 4096)
                    .into_iter()
                    .map(|s| s.to_string())
                    .collect();

                // If single chunk and we have a streaming message, edit it in-place
                if chunks.len() == 1
                    && let Some(mid) = streaming_msg_id
                {
                    match bot
                        .edit_message_text(msg.chat.id, mid, &chunks[0])
                        .parse_mode(ParseMode::Html)
                        .await
                    {
                        Ok(_) => {}
                        Err(teloxide::RequestError::RetryAfter(secs)) => {
                            tracing::warn!(
                                "Telegram: edit rate-limited, waiting {}s",
                                secs.seconds()
                            );
                            tokio::time::sleep(secs.duration()).await;
                            if let Err(e) = bot
                                .edit_message_text(msg.chat.id, mid, &chunks[0])
                                .parse_mode(ParseMode::Html)
                                .await
                            {
                                tracing::warn!(
                                    "Telegram: edit retry failed ({e}), falling back to delete+send"
                                );
                                let _ = bot.delete_message(msg.chat.id, mid).await;
                                let _ = send_html_or_plain(&bot, msg.chat.id, &chunks[0]).await;
                            }
                        }
                        Err(e) => {
                            tracing::warn!(
                                "Telegram: edit final failed ({e}), falling back to delete+send"
                            );
                            let _ = bot.delete_message(msg.chat.id, mid).await;
                            let _ = send_html_or_plain(&bot, msg.chat.id, &chunks[0]).await;
                        }
                    }
                } else {
                    // Multi-chunk or no streaming message — delete old, send new
                    if let Some(mid) = streaming_msg_id {
                        let _ = bot.delete_message(msg.chat.id, mid).await;
                    }
                    for chunk in &chunks {
                        let _ = send_html_or_plain(&bot, msg.chat.id, chunk).await;
                    }
                }
            } else if let Some(mid) = streaming_msg_id {
                // Empty final text — just clean up the streaming placeholder
                let _ = bot.delete_message(msg.chat.id, mid).await;
            }

            // Record the bot's text reply into channel_messages for group chats
            // so the recent() query that builds conversation context on the NEXT
            // turn sees both sides. Without this, `channel_msg_repo.recent()`
            // returns user messages only — the bot loads a one-sided transcript
            // on every group turn and effectively talks to itself in the dark.
            // DMs skip this: they already use the session's messages table
            // directly for context and don't touch channel_msg_repo.
            if !is_dm && !text_only.trim().is_empty() {
                let bot_display_name = telegram_state
                    .bot_username()
                    .await
                    .map(|u| format!("@{}", u))
                    .unwrap_or_else(|| "OpenCrabs".to_string());
                let cm = DbChannelMessage::new(
                    "telegram".to_string(),
                    msg.chat.id.0.to_string(),
                    Some(chat_title.to_string()),
                    "bot:opencrabs".to_string(),
                    bot_display_name,
                    text_only.clone(),
                    "text".to_string(),
                    None,
                );
                if let Err(e) = channel_msg_repo.insert(&cm).await {
                    tracing::warn!(
                        "Telegram: failed to record bot reply in channel_messages: {}",
                        e
                    );
                }
            }

            // If input was voice AND TTS is enabled, also send voice note after text
            if is_voice && voice_config.tts_enabled {
                tracing::info!(
                    "Telegram: TTS requested — synthesizing response text (len={})",
                    response.content.len()
                );
                match crate::channels::voice::synthesize(&response.content, &voice_config).await {
                    Ok(audio_bytes) => {
                        tracing::info!(
                            "Telegram: TTS succeeded — {} bytes of audio, sending to chat {}",
                            audio_bytes.len(),
                            msg.chat.id
                        );
                        match bot
                            .send_voice(msg.chat.id, InputFile::memory(audio_bytes))
                            .await
                        {
                            Ok(m) => {
                                tracing::info!(
                                    "Telegram: voice message delivered (msg_id={})",
                                    m.id
                                );
                                // Record the delivered voice message ID in
                                // the isolated voice_msg_ids list. Cleanup
                                // paths do not touch this list. See the
                                // field doc on StreamingState.
                                let mut s = streaming.lock().unwrap_or_else(|e| e.into_inner());
                                s.voice_msg_ids.push(m.id);
                            }
                            Err(e) => {
                                tracing::error!("Telegram: send_voice failed — {}: {:?}", e, e);
                            }
                        }
                    }
                    Err(e) => {
                        tracing::error!("Telegram: TTS synthesis failed: {:#}", e);
                    }
                }
            }
        }
        Err(ref e) if matches!(e, crate::brain::agent::AgentError::Cancelled) => {
            tracing::info!("Telegram: agent call cancelled for session {}", session_id);
            // Silently clean up — user already received "Operation cancelled." from /stop
            if let Some(mid) = streaming_msg_id {
                let _ = bot.delete_message(msg.chat.id, mid).await;
            }
        }
        Err(e) => {
            tracing::error!("Telegram: agent error: {}", e);
            // If a streaming message was started, edit it to show the error
            if let Some(mid) = streaming_msg_id {
                let _ = bot
                    .edit_message_text(msg.chat.id, mid, format!("Error: {}", e))
                    .await;
            } else {
                bot.send_message(msg.chat.id, format!("Error: {}", e))
                    .await?;
            }
        }
    }

    Ok(())
}

/// Resume an interrupted session with full streaming (typing, tool messages, edit loop).
/// Called from ui.rs on startup when pending Telegram requests are detected.
pub(crate) async fn resume_session(
    bot: Bot,
    chat_id: ChatId,
    session_id: Uuid,
    prompt: String,
    agent: Arc<AgentService>,
    telegram_state: Arc<TelegramState>,
) -> anyhow::Result<()> {
    tracing::info!(
        "Telegram: resume_session {} with full streaming pipeline",
        session_id
    );

    // ── Typing indicator ────────────────────────────────────────────────────
    let typing_cancel = CancellationToken::new();
    let _typing_guard = TypingGuard(typing_cancel.clone());
    tokio::spawn({
        let bot = bot.clone();
        let cancel = typing_cancel.clone();
        async move {
            loop {
                tokio::select! {
                    _ = cancel.cancelled() => break,
                    _ = tokio::time::sleep(std::time::Duration::from_secs(4)) => {
                        let _ = bot.send_chat_action(chat_id, ChatAction::Typing).await;
                    }
                }
            }
        }
    });

    // ── Streaming setup ────────────────────────────────────────────────────
    let streaming = Arc::new(std::sync::Mutex::new(StreamingState {
        msg_id: None,
        thinking: String::new(),
        tool_msgs: Vec::new(),
        display_queue: Vec::new(),
        response: String::new(),
        dirty: false,
        recreate: false,
        status_msg_id: None,
        tool_round_count: 0,
        tools_started_at: Some(std::time::Instant::now()),
        quip_index: 0,
        status_shown_at: None,
        sent_intermediates: Vec::new(),
        intermediate_msg_ids: Vec::new(),
        voice_msg_ids: Vec::new(),
        processing: true,
    }));

    let edit_cancel = CancellationToken::new();

    // Edit loop — same as handle_message
    tokio::spawn({
        let bot = bot.clone();
        let st = streaming.clone();
        let cancel = edit_cancel.clone();
        async move {
            loop {
                tokio::select! {
                    _ = cancel.cancelled() => break,
                    _ = tokio::time::sleep(std::time::Duration::from_millis(1500)) => {
                        struct Snap {
                            dirty: bool,
                            recreate: bool,
                            response_text: String,
                            msg_id: Option<MessageId>,
                            display_items: Vec<DisplayItem>,
                        }

                        let snap = {
                            let mut s = st.lock().unwrap_or_else(|e| e.into_inner());
                            let has_display = !s.display_queue.is_empty();
                            if !s.dirty && !s.recreate && !has_display { continue; }
                            let items: Vec<DisplayItem> = s.display_queue.drain(..).collect();
                            let response_text = s.render();
                            let snap = Snap {
                                dirty: s.dirty,
                                recreate: s.recreate,
                                response_text,
                                msg_id: s.msg_id,
                                display_items: items,
                            };
                            s.dirty = false;
                            s.recreate = false;
                            snap
                        };

                        // Process display items (tools + intermediates)
                        for item in snap.display_items {
                            match item {
                                DisplayItem::NewTool(idx) => {
                                    let tool_info = {
                                        let s = st.lock().unwrap_or_else(|e| e.into_inner());
                                        s.tool_msgs.get(idx).map(|t| {
                                            let label = format!("**{}**{}", t.name, t.context);
                                            (label, t.completed, t.msg_id)
                                        })
                                    };
                                    if let Some((label, completed, existing_mid)) = tool_info {
                                        let text = match completed {
                                            None => format!("⚙️ {}", label),
                                            Some(true) => format!("✅ {}", label),
                                            Some(false) => format!("❌ {}", label),
                                        };
                                        let html = markdown_to_telegram_html(&text);
                                        if existing_mid.is_none()
                                            && let Ok(m) = bot
                                                .send_message(chat_id, &html)
                                                .parse_mode(ParseMode::Html)
                                                .await
                                        {
                                            let mut s = st.lock().unwrap_or_else(|e| e.into_inner());
                                            if let Some(tool) = s.tool_msgs.get_mut(idx) {
                                                tool.msg_id = Some(m.id);
                                            }
                                        }
                                    }
                                }
                                DisplayItem::Intermediate(text) => {
                                    let text = crate::utils::sanitize::strip_llm_artifacts(&text);
                                    let text = redact_secrets(&text);
                                    // Pre-send dedup — see handle_message.
                                    {
                                        let s = st.lock().unwrap_or_else(|e| e.into_inner());
                                        if s.sent_intermediates.iter().any(|prev| prev == &text) {
                                            tracing::info!(
                                                "Telegram resume: suppressing duplicate intermediate (len={})",
                                                text.len()
                                            );
                                            continue;
                                        }
                                    }
                                    let html = markdown_to_telegram_html(&text);
                                    if !html.is_empty() {
                                        let chunks: Vec<String> = split_message(&html, 4096)
                                            .into_iter()
                                            .map(|s| s.to_string())
                                            .collect();
                                        let mut sent_ids: Vec<MessageId> = Vec::new();
                                        let mut all_ok = true;
                                        for chunk in &chunks {
                                            match send_html_or_plain(&bot, chat_id, chunk).await {
                                                Ok(id) => sent_ids.push(id),
                                                Err(e) => {
                                                    tracing::warn!(
                                                        "Telegram (voice) edit-loop intermediate send failed ({e}) — NOT marking as delivered",
                                                    );
                                                    all_ok = false;
                                                    break;
                                                }
                                            }
                                        }
                                        if all_ok {
                                            let mut s = st.lock().unwrap_or_else(|e| e.into_inner());
                                            s.sent_intermediates.push(text.clone());
                                            s.intermediate_msg_ids.extend(sent_ids);
                                        }
                                    }
                                }
                            }
                        }

                        // Response message (streaming)
                        if snap.dirty || snap.recreate {
                            if snap.recreate
                                && let Some(old_mid) = snap.msg_id
                            {
                                let _ = bot.delete_message(chat_id, old_mid).await;
                                let mut s = st.lock().unwrap_or_else(|e| e.into_inner());
                                s.msg_id = None;
                            }
                            if !snap.response_text.is_empty() {
                                let current_msg_id = {
                                    let s = st.lock().unwrap_or_else(|e| e.into_inner());
                                    s.msg_id
                                };
                                if current_msg_id.is_none()
                                    && let Ok(m) = bot.send_message(chat_id, "\u{258b}").await
                                {
                                    let mut s = st.lock().unwrap_or_else(|e| e.into_inner());
                                    s.msg_id = Some(m.id);
                                }
                                let msg_id = {
                                    let s = st.lock().unwrap_or_else(|e| e.into_inner());
                                    s.msg_id
                                };
                                if let Some(mid) = msg_id {
                                    let html = markdown_to_telegram_html(&snap.response_text);
                                    let display = format!("{}\u{258b}", html);
                                    let _ = bot
                                        .edit_message_text(chat_id, mid, display)
                                        .parse_mode(ParseMode::Html)
                                        .await;
                                }
                            }
                        }

                        let _ = bot.send_chat_action(chat_id, ChatAction::Typing).await;
                    }
                }
            }
        }
    });

    // Progress callback — same as handle_message
    let progress_cb: ProgressCallback = {
        let st = streaming.clone();
        Arc::new(move |_sid, event| match event {
            ProgressEvent::ReasoningChunk { text } => {
                if let Ok(mut s) = st.lock() {
                    s.thinking.push_str(&text);
                    s.dirty = true;
                }
            }
            ProgressEvent::StreamingChunk { text } => {
                if let Ok(mut s) = st.lock() {
                    if !s.thinking.is_empty() {
                        s.thinking.clear();
                    }
                    s.response.push_str(&text);
                    s.dirty = true;
                    s.processing = false;
                }
            }
            ProgressEvent::ToolStarted {
                tool_name,
                tool_input,
            } => {
                if let Ok(mut s) = st.lock() {
                    s.thinking.clear();
                    if s.tools_started_at.is_none() {
                        s.tools_started_at = Some(std::time::Instant::now());
                    }
                    let ctx = tool_context(&tool_name, &tool_input);
                    let idx = s.tool_msgs.len();
                    s.tool_msgs.push(ToolMsg {
                        msg_id: None,
                        name: tool_name,
                        context: ctx,
                        completed: None,
                        dirty: true,
                    });
                    s.display_queue.push(DisplayItem::NewTool(idx));
                }
            }
            ProgressEvent::ToolCompleted {
                tool_name, success, ..
            } => {
                if let Ok(mut s) = st.lock() {
                    s.tool_round_count += 1;
                    if let Some(tool) = s
                        .tool_msgs
                        .iter_mut()
                        .rev()
                        .find(|t| t.name == tool_name && t.completed.is_none())
                    {
                        tool.completed = Some(success);
                        tool.dirty = true;
                    }
                    if s.msg_id.is_some() {
                        s.recreate = true;
                    }
                }
            }
            ProgressEvent::IntermediateText { text, reasoning } => {
                if let Ok(mut s) = st.lock() {
                    s.thinking.clear();
                    s.response.clear();
                    if s.msg_id.is_some() {
                        s.recreate = true;
                    }
                    let content = if text.is_empty() {
                        reasoning.unwrap_or_default()
                    } else {
                        text
                    };
                    if !content.is_empty() {
                        s.display_queue.push(DisplayItem::Intermediate(content));
                    }
                }
            }
            _ => {}
        })
    };

    // ── Agent call ──────────────────────────────────────────────────────────
    let cancel_token = CancellationToken::new();
    telegram_state
        .store_cancel_token(session_id, cancel_token.clone())
        .await;

    let chat_id_str = chat_id.0.to_string();
    let result = agent
        .send_message_with_tools_and_callback(
            session_id,
            prompt,
            None,
            Some(cancel_token.clone()),
            None, // no approval callback for resume
            Some(progress_cb),
            "telegram",
            Some(&chat_id_str),
        )
        .await;

    telegram_state.remove_cancel_token(session_id).await;
    edit_cancel.cancel();

    // ── Final delivery ─────────────────────────────────────────────────────
    let (streaming_msg_id, status_msg_id, remaining_display) = {
        let mut s = streaming.lock().unwrap_or_else(|e| e.into_inner());
        let display: Vec<DisplayItem> = s.display_queue.drain(..).collect();
        (s.msg_id, s.status_msg_id, display)
    };
    if let Some(mid) = status_msg_id {
        let _ = bot.delete_message(chat_id, mid).await;
    }

    if cancel_token.is_cancelled() {
        tracing::info!(
            "Telegram: resume for session {} cancelled by new message",
            session_id
        );
        // Only delete the streaming placeholder — keep prior
        // intermediate + tool-call history visible. See the matching
        // block in handle_message() for rationale.
        if let Some(mid) = streaming_msg_id {
            let _ = bot.delete_message(chat_id, mid).await;
        }
        return Ok(());
    }

    // Send remaining display items
    for item in remaining_display {
        match item {
            DisplayItem::NewTool(idx) => {
                let tool_info = {
                    let s = streaming.lock().unwrap_or_else(|e| e.into_inner());
                    s.tool_msgs.get(idx).map(|t| {
                        let label = format!("**{}**{}", t.name, t.context);
                        (label, t.completed, t.msg_id)
                    })
                };
                if let Some((label, completed, existing_mid)) = tool_info {
                    let text = match completed {
                        None => format!("⚙️ {}", label),
                        Some(true) => format!("✅ {}", label),
                        Some(false) => format!("❌ {}", label),
                    };
                    let html = markdown_to_telegram_html(&text);
                    if existing_mid.is_none() {
                        let _ = bot
                            .send_message(chat_id, &html)
                            .parse_mode(ParseMode::Html)
                            .await;
                    }
                }
            }
            DisplayItem::Intermediate(text) => {
                let text = crate::utils::sanitize::strip_llm_artifacts(&text);
                let text = redact_secrets(&text);
                // Pre-send dedup — see handle_message.
                {
                    let s = streaming.lock().unwrap_or_else(|e| e.into_inner());
                    if s.sent_intermediates.iter().any(|prev| prev == &text) {
                        tracing::info!(
                            "Telegram resume: suppressing duplicate intermediate (len={})",
                            text.len()
                        );
                        continue;
                    }
                }
                let html = markdown_to_telegram_html(&text);
                if !html.is_empty() {
                    // Same chunk-and-confirm pattern as the group handler —
                    // don't mark as delivered unless every chunk succeeded,
                    // otherwise dedup will strip a message the user never saw.
                    let chunks: Vec<String> = split_message(&html, 4096)
                        .into_iter()
                        .map(|s| s.to_string())
                        .collect();
                    let mut sent_ids: Vec<MessageId> = Vec::new();
                    let mut all_ok = true;
                    for chunk in &chunks {
                        match send_html_or_plain(&bot, chat_id, chunk).await {
                            Ok(id) => sent_ids.push(id),
                            Err(e) => {
                                tracing::warn!(
                                    "Telegram (DM) intermediate send failed ({e}) — NOT marking as delivered",
                                );
                                all_ok = false;
                                break;
                            }
                        }
                    }
                    if all_ok {
                        let mut s = streaming.lock().unwrap_or_else(|e| e.into_inner());
                        s.sent_intermediates.push(text.clone());
                        s.intermediate_msg_ids.extend(sent_ids);
                    }
                }
            }
        }
    }

    match result {
        Ok(response) => {
            let (text_only, img_paths) = crate::utils::extract_img_markers(&response.content);
            let text_only = crate::utils::sanitize::strip_llm_artifacts(&text_only);
            let text_only = redact_secrets(&text_only);

            // Dedup intermediates already delivered so we don't duplicate
            // them when editing the streaming placeholder with the final.
            let sent = {
                let s = streaming.lock().unwrap_or_else(|e| e.into_inner());
                s.sent_intermediates.clone()
            };
            let text_only = if !sent.is_empty() {
                let mut remaining = text_only.clone();
                for intermediate in &sent {
                    remaining = remaining.replace(intermediate.as_str(), "");
                }
                remaining.trim().to_string()
            } else {
                text_only
            };

            for img_path in img_paths {
                if let Ok(bytes) = tokio::fs::read(&img_path).await {
                    let _ = bot.send_photo(chat_id, InputFile::memory(bytes)).await;
                }
            }

            let html = markdown_to_telegram_html(&text_only);
            if !html.is_empty() {
                let chunks: Vec<String> = split_message(&html, 4096)
                    .into_iter()
                    .map(|s| s.to_string())
                    .collect();

                if chunks.len() == 1
                    && let Some(mid) = streaming_msg_id
                {
                    if let Err(e) = bot
                        .edit_message_text(chat_id, mid, &chunks[0])
                        .parse_mode(ParseMode::Html)
                        .await
                    {
                        tracing::warn!("Telegram resume: edit failed ({e}), falling back to send");
                        let _ = bot.delete_message(chat_id, mid).await;
                        let _ = send_html_or_plain(&bot, chat_id, &chunks[0]).await;
                    }
                } else {
                    if let Some(mid) = streaming_msg_id {
                        let _ = bot.delete_message(chat_id, mid).await;
                    }
                    for chunk in &chunks {
                        let _ = send_html_or_plain(&bot, chat_id, chunk).await;
                    }
                }
            } else if let Some(mid) = streaming_msg_id {
                let _ = bot.delete_message(chat_id, mid).await;
            }

            tracing::info!(
                "Telegram: resume completed for session {} — {} chars delivered",
                session_id,
                response.content.len()
            );
        }
        Err(crate::brain::agent::AgentError::Cancelled) => {
            tracing::info!("Telegram: resume cancelled for session {}", session_id);
            if let Some(mid) = streaming_msg_id {
                let _ = bot.delete_message(chat_id, mid).await;
            }
        }
        Err(e) => {
            tracing::error!("Telegram: resume error for session {}: {}", session_id, e);
            if let Some(mid) = streaming_msg_id {
                let _ = bot
                    .edit_message_text(chat_id, mid, format!("Error: {}", e))
                    .await;
            } else {
                let _ = bot.send_message(chat_id, format!("Error: {}", e)).await;
            }
        }
    }

    Ok(())
}

/// Convert simple markdown (`*bold*`, `` `code` ``) to Telegram HTML.
pub(crate) fn md_to_html(s: &str) -> String {
    // Replace `code` with <code>code</code>, then *bold* with <b>bold</b>
    let mut out = String::with_capacity(s.len());
    let mut chars = s.chars().peekable();
    while let Some(c) = chars.next() {
        if c == '`' {
            let code: String = chars.by_ref().take_while(|&ch| ch != '`').collect();
            out.push_str("<code>");
            out.push_str(&code);
            out.push_str("</code>");
        } else if c == '*' {
            let bold: String = chars.by_ref().take_while(|&ch| ch != '*').collect();
            out.push_str("<b>");
            out.push_str(&bold);
            out.push_str("</b>");
        } else {
            out.push(c);
        }
    }
    out
}

/// Shorthand — delegates to the shared utility in `crate::utils`.
fn tool_context(name: &str, input: &serde_json::Value) -> String {
    crate::utils::tool_context_hint(name, input)
}

/// Send an HTML message, falling back to plain text if Telegram rejects the HTML.
/// Returns the resulting `MessageId` so callers that need to track or later delete
/// the message (e.g. intermediate cleanup on cancellation) can do so.
async fn send_html_or_plain(
    bot: &Bot,
    chat_id: ChatId,
    html: &str,
) -> std::result::Result<MessageId, teloxide::RequestError> {
    match bot
        .send_message(chat_id, html)
        .parse_mode(ParseMode::Html)
        .await
    {
        Ok(m) => Ok(m.id),
        Err(teloxide::RequestError::RetryAfter(secs)) => {
            tracing::warn!(
                "Telegram: HTML send rate-limited, waiting {}s before retry",
                secs.seconds()
            );
            tokio::time::sleep(secs.duration()).await;
            // Retry as HTML after waiting
            match bot
                .send_message(chat_id, html)
                .parse_mode(ParseMode::Html)
                .await
            {
                Ok(m) => Ok(m.id),
                Err(e) => {
                    tracing::warn!("Telegram: HTML retry failed ({e}), sending as plain text");
                    let plain = strip_html_tags(html);
                    bot.send_message(chat_id, plain).await.map(|m| m.id)
                }
            }
        }
        Err(e) => {
            tracing::warn!("Telegram: HTML send failed ({e}), retrying as plain text");
            let plain = strip_html_tags(html);
            bot.send_message(chat_id, plain).await.map(|m| m.id)
        }
    }
}

fn strip_html_tags(html: &str) -> String {
    html.replace("<b>", "")
        .replace("</b>", "")
        .replace("<i>", "")
        .replace("</i>", "")
        .replace("<code>", "")
        .replace("</code>", "")
        .replace("<pre>", "")
        .replace("</pre>", "")
        .replace("&lt;", "<")
        .replace("&gt;", ">")
        .replace("&amp;", "&")
}

/// Convert markdown to Telegram-safe HTML.
/// Handles: code blocks, inline code, bold, italic, underscore italic,
/// strikethrough, headers, links, and list items. Escapes HTML entities.
pub(crate) fn markdown_to_telegram_html(text: &str) -> String {
    let mut result = String::with_capacity(text.len() + 256);
    let mut in_code_block = false;
    let mut code_lang;

    for line in text.lines() {
        if line.starts_with("```") {
            if in_code_block {
                result.push_str("</code></pre>\n");
                in_code_block = false;
            } else {
                code_lang = line.trim_start_matches('`').trim().to_string();
                if code_lang.is_empty() {
                    result.push_str("<pre><code>");
                } else {
                    result.push_str(&format!(
                        "<pre><code class=\"language-{}\">",
                        escape_html(&code_lang)
                    ));
                }
                in_code_block = true;
            }
            continue;
        }

        if in_code_block {
            result.push_str(&escape_html(line));
            result.push('\n');
            continue;
        }

        // Headers: # → bold
        let trimmed = line.trim_start();
        if trimmed.starts_with('#') {
            let content = trimmed.trim_start_matches('#').trim();
            let escaped = escape_html(content);
            result.push_str(&format!("<b>{}</b>\n", format_inline(&escaped)));
            continue;
        }

        // List items: - or * at start of line → bullet
        if (trimmed.starts_with("- ") || trimmed.starts_with("* ")) && trimmed.len() > 2 {
            let content = &trimmed[2..];
            let escaped = escape_html(content);
            // Preserve leading indent
            let indent = line.len() - trimmed.len();
            let spaces = &line[..indent];
            result.push_str(&format!(
                "{}• {}\n",
                escape_html(spaces),
                format_inline(&escaped)
            ));
            continue;
        }

        let escaped = escape_html(line);
        let formatted = format_inline(&escaped);
        result.push_str(&formatted);
        result.push('\n');
    }

    if in_code_block {
        result.push_str("</code></pre>\n");
    }

    result.trim_end().to_string()
}

/// Escape HTML special characters
fn escape_html(text: &str) -> String {
    text.replace('&', "&amp;")
        .replace('<', "&lt;")
        .replace('>', "&gt;")
}

/// Apply inline formatting: `code`, **bold**, *italic*, _italic_, ~~strikethrough~~, [text](url)
fn format_inline(text: &str) -> String {
    // First pass: convert markdown links [text](url) → <a href="url">text</a>
    // Links are processed first because their syntax contains special chars
    let text = convert_links(text);

    let mut result = String::new();
    let chars: Vec<char> = text.chars().collect();
    let mut i = 0;

    while i < chars.len() {
        if chars[i] == '`' {
            if let Some(end) = chars[i + 1..].iter().position(|&c| c == '`') {
                let code: String = chars[i + 1..i + 1 + end].iter().collect();
                result.push_str(&format!("<code>{}</code>", code));
                i += end + 2;
                continue;
            }
        } else if chars[i] == '~' && i + 1 < chars.len() && chars[i + 1] == '~' {
            // ~~strikethrough~~
            if let Some(end) = find_closing_marker(&chars[i + 2..], &['~', '~']) {
                let inner: String = chars[i + 2..i + 2 + end].iter().collect();
                result.push_str(&format!("<s>{}</s>", inner));
                i += end + 4;
                continue;
            }
        } else if chars[i] == '*' && i + 1 < chars.len() && chars[i + 1] == '*' {
            // **bold**
            if let Some(end) = find_closing_marker(&chars[i + 2..], &['*', '*']) {
                let inner: String = chars[i + 2..i + 2 + end].iter().collect();
                result.push_str(&format!("<b>{}</b>", inner));
                i += end + 4;
                continue;
            }
        } else if chars[i] == '_' && i + 1 < chars.len() && chars[i + 1] == '_' {
            // __bold__ (underscore bold)
            if let Some(end) = find_closing_marker(&chars[i + 2..], &['_', '_']) {
                let inner: String = chars[i + 2..i + 2 + end].iter().collect();
                result.push_str(&format!("<b>{}</b>", inner));
                i += end + 4;
                continue;
            }
        } else if chars[i] == '*' {
            // *italic*
            if let Some(end) = chars[i + 1..].iter().position(|&c| c == '*') {
                let inner: String = chars[i + 1..i + 1 + end].iter().collect();
                result.push_str(&format!("<i>{}</i>", inner));
                i += end + 2;
                continue;
            }
        } else if chars[i] == '_' {
            // _italic_ — only match if not part of a word (e.g. my_var should stay)
            let prev_alnum = i > 0 && chars[i - 1].is_alphanumeric();
            if !prev_alnum && let Some(end) = chars[i + 1..].iter().position(|&c| c == '_') {
                let next_alnum =
                    i + 1 + end + 1 < chars.len() && chars[i + 1 + end + 1].is_alphanumeric();
                if !next_alnum && end > 0 {
                    let inner: String = chars[i + 1..i + 1 + end].iter().collect();
                    result.push_str(&format!("<i>{}</i>", inner));
                    i += end + 2;
                    continue;
                }
            }
        }
        result.push(chars[i]);
        i += 1;
    }
    result
}

/// Convert markdown links [text](url) to Telegram HTML <a> tags.
/// Operates on already-HTML-escaped text, so we must unescape the URL.
fn convert_links(text: &str) -> String {
    let mut result = String::new();
    let mut rest = text;
    while let Some(open) = rest.find('[') {
        result.push_str(&rest[..open]);
        let after_open = &rest[open + 1..];
        if let Some(close) = after_open.find("](") {
            let link_text = &after_open[..close];
            let after_paren = &after_open[close + 2..];
            if let Some(end_paren) = after_paren.find(')') {
                let url = &after_paren[..end_paren];
                // Unescape HTML entities in URL (escape_html ran before format_inline)
                let clean_url = url
                    .replace("&amp;", "&")
                    .replace("&lt;", "<")
                    .replace("&gt;", ">");
                result.push_str(&format!("<a href=\"{}\">{}</a>", clean_url, link_text));
                rest = &after_paren[end_paren + 1..];
                continue;
            }
        }
        // Not a valid link, emit the '[' and continue
        result.push('[');
        rest = after_open;
    }
    result.push_str(rest);
    result
}

/// Find closing double-char marker (e.g. **) in a char slice
fn find_closing_marker(chars: &[char], marker: &[char]) -> Option<usize> {
    if marker.len() != 2 {
        return None;
    }
    (0..chars.len().saturating_sub(1)).find(|&i| chars[i] == marker[0] && chars[i + 1] == marker[1])
}

/// Split a message into chunks that fit Telegram's 4096 char limit
pub(crate) fn split_message(text: &str, max_len: usize) -> Vec<&str> {
    if text.len() <= max_len {
        return vec![text];
    }
    let mut chunks = Vec::new();
    let mut start = 0;
    while start < text.len() {
        let mut end = (start + max_len).min(text.len());
        // Ensure end falls on a char boundary (back up if inside a multi-byte char)
        while end < text.len() && !text.is_char_boundary(end) {
            end -= 1;
        }
        let break_at = if end < text.len() {
            text[start..end]
                .rfind('\n')
                .filter(|&pos| pos > end - start - 200)
                .map(|pos| start + pos + 1)
                .unwrap_or(end)
        } else {
            end
        };
        chunks.push(&text[start..break_at]);
        start = break_at;
    }
    chunks
}

/// Build an `ApprovalCallback` that sends an inline-keyboard message to Telegram
/// and waits (up to 5 min) for the user to tap Yes, Always, or No.
pub(crate) fn make_approval_callback(
    state: Arc<super::TelegramState>,
) -> crate::brain::agent::ApprovalCallback {
    use crate::brain::agent::ToolApprovalInfo;
    use crate::utils::{check_approval_policy, persist_auto_session_policy};
    use teloxide::payloads::SendMessageSetters;
    use teloxide::prelude::Requester;
    use teloxide::types::{ChatId, InlineKeyboardButton, InlineKeyboardMarkup, ParseMode};
    use tokio::sync::oneshot;

    Arc::new(move |info: ToolApprovalInfo| {
        let state = state.clone();
        Box::pin(async move {
            // Respect config-level approval policy (single source of truth)
            if let Some(result) = check_approval_policy() {
                return Ok(result);
            }

            // Find the chat this session is active in
            let chat_id = match state.session_chat(info.session_id).await {
                Some(id) => id,
                None => match state.owner_chat_id().await {
                    Some(id) => id,
                    None => {
                        tracing::warn!(
                            "Telegram approval: no chat_id for session {}",
                            info.session_id
                        );
                        return Ok((false, false));
                    }
                },
            };

            let bot = match state.bot().await {
                Some(b) => b,
                None => {
                    tracing::warn!("Telegram approval: bot not connected");
                    return Ok((false, false));
                }
            };

            // Build unique approval id
            let approval_id = uuid::Uuid::new_v4().to_string();

            // Build inline keyboard — Yes / Always (session) / YOLO (permanent) / No
            let keyboard = InlineKeyboardMarkup::new(vec![
                vec![
                    InlineKeyboardButton::callback("✅ Yes", format!("approve:{}", approval_id)),
                    InlineKeyboardButton::callback(
                        "🔁 Always (session)",
                        format!("always:{}", approval_id),
                    ),
                ],
                vec![
                    InlineKeyboardButton::callback(
                        "🔥 YOLO (permanent)",
                        format!("yolo:{}", approval_id),
                    ),
                    InlineKeyboardButton::callback("❌ No", format!("deny:{}", approval_id)),
                ],
            ]);

            // Format message — redact secrets before display, truncate to fit Telegram limit
            let safe_input = crate::utils::redact_tool_input(&info.tool_input);
            let mut input_pretty = serde_json::to_string_pretty(&safe_input)
                .unwrap_or_else(|_| safe_input.to_string());
            if input_pretty.len() > 3500 {
                input_pretty.truncate(3500);
                input_pretty.push_str("\n... [truncated]");
            }
            let text = format!(
                "🔐 <b>Tool Approval Required</b>\n\nTool: <code>{}</code>\nInput:\n<pre>{}</pre>",
                info.tool_name,
                escape_html(&input_pretty),
            );

            // Register oneshot channel BEFORE sending the message to prevent
            // race condition where user clicks before registration completes
            let (tx, rx) = oneshot::channel();
            state
                .register_pending_approval(approval_id.clone(), tx)
                .await;
            tracing::info!(
                "Telegram approval: registered pending id={}, sending to chat={}",
                approval_id,
                chat_id
            );

            match bot
                .send_message(ChatId(chat_id), &text)
                .parse_mode(ParseMode::Html)
                .reply_markup(keyboard)
                .await
            {
                Ok(_) => {
                    tracing::info!(
                        "Telegram approval: message sent, waiting for response (id={})",
                        approval_id
                    );
                }
                Err(e) => {
                    tracing::error!("Telegram approval: failed to send message: {}", e);
                    return Ok((false, false));
                }
            }

            // Wait up to 5 minutes
            match tokio::time::timeout(std::time::Duration::from_secs(300), rx).await {
                Ok(Ok((approved, always))) => {
                    tracing::info!(
                        "Telegram approval: user responded id={}, approved={}, always={}",
                        approval_id,
                        approved,
                        always
                    );
                    if always {
                        persist_auto_session_policy();
                    }
                    Ok((approved, always))
                }
                Ok(Err(_)) => {
                    tracing::warn!(
                        "Telegram approval: oneshot channel closed (id={})",
                        approval_id
                    );
                    Ok((false, false))
                }
                Err(_) => {
                    tracing::warn!(
                        "Telegram approval: 5-minute timeout — auto-denying (id={})",
                        approval_id
                    );
                    Ok((false, false))
                }
            }
        })
    })
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_split_short_message() {
        let chunks = split_message("hello", 4096);
        assert_eq!(chunks, vec!["hello"]);
    }

    #[test]
    fn test_split_long_message() {
        let text = "a\n".repeat(3000);
        let chunks = split_message(&text, 4096);
        assert!(chunks.len() >= 2);
        for chunk in &chunks {
            assert!(chunk.len() <= 4096);
        }
        let joined: String = chunks.into_iter().collect();
        assert_eq!(joined, text);
    }

    #[test]
    fn test_split_no_newlines() {
        let text = "a".repeat(5000);
        let chunks = split_message(&text, 4096);
        assert_eq!(chunks.len(), 2);
        assert_eq!(chunks[0].len(), 4096);
        assert_eq!(chunks[1].len(), 904);
    }

    #[test]
    fn test_markdown_to_telegram_html_bold() {
        let html = markdown_to_telegram_html("**hello**");
        assert!(html.contains("<b>hello</b>"));
    }

    #[test]
    fn test_markdown_to_telegram_html_code_block() {
        let md = "```rust\nfn main() {}\n```";
        let html = markdown_to_telegram_html(md);
        assert!(html.contains("<pre><code"));
        assert!(html.contains("fn main()"));
        assert!(html.contains("</code></pre>"));
    }

    #[test]
    fn test_markdown_to_telegram_html_inline_code() {
        let html = markdown_to_telegram_html("use `cargo build`");
        assert!(html.contains("<code>cargo build</code>"));
    }

    #[test]
    fn test_escape_html() {
        assert_eq!(
            escape_html("<script>alert('xss')</script>"),
            "&lt;script&gt;alert('xss')&lt;/script&gt;"
        );
        assert_eq!(escape_html("a & b"), "a &amp; b");
    }

    #[test]
    fn test_img_marker_format() {
        // Verify the <<IMG:path>> marker format used for photo attachments
        let path = "/tmp/tg_photo_abc.jpg";
        let caption = "What's in this image?";
        let text = format!("<<IMG:{}>> {}", path, caption);
        assert!(text.starts_with("<<IMG:"));
        assert!(text.contains(path));
        assert!(text.contains(caption));
    }
}