From 71191a10b08c20d2c922f593cf6536ad1ca17eba Mon Sep 17 00:00:00 2001 From: andy Date: Wed, 22 Jul 2026 14:44:21 +0700 Subject: [PATCH] =?UTF-8?q?=E5=A2=9E=E5=BC=BASuperAgent=E5=A4=B1=E8=B4=A5?= =?UTF-8?q?=E5=AE=89=E5=85=A8=E8=AF=8A=E6=96=AD=E6=97=A5=E5=BF=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../impl/SuperAgentOpenApiClientImpl.java | 232 +++++++++++++++--- .../impl/SuperAgentOpenApiSseParser.java | 21 ++ .../impl/SuperAgentOpenApiClientImplTest.java | 152 ++++++++++++ 3 files changed, 373 insertions(+), 32 deletions(-) diff --git a/server/src/main/java/cn/nianxx/thhotel/integrations/ai/superagent/service/impl/SuperAgentOpenApiClientImpl.java b/server/src/main/java/cn/nianxx/thhotel/integrations/ai/superagent/service/impl/SuperAgentOpenApiClientImpl.java index a78e7d8..61e1c7a 100644 --- a/server/src/main/java/cn/nianxx/thhotel/integrations/ai/superagent/service/impl/SuperAgentOpenApiClientImpl.java +++ b/server/src/main/java/cn/nianxx/thhotel/integrations/ai/superagent/service/impl/SuperAgentOpenApiClientImpl.java @@ -20,6 +20,9 @@ import java.util.Base64; import java.util.LinkedHashMap; import java.util.Map; import java.util.function.Consumer; +import java.util.regex.Pattern; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; @@ -29,10 +32,19 @@ import org.springframework.stereotype.Service; @Service public class SuperAgentOpenApiClientImpl implements SuperAgentOpenApiClient { + private static final Logger log = LoggerFactory.getLogger(SuperAgentOpenApiClientImpl.class); private static final String CSRF_HEADER_NAME = "X-CSRF-Token"; private static final String CSRF_COOKIE_NAME = "csrf_token"; private static final int CSRF_TOKEN_BYTES = 32; + private static final int SAFE_LOG_VALUE_MAX_LENGTH = 512; private static final SecureRandom CSRF_RANDOM = new SecureRandom(); + private static final Pattern JSON_SECRET_PATTERN = Pattern.compile( + "(?i)(\"(?:api[_-]?key|token|secret|password|cookie|authorization|csrf)\"\\s*:\\s*\")[^\"]*(\")"); + private static final Pattern HEADER_SECRET_PATTERN = Pattern.compile( + "(?i)((?:authorization|cookie|x-csrf-token|csrf[_-]?token)\\s*:\\s*)[^\\s,;]+"); + private static final Pattern TEXT_SECRET_PATTERN = Pattern.compile( + "(?i)((?:api[_-]?key|token|secret|password|cookie|authorization|csrf[_-]?token)\\s*[=:]\\s*)[^\\s,;}]+"); + private static final Pattern URL_PATTERN = Pattern.compile("(?i)\\b(?:https?|oss)://[^\\s,;\"'<>]+"); private final SuperAgentOpenApiProperties properties; private final SuperAgentOpenApiSseParser sseParser; @@ -104,29 +116,38 @@ public class SuperAgentOpenApiClientImpl implements SuperAgentOpenApiClient { */ private String createSession(HttpClient httpClient, URI baseUri, SuperAgentOpenApiMessageRequest request) throws Exception { - Map body = new LinkedHashMap<>(); - body.put("external_subject_id", textOrDefault(request.externalSubjectId(), properties.getExternalSubjectId())); - body.put("idempotency_key", request.idempotencyKey() + "-session"); - body.put("metadata", request.metadata()); - HttpRequest httpRequest = baseRequest(baseUri.resolve("/api/open/agent-sessions"), request.requestId()) - .header("Content-Type", "application/json") - .header("Accept", "application/json") - .POST(HttpRequest.BodyPublishers.ofString(objectMapper.writeValueAsString(body), StandardCharsets.UTF_8)) - .build(); - HttpResponse httpResponse = httpClient.send( - httpRequest, - HttpResponse.BodyHandlers.ofString(StandardCharsets.UTF_8)); - require2xx(httpResponse.statusCode(), httpResponse.body()); - String response = httpResponse.body(); - JsonNode json = objectMapper.readTree(response == null ? "{}" : response); - String sessionId = text(json, "session_id"); - if (sessionId == null) { - sessionId = text(json, "id"); + try { + Map body = new LinkedHashMap<>(); + body.put("external_subject_id", textOrDefault(request.externalSubjectId(), properties.getExternalSubjectId())); + body.put("idempotency_key", request.idempotencyKey() + "-session"); + body.put("metadata", request.metadata()); + HttpRequest httpRequest = baseRequest(baseUri.resolve("/api/open/agent-sessions"), request.requestId()) + .header("Content-Type", "application/json") + .header("Accept", "application/json") + .POST(HttpRequest.BodyPublishers.ofString(objectMapper.writeValueAsString(body), StandardCharsets.UTF_8)) + .build(); + HttpResponse httpResponse = httpClient.send( + httpRequest, + HttpResponse.BodyHandlers.ofString(StandardCharsets.UTF_8)); + require2xx(httpResponse.statusCode(), httpResponse.body()); + String response = httpResponse.body(); + JsonNode json = objectMapper.readTree(response == null ? "{}" : response); + String sessionId = text(json, "session_id"); + if (sessionId == null) { + sessionId = text(json, "id"); + } + if (sessionId == null) { + throw new SuperAgentOpenApiException("SuperAgent 创建 session 响应缺少 session_id。"); + } + return sessionId; + } catch (SuperAgentOpenApiException exception) { + logOpenApiFailure("create_session", request, null, null, exception); + throw exception; + } catch (Exception exception) { + restoreInterruptedFlag(exception); + logOpenApiFailure("create_session", request, null, null, exception); + throw exception; } - if (sessionId == null) { - throw new SuperAgentOpenApiException("SuperAgent 创建 session 响应缺少 session_id。"); - } - return sessionId; } /** @@ -148,22 +169,47 @@ public class SuperAgentOpenApiClientImpl implements SuperAgentOpenApiClient { .header("Accept", "text/event-stream") .POST(HttpRequest.BodyPublishers.ofString(objectMapper.writeValueAsString(body), StandardCharsets.UTF_8)) .build(); - HttpResponse httpResponse = httpClient.send( - httpRequest, - HttpResponse.BodyHandlers.ofInputStream()); - require2xxOrClose(httpResponse.statusCode(), httpResponse.body()); SuperAgentOpenApiSseParser.ParsedState state = sseParser.newState(sessionId); + HttpResponse httpResponse; + try { + httpResponse = httpClient.send( + httpRequest, + HttpResponse.BodyHandlers.ofInputStream()); + require2xxOrClose(httpResponse.statusCode(), httpResponse.body()); + } catch (SuperAgentOpenApiException exception) { + logOpenApiFailure("messages_stream", request, sessionId, state, exception); + throw exception; + } catch (Exception exception) { + restoreInterruptedFlag(exception); + logOpenApiFailure("messages_stream", request, sessionId, state, exception); + throw exception; + } applyRunLocation(baseUri, httpResponse, state); try { consumeResponse(httpResponse.body(), state, traceConsumer); } catch (SuperAgentOpenApiException exception) { if (!state.endSeen() && state.runUri() != null) { - recover(httpClient, request.requestId(), state, traceConsumer); + try { + recover(httpClient, request.requestId(), state, traceConsumer); + } catch (SuperAgentOpenApiException recoveryException) { + logOpenApiFailure("sse_recover", request, sessionId, state, recoveryException); + throw recoveryException; + } catch (Exception recoveryException) { + restoreInterruptedFlag(recoveryException); + logOpenApiFailure("sse_recover", request, sessionId, state, recoveryException); + throw recoveryException; + } } else { + logOpenApiFailure("sse_consume", request, sessionId, state, exception); throw exception; } } - return sseParser.requireSuccessfulResult(state); + try { + return sseParser.requireSuccessfulResult(state); + } catch (SuperAgentOpenApiException exception) { + logOpenApiFailure("sse_result", request, sessionId, state, exception); + throw exception; + } } /** @@ -231,10 +277,7 @@ public class SuperAgentOpenApiClientImpl implements SuperAgentOpenApiClient { */ private void require2xx(int statusCode, String responseBody) { if (statusCode < 200 || statusCode >= 300) { - String safeBody = responseBody == null ? "" : responseBody; - if (safeBody.length() > 512) { - safeBody = safeBody.substring(0, 512); - } + String safeBody = responseBody == null ? "" : safeLogText(responseBody); throw new SuperAgentOpenApiException("SuperAgent Open API HTTP 调用失败，status=" + statusCode + "，body=" + safeBody); } } @@ -363,6 +406,131 @@ public class SuperAgentOpenApiClientImpl implements SuperAgentOpenApiClient { consumeResponse(response.body(), state, traceConsumer); } + /** + * HTTP 调用被中断时恢复中断标记，避免上层线程池误判当前线程仍可继续执行。 + */ + private void restoreInterruptedFlag(Exception exception) { + if (exception instanceof InterruptedException) { + Thread.currentThread().interrupt(); + } + } + + /** + * 记录 SuperAgent 失败边界诊断。只打印可定位的安全摘要，不打印邮件正文、附件 URL、鉴权头或 Secret。 + */ + private void logOpenApiFailure( + String phase, + SuperAgentOpenApiMessageRequest request, + String sessionId, + SuperAgentOpenApiSseParser.ParsedState state, + Exception exception) { + if (!log.isWarnEnabled()) { + return; + } + log.warn( + "SuperAgent Open API failed. phase={}, request_id={}, external_subject_id={}, " + + "idempotency_key={}, message_chars={}, session_id={}, run_id={}, run_uri_path={}, " + + "last_event_id={}, event_types={}, failure_code={}, failure_message={}, metadata={}, " + + "safe_error_summary={}, exception_chain={}", + safeLogText(phase), + safeLogText(request.requestId()), + safeLogText(textOrDefault(request.externalSubjectId(), properties.getExternalSubjectId())), + safeLogText(request.idempotencyKey()), + request.message() == null ? 0 : request.message().length(), + safeLogText(sessionId), + state == null ? null : safeLogText(state.runId()), + state == null ? null : safeUriPath(state.runUri()), + state == null ? null : safeLogText(state.lastEventId()), + state == null ? null : state.eventTypes(), + state == null ? null : safeLogText(state.failureCode()), + state == null ? null : safeLogText(state.failureMessage()), + safeMetadataSummary(request.metadata()), + safeLogText(exception.getMessage()), + safeExceptionChain(exception)); + } + + /** + * 输出可排查的 metadata 摘要，只保留内部 ID / hotel 等定位字段。 + */ + private Map safeMetadataSummary(Map metadata) { + Map summary = new LinkedHashMap<>(); + if (metadata == null || metadata.isEmpty()) { + return summary; + } + putSafeMetadata(summary, metadata, "debug_run_id"); + putSafeMetadata(summary, metadata, "dispatch_run_id"); + putSafeMetadata(summary, metadata, "source_message_id"); + putSafeMetadata(summary, metadata, "hotel_id"); + putSafeMetadata(summary, metadata, "external_message_id"); + putSafeMetadata(summary, metadata, "provider"); + putSafeMetadata(summary, metadata, "schema_version"); + return summary; + } + + /** + * 追加单个安全 metadata 字段。 + */ + private void putSafeMetadata(Map summary, Map metadata, String fieldName) { + Object value = metadata.get(fieldName); + if (value == null) { + return; + } + summary.put(fieldName, safeLogText(String.valueOf(value))); + } + + /** + * 生成异常链短摘要，避免日志输出完整堆栈和潜在敏感内容。 + */ + private String safeExceptionChain(Throwable throwable) { + StringBuilder builder = new StringBuilder(); + Throwable current = throwable; + int depth = 0; + while (current != null && depth < 4) { + if (!builder.isEmpty()) { + builder.append(" <- "); + } + builder.append(current.getClass().getSimpleName()); + String message = safeLogText(current.getMessage()); + if (message != null && !message.isBlank()) { + builder.append(": ").append(message); + } + current = current.getCause(); + depth++; + } + return builder.toString(); + } + + /** + * 输出 URI 的路径部分，避免在日志里写入完整域名、签名参数或外链。 + */ + private String safeUriPath(String uri) { + if (blank(uri)) { + return null; + } + try { + URI parsedUri = URI.create(uri); + return safeLogText(parsedUri.getPath()); + } catch (Exception exception) { + return safeLogText(uri); + } + } + + /** + * 对日志文本做 Secret / URL 脱敏和长度截断。 + */ + private String safeLogText(String value) { + if (value == null) { + return null; + } + String sanitized = JSON_SECRET_PATTERN.matcher(value).replaceAll("$1***$2"); + sanitized = HEADER_SECRET_PATTERN.matcher(sanitized).replaceAll("$1***"); + sanitized = TEXT_SECRET_PATTERN.matcher(sanitized).replaceAll("$1***"); + sanitized = URL_PATTERN.matcher(sanitized).replaceAll("***URL***"); + return sanitized.length() > SAFE_LOG_VALUE_MAX_LENGTH + ? sanitized.substring(0, SAFE_LOG_VALUE_MAX_LENGTH) + : sanitized; + } + /** * 读取 JSON 文本字段。 */ diff --git a/server/src/main/java/cn/nianxx/thhotel/integrations/ai/superagent/service/impl/SuperAgentOpenApiSseParser.java b/server/src/main/java/cn/nianxx/thhotel/integrations/ai/superagent/service/impl/SuperAgentOpenApiSseParser.java index ac8dcb4..b976942 100644 --- a/server/src/main/java/cn/nianxx/thhotel/integrations/ai/superagent/service/impl/SuperAgentOpenApiSseParser.java +++ b/server/src/main/java/cn/nianxx/thhotel/integrations/ai/superagent/service/impl/SuperAgentOpenApiSseParser.java @@ -589,5 +589,26 @@ public class SuperAgentOpenApiSseParser { public String lastEventId() { return lastEventId; } + + /** + * 返回已消费的 SSE 事件类型，用于失败诊断。 + */ + public List eventTypes() { + return List.copyOf(eventTypes); + } + + /** + * 返回 SSE 失败事件中的错误码，用于失败诊断。 + */ + public String failureCode() { + return failureCode; + } + + /** + * 返回 SSE 失败事件中的错误摘要，用于失败诊断。 + */ + public String failureMessage() { + return failureMessage; + } } } diff --git 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 index 9f9ac62..0734f32 100644 --- 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 @@ -35,7 +35,11 @@ import javax.net.ssl.SSLContext; import javax.net.ssl.SSLParameters; import javax.net.ssl.SSLSession; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.springframework.boot.test.system.CapturedOutput; +import org.springframework.boot.test.system.OutputCaptureExtension; +@ExtendWith(OutputCaptureExtension.class) class SuperAgentOpenApiClientImplTest { @Test @@ -232,6 +236,50 @@ class SuperAgentOpenApiClientImplTest { assertThat(httpClient.streamPostCount()).isEqualTo(1); } + @Test + void shouldLogSafeRequestAndResponseSummaryWhenSseRunFails(CapturedOutput output) throws Exception { + FailureSseHttpClient httpClient = new FailureSseHttpClient(); + SuperAgentOpenApiProperties properties = openApiProperties(); + properties.setApiKey("secret-open-api-key"); + properties.setExternalSubjectId("th-hotel-agentbus-source-message"); + SuperAgentOpenApiClientImpl client = new SuperAgentOpenApiClientImpl( + properties, + new SuperAgentOpenApiSseParser(new ObjectMapper()), + new ObjectMapper(), + httpClient); + + assertThatThrownBy(() -> client.invokeMailDebug(new SuperAgentMailDebugRequest( + "MAIL_BODY_SECRET_TOKEN https://oss.secret/mail-body.html", + "debug-eml-2079814223967580162", + Map.of( + "debug_run_id", "2079814223967580162", + "source_message_id", "2079159429322088449", + "external_message_id", "AAMk-secret-message-id")))) + .isInstanceOf(SuperAgentOpenApiException.class) + .hasMessageContaining("open_agent_run_failed"); + + assertThat(httpClient.streamPostCount()).isEqualTo(1); + assertThat(output) + .contains("SuperAgent Open API failed") + .contains("phase=sse_result") + .contains("request_id=debug-eml-2079814223967580162") + .contains("external_subject_id=th-hotel-agentbus-source-message") + .contains("idempotency_key=debug-eml-2079814223967580162") + .contains("session_id=session-failure-001") + .contains("run_id=run-failure-001") + .contains("last_event_id=evt-3") + .contains("failure_code=open_agent_run_failed") + .contains("metadata={debug_run_id=2079814223967580162, source_message_id=2079159429322088449"); + assertThat(output) + .doesNotContain("MAIL_BODY_SECRET_TOKEN") + .doesNotContain("secret-open-api-key") + .doesNotContain("secret-token") + .doesNotContain("https://oss.secret") + .doesNotContain("Authorization") + .doesNotContain("Cookie") + .doesNotContain("csrf_token"); + } + private SuperAgentOpenApiProperties openApiProperties() { SuperAgentOpenApiProperties properties = new SuperAgentOpenApiProperties(); properties.setEnabled(true); @@ -354,6 +402,110 @@ class SuperAgentOpenApiClientImplTest { } } + private static final class FailureSseHttpClient extends HttpClient { + private final AtomicInteger streamPostCount = new AtomicInteger(); + + private int streamPostCount() { + return streamPostCount.get(); + } + + @Override + public Optional cookieHandler() { + return Optional.empty(); + } + + @Override + public Optional connectTimeout() { + return Optional.empty(); + } + + @Override + public Redirect followRedirects() { + return Redirect.NEVER; + } + + @Override + public Optional proxy() { + return Optional.empty(); + } + + @Override + public SSLContext sslContext() { + try { + return SSLContext.getDefault(); + } catch (NoSuchAlgorithmException exception) { + throw new IllegalStateException(exception); + } + } + + @Override + public SSLParameters sslParameters() { + return new SSLParameters(); + } + + @Override + public Optional authenticator() { + return Optional.empty(); + } + + @Override + public Version version() { + return Version.HTTP_1_1; + } + + @Override + public Optional executor() { + return Optional.empty(); + } + + @Override + public HttpResponse send(HttpRequest request, HttpResponse.BodyHandler responseBodyHandler) { + String path = request.uri().getPath(); + if (path.endsWith("/api/open/agent-sessions")) { + return response(request, 200, "{\"session_id\":\"session-failure-001\"}"); + } + if (path.endsWith("/api/open/agent-sessions/session-failure-001/messages/stream")) { + streamPostCount.incrementAndGet(); + byte[] response = """ + id: evt-1 + event: metadata + data: {"run_id":"run-failure-001","resolved_profile_id":"profile-failure-001"} + + id: evt-2 + event: error + data: {"code":"open_agent_run_failed","message":"Open agent run failed with token=secret-token and https://oss.secret/file.pdf"} + + id: evt-3 + event: end + data: {} + + """.getBytes(StandardCharsets.UTF_8); + return response(request, 200, new ByteArrayInputStream(response)); + } + throw new AssertionError("Unexpected request path: " + path); + } + + @Override + public CompletableFuture> sendAsync( + HttpRequest request, + HttpResponse.BodyHandler responseBodyHandler) { + return CompletableFuture.failedFuture(new UnsupportedOperationException()); + } + + @Override + public CompletableFuture> sendAsync( + HttpRequest request, + HttpResponse.BodyHandler responseBodyHandler, + HttpResponse.PushPromiseHandler pushPromiseHandler) { + return CompletableFuture.failedFuture(new UnsupportedOperationException()); + } + + @SuppressWarnings("unchecked") + private HttpResponse response(HttpRequest request, int statusCode, Object body) { + return new SimpleHttpResponse<>(request, statusCode, (T) body); + } + } + private record SimpleHttpResponse( HttpRequest request, int statusCode,