调整ai问答记录，没有回答完，后台也继续处理直到问答完成

backend-java/src/main/java/com/pharmacopoeia/controller/ChatController.java

@@ -8,6 +8,8 @@ import com.pharmacopoeia.dto.MultimodalChatRequest;
 import com.pharmacopoeia.service.*;
 import com.pharmacopoeia.util.IpUtils;
 import jakarta.servlet.http.HttpServletRequest;
+import lombok.Builder;
+import lombok.Data;
 import lombok.extern.slf4j.Slf4j;
 import org.jetbrains.annotations.NotNull;
 import org.springframework.http.MediaType;
@@ -353,7 +355,8 @@ public class ChatController {
     }
 
     @PostMapping(value = "/stream-image", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
-    public Flux<ServerSentEvent<String>> chatStreamImage(@RequestBody ImageChatRequest request) {
+    public Flux<ServerSentEvent<String>> chatStreamImage(@RequestBody ImageChatRequest request,
+                                                         HttpServletRequest servletRequest) {
         final String userKey = getCurrentUserKey();
         final String cid = request.getConversationId() != null && !request.getConversationId().isBlank()
                 ? request.getConversationId()
@@ -366,73 +369,36 @@ public class ChatController {
             );
         }
 
-        // 先发状态，不等图片分析完成
-        return Flux.just(
-                ServerSentEvent.<String>builder().event("status").data("正在分析图片（OCR 文字识别）...").build()
-        ).concatWith(
-            llmService.analyzeImageStream(request.getImageBase64(), request.getMimeType(),
-                    "请分析这张图片，提取其中所有文字信息（OCR），特别是药品名称、成分、用法用量等。简要输出。")
-                    .collectList()
-                    .flatMapMany(tokens -> {
-                        String ocrText = String.join("", tokens);
-
-                        String query = (!request.getMessage().isBlank())
-                                ? request.getMessage() + "\n\n（图片OCR提取内容：" + ocrText + "）"
-                                : ocrText;
-                        final String intent = retrieverService.classifyIntent(query);
-
-                        return Flux.just(
-                                ServerSentEvent.<String>builder().event("status").data("图片分析完成，正在检索药典知识库...").build(),
-                                ServerSentEvent.<String>builder().event("intent").data(intent).build()
-                        ).concatWith(retrieverService.searchReactive(query, intent, 20)
-                                .map(docs -> rerankerService.rerank(docs, query, 5))
-                                .flatMapMany(docs -> {
-                                    final List<Map<String, String>> messages = promptService.buildPrompt(query, docs, intent);
-                                    String imageContext = "\n\n【图片分析结果】\n" + ocrText + "\n";
-                                    messages.getFirst().put("content", messages.getFirst().get("content") + imageContext);
-
-                                    var fullAnswerBuf = new StringBuilder();
-
-                                    Flux<ServerSentEvent<String>> contentFlux = llmService.chatStream(messages, true)
-                                            .map(token -> {
-                                                fullAnswerBuf.append(token);
-                                                return ServerSentEvent.<String>builder().data(token).build();
-                                            });
-
-                                    Flux<ServerSentEvent<String>> tailFlux = Flux.defer(() -> {
-                                        String finalAnswer = cleanAnswer(fullAnswerBuf.toString());
-                                        final List<Map<String, Object>> sources = buildSources(docs);
-                                        final List<Map<String, Object>> brandRecs = brandRecommendService.match(sources, finalAnswer);
-                                        String meta;
-                                        try {
-                                            meta = new com.fasterxml.jackson.databind.ObjectMapper().writeValueAsString(Map.of(
-                                                    "intent", intent,
-                                                    "sources", sources,
-                                                    "conversation_id", cid,
-                                                    "ocr_text", ocrText
-                                            ));
-                                        } catch (Exception e) {
-                                            meta = "{}";
-                                        }
-                                        persistenceService.saveMessage(userKey, cid, "user",
-                                                request.getMessage().isBlank() ? "[图片]" : "[图片] " + request.getMessage(),
-                                                intent, null);
-                                        persistenceService.saveMessage(userKey, cid, "assistant", finalAnswer, intent, sources, brandRecs);
-                                        return Flux.just(
-                                                buildBrandRecommendEvent(brandRecs),
-                                                ServerSentEvent.<String>builder().event("meta").data(meta).build()
-                                        );
-                                    });
-
-                                    Flux<ServerSentEvent<String>> headFlux = Flux.just(
-                                            ServerSentEvent.<String>builder().event("status")
-                                                    .data("已匹配 " + docs.size() + " 条药典资料，生成回答中（已启用联网搜索）...").build()
-                                    );
-
-                                    return Flux.concat(headFlux, contentFlux, tailFlux);
-                                }));
-                    })
-        );
+        final String chatRequestId = servletRequest.getHeader("X-Request-Id");
+        final String chatIp = IpUtils.getClientIp(servletRequest);
+        final String chatUa = servletRequest.getHeader("User-Agent");
+        final long t0 = System.currentTimeMillis();
+
+        // 复制为 effectively final，供 lambda 捕获
+        final String msg = request.getMessage() != null ? request.getMessage() : "";
+        final String imageBase64 = request.getImageBase64();
+        final String mimeType = request.getMimeType();
+
+        DetachedCfg cfg = DetachedCfg.builder()
+                .cid(cid).userKey(userKey)
+                .requestId(chatRequestId).ip(chatIp).ua(chatUa)
+                .t0(t0).endpoint("stream-image")
+                .rawMsg(msg)
+                .hasMedia(true).mediaLabel("图片")
+                .ocrSupplier(() -> llmService.analyzeImageStream(imageBase64, mimeType,
+                        "请分析这张图片，提取其中所有文字信息（OCR），特别是药品名称、成分、用法用量等关键药学信息。简要输出即可。"))
+                .ocrStartStatus("正在分析图片（OCR 文字识别）...")
+                .ocrHeaderData(null)
+                .ocrTrailingSeparator(null)
+                .ocrDoneStatus("图片分析完成，正在检索药典知识库...")
+                .ragHeaderData(null)
+                .queryBuilder(ocrText -> (!msg.isBlank())
+                        ? msg + "\n\n（图片OCR提取内容：" + ocrText + "）"
+                        : ocrText)
+                .userMsgBuilder(ocrText -> msg.isBlank() ? "[图片]" : "[图片] " + msg)
+                .enableSearchFn(ocrText -> true)
+                .build();
+        return buildDetachedPipeline(cfg);
     }
 
     // ============================================================
@@ -499,7 +465,8 @@ public class ChatController {
     }
 
     @PostMapping(value = "/stream-multimodal", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
-    public Flux<ServerSentEvent<String>> chatStreamMultimodal(@RequestBody MultimodalChatRequest request) {
+    public Flux<ServerSentEvent<String>> chatStreamMultimodal(@RequestBody MultimodalChatRequest request,
+                                                              HttpServletRequest servletRequest) {
         final String userKey = getCurrentUserKey();
         final String cid = request.getConversationId() != null && !request.getConversationId().isBlank()
                 ? request.getConversationId()
@@ -508,115 +475,244 @@ public class ChatController {
                 && request.getMediaType() != null && !request.getMediaType().isBlank();
         final String mediaLabel = hasMedia && "video".equals(request.getMediaType()) ? "视频" : "图片";
 
+        final String chatRequestId = servletRequest.getHeader("X-Request-Id");
+        final String chatIp = IpUtils.getClientIp(servletRequest);
+        final String chatUa = servletRequest.getHeader("User-Agent");
+        final long t0 = System.currentTimeMillis();
+
+        // 复制为 effectively final，供 lambda 捕获
+        final String rawMsg = request.getMessage();
+        final String mediaBase64 = request.getMediaBase64();
+        final String mediaType = request.getMediaType();
+        final String mediaMime = request.getMediaMime();
+
         if (hasMedia) {
-            // 先发 OCR section header
-            return Flux.just(
-                    ServerSentEvent.<String>builder().event("status")
-                            .data("🔍 正在分析" + mediaLabel + "...").build(),
-                    ServerSentEvent.<String>builder().data("【📷 " + mediaLabel + "分析】\n\n").build()
-            ).concatWith(
-                    // 用 collectList 收集流式结果，同时攒 OCR 文本，全程在管道内无阻塞
-                    llmService.analyzeMediaStream(request.getMediaBase64(), request.getMediaType(),
-                            request.getMediaMime(), "")
-                            .flatMap(token -> Flux.just(
-                                    ServerSentEvent.<String>builder().data(token).build()  // 前端实时看到
-                            ))
-                            .concatWith(Flux.just(ServerSentEvent.<String>builder().data("\n\n").build()))
-                            // collectList 后再 flatMapMany 接 RAG——Reactor 取消时自动 abort
-                            .collectList()
-                            .flatMapMany(events -> {
-                                // 从已发送的事件中拼回 OCR 文本（不额外调 API）
-                                StringBuilder ocrBuilder = new StringBuilder();
-                                for (ServerSentEvent<String> e : events) {
-                                    String d = e.data();
-                                    if (d != null && !"\n\n".equals(d)) {
-                                        ocrBuilder.append(d);
-                                    }
-                                }
-                                String mediaOcr = ocrBuilder.toString().trim();
-                                // 不重复发送 OCR 事件（已经通过 analyzeMediaStream 实时发送过了）
-                                return buildRagPipeline(cid,
-                                        request.getMessage(), mediaOcr, userKey, mediaLabel);
-                            })
-            );
+            DetachedCfg cfg = DetachedCfg.builder()
+                    .cid(cid).userKey(userKey)
+                    .requestId(chatRequestId).ip(chatIp).ua(chatUa)
+                    .t0(t0).endpoint("stream-multimodal")
+                    .rawMsg(rawMsg)
+                    .hasMedia(true).mediaLabel(mediaLabel)
+                    .ocrSupplier(() -> llmService.analyzeMediaStream(mediaBase64, mediaType, mediaMime, ""))
+                    .ocrStartStatus("🔍 正在分析" + mediaLabel + "...")
+                    .ocrHeaderData("【📷 " + mediaLabel + "分析】\n\n")
+                    .ocrTrailingSeparator("\n\n")
+                    .ocrDoneStatus("📚 检索药典知识库...")
+                    .ragHeaderData("\n【📚 药典参考回答】\n\n")
+                    .queryBuilder(ocrText -> {
+                        String rq = rawMsg != null ? rawMsg.trim() : "";
+                        if (!rq.isEmpty() && ocrText != null && !ocrText.isEmpty()) {
+                            return rq + "\n\n（" + mediaLabel + "OCR提取内容：" + ocrText + "）";
+                        } else if (ocrText != null && !ocrText.isEmpty()) {
+                            return ocrText;
+                        } else if (rq.isEmpty()) {
+                            return "请介绍一下自己";
+                        }
+                        return rq;
+                    })
+                    .userMsgBuilder(ocrText -> (rawMsg != null && !rawMsg.isBlank()) ? rawMsg
+                            : (ocrText != null && !ocrText.isEmpty()) ? "[" + mediaLabel + "]" : "")
+                    .enableSearchFn(ocrText -> (ocrText != null && !ocrText.isEmpty()) || props.isEnableWebSearch())
+                    .build();
+            return buildDetachedPipeline(cfg);
         } else {
-            return buildRagPipeline(cid, request.getMessage(), "", userKey, "");
+            DetachedCfg cfg = DetachedCfg.builder()
+                    .cid(cid).userKey(userKey)
+                    .requestId(chatRequestId).ip(chatIp).ua(chatUa)
+                    .t0(t0).endpoint("stream-multimodal")
+                    .rawMsg(rawMsg)
+                    .hasMedia(false).mediaLabel("")
+                    .ocrSupplier(null)
+                    .ocrStartStatus(null)
+                    .ocrHeaderData(null)
+                    .ocrTrailingSeparator(null)
+                    .ocrDoneStatus(null)
+                    .ragHeaderData("\n【📚 药典参考回答】\n\n")
+                    .queryBuilder(ocrText -> {
+                        String rq = rawMsg != null ? rawMsg.trim() : "";
+                        return rq.isEmpty() ? "请介绍一下自己" : rq;
+                    })
+                    .userMsgBuilder(ocrText -> (rawMsg != null && !rawMsg.isBlank()) ? rawMsg : "")
+                    .enableSearchFn(ocrText -> props.isEnableWebSearch())
+                    .build();
+            return buildDetachedPipeline(cfg);
         }
     }
 
-    /** 构建 RAG → 百炼流式管道（纯 Reactor，零裸 subscribe） */
-    private Flux<ServerSentEvent<String>> buildRagPipeline(String cid, String rawMsg,
-                                                          String ocrText, String userKey,
-                                                          String mediaLabel) {
-        String rq = rawMsg != null ? rawMsg.trim() : "";
-        if (!rq.isEmpty() && ocrText != null && !ocrText.isEmpty()) {
-            rq = rq + "\n\n（" + mediaLabel + "OCR提取内容：" + ocrText + "）";
-        } else if (ocrText != null && !ocrText.isEmpty()) {
-            rq = ocrText;
-        } else if (rq.isEmpty()) {
-            rq = "请介绍一下自己";
-        }
-        final String query = rq;
-        final String intent = retrieverService.classifyIntent(query);
-        final boolean enableSearch = (ocrText != null && !ocrText.isEmpty()) || props.isEnableWebSearch();
-
-        // 如果等待检索，先发状态
-        Flux<ServerSentEvent<String>> prefixFlux = (ocrText != null && !ocrText.isEmpty())
-                ? Flux.just(ServerSentEvent.<String>builder().event("status")
-                        .data("📚 检索药典知识库...").build())
-                : Flux.just(ServerSentEvent.<String>builder().event("intent").data(intent).build());
-
-        Flux<ServerSentEvent<String>> intentFlux = (ocrText != null && !ocrText.isEmpty())
-                ? Flux.just(ServerSentEvent.<String>builder().event("intent").data(intent).build())
-                : Flux.empty();
-
-        return prefixFlux.concatWith(intentFlux)
-                .concatWith(retrieverService.searchReactive(query, intent, 20)
-                        .map(docs -> rerankerService.rerank(docs, query, 5))
-                        .flatMapMany(docs -> {
-                            final List<Map<String, String>> messages = promptService.buildPrompt(query, docs, intent);
-                            if (ocrText != null && !ocrText.isEmpty()) {
-                                messages.getFirst().put("content",
-                                        messages.getFirst().get("content") + "\n\n【" + mediaLabel + "分析结果】\n" + ocrText + "\n");
-                            }
-
-                            var fullAnswerBuf = new StringBuilder();
+    /** 图片/多模态流式两端点的事件文案与行为差异，供 buildDetachedPipeline 使用 */
+    @Data
+    @Builder
+    private static class DetachedCfg {
+        String cid, userKey, requestId, ip, ua, endpoint;
+        long t0;
+        String rawMsg;
+        boolean hasMedia;
+        String mediaLabel;
+        /** null = 无媒体，不跑 OCR 阶段 */
+        java.util.function.Supplier<Flux<String>> ocrSupplier;
+        /** null = 不发 */
+        String ocrStartStatus, ocrHeaderData, ocrTrailingSeparator, ocrDoneStatus, ragHeaderData;
+        /** ocrText -> query */
+        java.util.function.Function<String, String> queryBuilder;
+        /** ocrText -> userMsg */
+        java.util.function.Function<String, String> userMsgBuilder;
+        /** ocrText -> 是否启用联网搜索 */
+        java.util.function.Function<String, Boolean> enableSearchFn;
+    }
 
-                            Flux<ServerSentEvent<String>> contentFlux = llmService.chatStream(messages, enableSearch)
-                                    .map(token -> {
-                                        fullAnswerBuf.append(token);
-                                        return ServerSentEvent.<String>builder().data(token).build();
-                                    });
+    /**
+     * 图片/多模态统一后台解耦管道（镜像 /stream 的 L2）。
+     * OCR(可选) → RAG → LLM → 存库 整条在 Schedulers.boundedElastic 后台独立订阅，不绑 SSE 连接；
+     * 前端断开后后台继续跑完并存库，用户重进 /recent-messages 仍能拿到。
+     * 存库在 chatStream 的 doOnComplete 里，不受 clientGone 影响；clientGone 只 gate 是否往 sink emit。
+     * SSE 事件序列与改造前逐字一致。
+     */
+    private Flux<ServerSentEvent<String>> buildDetachedPipeline(DetachedCfg cfg) {
+        final java.util.concurrent.atomic.AtomicBoolean clientGone = new java.util.concurrent.atomic.AtomicBoolean(false);
+        final Sinks.Many<ServerSentEvent<String>> sink = Sinks.many().multicast().onBackpressureBuffer();
+        final StringBuilder ocrBuf = new StringBuilder();
+        final StringBuilder answerBuf = new StringBuilder();
+        final String[] queryBox = new String[1];
+        final String[] intentBox = new String[1];
+        final boolean[] enableSearchBox = new boolean[1];
+        final java.util.Map<String, Long> timings = new java.util.concurrent.ConcurrentHashMap<>();
+        final long t0 = cfg.getT0();
 
-                            Flux<ServerSentEvent<String>> tailFlux = Flux.defer(() -> {
-                                final List<Map<String, Object>> sources = buildSources(docs);
-                                String finalAnswer = cleanAnswer(fullAnswerBuf.toString());
-                                final List<Map<String, Object>> brandRecs = brandRecommendService.match(sources, finalAnswer);
-                                String meta;
-                                try {
-                                    meta = new com.fasterxml.jackson.databind.ObjectMapper().writeValueAsString(Map.of(
-                                            "intent", intent, "sources", sources, "conversation_id", cid,
-                                            "ocr_text", ocrText != null ? ocrText : ""));
-                                } catch (Exception e) { meta = "{}"; }
-                                String userMsg = (rawMsg != null && !rawMsg.isBlank()) ? rawMsg
-                                        : (ocrText != null && !ocrText.isEmpty()) ? "[" + mediaLabel + "]" : "";
-                                persistenceService.saveMessage(userKey, cid, "user", userMsg, intent, null);
-                                persistenceService.saveMessage(userKey, cid, "assistant", finalAnswer, intent, sources, brandRecs);
-                                return Flux.just(
-                                        buildBrandRecommendEvent(brandRecs),
-                                        ServerSentEvent.<String>builder().event("meta").data(meta).build()
-                                );
-                            });
+        // 阶段1：前置事件（同步 emit，进入 sink 缓冲）
+        if (cfg.isHasMedia()) {
+            if (cfg.getOcrStartStatus() != null) {
+                sink.tryEmitNext(ServerSentEvent.<String>builder().event("status").data(cfg.getOcrStartStatus()).build());
+            }
+            if (cfg.getOcrHeaderData() != null) {
+                sink.tryEmitNext(ServerSentEvent.<String>builder().data(cfg.getOcrHeaderData()).build());
+            }
+        } else {
+            // 无媒体：query/intent 前置计算并先发 intent
+            String q = cfg.getQueryBuilder().apply("");
+            queryBox[0] = q;
+            intentBox[0] = retrieverService.classifyIntent(q);
+            enableSearchBox[0] = cfg.getEnableSearchFn().apply("");
+            sink.tryEmitNext(ServerSentEvent.<String>builder().event("intent").data(intentBox[0]).build());
+        }
 
-                            Flux<ServerSentEvent<String>> headFlux = Flux.just(
-                                    ServerSentEvent.<String>builder().data("\n【📚 药典参考回答】\n\n").build(),
-                                    ServerSentEvent.<String>builder().event("status")
+        // 阶段2：OCR 流（仅 hasMedia），token 实时下发 sink 并累积
+        Mono<Void> ocrStage = (cfg.isHasMedia() && cfg.getOcrSupplier() != null)
+                ? cfg.getOcrSupplier().get()
+                    .doOnNext(t -> {
+                        ocrBuf.append(t);
+                        if (!clientGone.get()) {
+                            sink.tryEmitNext(ServerSentEvent.<String>builder().data(t).build());
+                        }
+                    })
+                    .doOnComplete(() -> timings.put("ocr_ms", System.currentTimeMillis() - t0))
+                    .then()
+                : Mono.empty();
+
+        Mono<Void> pipeline = ocrStage
+                // 阶段3：OCR 后事件 + 计算 query/intent/enableSearch（用 box 暂存，供 stage4 defer 读取）
+                .then(Mono.<Void>fromRunnable(() -> {
+                    if (cfg.isHasMedia()) {
+                        if (!clientGone.get() && cfg.getOcrTrailingSeparator() != null) {
+                            sink.tryEmitNext(ServerSentEvent.<String>builder().data(cfg.getOcrTrailingSeparator()).build());
+                        }
+                        String ocrText = ocrBuf.toString().trim();
+                        String q = cfg.getQueryBuilder().apply(ocrText);
+                        queryBox[0] = q;
+                        intentBox[0] = retrieverService.classifyIntent(q);
+                        enableSearchBox[0] = cfg.getEnableSearchFn().apply(ocrText);
+                        if (!clientGone.get()) {
+                            if (cfg.getOcrDoneStatus() != null) {
+                                sink.tryEmitNext(ServerSentEvent.<String>builder().event("status").data(cfg.getOcrDoneStatus()).build());
+                            }
+                            sink.tryEmitNext(ServerSentEvent.<String>builder().event("intent").data(intentBox[0]).build());
+                        }
+                    }
+                }))
+                // 阶段4：检索 + rerank + LLM 流式（defer 读 box，避免 eager 求值时空指针）
+                .then(Mono.defer(() -> {
+                    final String query = queryBox[0];
+                    final String intent = intentBox[0];
+                    final boolean enableSearch = enableSearchBox[0];
+                    return retrieverService.searchReactive(query, intent, 20)
+                            .map(docs -> rerankerService.rerank(docs, query, 5))
+                            .flatMapMany(docs -> {
+                                long ts = System.currentTimeMillis();
+                                timings.put("search_rerank_ms", ts - t0);
+                                final List<Map<String, String>> messages = promptService.buildPrompt(query, docs, intent);
+                                if (cfg.isHasMedia()) {
+                                    String ocrText = ocrBuf.toString().trim();
+                                    if (ocrText != null && !ocrText.isEmpty()) {
+                                        messages.getFirst().put("content",
+                                                messages.getFirst().get("content") + "\n\n【" + cfg.getMediaLabel() + "分析结果】\n" + ocrText + "\n");
+                                    }
+                                }
+                                if (!clientGone.get()) {
+                                    if (cfg.getRagHeaderData() != null) {
+                                        sink.tryEmitNext(ServerSentEvent.<String>builder().data(cfg.getRagHeaderData()).build());
+                                    }
+                                    sink.tryEmitNext(ServerSentEvent.<String>builder().event("status")
                                             .data("已匹配 " + docs.size() + " 条药典资料，生成回答中"
-                                                    + (enableSearch ? "（已启用联网搜索）" : "") + "...").build()
-                            );
+                                                    + (enableSearch ? "（已启用联网搜索）" : "") + "...").build());
+                                }
+                                final long tllm = System.currentTimeMillis();
+                                return llmService.chatStream(messages, enableSearch)
+                                        .doOnNext(token -> {
+                                            answerBuf.append(token);
+                                            if (!clientGone.get()) {
+                                                sink.tryEmitNext(ServerSentEvent.<String>builder().data(token).build());
+                                            }
+                                        })
+                                        .doOnComplete(() -> {
+                                            long t4 = System.currentTimeMillis();
+                                            timings.put("llm_ms", t4 - tllm);
+                                            timings.put("total_internal_ms", t4 - t0);
+                                            String finalAnswer = cleanAnswer(answerBuf.toString());
+                                            List<Map<String, Object>> sources = buildSources(docs);
+                                            List<Map<String, Object>> brandRecs = brandRecommendService.match(sources, finalAnswer);
+                                            String ocrText = cfg.isHasMedia() ? ocrBuf.toString().trim() : "";
+                                            String userMsg = cfg.getUserMsgBuilder().apply(ocrText);
+                                            persistenceService.saveMessage(cfg.getUserKey(), cfg.getCid(), "user", userMsg, intent, null);
+                                            persistenceService.saveMessage(cfg.getUserKey(), cfg.getCid(), "assistant", finalAnswer, intent, sources, brandRecs);
+                                            if (!clientGone.get()) {
+                                                sink.tryEmitNext(buildBrandRecommendEvent(brandRecs));
+                                                String meta;
+                                                try {
+                                                    meta = new com.fasterxml.jackson.databind.ObjectMapper().writeValueAsString(Map.of(
+                                                            "intent", intent,
+                                                            "sources", sources,
+                                                            "conversation_id", cfg.getCid(),
+                                                            "ocr_text", ocrText != null ? ocrText : ""
+                                                    ));
+                                                } catch (Exception e) {
+                                                    meta = "{}";
+                                                }
+                                                sink.tryEmitNext(ServerSentEvent.<String>builder().event("meta").data(meta).build());
+                                                sink.tryEmitComplete();
+                                            }
+                                        });
+                            }).then();
+                }))
+                .doOnError(e -> {
+                    log.error("[detached] error: cid={}, endpoint={}, error={}", cfg.getCid(), cfg.getEndpoint(), e.getMessage());
+                    if (!clientGone.get()) {
+                        sink.tryEmitError(e);
+                    }
+                })
+                .doFinally(sig -> recordChatTiming(cfg.getUserKey(), cfg.getRequestId(), cfg.getIp(), cfg.getUa(),
+                        cfg.getEndpoint(), t0, sig, timings));
 
-                            return Flux.concat(headFlux, contentFlux, tailFlux);
-                        }));
+        // 后台独立订阅（不绑 SSE），subscribeOn boundedElastic 不占 Netty 线程
+        pipeline.subscribeOn(Schedulers.boundedElastic())
+                .subscribe(v -> {}, e -> log.error("[detached] 后台 pipeline 出错: cid={}, endpoint={}, error={}",
+                        cfg.getCid(), cfg.getEndpoint(), e.getMessage()));
+
+        return sink.asFlux().doFinally(sig -> {
+            if (sig == SignalType.CANCEL) {
+                clientGone.set(true);
+                log.info("[detached] 客户端断开，后台继续生成并存库: cid={}, endpoint={}, request_id={}",
+                        cfg.getCid(), cfg.getEndpoint(), cfg.getRequestId());
+            }
+        });
     }
 
     // ============================================================