fix(feishu): own WebSocket readiness and resource lifecycle (#659)

Contributed by RobinZhiBin <robinzhi@gmail.com> (@RobinZhiBin). Wait for handshake readiness, release transport resources, and validate liveness using protocol pings. Includes both original commits; verified by local socket and channel health regression tests.

docs/compatibility/feishu-websocket-lifecycle.md

@@ -0,0 +1,52 @@
+# Feishu WebSocket lifecycle (oapi-sdk 2.7.1)
+
+The adapter owns reconnect scheduling. SDK `autoReconnect` stays false. SDK
+`start()` initiates a handshake and returns; only `awaitReady` success makes the
+adapter CONNECTED. Startup failures and actual socket failures enter the existing
+single exponential-backoff loop. A 15-second probe observes the SDK socket;
+OkHttp protocol ping/pong (20 seconds) detects half-open transports without
+business messages. A quiet channel does not imply a broken connection.
+
+`FeishuWebSocketClient` is a deliberately narrow, version-bound compatibility
+bridge. SDK 2.7.1 has a private constructor and no public resource-disposal or
+transport-status APIs. Its `close()` does not shut down the executor. The bridge
+validates `executor`, `httpClient`, and `conn` field types. Reads of the non-volatile
+`conn` field acquire the SDK monitor used by `disconnect()`. It configures the dedicated
+HTTP client before use, and releases SDK/HTTP executors and idle connections on
+permanent close. It cancels the underlying socket after SDK close (which only
+enqueues a graceful close), outside the SDK monitor to allow failure callbacks.
+It cancels HTTP calls and interrupts the starter before taking
+the lifecycle lock, so a pending handshake cannot prevent shutdown. Do not remove
+these contract tests or upgrade the SDK without rechecking its lifecycle.
+
+Feishu opts out of message-age staleness with Duration.ZERO. Other adapters retain
+their existing thresholds.
+ERROR recovery remains active; long-poll adapters retain their existing positive
+thresholds. The legacy Feishu `silent_disconnect_threshold_seconds` setting no
+longer triggers reconnects. A transport reconnect reuses the HTTP token client;
+token refresh already has its own schedule. Permanent stop shuts down that client.
+
+Regression tests use a loopback-only HTTP/WebSocket peer and synthetic credentials,
+with no model calls or live Feishu messages. They cover delayed/rejected handshakes,
+TCP resets, peer close, missing pongs, stop during startup, close-before-start,
+close between handshake and success, and 25 real SDK start/stop cycles. Each old
+SDK and HTTP executor must terminate; quiet channels, ERROR cooldown, and Weixin
+five-minute stale recovery are verified separately.
+
+## Validation
+
+Run the lifecycle and existing channel regressions on Java 21:
+
+```sh
+JAVA_TOOL_OPTIONS=-Djdk.httpclient.allowRestrictedHeaders=connection \
+  mvn -pl mateclaw-server -am \
+  -Dtest='Feishu*Test,ChannelHealthMonitor*Test,ChannelAdapterStalenessTest' \
+  -Dsurefire.failIfNoSpecifiedTests=false test
+```
+
+The JVM option matches the application startup HTTP setting; the broader existing
+Feishu HTTP tests also run outside the application main method. The lifecycle
+tests do not need credentials or Internet access. The native SDK
+resource bridge is scoped to the pinned oapi-sdk 2.7.1; review the reflected field
+contracts before changing that dependency. Local socket tests do not establish
+real-user delivery or multi-day reliability.

mateclaw-server/src/main/java/vip/mate/channel/AbstractChannelAdapter.java

@@ -159,6 +159,7 @@ public abstract class AbstractChannelAdapter implements ChannelAdapter {
      * 重连成功回调
      */
     protected void onReconnectSuccess() {
+        if (!running.get()) return;
         backoff.reset();
         connectionState.set(ConnectionState.CONNECTED);
         lastEventTimeMs.set(System.currentTimeMillis());
@@ -171,6 +172,7 @@ public abstract class AbstractChannelAdapter implements ChannelAdapter {
      * 重连失败回调
      */
     protected void onReconnectFailed(Exception e) {
+        if (!running.get()) return;
         lastError = e.getMessage();
         log.warn("[{}] Reconnect failed for {}: {} (attempt #{})",
                 getChannelType(), channelEntity.getName(), e.getMessage(), backoff.getAttempts());
@@ -182,11 +184,14 @@ public abstract class AbstractChannelAdapter implements ChannelAdapter {
         if (running.compareAndSet(false, true)) {
             log.info("[{}] Starting channel: {}", getChannelType(), channelEntity.getName());
             try {
+                connectionState.set(ConnectionState.RECONNECTING);
                 doStart();
-                connectionState.set(ConnectionState.CONNECTED);
-                lastEventTimeMs.set(System.currentTimeMillis());
-                lastError = null;
-                backoff.reset();
+                if (!connectsAsynchronously()) {
+                    connectionState.set(ConnectionState.CONNECTED);
+                    lastEventTimeMs.set(System.currentTimeMillis());
+                    lastError = null;
+                    backoff.reset();
+                }
                 log.info("[{}] Channel started successfully: {}", getChannelType(), channelEntity.getName());
             } catch (Exception e) {
                 running.set(false);
@@ -198,6 +203,11 @@ public abstract class AbstractChannelAdapter implements ChannelAdapter {
         }
     }
 
+    /** Async transports report CONNECTED only after their handshake completes. */
+    protected boolean connectsAsynchronously() {
+        return false;
+    }
+
     @Override
     public void stop() {
         if (running.compareAndSet(true, false)) {

mateclaw-server/src/main/java/vip/mate/channel/ChannelAdapter.java

@@ -248,6 +248,8 @@ public interface ChannelAdapter {
     /**
      * RFC-024 Change 2：本 adapter 认为"多久没活动就视作 stale 需要重启"的阈值。
      *
+     * <p>返回 Duration.ZERO 可禁用消息静默检查（无状态 HTTP 或自带传输心跳的渠道）。
+     * ERROR 状态的恢复不受影响。
      * <p>通用默认 60 分钟；长轮询类渠道（如 iLink 微信）应覆盖为 5 分钟，
      * 这样代理/NAT 侧 2–5 分钟 idle-close 把连接切断后，
      * {@code ChannelHealthMonitor} 能在几分钟内触发自动重启，而非静默等整小时。</p>

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

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();
-    }
-
-    /**
-     * 启动静默断连看门狗。
-     * <p>
-     * SDK 内部有 ping/pong 心跳，正常情况下连接断开会触发 start() 返回 → onDisconnected。
-     * 但少数场景 TCP 层认为连接还在但事件不再流入（NAT 超时、半开连接、对端 hang 死），
-     * SDK 也察觉不到。看门狗每 60s 检查一次「最近事件时间」，超过阈值就强制重连。
-     * <p>
-     * 静默判定有两个前置条件：
-     * 1. 已经收到过至少一个事件（避免新连接立即触发误报）
-     * 2. silent_disconnect_threshold_seconds &gt; 0（设为 0 可禁用看门狗）
-     * <p>
-     * 默认阈值 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 连接。
-     * <p>
-     * 必须主动关闭底层 WebSocket 连接并触发 SDK 的内部清理（停止 pingLoop、
-     * 释放 ExecutorService）。仅置空引用会导致旧连接的 pingLoop 和线程池
-     * 持续运行，造成文件描述符和线程泄漏，最终使新连接无法建立。
-     * <p>
-     * 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 分钟前的消息再处理一次没意义）

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);
+        }
+    }
+}

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);
+    }
+}

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<Boolean>();
+                Thread observer = new Thread(() -> result.set(client.isConnected()), "transport-status-reader");
+                try {
+                    synchronized (sdk) {
+                        observer.start();
+                        long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(3);
+                        while (observer.getState() != Thread.State.BLOCKED
+                                && observer.getState() != Thread.State.TERMINATED
+                                && System.nanoTime() < deadline) {
+                            Thread.sleep(5);
+                        }
+                        assertEquals(Thread.State.BLOCKED, observer.getState(),
+                                "conn is non-volatile: reads must acquire the monitor used by SDK disconnect()");
+                        assertNull(result.get(), "status read must not bypass the SDK monitor");
+                    }
+                } finally {
+                    observer.join(3_000);
+                }
+                assertFalse(observer.isAlive());
+                assertEquals(Boolean.TRUE, result.get());
+            }
+        }
+    }
+    @Test
+    void permanentCloseCancelsTransportWithoutWaitingForPeerClose() throws Exception {
+        Client sdk = new Client.Builder("local", "local").autoReconnect(false).build();
+        var socket = mock(com.lark.oapi.okhttp.WebSocket.class);
+        var connection = Client.class.getDeclaredField("conn");
+        connection.setAccessible(true);
+        connection.set(sdk, socket);
+        try (var client = new FeishuWebSocketClient(sdk)) {
+            client.close();
+            org.mockito.Mockito.verify(socket).cancel();
+        }
+    }
+    @Test
+    void sdkCloseFailureStillCancelsSocketAndReleasesExecutors() throws Exception {
+        Client sdk = new Client.Builder("local", "local").autoReconnect(false).build();
+        var socket = mock(com.lark.oapi.okhttp.WebSocket.class);
+        org.mockito.Mockito.when(socket.close(org.mockito.ArgumentMatchers.anyInt(),
+                org.mockito.ArgumentMatchers.anyString())).thenThrow(new IllegalStateException("close failed"));
+        var connection = Client.class.getDeclaredField("conn");
+        connection.setAccessible(true);
+        connection.set(sdk, socket);
+        try (var client = new FeishuWebSocketClient(sdk)) {
+            assertThrows(IllegalStateException.class, client::close);
+            org.mockito.Mockito.verify(socket).cancel();
+            FeishuWebSocketTransportTest.assertReleased(sdk);
+        }
+    }
+}

mateclaw-server/src/test/java/vip/mate/channel/feishu/FeishuWebSocketTransportTest.java

@@ -0,0 +1,171 @@
+package vip.mate.channel.feishu;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.lark.oapi.okhttp.OkHttpClient;
+import com.lark.oapi.ws.Client;
+import org.junit.jupiter.api.Test;
+import vip.mate.channel.AbstractChannelAdapter.ConnectionState;
+import vip.mate.channel.ChannelMessageRouter;
+import vip.mate.channel.ExponentialBackoff;
+import vip.mate.channel.model.ChannelEntity;
+
+import java.net.Socket;
+import java.time.Duration;
+import java.util.*;
+import java.util.concurrent.*;
+import java.util.function.BooleanSupplier;
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.mock;
+
+class FeishuWebSocketTransportTest {
+    static Object field(Object target, String name) throws Exception {
+        var f=Client.class.getDeclaredField(name);f.setAccessible(true);return f.get(target);
+    }
+    static void await(BooleanSupplier predicate) throws Exception {
+        long end=System.nanoTime()+TimeUnit.SECONDS.toNanos(4);
+        while(!predicate.getAsBoolean() && System.nanoTime()<end) Thread.sleep(10);
+        assertTrue(predicate.getAsBoolean(),"condition did not become true");
+    }
+    static class TestAdapter extends FeishuChannelAdapter {
+        final LocalFeishuSocketServer server;
+        final List<Client> sdks=new CopyOnWriteArrayList<>();
+        TestAdapter(LocalFeishuSocketServer server) {
+            super(entity(),mock(ChannelMessageRouter.class),new ObjectMapper());this.server=server;
+            backoff=new ExponentialBackoff(50,100,2,-1);
+        }
+        static ChannelEntity entity() {
+            ChannelEntity e=new ChannelEntity();e.setId(73L);e.setName("local transport");
+            e.setConfigJson("{\"app_id\":\"local\",\"app_secret\":\"local\"}");return e;
+        }
+        @Override protected void doStart() { startWebSocket("local","local"); }
+        @Override FeishuWebSocketClient createWsClient(String id,String secret) {
+            Client sdk=new Client.Builder(id,secret).domain(server.domain()).autoReconnect(false).build();
+            sdks.add(sdk);return new FeishuWebSocketClient(sdk,100);
+        }
+    }
+    static void assertReleased(Client sdk) throws Exception {
+        ExecutorService pool=(ExecutorService)field(sdk,"executor");
+        OkHttpClient http=(OkHttpClient)field(sdk,"httpClient");
+        assertTrue(pool.awaitTermination(3,TimeUnit.SECONDS));
+        assertTrue(http.dispatcher().executorService().awaitTermination(3,TimeUnit.SECONDS));
+        assertEquals(0,http.connectionPool().connectionCount());
+    }
+    @Test void delayedHandshakeStaysReconnectingThenUsesOnlyOneClient() throws Exception {
+        try(var server=new LocalFeishuSocketServer()) {
+            server.handshakeDelayMs=250;
+            var adapter=new TestAdapter(server);
+            try {
+                adapter.start();assertEquals(ConnectionState.RECONNECTING,adapter.getConnectionState().get());
+                await(()->adapter.getConnectionState().get()==ConnectionState.CONNECTED);
+                adapter.getLastEventTimeMs().set(System.currentTimeMillis()-Duration.ofHours(3).toMillis());
+                for(int i=0;i<10;i++) adapter.checkWebSocketTransport();
+                assertEquals(1,adapter.sdks.size());assertEquals(1,server.handshakes.get());
+            } finally { adapter.stop(); }
+            assertReleased(adapter.sdks.getFirst());
+        }
+    }
+    @Test void closeBetweenHandshakeAndSuccessDoesNotGetStuckReconnecting() throws Exception {
+        try(var server=new LocalFeishuSocketServer()) {
+            var successReached=new CountDownLatch(1);
+            var allowSuccess=new CountDownLatch(1);
+            var adapter=new TestAdapter(server) {
+                @Override protected void onReconnectSuccess() {
+                    successReached.countDown();
+                    try { assertTrue(allowSuccess.await(3,TimeUnit.SECONDS)); }
+                    catch(InterruptedException e) {Thread.currentThread().interrupt();throw new IllegalStateException(e);}
+                    super.onReconnectSuccess();
+                }
+            };
+            try {
+                adapter.start();assertTrue(successReached.await(3,TimeUnit.SECONDS));
+                server.awaitConnection().close();
+                await(()->{try{return field(adapter.sdks.getFirst(),"conn")==null;}catch(Exception e){throw new RuntimeException(e);}});
+                allowSuccess.countDown();
+                await(()->adapter.sdks.size()==2 && adapter.getConnectionState().get()==ConnectionState.CONNECTED);
+                assertReleased(adapter.sdks.getFirst());
+            } finally { allowSuccess.countDown();adapter.stop(); }
+        }
+    }
+    @Test void tcpAbortReconnectsOnceAndReleasesOldClient() throws Exception { reconnect(false); }
+    @Test void peerCloseReconnectsOnceAndReleasesOldClient() throws Exception { reconnect(true); }
+    private void reconnect(boolean graceful) throws Exception {
+        try(var server=new LocalFeishuSocketServer()) {
+            var adapter=new TestAdapter(server);
+            try {
+                adapter.start();await(()->adapter.getConnectionState().get()==ConnectionState.CONNECTED);
+                Socket old=server.awaitConnection();
+                if(graceful) LocalFeishuSocketServer.frame(old,8,new byte[]{3,(byte)232});
+                else old.setSoLinger(true,0);
+                old.close();
+                await(()->{adapter.checkWebSocketTransport();return adapter.sdks.size()==2 && adapter.getConnectionState().get()==ConnectionState.CONNECTED;});
+                assertEquals(2,server.handshakes.get());assertReleased(adapter.sdks.getFirst());
+            } finally { adapter.stop(); }
+            for(Client sdk:adapter.sdks) assertReleased(sdk);
+        }
+    }
+    @Test void protocolPingDetectsHalfOpenSocketWithoutBusinessMessages() throws Exception {
+        try(var server=new LocalFeishuSocketServer()) {
+            var adapter=new TestAdapter(server);
+            try {
+                adapter.start();await(()->adapter.getConnectionState().get()==ConnectionState.CONNECTED);
+                server.replyToPing=false;
+                await(()->{adapter.checkWebSocketTransport();return adapter.sdks.size()>=2;});
+                server.replyToPing=true;
+                await(()->adapter.getConnectionState().get()==ConnectionState.CONNECTED);
+                assertReleased(adapter.sdks.getFirst());
+            } finally { adapter.stop(); }
+        }
+    }
+    @Test void rejectedHandshakeIsRetriedWithoutLeakingTheFailedClient() throws Exception {
+        try(var server=new LocalFeishuSocketServer()) {
+            server.rejectHandshake=true;
+            var adapter=new TestAdapter(server);
+            try {
+                adapter.start();await(()->adapter.sdks.size()>=2);
+                server.rejectHandshake=false;
+                await(()->adapter.getConnectionState().get()==ConnectionState.CONNECTED);
+                assertReleased(adapter.sdks.getFirst());
+            } finally { adapter.stop(); }
+            for(Client sdk:adapter.sdks) assertReleased(sdk);
+        }
+    }
+    @Test void repeatedRealStartStopDoesNotAccumulateExecutorThreads() throws Exception {
+        try(var server=new LocalFeishuSocketServer()) {
+            for(int i=0;i<25;i++) {
+                var adapter=new TestAdapter(server);
+                try { adapter.start();await(()->adapter.getConnectionState().get()==ConnectionState.CONNECTED); }
+                finally { adapter.stop(); }
+                assertEquals(1,adapter.sdks.size());assertReleased(adapter.sdks.getFirst());
+            }
+            assertEquals(25,server.handshakes.get());
+        }
+    }
+    @Test void stopDuringDelayedHandshakeReturnsWithinManagerDeadline() throws Exception {
+        try(var server=new LocalFeishuSocketServer()) {
+            server.handshakeDelayMs=2_000;
+            var adapter=new TestAdapter(server);adapter.start();
+            assertTrue(server.endpointRequested.await(2,TimeUnit.SECONDS));
+            assertTimeoutPreemptively(Duration.ofSeconds(3),adapter::stop);
+            Thread.sleep(100);
+            assertFalse(adapter.isRunning());assertEquals(ConnectionState.DISCONNECTED,adapter.getConnectionState().get());
+            for(Client sdk:adapter.sdks) assertReleased(sdk);
+        }
+    }
+    @Test void stopCancelsStalledEndpointRequest() throws Exception {
+        try(var server=new LocalFeishuSocketServer()) {
+            server.stallEndpoint=true;
+            var adapter=new TestAdapter(server);adapter.start();
+            assertTrue(server.endpointRequested.await(2,TimeUnit.SECONDS));
+            assertTimeoutPreemptively(Duration.ofSeconds(3),adapter::stop);
+            assertFalse(adapter.isRunning());for(Client sdk:adapter.sdks) assertReleased(sdk);
+        }
+    }
+    @Test void closeBeforeStartCannotReopenSdkOrCreateThreads() throws Exception {
+        try(var server=new LocalFeishuSocketServer()) {
+            Client sdk=new Client.Builder("local","local").domain(server.domain()).autoReconnect(false).build();
+            var client=new FeishuWebSocketClient(sdk);client.close();client.close();
+            assertThrows(IllegalStateException.class,client::start);
+            assertEquals(0,server.handshakes.get());assertReleased(sdk);
+        }
+    }
+}

mateclaw-server/src/test/java/vip/mate/channel/feishu/LocalFeishuSocketServer.java

@@ -0,0 +1,96 @@
+package vip.mate.channel.feishu;
+
+import java.io.*;
+import java.net.*;
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import java.util.*;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/** Loopback-only HTTP endpoint + RFC6455 peer; never contacts Feishu or a model. */
+final class LocalFeishuSocketServer implements AutoCloseable {
+    final ServerSocket listener = new ServerSocket(0, 50, InetAddress.getLoopbackAddress());
+    final Set<Socket> sockets = ConcurrentHashMap.newKeySet();
+    final BlockingQueue<Socket> connections = new LinkedBlockingQueue<>();
+    final AtomicInteger handshakes = new AtomicInteger();
+    final ExecutorService threads = Executors.newCachedThreadPool();
+    volatile long handshakeDelayMs;
+    volatile boolean rejectHandshake;
+    volatile boolean replyToPing = true;
+    volatile boolean stallEndpoint;
+    final CountDownLatch endpointRequested = new CountDownLatch(1);
+
+    LocalFeishuSocketServer() throws IOException {
+        threads.submit(() -> {
+            try {
+                while (!listener.isClosed()) {
+                    Socket socket = listener.accept(); sockets.add(socket);
+                    threads.submit(() -> serve(socket));
+                }
+            } catch (IOException ignored) { }
+        });
+    }
+    String domain() { return "http://127.0.0.1:" + listener.getLocalPort(); }
+
+    private void serve(Socket socket) {
+        try (socket) {
+            InputStream in = socket.getInputStream();
+            ByteArrayOutputStream header = new ByteArrayOutputStream();
+            while (!header.toString(StandardCharsets.US_ASCII).endsWith("\r\n\r\n")) {
+                int next = in.read(); if (next < 0) return; header.write(next);
+                if (header.size() > 16_384) throw new IOException("oversize request");
+            }
+            String headers = header.toString(StandardCharsets.US_ASCII);
+            if (headers.startsWith("POST ")) {
+                endpointRequested.countDown();
+                if (stallEndpoint) { while (in.read() >= 0) {} return; }
+                String body = "{\"code\":0,\"data\":{\"URL\":\"ws://127.0.0.1:" + listener.getLocalPort()
+                        + "/socket?device_id=local&service_id=1\",\"ClientConfig\":{\"ReconnectCount\":0,\"ReconnectInterval\":0,\"ReconnectNonce\":0,\"PingInterval\":30}}}";
+                write(socket, "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: "
+                        + body.getBytes(StandardCharsets.UTF_8).length + "\r\nConnection: close\r\n\r\n" + body);
+                return;
+            }
+            if (handshakeDelayMs > 0) Thread.sleep(handshakeDelayMs);
+            if (rejectHandshake) { write(socket, "HTTP/1.1 503 Unavailable\r\nContent-Length: 0\r\nConnection: close\r\n\r\n"); return; }
+            String key = Arrays.stream(headers.split("\r\n")).filter(s -> s.toLowerCase(Locale.ROOT).startsWith("sec-websocket-key:"))
+                    .findFirst().orElseThrow().split(":",2)[1].trim();
+            String accept = Base64.getEncoder().encodeToString(MessageDigest.getInstance("SHA-1")
+                    .digest((key + "258EAFA5-E914-47DA-95CA-C5AB0DC85B11").getBytes(StandardCharsets.US_ASCII)));
+            write(socket,"HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: " + accept + "\r\n\r\n");
+            handshakes.incrementAndGet(); connections.add(socket);
+            while (true) {
+                int first=in.read(); if(first<0) return;
+                int second=in.read(); if(second<0) return;
+                int length=second&127;
+                if(length==126) length=(in.read()<<8)|in.read();
+                if(length==127) throw new IOException("unexpected test frame length");
+                byte[] mask=(second&128)!=0 ? in.readNBytes(4) : new byte[0];
+                byte[] payload=in.readNBytes(length);
+                for(int i=0;i<payload.length && mask.length==4;i++) payload[i]^=mask[i%4];
+                int opcode=first&15;
+                if(opcode==8) { frame(socket,8,payload);return; }
+                if(opcode==9 && replyToPing) frame(socket,10,payload);
+            }
+        } catch (Exception ignored) {
+            // Tests assert observed handshake/readiness/failure. Peer close is expected.
+        } finally { sockets.remove(socket); }
+    }
+    Socket awaitConnection() throws InterruptedException {
+        Socket socket=connections.poll(3,TimeUnit.SECONDS);
+        if(socket==null) throw new AssertionError("No WebSocket handshake");
+        return socket;
+    }
+    static void frame(Socket socket,int opcode,byte[] payload) throws IOException {
+        synchronized(socket) {
+            OutputStream out=socket.getOutputStream();out.write(128|opcode);out.write(payload.length);out.write(payload);out.flush();
+        }
+    }
+    private static void write(Socket socket,String text) throws IOException {
+        socket.getOutputStream().write(text.getBytes(StandardCharsets.UTF_8));socket.getOutputStream().flush();
+    }
+    @Override public void close() throws Exception {
+        listener.close();for(Socket s:sockets) s.close();threads.shutdownNow();
+        if(!threads.awaitTermination(3,TimeUnit.SECONDS)) throw new AssertionError("Fixture threads did not stop");
+    }
+}