diff --git a/docs/project/go-live-notes.md b/docs/project/go-live-notes.md index 7957404..cab88a0 100644 --- a/docs/project/go-live-notes.md +++ b/docs/project/go-live-notes.md @@ -130,6 +130,7 @@ - `AGENTBUS_DEFAULT_HOTEL_ID` 是 M005 前旧变量，当前后端不再读取；系统酒店来自 `platform_hotel` 唯一 `ACTIVE` 酒店。 - 当前实现不发送 ACK、不发送 `task.result`、不自动回复客户。 - M007 自动分发即使开启，也只能在 SourceMessage 入库后通过 dispatch / outbox 异步调用 SuperAgent，不能在 AgentBus WebSocket 回调内同步等待外部返回。 +- M007 dispatch 创建使用 SourceMessage + dispatch_source 幂等；重复 `RECEIVED` 投递会尝试补偿缺失 outbox，锁已过期的 `RUNNING` 记录会被 worker 重新领取。 ### 3.5 SuperAgent HMAC diff --git a/docs/project/requirements/M007-agentbus-superagent-auto-dispatch-v1.md b/docs/project/requirements/M007-agentbus-superagent-auto-dispatch-v1.md index 58a686d..9ba8dbe 100644 --- a/docs/project/requirements/M007-agentbus-superagent-auto-dispatch-v1.md +++ b/docs/project/requirements/M007-agentbus-superagent-auto-dispatch-v1.md @@ -44,8 +44,8 @@ AgentBus WebSocket 收到邮件 | --- | --- | | 来源 provider | 必须是 `AGENTBUS` | | 捕获状态 | 必须是 `RECEIVED` | -| 幂等结果 | 仅 `SourceMessageCaptureResult.newlyCreated=true` 时触发 | -| 重复投递 | 不重复创建 dispatch；如 payload changed，只记录 SourceMessage 重复诊断 | +| 幂等结果 | `RECEIVED` 状态会尝试幂等创建 dispatch；数据库唯一键保证不会重复创建 | +| 重复投递 | 会再次尝试 `insertIfAbsent` 补偿可能缺失的 dispatch；如 payload changed，只记录 SourceMessage 重复诊断 | | Debug EML | 不参与生产自动分发 | | FAILED Inbox | 不分发，保留入库失败原因供排查 | @@ -150,7 +150,7 @@ AgentBus 自动分发发送给 SuperAgent 的 message 第一版应基于 SourceM ## 8. 状态流转 ```text -PENDING / RETRYABLE_FAILED +PENDING / RETRYABLE_FAILED / 过期 RUNNING → RUNNING → SUCCEEDED @@ -191,11 +191,13 @@ RUNNING / RETRYABLE_FAILED ## 10. 验收标准 -- AgentBus 新邮件入库成功后，在开启配置时创建一条 dispatch run。 +- AgentBus 新邮件入库成功后，在开启配置时幂等创建一条 dispatch run。 - 重复 AgentBus 邮件投递不会创建第二条 dispatch run。 +- 如首次入库后 dispatch 创建失败，后续重复 `RECEIVED` 投递可以补偿创建缺失的 dispatch run。 +- worker 崩溃后，锁过期的 `RUNNING` dispatch 可以被重新领取处理。 - Debug EML 不会进入生产 dispatch run。 - `FAILED` SourceMessage 不会触发 dispatch。 -- worker 能领取 `PENDING` 记录并调用 SuperAgent Open API。 +- worker 能领取 `PENDING`、`RETRYABLE_FAILED` 和锁已过期的 `RUNNING` 记录并调用 SuperAgent Open API。 - SuperAgent 调用成功时保存 session、run、raw answer、parsed json、trace 摘要和 `SUCCEEDED` 状态。 - SuperAgent SSE 缺少 `end`、缺少最终内容或 `run.completed` 时不能成功。 - SSE EOF 后如果已有 `run_id`，使用 `/events` 恢复，不重发初始 POST。 @@ -209,8 +211,8 @@ RUNNING / RETRYABLE_FAILED - 新增 `platform_superagent_dispatch_run` 表和 `V20__create_superagent_dispatch_run.sql`。 - 新增 `SuperAgentDispatchRunEntity`、Mapper、Repository、Service 和配置类。 -- AgentBus 捕获 `AGENTBUS + RECEIVED + newlyCreated=true` 后，在配置开启时创建 dispatch run。 -- worker 处理 `PENDING / RETRYABLE_FAILED`，调用共享 SuperAgent Open API client。 +- AgentBus 捕获 `AGENTBUS + RECEIVED` 后，在配置开启时幂等创建 dispatch run；重复投递可补偿缺失 outbox。 +- worker 处理 `PENDING / RETRYABLE_FAILED / 锁已过期 RUNNING`，调用共享 SuperAgent Open API client。 - Open API client 支持 `Content-Location`、`run_id`、SSE `id`、`Last-Event-ID`、`GET /runs/{run_id}` 和 `GET /runs/{run_id}/events` 恢复。 - Debug EML 继续复用共享 Open API client。 - dispatch 成功保存 session、run、last event id、raw answer、parsed json、trace 摘要和状态。 diff --git a/server/src/main/java/cn/nianxx/thhotel/integrations/ai/superagent/repository/MybatisSuperAgentDispatchRunRepository.java b/server/src/main/java/cn/nianxx/thhotel/integrations/ai/superagent/repository/MybatisSuperAgentDispatchRunRepository.java index c320ad2..85251ed 100644 --- a/server/src/main/java/cn/nianxx/thhotel/integrations/ai/superagent/repository/MybatisSuperAgentDispatchRunRepository.java +++ b/server/src/main/java/cn/nianxx/thhotel/integrations/ai/superagent/repository/MybatisSuperAgentDispatchRunRepository.java @@ -47,7 +47,7 @@ public class MybatisSuperAgentDispatchRunRepository implements SuperAgentDispatc } /** - * 抢占 PENDING/RETRYABLE_FAILED 记录。先查候选再逐条条件更新，降低多 worker 重复执行概率。 + * 抢占 PENDING/RETRYABLE_FAILED 或锁已过期的 RUNNING 记录。先查候选再逐条条件更新，降低多 worker 重复执行概率。 */ @Override public List claimDue( diff --git a/server/src/main/java/cn/nianxx/thhotel/integrations/ai/superagent/service/impl/SuperAgentDispatchServiceImpl.java b/server/src/main/java/cn/nianxx/thhotel/integrations/ai/superagent/service/impl/SuperAgentDispatchServiceImpl.java index a9bf3bc..9d3315a 100644 --- a/server/src/main/java/cn/nianxx/thhotel/integrations/ai/superagent/service/impl/SuperAgentDispatchServiceImpl.java +++ b/server/src/main/java/cn/nianxx/thhotel/integrations/ai/superagent/service/impl/SuperAgentDispatchServiceImpl.java @@ -80,7 +80,6 @@ public class SuperAgentDispatchServiceImpl implements SuperAgentDispatchService return; } if (captureResult.inboxId() == null - || !captureResult.created() || !CAPTURE_STATUS_RECEIVED.equals(captureResult.captureStatus())) { return; } @@ -116,7 +115,10 @@ public class SuperAgentDispatchServiceImpl implements SuperAgentDispatchService LocalDateTime now = nowUtc(); List runs = runRepository.claimDue( "local-worker", - List.of(SuperAgentDispatchStatus.PENDING.code(), SuperAgentDispatchStatus.RETRYABLE_FAILED.code()), + List.of( + SuperAgentDispatchStatus.PENDING.code(), + SuperAgentDispatchStatus.RETRYABLE_FAILED.code(), + SuperAgentDispatchStatus.RUNNING.code()), now, now.plus(lockTtl()), safeBatchSize()); 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 5330a47..6d05049 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 @@ -7,17 +7,18 @@ import cn.nianxx.thhotel.integrations.ai.superagent.common.result.SuperAgentOpen import cn.nianxx.thhotel.integrations.ai.superagent.service.SuperAgentOpenApiClient; import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; -import java.io.InputStream; import java.io.IOException; -import java.nio.charset.StandardCharsets; +import java.io.InputStream; +import java.io.InputStreamReader; import java.net.URI; +import java.nio.charset.StandardCharsets; import java.net.http.HttpClient; import java.net.http.HttpRequest; import java.net.http.HttpResponse; -import java.io.InputStreamReader; import java.util.LinkedHashMap; import java.util.Map; import java.util.function.Consumer; +import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; /** @@ -29,17 +30,31 @@ public class SuperAgentOpenApiClientImpl implements SuperAgentOpenApiClient { private final SuperAgentOpenApiProperties properties; private final SuperAgentOpenApiSseParser sseParser; private final ObjectMapper objectMapper; + private final HttpClient injectedHttpClient; /** * 注入 SuperAgent 配置、SSE 解析器和 JSON 工具。 */ + @Autowired public SuperAgentOpenApiClientImpl( SuperAgentOpenApiProperties properties, SuperAgentOpenApiSseParser sseParser, ObjectMapper objectMapper) { + this(properties, sseParser, objectMapper, null); + } + + /** + * 注入可替换的 HTTP 客户端，供单元测试验证低层 HTTP 边界行为。 + */ + SuperAgentOpenApiClientImpl( + SuperAgentOpenApiProperties properties, + SuperAgentOpenApiSseParser sseParser, + ObjectMapper objectMapper, + HttpClient injectedHttpClient) { this.properties = properties; this.sseParser = sseParser; this.objectMapper = objectMapper; + this.injectedHttpClient = injectedHttpClient; } /** @@ -129,7 +144,7 @@ public class SuperAgentOpenApiClientImpl implements SuperAgentOpenApiClient { HttpResponse httpResponse = httpClient.send( httpRequest, HttpResponse.BodyHandlers.ofInputStream()); - require2xx(httpResponse.statusCode(), null); + require2xxOrClose(httpResponse.statusCode(), httpResponse.body()); SuperAgentOpenApiSseParser.ParsedState state = sseParser.newState(sessionId); applyRunLocation(baseUri, httpResponse, state); try { @@ -160,6 +175,9 @@ public class SuperAgentOpenApiClientImpl implements SuperAgentOpenApiClient { * 构造 JDK HttpClient；SSE 请求不设置整体 read timeout，避免长任务被固定超时截断。 */ private HttpClient httpClient() { + if (injectedHttpClient != null) { + return injectedHttpClient; + } return HttpClient.newBuilder() .connectTimeout(properties.getConnectTimeout()) .version(HttpClient.Version.HTTP_1_1) @@ -202,6 +220,31 @@ public class SuperAgentOpenApiClientImpl implements SuperAgentOpenApiClient { } } + /** + * 校验流式响应状态码；失败时必须关闭响应流，避免连接资源泄漏。 + */ + private void require2xxOrClose(int statusCode, InputStream responseBody) { + if (statusCode >= 200 && statusCode < 300) { + return; + } + closeQuietly(responseBody); + require2xx(statusCode, null); + } + + /** + * 静默关闭错误响应流；关闭异常不覆盖原始 HTTP 状态错误。 + */ + private void closeQuietly(InputStream responseBody) { + if (responseBody == null) { + return; + } + try { + responseBody.close(); + } catch (IOException exception) { + // 错误响应流关闭失败不覆盖原始 HTTP 状态错误。 + } + } + /** * 从 Content-Location 保存 run URI 和 runId，供 EOF 后恢复使用。 */ @@ -297,7 +340,7 @@ public class SuperAgentOpenApiClientImpl implements SuperAgentOpenApiClient { HttpResponse response = httpClient.send( builder.build(), HttpResponse.BodyHandlers.ofInputStream()); - require2xx(response.statusCode(), null); + require2xxOrClose(response.statusCode(), response.body()); consumeResponse(response.body(), state, traceConsumer); } diff --git a/server/src/test/java/cn/nianxx/thhotel/integrations/ai/superagent/service/impl/SuperAgentDispatchServiceImplTest.java b/server/src/test/java/cn/nianxx/thhotel/integrations/ai/superagent/service/impl/SuperAgentDispatchServiceImplTest.java index c0ea830..768c03e 100644 --- a/server/src/test/java/cn/nianxx/thhotel/integrations/ai/superagent/service/impl/SuperAgentDispatchServiceImplTest.java +++ b/server/src/test/java/cn/nianxx/thhotel/integrations/ai/superagent/service/impl/SuperAgentDispatchServiceImplTest.java @@ -2,6 +2,7 @@ package cn.nianxx.thhotel.integrations.ai.superagent.service.impl; import static org.assertj.core.api.Assertions.assertThat; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyInt; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; @@ -66,13 +67,46 @@ class SuperAgentDispatchServiceImplTest { true, false, "FAILED")); + + verify(runRepository, never()).insertIfAbsent(any(SuperAgentDispatchRunDraft.class)); + } + + @Test + void shouldCreateDispatchForDuplicateReceivedSourceMessageToCompensateMissingOutbox() { + SuperAgentDispatchRunRepository runRepository = mock(SuperAgentDispatchRunRepository.class); + when(runRepository.insertIfAbsent(any(SuperAgentDispatchRunDraft.class))).thenReturn(99002L); + SuperAgentDispatchServiceImpl service = service(runRepository, mock(SourceMessageInboxRepository.class), + mock(SuperAgentOpenApiClient.class), properties(true, false)); + service.enqueueAgentBusRealtime(agentBusCommand("AGENTBUS"), new SourceMessageCaptureResult( 88004L, false, false, "RECEIVED")); - verify(runRepository, never()).insertIfAbsent(any(SuperAgentDispatchRunDraft.class)); + ArgumentCaptor captor = ArgumentCaptor.forClass(SuperAgentDispatchRunDraft.class); + verify(runRepository).insertIfAbsent(captor.capture()); + assertThat(captor.getValue().sourceMessageId()).isEqualTo(88004L); + assertThat(captor.getValue().dispatchSource()).isEqualTo(SuperAgentDispatchSource.AGENTBUS_REALTIME.code()); + } + + @Test + void shouldClaimExpiredRunningDispatchesForCrashRecovery() { + SuperAgentDispatchRunRepository runRepository = mock(SuperAgentDispatchRunRepository.class); + when(runRepository.claimDue(any(), any(), any(), any(), anyInt())).thenReturn(List.of()); + SuperAgentDispatchServiceImpl service = service(runRepository, mock(SourceMessageInboxRepository.class), + mock(SuperAgentOpenApiClient.class), properties(true, true)); + + service.processDueDispatches(); + + @SuppressWarnings("unchecked") + ArgumentCaptor> statusesCaptor = ArgumentCaptor.forClass(List.class); + verify(runRepository).claimDue(any(), statusesCaptor.capture(), any(), any(), anyInt()); + assertThat(statusesCaptor.getValue()) + .contains( + SuperAgentDispatchStatus.PENDING.code(), + SuperAgentDispatchStatus.RETRYABLE_FAILED.code(), + SuperAgentDispatchStatus.RUNNING.code()); } @Test 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 038c11f..65a221a 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 @@ -1,22 +1,39 @@ package cn.nianxx.thhotel.integrations.ai.superagent.service.impl; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; 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.ByteArrayInputStream; +import java.io.IOException; import java.io.OutputStream; +import java.net.Authenticator; +import java.net.CookieHandler; import java.net.InetAddress; import java.net.InetSocketAddress; +import java.net.ProxySelector; +import java.net.http.HttpClient; +import java.net.http.HttpHeaders; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; import java.nio.charset.StandardCharsets; +import java.security.NoSuchAlgorithmException; import java.time.Duration; import java.util.ArrayList; import java.util.List; import java.util.Map; +import java.util.Optional; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.Executor; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; +import javax.net.ssl.SSLContext; +import javax.net.ssl.SSLParameters; +import javax.net.ssl.SSLSession; import org.junit.jupiter.api.Test; class SuperAgentOpenApiClientImplTest { @@ -180,4 +197,178 @@ class SuperAgentOpenApiClientImplTest { server.stop(0); } } + + @Test + void shouldCloseStreamingResponseBodyWhenStatusIsNot2xx() { + TrackingInputStream responseBody = new TrackingInputStream("bad gateway".getBytes(StandardCharsets.UTF_8)); + FakeHttpClient httpClient = new FakeHttpClient(responseBody); + SuperAgentOpenApiClientImpl client = new SuperAgentOpenApiClientImpl( + openApiProperties(), + new SuperAgentOpenApiSseParser(new ObjectMapper()), + new ObjectMapper(), + httpClient); + + assertThatThrownBy(() -> client.invokeMailDebug(new SuperAgentMailDebugRequest( + "debug message", + "debug-500-idempotency", + Map.of("source", "unit-test")))) + .isInstanceOf(SuperAgentOpenApiException.class) + .hasMessageContaining("status=502"); + assertThat(responseBody.closed()).isTrue(); + assertThat(httpClient.streamPostCount()).isEqualTo(1); + } + + private SuperAgentOpenApiProperties openApiProperties() { + SuperAgentOpenApiProperties properties = new SuperAgentOpenApiProperties(); + properties.setEnabled(true); + properties.setBaseUrl("http://superagent.test"); + properties.setApiKey("df_open_test"); + properties.setExternalSubjectId("debug-subject"); + return properties; + } + + private static final class TrackingInputStream extends ByteArrayInputStream { + private boolean closed; + + private TrackingInputStream(byte[] buffer) { + super(buffer); + } + + @Override + public void close() throws IOException { + closed = true; + super.close(); + } + + private boolean closed() { + return closed; + } + } + + private static final class FakeHttpClient extends HttpClient { + private final TrackingInputStream streamingBody; + private final AtomicInteger streamPostCount = new AtomicInteger(); + + private FakeHttpClient(TrackingInputStream streamingBody) { + this.streamingBody = streamingBody; + } + + 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-http-500\"}"); + } + if (path.endsWith("/api/open/agent-sessions/session-http-500/messages/stream")) { + streamPostCount.incrementAndGet(); + return response(request, 502, streamingBody); + } + 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, + T body + ) implements HttpResponse { + + @Override + public Optional> previousResponse() { + return Optional.empty(); + } + + @Override + public HttpHeaders headers() { + return HttpHeaders.of(Map.of(), (name, value) -> true); + } + + @Override + public Optional sslSession() { + return Optional.empty(); + } + + @Override + public java.net.URI uri() { + return request.uri(); + } + + @Override + public HttpClient.Version version() { + return HttpClient.Version.HTTP_1_1; + } + } }