修复SuperAgent自动分发恢复逻辑

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
 

docs/project/requirements/M007-agentbus-superagent-auto-dispatch-v1.md

@@ -44,8 +44,8 @@ AgentBus WebSocket 收到邮件
 | --- | --- |
 | 来源 provider | 必须是 `AGENTBUS` |
 | 捕获状态 | 必须是 `RECEIVED` |
-| 幂等结果 | 仅 `SourceMessageCaptureResult.newlyCreated=true` 时触发 |
+| 幂等结果 | `RECEIVED` 状态会尝试幂等创建 dispatch；数据库唯一键保证不会重复创建 |
-| 重复投递 | 不重复创建 dispatch；如 payload changed，只记录 SourceMessage 重复诊断 |
+| 重复投递 | 会再次尝试 `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。
+- AgentBus 捕获 `AGENTBUS + RECEIVED` 后，在配置开启时幂等创建 dispatch run；重复投递可补偿缺失 outbox。
-- worker 处理 `PENDING / RETRYABLE_FAILED`，调用共享 SuperAgent Open API client。
+- 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 摘要和状态。

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<SuperAgentDispatchRunSnapshot> claimDue(

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<SuperAgentDispatchRunSnapshot> 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());

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<InputStream> 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<InputStream> response = httpClient.send(
                 builder.build(),
                 HttpResponse.BodyHandlers.ofInputStream());
-        require2xx(response.statusCode(), null);
+        require2xxOrClose(response.statusCode(), response.body());
         consumeResponse(response.body(), state, traceConsumer);
     }
 

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<SuperAgentDispatchRunDraft> 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<List<String>> 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

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> cookieHandler() {
+            return Optional.empty();
+        }
+
+        @Override
+        public Optional<Duration> connectTimeout() {
+            return Optional.empty();
+        }
+
+        @Override
+        public Redirect followRedirects() {
+            return Redirect.NEVER;
+        }
+
+        @Override
+        public Optional<ProxySelector> 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> authenticator() {
+            return Optional.empty();
+        }
+
+        @Override
+        public Version version() {
+            return Version.HTTP_1_1;
+        }
+
+        @Override
+        public Optional<Executor> executor() {
+            return Optional.empty();
+        }
+
+        @Override
+        public <T> HttpResponse<T> send(HttpRequest request, HttpResponse.BodyHandler<T> 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 <T> CompletableFuture<HttpResponse<T>> sendAsync(
+                HttpRequest request,
+                HttpResponse.BodyHandler<T> responseBodyHandler) {
+            return CompletableFuture.failedFuture(new UnsupportedOperationException());
+        }
+
+        @Override
+        public <T> CompletableFuture<HttpResponse<T>> sendAsync(
+                HttpRequest request,
+                HttpResponse.BodyHandler<T> responseBodyHandler,
+                HttpResponse.PushPromiseHandler<T> pushPromiseHandler) {
+            return CompletableFuture.failedFuture(new UnsupportedOperationException());
+        }
+
+        @SuppressWarnings("unchecked")
+        private <T> HttpResponse<T> response(HttpRequest request, int statusCode, Object body) {
+            return new SimpleHttpResponse<>(request, statusCode, (T) body);
+        }
+    }
+
+    private record SimpleHttpResponse<T>(
+            HttpRequest request,
+            int statusCode,
+            T body
+    ) implements HttpResponse<T> {
+
+        @Override
+        public Optional<HttpResponse<T>> previousResponse() {
+            return Optional.empty();
+        }
+
+        @Override
+        public HttpHeaders headers() {
+            return HttpHeaders.of(Map.of(), (name, value) -> true);
+        }
+
+        @Override
+        public Optional<SSLSession> sslSession() {
+            return Optional.empty();
+        }
+
+        @Override
+        public java.net.URI uri() {
+            return request.uri();
+        }
+
+        @Override
+        public HttpClient.Version version() {
+            return HttpClient.Version.HTTP_1_1;
+        }
+    }
 }