package com.wok.supportbot.app;

import com.wok.supportbot.advisor.ContentSafetyAdvisor;
import com.wok.supportbot.advisor.MyLoggerAdvisor;
import com.wok.supportbot.chatmemory.DatabaseChatMemory;
import com.wok.supportbot.config.ChatModelFactory;
import com.wok.supportbot.config.SimpleCircuitBreaker;
import com.wok.supportbot.entity.AiModelConfig;
import com.wok.supportbot.entity.LlmCallTrace;
import com.wok.supportbot.mcp.McpToolCallback;
import com.wok.supportbot.mcp.McpToolCallback.ToolCallEvent;
import com.wok.supportbot.mcp.McpToolCallbackAdapter;
import com.wok.supportbot.service.AiModelConfigService;
import com.wok.supportbot.service.ContentSafetyService;
import com.wok.supportbot.service.LlmCallTraceService;
import com.fasterxml.jackson.databind.ObjectMapper;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.ai.chat.client.ChatClient;
import org.springframework.ai.chat.client.advisor.MessageChatMemoryAdvisor;
import org.springframework.ai.chat.messages.Message;
import org.springframework.ai.chat.messages.UserMessage;
import org.springframework.ai.content.Media;
import org.springframework.ai.chat.metadata.Usage;
import org.springframework.ai.chat.model.ChatModel;
import org.springframework.ai.chat.model.ChatResponse;
import org.springframework.ai.document.Document;
import org.springframework.ai.tool.ToolCallback;
import org.springframework.ai.vectorstore.VectorStore;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.util.MimeType;
import org.springframework.util.MimeTypeUtils;
import org.springframework.util.StringUtils;
import reactor.core.Disposable;
import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxSink;
import reactor.core.publisher.SignalType;

import java.util.ArrayList;
import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;

import static org.springframework.ai.chat.memory.ChatMemory.CONVERSATION_ID;

/**
 * AI 对话执行层 —— ChatClient 构建、缓存管理与 LLM 调用执行。
 * <p>
 * 本类负责 ChatClient 生命周期（按 appType + allowedMcpTools 缓存，LRU 淘汰），
 * Advisor 链装配（ContentSafetyAdvisor → MessageChatMemoryAdvisor → MyLoggerAdvisor），
 * 以及同步/流式对话执行（含熔断保护）。对话编排决策由 {@link ChatPipeline} 完成。
 * <p>
 * 核心方法：
 * <ul>
 *   <li>{@link #chat(ChatContext)} — 同步对话（返回纯文本）</li>
 *   <li>{@link #chatWithEvents(ChatContext)} — 同步对话 + MCP 工具调用事件</li>
 *   <li>{@link #chatStream(ChatContext)} — 流式对话（返回 Flux&lt;String&gt;）</li>
 * </ul>
 * <p>
 * {@code @pipeline} execution-layer order=2<br>
 * {@code @pipeline-step} 熔断检查: SimpleCircuitBreaker(3次失败/5分钟恢复) → 熔断时返回降级提示<br>
 * {@code @pipeline-step} Advisor链: ContentSafetyAdvisor(HIGHEST) → MessageChatMemoryAdvisor → MyLoggerAdvisor<br>
 * {@code @pipeline-step} ChatClient.call/stream: LLM 大模型调用 → 成功/失败记录到熔断器<br>
 * 同步至: frontend/src/views/PipelineFlow.vue Advisor 子图 & 熔断节点
 *
 * @author lyx
 * @version 1.0.0
 * @date 2025/06/27 14:11
 */
@Component
@Slf4j
public class AssistantApp {

    @Resource
    private ContentSafetyAdvisor contentSafetyAdvisor;

    @Resource
    private McpToolCallbackAdapter mcpToolCallbackAdapter;

    @Resource
    private ChatPipeline chatPipeline;

    @Resource
    private AiModelConfigService aiModelConfigService;

    @Resource
    private LlmCallTraceService llmCallTraceService;

    @Resource
    private ContentSafetyService contentSafetyService;

    /** MCP 工具开关，默认启用，可通过 application.yml 的 chat.mcp.enabled 关闭 */
    @Value("${chat.mcp.enabled:true}")
    private boolean enableMcpTools;

    private final ChatModelFactory chatModelFactory;

    private final DatabaseChatMemory chatMemory;

    /** ChatClient 缓存上限，超出后按 LRU 自动淘汰最早最少使用的实例 */
    private static final int CHAT_CLIENT_CACHE_MAX = 32;

    /** ChatClient 缓存（LinkedHashMap + removeEldestEntry 实现 LRU 淘汰） */
    private final Map<String, ChatClient> chatClientCache =
            Collections.synchronizedMap(new LinkedHashMap<>(16, 0.75f, true) {
                @Override
                protected boolean removeEldestEntry(Map.Entry<String, ChatClient> eldest) {
                    if (size() > CHAT_CLIENT_CACHE_MAX) {
                        log.info("ChatClient 缓存淘汰: key={}", eldest.getKey());
                        return true;
                    }
                    return false;
                }
            });
    private final SimpleCircuitBreaker aiCircuitBreaker = new SimpleCircuitBreaker(3, 5 * 60 * 1000L, "AI");
    /** 全局 AI 调用的熔断 key（非模型配置级，整体熔断） */
    private static final Long AI_CIRCUIT_KEY = -1L;

    /** AI 熔断时的降级提示语 */
    private static final String CIRCUIT_OPEN_MESSAGE = "AI 服务暂时不可用，请稍后重试。";

    private static final String SYSTEM_PROMPT = "";

    /** 尾部空白缓冲上限：超过后强制发出，避免纯空白输出导致 buffer 无界增长 */
    private static final int MAX_TRAILING_WHITESPACE_BUFFER = 256;

    /** AI 回复落库的最大字符数，超过则保留头部 + 尾部 */
    private static final int MAX_AI_RESPONSE_CHARS = 2000;

    /** 截断时保留的头部字符数 */
    private static final int AI_RESPONSE_HEAD_CHARS = 1500;

    /** 截断时保留的尾部字符数 */
    private static final int AI_RESPONSE_TAIL_CHARS = 500;

    /** 埋点 JSON 序列化器（构建 ragHitsJson / toolCallsJson / historyMessagesJson） */
    private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();

    /** 历史消息每轮内容截断长度（避免 trace 行过大） */
    private static final int HISTORY_MESSAGE_MAX_CHARS = 200;

    /** 历史消息记录上限（条数） */
    private static final int HISTORY_MESSAGE_MAX_COUNT = 10;

    /** 错误消息截断长度 */
    private static final int ERROR_MESSAGE_MAX_CHARS = 500;

    /** MCP 工具调用结果落库截断长度（避免返回数据过大撑爆 trace 行） */
    private static final int TOOL_CALL_RESULT_MAX_CHARS = 2000;

    /** MCP 工具调用入参落库截断长度 */
    private static final int TOOL_CALL_INPUT_MAX_CHARS = 500;

    /**
     * 埋点附加元信息：承载错误分类/消息、token 用量与 MCP 工具调用事件，
     * 避免 recordTrace 参数过多。
     */
    private record TraceMeta(
            String errorType,
            String errorMessage,
            Integer promptTokens,
            Integer completionTokens,
            Integer totalTokens,
            List<ToolCallEvent> mcpEvents
    ) {}

    /**
     * 初始化 ChatClient
     *
     * @param chatModelFactory
     */
    public AssistantApp(ChatModelFactory chatModelFactory, DatabaseChatMemory chatMemory) {
        this.chatModelFactory = chatModelFactory;
        this.chatMemory = chatMemory;
    }

    private ChatClient getChatClient(String appType) {
        return getChatClient(appType, enableMcpTools ? List.of("*") : null);
    }

    /**
     * 获取 ChatClient（支持按角色过滤 MCP 工具），缓存上限 32 个，按 LRU 淘汰。
     */
    private ChatClient getChatClient(String appType, List<String> allowedMcpTools) {
        // 仅 null（无角色）降级为 ["*"]（向后兼容允许全部）；空列表表示有角色但未授权，保持空（不注册任何工具）
        final List<String> effectiveTools;
        if (enableMcpTools && allowedMcpTools == null) {
            effectiveTools = List.of("*");
        } else {
            effectiveTools = allowedMcpTools;
        }
        String cacheKey = appType + ":" + (effectiveTools == null ? "none" : String.join(",", effectiveTools));

        ChatClient cached = chatClientCache.get(cacheKey);
        if (cached != null) return cached;

        return chatClientCache.computeIfAbsent(cacheKey, key -> {
            ChatModel chatModel = chatModelFactory.getChatModel(appType);
            var builder = ChatClient.builder(chatModel);
            if (StringUtils.hasText(SYSTEM_PROMPT)) {
                builder.defaultSystem(SYSTEM_PROMPT);
            }
            // 注册 MCP 工具（按角色权限过滤）
            if (enableMcpTools && effectiveTools != null && !effectiveTools.isEmpty()) {
                ToolCallback[] mcpTools = mcpToolCallbackAdapter.getFilteredToolCallbacks(effectiveTools);
                if (mcpTools.length > 0) {
                    builder.defaultToolCallbacks(mcpTools);
                    log.info("已注册 {} 个 MCP 工具到 ChatClient [{}]", mcpTools.length, cacheKey);
                    for (ToolCallback tc : mcpTools) {
                        log.info("   工具: {} — {}", tc.getToolDefinition().name(), tc.getToolDefinition().description());
                    }
                } else {
                    log.warn("MCP 工具已启用但无可注册的工具 [{}]，请检查：", cacheKey);
                    log.warn("  1. 是否在「MCP 服务管理」页面添加了 MCP Server 配置？");
                    log.warn("  2. 配置是否已启用（is_active=true）？");
                    log.warn("  3. MCP Server 是否连接成功（点击「测试连接」验证）？");
                    log.warn("  4. MCP Server 是否暴露了工具（listTools 返回非空）？");
                }
            } else {
                log.info("MCP 工具未启用 [enableMcpTools={}, effectiveTools={}]", enableMcpTools, effectiveTools);
            }
            return builder
                    .defaultAdvisors(
                            contentSafetyAdvisor,
                            MessageChatMemoryAdvisor.builder(chatMemory).build(),
                            new MyLoggerAdvisor()
                    )
                    .build();
        });
    }

    public void clearCache() {
        chatClientCache.clear();
        log.info("AssistantApp ChatClient cache cleared");
    }

    /**
     * 构造传给模型的用户消息：无图片时返回纯文本 UserMessage；
     * 有图片时构造多模态 UserMessage（文本 + 图片 Media）。
     */
    private Message buildUserMessage(ChatRequest req) {
        List<String> imageUrls = req.ctx().imageUrls();
        if (imageUrls == null || imageUrls.isEmpty()) {
            return new UserMessage(req.finalMessage());
        }
        List<Media> mediaList = imageUrls.stream()
                .map(this::toImageMedia)
                .filter(m -> m != null)
                .toList();
        if (mediaList.isEmpty()) {
            return new UserMessage(req.finalMessage());
        }
        return UserMessage.builder()
                .text(req.finalMessage())
                .media(mediaList)
                .build();
    }

    /** 将图片 URL 转换为 Media（MIME 类型按扩展名推断），URL 非法时返回 null */
    private Media toImageMedia(String url) {
        try {
            return Media.builder()
                    .mimeType(resolveImageMimeType(url))
                    .data(url)
                    .build();
        } catch (Exception e) {
            log.warn("图片 URL 非法，跳过: {}", url);
            return null;
        }
    }

    /** 根据图片 URL 扩展名推断 MIME 类型 */
    private MimeType resolveImageMimeType(String url) {
        String path = url;
        int q = path.indexOf('?');
        if (q >= 0) {
            path = path.substring(0, q);
        }
        int dot = path.lastIndexOf('.');
        if (dot < 0) {
            return MimeTypeUtils.IMAGE_PNG;
        }
        String ext = path.substring(dot + 1).toLowerCase();
        if ("jpg".equals(ext)) {
            ext = "jpeg";
        }
        return MimeTypeUtils.parseMimeType("image/" + ext);
    }

    // ==================== 统一入口（新） ====================

    /**
     * 同步对话（新入口，委托 {@link ChatPipeline} 编排）。
     *
     * @param ctx 对话上下文
     * @return AI 回答文本
     */
    public String chat(ChatContext ctx) {
        return chatWithEvents(ctx).text();
    }

    /**
     * 同步对话 + MCP 工具调用事件（新入口）。
     * 比 {@link #chat(ChatContext)} 多返回本次触发的工具调用事件，供需要展示调用过程的场景。
     *
     * @param ctx 对话上下文
     * @return 回答文本 + MCP 事件
     */
    public ChatResult chatWithEvents(ChatContext ctx) {
        long startNanos = System.nanoTime();
        // 熔断：全局 AI 调用处于熔断状态，直接返回降级提示（不做 buildRequest，避免熔断期间仍走意图路由/检索）
        if (aiCircuitBreaker.isOpen(AI_CIRCUIT_KEY)) {
            log.warn("AI 调用熔断中，返回降级提示");
            recordTrace(ctx, null, CIRCUIT_OPEN_MESSAGE, 0, "BYPASS",
                    new TraceMeta("CIRCUIT_BREAK", "AI 服务熔断降级", null, null, null, null));
            return new ChatResult(CIRCUIT_OPEN_MESSAGE, List.of());
        }
        ChatRequest req = chatPipeline.buildRequest(ctx);
        if (req.faqHit()) {
            String faqAnswer = req.faqAnswer().get();
            recordTrace(ctx, req, faqAnswer, 0, "FAQ",
                    new TraceMeta(null, null, null, null, null, null));
            return new ChatResult(faqAnswer, List.of());
        }
        // 显式事件收集器 + 轮次计数器，通过 toolContext 传给 McpToolCallback，规避 Reactor 跨线程丢 ThreadLocal 的问题
        List<ToolCallEvent> events = new CopyOnWriteArrayList<>();
        AtomicInteger rounds = new AtomicInteger(0);
        try {
            ChatClient.ChatClientRequestSpec spec = getChatClient(ctx.appType(), ctx.allowedMcpTools())
                    .prompt()
                    .messages(buildUserMessage(req))
                    .advisors(s -> s.param(CONVERSATION_ID, ctx.chatId()));
            if (StringUtils.hasText(req.finalSystemPrompt())) {
                spec = spec.system(req.finalSystemPrompt());
            }
            spec = spec.toolContext(Map.of(
                    McpToolCallback.MCP_EVENTS_KEY, events,
                    McpToolCallback.MCP_ROUNDS_KEY, rounds));
            ChatResponse response = spec.call().chatResponse();
            String text = response.getResult().getOutput().getText();
            Usage usage = response.getMetadata() != null ? response.getMetadata().getUsage() : null;
            aiCircuitBreaker.recordSuccess(AI_CIRCUIT_KEY);
            recordTrace(ctx, req, text, elapsedMillis(startNanos), "COMPLETE",
                    new TraceMeta(null, null,
                            usage != null ? usage.getPromptTokens() : null,
                            usage != null ? usage.getCompletionTokens() : null,
                            usage != null ? usage.getTotalTokens() : null,
                            events));

            // 推荐问题已不再由主回复同步生成，改由 SuggestionGenerator 异步按需生成
            return new ChatResult(text, events, List.of());
        } catch (Exception e) {
            aiCircuitBreaker.recordFailure(AI_CIRCUIT_KEY);
            log.error("AI 同步调用失败: chatId={}, error={}", ctx.chatId(), e.getMessage());
            String fallback = "抱歉，AI 服务调用失败：" + e.getMessage();
            recordTrace(ctx, req, fallback, elapsedMillis(startNanos), "ERROR",
                    new TraceMeta(classifyError(e), maskError(e.getMessage()), null, null, null, events));
            return new ChatResult(fallback, List.of());
        }
    }

    /**
     * 流式对话（新入口，委托 {@link ChatPipeline} 编排）。
     * <p>
     * 注意：返回类型为 Flux&lt;String&gt;，Spring WebFlux 会对每个元素做 data: 帧包装。
     * 因此 Flux 元素必须只是纯文本内容，不应包含 event: / data: 等 SSE 协议行。
     *
     * @param ctx 对话上下文
     * @return 纯文本流式回答（每个元素为一段自然语言文本）
     */
    public Flux<String> chatStream(ChatContext ctx) {
        long startNanos = System.nanoTime();
        // 熔断：全局 AI 调用处于熔断状态（不做 buildRequest，避免熔断期间仍走意图路由/检索）
        if (aiCircuitBreaker.isOpen(AI_CIRCUIT_KEY)) {
            log.warn("AI 调用熔断中（流式），返回降级提示");
            recordTrace(ctx, null, CIRCUIT_OPEN_MESSAGE, 0, "BYPASS",
                    new TraceMeta("CIRCUIT_BREAK", "AI 服务熔断降级", null, null, null, null));
            return Flux.just(CIRCUIT_OPEN_MESSAGE);
        }
        ChatRequest req = chatPipeline.buildRequest(ctx);
        if (req.faqHit()) {
            // FAQ 命中：整段答案原样输出，由 SSE 编码器处理内部换行，
            // 后端不做任何格式增删（不拆行、不加换行、不补空格）。
            String faqAnswer = req.faqAnswer().get();
            recordTrace(ctx, req, faqAnswer, 0, "FAQ",
                    new TraceMeta(null, null, null, null, null, null));
            return Flux.just(faqAnswer);
        }
        // 显式事件收集器 + 轮次计数器，通过 toolContext 传给 McpToolCallback，规避 Reactor 跨线程丢 ThreadLocal 的问题
        List<ToolCallEvent> events = new CopyOnWriteArrayList<>();
        AtomicInteger rounds = new AtomicInteger(0);
        ChatClient.ChatClientRequestSpec spec = getChatClient(ctx.appType(), ctx.allowedMcpTools())
                .prompt()
                .messages(buildUserMessage(req))
                .advisors(s -> s.param(CONVERSATION_ID, ctx.chatId()));
        if (StringUtils.hasText(req.finalSystemPrompt())) {
            spec = spec.system(req.finalSystemPrompt());
        }
        spec = spec.toolContext(Map.of(
                McpToolCallback.MCP_EVENTS_KEY, events,
                McpToolCallback.MCP_ROUNDS_KEY, rounds));
        // 改为 chatResponse 流以采集 token 用量，再映射回纯文本流
        AtomicReference<Usage> usageRef = new AtomicReference<>();
        AtomicReference<String> errorTypeRef = new AtomicReference<>();
        AtomicReference<String> errorMessageRef = new AtomicReference<>();
        Flux<ChatResponse> responseFlux = spec.stream().chatResponse();
        Flux<String> rawStream = responseFlux
                .doOnNext(r -> {
                    if (r != null && r.getMetadata() != null && r.getMetadata().getUsage() != null) {
                        usageRef.set(r.getMetadata().getUsage());
                    }
                })
                .map(r -> {
                    String out = r != null && r.getResult() != null && r.getResult().getOutput() != null
                            ? r.getResult().getOutput().getText() : "";
                    return out != null ? out : "";
                });
        // 聚合所有分片用于埋点（在 doFinally 时取完整回复文本）
        StringBuilder aggregated = new StringBuilder();
        return preserveTrailingWhitespace(rawStream)
                .doOnNext(aggregated::append)
                .doOnComplete(() -> aiCircuitBreaker.recordSuccess(AI_CIRCUIT_KEY))
                .doOnError(e -> {
                    aiCircuitBreaker.recordFailure(AI_CIRCUIT_KEY);
                    errorTypeRef.set(classifyError(e));
                    errorMessageRef.set(maskError(e.getMessage()));
                    log.error("AI 流式调用失败: chatId={}, error={}", ctx.chatId(), e.getMessage());
                })
                .doFinally(signalType -> {
                    // 流式埋点：按终止信号区分状态，断连/异常也落库（events 由 toolContext 显式收集，跨线程安全）
                    String status = signalType == SignalType.ON_COMPLETE ? "COMPLETE"
                            : signalType == SignalType.ON_ERROR ? "ERROR" : "CANCEL";
                    Usage usage = usageRef.get();
                    recordTrace(ctx, req, aggregated.toString(), elapsedMillis(startNanos), status,
                            new TraceMeta(errorTypeRef.get(), errorMessageRef.get(),
                                    usage != null ? usage.getPromptTokens() : null,
                                    usage != null ? usage.getCompletionTokens() : null,
                                    usage != null ? usage.getTotalTokens() : null,
                                    events));
                })
                .onErrorResume(e -> Flux.just("抱歉，AI 服务调用失败：" + e.getMessage()));
    }

    /**
     * 流式对话（OpenAI Chat Completions 标准 SSE 格式）。
     * <p>
     * 复用 {@link #chatStream(ChatContext)} 的完整编排逻辑（熔断早退 / FAQ 命中早退 /
     * 正常流式调用 / 空白缓冲 / 埋点），差异在于把每个文本片段包装为 OpenAI 标准 JSON chunk：
     * 首片 delta 携带 role=assistant，流结束时追加 finish_reason=stop 的 chunk 与 [DONE]。
     * <p>
     * 每个 Flux 元素即一个完整 JSON 字符串，Spring WebFlux 自动加 data: 前缀。
     *
     * @param ctx 对话上下文
     * @return OpenAI 标准格式的流式回答
     */
    public Flux<String> chatStreamOpenAi(ChatContext ctx) {
        long startNanos = System.nanoTime();
        // OpenAI 标准 chunk 的公共元信息：同一次流式回答共享 id / created / model
        String completionId = "chatcmpl-" + UUID.randomUUID().toString().replace("-", "");
        long created = System.currentTimeMillis() / 1000;
        // 活跃模型配置可能不存在（返回 null），回退 unknown
        AiModelConfig cfg = null;
        try {
            cfg = aiModelConfigService.getActiveConfigWithFullKey(ctx.appType());
        } catch (Exception e) {
            log.warn("获取活跃模型配置失败，model 回退 unknown: chatId={}, error={}", ctx.chatId(), e.getMessage());
        }
        String model = (cfg != null && cfg.getModelName() != null) ? cfg.getModelName() : "unknown";
        // 熔断：全局 AI 调用处于熔断状态（不做 buildRequest，避免熔断期间仍走意图路由/检索）
        if (aiCircuitBreaker.isOpen(AI_CIRCUIT_KEY)) {
            log.warn("AI 调用熔断中（OpenAI 流式），返回降级提示");
            recordTrace(ctx, null, CIRCUIT_OPEN_MESSAGE, 0, "BYPASS",
                    new TraceMeta("CIRCUIT_BREAK", "AI 服务熔断降级", null, null, null, null));
            return openAiFallbackStream(completionId, model, created, CIRCUIT_OPEN_MESSAGE, true);
        }
        ChatRequest req = chatPipeline.buildRequest(ctx);
        if (req.faqHit()) {
            // FAQ 命中：整段答案包装为 OpenAI chunk，随后追加 stop + [DONE]
            String faqAnswer = req.faqAnswer().get();
            recordTrace(ctx, req, faqAnswer, 0, "FAQ",
                    new TraceMeta(null, null, null, null, null, null));
            return openAiFallbackStream(completionId, model, created, faqAnswer, true);
        }
        // 显式事件收集器 + 轮次计数器，通过 toolContext 传给 McpToolCallback，规避 Reactor 跨线程丢 ThreadLocal 的问题
        List<ToolCallEvent> events = new CopyOnWriteArrayList<>();
        AtomicInteger rounds = new AtomicInteger(0);
        ChatClient.ChatClientRequestSpec spec = getChatClient(ctx.appType(), ctx.allowedMcpTools())
                .prompt()
                .messages(buildUserMessage(req))
                .advisors(s -> s.param(CONVERSATION_ID, ctx.chatId()));
        if (StringUtils.hasText(req.finalSystemPrompt())) {
            spec = spec.system(req.finalSystemPrompt());
        }
        spec = spec.toolContext(Map.of(
                McpToolCallback.MCP_EVENTS_KEY, events,
                McpToolCallback.MCP_ROUNDS_KEY, rounds));
        // 改为 chatResponse 流以采集 token 用量，再映射回纯文本流
        AtomicReference<Usage> usageRef = new AtomicReference<>();
        AtomicReference<String> errorTypeRef = new AtomicReference<>();
        AtomicReference<String> errorMessageRef = new AtomicReference<>();
        Flux<ChatResponse> responseFlux = spec.stream().chatResponse();
        Flux<String> rawStream = responseFlux
                .doOnNext(r -> {
                    if (r != null && r.getMetadata() != null && r.getMetadata().getUsage() != null) {
                        usageRef.set(r.getMetadata().getUsage());
                    }
                })
                .map(r -> {
                    String out = r != null && r.getResult() != null && r.getResult().getOutput() != null
                            ? r.getResult().getOutput().getText() : "";
                    return out != null ? out : "";
                });
        // 聚合所有分片用于埋点（在 doFinally 时取完整回复文本）
        StringBuilder aggregated = new StringBuilder();
        return preserveTrailingWhitespace(rawStream)
                .doOnNext(aggregated::append)
                .map(chunk -> buildOpenAiChunk(completionId, model, created, chunk, false, null))
                .doOnComplete(() -> aiCircuitBreaker.recordSuccess(AI_CIRCUIT_KEY))
                .doOnError(e -> {
                    aiCircuitBreaker.recordFailure(AI_CIRCUIT_KEY);
                    errorTypeRef.set(classifyError(e));
                    errorMessageRef.set(maskError(e.getMessage()));
                    log.error("AI 流式调用失败（OpenAI）: chatId={}, error={}", ctx.chatId(), e.getMessage());
                })
                .doFinally(signalType -> {
                    // 流式埋点：按终止信号区分状态，断连/异常也落库（events 由 toolContext 显式收集，跨线程安全）
                    String status = signalType == SignalType.ON_COMPLETE ? "COMPLETE"
                            : signalType == SignalType.ON_ERROR ? "ERROR" : "CANCEL";
                    Usage usage = usageRef.get();
                    recordTrace(ctx, req, aggregated.toString(), elapsedMillis(startNanos), status,
                            new TraceMeta(errorTypeRef.get(), errorMessageRef.get(),
                                    usage != null ? usage.getPromptTokens() : null,
                                    usage != null ? usage.getCompletionTokens() : null,
                                    usage != null ? usage.getTotalTokens() : null,
                                    events));
                })
                // 首片（仅 role=assistant、无 content）在流订阅时立即发出，确保 SSE 响应头/首字节及时 flush。
                // 推理模型（如 doubao-seed）思考阶段 delta.content 为空、被 preserveTrailingWhitespace 吞掉，
                // 若不提前发首片，思考阶段将无任何字节输出，前端等待首字节会触发 60s 超时。
                .startWith(buildOpenAiChunk(completionId, model, created, "", true, null))
                // 流正常结束时追加 finish_reason=stop 的 chunk 与 [DONE]
                .concatWith(Flux.just(
                        buildOpenAiChunk(completionId, model, created, "", false, "stop"),
                        "[DONE]"))
                // 错误兜底：脱敏错误信息，避免泄露内部细节（首片 role 已提前发出，此处不再带 role）
                .onErrorResume(e -> openAiFallbackStream(completionId, model, created,
                        "抱歉，AI 服务调用失败：" + maskError(e.getMessage()), false));
    }

    /**
     * 组装单个 OpenAI Chat Completions 流式 chunk（JSON 字符串）。
     * <p>
     * 字段顺序固定为 id/object/created/model/choices，delta 内 role 在 content 前；
     * choices 使用 LinkedHashMap 以支持 finish_reason=null（Map.of 不允许 null 值）。
     *
     * @param id           chunk 唯一 ID（chatcmpl-xxx）
     * @param model        模型名称
     * @param created      创建时间（epoch 秒）
     * @param content      文本片段（stop 片传空串）
     * @param first        是否首片（首片 delta 携带 role=assistant）
     * @param finishReason 结束原因（中间片为 null，stop 片为 "stop"）
     * @return OpenAI 标准 chunk 的 JSON 字符串
     */
    private String buildOpenAiChunk(String id, String model, long created, String content, boolean first, String finishReason) {
        // delta：首片带 role=assistant（role 在 content 前），后续片仅 content，stop 片为空对象
        Map<String, Object> delta = new LinkedHashMap<>();
        if (first) {
            delta.put("role", "assistant");
        }
        if (content != null && !content.isEmpty()) {
            delta.put("content", content);
        }
        Map<String, Object> choice = new LinkedHashMap<>();
        choice.put("index", 0);
        choice.put("delta", delta);
        choice.put("finish_reason", finishReason);
        Map<String, Object> chunk = new LinkedHashMap<>();
        chunk.put("id", id);
        chunk.put("object", "chat.completion.chunk");
        chunk.put("created", created);
        chunk.put("model", model);
        chunk.put("choices", List.of(choice));
        try {
            return OBJECT_MAPPER.writeValueAsString(chunk);
        } catch (Exception e) {
            log.warn("序列化 OpenAI chunk 失败: {}", e.getMessage());
            return "{}";
        }
    }

    /**
     * 组装 OpenAI 格式的早退/兜底流：内容 chunk + finish_reason=stop + [DONE]。
     * 用于熔断降级、FAQ 命中与错误兜底三种场景。
     *
     * @param id       chunk 唯一 ID
     * @param model    模型名称
     * @param created  创建时间（epoch 秒）
     * @param content  完整回复文本
     * @param withRole 首片是否携带 role=assistant（熔断/FAQ 早退为 true；错误兜底时取决于此前是否已发出内容片）
     * @return OpenAI 标准格式的流
     */
    private Flux<String> openAiFallbackStream(String id, String model, long created, String content, boolean withRole) {
        return Flux.just(
                buildOpenAiChunk(id, model, created, content, withRole, null),
                buildOpenAiChunk(id, model, created, "", false, "stop"),
                "[DONE]");
    }

    /**
     * 缓冲以空白字符结尾的 chunk，将其与下一个 chunk 合并后再发出。
     * <p>
     * 原因：前端 SSE 解析会对每行执行 trim()，如果 chunk 以空白结尾（如 Markdown 标题 "## "），
     * 行尾空白会被削掉，导致 "## " + "一、..." 拼成 "##一、..."，破坏 Markdown 渲染。
     * 本方法不改变大模型输出的文本内容，只调整 chunk 边界以避开 SSE 的空白截断。
     *
     * @param source 原始大模型输出流
     * @return 调整后的流，每个元素均不以空白字符结尾（流末尾除外）
     */
    private static Flux<String> preserveTrailingWhitespace(Flux<String> source) {
        return Flux.create(sink -> {
            StringBuilder buffer = new StringBuilder();
            // 捕获内部订阅，下游取消/释放时同步取消上游，避免资源泄漏
            Disposable subscription = source.subscribe(
                chunk -> {
                    if (chunk == null) {
                        return;
                    }
                    buffer.append(chunk);
                    String current = buffer.toString();
                    if (current.isEmpty()) {
                        return;
                    }
                    char last = current.charAt(current.length() - 1);
                    // 当前累积内容不以空白结尾，可以安全发出
                    if (!Character.isWhitespace(last)) {
                        sink.next(current);
                        buffer.setLength(0);
                    } else if (buffer.length() >= MAX_TRAILING_WHITESPACE_BUFFER) {
                        // 尾部空白过长（如纯空白输出），强制发出以限制内存占用
                        sink.next(current);
                        buffer.setLength(0);
                    }
                    // 若以空白结尾，继续缓存，等下一个 chunk
                },
                e -> {
                    // 出错前先把已缓冲内容发出，避免已生成文本丢失
                    if (buffer.length() > 0) {
                        sink.next(buffer.toString());
                        buffer.setLength(0);
                    }
                    sink.error(e);
                },
                () -> {
                    if (buffer.length() > 0) {
                        sink.next(buffer.toString());
                    }
                    sink.complete();
                }
            );
            sink.onCancel(subscription);
            sink.onDispose(subscription);
        }, FluxSink.OverflowStrategy.BUFFER);
    }

    /**
     * 组装并异步写入一条 LLM 调用追踪记录。
     * <p>
     * 所有字段均在调用现场（非异步线程）组装为实体，避免把 ChatRequest/ChatContext
     * 全量对象塞进异步队列。落库前对用户消息与 AI 回复做敏感词脱敏与长度截断。
     *
     * @param ctx          对话上下文
     * @param req          编排决策（熔断早退时为 null）
     * @param responseText AI 回复文本
     * @param latencyMs    耗时（毫秒）
     * @param status       状态：COMPLETE / ERROR / CANCEL / FAQ / BYPASS
     */
    private void recordTrace(ChatContext ctx, ChatRequest req, String responseText, long latencyMs,
                             String status, TraceMeta meta) {
        try {
            // 走缓存的活跃配置（仅取模型元信息，不落 apiKey），避免每次对话在 Reactor 线程同步查库
            AiModelConfig cfg = aiModelConfigService.getActiveConfigWithFullKey(ctx.appType());
            // AI 回复脱敏 + 截断（保留头部 + 尾部，避免丢失末尾的推荐问题块）
            String maskedResponse = contentSafetyService.mask(responseText);
            String aiResponse = maskedResponse;
            boolean truncated = false;
            if (maskedResponse != null && maskedResponse.length() > MAX_AI_RESPONSE_CHARS) {
                aiResponse = maskedResponse.substring(0, AI_RESPONSE_HEAD_CHARS)
                        + "\n\n…（内容过长已截断）…\n\n"
                        + maskedResponse.substring(maskedResponse.length() - AI_RESPONSE_TAIL_CHARS);
                truncated = true;
            }

            // FAQ 命中详情（faqMatchResult 来自编排决策，未命中为 null）
            Long faqId = null;
            String faqQuestion = null;
            String faqMatchType = null;
            Double faqScore = null;
            if (req != null && req.faqMatchResult() != null && req.faqMatchResult().getFaq() != null) {
                faqId = req.faqMatchResult().getFaq().getId();
                faqQuestion = req.faqMatchResult().getFaq().getQuestion();
                faqMatchType = req.faqMatchResult().getMatchType();
                faqScore = req.faqMatchResult().getScore();
            }

            // 历史消息（一次读取，同时用于 JSON 与条数）
            List<Message> history = safeGetHistory(ctx.chatId());

            LlmCallTrace trace = LlmCallTrace.builder()
                    .conversationId(ctx.chatId())
                    .roleId(ctx.roleId())
                    .roleName(ctx.roleName())
                    .accountId(ctx.accountId())
                    .apiKeyId(ctx.apiKeyId())
                    .intent(req != null ? req.intent() : null)
                    .enableRag(ctx.enableRag())
                    .systemPrompt(req != null ? contentSafetyService.mask(req.finalSystemPrompt()) : null)
                    .globalPrompt(req != null ? contentSafetyService.mask(req.globalPrompt()) : null)
                    .rolePrompt(contentSafetyService.mask(ctx.systemPrompt()))
                    .userMessage(contentSafetyService.mask(ctx.message()))
                    .aiResponse(aiResponse)
                    .aiResponseTruncated(truncated)
                    .ragContext(req != null ? contentSafetyService.mask(req.ragContextText()) : null)
                    .ragHitsJson(buildRagHitsJson(req))
                    .faqHit(req != null ? req.faqHit() : false)
                    .faqId(faqId)
                    .faqQuestion(faqQuestion)
                    .faqMatchType(faqMatchType)
                    .faqScore(faqScore)
                    .searchMode(req != null ? req.searchMode() : null)
                    .hitCount(req != null ? req.hitCount() : null)
                    .toolCallsJson(buildToolCallsJson(meta != null ? meta.mcpEvents() : null))
                    .historyMessagesJson(buildHistoryJson(history))
                    .historyTurns(history != null ? history.size() : null)
                    .modelName(cfg != null ? cfg.getModelName() : null)
                    .provider(cfg != null ? cfg.getProvider() : null)
                    .temperature(cfg != null ? cfg.getTemperature() : null)
                    .maxTokens(cfg != null ? cfg.getMaxTokens() : null)
                    .promptTokens(meta != null ? meta.promptTokens() : null)
                    .completionTokens(meta != null ? meta.completionTokens() : null)
                    .totalTokens(meta != null ? meta.totalTokens() : null)
                    .errorType(meta != null ? meta.errorType() : null)
                    .errorMessage(meta != null ? contentSafetyService.mask(meta.errorMessage()) : null)
                    .latencyMs((int) latencyMs)
                    .status(status)
                    .build();
            llmCallTraceService.recordAsync(trace);
        } catch (Exception e) {
            log.warn("构造 LLM 调用追踪失败（不影响主流程）: chatId={}, error={}", ctx.chatId(), e.getMessage());
        }
    }

    /**
     * 序列化 RAG 命中文档片段为 JSON（含 documentId/title/chunkIndex/sourceName/score/searchMode）。
     * 无命中返回 null。
     */
    private String buildRagHitsJson(ChatRequest req) {
        if (req == null || req.hitDocuments() == null || req.hitDocuments().isEmpty()) {
            return null;
        }
        try {
            List<Map<String, Object>> items = new ArrayList<>();
            for (Document doc : req.hitDocuments()) {
                Map<String, Object> meta = doc.getMetadata();
                Map<String, Object> item = new LinkedHashMap<>();
                item.put("documentId", meta.get("documentId"));
                item.put("title", meta.get("title"));
                item.put("chunkIndex", meta.get("chunkIndex"));
                item.put("sourceName", meta.get("sourceName"));
                // 距离字段在不同检索实现下可能是 distance 或 score，二者取一
                Object score = meta.get("distance") != null ? meta.get("distance") : meta.get("score");
                item.put("score", score);
                item.put("searchMode", req.searchMode());
                items.add(item);
            }
            return OBJECT_MAPPER.writeValueAsString(items);
        } catch (Exception e) {
            log.warn("序列化 RAG 命中片段失败: {}", e.getMessage());
            return null;
        }
    }

    /**
     * 序列化 MCP 工具调用事件为 JSON（input/result 先脱敏再截断，避免返回数据过大撑爆 trace 行）。
     * 无事件返回 null。
     */
    private String buildToolCallsJson(List<ToolCallEvent> events) {
        if (events == null || events.isEmpty()) {
            return null;
        }
        try {
            List<Map<String, Object>> items = new ArrayList<>();
            for (ToolCallEvent e : events) {
                Map<String, Object> item = new LinkedHashMap<>();
                item.put("tool", e.tool());
                item.put("input", truncateToolCallText(contentSafetyService.mask(e.input()), TOOL_CALL_INPUT_MAX_CHARS));
                item.put("result", truncateToolCallText(contentSafetyService.mask(e.result()), TOOL_CALL_RESULT_MAX_CHARS));
                item.put("latencyMs", e.latencyMs());
                item.put("error", e.error());
                items.add(item);
            }
            return OBJECT_MAPPER.writeValueAsString(items);
        } catch (Exception ex) {
            log.warn("序列化 MCP 工具调用事件失败: {}", ex.getMessage());
            return null;
        }
    }

    /**
     * 对工具调用 input/result 做长度截断（保留头部 + 截断标记）。
     *
     * @param text     脱敏后的文本
     * @param maxChars 最大保留字符数
     * @return 未超限返回原文本；超限返回前 maxChars 个字符 + 截断标记
     */
    private String truncateToolCallText(String text, int maxChars) {
        if (text == null || text.length() <= maxChars) {
            return text;
        }
        int total = text.length();
        return text.substring(0, maxChars) + "…（共 " + total + " 字符，已截断）";
    }

    /** 从会话记忆读取最近若干条历史消息（失败返回 null）。 */
    private List<Message> safeGetHistory(String chatId) {
        try {
            return chatMemory.get(chatId, HISTORY_MESSAGE_MAX_COUNT);
        } catch (Exception e) {
            return null;
        }
    }

    /** 将历史消息序列化为 JSON（每轮内容截断）。无历史返回 null。 */
    private String buildHistoryJson(List<Message> history) {
        if (history == null || history.isEmpty()) {
            return null;
        }
        try {
            List<Map<String, String>> items = new ArrayList<>();
            for (Message m : history) {
                String role = m.getMessageType() != null ? m.getMessageType().name() : "unknown";
                String content = m.getText() != null ? m.getText() : "";
                if (content.length() > HISTORY_MESSAGE_MAX_CHARS) {
                    content = content.substring(0, HISTORY_MESSAGE_MAX_CHARS) + "…";
                }
                items.add(Map.of("role", role, "content", content));
            }
            return OBJECT_MAPPER.writeValueAsString(items);
        } catch (Exception e) {
            return null;
        }
    }

    /**
     * 按异常类型分类错误，供 trace.error_type 落库。
     */
    private String classifyError(Throwable t) {
        if (t == null) {
            return "UNKNOWN";
        }
        String name = t.getClass().getSimpleName();
        String msg = t.getMessage() == null ? "" : t.getMessage().toLowerCase();
        if (name.contains("Mcp") || msg.contains("mcp") || msg.contains("tool")) {
            return "MCP";
        }
        if (name.contains("CircuitBreaker") || msg.contains("circuit")) {
            return "CIRCUIT_BREAK";
        }
        if (name.contains("Validation") || name.contains("IllegalArgument")) {
            return "VALIDATION";
        }
        if (name.contains("Ai") || name.contains("OpenAi") || name.contains("DashScope")
                || name.contains("Http") || name.contains("Timeout") || msg.contains("timeout")) {
            return "LLM_API";
        }
        return "UNKNOWN";
    }

    /** 对异常消息做脱敏并截断，供 trace.error_message 落库。 */
    private String maskError(String message) {
        if (message == null || message.isBlank()) {
            return null;
        }
        String masked = contentSafetyService.mask(message);
        if (masked != null && masked.length() > ERROR_MESSAGE_MAX_CHARS) {
            return masked.substring(0, ERROR_MESSAGE_MAX_CHARS) + "…";
        }
        return masked;
    }

    /**
     * 计算自 startNanos 起的耗时（毫秒）。
     */
    private long elapsedMillis(long startNanos) {
        return (System.nanoTime() - startNanos) / 1_000_000;
    }

    /**
     * 统一检索引用来源（新入口，委托 {@link ChatPipeline#retrieveSources}）。
     *
     * @param ctx 对话上下文（使用 message / rewriteStrategy / categoryIds）
     * @return 命中的知识库片段，含 metadata（documentId/title/sourceName/chunkIndex/distance）
     */
    public List<Document> retrieveSources(ChatContext ctx) {
        return chatPipeline.retrieveSources(ctx);
    }

    // ==================== 内部工具方法已移除，不再需要 ====================
}