mirror of
https://gitee.com/mateos/mateclaw.git
synced 2026-09-13 03:13:41 +08:00
fix(wecom): harden streaming reply management
This commit is contained in:
mateaix
parent
e756219bdc
commit
be836b6f0a
@ -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<String, String> streamLastContent = new ConcurrentHashMap<>();
|
||||
|
||||
/**
|
||||
* Send a markdown bubble bound to an inbound frame via
|
||||
* {@code aibot_respond_msg}.
|
||||
*
|
||||
* <p>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).
|
||||
*
|
||||
* <p>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<String, Object> body = Map.of(
|
||||
"msgtype", "markdown",
|
||||
"markdown", Map.of("content", content)
|
||||
);
|
||||
Map<String, Object> 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);
|
||||
}
|
||||
|
||||
|
||||
@ -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());
|
||||
|
||||
@ -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());
|
||||
}
|
||||
|
||||
@ -99,13 +99,29 @@ public class ToolGuardCardRenderer implements WeComCardRenderer {
|
||||
* to stay inside WeCom's main_title.desc limit
|
||||
*/
|
||||
public static Map<String, Object> 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<String, Object> buildResolvedCard(String taskId, String title, String desc,
|
||||
String actionUrl) {
|
||||
Map<String, Object> mainTitle = new LinkedHashMap<>();
|
||||
mainTitle.put("title", title == null ? "" : title);
|
||||
mainTitle.put("desc", truncate(desc == null ? "" : desc, 30));
|
||||
|
||||
Map<String, Object> 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<String, Object> card = new LinkedHashMap<>();
|
||||
card.put("card_type", "text_notice");
|
||||
|
||||
@ -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 {
|
||||
|
||||
@ -16,8 +16,8 @@ import static org.mockito.Mockito.*;
|
||||
/**
|
||||
* Verify the WeComKeepaliveScheduler bookkeeping + force-finish path.
|
||||
*
|
||||
* <p>The 20s/180s timing constants come from QwenPaw and are already
|
||||
* validated empirically in production; we don't re-test the exact
|
||||
* <p>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:
|
||||
* <ul>
|
||||
@ -114,7 +114,7 @@ class WeComKeepaliveSchedulerTest {
|
||||
tick.invoke(scheduler, state);
|
||||
|
||||
verify(adapter, times(1)).replyStreamFinishForKeepalive(
|
||||
eq("req-x"), eq("stream-x"), eq(WeComKeepaliveScheduler.PROCESSING_TEXT));
|
||||
eq("req-x"), eq("stream-x"), eq(WeComKeepaliveScheduler.FORCE_FINISH_TEXT));
|
||||
verify(adapter, times(1)).invalidateReplyContext(eq("user-alice"), eq("stream-x"));
|
||||
verify(adapter, never()).replyStreamRefreshForKeepalive(any(), any(), any());
|
||||
// After force-finish, the stream is removed from the tracker
|
||||
@ -142,11 +142,15 @@ class WeComKeepaliveSchedulerTest {
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("constants match the QwenPaw-verified values (20s refresh / 180s ceiling)")
|
||||
@DisplayName("constants match the empirically-verified values (20s refresh / 180s ceiling)")
|
||||
void constantsMatch() {
|
||||
assertEquals(20L, WeComKeepaliveScheduler.REFRESH_INTERVAL_SECONDS);
|
||||
assertEquals(180L, WeComKeepaliveScheduler.MAX_DURATION_SECONDS);
|
||||
assertEquals("🤔 思考中...", WeComKeepaliveScheduler.PROCESSING_TEXT);
|
||||
// Force-finish must NOT reuse the processing placeholder — the sealed
|
||||
// bubble is the last thing the user sees until the pushed reply lands.
|
||||
assertNotEquals(WeComKeepaliveScheduler.PROCESSING_TEXT,
|
||||
WeComKeepaliveScheduler.FORCE_FINISH_TEXT);
|
||||
}
|
||||
|
||||
// Pull a tracked StreamState by streamId via reflection. The states map
|
||||
|
||||
@ -141,6 +141,44 @@ class WeComProcessStreamTest {
|
||||
"default config must not leave standalone tool messages");
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("multi-segment reply: segments after the first ride the inbound frame, not proactive push")
|
||||
@SuppressWarnings("unchecked")
|
||||
void multiSegmentRidesInboundFrame() throws Exception {
|
||||
TestableAdapter adapter = newAdapter("{}");
|
||||
seedReplyContext(adapter, "alice", "req-1", "stream-1");
|
||||
|
||||
// Long enough to exceed the 2048-char platform limit → at least 2 segments.
|
||||
StringBuilder sb = new StringBuilder();
|
||||
for (int i = 0; i < 80; i++) {
|
||||
sb.append("第").append(i)
|
||||
.append("行：这是一段足够长的中文内容，用来撑破企业微信单条消息的长度上限，验证分段发送路径。\n");
|
||||
}
|
||||
adapter.renderAndSend("alice", sb.toString());
|
||||
|
||||
List<Map<String, Object>> frames = adapter.drainFrames();
|
||||
assertTrue(frames.size() >= 2, "expected >= 2 outbound frames, got " + frames.size());
|
||||
|
||||
// Segment 1 overwrites the stream bubble with finish=true.
|
||||
Map<String, Object> firstBody = (Map<String, Object>) frames.get(0).get("body");
|
||||
assertEquals("stream", firstBody.get("msgtype"), "first segment must close the stream bubble");
|
||||
|
||||
// Segments 2+ must be markdown replies bound to the SAME inbound frame —
|
||||
// aibot_send_msg is rejected in group chats, so any proactive push here
|
||||
// would silently lose the segment for group users.
|
||||
for (int i = 1; i < frames.size(); i++) {
|
||||
Map<String, Object> frame = frames.get(i);
|
||||
assertEquals("aibot_respond_msg", frame.get("cmd"),
|
||||
"segment #" + (i + 1) + " must ride the inbound frame reply slot");
|
||||
Map<String, Object> headers = (Map<String, Object>) frame.get("headers");
|
||||
assertEquals("req-1", headers.get("req_id"));
|
||||
Map<String, Object> body = (Map<String, Object>) frame.get("body");
|
||||
assertEquals("markdown", body.get("msgtype"));
|
||||
}
|
||||
assertTrue(frames.stream().noneMatch(f -> "aibot_send_msg".equals(f.get("cmd"))),
|
||||
"no segment may fall back to proactive push while a reply context exists");
|
||||
}
|
||||
|
||||
// ==================== helpers ====================
|
||||
|
||||
private static ChannelMessage inbound(String sender) {
|
||||
|
||||
Loading…
Reference in New Issue
Block a user