实现Debug EML实时Trace调试链路

2026-07-12 14:59:03 +08:00
parent eee37315a0
commit 8d8fae670a
20 changed files with 1571 additions and 76 deletions
--- a/server/src/test/java/cn/nianxx/thhotel/integrations/ai/superagent/service/impl/SuperAgentOpenApiClientImplTest.java
+++ b/server/src/test/java/cn/nianxx/thhotel/integrations/ai/superagent/service/impl/SuperAgentOpenApiClientImplTest.java
@@ -0,0 +1,88 @@
+package cn.nianxx.thhotel.integrations.ai.superagent.service.impl;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import cn.nianxx.thhotel.integrations.ai.superagent.common.request.SuperAgentMailDebugRequest;
+import cn.nianxx.thhotel.integrations.ai.superagent.common.result.SuperAgentOpenApiResult;
+import cn.nianxx.thhotel.integrations.ai.superagent.common.result.SuperAgentOpenApiTraceEvent;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.sun.net.httpserver.HttpServer;
+import java.io.OutputStream;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicReference;
+import org.junit.jupiter.api.Test;
+
+class SuperAgentOpenApiClientImplTest {
+
+    @Test
+    void shouldRequestStreamingMessagesWithIncludeTraceAndEmitPublicTraceEvents() throws Exception {
+        HttpServer server = HttpServer.create(new InetSocketAddress(InetAddress.getLoopbackAddress(), 0), 0);
+        AtomicReference<String> streamQuery = new AtomicReference<>();
+        server.createContext("/api/open/agent-sessions", exchange -> {
+            byte[] response = "{\"session_id\":\"session-http-001\"}".getBytes(StandardCharsets.UTF_8);
+            exchange.getResponseHeaders().add("Content-Type", "application/json");
+            exchange.sendResponseHeaders(200, response.length);
+            try (OutputStream outputStream = exchange.getResponseBody()) {
+                outputStream.write(response);
+            }
+        });
+        server.createContext("/api/open/agent-sessions/session-http-001/messages/stream", exchange -> {
+            streamQuery.set(exchange.getRequestURI().getRawQuery());
+            byte[] response = """
+                    event: metadata
+                    data: {"run_id":"run-http-001","resolved_profile_id":"profile-http","resolved_profile_version_id":"version-http"}
+
+                    event: trace
+                    data: {"event":"reasoning.summary","run_id":"run-http-001","text":"正在分析 Debug 邮件。","ts":"2026-07-11T10:00:01Z"}
+
+                    event: values
+                    data: {"messages":[{"type":"ai","content":"{\\"route_code\\":\\"S10\\"}","response_metadata":{"finish_reason":"stop","model_name":"debug-model"},"usage_metadata":{"input_tokens":9,"output_tokens":4,"total_tokens":13}}]}
+
+                    event: end
+                    data: {}
+
+                    """.getBytes(StandardCharsets.UTF_8);
+            exchange.getResponseHeaders().add("Content-Type", "text/event-stream");
+            exchange.sendResponseHeaders(200, response.length);
+            try (OutputStream outputStream = exchange.getResponseBody()) {
+                outputStream.write(response);
+            }
+        });
+        server.start();
+        try {
+            SuperAgentOpenApiProperties properties = new SuperAgentOpenApiProperties();
+            properties.setEnabled(true);
+            properties.setBaseUrl("http://127.0.0.1:" + server.getAddress().getPort());
+            properties.setApiKey("df_open_test");
+            properties.setExternalSubjectId("debug-subject");
+            properties.setConnectTimeout(Duration.ofSeconds(5));
+            properties.setReadTimeout(Duration.ofSeconds(5));
+            SuperAgentOpenApiClientImpl client = new SuperAgentOpenApiClientImpl(
+                    properties,
+                    new SuperAgentOpenApiSseParser(new ObjectMapper()),
+                    new ObjectMapper());
+            List<SuperAgentOpenApiTraceEvent> traceEvents = new ArrayList<>();
+
+            SuperAgentOpenApiResult result = client.invokeMailDebug(new SuperAgentMailDebugRequest(
+                    "debug message",
+                    "debug-idempotency",
+                    Map.of("source", "unit-test")), traceEvents::add);
+
+            assertThat(streamQuery.get()).isEqualTo("include_trace=true");
+            assertThat(result.sessionId()).isEqualTo("session-http-001");
+            assertThat(result.runId()).isEqualTo("run-http-001");
+            assertThat(result.traceEvents()).hasSize(1);
+            assertThat(traceEvents).hasSize(1);
+            assertThat(traceEvents.get(0).event()).isEqualTo("reasoning.summary");
+            assertThat(traceEvents.get(0).text()).isEqualTo("正在分析 Debug 邮件。");
+        } finally {
+            server.stop(0);
+        }
+    }
+}
--- a/server/src/test/java/cn/nianxx/thhotel/integrations/ai/superagent/service/impl/SuperAgentOpenApiSseParserTest.java
+++ b/server/src/test/java/cn/nianxx/thhotel/integrations/ai/superagent/service/impl/SuperAgentOpenApiSseParserTest.java
@@ -4,6 +4,9 @@ import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;

 import cn.nianxx.thhotel.integrations.ai.superagent.common.result.SuperAgentOpenApiResult;
+import cn.nianxx.thhotel.integrations.ai.superagent.common.result.SuperAgentOpenApiTraceEvent;
+import java.util.ArrayList;
+import java.util.List;
 import org.junit.jupiter.api.Test;

 class SuperAgentOpenApiSseParserTest {
@@ -164,4 +167,41 @@ class SuperAgentOpenApiSseParserTest {

        assertThat(result.rawAnswer()).isEqualTo("{\"ai_task_results\":[{\"task_type\":\"Cancel Booking\"}]}");
    }
+
+    @Test
+    void shouldCollectPublicTraceEventsAndKeepFinalAnswer() {
+        String sse = """
+                event: metadata
+                data: {"run_id":"run-debug-trace","resolved_profile_id":"profile-debug"}
+
+                event: trace
+                data: {"event":"reasoning.summary","run_id":"run-debug-trace","text":"正在分析邮件和附件。","ts":"2026-07-11T10:00:01Z"}
+
+                event: trace
+                data: {"event":"tool.call.started","run_id":"run-debug-trace","message_id":"ai-1","tool_calls":[{"tool_call_id":"tool-1","name":"th_hotel_query_case_context","input":{"summary":"{\\"group_code\\":\\"G001\\"}"}}],"ts":"2026-07-11T10:00:02Z"}
+
+                event: trace
+                data: {"event":"tool.call.completed","run_id":"run-debug-trace","tool_call_id":"tool-1","name":"th_hotel_query_case_context","output":{"summary":"matched 1 case"},"ts":"2026-07-11T10:00:03Z"}
+
+                event: values
+                data: {"messages":[{"type":"ai","content":"{\\"route_code\\":\\"S10\\"}","response_metadata":{"finish_reason":"stop","model_name":"debug-model"},"usage_metadata":{"input_tokens":31,"output_tokens":5,"total_tokens":36}}]}
+
+                event: end
+                data: {}
+
+                """;
+        List<SuperAgentOpenApiTraceEvent> emittedTraceEvents = new ArrayList<>();
+
+        SuperAgentOpenApiResult result = parser.parse("session-debug-trace", sse, emittedTraceEvents::add);
+
+        assertThat(result.rawAnswer()).isEqualTo("{\"route_code\":\"S10\"}");
+        assertThat(result.eventTypes()).containsExactly("metadata", "trace", "values", "end");
+        assertThat(result.traceEvents()).hasSize(3);
+        assertThat(emittedTraceEvents).hasSize(3);
+        assertThat(result.traceEvents().get(0).event()).isEqualTo("reasoning.summary");
+        assertThat(result.traceEvents().get(0).text()).isEqualTo("正在分析邮件和附件。");
+        assertThat(result.traceEvents().get(1).toolName()).isEqualTo("th_hotel_query_case_context");
+        assertThat(result.traceEvents().get(1).inputSummary()).isEqualTo("{\"group_code\":\"G001\"}");
+        assertThat(result.traceEvents().get(2).outputSummary()).isEqualTo("matched 1 case");
+    }
 }
--- a/server/src/test/java/cn/nianxx/thhotel/platform/debug/control/DebugEmlSuperAgentControllerTest.java
+++ b/server/src/test/java/cn/nianxx/thhotel/platform/debug/control/DebugEmlSuperAgentControllerTest.java
@@ -7,20 +7,27 @@ import static org.hamcrest.Matchers.not;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.reset;
 import static org.mockito.Mockito.when;
+import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.asyncDispatch;
 import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.multipart;
 import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.content;
 import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.jsonPath;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.request;
 import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;

 import cn.nianxx.thhotel.ThHotelApplication;
 import cn.nianxx.thhotel.integrations.ai.superagent.common.result.SuperAgentOpenApiResult;
+import cn.nianxx.thhotel.integrations.ai.superagent.common.result.SuperAgentOpenApiTraceEvent;
 import cn.nianxx.thhotel.integrations.ai.superagent.service.SuperAgentOpenApiClient;
 import cn.nianxx.thhotel.integrations.ai.superagent.service.impl.SuperAgentOpenApiException;
 import cn.nianxx.thhotel.integrations.storage.aliyunoss.common.request.ObjectStoragePutRequest;
 import cn.nianxx.thhotel.integrations.storage.aliyunoss.common.result.ObjectStoragePutResult;
 import cn.nianxx.thhotel.integrations.storage.aliyunoss.service.ObjectStorageService;
+import cn.nianxx.thhotel.platform.debug.service.DebugEmlSuperAgentRunService;
+import java.io.IOException;
+import java.io.OutputStream;
 import java.nio.charset.StandardCharsets;
 import java.util.List;
+import java.util.function.Consumer;
 import java.util.concurrent.atomic.AtomicInteger;
 import org.junit.jupiter.api.Test;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -32,6 +39,7 @@ import org.springframework.jdbc.core.JdbcTemplate;
 import org.springframework.mock.web.MockMultipartFile;
 import org.springframework.test.context.ActiveProfiles;
 import org.springframework.test.web.servlet.MockMvc;
+import org.springframework.test.web.servlet.MvcResult;
 import org.springframework.web.client.RestClientResponseException;

@SpringBootTest(
@@ -56,6 +64,9 @@ class DebugEmlSuperAgentControllerTest {
    @Autowired
    private JdbcTemplate jdbcTemplate;

+    @Autowired
+    private DebugEmlSuperAgentRunService runService;
+
    @MockBean
    private ObjectStorageService objectStorageService;

@@ -71,6 +82,17 @@ class DebugEmlSuperAgentControllerTest {
                .andExpect(content().string(not(containsString("test-debug-upload-key"))));
    }

+    @Test
+    void shouldRejectStreamUploadWhenDebugKeyMissingBeforeStartingSse() throws Exception {
+        mockMvc.perform(multipart(ENDPOINT + "/stream")
+                        .file(emlFile())
+                        .param("hotel_id", "HOTEL-TEST"))
+                .andExpect(status().isUnauthorized())
+                .andExpect(content().contentTypeCompatibleWith(MediaType.APPLICATION_JSON))
+                .andExpect(jsonPath("$.error_code").value("DEBUG_UPLOAD_KEY_INVALID"))
+                .andExpect(content().string(not(containsString("test-debug-upload-key"))));
+    }
+
    @Test
    void shouldUploadEmlToOssCaptureSourceMessageAndReturnSuperAgentResult() throws Exception {
        when(objectStorageService.putObject(any())).thenAnswer(invocation -> {
@@ -158,6 +180,139 @@ class DebugEmlSuperAgentControllerTest {
        org.assertj.core.api.Assertions.assertThat(debugRunCount).isEqualTo(1L);
    }

+    @Test
+    void shouldStreamDebugStagesSuperAgentTraceAndFinalResult() throws Exception {
+        when(objectStorageService.putObject(any())).thenAnswer(invocation -> {
+            ObjectStoragePutRequest request = invocation.getArgument(0);
+            return new ObjectStoragePutResult(
+                    request.objectKey(),
+                    "https://oss.example.test/" + request.objectKey(),
+                    request.contentType(),
+                    request.sizeBytes());
+        });
+        when(superAgentOpenApiClient.invokeMailDebug(any(), any())).thenAnswer(invocation -> {
+            @SuppressWarnings("unchecked")
+            Consumer<SuperAgentOpenApiTraceEvent> traceConsumer = invocation.getArgument(1);
+            traceConsumer.accept(new SuperAgentOpenApiTraceEvent(
+                    "reasoning.summary",
+                    "run-stream-001",
+                    null,
+                    null,
+                    null,
+                    "正在分析 Debug 邮件。",
+                    null,
+                    null,
+                    null,
+                    "2026-07-11T10:00:01Z"));
+            traceConsumer.accept(new SuperAgentOpenApiTraceEvent(
+                    "tool.call.started",
+                    "run-stream-001",
+                    "ai-1",
+                    "tool-1",
+                    "th_hotel_query_case_context",
+                    null,
+                    "{\"group_code\":\"G001\"}",
+                    null,
+                    null,
+                    "2026-07-11T10:00:02Z"));
+            return new SuperAgentOpenApiResult(
+                    "session-stream-001",
+                    "run-stream-001",
+                    "profile-debug",
+                    "profile-version-debug",
+                    "debug-model",
+                    "{\"route_code\":\"S10\"}",
+                    11,
+                    7,
+                    18,
+                    List.of("metadata", "trace", "values", "end"),
+                    List.of());
+        });
+
+        MvcResult mvcResult = mockMvc.perform(multipart(ENDPOINT + "/stream")
+                        .file(emlFile())
+                        .param("hotel_id", "HOTEL-TEST")
+                        .param("run_label", "stream-debug-upload")
+                        .header("X-TH-Hotel-Debug-Upload-Key", "test-debug-upload-key"))
+                .andExpect(request().asyncStarted())
+                .andReturn();
+
+        mockMvc.perform(asyncDispatch(mvcResult))
+                .andExpect(status().isOk())
+                .andExpect(content().contentTypeCompatibleWith(MediaType.TEXT_EVENT_STREAM))
+                .andExpect(content().string(containsString("event: debug_stage")))
+                .andExpect(content().string(containsString("\"status\":\"PARSING_EML\"")))
+                .andExpect(content().string(containsString("\"status\":\"CALLING_SUPERAGENT\"")))
+                .andExpect(content().string(containsString("event: superagent_trace")))
+                .andExpect(content().string(containsString("\"event\":\"reasoning.summary\"")))
+                .andExpect(content().string(containsString("\"event\":\"tool.call.started\"")))
+                .andExpect(content().string(containsString("\"tool_name\":\"th_hotel_query_case_context\"")))
+                .andExpect(content().string(containsString("event: superagent_result")))
+                .andExpect(content().string(containsString("\"superagent_run_id\":\"run-stream-001\"")))
+                .andExpect(content().string(containsString("\"superagent_parsed_json\":{\"route_code\":\"S10\"}")))
+                .andExpect(content().string(containsString("event: done")))
+                .andExpect(content().string(not(containsString("test-debug-upload-key"))));
+    }
+
+    @Test
+    void shouldKeepBusinessRunStatusWhenSseClientDisconnects() {
+        when(objectStorageService.putObject(any())).thenAnswer(invocation -> {
+            ObjectStoragePutRequest request = invocation.getArgument(0);
+            return new ObjectStoragePutResult(
+                    request.objectKey(),
+                    "https://oss.example.test/" + request.objectKey(),
+                    request.contentType(),
+                    request.sizeBytes());
+        });
+        when(superAgentOpenApiClient.invokeMailDebug(any(), any())).thenAnswer(invocation -> {
+            @SuppressWarnings("unchecked")
+            Consumer<SuperAgentOpenApiTraceEvent> traceConsumer = invocation.getArgument(1);
+            traceConsumer.accept(new SuperAgentOpenApiTraceEvent(
+                    "reasoning.summary",
+                    "run-stream-disconnect",
+                    null,
+                    null,
+                    null,
+                    "正在分析 Debug 邮件。",
+                    null,
+                    null,
+                    null,
+                    "2026-07-11T10:00:01Z"));
+            return new SuperAgentOpenApiResult(
+                    "session-stream-disconnect",
+                    "run-stream-disconnect",
+                    "profile-debug",
+                    "profile-version-debug",
+                    "debug-model",
+                    "{\"route_code\":\"S10\"}",
+                    11,
+                    7,
+                    18,
+                    List.of("metadata", "trace", "values", "end"),
+                    List.of());
+        });
+
+        org.assertj.core.api.Assertions.assertThatCode(() -> runService.uploadAndRunStream(
+                        "test-debug-upload-key",
+                        emlFile(),
+                        "HOTEL-TEST",
+                        "stream-client-disconnected",
+                        new FailingOutputStream()))
+                .doesNotThrowAnyException();
+
+        Long debugRunCount = jdbcTemplate.queryForObject("""
+                SELECT COUNT(*)
+                FROM platform_debug_eml_superagent_run
+                WHERE hotel_id = 'HOTEL-TEST'
+                  AND run_status = 'SUPERAGENT_SUCCEEDED'
+                  AND superagent_session_id = 'session-stream-disconnect'
+                  AND superagent_run_id = 'run-stream-disconnect'
+                  AND run_label = 'stream-client-disconnected'
+                  AND safe_error_summary IS NULL
+                """, Long.class);
+        org.assertj.core.api.Assertions.assertThat(debugRunCount).isEqualTo(1L);
+    }
+
    @Test
    void shouldRecordPhaseBeforeUploadingOriginalEmlToOss() throws Exception {
        AtomicInteger uploadIndex = new AtomicInteger();
@@ -408,6 +563,19 @@ class DebugEmlSuperAgentControllerTest {
                unsafeHtmlEmlBytes());
    }

+    private static final class FailingOutputStream extends OutputStream {
+
+        @Override
+        public void write(int b) throws IOException {
+            throw new IOException("client disconnected");
+        }
+
+        @Override
+        public void write(byte[] b, int off, int len) throws IOException {
+            throw new IOException("client disconnected");
+        }
+    }
+
    private byte[] emlBytes() {
        return """
                From: Guest <guest@example.test>