From be836b6f0a7f9eb7c95526873e5ea25b6e0b3e36 Mon Sep 17 00:00:00 2001 From: mateaix <57164338+mateaix@users.noreply.github.com> Date: Tue, 21 Jul 2026 21:48:30 +0800 Subject: [PATCH] fix(wecom): harden streaming reply management --- .../channel/wecom/WeComChannelAdapter.java | 92 ++++++++++++++++++- .../wecom/WeComKeepaliveScheduler.java | 12 ++- .../tool_guard/ToolGuardCardHandler.java | 9 +- .../tool_guard/ToolGuardCardRenderer.java | 18 +++- .../channel/wecom/ReplyStreamDedupTest.java | 20 ++++ .../wecom/WeComKeepaliveSchedulerTest.java | 12 ++- .../channel/wecom/WeComProcessStreamTest.java | 38 ++++++++ 7 files changed, 186 insertions(+), 15 deletions(-) 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 6d6ff3b8..ea447426 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 @@ -12,11 +12,13 @@ import vip.mate.workspace.core.service.ChatUploadLocationResolver; import vip.mate.channel.ExponentialBackoff; import vip.mate.channel.media.InboundMediaDownloader; import vip.mate.channel.model.ChannelEntity; +import vip.mate.channel.wecom.cards.tool_guard.ToolGuardCardRenderer; import vip.mate.workspace.conversation.model.MessageContentPart; import javax.crypto.Cipher; import javax.crypto.spec.IvParameterSpec; import javax.crypto.spec.SecretKeySpec; +import java.io.ByteArrayOutputStream; import java.io.InputStream; import java.net.URI; import java.net.http.HttpClient; @@ -24,6 +26,7 @@ import java.net.http.HttpRequest; import java.net.http.HttpResponse; import java.net.http.WebSocket; import java.nio.ByteBuffer; +import java.nio.charset.StandardCharsets; import java.nio.file.Files; import java.nio.file.Path; import java.security.MessageDigest; @@ -196,6 +199,15 @@ public class WeComChannelAdapter extends AbstractChannelAdapter implements Strea /** WebSocket 消息碎片缓冲区 */ private final StringBuilder wsBuffer = new StringBuilder(); + /** + * Raw-byte accumulator for fragmented binary WS frames. Bytes are only + * decoded (as UTF-8) once the final fragment arrives — decoding each + * fragment separately would corrupt any multi-byte character split + * across a fragment boundary. Accessed only from the JDK WebSocket + * listener callbacks, which are delivered serially per socket. + */ + private final ByteArrayOutputStream wsBinaryBuffer = new ByteArrayOutputStream(); + /** 请求 ID 计数器 */ private final AtomicInteger reqIdCounter = new AtomicInteger(0); @@ -498,6 +510,9 @@ public class WeComChannelAdapter extends AbstractChannelAdapter implements Strea pendingFrames.clear(); replyContexts.clear(); streamLastContent.clear(); + // 断线时可能残留半截帧碎片，清空以免污染下一个连接的首帧 + wsBuffer.setLength(0); + wsBinaryBuffer.reset(); if (keepaliveScheduler != null) { keepaliveScheduler.shutdownAll(); } @@ -761,8 +776,10 @@ public class WeComChannelAdapter extends AbstractChannelAdapter implements Strea } byte[] bytes = new byte[data.remaining()]; data.get(bytes); - wsBuffer.append(new String(bytes)); + wsBinaryBuffer.write(bytes, 0, bytes.length); if (last) { + wsBuffer.append(new String(wsBinaryBuffer.toByteArray(), StandardCharsets.UTF_8)); + wsBinaryBuffer.reset(); String fullMessage = wsBuffer.toString(); wsBuffer.setLength(0); handleWebSocketFrame(fullMessage); @@ -1455,6 +1472,10 @@ public class WeComChannelAdapter extends AbstractChannelAdapter implements Strea try { replyStream(ctx.frameReqId(), ctx.processingStreamId(), progress.snapshot(), false); } catch (Exception e) { + // Reset the throttle window so the next delta retries the + // overwrite immediately instead of waiting out the min + // interval — a failed push means the bubble is stale. + lastFlushAt[0] = 0L; log.debug("[wecom] progress overwrite failed: {}", e.getMessage()); } }).blockLast(Duration.ofMinutes(10)); @@ -1516,10 +1537,10 @@ public class WeComChannelAdapter extends AbstractChannelAdapter implements Strea return; } - // 检查是否有 pending frame（用于 reply_stream 覆盖"思考中..."） - // sendMessage 被 renderAndSend 调用时，尝试用 reply_stream 覆盖 - // 但由于 rawPayload 信息在 ChannelMessageRouter 层已丢失， - // 这里走 send_message 主动推送路径 + // 无 frame 上下文的通用发送入口（cron/异步通知等主动推送场景）。 + // 带 WeComReplyContext 的回复路径（renderAndSend / sendContentParts） + // 已直接绑定入站 frame 发送，不再落到这里；此处保留 + // send_message 主动推送 + 群聊缓存 reqId 兜底。 sendMessageToChat(targetId, content); } @@ -1655,6 +1676,10 @@ public class WeComChannelAdapter extends AbstractChannelAdapter implements Strea && !ctx.processingStreamId().isBlank()) { replyStream(ctx.frameReqId(), ctx.processingStreamId(), segment, true); first = false; + } else if (ctx != null && ctx.frameReqId() != null && !ctx.frameReqId().isBlank()) { + // 后续分段绑定同一入站 frame 回复——群聊拒收主动推送， + // 走 frame 回复在群聊/单聊都可达且保持顺序 + replyMarkdown(ctx.frameReqId(), segment); } else { sendMessage(targetId, segment); } @@ -1773,6 +1798,10 @@ public class WeComChannelAdapter extends AbstractChannelAdapter implements Strea && !ctx.processingStreamId().isBlank()) { replyStream(ctx.frameReqId(), ctx.processingStreamId(), rewritten, true); firstText = false; + } else if (ctx != null && ctx.frameReqId() != null + && !ctx.frameReqId().isBlank()) { + // 后续文本绑定同一入站 frame 回复（群聊拒收主动推送） + replyMarkdown(ctx.frameReqId(), rewritten); } else { sendMessage(targetId, rewritten); } @@ -2075,6 +2104,42 @@ public class WeComChannelAdapter extends AbstractChannelAdapter implements Strea */ private final ConcurrentHashMap streamLastContent = new ConcurrentHashMap<>(); + /** + * Send a markdown bubble bound to an inbound frame via + * {@code aibot_respond_msg}. + * + *

Used for reply segments after the first when a live + * {@link WeComReplyContext} exists: the platform rejects + * {@code aibot_send_msg} in group chats, so segments pushed actively + * would silently vanish there. Riding the inbound frame's reply slot + * works in both group and single chats, and keeps segment ordering + * behind the stream bubble (same per-reqId serial worker queue). + * + *

Failures are logged, not propagated — one bad segment must not + * abort the remaining segments of a long reply. + */ + private void replyMarkdown(String frameReqId, String content) { + if (frameReqId == null || frameReqId.isBlank() + || content == null || content.isBlank()) { + return; + } + try { + Map body = Map.of( + "msgtype", "markdown", + "markdown", Map.of("content", content) + ); + Map frame = Map.of( + "cmd", CMD_RESPONSE, + "headers", Map.of("req_id", frameReqId), + "body", body + ); + sendFrameWithAck(frameReqId, frame); + } catch (Exception e) { + log.error("[wecom] Failed to send reply segment via frame {}: {}", + frameReqId, e.getMessage()); + } + } + // ==================== 上传大小预校验（WeCom 平台限制） ==================== /** WeCom hard limits — verified empirically; sources differ slightly. */ @@ -2252,6 +2317,16 @@ public class WeComChannelAdapter extends AbstractChannelAdapter implements Strea sendFrameWithAck(eventReqId, frame); } + /** + * URL for the resolved approval card's mandatory {@code card_action} + * link (WeCom rejects {@code text_notice} cards without a type-1/2 + * action). Reads channel config {@code card_action_url}; public so the + * card handler (different package) can pass it to the renderer. + */ + public String resolvedCardActionUrl() { + return getConfigString("card_action_url", ToolGuardCardRenderer.DEFAULT_CARD_ACTION_URL); + } + /** * Keepalive refresh tick (called by {@link WeComKeepaliveScheduler} * every 20s). Sends {@code finish=false} on the existing stream so @@ -2262,6 +2337,13 @@ public class WeComChannelAdapter extends AbstractChannelAdapter implements Strea * not be triggering refresh ticks. */ public void replyStreamRefreshForKeepalive(String reqId, String streamId, String text) { + // Bypass chunk dedup: a refresh whose text equals the previous chunk + // (e.g. the static placeholder when no progress supplier is attached) + // would otherwise be swallowed by replyStream's dedup guard — no + // network frame goes out, the server-side TTL is NOT reset, and the + // slot dies exactly the way this keepalive exists to prevent. Clearing + // the dedup slot first guarantees every tick produces a real frame. + streamLastContent.remove(streamId); replyStream(reqId, streamId, text, false); } diff --git a/mateclaw-server/src/main/java/vip/mate/channel/wecom/WeComKeepaliveScheduler.java b/mateclaw-server/src/main/java/vip/mate/channel/wecom/WeComKeepaliveScheduler.java index 600b7c65..7724d836 100644 --- a/mateclaw-server/src/main/java/vip/mate/channel/wecom/WeComKeepaliveScheduler.java +++ b/mateclaw-server/src/main/java/vip/mate/channel/wecom/WeComKeepaliveScheduler.java @@ -48,9 +48,17 @@ public class WeComKeepaliveScheduler { /** Hard ceiling — after this many seconds, force-finish the stream. */ static final long MAX_DURATION_SECONDS = 180; - /** Placeholder text written on every refresh tick + on force-finish. */ + /** Placeholder text written on every refresh tick (when no live progress supplier is attached). */ static final String PROCESSING_TEXT = "🤔 思考中..."; + /** + * Text written when the 180s ceiling force-finishes the stream. The real + * answer will arrive later as a separate pushed bubble (the reply context + * is invalidated below), so the sealed bubble must tell the user the + * reply is still coming — freezing it on "思考中..." reads as a hang. + */ + static final String FORCE_FINISH_TEXT = "⏳ 任务耗时较长，仍在处理中，结果稍后送达"; + /** One-shot state per active stream. Held by reference inside the scheduled task. */ private static final class StreamState { final WeComChannelAdapter adapter; @@ -159,7 +167,7 @@ public class WeComKeepaliveScheduler { // replyContext entry so the eventual real reply takes the // fresh-stream path. try { - st.adapter.replyStreamFinishForKeepalive(st.reqId, st.streamId, PROCESSING_TEXT); + st.adapter.replyStreamFinishForKeepalive(st.reqId, st.streamId, FORCE_FINISH_TEXT); } catch (Exception e) { log.debug("[wecom-keepalive] force-finish replyStream failed for {}: {}", st.streamId, e.getMessage()); diff --git a/mateclaw-server/src/main/java/vip/mate/channel/wecom/cards/tool_guard/ToolGuardCardHandler.java b/mateclaw-server/src/main/java/vip/mate/channel/wecom/cards/tool_guard/ToolGuardCardHandler.java index ac28885d..6f3d8202 100644 --- a/mateclaw-server/src/main/java/vip/mate/channel/wecom/cards/tool_guard/ToolGuardCardHandler.java +++ b/mateclaw-server/src/main/java/vip/mate/channel/wecom/cards/tool_guard/ToolGuardCardHandler.java @@ -199,7 +199,8 @@ public class ToolGuardCardHandler implements WeComCardHandler { : "Tool " + toolName + " 已拒绝"; try { adapter.updateTemplateCard(eventReqId, - ToolGuardCardRenderer.buildResolvedCard(taskId, title, desc)); + ToolGuardCardRenderer.buildResolvedCard(taskId, title, desc, + adapter.resolvedCardActionUrl())); } catch (Exception e) { log.warn("[wecom-toolguard] update_template_card (resolved) failed: {}", e.getMessage()); } @@ -212,7 +213,8 @@ public class ToolGuardCardHandler implements WeComCardHandler { adapter.updateTemplateCard(eventReqId, ToolGuardCardRenderer.buildResolvedCard(taskId, "❌ 仅原请求者可审批", - "请由 " + requesterLabel + " 操作")); + "请由 " + requesterLabel + " 操作", + adapter.resolvedCardActionUrl())); } catch (Exception e) { log.warn("[wecom-toolguard] update_template_card (unauthorised) failed: {}", e.getMessage()); } @@ -224,7 +226,8 @@ public class ToolGuardCardHandler implements WeComCardHandler { adapter.updateTemplateCard(eventReqId, ToolGuardCardRenderer.buildResolvedCard(taskId, "⌛ 审批已过期", - "Tool " + toolName + " 的审批已过期或被处理")); + "Tool " + toolName + " 的审批已过期或被处理", + adapter.resolvedCardActionUrl())); } catch (Exception e) { log.warn("[wecom-toolguard] update_template_card (expired) failed: {}", e.getMessage()); } diff --git a/mateclaw-server/src/main/java/vip/mate/channel/wecom/cards/tool_guard/ToolGuardCardRenderer.java b/mateclaw-server/src/main/java/vip/mate/channel/wecom/cards/tool_guard/ToolGuardCardRenderer.java index 6498c3e4..8f502e8c 100644 --- a/mateclaw-server/src/main/java/vip/mate/channel/wecom/cards/tool_guard/ToolGuardCardRenderer.java +++ b/mateclaw-server/src/main/java/vip/mate/channel/wecom/cards/tool_guard/ToolGuardCardRenderer.java @@ -99,13 +99,29 @@ public class ToolGuardCardRenderer implements WeComCardRenderer { * to stay inside WeCom's main_title.desc limit */ public static Map buildResolvedCard(String taskId, String title, String desc) { + return buildResolvedCard(taskId, title, desc, DEFAULT_CARD_ACTION_URL); + } + + /** + * Fallback for the mandatory {@code card_action} link when the channel + * config doesn't provide one ({@code card_action_url}). + */ + public static final String DEFAULT_CARD_ACTION_URL = "https://mateclaw.vip"; + + /** + * Same as {@link #buildResolvedCard(String, String, String)} but with an + * explicit {@code card_action} URL (per-channel configurable). + */ + public static Map buildResolvedCard(String taskId, String title, String desc, + String actionUrl) { Map mainTitle = new LinkedHashMap<>(); mainTitle.put("title", title == null ? "" : title); mainTitle.put("desc", truncate(desc == null ? "" : desc, 30)); Map cardAction = new LinkedHashMap<>(); cardAction.put("type", 1); - cardAction.put("url", "https://mateclaw.vip"); + cardAction.put("url", (actionUrl == null || actionUrl.isBlank()) + ? DEFAULT_CARD_ACTION_URL : actionUrl); Map card = new LinkedHashMap<>(); card.put("card_type", "text_notice"); diff --git a/mateclaw-server/src/test/java/vip/mate/channel/wecom/ReplyStreamDedupTest.java b/mateclaw-server/src/test/java/vip/mate/channel/wecom/ReplyStreamDedupTest.java index cd26664c..0ece8882 100644 --- a/mateclaw-server/src/test/java/vip/mate/channel/wecom/ReplyStreamDedupTest.java +++ b/mateclaw-server/src/test/java/vip/mate/channel/wecom/ReplyStreamDedupTest.java @@ -136,6 +136,26 @@ class ReplyStreamDedupTest { "dedup memory must be per-streamId — same content on a different stream still dispatches"); } + @Test + @DisplayName("keepalive refresh bypasses dedup — identical placeholder text still dispatches") + void keepaliveRefreshBypassesDedup() throws Exception { + Method m = WeComChannelAdapter.class.getDeclaredMethod( + "replyStream", String.class, String.class, String.class, boolean.class); + m.setAccessible(true); + // Initial placeholder chunk primes the dedup slot with this exact text. + m.invoke(adapter, "rid", "stream-1", "🤔 思考中...", false); + // Keepalive ticks re-send the SAME static placeholder. If they were + // deduplicated, no frame would reach the server and the stream slot's + // TTL would never reset — the exact failure keepalive exists to prevent. + adapter.replyStreamRefreshForKeepalive("rid", "stream-1", "🤔 思考中..."); + adapter.replyStreamRefreshForKeepalive("rid", "stream-1", "🤔 思考中..."); + + for (int i = 0; i < 3; i++) { + assertNotNull(sentFrames.poll(500, TimeUnit.MILLISECONDS), + "expected frame #" + (i + 1) + " — keepalive refreshes must not be deduplicated"); + } + } + @Test @DisplayName("after finish=true, the dedup slot is cleared so the next stream with same content goes") void finishClearsDedupSlot() throws Exception { diff --git a/mateclaw-server/src/test/java/vip/mate/channel/wecom/WeComKeepaliveSchedulerTest.java b/mateclaw-server/src/test/java/vip/mate/channel/wecom/WeComKeepaliveSchedulerTest.java index 56024609..ddf5ff20 100644 --- a/mateclaw-server/src/test/java/vip/mate/channel/wecom/WeComKeepaliveSchedulerTest.java +++ b/mateclaw-server/src/test/java/vip/mate/channel/wecom/WeComKeepaliveSchedulerTest.java @@ -16,8 +16,8 @@ import static org.mockito.Mockito.*; /** * Verify the WeComKeepaliveScheduler bookkeeping + force-finish path. * - *

The 20s/180s timing constants come from QwenPaw and are already - * validated empirically in production; we don't re-test the exact + *

The 20s/180s timing constants were chosen against the empirically + * observed stream-slot TTL and are validated in production; we don't re-test the exact * scheduling intervals here (would require either real wall-clock waits * or invasive ScheduledExecutor mocking). Instead we cover: *