//! 流式接收 LLM 响应

use std::collections::HashMap;
use std::sync::Arc;
use std::sync::atomic::AtomicBool;
use std::time::Duration;

use tauri::{AppHandle, Emitter, Manager};
use tracing::warn;

use df_ai::provider::{CompletionRequest, LlmProvider};

use super::{AiChatEvent, ToolCallDraft};

/// 从 anyhow 错误中尽力提取 HTTP 状态码/错误分类，供诊断拼接。
///
/// `provider.stream()` 失败有两类来源：
/// - 业务层：provider 在 non-2xx 时 `bail!("LLM 流式 API 错误 {status}: {body}")`，原始文本已含状态码，
///   此处用正则从文本抠 `HTTP <code>` 或裸 `<code>`（provider 串已含 status code）。
/// - 传输层（reqwest，不直接命名以避免给 src-tauri 加依赖）：从 Display 文本识别
///   timeout/connect 关键词分类。
///
/// 返回 `(status_or_class, raw)`：status 取文本中首个三位数；否则按关键词给 timeout/connect/unknown。
/// 从字符串(非 anyhow)提取诊断信息,用于 mpsc channel 中来自专用线程的错误字符串。
fn extract_error_diag_from_str(raw: &str) -> (String, String) {
    // 1) 抠 HTTP 状态码
    const HTTP_CODES: &[&str] = &[
        "400", "401", "403", "404", "408", "409", "413",
        "422", "429", "500", "502", "503", "504",
    ];
    let chars: Vec<char> = raw.chars().collect();
    let n = chars.len();
    let mut i = 0;
    while i + 3 <= n {
        if chars[i].is_ascii_digit() && chars[i + 1].is_ascii_digit() && chars[i + 2].is_ascii_digit() {
            let code: String = chars[i..i + 3].iter().collect();
            let prev_ok = i == 0 || !chars[i - 1].is_ascii_digit();
            let next_ok = i + 3 == n || !chars[i + 3].is_ascii_digit();
            if prev_ok && next_ok && HTTP_CODES.contains(&code.as_str()) {
                return (format!("HTTP {}", code), raw.to_string());
            }
            i += 3;
            continue;
        }
        i += 1;
    }

    let lower = raw.to_lowercase();
    let class = if lower.contains("timeout") || lower.contains("超时") {
        "timeout"
    } else if lower.contains("connect") || lower.contains("dns") || lower.contains("resolve") {
        "connect"
    } else if lower.contains("timed out") {
        "timeout"
    } else {
        "unknown"
    };
    (class.to_string(), raw.to_string())
}

/// 从 anyhow::Error 提取诊断信息
#[allow(dead_code)]
fn extract_error_diag(e: &anyhow::Error) -> (String, String) {
    extract_error_diag_from_str(&e.to_string())
}

/// AiError 诊断消息的上下文，区分「建连/首字节失败」与「流中途断」两类。
#[derive(Copy, Clone, Eq, PartialEq, Debug)]
pub(crate) enum DiagKind {
    /// `provider.stream()` 直接返回 Err：连接/鉴权/HTTP non-2xx 等
    #[allow(dead_code)]
    Init,
    /// 流已建立，next() 返回 Err：SSE 传输断/解析错等
    MidStream,
}

/// 拼接 AiError 的可读诊断文本（纯函数，便于单测）。
///
/// 格式：`[<provider_name>] <上下文>(<status_or_class>): <raw>`
/// - provider_name：`provider.name()`（anthropic 协议为 "anthropic-compat"，
///   openai 兼容为模型名）。当前 `LlmProvider` trait 未暴露 base_url/provider_type，
///   不改签名的前提下这是唯一可得的 provider 标识。
/// - status_or_class：`HTTP 4xx/5xx` 或 `timeout`/`connect`/`unknown` 分类。
/// - raw：anyhow 原始错误文本。
pub(crate) fn fmt_diag(provider_name: &str, kind: DiagKind, status_or_class: &str, raw: &str) -> String {
    let ctx = match kind {
        DiagKind::Init => "AI 调用失败",
        DiagKind::MidStream => "流式接收错误",
    };
    format!("[{}] {}({}): {}", provider_name, ctx, status_or_class, raw)
}

/// 流式接收结果。显式区分三类出口，避免调用方对 None 的歧义推断
/// (UX-2025-04 / CR-30-1 / CR-30-2 / 决策 F-260616-07 a1)。
pub(crate) enum StreamResult {
    /// 正常完成（流尽 + finished 信号，或用户主动停止）。incomplete=false。
    /// 调用方按正常流程入库 + emit AiCompleted。
    Complete {
        text: String,
        tool_calls: HashMap<u32, ToolCallDraft>,
        usage: df_ai::provider::TokenUsage,
        /// DeepSeek thinking 模式推理内容（需回传到下一轮请求）
        reasoning_content: Option<String>,
    },
    /// 流中途断（MidStream chunk Err / idle timeout / provider stream-error / 有 partial_text
    /// 但流尽未收到 finished）。已 emit 过 AiTextDelta（前端 currentText 已累积），
    /// **不 emit AiError**——调用方保文入库 + emit AiCompleted(incomplete=true) + 系统提示。
    /// 调用方**不重试**（决策 a1）。
    Partial {
        text: String,
        tool_calls: HashMap<u32, ToolCallDraft>,
        usage: df_ai::provider::TokenUsage,
        /// DeepSeek thinking 模式推理内容（部分累积）
        reasoning_content: Option<String>,
    },
    /// Init 失败（provider.stream() Err：建连/鉴权/HTTP non-2xx,或流建立后空文本各类中断）。
    /// **不再 emit AiError**——重试可恢复路径的 emit 权交调用方(agentic/mod.rs),
    /// 由 agentic 在重试耗尽/Fatal 时统一 emit 最终 AiError(单气泡),避免 N 次重试 push N+1 气泡。
    /// `retryable`: 据状态码分类(retry::is_status_retryable 镜像)：true=5xx/429/timeout/connect
    /// 可重试；false=4xx(非429)/鉴权/参数错 Fatal 立即放弃。调用方据此决定是否重试。
    /// `error`: 已格式化的错误文本(fmt_diag 拼接 provider 标识 + 上下文 + 状态 + 原文),
    /// 供 agentic 重试耗尽/Fatal 时 emit AiError 用,保留原诊断信息不丢失。
    InitFailed { retryable: bool, error: String },
}

/// 将 chunk 携带的 tool_calls delta 累加到累积表。
///
/// 每个 `ToolCallDelta` 按 `index` 归位到同一个 `ToolCallDraft`：
/// - `id` 覆盖(后到的 id 视为权威,匹配 OpenAI/Anthropic 协议行为)
/// - `function_name` / `function_arguments` 增量拼接(流式分片到达)
///
/// 抽取自 stream_llm 的 chunk match arm(原最深 7 层嵌套点),纯状态累加无 emit/return 副作用。
fn accumulate_tool_calls(
    tc_deltas: &[df_ai::provider::ToolCallDelta],
    tool_calls_acc: &mut HashMap<u32, ToolCallDraft>,
) {
    for tc_delta in tc_deltas {
        let draft = tool_calls_acc.entry(tc_delta.index).or_default();
        if let Some(id) = &tc_delta.id {
            draft.id = id.clone();
        }
        if let Some(name) = &tc_delta.function_name {
            draft.name.push_str(name);
        }
        if let Some(args) = &tc_delta.function_arguments {
            draft.args.push_str(args);
        }
    }
}

/// 双写 emit: Tauri 事件 + ai_event_bus 总线(跨端透传)。消费 ev。
fn emit_ai_chat_event(app_handle: &AppHandle, ev: AiChatEvent) {
    let _ = app_handle.emit("ai-chat-event", ev.clone());
    let _ = app_handle.state::<crate::state::AppState>().ai_event_bus.publish_event(ev);
}

/// flush 合批的 AiTextDelta(双写 emit + event_bus)。空 buffer 不 emit。
fn flush_delta(app_handle: &AppHandle, conv_id: &str, pending_delta: &mut String) {
    if pending_delta.is_empty() {
        return;
    }
    let delta = std::mem::take(pending_delta);
    emit_ai_chat_event(
        app_handle,
        AiChatEvent::AiTextDelta {
            delta,
            conversation_id: Some(conv_id.to_string()),
        },
    );
}

/// 流式接收 LLM 响应。
///
/// 三类异常处理（返回 StreamResult 显式区分出口）：
/// - Init 失败（provider.stream() Err）：UX-260618-15 **不 emit AiError**，返回
///   InitFailed{retryable, error}。retryable 据状态码分类(retry::is_status_retryable 镜像)：
///   5xx/429/timeout/connect=true，4xx(非429)/鉴权/参数错=false Fatal。调用方走流前重试(仅
///   retryable=true,≤max_retries 次, 复用 retry::backoff_delay 退避 + 30s 总预算),重试耗尽或
///   Fatal 时由调用方 emit 最终 AiError(error 字段携带诊断文本,单气泡聚合)。
/// - MidStream 失败（流已建立后断/timeout/err 事件/有 partial_text 但流尽未 finished）：
///   **不 emit AiError**，返回 Partial{...} 保文。调用方不重试，入库 + AiCompleted(incomplete)
///   + 系统提示网络中断。空文本仍返回 InitFailed{retryable=true}(无文可保,交重试,emit 权归调用方)。
/// - 用户停止（stop_flag）：返回 Complete（incomplete 语义非异常中断）。
///
/// 心跳与停止响应（B-260615-02 / B-260615-04）：单 `tokio::select!` 三分支
/// - `stream.next()`：正常 chunk 处理
/// - `heartbeat.tick()`（30s）：静默期（如工具执行后等下一轮首 chunk）emit `AiHeartbeat`，
///   前端 watchdog 据此 reset，区分「LLM 在跑」与「真断」（避免空气泡误报中断）
/// - `stop_notify.notified()`：用户点停止即时打断，不再等 chunk 到或 120s idle timeout
pub(crate) async fn stream_llm(
    provider: Arc<dyn LlmProvider>,
    request: CompletionRequest,
    app_handle: &AppHandle,
    stop_flag: &AtomicBool,
    notify: &tokio::sync::Notify,
    conv_id: &str,
) -> StreamResult {
    /// 流式读取空闲超时:超过此时长无任何 chunk 即判定连接已断
    /// 原 120s 太长,后端挂死时用户等太久。改为 45s,正常流式每 chunk 间隔
    /// 远不到 45s(即使模型思考间隙),超时即判为连接异常走保文/报错路径
    const STREAM_IDLE_TIMEOUT: Duration = Duration::from_secs(45);
    /// 首 chunk 超时(等首字节)。
    /// 正常 provider 首 token <3s(即使 48 工具定义 + 长上下文),10s 足够宽容;
    /// 已有数据后中途静默仍用 STREAM_IDLE_TIMEOUT(120s,允许模型思考间隙)。
    const FIRST_CHUNK_TIMEOUT: Duration = Duration::from_secs(10);
    /// 心跳间隔：静默期向前端报「LLM 仍在跑」，reset watchdog
    const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(30);
    /// 后端 delta 合批窗口。50ms 内的多次 AiTextDelta 合并为一条再 emit,
    /// 降低 IPC 事件风暴(前端 rAF 只节流渲染不节流 IPC);合并后前端 currentText
    /// 逐条累加,最终一致。
    const DELTA_FLUSH_INTERVAL: Duration = Duration::from_millis(50);

    // ================================================================
    // BUG-2026-07-17 根治: 整个 LLM 流式请求运行在独立 OS 线程的
    // 专用 tokio runtime 上,通过 mpsc channel 将 chunk 流式送回主 runtime。
    // ================================================================
    //
    // 根因: Tauri 在 Windows 上使用单线程 tokio runtime。provider.stream()
    // 内部(eventsource-stream/reqwest TLS/网络建连)存在同步阻塞操作,
    // 阻塞唯一工作线程导致 tokio 计时器系统无法推进,timeout 不触发。
    //
    // 方案:
    // 1. 独立 OS 线程 + 多线程 tokio runtime 执行完整 LLM 请求
    //    (provider.stream + stream.next 循环),隔离同步阻塞。
    // 2. 通过 mpsc channel 将 StreamChunk 逐块传回主 runtime。
    // 3. reqwest Response/Stream 的生命周期不跨 runtime,避免
    //    "error decoding response body"(runtime 状态不兼容)。
    //
    // 资源: 专用线程运行至 LLM 流结束或 mpsc receiver 被 drop(超时/停止)。
    let (chunk_tx, mut chunk_rx) = tokio::sync::mpsc::channel::<
        std::result::Result<df_ai::provider::StreamChunk, String>
    >(64);
    let provider_for_thread = provider.clone();
    let conv_id_owned = conv_id.to_string();

    std::thread::spawn(move || {
        let rt = match tokio::runtime::Runtime::new() {
            Ok(rt) => rt,
            Err(e) => {
                tracing::error!(
                    conv_id = %conv_id_owned,
                    "[ai] 创建专用 tokio runtime 失败: {}",
                    e,
                );
                return;
            }
        };
        // BUG-2026-07-17 双重保险:在专用 runtime 内部也加 120s 超时。
        // 即使 provider.stream() 在专用线程上意外阻塞(理论不应发生,因专用 runtime
        // 多线程可推进独立 timer wheel),此 timeout 保证线程最终退出不永久挂起。
        rt.block_on(async move {
            use futures::StreamExt;
            let stream_result = tokio::time::timeout(
                std::time::Duration::from_secs(120),
                provider_for_thread.stream(request),
            ).await;
            match stream_result {
                Ok(Ok(mut stream)) => {
                    while let Some(chunk_result) = stream.next().await {
                        let sent = match chunk_result {
                            Ok(chunk) => chunk_tx.send(Ok(chunk)).await,
                            Err(e) => chunk_tx.send(Err(e.to_string())).await,
                        };
                        if sent.is_err() {
                            // receiver dropped (timeout/stop/error) — 停止发送
                            break;
                        }
                    }
                }
                Ok(Err(e)) => {
                    let _ = chunk_tx.send(Err(e.to_string())).await;
                }
                Err(_elapsed) => {
                    tracing::error!(
                        conv_id = %conv_id_owned,
                        "[ai] 专用 runtime 内部超时(120s),provider.stream 未返回"
                    );
                    let _ = chunk_tx.send(Err("LLM 调用超时(120s)".to_string())).await;
                }
            }
        });
    });

    // 在主 runtime 上通过 mpsc 接收 chunk,应用 timeout/heartbeat/stop 逻辑
    let mut full_text = String::new();
    let mut tool_calls_acc: HashMap<u32, ToolCallDraft> = HashMap::new();
    let mut finished_received = false;
    let mut stopped = false;
    let mut final_usage: Option<df_ai::provider::TokenUsage> = None;
    // DeepSeek thinking 模式推理内容累积（多轮需回传）
    let mut reasoning_content_acc: Option<String> = None;
    // per-conv delta 合批累加器。pending_delta 累积未 flush 的文本,
    // next_flush_at 记录下次 flush 的绝对时刻(首个入 buffer 时置 now+50ms)。
    let mut pending_delta = String::new();
    let mut next_flush_at: Option<tokio::time::Instant> = None;

    // heartbeat interval 提至 loop 外复用,避免每轮重建计时器
    // (每轮重建会丢已积累的节拍,且 interval 首次 tick 立即返回的特性会被误用)。
    // tokio interval 首 tick 立即返回——此处先丢弃首 tick,让心跳等满首个 30s 静默期才发
    // (心跳语义是"静默期仍在跑",循环入口立即报无意义且会与 rx.recv() 抢分支错过首 chunk)。
    let mut heartbeat = tokio::time::interval(HEARTBEAT_INTERVAL);
    heartbeat.tick().await; // 丢弃首 tick

    // BUG-2026-07-14 根治: wall-clock deadline 替代 select! 内 timeout 包裹。
    // idle_deadline 是绝对时刻:"下一次 chunk 必须在此前到达,否则判定超时"。
    // - 首 chunk 阶段: idle_deadline = now + 10s (FIRST_CHUNK_TIMEOUT)
    // - 收到首 chunk 后: idle_deadline = now + 120s (STREAM_IDLE_TIMEOUT),每次收到 chunk 更新
    // 心跳打断 select! 后回到循环顶部,deadline 检查在 select! 之前(不受竞争影响)。
    let mut idle_deadline = tokio::time::Instant::now() + FIRST_CHUNK_TIMEOUT;
    let mut first_chunk_done = false;

    loop {
        // stop 即时打断。
        if stop_flag.load(std::sync::atomic::Ordering::SeqCst) {
            stopped = true;
            break;
        }

        // BUG-2026-07-14 根治: wall-clock deadline 检查(不依赖 select! timeout 语义)。
        if tokio::time::Instant::now() >= idle_deadline {
            if !first_chunk_done {
                flush_delta(app_handle, conv_id, &mut pending_delta);
                return StreamResult::InitFailed {
                    retryable: true,
                    error: format!(
                        "流式响应超时（{}秒未收到首字节，服务无响应或请求体过大）",
                        FIRST_CHUNK_TIMEOUT.as_secs()
                    ),
                };
            }
            // 已收到 finish_reason 仅剩等 usage 帧:慢 provider 拖过 idle 窗口不降 Partial,
            // 按正常完成收尾(usage 缺失用默认 0,文本/tool_calls 已完整)。
            if finished_received {
                break;
            }
            warn!(
                provider = %provider.name(),
                conv_id = %conv_id,
                text_len = full_text.len(),
                "[ai] 流中途 idle timeout(deadline),保文不重试(incomplete)",
            );
            flush_delta(app_handle, conv_id, &mut pending_delta);
            return StreamResult::Partial {
                text: full_text,
                tool_calls: tool_calls_acc,
                usage: final_usage.unwrap_or_default(),
                reasoning_content: reasoning_content_acc,
            };
        }

        // 到点 flush 合批的 delta(单条 AiTextDelta 双写 emit + event_bus)。
        if let Some(deadline) = next_flush_at {
            if tokio::time::Instant::now() >= deadline {
                next_flush_at = None;
                flush_delta(app_handle, conv_id, &mut pending_delta);
            }
        }

        // 多分支 select!:
        // 1) chunk_rx.recv():来自专用线程的 LLM chunk (15s 保底 timeout,防 select! 死等)
        // 2) heartbeat.tick():静默期发 AiHeartbeat reset 前端 watchdog
        // 3) notify.notified():用户停止即时打断
        // 4) flush 定时器(sleep_until next_flush_at):尾部 delta 不因流静默而延迟。
        let flush_deadline = next_flush_at
            .unwrap_or_else(|| tokio::time::Instant::now() + Duration::from_secs(3600));
        let flush_fut = tokio::time::sleep_until(flush_deadline);
        tokio::pin!(flush_fut);

        tokio::select! {
            chunk_result = tokio::time::timeout(
                Duration::from_secs(15),
                chunk_rx.recv()
            ) => {
                match chunk_result {
                    Err(_elapsed) => {
                        // 15s 保底 timeout 触发。正常路径不会到这里:心跳 30s 会先触发 select! 返回。
                        if !first_chunk_done {
                            flush_delta(app_handle, conv_id, &mut pending_delta);
                            return StreamResult::InitFailed {
                                retryable: true,
                                error: format!(
                                    "流式响应超时（{}秒未收到首字节，服务无响应或请求体过大）",
                                    FIRST_CHUNK_TIMEOUT.as_secs()
                                ),
                            };
                        }
                        // 已收到 finish_reason 仅剩等 usage 帧:慢/坏 provider 拖过 15s 不降 Partial,
                        // 按正常完成收尾(与 idle_deadline 分支同口径,usage 缺失用默认 0)。
                        if finished_received {
                            break;
                        }
                        warn!(
                            provider = %provider.name(),
                            conv_id = %conv_id,
                            text_len = full_text.len(),
                            "[ai] 流中途 idle timeout,保文不重试(incomplete)",
                        );
                        flush_delta(app_handle, conv_id, &mut pending_delta);
                        return StreamResult::Partial {
                            text: full_text,
                            tool_calls: tool_calls_acc,
                            usage: final_usage.unwrap_or_default(),
                            reasoning_content: reasoning_content_acc,
                        };
                    }
                    Ok(None) => break, // mpsc channel 关闭 = 专用线程 stream 结束
                    Ok(Some(Ok(chunk))) => {
                        // BUG-2026-07-14 根治: 收到 chunk 更新 idle_deadline。
                        if !first_chunk_done {
                            first_chunk_done = true;
                        }
                        idle_deadline = tokio::time::Instant::now() + STREAM_IDLE_TIMEOUT;
                        if !chunk.delta.is_empty() {
                            full_text.push_str(&chunk.delta);
                            // 合批入 buffer,首个入 buffer 时置 flush 时刻(到点统一 emit 单条)。
                            pending_delta.push_str(&chunk.delta);
                            if next_flush_at.is_none() {
                                next_flush_at = Some(tokio::time::Instant::now() + DELTA_FLUSH_INTERVAL);
                            }
                        }
                        if let Some(tc_deltas) = &chunk.tool_calls {
                            accumulate_tool_calls(tc_deltas, &mut tool_calls_acc);
                        }
                        if let Some(u) = &chunk.usage {
                            final_usage = Some(u.clone());
                        }
                        if let Some(ref rc) = chunk.reasoning_content {
                            reasoning_content_acc.get_or_insert_with(String::new).push_str(rc);
                        }
                        // provider 流式错误事件
                        if let Some(err_msg) = &chunk.error {
                            // A2-B12 / G4.2:error 帧按 message 文本分类 retryable。
                            // 原硬编码 retryable:true,确定性请求错(鉴权/context_length/invalid_request)
                            // 被当瞬态空耗重试。现抠 HTTP 状态码/传输类走既有 classify_status_or_class
                            // 单一分类源;无码则按错误类型关键词保守降级(仅明确非重试降 Fatal)。
                            let (status_or_class, _raw) = extract_error_diag_from_str(err_msg);
                            let retryable = classify_error_frame_retryable(err_msg, &status_or_class);
                            let diag_class = if status_or_class == "unknown" {
                                "stream-error"
                            } else {
                                status_or_class.as_str()
                            };
                            warn!(
                                provider = %provider.name(),
                                conv_id = %conv_id,
                                error = %err_msg,
                                status_or_class = %status_or_class,
                                retryable = %retryable,
                                text_len = full_text.len(),
                                "[ai] provider 流式错误事件",
                            );
                            if full_text.is_empty() && tool_calls_acc.is_empty() {
                                flush_delta(app_handle, conv_id, &mut pending_delta);
                                return StreamResult::InitFailed {
                                    retryable,
                                    error: fmt_diag(
                                        provider.name(),
                                        DiagKind::MidStream,
                                        diag_class,
                                        err_msg,
                                    ),
                                };
                            }
                            flush_delta(app_handle, conv_id, &mut pending_delta);
                            return StreamResult::Partial {
                                text: full_text,
                                tool_calls: tool_calls_acc,
                                usage: final_usage.unwrap_or_default(),
                                reasoning_content: reasoning_content_acc,
                            };
                        }
                        if chunk.finished {
                            finished_received = true;
                            // usage 携带点:OpenAI 兼容流中 usage 挂在 [DONE] 帧(或 usage-only 帧),
                            // 而 finish_reason 帧(finished=true, usage=None)在其之前到达。
                            // 若此刻 break 会错过 [DONE] 帧的 usage → 真实 completion_tokens 丢失,
                            // 前端 token 显示 0。拿到真实 usage 才停;usage 仍 None 则继续读到
                            // usage-only/[DONE] 帧或通道关闭(Ok(None) 兜底退出,不断连误判)。
                            if chunk.usage.is_some() {
                                break;
                            }
                        }
                    }
                    Ok(Some(Err(err_str))) => {
                        // 专用线程上报的流错误(SSE 解析失败/HTTP 错误等)
                        let (status_or_class, raw) = extract_error_diag_from_str(&err_str);
                        warn!(
                            provider = %provider.name(),
                            status = %status_or_class,
                            conv_id = %conv_id,
                            error = %raw,
                            text_len = full_text.len(),
                            "[ai] 流式接收中途错误(from 专用线程)",
                        );
                        if full_text.is_empty() && tool_calls_acc.is_empty() {
                            flush_delta(app_handle, conv_id, &mut pending_delta);
                            return StreamResult::InitFailed {
                                retryable: classify_status_with_body(&status_or_class, &raw),
                                error: fmt_diag(
                                    provider.name(),
                                    DiagKind::MidStream,
                                    &status_or_class,
                                    &raw,
                                ),
                            };
                        }
                        flush_delta(app_handle, conv_id, &mut pending_delta);
                        return StreamResult::Partial {
                            text: full_text,
                            tool_calls: tool_calls_acc,
                            usage: final_usage.unwrap_or_default(),
                            reasoning_content: reasoning_content_acc,
                        };
                    }
                }
            }
            // 2) 心跳:静默期 30s 发 AiHeartbeat,前端 watchdog reset
            _ = heartbeat.tick() => {
                emit_ai_chat_event(app_handle, AiChatEvent::AiHeartbeat {
                    conversation_id: Some(conv_id.to_string()),
                });
            }
            // 3) 即时停止唤醒
            _ = notify.notified() => {
                if stop_flag.load(std::sync::atomic::Ordering::SeqCst) {
                    stopped = true;
                    break;
                }
            }
            // 4) delta 合批到点 flush(流静默/工具执行等待期也能按时发出尾部文本)
            _ = &mut flush_fut => {
                if next_flush_at.is_some() {
                    next_flush_at = None;
                    flush_delta(app_handle, conv_id, &mut pending_delta);
                }
            }
        }
    }

    // 用户停止
    if stopped {
        flush_delta(app_handle, conv_id, &mut pending_delta);
        return StreamResult::Complete {
            text: full_text,
            tool_calls: tool_calls_acc,
            usage: final_usage.unwrap_or_default(),
            reasoning_content: reasoning_content_acc,
        };
    }

    // 断连检测
    if !finished_received {
        if full_text.is_empty() && tool_calls_acc.is_empty() {
            return StreamResult::InitFailed {
                retryable: true,
                error: "流式响应意外中断（未收到完成信号，已丢弃残缺响应）".to_string(),
            };
        }
        warn!(
            provider = %provider.name(),
            conv_id = %conv_id,
            text_len = full_text.len(),
            "[ai] 流尽未收到 finished 但有 partial_text,保文不重试(incomplete)",
        );
        flush_delta(app_handle, conv_id, &mut pending_delta);
        return StreamResult::Partial {
            text: full_text,
            tool_calls: tool_calls_acc,
            usage: final_usage.unwrap_or_default(),
            reasoning_content: reasoning_content_acc,
        };
    }

    // 正常完成
    flush_delta(app_handle, conv_id, &mut pending_delta);
    StreamResult::Complete {
        text: full_text,
        tool_calls: tool_calls_acc,
        usage: final_usage.unwrap_or_default(),
        reasoning_content: reasoning_content_acc,
    }
}

/// 据 extract_error_diag 产出的 status_or_class 字符串镜像 retry::is_status_retryable 分类。
///
/// status_or_class 取值: "HTTP 5xx"/"HTTP 4xx"(白名单码) / "timeout" / "connect" / "unknown"。
/// - HTTP 429 / 5xx → retryable=true(复用 retry::is_status_retryable 逻辑)
/// - HTTP 4xx(非429) → retryable=false Fatal(401/403/404/422 等)
/// - timeout / connect → retryable=true(瞬态,可重试)
/// - unknown / 其他 → retryable=true(保守可重试,避免误判 Fatal 错杀)
///
/// 抽公共分类函数(决策 F-260616-07 a1 "复用 retry.rs 错误分类逻辑或镜像"),
/// 避免 stream_recv 与 agentic 各自重写状态码解析。
fn classify_status_or_class(status_or_class: &str) -> bool {
    use df_ai::retry::is_status_retryable;
    // 抠 status_or_class 内三位数字状态码("HTTP 401" → 401)
    let chars: Vec<char> = status_or_class.chars().collect();
    let mut i = 0;
    let n = chars.len();
    while i + 3 <= n {
        if chars[i].is_ascii_digit()
            && chars[i + 1].is_ascii_digit()
            && chars[i + 2].is_ascii_digit()
        {
            let code: String = chars[i..i + 3].iter().collect();
            if let Ok(status) = code.parse::<u16>() {
                return is_status_retryable(status);
            }
            i += 3;
            continue;
        }
        i += 1;
    }
    // 无三位数字码:按文本分类(timeout/connect/transport/decode 可重试,其余 unknown 保守不重试)
    // Anthropic 中文参数错误(如 `[1214][messages 参数非法]`)无 HTTP 数字前缀,
    // 走此文本分支,原 || true 致 retryable=true 空耗 4 次重试浪费 token。
    // 删 || true 对齐 retry::is_status_retryable 未知码不可重试语义;timeout/connect 瞬态仍可重试。
    //
    // BUG-2026-07-17 根治:SSE 解码/传输错误(error decoding response body / Transport error)
    // 是传输层错误,不是 Fatal。原实现未识别这些关键词,返 unknown→Fatal,致连续重试 4 次均
    // 立即放弃。现补全传输层关键词,让这类错误走可重试路径。
    let lower = status_or_class.to_lowercase();
    lower.contains("timeout")
        || lower.contains("connect")
        || lower.contains("transport")
        || lower.contains("decoding")
        || lower.contains("decode")
        || lower.contains("body")
        || lower.contains("eof")
        || lower.contains("reset")
        || lower.contains("broken pipe")
}

/// 429 额度/配额耗尽特征检测:区分「额度耗尽型 429」(确定性,重试必失败)与
/// 「普通瞬时限流 429」(可重试)。实证:anthropic/GLM 周月额度耗尽返回
/// `[1310] 您已达到每周/每月使用上限,限额将在 X 重置`(code 1310)。
/// 特征命中 → Fatal 不重试;未命中 → 交 is_status_retryable 正常判定。
/// 入参 combined:状态码串 + 错误体原文(如 "HTTP 429 [1310] 已达到上限...")。
fn is_quota_exhaustion(combined: &str) -> bool {
    let lower = combined.to_lowercase();
    let quota_markers = [
        "额度", "限额", "达到上限", "使用上限", "重置",
        "quota", "limit exceeded", "usage limit", "quota exhausted",
        "insufficient_quota",
        // 注意:不含 rate_limit_error——它是标准 429 错误类型(普通瞬时限流也带),
        // 误配会把可恢复的限流当额度耗尽降 Fatal。额度语义用上面明确词。
    ];
    // 仅当文本同时含「429」与额度特征,才判定额度耗尽(避免误伤普通限流)。
    lower.contains("429")
        && quota_markers.iter().any(|m| lower.contains(m))
}

/// 额度耗尽型 429 → Fatal(确定性,不重试);否则交 classify_status_or_class 正常判定。
/// 入参 body:错误体原文(provider bail 文本 / error 帧 message)。
fn classify_status_with_body(status_or_class: &str, body: &str) -> bool {
    if is_quota_exhaustion(&format!("{status_or_class} {body}")) {
        return false;
    }
    classify_status_or_class(status_or_class)
}

/// 据 provider 流式 error 帧的 message 文本分类是否可重试(A2-B12 / G4.2)。
///
/// openai/anthropic helper 仅将 error 帧的 `message` 字段透传进 chunk.error
/// (结构化 `type` 未上抛),故此处按 message 文本保守分类,对齐 Err 路径单一分类源:
/// - 内含显式 HTTP 状态码或传输类(timeout/connect) → 复用 classify_status_or_class
/// - 无状态码时按错误类型关键词:明确非重试签名(authentication / context_length /
///   invalid_request / 鉴权 / 参数错等) → Fatal
/// - 其余 → retryable=true(保守,防误判瞬态为 Fatal)
fn classify_error_frame_retryable(msg: &str, status_or_class: &str) -> bool {
    // 已有状态码/传输类 → 走既有单一分类源(显式 4xx Fatal,5xx/429/timeout/connect 可重试)
    // 额度耗尽型 429(如 [1310] 已达周/月上限)确定性失败,经 body 感知识别降 Fatal,
    // 不再被普通 429 路径误判可重试空耗。
    if status_or_class != "unknown" {
        return classify_status_with_body(status_or_class, msg);
    }
    // 无状态码:按错误类型关键词明确非重试签名降级 Fatal
    let lower = msg.to_lowercase();
    const FATAL_KEYWORDS: &[&str] = &[
        // OpenAI / Anthropic error.type
        "authentication_error",
        "invalid_request_error",
        "context_length_exceeded",
        "permission_error",
        "not_found_error",
        "request_too_large",
        "insufficient_quota",
        // OpenAI 消息文本签名
        "authentication",
        "invalid_request",
        "invalid request",
        "context length",
        "api key",
        "api-key",
        "api_key",
        "unauthorized",
        // 中文签名Anthropic 中文参数错误)
        "鉴权",
        "认证失败",
        "未授权",
        "密钥",
        "参数非法",
        "参数错误",
        "上下文长度",
        "输入超长",
    ];
    !FATAL_KEYWORDS.iter().any(|k| lower.contains(k))
}

// ============================================================
// 单测：诊断提取/格式化（纯函数，不发 HTTP、不依赖 app_handle）
// ============================================================

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

    // ---- classify_status_or_class：CR-30-1 retryable 分类(镜像 retry::is_status_retryable) ----

    /// 5xx → retryable=true
    #[test]
    fn classify_5xx_retryable() {
        assert!(classify_status_or_class("HTTP 500"));
        assert!(classify_status_or_class("HTTP 503"));
    }

    /// 429 → retryable=true(限流可重试)
    #[test]
    fn classify_429_retryable() {
        assert!(classify_status_or_class("HTTP 429"));
    }

    /// 4xx(非429) → retryable=false Fatal(鉴权/参数错立即放弃)
    #[test]
    fn classify_4xx_fatal() {
        assert!(!classify_status_or_class("HTTP 401"));
        assert!(!classify_status_or_class("HTTP 403"));
        assert!(!classify_status_or_class("HTTP 404"));
        assert!(!classify_status_or_class("HTTP 422"));
    }

    /// timeout / connect → retryable=true(瞬态)
    #[test]
    fn classify_transient_retryable() {
        assert!(classify_status_or_class("timeout"));
        assert!(classify_status_or_class("connect"));
    }

    /// unknown → 保守 retryable=false防 Anthropic 中文参数错误等 Fatal 空耗重试)
    #[test]
    fn classify_unknown_not_retryable() {
        assert!(!classify_status_or_class("unknown"));
    }

    // ---- classify_error_frame_retryable：A2-B12 / G4.2 error 帧分类 ----

    /// error 帧 message 含显式 4xx → Fatal(接入 classify_status_or_class 单一分类源)
    #[test]
    fn error_frame_http_4xx_fatal() {
        let (s, _) = extract_error_diag_from_str("LLM 流式 API 错误 401: Unauthorized");
        assert!(!classify_error_frame_retryable("LLM 流式 API 错误 401: Unauthorized", &s));
    }

    /// error 帧 message 含显式 5xx / 429 → retryable
    #[test]
    fn error_frame_http_5xx_retryable() {
        let (s, _) = extract_error_diag_from_str("LLM 流式 API 错误 500: Internal Server Error");
        assert!(classify_error_frame_retryable("LLM 流式 API 错误 500: Internal Server Error", &s));
        let (s2, _) = extract_error_diag_from_str("rate limit 429");
        assert!(classify_error_frame_retryable("rate limit 429", &s2));
    }

    /// error 帧 message 含 timeout → retryable(瞬态)
    #[test]
    fn error_frame_timeout_retryable() {
        let (s, _) = extract_error_diag_from_str("operation timed out");
        assert!(classify_error_frame_retryable("operation timed out", &s));
    }

    /// authentication / api key 签名 → Fatal
    #[test]
    fn error_frame_authentication_fatal() {
        let (s, _) = extract_error_diag_from_str("Incorrect API key provided: sk-xxx");
        assert!(!classify_error_frame_retryable("Incorrect API key provided: sk-xxx", &s));
        let (s2, _) = extract_error_diag_from_str("invalid x-api-key");
        assert!(!classify_error_frame_retryable("invalid x-api-key", &s2));
    }

    /// context_length_exceeded / context length 签名 → Fatal
    #[test]
    fn error_frame_context_length_fatal() {
        let (s, _) = extract_error_diag_from_str("This model's maximum context length is 128000 tokens");
        assert!(!classify_error_frame_retryable("This model's maximum context length is 128000 tokens", &s));
    }

    /// invalid_request 签名 → Fatal
    #[test]
    fn error_frame_invalid_request_fatal() {
        let (s, _) = extract_error_diag_from_str("invalid_request_error: the model is not supported");
        assert!(!classify_error_frame_retryable("invalid_request_error: the model is not supported", &s));
    }

    /// 无明确签名(overloaded / stream error 等瞬态语义)→ 保守 retryable=true(防误判 Fatal 错杀)
    #[test]
    fn error_frame_unknown_conservative_retryable() {
        let (s, _) = extract_error_diag_from_str("overloaded_error: Overloaded");
        assert!(classify_error_frame_retryable("overloaded_error: Overloaded", &s));
        let (s2, _) = extract_error_diag_from_str("stream error");
        assert!(classify_error_frame_retryable("stream error", &s2));
    }

    // ---- is_quota_exhaustion / classify_status_with_body：额度耗尽型 429 不重试 ----

    /// 额度耗尽型 429（anthropic/GLM 周月上限，含 [1310] 与「使用上限/重置」特征）→ 判定额度耗尽
    #[test]
    fn quota_exhaustion_429_detected() {
        let combined = "HTTP 429 [1310] 您已达到每周/每月使用上限,限额将在 2026-08-14 重置";
        assert!(is_quota_exhaustion(combined));
        assert!(!classify_status_with_body("HTTP 429", combined));
    }

    /// 普通瞬时限流 429（无额度特征）→ 不判定额度耗尽,仍可重试
    #[test]
    fn transient_429_not_quota() {
        let combined = "HTTP 429 请求过于频繁,请稍后重试";
        assert!(!is_quota_exhaustion(combined));
        assert!(classify_status_with_body("HTTP 429", combined));
    }

    /// 额度关键词但无 429 状态码 → 不判定额度耗尽(防误伤普通 5xx)
    #[test]
    fn quota_marker_without_429_not_detected() {
        assert!(!is_quota_exhaustion("HTTP 500 达到上限"));
    }

    /// classify_status_with_body 非 429 路径行为不变(5xx 仍可重试)
    #[test]
    fn classify_status_with_body_5xx_retryable() {
        assert!(classify_status_with_body("HTTP 500", "Internal Server Error"));
    }

    /// 普通瞬时限流 429 的标准 type=rate_limit_error 不应误判额度耗尽(仍可重试,防误伤)
    #[test]
    fn standard_rate_limit_error_not_quota() {
        let combined = r#"HTTP 429 {"type":"rate_limit_error","message":"请求过于频繁,请稍后重试"}"#;
        assert!(!is_quota_exhaustion(combined));
        assert!(classify_status_with_body("HTTP 429", combined));
    }

    /// error 帧 message 含额度耗尽型 429 → 经 classify_error_frame_retryable 降 Fatal 不重试
    #[test]
    fn error_frame_quota_429_fatal() {
        let msg = "HTTP 429 [1310] 您已达到每周/每月使用上限,限额将在 2026-08-14 重置";
        let (s, _) = extract_error_diag_from_str(msg);
        assert_eq!(s, "HTTP 429");
        assert!(!classify_error_frame_retryable(msg, &s));
    }

    // ---- extract_error_diag：业务层 bail（provider 串已含状态码）----

    /// provider 在 non-2xx bail 的典型串：抠出 401（鉴权失败/Key 错）
    #[test]
    fn diag_extracts_401_from_provider_bail() {
        let e = anyhow::anyhow!("LLM 流式 API 错误 401: Unauthorized");
        let (status, raw) = extract_error_diag(&e);
        assert_eq!(status, "HTTP 401");
        assert!(raw.contains("401"));
    }

    /// 404 = base_url/endpoint 不对
    #[test]
    fn diag_extracts_404_from_provider_bail() {
        let e = anyhow::anyhow!("LLM 流式 API 错误 404: Not Found");
        assert_eq!(extract_error_diag(&e).0, "HTTP 404");
    }

    /// 429 = 限流
    #[test]
    fn diag_extracts_429_from_provider_bail() {
        let e = anyhow::anyhow!("LLM 流式 API 错误 429: rate limit");
        assert_eq!(extract_error_diag(&e).0, "HTTP 429");
    }

    /// 5xx = 上游服务端错
    #[test]
    fn diag_extracts_500_from_provider_bail() {
        let e = anyhow::anyhow!("LLM 流式 API 错误 500: Internal Server Error");
        assert_eq!(extract_error_diag(&e).0, "HTTP 500");
    }

    // ---- extract_error_diag：传输层（reqwest Display 文本，不命名 reqwest）----

    /// 超时：连接成功但响应慢/静默断
    #[test]
    fn diag_classifies_timeout() {
        let e = anyhow::anyhow!("error sending request for url (https://api.x.com/v1/chat/completions): operation timed out");
        let (status, _) = extract_error_diag(&e);
        assert_eq!(status, "timeout");
    }

    /// 中文「超时」关键词
    #[test]
    fn diag_classifies_timeout_cn() {
        let e = anyhow::anyhow!("请求超时");
        assert_eq!(extract_error_diag(&e).0, "timeout");
    }

    /// 连接失败：DNS 解析失败 / 端点不通
    #[test]
    fn diag_classifies_connect() {
        let e = anyhow::anyhow!("dns error: failed to lookup address information");
        assert_eq!(extract_error_diag(&e).0, "connect");
    }

    /// resolve 关键词也归 connect
    #[test]
    fn diag_classifies_connect_resolve() {
        let e = anyhow::anyhow!("error connecting: resolve failed");
        assert_eq!(extract_error_diag(&e).0, "connect");
    }

    // ---- extract_error_diag：边界 ----

    /// 三位数但不在已知 HTTP 状态白名单（如 "200"）→ 不当状态码，走分类
    #[test]
    fn diag_ignores_non_http_status_number() {
        let e = anyhow::anyhow!("成功 200 条记录");
        // 200 不在白名单，应落到 unknown
        assert_eq!(extract_error_diag(&e).0, "unknown");
    }

    /// 完全无特征文本 → unknown，且 raw 原样返回
    #[test]
    fn diag_unknown_preserves_raw() {
        let e = anyhow::anyhow!("奇怪的错误 xyz");
        let (status, raw) = extract_error_diag(&e);
        assert_eq!(status, "unknown");
        assert_eq!(raw, "奇怪的错误 xyz");
    }

    /// 状态码出现在更长数字串里也不误匹配（边界判断：前后必须非数字）
    #[test]
    fn diag_does_not_match_status_inside_longer_digits() {
        // 边界判断后,白名单码嵌在长数字串内不再误命中(原窗口切片会从 14012 抠出 401)
        let e = anyhow::anyhow!("port 14012 used");
        assert_eq!(extract_error_diag(&e).0, "unknown");
        // 纯长数字串(无白名单码)同样不命中
        let e2 = anyhow::anyhow!("port 99999 used");
        assert_eq!(extract_error_diag(&e2).0, "unknown");
    }

    // ---- fmt_diag：两个上下文 ----

    #[test]
    fn fmt_diag_init_branch() {
        let s = fmt_diag("anthropic-compat", DiagKind::Init, "HTTP 401", "Unauthorized");
        assert_eq!(s, "[anthropic-compat] AI 调用失败(HTTP 401): Unauthorized");
    }

    /// 状态码紧贴中文字符(多字节 UTF-8)时仍能正确抠出
    /// (原按字节窗口切片在中文边界脆弱,改为 char 迭代后安全)
    #[test]
    fn diag_extracts_code_adjacent_to_multibyte_chars() {
        let e = anyhow::anyhow!("错误401Unauthorized");
        assert_eq!(extract_error_diag(&e).0, "HTTP 401");
        // 中文在前:状态码前字符是多字节中文,char 迭代边界判断正确
        let e2 = anyhow::anyhow!("请求失败：401");
        assert_eq!(extract_error_diag(&e2).0, "HTTP 401");
    }

    #[test]
    fn fmt_diag_midstream_branch() {
        let s = fmt_diag("gpt-4o", DiagKind::MidStream, "timeout", "operation timed out");
        assert_eq!(s, "[gpt-4o] 流式接收错误(timeout): operation timed out");
    }

    /// provider 名/错误文本含特殊字符仍按模板拼接（无格式注入风险）
    #[test]
    fn fmt_diag_handles_special_chars() {
        let s = fmt_diag("model/x", DiagKind::Init, "unknown", "err: {json} \"quoted\"");
        assert_eq!(s, "[model/x] AI 调用失败(unknown): err: {json} \"quoted\"");
    }
}