mirror of
https://gitee.com/mateos/mateclaw.git
synced 2026-09-13 03:13:41 +08:00
feat(wecom): event-driven progress bubble with live tool trace
This commit is contained in:
matevip
parent
beb1a8c243
commit
bf6bed5511
@ -2,9 +2,12 @@ package vip.mate.channel.wecom;
|
||||
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import reactor.core.publisher.Flux;
|
||||
import vip.mate.agent.AgentService.StreamDelta;
|
||||
import vip.mate.channel.AbstractChannelAdapter;
|
||||
import vip.mate.channel.ChannelMessage;
|
||||
import vip.mate.channel.ChannelMessageRouter;
|
||||
import vip.mate.channel.StreamingChannelAdapter;
|
||||
import vip.mate.workspace.core.service.ChatUploadLocationResolver;
|
||||
import vip.mate.channel.ExponentialBackoff;
|
||||
import vip.mate.channel.media.InboundMediaDownloader;
|
||||
@ -55,12 +58,17 @@ import java.util.concurrent.atomic.AtomicInteger;
|
||||
* <li>welcome_text: 欢迎消息（可选）</li>
|
||||
* <li>media_download_enabled: 是否下载媒体文件（默认 true）</li>
|
||||
* <li>media_dir: 媒体文件保存目录（默认 data/media）</li>
|
||||
* <li>stream_progress: 处理期间是否在气泡内展示实时进度（默认 true；
|
||||
* false 退化为"累积后一次性发送"）</li>
|
||||
* <li>progress_interval_ms: 进度覆写最小间隔（默认 500ms）</li>
|
||||
* <li>filter_thinking: false 时思考内容流式进入进度气泡（默认 true）</li>
|
||||
* <li>filter_tool_messages: false 时每次工具调用发独立留痕消息（默认 true）</li>
|
||||
* </ul>
|
||||
*
|
||||
* @author MateClaw Team
|
||||
*/
|
||||
@Slf4j
|
||||
public class WeComChannelAdapter extends AbstractChannelAdapter {
|
||||
public class WeComChannelAdapter extends AbstractChannelAdapter implements StreamingChannelAdapter {
|
||||
|
||||
public static final String CHANNEL_TYPE = "wecom";
|
||||
|
||||
@ -1345,6 +1353,162 @@ public class WeComChannelAdapter extends AbstractChannelAdapter {
|
||||
}
|
||||
}
|
||||
|
||||
// ==================== StreamingChannelAdapter ====================
|
||||
|
||||
/** Minimum interval between progress overwrites of the stream bubble. */
|
||||
private static final long PROGRESS_MIN_INTERVAL_MS = 500;
|
||||
|
||||
/** Max chars of a tool-argument summary in a standalone tool message. */
|
||||
private static final int TOOL_ARGS_SUMMARY_MAX = 120;
|
||||
|
||||
/**
|
||||
* 流式处理 Agent 事件并渲染到企业微信
|
||||
* <p>
|
||||
* 渲染策略：复用入站时发出的 "🤔 思考中..." reply_stream 气泡，随
|
||||
* agent 事件（思考、工具调用、计划步骤、内容产出）持续覆写为实时进度，
|
||||
* 流结束后原地渐变为最终答案（走 renderAndSend 的分段逻辑）。
|
||||
* <ul>
|
||||
* <li>覆写节流 {@link #PROGRESS_MIN_INTERVAL_MS}；工具/审批等关键事件立即刷新</li>
|
||||
* <li>静默期由 {@link WeComKeepaliveScheduler} 每 20s 用进度快照续期</li>
|
||||
* <li>{@code stream_progress=false} 或无可用气泡（语音入站等）时退化为
|
||||
* "累积后一次性发送"，与改造前行为一致</li>
|
||||
* <li>{@code filter_tool_messages=false} 时，每次工具调用另发独立消息留痕</li>
|
||||
* <li>{@code filter_thinking=false} 时，思考内容以引用块进入进度气泡</li>
|
||||
* </ul>
|
||||
*/
|
||||
@Override
|
||||
public String processStream(Flux<StreamDelta> stream, ChannelMessage message, String conversationId) {
|
||||
String replyTarget = message.getReplyToken() != null ? message.getReplyToken()
|
||||
: (message.getChatId() != null ? message.getChatId() : message.getSenderId());
|
||||
WeComReplyContext ctx = replyContexts.get(replyTarget);
|
||||
boolean progressEnabled = getConfigBoolean("stream_progress", true);
|
||||
boolean bubbleUsable = ctx != null && ctx.processingStreamId() != null
|
||||
&& !ctx.processingStreamId().isBlank()
|
||||
&& ctx.frameReqId() != null && !ctx.frameReqId().isBlank();
|
||||
|
||||
StringBuilder contentAccumulator = new StringBuilder();
|
||||
if (!progressEnabled || !bubbleUsable) {
|
||||
// Degraded path — identical to the pre-streaming sync behavior:
|
||||
// accumulate content only (persistOnly deltas included, matching
|
||||
// the router's legacy collector) and send once at the end.
|
||||
stream.doOnNext(delta -> {
|
||||
if (!delta.isEvent() && delta.content() != null) {
|
||||
contentAccumulator.append(delta.content());
|
||||
}
|
||||
})
|
||||
.blockLast(Duration.ofMinutes(10));
|
||||
} else {
|
||||
consumeWithProgress(stream, message, replyTarget, ctx, contentAccumulator);
|
||||
}
|
||||
|
||||
String finalContent = contentAccumulator.toString();
|
||||
if (!finalContent.isBlank()) {
|
||||
renderAndSend(replyTarget, finalContent);
|
||||
}
|
||||
return finalContent;
|
||||
}
|
||||
|
||||
/** Event-driven progress rendering into the existing processing-stream bubble. */
|
||||
private void consumeWithProgress(Flux<StreamDelta> stream, ChannelMessage message,
|
||||
String replyTarget, WeComReplyContext ctx,
|
||||
StringBuilder contentAccumulator) {
|
||||
boolean showThinking = !getConfigBoolean("filter_thinking", true);
|
||||
boolean standaloneToolMessages = !getConfigBoolean("filter_tool_messages", true);
|
||||
long minIntervalMs = getConfigLong("progress_interval_ms", PROGRESS_MIN_INTERVAL_MS);
|
||||
|
||||
WeComProgressRenderer progress = new WeComProgressRenderer(
|
||||
System.currentTimeMillis(), showThinking);
|
||||
if (keepaliveScheduler != null) {
|
||||
// Silent stretches (long LLM calls with no events) keep showing a
|
||||
// fresh elapsed-time snapshot instead of the static placeholder.
|
||||
keepaliveScheduler.attachTextSupplier(ctx.processingStreamId(), progress::snapshot);
|
||||
}
|
||||
|
||||
final long[] lastFlushAt = {0L};
|
||||
stream.doOnNext(delta -> {
|
||||
boolean flushNow = false;
|
||||
if (delta.isEvent()) {
|
||||
flushNow = progress.onEvent(delta.eventType(), delta.eventData());
|
||||
if (standaloneToolMessages) {
|
||||
maybeSendToolEventMessage(replyTarget, delta.eventType(), delta.eventData());
|
||||
}
|
||||
} else {
|
||||
if (delta.thinking() != null) {
|
||||
progress.onThinkingDelta(delta.thinking());
|
||||
}
|
||||
if (delta.content() != null) {
|
||||
contentAccumulator.append(delta.content());
|
||||
progress.onContentDelta(delta.content());
|
||||
}
|
||||
}
|
||||
long now = System.currentTimeMillis();
|
||||
if (!flushNow && now - lastFlushAt[0] < minIntervalMs) {
|
||||
return;
|
||||
}
|
||||
// The keepalive force-finish (180s ceiling) evicts the reply
|
||||
// context; once that happens the stream slot is closed and
|
||||
// further overwrites would be silently rejected — stop pushing.
|
||||
if (replyContexts.get(replyTarget) != ctx) {
|
||||
return;
|
||||
}
|
||||
lastFlushAt[0] = now;
|
||||
try {
|
||||
replyStream(ctx.frameReqId(), ctx.processingStreamId(), progress.snapshot(), false);
|
||||
} catch (Exception e) {
|
||||
log.debug("[wecom] progress overwrite failed: {}", e.getMessage());
|
||||
}
|
||||
}).blockLast(Duration.ofMinutes(10));
|
||||
}
|
||||
|
||||
/**
|
||||
* Standalone tool-call trace messages, sent only when the channel's
|
||||
* {@code filter_tool_messages} toggle is off: the user opted into seeing
|
||||
* the tool trail as persistent bubbles (the progress bubble alone is
|
||||
* transient — the final answer overwrites it).
|
||||
*/
|
||||
private void maybeSendToolEventMessage(String replyTarget, String eventType,
|
||||
Map<String, Object> data) {
|
||||
if (data == null || eventType == null) {
|
||||
return;
|
||||
}
|
||||
Object toolName = data.get("toolName");
|
||||
if (toolName == null) {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
switch (eventType) {
|
||||
case "tool_call_started" -> {
|
||||
String args = summarizeToolArgs(data.get("arguments"));
|
||||
sendMessage(replyTarget, "🔧 调用工具 `" + toolName + "`"
|
||||
+ (args.isEmpty() ? "" : "\n> " + args));
|
||||
}
|
||||
case "tool_call_completed" -> {
|
||||
boolean success = !Boolean.FALSE.equals(data.get("success"));
|
||||
sendMessage(replyTarget, (success ? "✅ `" : "❌ `") + toolName
|
||||
+ (success ? "` 完成" : "` 失败"));
|
||||
}
|
||||
default -> {
|
||||
// Other events carry no standalone tool trace.
|
||||
}
|
||||
}
|
||||
} catch (Exception e) {
|
||||
log.debug("[wecom] tool trace message failed: {}", e.getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
private static String summarizeToolArgs(Object arguments) {
|
||||
if (arguments == null) {
|
||||
return "";
|
||||
}
|
||||
String text = arguments.toString().replaceAll("\\s+", " ").trim();
|
||||
if (text.isEmpty() || "{}".equals(text)) {
|
||||
return "";
|
||||
}
|
||||
return text.length() > TOOL_ARGS_SUMMARY_MAX
|
||||
? text.substring(0, TOOL_ARGS_SUMMARY_MAX) + "…"
|
||||
: text;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void sendMessage(String targetId, String content) {
|
||||
if (webSocket == null) {
|
||||
|
||||
@ -9,6 +9,7 @@ import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.ScheduledExecutorService;
|
||||
import java.util.concurrent.ScheduledFuture;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.function.Supplier;
|
||||
|
||||
/**
|
||||
* Periodically refreshes a WeCom AI Bot {@code stream} reply with the
|
||||
@ -58,6 +59,10 @@ public class WeComKeepaliveScheduler {
|
||||
final String replyToken;
|
||||
final long startedAt;
|
||||
volatile ScheduledFuture<?> future;
|
||||
/** Optional live progress text source; when set, refresh ticks write
|
||||
* its current snapshot instead of the static placeholder so the
|
||||
* bubble keeps showing elapsed time / tool state between events. */
|
||||
volatile Supplier<String> textSupplier;
|
||||
StreamState(WeComChannelAdapter a, String r, String s, String t) {
|
||||
this.adapter = a; this.reqId = r; this.streamId = s; this.replyToken = t;
|
||||
this.startedAt = System.currentTimeMillis();
|
||||
@ -105,6 +110,22 @@ public class WeComKeepaliveScheduler {
|
||||
log.debug("[wecom-keepalive] started for stream={} reqId={}", streamId, reqId);
|
||||
}
|
||||
|
||||
/**
|
||||
* Attach a live progress text source to an already-tracked stream.
|
||||
* Subsequent refresh ticks write the supplier's snapshot instead of the
|
||||
* static placeholder. No-op when the stream is not tracked (already
|
||||
* stopped or force-finished).
|
||||
*/
|
||||
public void attachTextSupplier(String streamId, Supplier<String> supplier) {
|
||||
if (streamId == null || streamId.isBlank()) {
|
||||
return;
|
||||
}
|
||||
StreamState st = states.get(streamId);
|
||||
if (st != null) {
|
||||
st.textSupplier = supplier;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Stop keepalive for a stream — call this immediately before sending
|
||||
* the real reply so the next refresh tick doesn't race the
|
||||
@ -155,12 +176,29 @@ public class WeComKeepaliveScheduler {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
st.adapter.replyStreamRefreshForKeepalive(st.reqId, st.streamId, PROCESSING_TEXT);
|
||||
st.adapter.replyStreamRefreshForKeepalive(st.reqId, st.streamId, refreshText(st));
|
||||
} catch (Exception e) {
|
||||
log.debug("[wecom-keepalive] refresh failed for {}: {}", st.streamId, e.getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
/** Current refresh text: live progress snapshot when attached, static placeholder otherwise. */
|
||||
private String refreshText(StreamState st) {
|
||||
Supplier<String> supplier = st.textSupplier;
|
||||
if (supplier != null) {
|
||||
try {
|
||||
String text = supplier.get();
|
||||
if (text != null && !text.isBlank()) {
|
||||
return text;
|
||||
}
|
||||
} catch (Exception e) {
|
||||
log.debug("[wecom-keepalive] progress supplier failed for {}: {}",
|
||||
st.streamId, e.getMessage());
|
||||
}
|
||||
}
|
||||
return PROCESSING_TEXT;
|
||||
}
|
||||
|
||||
// ---- Test hooks ----
|
||||
|
||||
int activeStreamCount() { return states.size(); }
|
||||
|
||||
@ -0,0 +1,241 @@
|
||||
package vip.mate.channel.wecom;
|
||||
|
||||
import java.util.ArrayDeque;
|
||||
import java.util.Deque;
|
||||
import java.util.Map;
|
||||
|
||||
/**
|
||||
* Builds the live progress text shown in the WeCom stream bubble while an
|
||||
* agent turn is running: a status header (thinking / tool call / replying,
|
||||
* with elapsed time), the most recent tool-call lines, an optional rolling
|
||||
* window of reasoning text, and the tail of the answer produced so far.
|
||||
* <p>
|
||||
* Mutations arrive from the stream-consuming thread; {@link #snapshot()} may
|
||||
* also be called from the keepalive scheduler thread, so all state access is
|
||||
* synchronized on this instance.
|
||||
*/
|
||||
final class WeComProgressRenderer {
|
||||
|
||||
/** Max completed/running tool lines rendered before collapsing to a counter. */
|
||||
private static final int MAX_TOOL_LINES = 3;
|
||||
|
||||
/** Rolling window (chars) of reasoning text when thinking display is on. */
|
||||
private static final int THINKING_WINDOW = 500;
|
||||
|
||||
/** Tail window (chars) of the streamed answer kept inside the bubble.
|
||||
* The full answer is delivered by the final render pass; the bubble only
|
||||
* needs enough to show live progress while staying under the 2048 limit. */
|
||||
private static final int ANSWER_WINDOW = 1200;
|
||||
|
||||
private record ToolLine(String callId, String name, long startedAt,
|
||||
Long finishedAt, boolean success) {
|
||||
}
|
||||
|
||||
private final long startedAtMillis;
|
||||
private final boolean showThinking;
|
||||
|
||||
private final Deque<ToolLine> toolLines = new ArrayDeque<>();
|
||||
private int collapsedToolCount;
|
||||
private final StringBuilder thinkingTail = new StringBuilder();
|
||||
private final StringBuilder answerTail = new StringBuilder();
|
||||
private boolean thinkingSeen;
|
||||
private boolean contentSeen;
|
||||
private boolean approvalPending;
|
||||
private String planStepLine;
|
||||
|
||||
WeComProgressRenderer(long startedAtMillis, boolean showThinking) {
|
||||
this.startedAtMillis = startedAtMillis;
|
||||
this.showThinking = showThinking;
|
||||
}
|
||||
|
||||
synchronized void onThinkingDelta(String delta) {
|
||||
thinkingSeen = true;
|
||||
if (showThinking && delta != null && !delta.isEmpty()) {
|
||||
thinkingTail.append(delta);
|
||||
trimLeading(thinkingTail, THINKING_WINDOW);
|
||||
}
|
||||
}
|
||||
|
||||
synchronized void onContentDelta(String delta) {
|
||||
contentSeen = true;
|
||||
if (delta != null && !delta.isEmpty()) {
|
||||
answerTail.append(delta);
|
||||
trimLeading(answerTail, ANSWER_WINDOW);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Consume a graph event. Returns true when the event changes what the
|
||||
* bubble shows in a way worth flushing immediately (tool transitions,
|
||||
* plan steps, approval waits) rather than waiting for the throttle tick.
|
||||
*/
|
||||
synchronized boolean onEvent(String eventType, Map<String, Object> data) {
|
||||
if (eventType == null) {
|
||||
return false;
|
||||
}
|
||||
switch (eventType) {
|
||||
case "tool_call_started" -> {
|
||||
toolLines.addLast(new ToolLine(
|
||||
stringField(data, "toolCallId"),
|
||||
stringField(data, "toolName"),
|
||||
System.currentTimeMillis(), null, false));
|
||||
compactToolLines();
|
||||
return true;
|
||||
}
|
||||
case "tool_call_completed" -> {
|
||||
String callId = stringField(data, "toolCallId");
|
||||
boolean success = data == null || !Boolean.FALSE.equals(data.get("success"));
|
||||
markToolCompleted(callId, stringField(data, "toolName"), success);
|
||||
return true;
|
||||
}
|
||||
case "plan_step_started" -> {
|
||||
Object index = data != null ? data.get("index") : null;
|
||||
String title = stringField(data, "title");
|
||||
planStepLine = "📋 步骤" + (index != null ? " " + index : "")
|
||||
+ (title != null && !title.isBlank() ? ": " + title : "");
|
||||
return true;
|
||||
}
|
||||
case "tool_approval_requested" -> {
|
||||
approvalPending = true;
|
||||
return true;
|
||||
}
|
||||
default -> {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** Render the current progress text for the stream bubble. */
|
||||
synchronized String snapshot() {
|
||||
StringBuilder sb = new StringBuilder();
|
||||
sb.append(statusLine());
|
||||
if (planStepLine != null) {
|
||||
sb.append('\n').append(planStepLine);
|
||||
}
|
||||
appendToolLines(sb);
|
||||
if (showThinking && !contentSeen && thinkingTail.length() > 0) {
|
||||
sb.append("\n\n> 💭 ").append(thinkingTail.toString().replace("\n", "\n> "));
|
||||
}
|
||||
if (answerTail.length() > 0) {
|
||||
sb.append("\n\n").append(answerTail);
|
||||
}
|
||||
return sb.toString();
|
||||
}
|
||||
|
||||
private String statusLine() {
|
||||
if (approvalPending) {
|
||||
return "⏸️ 等待工具审批…（" + elapsed() + "）";
|
||||
}
|
||||
if (contentSeen) {
|
||||
return "✍️ 正在回复…（" + elapsed() + "）";
|
||||
}
|
||||
ToolLine running = lastRunningTool();
|
||||
if (running != null) {
|
||||
return "🔧 正在调用 " + displayName(running) + "…（" + elapsed() + "）";
|
||||
}
|
||||
if (thinkingSeen) {
|
||||
return "💭 思考中…（" + elapsed() + "）";
|
||||
}
|
||||
return "🤔 思考中…（" + elapsed() + "）";
|
||||
}
|
||||
|
||||
private void appendToolLines(StringBuilder sb) {
|
||||
if (collapsedToolCount > 0) {
|
||||
sb.append("\n…等 ").append(collapsedToolCount).append(" 项已完成");
|
||||
}
|
||||
for (ToolLine line : toolLines) {
|
||||
if (line.finishedAt() == null) {
|
||||
// The running tool is already the status header — skip here
|
||||
// unless content started (header shows "正在回复" instead).
|
||||
if (contentSeen || approvalPending) {
|
||||
sb.append("\n🔧 ").append(displayName(line)).append(" 运行中…");
|
||||
}
|
||||
} else {
|
||||
long seconds = Math.max(0, (line.finishedAt() - line.startedAt()) / 1000);
|
||||
sb.append('\n').append(line.success() ? "✅ " : "❌ ")
|
||||
.append(displayName(line))
|
||||
.append(line.success() ? " 完成" : " 失败")
|
||||
.append("（").append(seconds).append(" 秒）");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private ToolLine lastRunningTool() {
|
||||
ToolLine running = null;
|
||||
for (ToolLine line : toolLines) {
|
||||
if (line.finishedAt() == null) {
|
||||
running = line;
|
||||
}
|
||||
}
|
||||
return running;
|
||||
}
|
||||
|
||||
private void markToolCompleted(String callId, String toolName, boolean success) {
|
||||
ToolLine match = null;
|
||||
for (ToolLine line : toolLines) {
|
||||
if (line.finishedAt() != null) {
|
||||
continue;
|
||||
}
|
||||
boolean idMatch = callId != null && callId.equals(line.callId());
|
||||
boolean nameMatch = callId == null && toolName != null && toolName.equals(line.name());
|
||||
if (idMatch || nameMatch) {
|
||||
match = line;
|
||||
}
|
||||
}
|
||||
if (match == null) {
|
||||
toolLines.addLast(new ToolLine(callId, toolName,
|
||||
System.currentTimeMillis(), System.currentTimeMillis(), success));
|
||||
} else {
|
||||
ToolLine done = new ToolLine(match.callId(), match.name(),
|
||||
match.startedAt(), System.currentTimeMillis(), success);
|
||||
replaceLine(match, done);
|
||||
}
|
||||
compactToolLines();
|
||||
}
|
||||
|
||||
private void replaceLine(ToolLine oldLine, ToolLine newLine) {
|
||||
Deque<ToolLine> rebuilt = new ArrayDeque<>(toolLines.size());
|
||||
for (ToolLine line : toolLines) {
|
||||
rebuilt.addLast(line == oldLine ? newLine : line);
|
||||
}
|
||||
toolLines.clear();
|
||||
toolLines.addAll(rebuilt);
|
||||
}
|
||||
|
||||
/** Keep at most {@link #MAX_TOOL_LINES}; older completed lines collapse into a counter. */
|
||||
private void compactToolLines() {
|
||||
while (toolLines.size() > MAX_TOOL_LINES) {
|
||||
ToolLine oldest = toolLines.peekFirst();
|
||||
if (oldest != null && oldest.finishedAt() == null) {
|
||||
// Never collapse a still-running tool line.
|
||||
break;
|
||||
}
|
||||
toolLines.pollFirst();
|
||||
collapsedToolCount++;
|
||||
}
|
||||
}
|
||||
|
||||
private String elapsed() {
|
||||
long seconds = Math.max(0, (System.currentTimeMillis() - startedAtMillis) / 1000);
|
||||
if (seconds < 60) {
|
||||
return "已 " + seconds + " 秒";
|
||||
}
|
||||
return "已 " + (seconds / 60) + " 分 " + (seconds % 60) + " 秒";
|
||||
}
|
||||
|
||||
private static String displayName(ToolLine line) {
|
||||
return line.name() != null && !line.name().isBlank() ? line.name() : "工具";
|
||||
}
|
||||
|
||||
private static String stringField(Map<String, Object> data, String key) {
|
||||
Object value = data != null ? data.get(key) : null;
|
||||
return value != null ? value.toString() : null;
|
||||
}
|
||||
|
||||
private static void trimLeading(StringBuilder sb, int maxLen) {
|
||||
int excess = sb.length() - maxLen;
|
||||
if (excess > 0) {
|
||||
sb.delete(0, excess);
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,279 @@
|
||||
package vip.mate.channel.wecom;
|
||||
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import org.junit.jupiter.api.DisplayName;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.mockito.Mockito;
|
||||
import reactor.core.publisher.Flux;
|
||||
import vip.mate.agent.AgentService.StreamDelta;
|
||||
import vip.mate.channel.ChannelMessage;
|
||||
import vip.mate.channel.ChannelMessageRouter;
|
||||
import vip.mate.channel.model.ChannelEntity;
|
||||
import vip.mate.channel.notification.ApprovalNotificationService;
|
||||
import vip.mate.channel.wecom.cards.WeComCardDispatcher;
|
||||
|
||||
import java.lang.reflect.Constructor;
|
||||
import java.lang.reflect.Field;
|
||||
import java.lang.reflect.Method;
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.ExecutorService;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.LinkedBlockingQueue;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.*;
|
||||
|
||||
/**
|
||||
* Exercises {@link WeComChannelAdapter#processStream}: the progress-bubble
|
||||
* path (event-driven reply_stream overwrites + final finish=true), the
|
||||
* degraded accumulate-then-send path, and the standalone tool-trace messages
|
||||
* gated by {@code filter_tool_messages=false}.
|
||||
*/
|
||||
class WeComProcessStreamTest {
|
||||
|
||||
@Test
|
||||
@DisplayName("progress path overwrites the bubble with tool progress, then finishes with the answer")
|
||||
void progressPathStreamsAndFinishes() throws Exception {
|
||||
TestableAdapter adapter = newAdapter("{\"progress_interval_ms\": 0}");
|
||||
seedReplyContext(adapter, "alice", "req-1", "stream-1");
|
||||
|
||||
Flux<StreamDelta> stream = Flux.just(
|
||||
new StreamDelta(null, "先查时间"),
|
||||
StreamDelta.event("tool_call_started",
|
||||
Map.of("toolCallId", "c1", "toolName", "get_time")),
|
||||
StreamDelta.event("tool_call_completed",
|
||||
Map.of("toolCallId", "c1", "toolName", "get_time", "success", true)),
|
||||
new StreamDelta("现在是", null),
|
||||
new StreamDelta("下午三点。", null));
|
||||
|
||||
String result = adapter.processStream(stream, inbound("alice"), "wecom:alice");
|
||||
|
||||
assertEquals("现在是下午三点。", result);
|
||||
List<Map<String, Object>> frames = adapter.drainFrames();
|
||||
List<Map<String, Object>> streamBodies = streamBodies(frames);
|
||||
assertFalse(streamBodies.isEmpty(), "expected reply_stream progress frames");
|
||||
|
||||
boolean sawToolProgress = streamBodies.stream()
|
||||
.filter(s -> Boolean.FALSE.equals(s.get("finish")))
|
||||
.anyMatch(s -> String.valueOf(s.get("content")).contains("get_time"));
|
||||
assertTrue(sawToolProgress, "some non-final chunk should show the tool call");
|
||||
|
||||
Map<String, Object> finalChunk = streamBodies.get(streamBodies.size() - 1);
|
||||
assertEquals(Boolean.TRUE, finalChunk.get("finish"), "last chunk must close the stream");
|
||||
assertTrue(String.valueOf(finalChunk.get("content")).contains("现在是下午三点。"));
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("stream_progress=false degrades to accumulate-then-send with no interim overwrites")
|
||||
void progressDisabledDegrades() throws Exception {
|
||||
TestableAdapter adapter = newAdapter("{\"stream_progress\": false}");
|
||||
seedReplyContext(adapter, "alice", "req-1", "stream-1");
|
||||
|
||||
Flux<StreamDelta> stream = Flux.just(
|
||||
StreamDelta.event("tool_call_started", Map.of("toolCallId", "c1", "toolName", "t")),
|
||||
new StreamDelta("答案", null));
|
||||
|
||||
String result = adapter.processStream(stream, inbound("alice"), "wecom:alice");
|
||||
|
||||
assertEquals("答案", result);
|
||||
List<Map<String, Object>> streamBodies = streamBodies(adapter.drainFrames());
|
||||
// Only the final renderAndSend overwrite — no interim progress chunks.
|
||||
assertEquals(1, streamBodies.size(), "degraded path must not stream progress");
|
||||
assertEquals(Boolean.TRUE, streamBodies.get(0).get("finish"));
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("without a reply context the final answer goes out as a plain message")
|
||||
void noContextFallsBackToPlainSend() throws Exception {
|
||||
TestableAdapter adapter = newAdapter("{}");
|
||||
|
||||
Flux<StreamDelta> stream = Flux.just(new StreamDelta("答案", null));
|
||||
String result = adapter.processStream(stream, inbound("alice"), "wecom:alice");
|
||||
|
||||
assertEquals("答案", result);
|
||||
List<Map<String, Object>> frames = adapter.drainFrames();
|
||||
assertTrue(streamBodies(frames).isEmpty(), "no stream slot → no reply_stream frames");
|
||||
assertTrue(frames.stream().anyMatch(f -> "aibot_send_msg".equals(f.get("cmd"))),
|
||||
"answer must fall back to the proactive send path");
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("filter_tool_messages=false emits standalone tool trace messages")
|
||||
void toolTraceMessagesWhenUnfiltered() throws Exception {
|
||||
TestableAdapter adapter = newAdapter(
|
||||
"{\"progress_interval_ms\": 0, \"filter_tool_messages\": false}");
|
||||
seedReplyContext(adapter, "alice", "req-1", "stream-1");
|
||||
|
||||
Flux<StreamDelta> stream = Flux.just(
|
||||
StreamDelta.event("tool_call_started",
|
||||
Map.of("toolCallId", "c1", "toolName", "get_time", "arguments", "{\"tz\":\"cn\"}")),
|
||||
StreamDelta.event("tool_call_completed",
|
||||
Map.of("toolCallId", "c1", "toolName", "get_time", "success", true)),
|
||||
new StreamDelta("好了", null));
|
||||
|
||||
adapter.processStream(stream, inbound("alice"), "wecom:alice");
|
||||
|
||||
List<String> markdowns = markdownContents(adapter.drainFrames());
|
||||
assertTrue(markdowns.stream().anyMatch(t -> t.contains("调用工具") && t.contains("get_time")),
|
||||
"expected a standalone tool-start trace, got: " + markdowns);
|
||||
assertTrue(markdowns.stream().anyMatch(t -> t.contains("get_time") && t.contains("完成")),
|
||||
"expected a standalone tool-completion trace, got: " + markdowns);
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("default filters emit no standalone tool messages")
|
||||
void noToolTraceByDefault() throws Exception {
|
||||
TestableAdapter adapter = newAdapter("{\"progress_interval_ms\": 0}");
|
||||
seedReplyContext(adapter, "alice", "req-1", "stream-1");
|
||||
|
||||
Flux<StreamDelta> stream = Flux.just(
|
||||
StreamDelta.event("tool_call_started", Map.of("toolCallId", "c1", "toolName", "t1")),
|
||||
new StreamDelta("答案", null));
|
||||
|
||||
adapter.processStream(stream, inbound("alice"), "wecom:alice");
|
||||
|
||||
assertTrue(markdownContents(adapter.drainFrames()).stream().noneMatch(t -> t.contains("调用工具")),
|
||||
"default config must not leave standalone tool messages");
|
||||
}
|
||||
|
||||
// ==================== helpers ====================
|
||||
|
||||
private static ChannelMessage inbound(String sender) {
|
||||
return ChannelMessage.builder()
|
||||
.channelType("wecom")
|
||||
.senderId(sender)
|
||||
.replyToken(sender)
|
||||
.content("hi")
|
||||
.build();
|
||||
}
|
||||
|
||||
private static TestableAdapter newAdapter(String configJson) throws Exception {
|
||||
ChannelEntity entity = new ChannelEntity();
|
||||
entity.setId(1L);
|
||||
entity.setChannelType("wecom");
|
||||
entity.setConfigJson(configJson);
|
||||
TestableAdapter adapter = new TestableAdapter(
|
||||
entity,
|
||||
Mockito.mock(ChannelMessageRouter.class),
|
||||
new ObjectMapper(),
|
||||
Mockito.mock(ApprovalNotificationService.class),
|
||||
Mockito.mock(WeComCardDispatcher.class),
|
||||
Mockito.mock(WeComKeepaliveScheduler.class));
|
||||
|
||||
Field running = adapter.getClass().getSuperclass().getSuperclass()
|
||||
.getDeclaredField("running");
|
||||
running.setAccessible(true);
|
||||
((AtomicBoolean) running.get(adapter)).set(true);
|
||||
// sendMessage / sendOutboundFrame gate on a live WebSocket reference;
|
||||
// sendFrame is overridden so the mock is never actually written to.
|
||||
Field ws = WeComChannelAdapter.class.getDeclaredField("webSocket");
|
||||
ws.setAccessible(true);
|
||||
ws.set(adapter, Mockito.mock(java.net.http.WebSocket.class));
|
||||
Method ensure = WeComChannelAdapter.class.getDeclaredMethod("ensureReplyExecutor");
|
||||
ensure.setAccessible(true);
|
||||
ensure.invoke(adapter);
|
||||
Method open = WeComChannelAdapter.class.getDeclaredMethod("openReplyQueue");
|
||||
open.setAccessible(true);
|
||||
open.invoke(adapter);
|
||||
adapter.workerIdleTimeoutMs = 60_000L;
|
||||
return adapter;
|
||||
}
|
||||
|
||||
/** Insert a (frameReqId, processingStreamId) reply context for {@code replyToken}. */
|
||||
@SuppressWarnings("unchecked")
|
||||
private static void seedReplyContext(WeComChannelAdapter adapter, String replyToken,
|
||||
String reqId, String streamId) throws Exception {
|
||||
Field ctxField = WeComChannelAdapter.class.getDeclaredField("replyContexts");
|
||||
ctxField.setAccessible(true);
|
||||
Map<String, Object> contexts = (Map<String, Object>) ctxField.get(adapter);
|
||||
Class<?> ctxClass = Class.forName(
|
||||
"vip.mate.channel.wecom.WeComChannelAdapter$WeComReplyContext");
|
||||
Constructor<?> ctor = ctxClass.getDeclaredConstructor(String.class, String.class);
|
||||
ctor.setAccessible(true);
|
||||
contexts.put(replyToken, ctor.newInstance(reqId, streamId));
|
||||
}
|
||||
|
||||
/** Extract every reply_stream body (in dispatch order) from the captured frames. */
|
||||
@SuppressWarnings("unchecked")
|
||||
private static List<Map<String, Object>> streamBodies(List<Map<String, Object>> frames) {
|
||||
List<Map<String, Object>> out = new ArrayList<>();
|
||||
for (Map<String, Object> frame : frames) {
|
||||
Map<String, Object> body = (Map<String, Object>) frame.get("body");
|
||||
if (body != null && "stream".equals(body.get("msgtype"))) {
|
||||
out.add((Map<String, Object>) body.get("stream"));
|
||||
}
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
/** Extract every markdown message content from the captured frames. */
|
||||
@SuppressWarnings("unchecked")
|
||||
private static List<String> markdownContents(List<Map<String, Object>> frames) {
|
||||
List<String> out = new ArrayList<>();
|
||||
for (Map<String, Object> frame : frames) {
|
||||
Map<String, Object> body = (Map<String, Object>) frame.get("body");
|
||||
if (body != null && "markdown".equals(body.get("msgtype"))) {
|
||||
Map<String, Object> md = (Map<String, Object>) body.get("markdown");
|
||||
if (md != null) {
|
||||
out.add(String.valueOf(md.get("content")));
|
||||
}
|
||||
}
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
/** Frame-capturing adapter with auto-ACK, mirroring ReplyStreamDedupTest. */
|
||||
static class TestableAdapter extends WeComChannelAdapter {
|
||||
final LinkedBlockingQueue<Map<String, Object>> sentFrames = new LinkedBlockingQueue<>();
|
||||
private static final ExecutorService AUTOACK = Executors.newCachedThreadPool(r -> {
|
||||
Thread t = new Thread(r, "test-autoack-progress");
|
||||
t.setDaemon(true);
|
||||
return t;
|
||||
});
|
||||
|
||||
TestableAdapter(ChannelEntity entity, ChannelMessageRouter router,
|
||||
ObjectMapper mapper, ApprovalNotificationService approvalSvc,
|
||||
WeComCardDispatcher cardDispatcher, WeComKeepaliveScheduler keepalive) {
|
||||
super(entity, router, mapper, approvalSvc, cardDispatcher, keepalive);
|
||||
}
|
||||
|
||||
/** Drain all frames dispatched so far, waiting briefly for the async worker. */
|
||||
List<Map<String, Object>> drainFrames() throws InterruptedException {
|
||||
List<Map<String, Object>> out = new ArrayList<>();
|
||||
Map<String, Object> frame;
|
||||
while ((frame = sentFrames.poll(500, TimeUnit.MILLISECONDS)) != null) {
|
||||
out.add(frame);
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
@Override
|
||||
@SuppressWarnings("unchecked")
|
||||
void sendFrame(Map<String, Object> frame) {
|
||||
sentFrames.offer(frame);
|
||||
Map<String, Object> headers = (Map<String, Object>) frame.get("headers");
|
||||
if (headers == null) return;
|
||||
String reqId = (String) headers.get("req_id");
|
||||
if (reqId == null || reqId.isBlank()) return;
|
||||
AUTOACK.submit(() -> completeAckSoon(reqId));
|
||||
}
|
||||
|
||||
private void completeAckSoon(String reqId) {
|
||||
try {
|
||||
Thread.sleep(2);
|
||||
Field f = WeComChannelAdapter.class.getDeclaredField("pendingAcks");
|
||||
f.setAccessible(true);
|
||||
ConcurrentHashMap<String, CompletableFuture<Map<String, Object>>> pending =
|
||||
(ConcurrentHashMap<String, CompletableFuture<Map<String, Object>>>) f.get(this);
|
||||
CompletableFuture<Map<String, Object>> fut = pending.get(reqId);
|
||||
if (fut != null) fut.complete(Map.of("errcode", 0));
|
||||
} catch (Exception ignored) {
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,122 @@
|
||||
package vip.mate.channel.wecom;
|
||||
|
||||
import org.junit.jupiter.api.DisplayName;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import java.util.Map;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.*;
|
||||
|
||||
class WeComProgressRendererTest {
|
||||
|
||||
@Test
|
||||
@DisplayName("initial snapshot shows a thinking status with elapsed time")
|
||||
void initialSnapshot() {
|
||||
WeComProgressRenderer r = new WeComProgressRenderer(System.currentTimeMillis(), false);
|
||||
String s = r.snapshot();
|
||||
assertTrue(s.contains("思考中"), s);
|
||||
assertTrue(s.contains("已 "), s);
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("tool start flips status to the running tool and requests an immediate flush")
|
||||
void toolStartUpdatesStatus() {
|
||||
WeComProgressRenderer r = new WeComProgressRenderer(System.currentTimeMillis(), false);
|
||||
boolean flush = r.onEvent("tool_call_started",
|
||||
Map.of("toolCallId", "c1", "toolName", "get_weather"));
|
||||
assertTrue(flush);
|
||||
String s = r.snapshot();
|
||||
assertTrue(s.contains("正在调用 get_weather"), s);
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("tool completion renders a checked line with duration")
|
||||
void toolCompletionRendersLine() {
|
||||
WeComProgressRenderer r = new WeComProgressRenderer(System.currentTimeMillis(), false);
|
||||
r.onEvent("tool_call_started", Map.of("toolCallId", "c1", "toolName", "get_weather"));
|
||||
r.onEvent("tool_call_completed",
|
||||
Map.of("toolCallId", "c1", "toolName", "get_weather", "success", true));
|
||||
String s = r.snapshot();
|
||||
assertTrue(s.contains("✅ get_weather 完成"), s);
|
||||
assertFalse(s.contains("正在调用"), s);
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("failed tool renders a cross line")
|
||||
void toolFailureRendersCross() {
|
||||
WeComProgressRenderer r = new WeComProgressRenderer(System.currentTimeMillis(), false);
|
||||
r.onEvent("tool_call_started", Map.of("toolCallId", "c1", "toolName", "search"));
|
||||
r.onEvent("tool_call_completed",
|
||||
Map.of("toolCallId", "c1", "toolName", "search", "success", false));
|
||||
String s = r.snapshot();
|
||||
assertTrue(s.contains("❌ search 失败"), s);
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("older completed tools collapse into a counter beyond the display cap")
|
||||
void toolLinesCollapse() {
|
||||
WeComProgressRenderer r = new WeComProgressRenderer(System.currentTimeMillis(), false);
|
||||
for (int i = 1; i <= 5; i++) {
|
||||
String id = "c" + i;
|
||||
r.onEvent("tool_call_started", Map.of("toolCallId", id, "toolName", "tool" + i));
|
||||
r.onEvent("tool_call_completed", Map.of("toolCallId", id, "toolName", "tool" + i));
|
||||
}
|
||||
String s = r.snapshot();
|
||||
assertTrue(s.contains("…等 2 项已完成"), s);
|
||||
assertTrue(s.contains("tool5"), s);
|
||||
assertFalse(s.contains("tool1"), s);
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("thinking text appears as a quote block only when display is enabled")
|
||||
void thinkingDisplayGate() {
|
||||
WeComProgressRenderer shown = new WeComProgressRenderer(System.currentTimeMillis(), true);
|
||||
shown.onThinkingDelta("先查当前时间");
|
||||
assertTrue(shown.snapshot().contains("> 💭 先查当前时间"), shown.snapshot());
|
||||
|
||||
WeComProgressRenderer hidden = new WeComProgressRenderer(System.currentTimeMillis(), false);
|
||||
hidden.onThinkingDelta("先查当前时间");
|
||||
String s = hidden.snapshot();
|
||||
assertFalse(s.contains("先查当前时间"), s);
|
||||
assertTrue(s.contains("💭 思考中"), s);
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("content deltas switch status to replying and show the answer tail")
|
||||
void contentSwitchesToReplying() {
|
||||
WeComProgressRenderer r = new WeComProgressRenderer(System.currentTimeMillis(), true);
|
||||
r.onThinkingDelta("想一想");
|
||||
r.onContentDelta("今天天气晴。");
|
||||
String s = r.snapshot();
|
||||
assertTrue(s.contains("正在回复"), s);
|
||||
assertTrue(s.contains("今天天气晴。"), s);
|
||||
// Thinking quote is dropped once real content flows.
|
||||
assertFalse(s.contains("> 💭"), s);
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("approval request switches status to waiting")
|
||||
void approvalSwitchesStatus() {
|
||||
WeComProgressRenderer r = new WeComProgressRenderer(System.currentTimeMillis(), false);
|
||||
boolean flush = r.onEvent("tool_approval_requested", Map.of("toolName", "rm"));
|
||||
assertTrue(flush);
|
||||
assertTrue(r.snapshot().contains("等待工具审批"), r.snapshot());
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("answer tail stays bounded for very long streamed content")
|
||||
void answerTailBounded() {
|
||||
WeComProgressRenderer r = new WeComProgressRenderer(System.currentTimeMillis(), false);
|
||||
r.onContentDelta("x".repeat(5000));
|
||||
assertTrue(r.snapshot().length() < 2048,
|
||||
"snapshot must stay under the WeCom message limit");
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("unknown events are ignored without requesting a flush")
|
||||
void unknownEventsIgnored() {
|
||||
WeComProgressRenderer r = new WeComProgressRenderer(System.currentTimeMillis(), false);
|
||||
assertFalse(r.onEvent("_usage_final", Map.of()));
|
||||
assertFalse(r.onEvent(null, null));
|
||||
}
|
||||
}
|
||||
@ -3519,9 +3519,9 @@ export default {
|
||||
messageFilter: {
|
||||
title: 'Message Filter',
|
||||
filterThinking: 'Filter Thinking',
|
||||
filterThinkingTooltip: 'Filter <think> tag content before sending to users',
|
||||
filterThinkingTooltip: 'Yes: strip <think> content from the final answer only; No: show reasoning live (WeCom streams it in the progress bubble)',
|
||||
filterToolMessages: 'Filter Tool Messages',
|
||||
filterToolMessagesTooltip: 'Filter tool_call / tool_result and ReAct Action/Observation lines',
|
||||
filterToolMessagesTooltip: 'Yes: strip inline tool_call / Action tags from the final answer only; No: each tool call leaves a standalone trace message (supported on WeCom)',
|
||||
messageFormat: 'Message Format',
|
||||
formatAuto: 'Auto',
|
||||
formatMarkdown: 'Markdown',
|
||||
|
||||
@ -3619,9 +3619,9 @@ export default {
|
||||
messageFilter: {
|
||||
title: '消息过滤',
|
||||
filterThinking: '过滤思维链',
|
||||
filterThinkingTooltip: '发送给用户前过滤 <think> 标签内容',
|
||||
filterThinkingTooltip: '是：仅过滤最终答案中的 <think> 内容；否：思考过程实时展示（企业微信在进度气泡中流式显示）',
|
||||
filterToolMessages: '过滤工具消息',
|
||||
filterToolMessagesTooltip: '过滤 tool_call / tool_result 和 ReAct Action/Observation 行',
|
||||
filterToolMessagesTooltip: '是：仅过滤最终答案中的 tool_call / Action 等内联标签；否：每次工具调用以独立消息留痕（企业微信已支持）',
|
||||
messageFormat: '消息格式',
|
||||
formatAuto: '自动',
|
||||
formatMarkdown: 'Markdown',
|
||||
|
||||
Loading…
Reference in New Issue
Block a user