当发生异常时调用此方法，会执行以下操作：
- *
- * 使用 CAS 操作确保连接只被关闭一次
- * 调用 SseEmitter 的 completeWithError 方法，通知客户端连接异常终止
- *
- * @param throwable 导致连接关闭的异常对象
- */
- private void closeWithError(Throwable throwable) {
- // 使用 CAS 原子操作，确保只关闭一次
- if (closed.compareAndSet(false, true)) {
- emitter.completeWithError(throwable);
- }
- }
}
diff --git a/framework/src/test/java/com/nageoffer/ai/ragent/framework/web/SseEmitterSenderTest.java b/framework/src/test/java/com/nageoffer/ai/ragent/framework/web/SseEmitterSenderTest.java
new file mode 100644
index 000000000..1d474431a
--- /dev/null
+++ b/framework/src/test/java/com/nageoffer/ai/ragent/framework/web/SseEmitterSenderTest.java
@@ -0,0 +1,47 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.nageoffer.ai.ragent.framework.web;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
+
+import java.io.IOException;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+
+class SseEmitterSenderTest {
+
+ @Test
+ void sendIOExceptionShouldNotTriggerAsyncErrorDispatch() throws Exception {
+ SseEmitter emitter = mock(SseEmitter.class);
+ doThrow(new IOException("client disconnected"))
+ .when(emitter).send(any(SseEmitter.SseEventBuilder.class));
+ SseEmitterSender sender = new SseEmitterSender(emitter);
+
+ sender.sendEvent("message", "content");
+ sender.sendEvent("message", "ignored");
+
+ verify(emitter).send(any(SseEmitter.SseEventBuilder.class));
+ verify(emitter, never()).complete();
+ verify(emitter, never()).completeWithError(any());
+ }
+}
diff --git a/rag/src/main/java/com/nageoffer/ai/ragent/rag/dto/StreamErrorPayload.java b/rag/src/main/java/com/nageoffer/ai/ragent/rag/dto/StreamErrorPayload.java
new file mode 100644
index 000000000..63ad279b6
--- /dev/null
+++ b/rag/src/main/java/com/nageoffer/ai/ragent/rag/dto/StreamErrorPayload.java
@@ -0,0 +1,26 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.nageoffer.ai.ragent.rag.dto;
+
+/**
+ * 流式对话错误事件载荷
+ *
+ * @param error 面向用户的错误提示
+ */
+public record StreamErrorPayload(String error) {
+}
diff --git a/rag/src/main/java/com/nageoffer/ai/ragent/rag/enums/SSEEventType.java b/rag/src/main/java/com/nageoffer/ai/ragent/rag/enums/SSEEventType.java
index 87f748f50..d4959e859 100644
--- a/rag/src/main/java/com/nageoffer/ai/ragent/rag/enums/SSEEventType.java
+++ b/rag/src/main/java/com/nageoffer/ai/ragent/rag/enums/SSEEventType.java
@@ -50,6 +50,11 @@ public enum SSEEventType {
*/
CANCEL("cancel"),
+ /**
+ * 错误事件
+ */
+ ERROR("error"),
+
/**
* 拒绝事件
*/
diff --git a/rag/src/main/java/com/nageoffer/ai/ragent/rag/service/handler/StreamChatEventHandler.java b/rag/src/main/java/com/nageoffer/ai/ragent/rag/service/handler/StreamChatEventHandler.java
index ee3cd17df..764434e86 100644
--- a/rag/src/main/java/com/nageoffer/ai/ragent/rag/service/handler/StreamChatEventHandler.java
+++ b/rag/src/main/java/com/nageoffer/ai/ragent/rag/service/handler/StreamChatEventHandler.java
@@ -23,6 +23,7 @@
import com.nageoffer.ai.ragent.rag.dto.CompletionPayload;
import com.nageoffer.ai.ragent.rag.dto.MessageDelta;
import com.nageoffer.ai.ragent.rag.dto.MetaPayload;
+import com.nageoffer.ai.ragent.rag.dto.StreamErrorPayload;
import com.nageoffer.ai.ragent.rag.enums.SSEEventType;
import com.nageoffer.ai.ragent.framework.context.UserContext;
import com.nageoffer.ai.ragent.framework.convention.ChatMessage;
@@ -44,6 +45,7 @@ public class StreamChatEventHandler implements StreamCallback {
private static final String TYPE_THINK = "think";
private static final String TYPE_RESPONSE = "response";
+ private static final String ERROR_MESSAGE = "生成失败，请稍后重试";
private final int messageChunkSize;
private final SseEmitterSender sender;
@@ -237,8 +239,10 @@ public void onError(Throwable t) {
if (taskManager.isCancelled(taskId)) {
return;
}
+ log.error("流式对话失败，conversationId：{}，taskId：{}", conversationId, taskId, t);
taskManager.unregister(taskId);
- sender.fail(t);
+ sender.sendEvent(SSEEventType.ERROR.value(), new StreamErrorPayload(ERROR_MESSAGE));
+ sender.complete();
}
private void sendChunked(String type, String content) {
diff --git a/rag/src/test/java/com/nageoffer/ai/ragent/rag/service/handler/StreamChatEventHandlerTest.java b/rag/src/test/java/com/nageoffer/ai/ragent/rag/service/handler/StreamChatEventHandlerTest.java
new file mode 100644
index 000000000..839e5fb1d
--- /dev/null
+++ b/rag/src/test/java/com/nageoffer/ai/ragent/rag/service/handler/StreamChatEventHandlerTest.java
@@ -0,0 +1,97 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.nageoffer.ai.ragent.rag.service.handler;
+
+import com.nageoffer.ai.ragent.framework.web.StreamTaskManager;
+import com.nageoffer.ai.ragent.infra.config.AIModelProperties;
+import com.nageoffer.ai.ragent.rag.core.memory.ConversationMemoryService;
+import com.nageoffer.ai.ragent.rag.service.ConversationGroupService;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.springframework.test.util.ReflectionTestUtils;
+import org.springframework.web.servlet.mvc.method.annotation.ResponseBodyEmitter;
+import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
+
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+@ExtendWith(MockitoExtension.class)
+class StreamChatEventHandlerTest {
+
+ private static final String PUBLIC_ERROR_MESSAGE = "生成失败，请稍后重试";
+
+ @Mock
+ private SseEmitter emitter;
+
+ @Mock
+ private AIModelProperties modelProperties;
+
+ @Mock
+ private ConversationMemoryService memoryService;
+
+ @Mock
+ private ConversationGroupService conversationGroupService;
+
+ @Mock
+ private StreamTaskManager taskManager;
+
+ @Test
+ void businessFailureShouldSendSafeErrorEventAndCompleteNormally() throws Exception {
+ StreamChatEventHandler handler = new StreamChatEventHandler(StreamChatHandlerParams.builder()
+ .emitter(emitter)
+ .conversationId("conversation-1")
+ .taskId("task-1")
+ .modelProperties(modelProperties)
+ .memoryService(memoryService)
+ .conversationGroupService(conversationGroupService)
+ .taskManager(taskManager)
+ .build());
+
+ handler.onError(new IllegalStateException("database password leaked"));
+
+ ArgumentCaptor