From 78f6ca8d3332bde6fc5aa087fb37af510f954949 Mon Sep 17 00:00:00 2001 From: matevip Date: Sun, 10 May 2026 08:27:48 +0800 Subject: [PATCH] feat(channel): reply queue foundation + lifecycle gate --- .../mate/channel/AbstractChannelAdapter.java | 21 + .../java/vip/mate/channel/ChannelAdapter.java | 47 +++ .../java/vip/mate/channel/ChannelManager.java | 14 +- .../java/vip/mate/channel/SendContext.java | 63 +++ .../ApprovalNotificationService.java | 37 +- .../channel/wecom/WeComChannelAdapter.java | 376 ++++++++++++++++-- 6 files changed, 522 insertions(+), 36 deletions(-) create mode 100644 mateclaw-server/src/main/java/vip/mate/channel/SendContext.java diff --git a/mateclaw-server/src/main/java/vip/mate/channel/AbstractChannelAdapter.java b/mateclaw-server/src/main/java/vip/mate/channel/AbstractChannelAdapter.java index fbbb3e78..52ea9cd4 100644 --- a/mateclaw-server/src/main/java/vip/mate/channel/AbstractChannelAdapter.java +++ b/mateclaw-server/src/main/java/vip/mate/channel/AbstractChannelAdapter.java @@ -319,6 +319,27 @@ public abstract class AbstractChannelAdapter implements ChannelAdapter { } } + /** + * Approval notice rendering — primary implementation position. + * + *

Subclasses that support a native card surface (WeCom + * {@code button_interaction}, DingTalk {@code ActionCard}, etc.) + * override this method and may call + * {@code super.sendApprovalNotice(...)} to fall back to the text + * path on render failure / payload-too-large / etc. + * + *

Lives on the abstract class rather than as an interface + * default method so the {@code super.x(...)} call from subclasses + * resolves cleanly via Java's normal class inheritance — see + * RFC-32 §2.0.4 (C-4 fix). + */ + @Override + public void sendApprovalNotice(String targetId, + vip.mate.channel.notification.ApprovalNotice notice) { + sendMessage(targetId, + vip.mate.channel.notification.ApprovalNotificationService.staticBuildText(notice)); + } + // ==================== 模板方法（子类实现） ==================== /** diff --git a/mateclaw-server/src/main/java/vip/mate/channel/ChannelAdapter.java b/mateclaw-server/src/main/java/vip/mate/channel/ChannelAdapter.java index 73f62d1b..fcf6b292 100644 --- a/mateclaw-server/src/main/java/vip/mate/channel/ChannelAdapter.java +++ b/mateclaw-server/src/main/java/vip/mate/channel/ChannelAdapter.java @@ -95,6 +95,53 @@ public interface ChannelAdapter { sendMessage(targetId, content); } + /** + * Extended render-and-send overload that carries an optional + * {@link SendContext} side-channel (e.g. WeCom AI Bot + * {@code feedback.id} for like/dislike collection). + * + *

Default implementation ignores {@code ctx} and falls back to + * {@link #renderAndSend(String, String)}, so existing channel + * adapters and callers see no behavior change. Channels that want + * to consume {@code SendContext} fields override this overload. + * + *

This was introduced as part of PR-0 (RFC-32 §2.0.3) to give + * {@code ChannelMessageRouter} a way to thread the pre-allocated + * feedback id (registered against the persisted + * {@code mate_message.id}) down to the WeCom adapter without + * widening the legacy two-arg signature. + */ + default void renderAndSend(String targetId, String content, SendContext ctx) { + renderAndSend(targetId, content); + } + + /** + * Render and deliver an approval notice. Channels that support a + * native interactive surface (WeCom {@code button_interaction}, + * DingTalk {@code ActionCard}, etc.) override this to skip the + * text path entirely. + * + *

Primary implementation lives on + * {@link AbstractChannelAdapter}, which keeps the bytewise + * fallback (markdown text → {@link #sendMessage}). Adapters that + * inherit from {@code AbstractChannelAdapter} can call + * {@code super.sendApprovalNotice(...)} to fall back; the default + * here is just a safety net for adapters that, for some reason, + * implement {@link ChannelAdapter} directly. + * + *

Introduced in PR-0 (RFC-32 §2.0.3) so the router does not + * need to know which channel renders cards vs text: + * 

+     *     ApprovalNotice notice = approvalNotificationService.buildNotice(pending);
+     *     adapter.sendApprovalNotice(replyTarget, notice);
+     *
+ */ + default void sendApprovalNotice(String targetId, + vip.mate.channel.notification.ApprovalNotice notice) { + sendMessage(targetId, + vip.mate.channel.notification.ApprovalNotificationService.staticBuildText(notice)); + } + // ==================== 主动推送 ==================== /** diff --git a/mateclaw-server/src/main/java/vip/mate/channel/ChannelManager.java b/mateclaw-server/src/main/java/vip/mate/channel/ChannelManager.java index b7805453..3440dc0d 100644 --- a/mateclaw-server/src/main/java/vip/mate/channel/ChannelManager.java +++ b/mateclaw-server/src/main/java/vip/mate/channel/ChannelManager.java @@ -46,6 +46,17 @@ public class ChannelManager { private final ObjectMapper objectMapper; private final vip.mate.tool.document.GeneratedFileCache generatedFileCache; + /** + * Approval notification renderer — used by WeCom adapter (PR-0 + * threading; PR-1 will switch the WeCom override to render a + * {@code button_interaction} card via this service's card builder). + * Other adapters keep using the text path on + * {@link AbstractChannelAdapter}, which calls + * {@code ApprovalNotificationService.staticBuildText} so this + * field is currently consumed only by WeCom. + */ + private final vip.mate.channel.notification.ApprovalNotificationService approvalNotificationService; + /** 运行中的渠道适配器：channelId -> adapter */ private final Map activeAdapters = new HashMap<>(); @@ -455,7 +466,8 @@ public class ChannelManager { case "feishu" -> new FeishuChannelAdapter(channel, messageRouter, objectMapper); case "telegram" -> new TelegramChannelAdapter(channel, messageRouter, objectMapper); case "discord" -> new DiscordChannelAdapter(channel, messageRouter, objectMapper); - case "wecom" -> new WeComChannelAdapter(channel, messageRouter, objectMapper); + case "wecom" -> new WeComChannelAdapter(channel, messageRouter, objectMapper, + approvalNotificationService); case "qq" -> new QQChannelAdapter(channel, messageRouter, objectMapper); case "weixin" -> new WeixinChannelAdapter(channel, messageRouter, objectMapper); case "slack" -> new vip.mate.channel.slack.SlackChannelAdapter(channel, messageRouter, objectMapper); diff --git a/mateclaw-server/src/main/java/vip/mate/channel/SendContext.java b/mateclaw-server/src/main/java/vip/mate/channel/SendContext.java new file mode 100644 index 00000000..d72b9ad8 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/channel/SendContext.java @@ -0,0 +1,63 @@ +package vip.mate.channel; + +import java.util.Map; + +/** + * Per-send context used to thread optional metadata from + * {@link ChannelMessageRouter} down into channel-specific renderers + * without having to expand {@code renderAndSend} signatures every time + * a new channel needs a side-channel value. + * + *

Currently used by: + *

+ * + *

Adapters that don't need any of this can ignore the parameter: + * the default {@code renderAndSend(targetId, content, ctx)} on + * {@link ChannelAdapter} delegates to the legacy two-arg version and + * drops {@code ctx} entirely. + * + * @param feedbackId optional like/dislike correlation id; null if + * the channel does not collect feedback or the + * assistant message could not be persisted + * @param savedMessageId persisted {@code mate_message.id} for the + * assistant reply; null in error / streaming + * passthrough paths where no row was created + * @param extra open-ended map; must never be null — use + * {@link #empty()} if no extras + */ +public record SendContext( + String feedbackId, + Long savedMessageId, + Map extra +) { + + public SendContext { + // Defensive: a null extra map breaks downstream get-or-default lookups. + if (extra == null) { + extra = Map.of(); + } + } + + private static final SendContext EMPTY = new SendContext(null, null, Map.of()); + + /** + * Reusable empty context. Adapters and callers that have no + * side-channel hints to thread should use this rather than + * allocating a fresh empty record per message. + */ + public static SendContext empty() { + return EMPTY; + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/channel/notification/ApprovalNotificationService.java b/mateclaw-server/src/main/java/vip/mate/channel/notification/ApprovalNotificationService.java index 085a46e0..36e83d77 100644 --- a/mateclaw-server/src/main/java/vip/mate/channel/notification/ApprovalNotificationService.java +++ b/mateclaw-server/src/main/java/vip/mate/channel/notification/ApprovalNotificationService.java @@ -57,29 +57,42 @@ public class ApprovalNotificationService { } /** - * 从 ApprovalNotice 构建文本 + * 从 ApprovalNotice 构建文本（实例方法，保留供历史调用方使用） */ public String buildApprovalText(ApprovalNotice notice) { + return staticBuildText(notice); + } + + /** + * Static text renderer — used by the {@code AbstractChannelAdapter} + * default {@code sendApprovalNotice} implementation, which has no + * Spring-managed reference to the service instance. + * + *

Logic is identical to {@link #buildApprovalText(ApprovalNotice)}; + * the instance method delegates here so the two paths can never + * drift. + */ + public static String staticBuildText(ApprovalNotice notice) { StringBuilder sb = new StringBuilder(); sb.append("🔐 **工具需要审批**\n\n"); sb.append("**工具名称**: ").append(notice.toolName()).append("\n"); - // 风险等级 + // Risk severity if (notice.maxSeverity() != null) { - sb.append("**风险等级**: ").append(severityLabel(notice.maxSeverity())).append("\n"); + sb.append("**风险等级**: ").append(staticSeverityLabel(notice.maxSeverity())).append("\n"); } - // 摘要 + // Summary if (notice.summary() != null && !notice.summary().isEmpty()) { sb.append("**摘要**: ").append(notice.summary()).append("\n"); } - // 参数预览 + // Args preview if (notice.argumentsPreview() != null && !notice.argumentsPreview().isEmpty()) { sb.append("**参数**: `").append(notice.argumentsPreview()).append("`\n"); } - // Findings 摘要（最多显示 3 条） + // Findings (top 3) if (notice.findings() != null && !notice.findings().isEmpty()) { sb.append("\n**发现的问题**:\n"); int shown = 0; @@ -100,6 +113,18 @@ public class ApprovalNotificationService { return sb.toString(); } + private static String staticSeverityLabel(String severity) { + if (severity == null) return ""; + return switch (severity) { + case "CRITICAL" -> "🔴 CRITICAL"; + case "HIGH" -> "🟠 HIGH"; + case "MEDIUM" -> "🟡 MEDIUM"; + case "LOW" -> "🔵 LOW"; + case "INFO" -> "⚪ INFO"; + default -> severity; + }; + } + /** * 构建 Web SSE 事件数据 */ diff --git a/mateclaw-server/src/main/java/vip/mate/channel/wecom/WeComChannelAdapter.java b/mateclaw-server/src/main/java/vip/mate/channel/wecom/WeComChannelAdapter.java index 2aefe81f..1395bfb9 100644 --- a/mateclaw-server/src/main/java/vip/mate/channel/wecom/WeComChannelAdapter.java +++ b/mateclaw-server/src/main/java/vip/mate/channel/wecom/WeComChannelAdapter.java @@ -114,18 +114,54 @@ public class WeComChannelAdapter extends AbstractChannelAdapter { /** 消息去重集合 */ private final Set processedMessageIds = ConcurrentHashMap.newKeySet(); - /** 回复 ACK 等待：reqId -> CompletableFuture */ + /** 回复 ACK 等待：reqId -> CompletableFuture（在 reqIdWorker 串行内 put，避免同 reqId 多次发送时撞 key） */ private final ConcurrentHashMap>> pendingAcks = new ConcurrentHashMap<>(); - /** 回复队列：reqId -> 串行队列（保证同一 reqId 的回复按序发送） */ - private final ConcurrentHashMap> replyQueues = new ConcurrentHashMap<>(); + /** + * Per-reqId 串行回复队列状态。Key=reqId，value 是 {@link ReplyQueueState} + * 包装的 (queue, closed) 二元组，用来在 idle-close 与 late-offer 之间提供 + * compute-bin-lock 级原子化（RFC-32 §2.4.1 a-2 / R-5 修正）。 + * + *

{@code closed} 是防御性标志位：worker 在 idle compute 退出时会把 entry + * 从 map 删掉，所以正常路径上 enqueue 看不到一个"closed=true 还在 map 里"的 + * state；保留这个标志为未来重构兜底，避免任何破坏"close = remove entry" + * 耦合的改动让 silent drop 复活。 + */ + private final ConcurrentHashMap replyQueues = new ConcurrentHashMap<>(); - /** 回复队列处理线程池 */ - private final ExecutorService replyExecutor = Executors.newCachedThreadPool(r -> { - Thread t = new Thread(r, "wecom-reply"); - t.setDaemon(true); - return t; - }); + /** + * 回复队列 worker 池。volatile + 非 final 是 RFC-32 §2.4.1 a-1 / R-2 修正 + * 的一部分：stop/重连时需要 {@link ExecutorService#shutdownNow()} 中断 + * worker 阻塞中的 {@code queue.poll(60s)}，但 cached pool 一旦 shutdown + * 就不能重用，所以必须能在 {@link #ensureReplyExecutor()} 里重建。 + */ + private volatile ExecutorService replyExecutor; + + /** + * Lifecycle gate：控制 {@link #sendFrameWithAck} 是否接受新任务。 + * + *

必须由 transport-ready 信号 触发置 true（即认证成功后的 + * {@link #markReady()}），而不是 executor-ready（{@link #ensureReplyExecutor()}）。 + * 否则会出现"executor 活的、accepting=true、但 webSocket=null"的窗口—— + * worker 调 {@link #sendFrame} 看到 {@code webSocket==null} 就 warn 后默默 return， + * 让 caller 等 5s 假超时（RFC-32 §2.4.1 a-1 / R-7 修正）。 + */ + private final AtomicBoolean replyQueueAccepting = new AtomicBoolean(false); + + /** + * Per-reqId 回复队列状态。 + * + * @param queue 串行回复任务队列 + * @param closed 防御性 closed 标志（详见 {@link #replyQueues} 注释） + */ + private record ReplyQueueState( + LinkedBlockingQueue queue, + AtomicBoolean closed + ) { + static ReplyQueueState fresh() { + return new ReplyQueueState(new LinkedBlockingQueue<>(), new AtomicBoolean(false)); + } + } /** WebSocket 消息碎片缓冲区 */ private final StringBuilder wsBuffer = new StringBuilder(); @@ -153,10 +189,26 @@ public class WeComChannelAdapter extends AbstractChannelAdapter { */ private final AtomicBoolean disconnectInflight = new AtomicBoolean(false); + /** + * Optional approval-notification renderer. + * + *

Wired in PR-0 for forward compatibility. PR-1 will use this to + * build {@code button_interaction} card payloads for tool-guard + * approvals; PR-0 keeps the field but does not consume it (the + * default text path on {@link AbstractChannelAdapter#sendApprovalNotice} + * still handles approvals). {@code null}-tolerant: if Spring DI + * cannot find the bean (unit-test contexts, hot-swap edge cases, + * etc.) the adapter still functions in PR-0 fashion. + */ + @SuppressWarnings("unused") // consumed in PR-1 + private final vip.mate.channel.notification.ApprovalNotificationService approvalNotificationService; + public WeComChannelAdapter(ChannelEntity channelEntity, ChannelMessageRouter messageRouter, - ObjectMapper objectMapper) { + ObjectMapper objectMapper, + vip.mate.channel.notification.ApprovalNotificationService approvalNotificationService) { super(channelEntity, messageRouter, objectMapper); + this.approvalNotificationService = approvalNotificationService; // Default to 8 bounded attempts (~4 minutes total at 2s..30s exponential) // so the UI eventually settles in ERROR instead of getting stuck in // RECONNECTING forever. User config still overrides (-1 = infinite). @@ -185,6 +237,11 @@ public class WeComChannelAdapter extends AbstractChannelAdapter { .connectTimeout(Duration.ofSeconds(10)) .build(); + // Build the reply-queue worker pool BEFORE the WS handshake kicks off, so + // any inbound auth_succeed → markReady → openReplyQueue path finds a live + // executor to schedule against. The gate stays closed until markReady runs. + ensureReplyExecutor(); + connectWebSocket(botId, secret); log.info("[wecom] WeCom bot channel initialized: botId={}, maxReconnectAttempts={}", @@ -211,6 +268,11 @@ public class WeComChannelAdapter extends AbstractChannelAdapter { .connectTimeout(Duration.ofSeconds(10)) .build(); + // Re-arm the reply-queue worker pool BEFORE attempting the new + // handshake. accepting flag stays false until the new connection's + // auth_succeed fires markReady → openReplyQueue. + ensureReplyExecutor(); + String botId = getConfigString("bot_id"); String secret = getConfigString("secret"); connectWebSocket(botId, secret); @@ -229,6 +291,16 @@ public class WeComChannelAdapter extends AbstractChannelAdapter { * "Disable + Enable" did to recover. */ private void releaseConnectionResources(String reason) { + // ============================================================================ + // RFC-32 §2.4.1 a-3 / R-6 + R-8 修正：必须按 step 0~4 顺序，不是尾部追加。 + // step 0 (replyQueueAccepting=false) 必须在 ws.close()/wsThread.join() 之前； + // 否则在 ws teardown 期间还会有 keepalive / 最终回复 / proactiveSend 漏进 enqueue。 + // ============================================================================ + + // ---- Step 0：先关 lifecycle gate，让任何后续 sendFrameWithAck 立刻 fast-fail ---- + replyQueueAccepting.set(false); + + // ---- 现有的 ws/heartbeat teardown（功能未变；插在 step 0 之后、step 1 之前） ---- if (heartbeatFuture != null) { heartbeatFuture.cancel(false); heartbeatFuture = null; @@ -253,10 +325,49 @@ public class WeComChannelAdapter extends AbstractChannelAdapter { } wsThread = null; } - pendingAcks.forEach((k, f) -> - f.completeExceptionally(new RuntimeException("Channel " + reason))); - pendingAcks.clear(); + + // ---- Step 1：第一次 drain replyQueues ---- + // forEach 是 weakly-consistent 迭代器，可能错过 step 0 之前刚提交但还没出 compute + // 的 enqueue —— step 3 会再 drain 一次兜底。 + replyQueues.forEach((rid, state) -> { + state.closed().set(true); + ReplyTask t; + while ((t = state.queue().poll()) != null) { + if (!t.future().isDone()) { + t.future().completeExceptionally(new IllegalStateException("Channel " + reason)); + } + } + }); + + // ---- Step 2：shutdownNow 中断 worker 阻塞中的 poll(60s) + 拒绝后续 submit ---- + ExecutorService oldExecutor = this.replyExecutor; + if (oldExecutor != null) { + oldExecutor.shutdownNow(); + this.replyExecutor = null; + } + + // ---- Step 3：second drain，捕获 step 1 与 step 2 之间的窗口期残留 ---- + // 此刻 shutdownNow 已经把任何新 fresh state 的 worker 拒掉，drain 是它们唯一退路。 + replyQueues.forEach((rid, state) -> { + state.closed().set(true); + ReplyTask t; + while ((t = state.queue().poll()) != null) { + if (!t.future().isDone()) { + t.future().completeExceptionally(new IllegalStateException("Channel " + reason)); + } + } + }); replyQueues.clear(); + + // ---- Step 4：pendingAcks 残留 ---- + pendingAcks.forEach((k, f) -> { + if (!f.isDone()) { + f.completeExceptionally(new IllegalStateException("Channel " + reason)); + } + }); + pendingAcks.clear(); + + // ---- 其他 per-connection 状态 ---- pendingFrames.clear(); replyContexts.clear(); missedPongCount.set(0); @@ -264,6 +375,119 @@ public class WeComChannelAdapter extends AbstractChannelAdapter { this.httpClient = null; } + // ==================================================================== + // RFC-32 §2.0.5 / §2.4.1 a-1: lifecycle gate plumbing + // ==================================================================== + + /** + * (Re)build the worker pool. Called from {@link #doStart()} and + * {@link #doReconnect()}. Does not touch the {@link #replyQueueAccepting} + * gate — that flag is controlled by the transport-ready signal + * ({@link #markReady()}). See §2.4.1 a-1 / R-7. + */ + private void ensureReplyExecutor() { + if (replyExecutor == null || replyExecutor.isShutdown()) { + replyExecutor = Executors.newCachedThreadPool(r -> { + Thread t = new Thread(r, "wecom-reply"); + t.setDaemon(true); + return t; + }); + } + } + + /** + * Open the {@link #replyQueueAccepting} lifecycle gate. Only + * called from {@link #markReady()} after auth_succeed. Until this + * runs, every {@link #sendFrameWithAck} call fast-fails the caller's + * future with {@link IllegalStateException}. + */ + private void openReplyQueue() { + replyQueueAccepting.set(true); + } + + /** + * Per-reqId serial worker. Started lazily by + * {@link #sendFrameWithAck} when a fresh {@link ReplyQueueState} is + * created. Exits when: + *

+ * + *

The compute-based idle-close fixes the TOCTOU race called out + * in RFC-32 §2.4.1 a-2 / R-5: enqueue's {@code compute} and + * worker's idle-close {@code compute} share the same bin lock, + * so offer and remove never interleave on the same key. + */ + private void reqIdWorker(String reqId, ReplyQueueState state) { + while (running.get() && !Thread.currentThread().isInterrupted()) { + ReplyTask task; + try { + task = state.queue().poll(60, TimeUnit.SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + break; // fall through to drainStateExceptionally + return + } + + if (task == null) { + // Atomic close — serialized against sendFrameWithAck.compute on + // the same reqId by ConcurrentHashMap's bin lock. + ReplyQueueState afterClose = replyQueues.compute(reqId, (k, current) -> { + if (current != state) return current; // (c) replaced — defensive exit + if (!current.queue().isEmpty()) return current; // (b) late offer — stay alive + current.closed().set(true); // (a) truly idle — close + return null; // (a) remove entry + }); + if (afterClose != state) return; // (a) or (c) — exit + continue; // (b) — keep going + } + + try { + pendingAcks.put(reqId, task.future()); + // orTimeout 5s 兜底，whenComplete 在完成时清 pendingAcks。 + // 用 (key, value) 双参 remove 避免误删后续 task 的注册。 + task.future().orTimeout(REPLY_ACK_TIMEOUT_MS, TimeUnit.MILLISECONDS) + .whenComplete((r, ex) -> pendingAcks.remove(reqId, task.future())); + sendFrame(task.frame()); + task.future().join(); // serialize: don't dequeue next until this is done + } catch (CompletionException ce) { + // join() 抛的是 orTimeout 注入的异常（典型：TimeoutException）—— + // task.future 已经 complete，无需手动 fail + log.debug("[wecom] reply task ACK failed for reqId={}: {}", reqId, ce.getCause()); + } catch (Exception e) { + // sendFrame 同步抛 → ACK 永远不会到 → 必须显式 fail，否则 caller future 永久 pending + if (!task.future().isDone()) { + task.future().completeExceptionally(e); + } + pendingAcks.remove(reqId, task.future()); + log.debug("[wecom] reply task send failed for reqId={}: {}", reqId, e.getMessage()); + } + } + + // running=false / interrupted: mark closed + drain leftover + state.closed().set(true); + drainStateExceptionally(reqId, state, "channel stopped"); + } + + /** + * Drain remaining tasks in a {@link ReplyQueueState} and best-effort + * remove the entry from {@link #replyQueues}. Used by worker exit + * paths (running=false / interrupt). For {@link #releaseConnectionResources} + * the drain is inlined (step 1 / step 3) to keep the ordering proof local. + */ + private void drainStateExceptionally(String reqId, ReplyQueueState state, String reason) { + ReplyTask t; + while ((t = state.queue().poll()) != null) { + if (!t.future().isDone()) { + t.future().completeExceptionally(new IllegalStateException(reason)); + } + } + replyQueues.remove(reqId, state); + } + // ==================== WebSocket 连接 ==================== /** @@ -354,6 +578,10 @@ public class WeComChannelAdapter extends AbstractChannelAdapter { reconnectFuture = null; } disconnectInflight.set(false); + // RFC-32 §2.4.1 a-1 / R-7: only NOW does sendFrameWithAck start + // accepting tasks — auth_succeed has just been observed and the + // WS is the canonical "transport ready" anchor. + openReplyQueue(); } /** @@ -1063,10 +1291,33 @@ public class WeComChannelAdapter extends AbstractChannelAdapter { * @param finish 是否结束流式消息 */ private void replyStream(String originalReqId, String streamId, String content, boolean finish) { + replyStream(originalReqId, streamId, content, finish, null); + } + + /** + * Streaming reply with optional WeCom feedback id attached on the + * final chunk (PR-2 hook installed in PR-0 so the protocol surface + * is stable). + * + *

Per WeCom AI Bot protocol (verified against the langbot + * reference implementation), {@code feedback.id} is only meaningful + * on the chunk where {@code finish=true}. We accept the parameter + * on every chunk for ergonomics but only emit the JSON field on + * the finishing chunk to avoid surfacing it where the server + * would ignore it. + * + *

Callers that don't need feedback collection pass {@code null} + * for {@code feedbackId} (or use the legacy 4-arg overload). + */ + private void replyStream(String originalReqId, String streamId, String content, + boolean finish, String feedbackId) { Map streamBody = new LinkedHashMap<>(); streamBody.put("id", streamId); streamBody.put("finish", finish); streamBody.put("content", content); + if (finish && feedbackId != null && !feedbackId.isBlank()) { + streamBody.put("feedback", Map.of("id", feedbackId)); + } Map body = Map.of( "msgtype", "stream", @@ -1407,27 +1658,94 @@ public class WeComChannelAdapter extends AbstractChannelAdapter { } /** - * 串行队列发送帧，等待 ACK（带超时） - *

- * 同一 reqId 的消息按顺序发送，每条等待 ACK 后再发下一条。 + * Serially send a frame on the WS and wait (in a per-reqId worker) + * for its ACK. Same {@code reqId} messages are guaranteed to be + * dispatched in arrival order: the worker reads from the queue, + * registers {@link #pendingAcks} only after the previous ACK + * settled, sends, then blocks on the future until the ACK arrives + * or {@link #REPLY_ACK_TIMEOUT_MS} elapses. + * + *

RFC-32 §2.4.1 a-2 / R-5/R-6/R-7 invariants this implements: + *

+ * + *

Returns the ACK future for callers that want to chain on + * success (e.g. extract {@code body} fields from the ACK frame). + * Existing fire-and-forget callers can ignore the return value; + * timeout/error handling lives inside the worker. */ - private void sendFrameWithAck(String reqId, Map frame) { + @SuppressWarnings("UnusedReturnValue") + private CompletableFuture> sendFrameWithAck(String reqId, Map frame) { CompletableFuture> ackFuture = new CompletableFuture<>(); - // 注册 ACK 等待 - pendingAcks.put(reqId, ackFuture); + // ---- Outer lifecycle check (fast-fail, no allocations beyond the future) ---- + if (!replyQueueAccepting.get()) { + ackFuture.completeExceptionally( + new IllegalStateException("WeCom channel not accepting reply tasks (lifecycle gate closed)")); + return ackFuture; + } - // 发送帧 - sendFrame(frame); + ReplyTask task = new ReplyTask(frame, ackFuture); + boolean[] offered = {false}; - // 等待 ACK（超时 5 秒，不阻塞当前线程 — fire and forget） - ackFuture.orTimeout(REPLY_ACK_TIMEOUT_MS, TimeUnit.MILLISECONDS) - .whenComplete((result, ex) -> { - pendingAcks.remove(reqId); - if (ex != null) { - log.debug("[wecom] Reply ACK timeout or error for reqId={}: {}", reqId, ex.getMessage()); - } - }); + try { + replyQueues.compute(reqId, (k, existing) -> { + // ---- Inner lifecycle check: gate may have flipped between outer check and bin lock ---- + if (!replyQueueAccepting.get()) { + return existing; // do NOT modify map; offered[0] stays false → fail below + } + + // ---- Reuse open state if present ---- + if (existing != null && !existing.closed().get()) { + existing.queue().offer(task); + offered[0] = true; + return existing; + } + + // ---- Need to start a fresh state. Defend against late-shutdown ---- + ExecutorService exec = this.replyExecutor; + if (exec == null || exec.isShutdown()) { + return existing; // executor torn down by release; fail below + } + + ReplyQueueState fresh = ReplyQueueState.fresh(); + fresh.queue().offer(task); + try { + exec.submit(() -> reqIdWorker(k, fresh)); + offered[0] = true; + return fresh; + } catch (RejectedExecutionException ree) { + // Race with shutdownNow between isShutdown check and submit + return existing; // do not write fresh; fail below + } + }); + } catch (Exception e) { + // compute lambda surfaced something we didn't expect — never let this leak as + // a hung future + ackFuture.completeExceptionally(e); + return ackFuture; + } + + // ---- Final guarantee: any path that didn't offer must fail-fast ---- + if (!offered[0] && !ackFuture.isDone()) { + ackFuture.completeExceptionally( + new IllegalStateException("WeCom channel transitioning, reply task rejected")); + } + return ackFuture; } // ==================== 媒体文件下载与 AES 解密 ====================