返回 Duration.ZERO 可禁用消息静默检查（无状态 HTTP 或自带传输心跳的渠道）。 + * ERROR 状态的恢复不受影响。 *
通用默认 60 分钟；长轮询类渠道（如 iLink 微信）应覆盖为 5 分钟， * 这样代理/NAT 侧 2–5 分钟 idle-close 把连接切断后， * {@code ChannelHealthMonitor} 能在几分钟内触发自动重启，而非静默等整小时。
diff --git a/mateclaw-server/src/main/java/vip/mate/channel/ChannelHealthMonitor.java b/mateclaw-server/src/main/java/vip/mate/channel/ChannelHealthMonitor.java index c0613793..52497b94 100644 --- a/mateclaw-server/src/main/java/vip/mate/channel/ChannelHealthMonitor.java +++ b/mateclaw-server/src/main/java/vip/mate/channel/ChannelHealthMonitor.java @@ -82,12 +82,12 @@ public class ChannelHealthMonitor { Duration staleThreshold; try { Duration d = adapter.stalenessThreshold(); - staleThreshold = (d == null || d.isNegative() || d.isZero()) ? DEFAULT_STALE_THRESHOLD : d; + staleThreshold = (d == null || d.isNegative()) ? DEFAULT_STALE_THRESHOLD : d; } catch (Exception ex) { staleThreshold = DEFAULT_STALE_THRESHOLD; } if (reason == null && state == AbstractChannelAdapter.ConnectionState.CONNECTED - && sinceLastEvent > staleThreshold.toMillis()) { + && !staleThreshold.isZero() && sinceLastEvent > staleThreshold.toMillis()) { reason = String.format("stale connection, no events for %ds (threshold=%ds)", sinceLastEvent / 1000, staleThreshold.toSeconds()); } diff --git a/mateclaw-server/src/main/java/vip/mate/channel/feishu/FeishuChannelAdapter.java b/mateclaw-server/src/main/java/vip/mate/channel/feishu/FeishuChannelAdapter.java index 045ea059..2f021790 100644 --- a/mateclaw-server/src/main/java/vip/mate/channel/feishu/FeishuChannelAdapter.java +++ b/mateclaw-server/src/main/java/vip/mate/channel/feishu/FeishuChannelAdapter.java @@ -63,7 +63,7 @@ import java.util.concurrent.TimeUnit; * - enable_nickname_cache: 是否通过 Contact API 获取用户昵称（默认 true） * - media_download_enabled: 是否下载消息中的媒体文件（默认 false） * - enable_quoted_context: 是否拉取被引用消息内容注入到 prompt（默认 true） - * - silent_disconnect_threshold_seconds: WebSocket 静默断连阈值（默认 1800，0 禁用） + * - silent_disconnect_threshold_seconds: 已弃用；健康检查使用协议心跳，不依赖消息频率 * - stale_event_threshold_seconds: 过滤旧事件阈值（默认 30，0 禁用） * - card_format: 卡片格式化模式 "auto"（默认）| "always" | "never" * auto: 根据内容自动检测；always: 全部包卡片；never: 全部纯文本（降级/调试用） @@ -126,22 +126,15 @@ public class FeishuChannelAdapter extends AbstractChannelAdapter implements Stre private static final int NICKNAME_CACHE_MAX = 500; /** WebSocket 客户端（websocket 模式） */ - private volatile com.lark.oapi.ws.Client wsClient; + private volatile FeishuWebSocketClient wsClient; + private final Object wsLifecycleLock = new Object(); /** WebSocket 连接线程 */ private volatile Thread wsThread; - /** WebSocket 静默断连看门狗：定期检查最近事件时间，超过阈值就强制重连 */ - private ScheduledFuture silentDisconnectWatchdog; - - /** 是否已收到至少一个事件（用于避免新连接立即触发静默超时） */ - private volatile boolean hasReceivedFirstEvent = false; - - /** 看门狗检查间隔（秒） */ - private static final long WATCHDOG_INTERVAL_SECONDS = 60L; - - /** 静默断连默认阈值（秒）：30 分钟无事件就视为可疑 */ - private static final long DEFAULT_SILENT_THRESHOLD_SECONDS = 1800L; + /** Poll transport failure; user-message silence is never a disconnect signal. */ + private ScheduledFuture transportWatchdog; + private static final long WATCHDOG_INTERVAL_SECONDS = 15L; /** 旧事件过滤默认阈值（秒）：超过 30 秒的事件视为重连后回放 */ private static final long DEFAULT_STALE_THRESHOLD_SECONDS = 30L; @@ -381,8 +374,39 @@ public class FeishuChannelAdapter extends AbstractChannelAdapter implements Stre // ==================== 生命周期 ==================== + @Override + public Duration stalenessThreshold() { + return Duration.ZERO; // Protocol ping/pong detects dead transports, even on quiet channels. + } + + @Override + protected void onReconnectSuccess() { + synchronized (wsLifecycleLock) { + if (!running.get()) return; + if (connectsAsynchronously() && (wsClient == null || !wsClient.isConnected())) { + throw new IllegalStateException("WebSocket closed before connection became ready"); + } + super.onReconnectSuccess(); + } + } + + @Override + protected boolean connectsAsynchronously() { + return "websocket".equals(getConfigString("connection_mode", "websocket")); + } + @Override protected void doStart() { + try { + initializeChannel(); + } catch (RuntimeException e) { + doStop(); + if (reconnectScheduler != null) reconnectScheduler.shutdownNow(); + throw e; + } + } + + private void initializeChannel() { String appId = getConfigString("app_id"); String appSecret = getConfigString("app_secret"); @@ -439,7 +463,9 @@ public class FeishuChannelAdapter extends AbstractChannelAdapter implements Stre // 关闭 WebSocket stopWebSocket(); + HttpClient previousHttpClient = this.httpClient; this.httpClient = null; + if (previousHttpClient != null) previousHttpClient.shutdownNow(); this.tenantAccessToken = null; // Reset bot-open-id state under the same lock getBotOpenId uses, so a // dispatch thread mid-fetch sees a consistent (cleared) view rather than @@ -462,27 +488,11 @@ public class FeishuChannelAdapter extends AbstractChannelAdapter implements Stre String appSecret = getConfigString("app_secret"); String connectionMode = getConfigString("connection_mode", "websocket"); - // 重新建立 HTTP 客户端 - this.httpClient = HttpClient.newBuilder() - .connectTimeout(Duration.ofSeconds(10)) - .build(); - - try { - refreshTenantAccessToken(); - } catch (Exception e) { - log.warn("[feishu] Token refresh during reconnect failed: {}", e.getMessage()); - } - - // Re-prefetch bot open_id after reconnect: same dispatch-thread latency - // concern as doStart, plus picks up a rotated app identity if any. - getBotOpenId(); - + // Token refresh has its own schedule; a WebSocket reconnect does not rotate HTTP state. if ("websocket".equals(connectionMode)) { log.info("[feishu] Reconnecting WebSocket..."); stopWebSocket(); - // 同步连接：在当前重连线程中直接阻塞调用 start() - // 连接成功 start() 会一直阻塞（不会返回到这里） - // 连接失败 start() 抛异常，由 AbstractChannelAdapter.scheduleReconnect 捕获并触发 onReconnectFailed + // Wait for the handshake; failures propagate to the adapter's single backoff loop. startWebSocketSync(appId, appSecret); } @@ -494,7 +504,7 @@ public class FeishuChannelAdapter extends AbstractChannelAdapter implements Stre /** * 创建 WebSocket 客户端实例（不启动连接） */ - private com.lark.oapi.ws.Client createWsClient(String appId, String appSecret) { + FeishuWebSocketClient createWsClient(String appId, String appSecret) { EventDispatcher eventDispatcher = EventDispatcher.newBuilder("", "") .onP2MessageReceiveV1(new ImService.P2MessageReceiveV1Handler() { @Override @@ -588,130 +598,94 @@ public class FeishuChannelAdapter extends AbstractChannelAdapter implements Stre }) .build(); - return new com.lark.oapi.ws.Client.Builder(appId, appSecret) + return new FeishuWebSocketClient(new com.lark.oapi.ws.Client.Builder(appId, appSecret) .eventHandler(eventDispatcher) .autoReconnect(false) // 由我们的 ExponentialBackoff 控制重连，不用 SDK 内置重连 .domain("lark".equals(getConfigString("domain", "feishu")) ? "https://open.larksuite.com" : "https://open.feishu.cn") - .build(); + .build()); } - /** - * 启动 WebSocket 长连接（异步，用于 doStart 首次启动） - * 在守护线程中运行，避免阻塞主线程。连接失败时触发 onDisconnected → 退避重连 - */ - private void startWebSocket(String appId, String appSecret) { - wsClient = createWsClient(appId, appSecret); - - wsThread = new Thread(() -> { - try { - log.info("[feishu] WebSocket connecting (long connection)..."); - wsClient.start(); - // start() blocks until disconnect; if it returns normally, it means disconnected - if (running.get()) { - onDisconnected("WebSocket connection ended"); + /** First connection is asynchronous; start() returning means handshake initiated. */ + void startWebSocket(String appId, String appSecret) { + synchronized (wsLifecycleLock) { + if (!running.get()) return; + FeishuWebSocketClient client = createWsClient(appId, appSecret); + wsClient = client; + wsThread = new Thread(() -> { + try { + client.start(); + synchronized (wsLifecycleLock) { + if (running.get() && wsClient == client) onReconnectSuccess(); + } + } catch (Exception e) { + synchronized (wsLifecycleLock) { + if (running.get() && wsClient == client) { + onDisconnected("WebSocket startup failed: " + e.getClass().getSimpleName()); + } + } } - } catch (Exception e) { - log.error("[feishu] WebSocket error: {}", e.getMessage(), e); - if (running.get()) { - onDisconnected("WebSocket error: " + e.getMessage()); - } - } - }, "feishu-ws-" + channelEntity.getId()); - wsThread.setDaemon(true); - wsThread.start(); - - startSilentDisconnectWatchdog(); - } - - /** - * 启动静默断连看门狗。 - *- * SDK 内部有 ping/pong 心跳，正常情况下连接断开会触发 start() 返回 → onDisconnected。 - * 但少数场景 TCP 层认为连接还在但事件不再流入（NAT 超时、半开连接、对端 hang 死）， - * SDK 也察觉不到。看门狗每 60s 检查一次「最近事件时间」，超过阈值就强制重连。 - *
- * 静默判定有两个前置条件： - * 1. 已经收到过至少一个事件（避免新连接立即触发误报） - * 2. silent_disconnect_threshold_seconds > 0（设为 0 可禁用看门狗） - *
- * 默认阈值 30 分钟。安静的渠道（一天没几条消息）建议调大到 1-2 小时； - * 高频渠道可以调小到 5-10 分钟以更快发现问题。 - */ - private void startSilentDisconnectWatchdog() { - long thresholdSec = getConfigLong("silent_disconnect_threshold_seconds", - DEFAULT_SILENT_THRESHOLD_SECONDS); - if (thresholdSec <= 0) { - log.debug("[feishu] Silent disconnect watchdog disabled (threshold=0)"); - return; - } - long thresholdMs = thresholdSec * 1000L; - - cancelSilentDisconnectWatchdog(); - silentDisconnectWatchdog = ensureReconnectScheduler().scheduleAtFixedRate(() -> { - if (!running.get() || wsClient == null) return; - if (!hasReceivedFirstEvent) return; // 没收到首个事件前不算静默 - - long silentMs = System.currentTimeMillis() - lastEventTimeMs.get(); - if (silentMs > thresholdMs) { - log.warn("[feishu] Silent WebSocket detected: no events for {}s (threshold {}s), forcing reconnect", - silentMs / 1000, thresholdSec); - onDisconnected("silent disconnect: no events for " + (silentMs / 1000) + "s"); - } - }, WATCHDOG_INTERVAL_SECONDS, WATCHDOG_INTERVAL_SECONDS, TimeUnit.SECONDS); - log.info("[feishu] Silent disconnect watchdog started (threshold {}s, check every {}s)", - thresholdSec, WATCHDOG_INTERVAL_SECONDS); - } - - private void cancelSilentDisconnectWatchdog() { - if (silentDisconnectWatchdog != null) { - silentDisconnectWatchdog.cancel(false); - silentDisconnectWatchdog = null; + }, "feishu-ws-" + channelEntity.getId()); + wsThread.setDaemon(true); + wsThread.start(); + startTransportWatchdog(); + } + } + + private void startTransportWatchdog() { + if (transportWatchdog != null) transportWatchdog.cancel(false); + transportWatchdog = ensureReconnectScheduler().scheduleWithFixedDelay( + this::checkWebSocketTransport, WATCHDOG_INTERVAL_SECONDS, + WATCHDOG_INTERVAL_SECONDS, TimeUnit.SECONDS); + } + + void checkWebSocketTransport() { + synchronized (wsLifecycleLock) { + FeishuWebSocketClient client = wsClient; + if (running.get() && client != null + && connectionState.get() == ConnectionState.CONNECTED && !client.isConnected()) { + // onDisconnected sets RECONNECTING before another probe can schedule a retry. + onDisconnected("WebSocket transport closed"); + } } } - /** - * 启动 WebSocket 长连接（同步，用于 doReconnect 重连线程） - * 在当前线程中阻塞调用 start()： - * - 连接成功后 start() 会一直阻塞（收消息），不会返回 - * - 连接失败 start() 抛异常，由 scheduleReconnect 的 catch 捕获 → onReconnectFailed → 退避递增 - */ private void startWebSocketSync(String appId, String appSecret) { - wsClient = createWsClient(appId, appSecret); - log.info("[feishu] WebSocket connecting (long connection)..."); - // 看门狗必须在 start() 之前启动，否则 start() 阻塞后这一行永远到不了， - // 一旦断连后重连这条路径，watchdog 就再也不会被恢复 - startSilentDisconnectWatchdog(); - wsClient.start(); // 阻塞：成功则永驻，失败则抛异常 + FeishuWebSocketClient client; + synchronized (wsLifecycleLock) { + if (!running.get()) throw new IllegalStateException("Channel stopped"); + client = createWsClient(appId, appSecret); + wsClient = client; + } + try { + client.start(); + synchronized (wsLifecycleLock) { + if (!running.get() || wsClient != client) throw new IllegalStateException("Channel stopped"); + startTransportWatchdog(); + } + } catch (Exception e) { + client.close(); + throw new IllegalStateException("WebSocket handshake failed", e); + } } - /** - * 关闭 WebSocket 连接。 - *
- * 必须主动关闭底层 WebSocket 连接并触发 SDK 的内部清理（停止 pingLoop、 - * 释放 ExecutorService）。仅置空引用会导致旧连接的 pingLoop 和线程池 - * 持续运行，造成文件描述符和线程泄漏，最终使新连接无法建立。 - *
- * SDK 2.7.0 起暴露了 public {@code close()} 入口，内部调用 protected
- * {@code disconnect()} 完成 {@code conn.close(1000) → executor.shutdown() →
- * 字段清零} 的全套清理。直接调用即可，无需反射。
- */
+ /** Permanently release the old SDK pool before a new client can be created. */
private void stopWebSocket() {
- cancelSilentDisconnectWatchdog();
- hasReceivedFirstEvent = false;
- if (wsThread != null) {
- wsThread.interrupt();
+ FeishuWebSocketClient previous;
+ Thread thread;
+ synchronized (wsLifecycleLock) {
+ if (transportWatchdog != null) {
+ transportWatchdog.cancel(false);
+ transportWatchdog = null;
+ }
+ previous = wsClient;
+ wsClient = null;
+ thread = wsThread;
wsThread = null;
}
- if (wsClient != null) {
- try {
- wsClient.close();
- } catch (Exception e) {
- log.warn("[feishu] WebSocket close failed: {}", e.getMessage());
- }
- }
- wsClient = null;
+ if (thread != null) thread.interrupt();
+ if (previous != null) previous.close();
}
/**
@@ -726,9 +700,8 @@ public class FeishuChannelAdapter extends AbstractChannelAdapter implements Stre
var message = eventBody.getMessage();
var sender = eventBody.getSender();
- // 任何事件到达都更新活跃时间戳，看门狗用它判断是否静默断连
+ // Record business activity without using message frequency as transport health.
touchActivity();
- hasReceivedFirstEvent = true;
// 旧事件过滤：SDK 在重连后可能回放历史事件，按 message.create_time 过滤
// 远超 stale 阈值的消息（典型场景：连接断了 5 分钟后恢复，5 分钟前的消息再处理一次没意义）
diff --git a/mateclaw-server/src/main/java/vip/mate/channel/feishu/FeishuWebSocketClient.java b/mateclaw-server/src/main/java/vip/mate/channel/feishu/FeishuWebSocketClient.java
new file mode 100644
index 00000000..8db58210
--- /dev/null
+++ b/mateclaw-server/src/main/java/vip/mate/channel/feishu/FeishuWebSocketClient.java
@@ -0,0 +1,128 @@
+package vip.mate.channel.feishu;
+
+import com.lark.oapi.okhttp.OkHttpClient;
+import com.lark.oapi.okhttp.WebSocket;
+import com.lark.oapi.ws.Client;
+
+import java.lang.reflect.Field;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * Owns one oapi-sdk 2.7.1 WebSocket client for its entire lifetime.
+ * The SDK's close() does not release its executors and exposes neither its
+ * transport nor an HTTP-client builder. Keep this version-bound compatibility
+ * bridge here; its field contract and real socket lifecycle are regression tested.
+ * Only the adapter reconnects (SDK autoReconnect must be false).
+ */
+final class FeishuWebSocketClient implements AutoCloseable {
+ private static final Field EXECUTOR = field("executor", ExecutorService.class);
+ private static final Field HTTP_CLIENT = field("httpClient", OkHttpClient.class);
+ private static final Field CONNECTION = field("conn", WebSocket.class);
+
+ private final Client client;
+ private final ExecutorService executor;
+ private final OkHttpClient httpClient;
+ private final AtomicBoolean closed = new AtomicBoolean();
+ private final Object lifecycleLock = new Object();
+ private volatile Thread starter;
+ private volatile boolean ready;
+
+ FeishuWebSocketClient(Client client) {
+ this(client, 20_000);
+ }
+
+ // Short protocol ping interval is used only by local socket tests.
+ FeishuWebSocketClient(Client client, long pingIntervalMs) {
+ this.client = client;
+ this.executor = (ExecutorService) read(EXECUTOR, client);
+ OkHttpClient original = (OkHttpClient) read(HTTP_CLIENT, client);
+ this.httpClient = original.newBuilder()
+ .pingInterval(pingIntervalMs, TimeUnit.MILLISECONDS)
+ .callTimeout(10, TimeUnit.SECONDS)
+ .build();
+ try {
+ HTTP_CLIENT.set(client, httpClient);
+ } catch (IllegalAccessException e) {
+ throw new IllegalStateException("Unsupported Feishu SDK HTTP client contract", e);
+ }
+ }
+
+ void start() throws Exception {
+ synchronized (lifecycleLock) {
+ if (closed.get()) throw new IllegalStateException("WebSocket client already closed");
+ starter = Thread.currentThread();
+ try {
+ // start() initiates an asynchronous handshake; returning is NOT a disconnect.
+ client.start();
+ client.awaitReady(10_000);
+ if (closed.get()) throw new IllegalStateException("WebSocket client closed during startup");
+ ready = true;
+ } catch (Exception e) {
+ close();
+ throw e;
+ } finally {
+ starter = null;
+ }
+ }
+ }
+
+ boolean isConnected() {
+ // SDK 2.7.1 conn is non-volatile; disconnect() clears it under this monitor.
+ synchronized (client) {
+ return ready && !closed.get() && read(CONNECTION, client) != null;
+ }
+ }
+
+ @Override
+ public void close() {
+ closed.set(true);
+ ready = false;
+ // Unblock endpoint requests/handshake before acquiring the lifecycle lock.
+ httpClient.dispatcher().cancelAll();
+ Thread thread = starter;
+ if (thread != null && thread != Thread.currentThread()) thread.interrupt();
+ synchronized (lifecycleLock) {
+ WebSocket connection = null;
+ try {
+ synchronized (client) {
+ connection = (WebSocket) read(CONNECTION, client);
+ client.close();
+ }
+ } finally {
+ try {
+ // SDK close() only enqueues a graceful close. Permanent disposal must
+ // abort the socket even when the peer never completes the handshake.
+ // Cancel outside the SDK monitor so failure callbacks can acquire it.
+ if (connection != null) connection.cancel();
+ } finally {
+ executor.shutdownNow();
+ httpClient.dispatcher().executorService().shutdownNow();
+ httpClient.connectionPool().evictAll();
+ }
+ }
+ }
+ }
+
+ private static Field field(String name, Class expectedType) {
+ try {
+ Field field = Client.class.getDeclaredField(name);
+ if (!expectedType.isAssignableFrom(field.getType())) {
+ throw new IllegalStateException("Unsupported Feishu SDK field: " + name);
+ }
+ field.setAccessible(true);
+ return field;
+ } catch (ReflectiveOperationException e) {
+ throw new IllegalStateException("Unsupported Feishu SDK lifecycle contract: " + name, e);
+ }
+ }
+
+ private static Object read(Field field, Client client) {
+ try {
+ return field.get(client);
+ } catch (IllegalAccessException e) {
+ throw new IllegalStateException("Cannot access Feishu SDK lifecycle resource", e);
+ }
+ }
+}
diff --git a/mateclaw-server/src/test/java/vip/mate/channel/ChannelHealthMonitorQuietTest.java b/mateclaw-server/src/test/java/vip/mate/channel/ChannelHealthMonitorQuietTest.java
new file mode 100644
index 00000000..61970fba
--- /dev/null
+++ b/mateclaw-server/src/test/java/vip/mate/channel/ChannelHealthMonitorQuietTest.java
@@ -0,0 +1,50 @@
+package vip.mate.channel;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.junit.jupiter.api.Test;
+import vip.mate.channel.feishu.FeishuChannelAdapter;
+import vip.mate.channel.model.ChannelEntity;
+import vip.mate.channel.weixin.WeixinChannelAdapter;
+import java.time.Duration;
+import java.util.List;
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+class ChannelHealthMonitorQuietTest {
+ private final ChannelManager manager=mock(ChannelManager.class);
+ private final ChannelHealthMonitor monitor=new ChannelHealthMonitor(manager);
+ private final ChannelMessageRouter router=mock(ChannelMessageRouter.class);
+ private final ObjectMapper json=new ObjectMapper();
+ private ChannelEntity entity() {
+ ChannelEntity e=new ChannelEntity();e.setId(42L);e.setName("health test");e.setConfigJson("{}");return e;
+ }
+ private void simulateActivityAge(AbstractChannelAdapter adapter,Duration age,AbstractChannelAdapter.ConnectionState state) {
+ adapter.getConnectionState().set(state);
+ adapter.getLastEventTimeMs().set(System.currentTimeMillis()-age.toMillis());
+ when(manager.getActiveAdapters()).thenReturn(List.of(adapter));
+ }
+ @Test void quietFeishuDoesNotTriggerHourlyRestart() {
+ var adapter=new FeishuChannelAdapter(entity(),router,json);
+ simulateActivityAge(adapter,Duration.ofHours(3),AbstractChannelAdapter.ConnectionState.CONNECTED);
+ monitor.checkHealth();verify(manager,never()).restartChannel(anyLong());
+ assertEquals(Duration.ZERO,adapter.stalenessThreshold());
+ }
+ @Test void idleExemptionStillRecoversErrorAndHonorsCooldown() {
+ var adapter=new FeishuChannelAdapter(entity(),router,json);
+ simulateActivityAge(adapter,Duration.ofMinutes(6),AbstractChannelAdapter.ConnectionState.ERROR);
+ monitor.checkHealth();monitor.checkHealth();
+ verify(manager,times(1)).restartChannel(42L);
+ }
+ @Test void idleExemptionDoesNotRestartRecentErrors() {
+ var adapter=new FeishuChannelAdapter(entity(),router,json);
+ simulateActivityAge(adapter,Duration.ofMinutes(2),AbstractChannelAdapter.ConnectionState.ERROR);
+ monitor.checkHealth();verify(manager,never()).restartChannel(anyLong());
+ }
+ @Test void longPollStalenessStillRestartsAfterFiveMinutes() {
+ // Real Weixin override, with no network startup or token configuration.
+ var adapter=mock(WeixinChannelAdapter.class,withSettings().useConstructor(entity(),router,json).defaultAnswer(CALLS_REAL_METHODS));
+ assertEquals(Duration.ofMinutes(5),adapter.stalenessThreshold());
+ simulateActivityAge(adapter,Duration.ofMinutes(6),AbstractChannelAdapter.ConnectionState.CONNECTED);
+ monitor.checkHealth();verify(manager).restartChannel(42L);
+ }
+}
diff --git a/mateclaw-server/src/test/java/vip/mate/channel/feishu/FeishuWebSocketLifecycleTest.java b/mateclaw-server/src/test/java/vip/mate/channel/feishu/FeishuWebSocketLifecycleTest.java
new file mode 100644
index 00000000..301c6238
--- /dev/null
+++ b/mateclaw-server/src/test/java/vip/mate/channel/feishu/FeishuWebSocketLifecycleTest.java
@@ -0,0 +1,84 @@
+package vip.mate.channel.feishu;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.lark.oapi.ws.Client;
+import org.junit.jupiter.api.Test;
+import vip.mate.channel.ChannelMessageRouter;
+import vip.mate.channel.model.ChannelEntity;
+import java.util.concurrent.*;
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.mock;
+
+class FeishuWebSocketLifecycleTest {
+ @Test void permanentlyClosingAdapterTerminatesSdkThreads() throws Exception {
+ ChannelEntity entity = new ChannelEntity(); entity.setId(42L); entity.setConfigJson("{}");
+ FeishuChannelAdapter adapter = new FeishuChannelAdapter(entity, mock(ChannelMessageRouter.class), new ObjectMapper());
+ Client sdk = new Client.Builder("local-test", "local-test").autoReconnect(false).build();
+ var f = Client.class.getDeclaredField("executor"); f.setAccessible(true);
+ ExecutorService executor = (ExecutorService) f.get(sdk);
+ executor.submit(() -> {}).get(2, TimeUnit.SECONDS);
+ var c = FeishuChannelAdapter.class.getDeclaredField("wsClient"); c.setAccessible(true); c.set(adapter, new FeishuWebSocketClient(sdk));
+ var stop = FeishuChannelAdapter.class.getDeclaredMethod("stopWebSocket"); stop.setAccessible(true);
+ try {
+ stop.invoke(adapter);
+ assertTrue(executor.awaitTermination(1, TimeUnit.SECONDS), "old SDK executor must terminate on permanent stop");
+ } finally { sdk.close(); executor.shutdownNow(); }
+ }
+ @Test
+ void transportStatusReadsUnderSdkMonitor() throws Exception {
+ try (var server = new LocalFeishuSocketServer()) {
+ Client sdk = new Client.Builder("local", "local")
+ .domain(server.domain()).autoReconnect(false).build();
+ try (var client = new FeishuWebSocketClient(sdk)) {
+ client.start();
+ var result = new java.util.concurrent.atomic.AtomicReference