feat(chat): cut over to single SSE endpoint

2026-07-22 10:01:12 +08:00
parent f8809cb7dd
commit bc36248cd8
36 changed files with 2187 additions and 945 deletions
@@ -0,0 +1,91 @@
+package com.superbiz.agent.config;
+
+import com.superbiz.agent.harness.core.DiagnosisHarnessCore;
+import com.superbiz.agent.harness.application.persistence.ChatRunStore;
+import com.superbiz.agent.harness.tool.contract.MysqlToolRequest;
+import com.superbiz.agent.harness.tool.mysql.MysqlSqlValidator;
+import com.superbiz.agent.harness.tool.mysql.MysqlSecurityException;
+import com.superbiz.agent.tool.LookupKnowledgeTool;
+import com.superbiz.agent.tool.RetrievedDocTracker;
+import com.superbiz.agent.service.KnowledgeContextPacker;
+import com.superbiz.agent.service.KnowledgeDocumentRetriever;
+import com.superbiz.agent.service.KnowledgeEvidencePostProcessor;
+import com.superbiz.agent.service.KnowledgeQueryTransformer;
+import com.superbiz.agent.service.LookupResultAssembler;
+import com.superbiz.agent.service.ToolInvocationRecorder;
+import com.superbiz.agent.repository.AgentStepRepository;
+import org.springframework.ai.chat.model.ChatModel;
+import org.springframework.boot.test.context.runner.ApplicationContextRunner;
+import org.springframework.data.redis.core.RedisTemplate;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.junit.jupiter.api.Test;
+
+import java.util.Map;
+import java.util.concurrent.ThreadPoolExecutor;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+
+class HarnessChatConfigurationTest {
+
+    @Test
+    void defaultsArePositiveAndExecutorsAreBounded() {
+        ChatHarnessProperties properties = new ChatHarnessProperties();
+        properties.validate();
+        HarnessChatConfiguration configuration = new HarnessChatConfiguration();
+
+        ThreadPoolExecutor worker = configuration.chatWorkerExecutor(properties);
+        ThreadPoolExecutor model = configuration.harnessModelExecutor(properties);
+        try {
+            assertTrue(worker.getQueue().remainingCapacity() > 0);
+            assertTrue(model.getQueue().remainingCapacity() > 0);
+            assertEquals(properties.getWorkerMaxPoolSize(), worker.getMaximumPoolSize());
+            assertEquals(properties.getModelMaxPoolSize(), model.getMaximumPoolSize());
+        } finally {
+            worker.shutdownNow();
+            model.shutdownNow();
+        }
+    }
+
+    @Test
+    void emptyMysqlConfigurationFailsClosedWithoutApplicationDatasource() {
+        MysqlToolProperties properties = new MysqlToolProperties();
+        HarnessChatConfiguration configuration = new HarnessChatConfiguration();
+        assertTrue(configuration.harnessMysqlDataSources(properties).isEmpty());
+        Map<String, com.superbiz.agent.harness.tool.mysql.MysqlDataSourceDefinition> definitions =
+                configuration.harnessMysqlDefinitions(properties);
+        MysqlSqlValidator validator = configuration.mysqlSqlValidator(definitions);
+
+        assertThrows(MysqlSecurityException.class, () -> validator.validate(
+                new MysqlToolRequest("application", "SELECT id FROM users", java.util.List.of())));
+    }
+
+    @Test
+    void harnessConfigurationBuildsOneApplicationGraphWithControlledDependencies() {
+        new ApplicationContextRunner()
+                .withUserConfiguration(HarnessChatConfiguration.class)
+                .withBean(ObjectMapper.class, ObjectMapper::new)
+                .withBean(ChatModel.class, () -> mock(ChatModel.class))
+                .withBean(RedisTemplate.class, () -> mock(RedisTemplate.class))
+                .withBean(LookupKnowledgeTool.class, () -> mock(LookupKnowledgeTool.class))
+                .withBean(KnowledgeQueryTransformer.class, () -> mock(KnowledgeQueryTransformer.class))
+                .withBean(KnowledgeDocumentRetriever.class, () -> mock(KnowledgeDocumentRetriever.class))
+                .withBean(KnowledgeEvidencePostProcessor.class, () -> mock(KnowledgeEvidencePostProcessor.class))
+                .withBean(KnowledgeContextPacker.class, () -> mock(KnowledgeContextPacker.class))
+                .withBean(LookupResultAssembler.class, () -> mock(LookupResultAssembler.class))
+                .withBean(ToolInvocationRecorder.class, () -> mock(ToolInvocationRecorder.class))
+                .withBean(RetrievedDocTracker.class, () -> mock(RetrievedDocTracker.class))
+                .withBean(AgentStepRepository.class, () -> mock(AgentStepRepository.class))
+                .withBean(ChatRunStore.class, () -> mock(ChatRunStore.class))
+                .withBean(MysqlToolProperties.class, MysqlToolProperties::new)
+                .run(context -> {
+                    assertTrue(context.isRunning());
+                    assertEquals(1, context.getBeansOfType(
+                            com.superbiz.agent.harness.application.ChatApplicationUseCase.class).size());
+                    assertTrue(context.containsBean("harnessModelExecutor"));
+                    assertTrue(context.containsBean("chatWorkerExecutor"));
+                });
+    }
+}
@@ -0,0 +1,17 @@
+package com.superbiz.agent.controller;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+class AiOpsControllerTest {
+
+    @Test
+    void metadataMessageCarriesSessionAndRunId() {
+        AiOpsController.SseMessage message = AiOpsController.SseMessage.metadata("session-1", "run-1");
+
+        assertEquals("metadata", message.getType());
+        assertEquals("session-1", message.getSessionId());
+        assertEquals("run-1", message.getRunId());
+    }
+}
@@ -1,12 +1,19 @@
 package com.superbiz.agent.controller;

-import com.superbiz.agent.service.ChatService;
+import com.superbiz.agent.config.ChatHarnessProperties;
+import com.superbiz.agent.harness.application.ChatApplicationUseCase;
 import org.junit.jupiter.api.Test;
 import org.springframework.http.ResponseEntity;
-import org.springframework.test.util.ReflectionTestUtils;
+
+import java.lang.reflect.Field;
+import java.util.Arrays;
+import java.util.Set;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;

 import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verifyNoInteractions;

@@ -14,28 +21,59 @@ class ChatControllerTest {

    @Test
    void blankChatRequestReturnsErrorBeforeCreatingRun() {
-        ChatController controller = new ChatController();
-        ChatService chatService = mock(ChatService.class);
-        ReflectionTestUtils.setField(controller, "chatService", chatService);
+        ThreadPoolExecutor executor = new ThreadPoolExecutor(
+                1, 1, 1, TimeUnit.SECONDS, new ArrayBlockingQueue<>(1));
+        ChatApplicationUseCase application = mock(ChatApplicationUseCase.class);
+        ChatController controller = new ChatController(
+                application, executor, new ChatHarnessProperties());

        ChatController.ChatRequest request = new ChatController.ChatRequest();
        request.setId("invalid-chat-session");
        request.setQuestion("   ");

-        ResponseEntity<ChatController.ApiResponse<ChatController.ChatResponse>> response = controller.chat(request);
+        ResponseEntity<?> response = controller.chat(request);

-        ChatController.ChatResponse body = response.getBody().getData();
-        assertFalse(body.isSuccess());
-        assertEquals("问题内容不能为空", body.getErrorMessage());
-        verifyNoInteractions(chatService);
+        assertEquals(400, response.getStatusCode().value());
+        verifyNoInteractions(application);
+        executor.shutdownNow();
    }

    @Test
-    void aiOpsMetadataMessageCarriesSessionAndRunId() {
-        ChatController.SseMessage message = ChatController.SseMessage.metadata("session-1", "run-1");
+    void controllerOwnsOnlyChatProtocolAndApplicationDependencies() {
+        Set<Class<?>> dependencyTypes = Arrays.stream(ChatController.class.getDeclaredFields())
+                .map(Field::getType)
+                .collect(java.util.stream.Collectors.toSet());

-        assertEquals("metadata", message.getType());
-        assertEquals("session-1", message.getSessionId());
-        assertEquals("run-1", message.getRunId());
+        assertEquals(Set.of(ChatApplicationUseCase.class, ThreadPoolExecutor.class, ChatHarnessProperties.class),
+                dependencyTypes);
+    }
+
+    @Test
+    void saturatedWorkerReturnsUnavailableBeforeCreatingRun() throws Exception {
+        ThreadPoolExecutor executor = new ThreadPoolExecutor(
+                1, 1, 1, TimeUnit.SECONDS, new ArrayBlockingQueue<>(1));
+        CountDownLatch release = new CountDownLatch(1);
+        executor.execute(() -> await(release));
+        executor.execute(() -> await(release));
+        ChatApplicationUseCase application = mock(ChatApplicationUseCase.class);
+        ChatController controller = new ChatController(
+                application, executor, new ChatHarnessProperties());
+        ChatController.ChatRequest request = new ChatController.ChatRequest();
+        request.setQuestion("hello");
+
+        ResponseEntity<?> response = controller.chat(request);
+
+        assertEquals(503, response.getStatusCode().value());
+        verifyNoInteractions(application);
+        release.countDown();
+        executor.shutdownNow();
+    }
+
+    private static void await(CountDownLatch release) {
+        try {
+            release.await(2, TimeUnit.SECONDS);
+        } catch (InterruptedException exception) {
+            Thread.currentThread().interrupt();
+        }
    }
 }
@@ -0,0 +1,36 @@
+package com.superbiz.agent.controller;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class FrontendChatSseContractTest {
+
+    private static final Path STATIC_ROOT = Path.of("src", "main", "resources", "static");
+
+    @Test
+    void bundledChatUsesOnlyTheNewSseConsumer() throws IOException {
+        String app = Files.readString(STATIC_ROOT.resolve("app.js"), StandardCharsets.UTF_8);
+        String html = Files.readString(STATIC_ROOT.resolve("index.html"), StandardCharsets.UTF_8);
+
+        assertTrue(app.contains("sendChatMessage"));
+        assertTrue(app.contains("/chat`"));
+        assertTrue(app.contains("case 'metadata'"));
+        assertTrue(app.contains("case 'status'"));
+        assertTrue(app.contains("case 'content'"));
+        assertTrue(app.contains("case 'failure'"));
+        assertTrue(app.contains("case 'done'"));
+        assertFalse(app.contains("/chat_stream"));
+        assertFalse(app.contains("sendQuickMessage"));
+        assertFalse(app.contains("sendStreamMessage"));
+        assertFalse(app.contains("currentMode"));
+        assertFalse(html.contains("modeSelectorBtn"));
+        assertFalse(html.contains("modeDropdown"));
+    }
+}
@@ -0,0 +1,143 @@
+package com.superbiz.agent.controller.sse;
+
+import com.superbiz.agent.harness.application.ChatApplicationResult;
+import com.superbiz.agent.harness.application.ChatApplicationException;
+import com.superbiz.agent.harness.application.ChatApplicationStatus;
+import com.superbiz.agent.harness.application.ChatContentType;
+import com.superbiz.agent.harness.application.ChatFailureCode;
+import com.superbiz.agent.harness.application.ChatRunControl;
+import com.superbiz.agent.harness.application.SystemChatContent;
+import com.superbiz.agent.harness.contract.IntentType;
+import com.superbiz.agent.harness.contract.ReleaseOutcome;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.util.ArrayList;
+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.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class ChatSseSessionTest {
+
+    @Test
+    void successEmitsStrictNamedEventSequenceWithTypedContent() {
+        RecordingSink sink = new RecordingSink();
+        ChatSseSession session = new ChatSseSession(sink);
+        RecordingRunControl control = new RecordingRunControl();
+
+        session.onStarted(control);
+        session.onStatus(ChatApplicationStatus.ROUTING);
+        session.complete(new ChatApplicationResult(
+                "session-1", "run-1", IntentType.SYSTEM_CHAT, ReleaseOutcome.SUCCESS,
+                ChatContentType.SYSTEM_CHAT, new SystemChatContent("hello")));
+
+        assertEquals(List.of("metadata", "status", "content", "done"), sink.names());
+        assertEquals(new ChatSseEvent.Metadata("session-1", "run-1"), sink.events.get(0).payload());
+        assertEquals(new ChatSseEvent.Content(
+                ChatContentType.SYSTEM_CHAT, new SystemChatContent("hello")),
+                sink.events.get(2).payload());
+        assertEquals(new ChatSseEvent.Done(ReleaseOutcome.SUCCESS), sink.events.get(3).payload());
+        assertEquals(1, sink.completions);
+        assertFalse(control.cancelled);
+    }
+
+    @Test
+    void technicalFailureEmitsSafeFailureAndFailedDoneWithoutContent() {
+        RecordingSink sink = new RecordingSink();
+        ChatSseSession session = new ChatSseSession(sink);
+        session.onStarted(new RecordingRunControl());
+
+        session.fail(new ChatApplicationException(
+                ChatFailureCode.ROUTING_UNAVAILABLE,
+                "当前暂时无法处理该请求，请稍后重试",
+                new IllegalStateException("provider-secret-detail")));
+
+        assertEquals(List.of("metadata", "failure", "done"), sink.names());
+        assertEquals(new ChatSseEvent.Failure(
+                ChatFailureCode.ROUTING_UNAVAILABLE,
+                "当前暂时无法处理该请求，请稍后重试"), sink.events.get(1).payload());
+        assertEquals(new ChatSseEvent.Done(ReleaseOutcome.FAILED), sink.events.get(2).payload());
+        assertFalse(sink.events.toString().contains("provider-secret-detail"));
+    }
+
+    @Test
+    void disconnectBeforeRunStartCancelsControlWhenPublishedWithoutSendingEvents() {
+        RecordingSink sink = new RecordingSink();
+        ChatSseSession session = new ChatSseSession(sink);
+        RecordingRunControl control = new RecordingRunControl();
+
+        assertFalse(session.disconnect());
+        session.onStarted(control);
+
+        assertTrue(control.cancelled);
+        assertTrue(sink.events.isEmpty());
+        assertEquals(0, sink.completions);
+    }
+
+    @Test
+    void statusSendFailureCancelsThePublishedRunAndBlocksLateTerminalEvents() {
+        RecordingSink sink = new RecordingSink();
+        sink.failOn = "status";
+        ChatSseSession session = new ChatSseSession(sink);
+        RecordingRunControl control = new RecordingRunControl();
+        session.onStarted(control);
+
+        assertThrows(ChatSseDisconnectedException.class,
+                () -> session.onStatus(ChatApplicationStatus.ROUTING));
+
+        assertTrue(control.cancelled);
+        assertFalse(session.complete(new ChatApplicationResult(
+                "session-1", "run-1", IntentType.SYSTEM_CHAT, ReleaseOutcome.SUCCESS,
+                ChatContentType.SYSTEM_CHAT, new SystemChatContent("late"))));
+        assertEquals(List.of("metadata"), sink.names());
+    }
+
+    private static final class RecordingSink implements ChatSseSink {
+        private final List<Event> events = new ArrayList<>();
+        private int completions;
+        private String failOn;
+
+        @Override
+        public void send(String name, Object payload) throws IOException {
+            if (name.equals(failOn)) {
+                throw new IOException("closed");
+            }
+            events.add(new Event(name, payload));
+        }
+
+        @Override
+        public void complete() {
+            completions++;
+        }
+
+        private List<String> names() {
+            return events.stream().map(Event::name).toList();
+        }
+    }
+
+    private static final class RecordingRunControl implements ChatRunControl {
+        private boolean cancelled;
+
+        @Override
+        public String sessionId() {
+            return "session-1";
+        }
+
+        @Override
+        public String runId() {
+            return "run-1";
+        }
+
+        @Override
+        public boolean cancelClientDisconnect() {
+            cancelled = true;
+            return true;
+        }
+    }
+
+    private record Event(String name, Object payload) {
+    }
+}