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

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

@@ -16,7 +16,10 @@ import org.springframework.http.codec.ServerSentEvent;
 import org.springframework.security.core.context.SecurityContextHolder;
 import org.springframework.web.bind.annotation.*;
 import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
 import reactor.core.publisher.SignalType;
+import reactor.core.publisher.Sinks;
+import reactor.core.scheduler.Schedulers;
 
 import org.springframework.jdbc.core.JdbcTemplate;
 import org.springframework.web.multipart.MultipartFile;
@@ -189,12 +192,14 @@ public class ChatController {
             if (kbFlux != null) return kbFlux;
         }
 
-        // 先发射 intent/status 事件，再用 flatMapMany 接回管道保持取消链完整。
-        // 纯 Reactor 管道（零裸 subscribe），连接断开时整条链路自动取消到百炼。
-        return Flux.just(
-                ServerSentEvent.<String>builder().event("intent").data(intent).build(),
-                ServerSentEvent.<String>builder().event("status").data("Retrieving...").build()
-        ).concatWith(retrieverService.searchReactive(query, intent, 20)
+        // L2：后台独立订阅 + Sinks 桥接。前端断开后后台继续把 LLM 跑完并存库，
+        // 用户下次进来 /recent-messages 仍能拿到。计时移到 pipeline 的 doFinally（真实完成）。
+        final java.util.concurrent.atomic.AtomicBoolean clientGone = new java.util.concurrent.atomic.AtomicBoolean(false);
+        final Sinks.Many<ServerSentEvent<String>> sink = Sinks.many().multicast().onBackpressureBuffer();
+        sink.tryEmitNext(ServerSentEvent.<String>builder().event("intent").data(intent).build());
+        sink.tryEmitNext(ServerSentEvent.<String>builder().event("status").data("Retrieving...").build());
+
+        Mono<Void> pipeline = retrieverService.searchReactive(query, intent, 20)
                 .map(docs -> rerankerService.rerank(docs, query, 5))
                 .flatMapMany(docs -> {
                     long t2 = System.currentTimeMillis();
@@ -207,58 +212,64 @@ public class ChatController {
                             t3 - t0, messages.stream().mapToInt(m -> m.get("content").length()).sum());
                     timings.put("prompt_ms", t3 - t0);
 
-                    // 用 StringBuilder 攒完整答案（Flux 内部同步操作，无线程安全问题）
-                    var fullAnswerBuf = new StringBuilder();
+                    sink.tryEmitNext(ServerSentEvent.<String>builder().event("status")
+                            .data("Matched " + docs.size() + " records, generating...").build());
 
-                    // 纯管道拼接，无裸 subscribe：
-                    //   [status] → [token1, token2, ...] → [meta] → complete
-                    Flux<ServerSentEvent<String>> contentFlux = llmService.chatStream(messages)
-                            .map(token -> {
+                    var fullAnswerBuf = new StringBuilder();
+                    return llmService.chatStream(messages)
+                            .doOnNext(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
-                            ));
-                        } catch (Exception e) {
-                            meta = "{}";
-                        }
-                        long t4 = System.currentTimeMillis();
-                        log.info("[chatStream] chatStream done, llmElapsed={}ms, total={}ms", t4 - t3, t4 - t0);
-                        timings.put("llm_ms", t4 - t3);
-                        timings.put("total_internal_ms", t4 - t0);
-                        persistenceService.saveMessage(userKey, cid, "user", query, intent, null);
-                        persistenceService.saveMessage(userKey, cid, "assistant", finalAnswer, intent, sources, brandRecs);
-                        qaCache.put(normalized, finalAnswer, intent, sources);
-
-                        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("Matched " + docs.size() + " records, generating...").build()
-                    );
-
-                    return Flux.concat(headFlux, contentFlux, tailFlux)
-                            .doOnError(e -> {
+                                if (!clientGone.get()) {
+                                    sink.tryEmitNext(ServerSentEvent.<String>builder().data(token).build());
+                                }
+                            })
+                            .doOnComplete(() -> {
+                                String finalAnswer = cleanAnswer(fullAnswerBuf.toString());
+                                final List<Map<String, Object>> sources = buildSources(docs);
+                                final List<Map<String, Object>> brandRecs = brandRecommendService.match(sources, finalAnswer);
                                 long t4 = System.currentTimeMillis();
-                                log.error("[chatStream] error, totalElapsed={}ms, error={}", t4 - t0, e.getMessage());
-                                qaCache.removePending(normalized);
+                                log.info("[chatStream] chatStream done, llmElapsed={}ms, total={}ms", t4 - t3, t4 - t0);
+                                timings.put("llm_ms", t4 - t3);
+                                timings.put("total_internal_ms", t4 - t0);
+                                persistenceService.saveMessage(userKey, cid, "user", query, intent, null);
+                                persistenceService.saveMessage(userKey, cid, "assistant", finalAnswer, intent, sources, brandRecs);
+                                qaCache.put(normalized, finalAnswer, intent, sources);
+                                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", cid
+                                        ));
+                                    } catch (Exception e) {
+                                        meta = "{}";
+                                    }
+                                    sink.tryEmitNext(ServerSentEvent.<String>builder().event("meta").data(meta).build());
+                                    sink.tryEmitComplete();
+                                }
                             });
-                }))
+                })
+                .then()
+                .doOnError(e -> {
+                    long t4 = System.currentTimeMillis();
+                    log.error("[chatStream] error, totalElapsed={}ms, error={}", t4 - t0, e.getMessage());
+                    qaCache.removePending(normalized);
+                    if (!clientGone.get()) {
+                        sink.tryEmitError(e);
+                    }
+                })
                 .doFinally(sig -> recordChatTiming(userKey, chatRequestId, chatIp, chatUa, "stream", t0, sig, timings));
+
+        // 后台跑完（不绑 SSE 连接），subscribeOn boundedElastic 不占 servlet/Netty 线程
+        pipeline.subscribeOn(Schedulers.boundedElastic())
+                .subscribe(v -> {}, e -> log.error("[chatStream] 后台 pipeline 出错: cid={}, error={}", cid, e.getMessage()));
+
+        return sink.asFlux().doFinally(sig -> {
+            if (sig == SignalType.CANCEL) {
+                clientGone.set(true);
+                log.info("[chatStream] 客户端断开，后台继续生成并存库: cid={}, request_id={}", cid, chatRequestId);
+            }
+        });
     }
 
     /** 将缓存命中结果以流式 SSE 形式返回 */