opencrabs 0.3.82

//! Intermediate-message delivery: standalone narration posts (rich-first
//! with HTML fallback), footer append to the last intermediate, the
//! rate-limit-retrying send wrapper and the HTML-or-plain send.
//!
//! Moved VERBATIM out of handler.rs (#471 phase 1, pure decomposition —
//! only visibility widened to pub(crate) so the handler glob re-export
//! keeps every existing call site and test import stable).

use super::handler::{DisplayItem, StreamingState};
use super::markdown::{markdown_to_telegram_html, split_message, strip_html_tags};
use super::send::message_in_thread;
use crate::utils::sanitize::redact_secrets;
use std::sync::Arc;
use teloxide::prelude::*;
use teloxide::types::{MessageId, ParseMode};

pub(crate) async fn flush_intermediates(
    bot: &Bot,
    chat: ChatId,
    thread_id: Option<teloxide::types::ThreadId>,
    streaming: &Arc<std::sync::Mutex<StreamingState>>,
) {
    let pending: Vec<DisplayItem> = {
        let mut s = streaming.lock().unwrap_or_else(|e| e.into_inner());
        s.display_queue
            .drain(..)
            .filter(|item| matches!(item, DisplayItem::Intermediate(_)))
            .collect()
    };
    for item in pending {
        if let DisplayItem::Intermediate(text) = item {
            let text = crate::utils::sanitize::strip_llm_artifacts(&text);
            let text = redact_secrets(&text);
            let (text, _img_paths) = crate::utils::extract_img_markers(&text);
            // Strip <<react:emoji>> too; see edit-loop site in handle_message.
            // flush_intermediates has no inbound user message in scope (called
            // from resume + follow-up paths), so the directive is stripped but
            // no reaction fires here (#261).
            let (text, _react_emoji) = crate::utils::extract_react_marker(&text);
            // #690 follow-up (#980): re-expand a collapsed table BEFORE the
            // dedup check, the #582 gate and the sends, so a collapsed report
            // passes the gate and both delivery paths render a grid instead of
            // raw pipes. Idempotent on well-formed tables.
            let text = super::rich::reflow_collapsed_tables(&text);
            {
                let s = streaming.lock().unwrap_or_else(|e| e.into_inner());
                if s.sent_intermediates.iter().any(|prev| prev == &text) {
                    continue;
                }
            }
            // Same gate the streaming path applies (#838). This function sent
            // everything queued, so "an intermediate needs a table to earn its
            // own message" held on one delivery path and not the other. A turn
            // routed through resume or follow-up dumped its working-out
            // verbatim, and the reply then arrived twice: once as the
            // intermediate, once as the final answer saying the same thing in
            // a different shape.
            //
            // Deliberately the shared function rather than a copy of its rule,
            // so the two paths cannot drift again.
            if !is_deliverable_rich_report(&text) {
                continue;
            }
            if let Some(id) = try_send_intermediate_rich(bot, chat, thread_id, &text).await {
                let mut s = streaming.lock().unwrap_or_else(|e| e.into_inner());
                s.sent_intermediates.push(text.clone());
                s.intermediate_msg_ids.push(id);
                continue;
            }
            let html = markdown_to_telegram_html(&text);
            if html.is_empty() {
                continue;
            }
            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, thread_id, chunk, "turn").await {
                    Ok(id) => sent_ids.push(id),
                    Err(e) => {
                        tracing::warn!(
                            "Telegram: flush intermediate send failed ({e}), leaving for edit loop"
                        );
                        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);
            }
        }
    }
}

/// 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.
/// Build the edited message body for appending the ctx/tok-s footer to the
/// last intermediate message.
///
/// Used when a turn's final response text deduped to empty because all of
/// it was already delivered as intermediate messages (the common tool-
/// using case). Rather than drop the footer (which left the user never
/// seeing ctx budget on Telegram — 2026-06-06) or send a standalone
/// footer bubble (removed in 7a0ca1c9), we edit the last intermediate
/// message to carry the footer inline.
///
/// Reconstructs the last chunk exactly as it was originally sent
/// (`markdown_to_telegram_html` + `split_message(_, 4096)` then `.last()`),
/// appends the footer, and returns `None` when:
/// - the footer or intermediate text is empty, OR
/// - the combined result would exceed Telegram's 4096-char cap (never
///   truncate real content to make room for metadata).
///
/// Pure + free function so the fit/reconstruct logic is unit-testable
/// without a live bot.
// Channel-unused since the ctx footer moved onto the flow message (the
// intermediate-footer append path went with the pre-block status bubble);
// kept because the reconstruct-last-chunk logic is nontrivial and its tests
// pin the split/fit contract meanwhile.
#[cfg_attr(not(test), allow(dead_code))]
pub(crate) fn build_last_intermediate_with_footer(
    last_intermediate_text: &str,
    footer: &str,
) -> Option<String> {
    if footer.is_empty() || last_intermediate_text.is_empty() {
        return None;
    }
    let html = markdown_to_telegram_html(last_intermediate_text);
    let chunks = split_message(&html, 4096);
    let last_chunk = chunks.last()?;
    let combined = format!("{last_chunk}\n\n{footer}");
    if combined.chars().count() > 4096 {
        None
    } else {
        Some(combined)
    }
}

/// Send a structured intermediate segment as a native rich message, returning
/// its id for tracking. Returns `None` when the text carries no rich structure
/// or the rich API rejects it — the caller then falls back to the HTML path.
pub(crate) async fn try_send_intermediate_rich(
    bot: &Bot,
    chat_id: ChatId,
    thread_id: Option<teloxide::types::ThreadId>,
    text: &str,
) -> Option<MessageId> {
    if !super::rich::should_send_native_rich(text) {
        return None;
    }
    match super::rich::api::send_rich_markdown_id(
        bot.api_url().as_str(),
        bot.token(),
        chat_id.0,
        thread_id,
        text,
        "turn",
        "-",
    )
    .await
    {
        Ok(id) => Some(MessageId(id)),
        Err(e) => {
            tracing::warn!("Telegram: intermediate rich send failed, using HTML: {e}");
            None
        }
    }
}

/// True when a folded intermediate is a substantial rich report worth
/// delivering as its OWN message rather than burying in the collapsed
/// processing log (#582). Keyed on a real markdown table plus some length, so
/// thin narration (no table) keeps folding and only report-shaped content —
/// which the model may emit before a tool call (e.g. text + `plan complete` in
/// one step) — is surfaced.
pub(crate) fn is_deliverable_rich_report(text: &str) -> bool {
    // #690 follow-up (#980): a table collapsed onto ONE line is invisible to
    // contains_table (which needs the header and separator each on their own
    // line), so a collapsed report would fail this gate and get buried in the
    // folded log as raw pipes. Reflow first — the same recovery the final-
    // response and HTML-render paths already apply. Idempotent.
    let reflowed = super::rich::reflow_collapsed_tables(text);
    super::rich::contains_table(&reflowed) && text.trim().chars().count() >= 200
}

/// Deliver `text` as its own message (rich-first, HTML fallback) and record it
/// in `sent_intermediates` so the final-response dedup will not resend it.
/// Returns true when something was delivered. Used to surface a rich report the
/// model emitted before a tool call, which folding would otherwise bury (#582).
pub(crate) async fn deliver_intermediate_message(
    bot: &Bot,
    chat: ChatId,
    thread_id: Option<teloxide::types::ThreadId>,
    streaming: &Arc<std::sync::Mutex<StreamingState>>,
    text: &str,
) -> bool {
    // #690 follow-up (#980): re-expand a collapsed table once, up front, so the
    // dedup record, the rich send and the HTML fallback all see the same
    // expanded shape. The HTML path reflows again internally but is idempotent.
    let expanded = super::rich::reflow_collapsed_tables(text);
    let text = expanded.as_str();
    {
        let s = streaming.lock().unwrap_or_else(|e| e.into_inner());
        if s.sent_intermediates.iter().any(|prev| prev == text) {
            return true;
        }
    }
    if let Some(id) = try_send_intermediate_rich(bot, chat, thread_id, text).await {
        let mut s = streaming.lock().unwrap_or_else(|e| e.into_inner());
        s.sent_intermediates.push(text.to_string());
        s.intermediate_msg_ids.push(id);
        return true;
    }
    let html = markdown_to_telegram_html(text);
    if html.is_empty() {
        return false;
    }
    let mut sent_ids: Vec<MessageId> = Vec::new();
    for chunk in split_message(&html, 4096) {
        match send_html_or_plain(bot, chat, thread_id, chunk, "turn").await {
            Ok(id) => sent_ids.push(id),
            Err(e) => {
                tracing::warn!("Telegram: rich-intermediate send failed ({e})");
                return false;
            }
        }
    }
    let mut s = streaming.lock().unwrap_or_else(|e| e.into_inner());
    s.sent_intermediates.push(text.to_string());
    s.intermediate_msg_ids.extend(sent_ids);
    true
}

/// Threshold for treating a Telegram 429 as a "long rate-limit" (#1110).
///
/// When Telegram returns `Retry-After: N` where N > this threshold, the chat
/// is flood-banned for hours (28442s = 7.9 hours observed). Retrying the
/// send ladder burns 90 seconds (3 × 30s clamped wait) for no gain: the
/// window won't clear in that time. Instead, bail immediately and let the
/// caller surface the rate-limit to the user.
///
/// One hour is the boundary: typical flood windows (placeholder-edit churn,
/// command bursts) are seconds and stay under the inline cap. Anything over
/// an hour is a multi-hour ban, not a throttle.
const LONG_RATE_LIMIT_THRESHOLD: std::time::Duration = std::time::Duration::from_secs(3600);

/// Run a Telegram send, waiting out `RetryAfter` (429) up to 3 attempts.
///
/// Command replies are programmatic: a per-chat rate limit (typically a
/// streaming turn editing its placeholder into the same chat) must DELAY
/// them, never drop them. The command branches used a bare `.await?`, so
/// the 429 propagated out of the handler and the reply vanished with a
/// single error log line — /models looked "stuck" while a turn streamed
/// and worked right after it completed (#297). Non-429 errors and
/// exhausted retries still propagate to the caller.
///
/// Long rate-limits (>1 hour) bail immediately without retrying (#1110).
pub(crate) async fn send_retrying_rate_limit<T, F, Fut>(
    what: &str,
    mut send: F,
) -> std::result::Result<T, teloxide::RequestError>
where
    F: FnMut() -> Fut,
    Fut: std::future::IntoFuture<Output = std::result::Result<T, teloxide::RequestError>>,
{
    const MAX_RETRIES: u32 = 3;
    let mut attempt = 0u32;
    loop {
        match send().await {
            Err(teloxide::RequestError::RetryAfter(secs)) => {
                let requested = secs.duration();
                // Long rate-limit (>1 hour): bail immediately, don't retry (#1110).
                // The chat is flood-banned for hours; retrying burns 90s for no gain.
                if requested > LONG_RATE_LIMIT_THRESHOLD {
                    tracing::error!(
                        "Telegram: {what} long rate-limit ({}s > {}s threshold) — bailing immediately, \
                         no retry ladder (#1110)",
                        requested.as_secs(),
                        LONG_RATE_LIMIT_THRESHOLD.as_secs()
                    );
                    return Err(teloxide::RequestError::RetryAfter(secs));
                }
                if attempt < MAX_RETRIES {
                    attempt += 1;
                    super::rate_limit::wait_out(
                        what,
                        requested,
                        &format!(" (attempt {attempt}/{MAX_RETRIES})"),
                    )
                    .await;
                } else {
                    tracing::error!(
                        "Telegram: {what} still rate-limited after {MAX_RETRIES} retries ({}s) — giving up",
                        requested.as_secs()
                    );
                    return Err(teloxide::RequestError::RetryAfter(secs));
                }
            }
            // No success line here (review F1): the wrapper is generic and
            // has no correlation fields, so its line carried nothing the
            // chokepoint telemetry doesn't already say with full fields.
            other => return other,
        }
    }
}

pub(crate) async fn send_html_or_plain(
    bot: &Bot,
    chat_id: ChatId,
    thread_id: Option<teloxide::types::ThreadId>,
    html: &str,
    origin: &str,
) -> std::result::Result<MessageId, teloxide::RequestError> {
    // Correlation telemetry (#1085 P1a, review F8): this is the chokepoint
    // carrying chunked final replies, command acks and error notices.
    // `origin` is threaded by the caller (turn | tool | cron | system) so
    // an outbox/cron send is never mislabeled "turn". Every exit logs;
    // metadata only, never content.
    let thread = thread_id.map(|t| t.0.0);
    let hash8 = super::telemetry::content_hash8(html);
    let len = html.len();
    let log_ok = |path: &str, m: &MessageId, len: usize, hash8: &str| {
        super::telemetry::log_send_success(
            origin,
            "-",
            "-",
            "html_or_plain",
            path,
            chat_id.0,
            thread,
            m.0,
            len,
            hash8,
        );
    };
    // HTML rides the shared retry ladder (#1085 P1b R1): up to 3 attempts
    // with `rate_limit::wait_out` between them (#297 delay-never-drop),
    // matching every other send path — previously this hand-rolled a single
    // retry. Only a final failure falls back to plain text, and the
    // fallback rides the same ladder so a 429 cannot drop it either.
    match send_retrying_rate_limit("HTML send", || {
        message_in_thread(bot, chat_id, thread_id, html).parse_mode(ParseMode::Html)
    })
    .await
    {
        Ok(m) => {
            log_ok("html", &m.id, len, &hash8);
            Ok(m.id)
        }
        Err(e) => {
            tracing::warn!("Telegram: HTML send failed after retries ({e}), sending as plain text");
            let plain = strip_html_tags(html);
            // Review F2: hash and len must describe the text that actually
            // landed on the wire (the stripped plain text), not the HTML
            // source — a duplicate-correlation query must match payloads.
            let plain_hash8 = super::telemetry::content_hash8(&plain);
            let plain_len = plain.len();
            match send_retrying_rate_limit("plain fallback", || {
                message_in_thread(bot, chat_id, thread_id, plain.as_str())
            })
            .await
            {
                Ok(m) => {
                    log_ok("plain_fallback", &m.id, plain_len, &plain_hash8);
                    Ok(m.id)
                }
                Err(e2) => {
                    super::telemetry::log_send_failure(
                        origin,
                        "-",
                        "-",
                        "html_or_plain",
                        "plain_fallback",
                        chat_id.0,
                        thread,
                        plain_len,
                        &plain_hash8,
                        &e2.to_string(),
                    );
                    Err(e2)
                }
            }
        }
    }
}