From 54eb77d7c75dddab3ef9bfd7af1f623a145888c3 Mon Sep 17 00:00:00 2001 From: mateaix <7333791@qq.com> Date: Sun, 9 Aug 2026 20:34:54 +0800 Subject: [PATCH] feat(feishu): show streaming execution progress --- .../channel/feishu/FeishuChannelAdapter.java | 110 +++++++- .../feishu/FeishuProgressRenderer.java | 260 ++++++++++++++++++ .../feishu/FeishuStreamingCardManager.java | 160 ++++++++--- .../src/main/resources/docs/en/channels.md | 7 +- .../src/main/resources/docs/zh/channels.md | 7 +- .../feishu/FeishuProcessStreamTest.java | 216 +++++++++++++++ .../FeishuStreamingCardManagerTest.java | 77 +++++- .../components/channels/ChannelEditModal.vue | 3 + mateclaw-ui/src/i18n/locales/en-US.ts | 2 + mateclaw-ui/src/i18n/locales/zh-CN.ts | 2 + mateclaw-ui/src/types/index.ts | 2 + 11 files changed, 795 insertions(+), 51 deletions(-) create mode 100644 mateclaw-server/src/main/java/vip/mate/channel/feishu/FeishuProgressRenderer.java create mode 100644 mateclaw-server/src/test/java/vip/mate/channel/feishu/FeishuProcessStreamTest.java diff --git a/mateclaw-server/src/main/java/vip/mate/channel/feishu/FeishuChannelAdapter.java b/mateclaw-server/src/main/java/vip/mate/channel/feishu/FeishuChannelAdapter.java index f82f9c34..045ea059 100644 --- a/mateclaw-server/src/main/java/vip/mate/channel/feishu/FeishuChannelAdapter.java +++ b/mateclaw-server/src/main/java/vip/mate/channel/feishu/FeishuChannelAdapter.java @@ -12,6 +12,7 @@ import vip.mate.channel.ChannelMessage; import vip.mate.workspace.core.service.ChatUploadLocationResolver; import vip.mate.channel.ChannelMessageRouter; import vip.mate.channel.ExponentialBackoff; +import vip.mate.channel.ProvisionalContentTracker; import vip.mate.channel.StreamingChannelAdapter; import vip.mate.channel.media.GeneratedFileScrubber; import vip.mate.channel.media.MediaSource; @@ -67,6 +68,10 @@ import java.util.concurrent.TimeUnit; * - card_format: 卡片格式化模式 "auto"（默认）| "always" | "never" * auto: 根据内容自动检测；always: 全部包卡片；never: 全部纯文本（降级/调试用） * - card_header: Markdown 卡片 header 文案，默认 "AI 助手"；设为空串可隐藏 header + * - card_streaming_enabled: 是否启用 CardKit 流式卡片（默认 true） + * - stream_progress: 是否在流式卡片中展示执行轨迹（默认 true） + * - filter_thinking: 是否隐藏原始思考文本（默认 true；状态与阶段轨迹仍展示） + * - filter_tool_messages: 是否隐藏工具名称与逐项状态（默认 true；仍展示汇总数量） * - require_mention: 群聊中是否需要 @机器人 才响应（默认 false） * true: 仅当消息中 @了机器人才处理；通过飞书 mentions 字段精确判断，无需配置 botPrefix * @@ -2596,29 +2601,70 @@ public class FeishuChannelAdapter extends AbstractChannelAdapter implements Stre } StringBuilder accumulator = new StringBuilder(); + boolean progressEnabled = getConfigBoolean("stream_progress", true); + FeishuProgressRenderer progress = progressEnabled + ? new FeishuProgressRenderer( + System.currentTimeMillis(), + !getConfigBoolean("filter_thinking", true), + !getConfigBoolean("filter_tool_messages", true)) + : null; + ProvisionalContentTracker narrationTracker = progressEnabled + ? new ProvisionalContentTracker("feishu") : null; try { stream.doOnNext(delta -> { - // segmentOnly narration is skipped: appending every - // ReAct iteration's "我来查一下…" into the card text is - // what makes the answer read as if it were sent twice. - if (StreamingChannelAdapter.contributesToFinalContent(delta)) { - accumulator.append(delta.content()); - streamingCardManager.appendContent(sessionKey, delta.content(), false); + if (!progressEnabled) { + // Legacy answer-only card mode. + if (StreamingChannelAdapter.contributesToFinalContent(delta)) { + accumulator.append(delta.content()); + streamingCardManager.appendContent(sessionKey, delta.content(), false); + } + return; } + + boolean forceFlush = false; + if (delta.isEvent()) { + if ("tool_call_completed".equals(delta.eventType())) { + narrationTracker.onToolObservation(); + } + forceFlush = progress.onEvent(delta.eventType(), delta.eventData()); + } else if (delta.segmentOnly()) { + String narration = delta.content() != null ? delta.content().trim() : ""; + if (!narration.isEmpty()) { + String publishable = narrationTracker.stageNarration(narration, delta.kind()); + if (publishable != null) progress.commitNarration(publishable); + progress.onPendingNarration(narration); + forceFlush = true; + } + } else { + if (delta.thinking() != null) progress.onThinkingDelta(delta.thinking()); + if (delta.content() != null) { + accumulator.append(delta.content()); + progress.onContentDelta(delta.content()); + } + } + streamingCardManager.updateContent(sessionKey, progress.snapshot(), forceFlush); }) .doOnError(err -> { log.error("[feishu-stream] stream error: sessionKey={}, err={}", sessionKey, err.getMessage()); - streamingCardManager.failCard(sessionKey, err.getMessage()); }) .blockLast(Duration.ofMinutes(5)); String finalContent = accumulator.toString(); - // Card streaming never touches renderAndSend, so the channel's - // message-filter config has to be applied here — otherwise - // filter_thinking / filter_tool_messages are inert on this path. + // Card streaming never touches renderAndSend, so apply the same + // outbound filters before the final card snapshot is assembled. String cardContent = filterOutboundContent(finalContent); if (cardContent.isBlank()) { + cardContent = ""; + } + if (progressEnabled) { + String heldNarration = narrationTracker.settle(!cardContent.isBlank()); + if (heldNarration != null && !sameOutboundText(heldNarration, cardContent)) { + progress.commitNarration(heldNarration); + } + progress.clearPendingNarration(); + cardContent = progress.completedSnapshot(cardContent); + } else if (cardContent.isBlank()) { cardContent = "（无回复内容）"; } // Strip any /api/v1/files/generated/{id} URLs out of the card @@ -2629,15 +2675,34 @@ public class FeishuChannelAdapter extends AbstractChannelAdapter implements Stre // actual file. Cache-miss URLs fall back to the user-facing // retry hint that GeneratedFileScrubber emits. String renderedContent = scrubAndSendAttachments(receiveId, cardContent); - streamingCardManager.finishCard(sessionKey, renderedContent); + FeishuStreamingCardManager.FinishResult finishResult = + streamingCardManager.finishCard(sessionKey, renderedContent); + if (!finishResult.success()) { + // The card was delivered but either its terminal content or + // streaming-mode close was rejected. A regular message is the + // only reliable fallback after both CardKit attempts fail. + log.warn("[feishu-stream] Card finalization incomplete (contentUpdated={}, closed={}); " + + "falling back to regular message: sessionKey={}", + finishResult.finalContentUpdated(), finishResult.streamingClosed(), sessionKey); + sendMessage(receiveId, renderedContent); + } + if (!finishResult.streamingClosed()) { + log.warn("[feishu-stream] Card streaming mode could not be closed after retry: sessionKey={}", + sessionKey); + } log.info("[feishu-stream] Card streaming completed: sessionKey={}, contentLen={}", sessionKey, renderedContent.length()); - return finalContent.isBlank() ? cardContent : finalContent; + // Execution-trace text is channel presentation only. Never return + // it to the router as assistant content or it will pollute the + // next turn's LLM history. Preserve the legacy empty placeholder + // only when progress rendering was explicitly disabled. + return progressEnabled + ? finalContent + : (finalContent.isBlank() ? cardContent : finalContent); } catch (Exception e) { log.error("[feishu-stream] Card streaming failed: sessionKey={}, err={}", sessionKey, e.getMessage(), e); - streamingCardManager.failCard(sessionKey, e.getMessage()); // Tag returned content with the "[错误] " prefix so // ChannelMessageRouter.isErrorReply flips status='error' on the @@ -2647,6 +2712,17 @@ public class FeishuChannelAdapter extends AbstractChannelAdapter implements Stre // as a valid assistant turn and re-trigger the same 400. String partial = accumulator.toString(); String errorPrefix = "[错误] Feishu CardKit streaming failed: " + e.getMessage(); + FeishuStreamingCardManager.FinishResult failureResult = + streamingCardManager.failCard(sessionKey, e.getMessage()); + if (!failureResult.success()) { + String fallbackError = partial.isBlank() + ? "⚠️ 处理失败：" + e.getMessage() + : partial + "\n\n⚠️ 处理失败：" + e.getMessage(); + log.warn("[feishu-stream] Error card finalization incomplete; sending regular fallback: " + + "sessionKey={}, contentUpdated={}, closed={}", + sessionKey, failureResult.finalContentUpdated(), failureResult.streamingClosed()); + sendMessage(receiveId, fallbackError); + } if (!partial.isBlank()) { return errorPrefix + "\n\n（已生成的部分内容，已忽略）\n" + partial; } @@ -2654,6 +2730,14 @@ public class FeishuChannelAdapter extends AbstractChannelAdapter implements Stre } } + /** Compare text after the same outbound filters the receiver sees. */ + private boolean sameOutboundText(String a, String b) { + if (a == null || b == null) return false; + String left = filterOutboundContent(a).trim(); + String right = filterOutboundContent(b).trim(); + return !left.isEmpty() && left.equals(right); + } + /** * Streaming fallback — accumulate all deltas, then send through the * existing {@link #sendMessage} path so the message goes out as a diff --git a/mateclaw-server/src/main/java/vip/mate/channel/feishu/FeishuProgressRenderer.java b/mateclaw-server/src/main/java/vip/mate/channel/feishu/FeishuProgressRenderer.java new file mode 100644 index 00000000..b4c30dc6 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/channel/feishu/FeishuProgressRenderer.java @@ -0,0 +1,260 @@ +package vip.mate.channel.feishu; + +import java.util.ArrayDeque; +import java.util.Deque; +import java.util.Map; + +/** + * Builds the execution trace rendered inside a Feishu CardKit streaming card. + * + *

The renderer deliberately separates user-visible progress from persisted + * assistant content. The adapter returns only the final answer to the router, + * while this class keeps a bounded live trace in the card: phase, plan step, + * tool transitions, optional model thinking, and grounded stage narration. + */ +final class FeishuProgressRenderer { + + private static final int MAX_TOOL_LINES = 3; + private static final int MAX_NARRATION_LINES = 3; + private static final int THINKING_WINDOW = 500; + 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 boolean showToolTrace; + private final Deque toolLines = new ArrayDeque<>(); + private final Deque committedNarrations = new ArrayDeque<>(); + private final StringBuilder thinkingTail = new StringBuilder(); + private final StringBuilder answerTail = new StringBuilder(); + + private int collapsedToolCount; + private boolean thinkingSeen; + private boolean contentSeen; + private boolean approvalPending; + private String planStepLine; + private String pendingNarration; + + FeishuProgressRenderer(long startedAtMillis, boolean showThinking, boolean showToolTrace) { + this.startedAtMillis = startedAtMillis; + this.showThinking = showThinking; + this.showToolTrace = showToolTrace; + } + + void onThinkingDelta(String delta) { + thinkingSeen = true; + if (showThinking && delta != null && !delta.isEmpty()) { + thinkingTail.append(delta); + trimLeading(thinkingTail, THINKING_WINDOW); + } + } + + void onContentDelta(String delta) { + contentSeen = true; + if (delta != null && !delta.isEmpty()) { + answerTail.append(delta); + trimLeading(answerTail, ANSWER_WINDOW); + } + } + + /** Returns true for transitions that should bypass the normal update throttle. */ + boolean onEvent(String eventType, Map 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; + } + } + } + + void onPendingNarration(String text) { + pendingNarration = normalize(text); + } + + void commitNarration(String text) { + String normalized = normalize(text); + if (normalized == null) return; + committedNarrations.addLast(normalized); + while (committedNarrations.size() > MAX_NARRATION_LINES) { + committedNarrations.removeFirst(); + } + } + + void clearPendingNarration() { + pendingNarration = null; + } + + boolean isApprovalPending() { + return approvalPending; + } + + String snapshot() { + StringBuilder sb = new StringBuilder(); + appendTrace(sb, statusLine(false), true); + if (answerTail.length() > 0) { + sb.append("\n\n---\n\n").append(answerTail); + } + return sb.toString(); + } + + String completedSnapshot(String finalAnswer) { + String answer = finalAnswer == null ? "" : finalAnswer.trim(); + StringBuilder sb = new StringBuilder(); + appendTrace(sb, statusLine(true), false); + if (!answer.isEmpty()) { + sb.append("\n\n---\n\n").append(answer); + } else if (approvalPending) { + sb.append("\n\n⏸️ 已暂停，等待工具审批。"); + } else { + sb.append("\n\n（本轮没有产生回复内容）"); + } + return sb.toString(); + } + + private void appendTrace(StringBuilder sb, String status, boolean includePending) { + sb.append("**执行轨迹**\n").append(status); + if (planStepLine != null) sb.append('\n').append(planStepLine); + appendToolLines(sb); + for (String narration : committedNarrations) { + sb.append("\n• ").append(narration); + } + if (includePending && pendingNarration != null) { + sb.append("\n• ").append(pendingNarration); + } + if (showThinking && thinkingTail.length() > 0) { + sb.append("\n\n> 💭 ") + .append(thinkingTail.toString().replace("\n", "\n> ")); + } + } + + private String statusLine(boolean completed) { + if (completed) return approvalPending ? "⏸️ 等待工具审批（" + elapsed() + "）" + : "✅ 已完成（" + elapsed() + "）"; + if (approvalPending) return "⏸️ 等待工具审批…（" + elapsed() + "）"; + if (contentSeen) return "✍️ 正在回复…（" + elapsed() + "）"; + ToolLine running = lastRunningTool(); + if (running != null) { + return showToolTrace + ? "🔧 正在调用 " + displayName(running) + "…（" + elapsed() + "）" + : "🔧 正在执行工具…（" + elapsed() + "）"; + } + return (thinkingSeen ? "💭" : "🤔") + " 思考中…（" + elapsed() + "）"; + } + + private void appendToolLines(StringBuilder sb) { + if (!showToolTrace) { + int completed = collapsedToolCount; + boolean running = false; + for (ToolLine line : toolLines) { + if (line.finishedAt() == null) running = true; + else completed++; + } + if (completed > 0) sb.append("\n✅ 已执行 ").append(completed).append(" 项工具"); + if (running && contentSeen) sb.append("\n🔧 工具运行中…"); + return; + } + if (collapsedToolCount > 0) sb.append("\n…等 ").append(collapsedToolCount).append(" 项已完成"); + for (ToolLine line : toolLines) { + if (line.finishedAt() == null) { + 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; + if ((callId != null && callId.equals(line.callId())) + || (callId == null && toolName != null && toolName.equals(line.name()))) { + match = line; + } + } + long now = System.currentTimeMillis(); + if (match == null) { + toolLines.addLast(new ToolLine(callId, toolName, now, now, success)); + } else { + Deque rebuilt = new ArrayDeque<>(toolLines.size()); + for (ToolLine line : toolLines) { + rebuilt.addLast(line == match + ? new ToolLine(match.callId(), match.name(), match.startedAt(), now, success) + : line); + } + toolLines.clear(); + toolLines.addAll(rebuilt); + } + compactToolLines(); + } + + private void compactToolLines() { + while (toolLines.size() > MAX_TOOL_LINES) { + ToolLine oldest = toolLines.peekFirst(); + if (oldest != null && oldest.finishedAt() == null) break; + toolLines.pollFirst(); + collapsedToolCount++; + } + } + + private String elapsed() { + long seconds = Math.max(0, (System.currentTimeMillis() - startedAtMillis) / 1000); + return seconds < 60 ? "已 " + seconds + " 秒" + : "已 " + (seconds / 60) + " 分 " + (seconds % 60) + " 秒"; + } + + private static String displayName(ToolLine line) { + return line.name() != null && !line.name().isBlank() ? line.name() : "工具"; + } + + private static String stringField(Map data, String key) { + Object value = data != null ? data.get(key) : null; + return value != null ? value.toString() : null; + } + + private static String normalize(String text) { + return text == null || text.isBlank() ? null : text.trim(); + } + + private static void trimLeading(StringBuilder sb, int maxLen) { + int excess = sb.length() - maxLen; + if (excess > 0) sb.delete(0, excess); + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/channel/feishu/FeishuStreamingCardManager.java b/mateclaw-server/src/main/java/vip/mate/channel/feishu/FeishuStreamingCardManager.java index 1b883057..9a406876 100644 --- a/mateclaw-server/src/main/java/vip/mate/channel/feishu/FeishuStreamingCardManager.java +++ b/mateclaw-server/src/main/java/vip/mate/channel/feishu/FeishuStreamingCardManager.java @@ -64,6 +64,13 @@ public class FeishuStreamingCardManager { /** Throttle window for {@link #appendContent}, ms — matches DingTalk AICard. */ static final long THROTTLE_INTERVAL_MS = 500; + /** + * Hard per-card operation spacing. Feishu allows at most 10 CardKit + * operations/second for one card; 120ms leaves a little clock/network + * jitter headroom while still letting phase transitions feel immediate. + */ + static final long PLATFORM_MIN_INTERVAL_MS = 120; + /** * Markdown element id baked into the initial streaming card. * Content-update calls reference this id. Public so tests can assert. @@ -91,6 +98,13 @@ public class FeishuStreamingCardManager { /** Terminal-state CAS guard — at most one of {finishCard, failCard} wins per session. */ enum Status { STREAMING, FINISHED, FAILED } + /** Result of the two independently fallible terminal CardKit operations. */ + public record FinishResult(boolean finalContentUpdated, boolean streamingClosed) { + public boolean success() { + return finalContentUpdated && streamingClosed; + } + } + /** * One in-flight streaming card. State is mutated by a single Reactor * thread per session (the one consuming the {@code Flux}), so all @@ -178,8 +192,8 @@ public class FeishuStreamingCardManager { } /** - * Append delta text to the running session. May flush immediately - * (force) or wait for the next throttle window. + * Append delta text to the running session. A forced update bypasses the + * normal 500ms UX throttle but still respects the platform hard limit. * *

No-op when {@code sessionKey} is unknown or the session has * already reached a terminal status — keeps the caller's @@ -195,11 +209,22 @@ public class FeishuStreamingCardManager { session.accumulated.append(contentDelta); } } - long now = currentTimeMs(); - if (!forceFlush && now - session.lastFlushMs < THROTTLE_INTERVAL_MS) { - return; - } - flush(session, now); + flushWithPolicy(session, forceFlush); + } + + /** + * Replace the streaming element with a full progress snapshot. + * + *

CardKit's content API expects the complete current text on every + * update. Agent progress is not append-only ("thinking" becomes "calling + * a tool", then "replying"), so treating snapshots as deltas duplicates + * the entire trace on every refresh. + */ + public boolean updateContent(String sessionKey, String fullContent, boolean forceFlush) { + CardSession session = activeSessions.get(sessionKey); + if (session == null || !session.isStreaming()) return false; + replaceAccumulated(session, fullContent != null ? fullContent : ""); + return flushWithPolicy(session, forceFlush); } /** @@ -207,20 +232,24 @@ public class FeishuStreamingCardManager { * a second call is a no-op. After return, the sessionKey is no * longer known to the manager. */ - public void finishCard(String sessionKey, String finalContent) { + public FinishResult finishCard(String sessionKey, String finalContent) { CardSession session = activeSessions.get(sessionKey); - if (session == null) return; + if (session == null) return new FinishResult(false, false); if (!session.status.compareAndSet(Status.STREAMING, Status.FINISHED)) { - return; + return new FinishResult(false, false); } + boolean contentUpdated = false; + boolean streamingClosed = false; try { replaceAccumulated(session, finalContent != null ? finalContent : ""); - flush(session, currentTimeMs()); - closeStreaming(session); + contentUpdated = flushWithRetry(session); + streamingClosed = closeStreamingWithRetry(session, summaryFor(finalContent)); + return new FinishResult(contentUpdated, streamingClosed); } finally { activeSessions.remove(sessionKey); - log.info("[feishu-stream] Card finished: sessionKey={}, contentLen={}", - sessionKey, finalContent == null ? 0 : finalContent.length()); + log.info("[feishu-stream] Card finished: sessionKey={}, contentLen={}, contentUpdated={}, closed={}", + sessionKey, finalContent == null ? 0 : finalContent.length(), + contentUpdated, streamingClosed); } } @@ -229,12 +258,14 @@ public class FeishuStreamingCardManager { * suffix; the card is closed so the typing animation stops. * Idempotent. */ - public void failCard(String sessionKey, String errorMessage) { + public FinishResult failCard(String sessionKey, String errorMessage) { CardSession session = activeSessions.get(sessionKey); - if (session == null) return; + if (session == null) return new FinishResult(false, false); if (!session.status.compareAndSet(Status.STREAMING, Status.FAILED)) { - return; + return new FinishResult(false, false); } + boolean contentUpdated = false; + boolean streamingClosed = false; try { String tail; synchronized (session) { @@ -246,11 +277,13 @@ public class FeishuStreamingCardManager { session.accumulated.setLength(0); session.accumulated.append(tail); } - flush(session, currentTimeMs()); - closeStreaming(session); + contentUpdated = flushWithRetry(session); + streamingClosed = closeStreamingWithRetry(session, "⚠️ 处理失败"); + return new FinishResult(contentUpdated, streamingClosed); } finally { activeSessions.remove(sessionKey); - log.warn("[feishu-stream] Card failed: sessionKey={}, error={}", sessionKey, errorMessage); + log.warn("[feishu-stream] Card failed: sessionKey={}, contentUpdated={}, closed={}, error={}", + sessionKey, contentUpdated, streamingClosed, errorMessage); } } @@ -272,7 +305,26 @@ public class FeishuStreamingCardManager { // Internal — flush + SDK seams // ------------------------------------------------------------------ - private void flush(CardSession session, long now) { + private boolean flushWithPolicy(CardSession session, boolean forceFlush) { + long now = currentTimeMs(); + long elapsed = now - session.lastFlushMs; + if (!forceFlush && elapsed < THROTTLE_INTERVAL_MS) { + return true; // latest snapshot is queued in session.accumulated + } + if (forceFlush && elapsed < PLATFORM_MIN_INTERVAL_MS) { + if (!pauseBeforeFlush(PLATFORM_MIN_INTERVAL_MS - elapsed)) return false; + now = currentTimeMs(); + } + return flush(session, now); + } + + /** One retry is enough to cover a transient rate-limit/network blip. */ + private boolean flushWithRetry(CardSession session) { + if (flushWithPolicy(session, true)) return true; + return flushWithPolicy(session, true); + } + + private boolean flush(CardSession session, long now) { String snapshot; synchronized (session) { snapshot = session.accumulated.toString(); @@ -281,27 +333,45 @@ public class FeishuStreamingCardManager { try { Client client = clientFactory.client(session.channelId); sdkPushElementContent(client, session.cardId, STREAM_ELEMENT_ID, snapshot, seq); - session.lastFlushMs = now; + return true; } catch (Exception e) { log.warn("[feishu-stream] flush failed: sessionKey={}, seq={}, err={}", session.sessionKey, seq, e.getMessage()); + return false; + } finally { + // Failed requests count against platform rate limits too. + session.lastFlushMs = now; } } - private void closeStreaming(CardSession session) { + private boolean closeStreamingWithRetry(CardSession session, String summary) { + if (closeStreaming(session, summary)) return true; + return closeStreaming(session, summary); + } + + private boolean closeStreaming(CardSession session, String summary) { + long elapsed = currentTimeMs() - session.lastFlushMs; + if (elapsed < PLATFORM_MIN_INTERVAL_MS + && !pauseBeforeFlush(PLATFORM_MIN_INTERVAL_MS - elapsed)) { + return false; + } int seq = session.sequence.incrementAndGet(); try { Client client = clientFactory.client(session.channelId); - sdkCloseStreamingMode(client, session.cardId, seq); + sdkCloseStreamingMode(client, session.cardId, seq, summary); + return true; } catch (Exception e) { log.warn("[feishu-stream] closeStreaming failed: sessionKey={}, err={}", session.sessionKey, e.getMessage()); + return false; + } finally { + session.lastFlushMs = currentTimeMs(); } } private void tryCloseStreamingSilently(Client client, String cardId) { try { - sdkCloseStreamingMode(client, cardId, 1); + sdkCloseStreamingMode(client, cardId, 1, "⚠️ 卡片发送失败"); } catch (Exception ignore) { // best-effort — already in an error path } @@ -314,6 +384,31 @@ public class FeishuStreamingCardManager { } } + private boolean pauseBeforeFlush(long millis) { + if (millis <= 0) return true; + try { + sleepMillis(millis); + return true; + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return false; + } + } + + /** Test seam for advancing a fake clock without real sleeping. */ + protected void sleepMillis(long millis) throws InterruptedException { + Thread.sleep(millis); + } + + static String summaryFor(String content) { + String preview = content == null ? "" : content + .replaceAll("[`*_>#~-]+", " ") + .replaceAll("\\s+", " ") + .trim(); + if (preview.isEmpty()) return "✅ 已完成"; + return preview.length() <= 80 ? preview : preview.substring(0, 77) + "..."; + } + // ------------------------------------------------------------------ // SDK seams (overridable in tests) // ------------------------------------------------------------------ @@ -373,15 +468,18 @@ public class FeishuStreamingCardManager { .build(); ContentCardElementResp resp = client.cardkit().v1().cardElement().content(req); if (!resp.success()) { - log.warn("[feishu-stream] cardElement.content failed: cardId={}, seq={}, code={}, msg={}", - abbrev(cardId), sequence, resp.getCode(), resp.getMsg()); + throw new IllegalStateException("cardElement.content failed: cardId=" + abbrev(cardId) + + ", seq=" + sequence + ", code=" + resp.getCode() + ", msg=" + resp.getMsg()); } } /** Flip streaming_mode=false so the receiving UI stops the typing animation. */ - protected void sdkCloseStreamingMode(Client client, String cardId, int sequence) throws Exception { + protected void sdkCloseStreamingMode(Client client, String cardId, int sequence, + String summary) throws Exception { Map settings = Map.of( - "config", Map.of("streaming_mode", false) + "config", Map.of( + "streaming_mode", false, + "summary", Map.of("content", summaryFor(summary))) ); SettingsCardReq req = SettingsCardReq.newBuilder() .cardId(cardId) @@ -393,8 +491,8 @@ public class FeishuStreamingCardManager { .build(); SettingsCardResp resp = client.cardkit().v1().card().settings(req); if (!resp.success()) { - log.warn("[feishu-stream] card.settings (close) failed: cardId={}, code={}, msg={}", - abbrev(cardId), resp.getCode(), resp.getMsg()); + throw new IllegalStateException("card.settings failed: cardId=" + abbrev(cardId) + + ", code=" + resp.getCode() + ", msg=" + resp.getMsg()); } } diff --git a/mateclaw-server/src/main/resources/docs/en/channels.md b/mateclaw-server/src/main/resources/docs/en/channels.md index a6d77e31..75d2c40c 100644 --- a/mateclaw-server/src/main/resources/docs/en/channels.md +++ b/mateclaw-server/src/main/resources/docs/en/channels.md @@ -295,8 +295,12 @@ Tool-guard approval flows arrive as a card with **Approve / Deny** buttons. Tapp Replies stream char-by-char into a **single card** instead of waiting for the whole answer before sending. - `card_streaming_enabled` (default `true`) -- The first token appears immediately; subsequent updates are throttled at 500ms +- The first token appears immediately; regular text updates coalesce at 500ms, while phase transitions refresh preferentially behind a 120ms platform safety limit +- `stream_progress` (default `true`) keeps thinking status, plan steps, tool progress, and stage narration in the same card; completion retains a bounded execution trace above the final answer +- Set `filter_thinking=false` to show raw model thinking; by default only status and stage progress are shown +- Set `filter_tool_messages=false` to show tool names and per-tool results; by default only the tool count is shown - On CardKit failure it falls back to accumulate-then-send +- Final update and close operations retry once; if they still fail, a regular Feishu message carries the answer #### Inbound voice transcription @@ -354,6 +358,7 @@ curl -X POST http://localhost:18088/api/v1/channels \ "card_format": "auto", "card_header": "AI 助手", "card_streaming_enabled": true, + "stream_progress": true, "media_download_enabled": true, "enable_done_reaction": true, "require_mention": false diff --git a/mateclaw-server/src/main/resources/docs/zh/channels.md b/mateclaw-server/src/main/resources/docs/zh/channels.md index 86b6edbe..19faa59d 100644 --- a/mateclaw-server/src/main/resources/docs/zh/channels.md +++ b/mateclaw-server/src/main/resources/docs/zh/channels.md @@ -295,8 +295,12 @@ JSON 卡片 payload 上限约 32 KB，超出后自动降级为纯文本。 回复逐字刷新进**同一张卡片**，而不是等整段生成完再发。 - `card_streaming_enabled`（默认 `true`） -- 首 token 立即出现，之后按 500ms 节流刷新 +- 首 token 立即出现；普通文本按 500ms 合并刷新，阶段切换遵守 120ms 平台硬限流后优先刷新 +- `stream_progress`（默认 `true`）：同一卡片会展示思考状态、计划步骤、工具进度和阶段旁白，完成后保留一份有界执行轨迹并追加最终回答 +- `filter_thinking=false` 时展示模型原始思考文本；默认仅展示状态与阶段轨迹，不暴露原始思考 +- `filter_tool_messages=false` 时展示工具名称和逐项结果；默认只展示工具执行数量 - CardKit 调用失败时自动回退到"先攒齐再一次性发出" +- 最终更新和关闭操作会自动重试一次；仍失败则通过普通飞书消息兜底 #### 入站语音转写 @@ -354,6 +358,7 @@ curl -X POST http://localhost:18088/api/v1/channels \ "card_format": "auto", "card_header": "AI 助手", "card_streaming_enabled": true, + "stream_progress": true, "media_download_enabled": true, "enable_done_reaction": true, "require_mention": false diff --git a/mateclaw-server/src/test/java/vip/mate/channel/feishu/FeishuProcessStreamTest.java b/mateclaw-server/src/test/java/vip/mate/channel/feishu/FeishuProcessStreamTest.java new file mode 100644 index 00000000..f8b2e263 --- /dev/null +++ b/mateclaw-server/src/test/java/vip/mate/channel/feishu/FeishuProcessStreamTest.java @@ -0,0 +1,216 @@ +package vip.mate.channel.feishu; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.lark.oapi.Client; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import reactor.core.publisher.Flux; +import vip.mate.agent.AgentService.StreamDelta; +import vip.mate.agent.ContentKind; +import vip.mate.channel.ChannelMessage; +import vip.mate.channel.ChannelMessageRouter; +import vip.mate.channel.model.ChannelEntity; + +import java.util.List; +import java.util.Map; +import java.util.concurrent.CopyOnWriteArrayList; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +/** End-to-end stream rendering at the Feishu adapter/CardKit boundary. */ +class FeishuProcessStreamTest { + + private static final class RecordingManager extends FeishuStreamingCardManager { + final List snapshots = new CopyOnWriteArrayList<>(); + volatile boolean failCompletedSnapshots; + volatile boolean failErrorSnapshots; + volatile boolean failClose; + + RecordingManager(FeishuClientFactory factory, ObjectMapper mapper) { + super(factory, mapper); + } + + @Override protected String sdkCreateCard(Client client, String initialText) { return "card_1"; } + @Override protected String sdkSendInteractiveMessage(Client client, String receiveIdType, + String receiveId, String cardId) { return "msg_1"; } + @Override protected void sdkPushElementContent(Client client, String cardId, String elementId, + String content, int sequence) { + if (failCompletedSnapshots && content.contains("✅ 已完成")) { + throw new IllegalStateException("simulated final update failure"); + } + if (failErrorSnapshots && content.contains("⚠️")) { + throw new IllegalStateException("simulated error update failure"); + } + snapshots.add(content); + } + @Override protected void sdkCloseStreamingMode(Client client, String cardId, int sequence, + String summary) { + if (failClose) throw new IllegalStateException("simulated close failure"); + } + @Override protected void sleepMillis(long millis) {} + } + + private static final class RecordingAdapter extends FeishuChannelAdapter { + final List fallbackMessages = new CopyOnWriteArrayList<>(); + + RecordingAdapter(ChannelEntity entity, ObjectMapper mapper, RecordingManager manager) { + super(entity, mock(ChannelMessageRouter.class), mapper, null, null, manager); + } + + @Override public void sendMessage(String targetId, String content) { + fallbackMessages.add(content); + } + } + + @Test + @DisplayName("default Feishu card shows a filtered execution trace and final answer") + void defaultTraceShowsGenericToolsWithoutRawThinking() { + Fixture f = fixture("{}"); + Flux 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)); + + assertEquals("现在是下午三点。", f.adapter.processStream(stream, inbound(), "feishu:test")); + + String finalCard = f.manager.snapshots.get(f.manager.snapshots.size() - 1); + assertTrue(finalCard.contains("执行轨迹")); + assertTrue(finalCard.contains("已执行 1 项工具")); + assertTrue(finalCard.contains("现在是下午三点。")); + assertFalse(finalCard.contains("get_time"), "default tool filter must hide tool identity"); + assertFalse(finalCard.contains("内部推理文本"), "default thinking filter must hide raw thinking"); + } + + @Test + @DisplayName("Feishu card honors unfiltered thinking and tool-detail settings") + void unfilteredTraceShowsThinkingAndToolName() { + Fixture f = fixture("{\"filter_thinking\":false,\"filter_tool_messages\":false}"); + Flux 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)); + + f.adapter.processStream(stream, inbound(), "feishu:test"); + + String finalCard = f.manager.snapshots.get(f.manager.snapshots.size() - 1); + assertTrue(finalCard.contains("先读取当前时间")); + assertTrue(finalCard.contains("get_time")); + assertTrue(finalCard.contains("完成。")); + } + + @Test + @DisplayName("pre-tool rehearsal remains live but is removed from the completed Feishu card") + void provisionalNarrationDoesNotBecomePermanent() { + Fixture f = fixture("{}"); + Flux stream = Flux.just( + StreamDelta.segmentOnly("预测温度是 29 度。", null, ContentKind.PRE_TOOL_NARRATION), + StreamDelta.event("tool_call_started", + Map.of("toolCallId", "c1", "toolName", "query_env")), + StreamDelta.event("tool_call_completed", + Map.of("toolCallId", "c1", "toolName", "query_env", "success", true)), + new StreamDelta("接口没有返回环境数据。", null)); + + f.adapter.processStream(stream, inbound(), "feishu:test"); + + assertTrue(f.manager.snapshots.stream().anyMatch(s -> s.contains("预测温度是 29 度")), + "provisional narration should be visible while work is in progress"); + String finalCard = f.manager.snapshots.get(f.manager.snapshots.size() - 1); + assertFalse(finalCard.contains("29 度"), "superseded rehearsal must not survive completion"); + assertTrue(finalCard.contains("接口没有返回环境数据")); + } + + @Test + @DisplayName("execution trace is presentation-only and an empty turn stays empty for persistence") + void emptyTurnDoesNotPersistTrace() { + Fixture f = fixture("{}"); + Flux stream = Flux.just( + StreamDelta.event("tool_call_started", + Map.of("toolCallId", "c1", "toolName", "approval_tool")), + StreamDelta.event("tool_approval_requested", Map.of("toolCallId", "c1"))); + + assertEquals("", f.adapter.processStream(stream, inbound(), "feishu:test"), + "the router must never persist the rendered execution trace as assistant content"); + String finalCard = f.manager.snapshots.get(f.manager.snapshots.size() - 1); + assertTrue(finalCard.contains("等待工具审批")); + } + + @Test + @DisplayName("failed terminal CardKit update falls back to a regular Feishu message") + void failedFinalCardUpdateFallsBackToRegularMessage() { + Fixture f = fixture("{}"); + f.manager.failCompletedSnapshots = true; + + assertEquals("最终答案", f.adapter.processStream( + Flux.just(new StreamDelta("最终答案", null)), inbound(), "feishu:test")); + + assertEquals(1, f.adapter.fallbackMessages.size()); + assertTrue(f.adapter.fallbackMessages.get(0).contains("最终答案")); + } + + @Test + @DisplayName("failed streaming close also falls back after retry") + void failedStreamingCloseFallsBackToRegularMessage() { + Fixture f = fixture("{}"); + f.manager.failClose = true; + + f.adapter.processStream(Flux.just(new StreamDelta("最终答案", null)), + inbound(), "feishu:test"); + + assertEquals(1, f.adapter.fallbackMessages.size()); + assertTrue(f.adapter.fallbackMessages.get(0).contains("最终答案")); + } + + @Test + @DisplayName("failed error-card update also sends a regular error fallback") + void failedErrorCardUpdateFallsBackToRegularMessage() { + Fixture f = fixture("{}"); + f.manager.failErrorSnapshots = true; + Flux stream = Flux.concat( + Flux.just(new StreamDelta("部分回答", null)), + Flux.error(new IllegalStateException("upstream failed"))); + + String result = f.adapter.processStream(stream, inbound(), "feishu:test"); + + assertTrue(result.startsWith("[错误]")); + assertEquals(1, f.adapter.fallbackMessages.size()); + assertTrue(f.adapter.fallbackMessages.get(0).contains("upstream failed")); + } + + private static ChannelMessage inbound() { + return ChannelMessage.builder() + .channelType("feishu") + .senderId("ou_user") + .replyToken("oc_chat") + .content("hi") + .build(); + } + + private static Fixture fixture(String configJson) { + ObjectMapper mapper = new ObjectMapper(); + FeishuClientFactory factory = mock(FeishuClientFactory.class); + when(factory.client(anyLong())).thenReturn(mock(Client.class)); + when(factory.client(any())).thenReturn(mock(Client.class)); + RecordingManager manager = new RecordingManager(factory, mapper); + + ChannelEntity entity = new ChannelEntity(); + entity.setId(1L); + entity.setChannelType("feishu"); + entity.setConfigJson(configJson); + RecordingAdapter adapter = new RecordingAdapter(entity, mapper, manager); + return new Fixture(adapter, manager); + } + + private record Fixture(RecordingAdapter adapter, RecordingManager manager) {} +} diff --git a/mateclaw-server/src/test/java/vip/mate/channel/feishu/FeishuStreamingCardManagerTest.java b/mateclaw-server/src/test/java/vip/mate/channel/feishu/FeishuStreamingCardManagerTest.java index ee8d206e..b8a7bc98 100644 --- a/mateclaw-server/src/test/java/vip/mate/channel/feishu/FeishuStreamingCardManagerTest.java +++ b/mateclaw-server/src/test/java/vip/mate/channel/feishu/FeishuStreamingCardManagerTest.java @@ -8,6 +8,7 @@ import org.junit.jupiter.api.Test; import java.util.List; import java.util.Map; +import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicReference; @@ -31,7 +32,7 @@ import static org.mockito.Mockito.when; *

Behaviour pinned: *