diff --git a/CHANGELOG.md b/CHANGELOG.md index a10f9fa1ad3..72ecbbf0e83 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -35,6 +35,7 @@ Docs: https://docs.openclaw.ai - Runtime/tool-state stability: recover from dangling Anthropic `tool_use` after compaction, serialize long-running Discord handler runs without blocking new inbound events, and prevent stale busy snapshots from suppressing stuck-channel recovery. (from #33630, #33583) Thanks @kevinWangSheng and @theotarr. - ACP/Discord startup hardening: clean up stuck ACP worker children on gateway restart, unbind stale ACP thread bindings during Discord startup reconciliation, and add per-thread listener watchdog timeouts so wedged turns cannot block later messages. (#33699) Thanks @dutifulbob. - Extensions/media local-root propagation: consistently forward `mediaLocalRoots` through extension `sendMedia` adapters (Google Chat, Slack, iMessage, Signal, WhatsApp), preserving non-local media behavior while restoring local attachment resolution from configured roots. Synthesis of #33581, #33545, #33540, #33536, #33528. Thanks @bmendonca3. +- Feishu/video media send contract: keep mp4-like outbound payloads on `msg_type: "media"` (including reply and reply-in-thread paths) so videos render as media instead of degrading to file-link behavior, while preserving existing non-video file subtype handling. (from #33720, #33808, #33678) Thanks @polooooo, @dingjianrui, and @kevinWangSheng. - Gateway/security default response headers: add `Permissions-Policy: camera=(), microphone=(), geolocation=()` to baseline gateway HTTP security headers for all responses. (#30186) thanks @habakan. - Plugins/startup loading: lazily initialize plugin runtime, split startup-critical plugin SDK imports into `openclaw/plugin-sdk/core` and `openclaw/plugin-sdk/telegram`, and preserve `api.runtime` reflection semantics for plugin compatibility. (#28620) thanks @hmemcpy. - Plugins/startup performance: reduce bursty plugin discovery/manifest overhead with short in-process caches, skip importing bundled memory plugins that are disabled by slot selection, and speed legacy root `openclaw/plugin-sdk` compatibility via runtime root-alias routing while preserving backward compatibility. Thanks @gumadeiras. diff --git a/extensions/feishu/src/media.test.ts b/extensions/feishu/src/media.test.ts index dd31b015404..336a2d425c4 100644 --- a/extensions/feishu/src/media.test.ts +++ b/extensions/feishu/src/media.test.ts @@ -113,7 +113,7 @@ describe("sendMediaFeishu msg_type routing", () => { messageResourceGetMock.mockResolvedValue(Buffer.from("resource-bytes")); }); - it("uses msg_type=file for mp4", async () => { + it("uses msg_type=media for mp4 video", async () => { await sendMediaFeishu({ cfg: {} as any, to: "user:ou_target", @@ -129,7 +129,7 @@ describe("sendMediaFeishu msg_type routing", () => { expect(messageCreateMock).toHaveBeenCalledWith( expect.objectContaining({ - data: expect.objectContaining({ msg_type: "file" }), + data: expect.objectContaining({ msg_type: "media" }), }), ); }); @@ -176,7 +176,7 @@ describe("sendMediaFeishu msg_type routing", () => { ); }); - it("uses msg_type=file when replying with mp4", async () => { + it("uses msg_type=media when replying with mp4", async () => { await sendMediaFeishu({ cfg: {} as any, to: "user:ou_target", @@ -188,7 +188,7 @@ describe("sendMediaFeishu msg_type routing", () => { expect(messageReplyMock).toHaveBeenCalledWith( expect.objectContaining({ path: { message_id: "om_parent" }, - data: expect.objectContaining({ msg_type: "file" }), + data: expect.objectContaining({ msg_type: "media" }), }), ); @@ -208,7 +208,10 @@ describe("sendMediaFeishu msg_type routing", () => { expect(messageReplyMock).toHaveBeenCalledWith( expect.objectContaining({ path: { message_id: "om_parent" }, - data: expect.objectContaining({ msg_type: "file", reply_in_thread: true }), + data: expect.objectContaining({ + msg_type: "media", + reply_in_thread: true, + }), }), ); }); diff --git a/extensions/feishu/src/media.ts b/extensions/feishu/src/media.ts index 42f98ab7305..41b6a7c6c4d 100644 --- a/extensions/feishu/src/media.ts +++ b/extensions/feishu/src/media.ts @@ -328,8 +328,8 @@ export async function sendFileFeishu(params: { cfg: ClawdbotConfig; to: string; fileKey: string; - /** Use "audio" for audio files, "file" for documents and video */ - msgType?: "file" | "audio"; + /** Use "audio" for audio, "media" for video (mp4), "file" for documents */ + msgType?: "file" | "audio" | "media"; replyToMessageId?: string; replyInThread?: boolean; accountId?: string; @@ -467,8 +467,8 @@ export async function sendMediaFeishu(params: { fileType, accountId, }); - // Feishu API: opus -> "audio", everything else (including video) -> "file" - const msgType = fileType === "opus" ? "audio" : "file"; + // Feishu API: opus -> "audio", mp4/video -> "media" (playable), others -> "file" + const msgType = fileType === "opus" ? "audio" : fileType === "mp4" ? "media" : "file"; return sendFileFeishu({ cfg, to, diff --git a/extensions/feishu/src/reply-dispatcher.test.ts b/extensions/feishu/src/reply-dispatcher.test.ts index ace7b2cc2db..7f25db5e417 100644 --- a/extensions/feishu/src/reply-dispatcher.test.ts +++ b/extensions/feishu/src/reply-dispatcher.test.ts @@ -26,6 +26,23 @@ vi.mock("./typing.js", () => ({ removeTypingIndicator: removeTypingIndicatorMock, })); vi.mock("./streaming-card.js", () => ({ + mergeStreamingText: (previousText: string | undefined, nextText: string | undefined) => { + const previous = typeof previousText === "string" ? previousText : ""; + const next = typeof nextText === "string" ? nextText : ""; + if (!next) { + return previous; + } + if (!previous || next === previous) { + return next; + } + if (next.startsWith(previous)) { + return next; + } + if (previous.startsWith(next)) { + return previous; + } + return `${previous}${next}`; + }, FeishuStreamingSession: class { active = false; start = vi.fn(async () => { @@ -244,6 +261,116 @@ describe("createFeishuReplyDispatcher streaming behavior", () => { expect(streamingInstances[0].close).toHaveBeenCalledWith("```md\npartial answer\n```"); }); + it("delivers distinct final payloads after streaming close", async () => { + createFeishuReplyDispatcher({ + cfg: {} as never, + agentId: "agent", + runtime: { log: vi.fn(), error: vi.fn() } as never, + chatId: "oc_chat", + }); + + const options = createReplyDispatcherWithTypingMock.mock.calls[0]?.[0]; + await options.deliver({ text: "```md\n完整回复第一段\n```" }, { kind: "final" }); + await options.deliver({ text: "```md\n完整回复第一段 + 第二段\n```" }, { kind: "final" }); + + expect(streamingInstances).toHaveLength(2); + expect(streamingInstances[0].close).toHaveBeenCalledTimes(1); + expect(streamingInstances[0].close).toHaveBeenCalledWith("```md\n完整回复第一段\n```"); + expect(streamingInstances[1].close).toHaveBeenCalledTimes(1); + expect(streamingInstances[1].close).toHaveBeenCalledWith("```md\n完整回复第一段 + 第二段\n```"); + expect(sendMessageFeishuMock).not.toHaveBeenCalled(); + expect(sendMarkdownCardFeishuMock).not.toHaveBeenCalled(); + }); + + it("skips exact duplicate final text after streaming close", async () => { + createFeishuReplyDispatcher({ + cfg: {} as never, + agentId: "agent", + runtime: { log: vi.fn(), error: vi.fn() } as never, + chatId: "oc_chat", + }); + + const options = createReplyDispatcherWithTypingMock.mock.calls[0]?.[0]; + await options.deliver({ text: "```md\n同一条回复\n```" }, { kind: "final" }); + await options.deliver({ text: "```md\n同一条回复\n```" }, { kind: "final" }); + + expect(streamingInstances).toHaveLength(1); + expect(streamingInstances[0].close).toHaveBeenCalledTimes(1); + expect(streamingInstances[0].close).toHaveBeenCalledWith("```md\n同一条回复\n```"); + expect(sendMessageFeishuMock).not.toHaveBeenCalled(); + expect(sendMarkdownCardFeishuMock).not.toHaveBeenCalled(); + }); + + it("suppresses duplicate final text while still sending media", async () => { + resolveFeishuAccountMock.mockReturnValue({ + accountId: "main", + appId: "app_id", + appSecret: "app_secret", + domain: "feishu", + config: { + renderMode: "auto", + streaming: false, + }, + }); + + createFeishuReplyDispatcher({ + cfg: {} as never, + agentId: "agent", + runtime: { log: vi.fn(), error: vi.fn() } as never, + chatId: "oc_chat", + }); + + const options = createReplyDispatcherWithTypingMock.mock.calls[0]?.[0]; + await options.deliver({ text: "plain final" }, { kind: "final" }); + await options.deliver( + { text: "plain final", mediaUrl: "https://example.com/a.png" }, + { kind: "final" }, + ); + + expect(sendMessageFeishuMock).toHaveBeenCalledTimes(1); + expect(sendMessageFeishuMock).toHaveBeenLastCalledWith( + expect.objectContaining({ + text: "plain final", + }), + ); + expect(sendMediaFeishuMock).toHaveBeenCalledTimes(1); + expect(sendMediaFeishuMock).toHaveBeenCalledWith( + expect.objectContaining({ + mediaUrl: "https://example.com/a.png", + }), + ); + }); + + it("treats block updates as delta chunks", async () => { + resolveFeishuAccountMock.mockReturnValue({ + accountId: "main", + appId: "app_id", + appSecret: "app_secret", + domain: "feishu", + config: { + renderMode: "card", + streaming: true, + }, + }); + + const result = createFeishuReplyDispatcher({ + cfg: {} as never, + agentId: "agent", + runtime: { log: vi.fn(), error: vi.fn() } as never, + chatId: "oc_chat", + }); + + const options = createReplyDispatcherWithTypingMock.mock.calls[0]?.[0]; + await options.onReplyStart?.(); + await result.replyOptions.onPartialReply?.({ text: "hello" }); + await options.deliver({ text: "lo world" }, { kind: "block" }); + await options.onIdle?.(); + + expect(streamingInstances).toHaveLength(1); + expect(streamingInstances[0].close).toHaveBeenCalledTimes(1); + expect(streamingInstances[0].close).toHaveBeenCalledWith("hellolo world"); + }); + it("sends media-only payloads as attachments", async () => { createFeishuReplyDispatcher({ cfg: {} as never, diff --git a/extensions/feishu/src/reply-dispatcher.ts b/extensions/feishu/src/reply-dispatcher.ts index 857e4cec023..58ca55eef28 100644 --- a/extensions/feishu/src/reply-dispatcher.ts +++ b/extensions/feishu/src/reply-dispatcher.ts @@ -13,7 +13,7 @@ import type { MentionTarget } from "./mention.js"; import { buildMentionedCardContent } from "./mention.js"; import { getFeishuRuntime } from "./runtime.js"; import { sendMarkdownCardFeishu, sendMessageFeishu } from "./send.js"; -import { FeishuStreamingSession } from "./streaming-card.js"; +import { FeishuStreamingSession, mergeStreamingText } from "./streaming-card.js"; import { resolveReceiveIdType } from "./targets.js"; import { addTypingIndicator, removeTypingIndicator, type TypingIndicatorState } from "./typing.js"; @@ -143,29 +143,16 @@ export function createFeishuReplyDispatcher(params: CreateFeishuReplyDispatcherP let streaming: FeishuStreamingSession | null = null; let streamText = ""; let lastPartial = ""; + let lastFinalText: string | null = null; let partialUpdateQueue: Promise = Promise.resolve(); let streamingStartPromise: Promise | null = null; - - const mergeStreamingText = (nextText: string) => { - if (!streamText) { - streamText = nextText; - return; - } - if (nextText.startsWith(streamText)) { - // Handle cumulative partial payloads where nextText already includes prior text. - streamText = nextText; - return; - } - if (streamText.endsWith(nextText)) { - return; - } - streamText += nextText; - }; + type StreamTextUpdateMode = "snapshot" | "delta"; const queueStreamingUpdate = ( nextText: string, options?: { dedupeWithLastPartial?: boolean; + mode?: StreamTextUpdateMode; }, ) => { if (!nextText) { @@ -177,7 +164,9 @@ export function createFeishuReplyDispatcher(params: CreateFeishuReplyDispatcherP if (options?.dedupeWithLastPartial) { lastPartial = nextText; } - mergeStreamingText(nextText); + const mode = options?.mode ?? "snapshot"; + streamText = + mode === "delta" ? `${streamText}${nextText}` : mergeStreamingText(streamText, nextText); partialUpdateQueue = partialUpdateQueue.then(async () => { if (streamingStartPromise) { await streamingStartPromise; @@ -241,6 +230,7 @@ export function createFeishuReplyDispatcher(params: CreateFeishuReplyDispatcherP responsePrefixContextProvider: prefixContext.responsePrefixContextProvider, humanDelay: core.channel.reply.resolveHumanDelayConfig(cfg, agentId), onReplyStart: () => { + lastFinalText = null; if (streamingEnabled && renderMode === "card") { startStreaming(); } @@ -256,12 +246,17 @@ export function createFeishuReplyDispatcher(params: CreateFeishuReplyDispatcherP : []; const hasText = Boolean(text.trim()); const hasMedia = mediaList.length > 0; + // Suppress only exact duplicate final text payloads to avoid + // dropping legitimate multi-part final replies. + const skipTextForDuplicateFinal = + info?.kind === "final" && hasText && lastFinalText === text; + const shouldDeliverText = hasText && !skipTextForDuplicateFinal; - if (!hasText && !hasMedia) { + if (!shouldDeliverText && !hasMedia) { return; } - if (hasText) { + if (shouldDeliverText) { const useCard = renderMode === "card" || (renderMode === "auto" && shouldUseCard(text)); if (info?.kind === "block") { @@ -287,11 +282,12 @@ export function createFeishuReplyDispatcher(params: CreateFeishuReplyDispatcherP if (info?.kind === "block") { // Some runtimes emit block payloads without onPartial/final callbacks. // Mirror block text into streamText so onIdle close still sends content. - queueStreamingUpdate(text); + queueStreamingUpdate(text, { mode: "delta" }); } if (info?.kind === "final") { - streamText = text; + streamText = mergeStreamingText(streamText, text); await closeStreaming(); + lastFinalText = text; } // Send media even when streaming handled the text if (hasMedia) { @@ -327,6 +323,9 @@ export function createFeishuReplyDispatcher(params: CreateFeishuReplyDispatcherP }); first = false; } + if (info?.kind === "final") { + lastFinalText = text; + } } else { const converted = core.channel.text.convertMarkdownTables(text, tableMode); for (const chunk of core.channel.text.chunkTextWithMode( @@ -345,6 +344,9 @@ export function createFeishuReplyDispatcher(params: CreateFeishuReplyDispatcherP }); first = false; } + if (info?.kind === "final") { + lastFinalText = text; + } } } @@ -387,7 +389,10 @@ export function createFeishuReplyDispatcher(params: CreateFeishuReplyDispatcherP if (!payload.text) { return; } - queueStreamingUpdate(payload.text, { dedupeWithLastPartial: true }); + queueStreamingUpdate(payload.text, { + dedupeWithLastPartial: true, + mode: "snapshot", + }); } : undefined, }, diff --git a/extensions/feishu/src/streaming-card.test.ts b/extensions/feishu/src/streaming-card.test.ts index 913a4633ada..f0276c0a91f 100644 --- a/extensions/feishu/src/streaming-card.test.ts +++ b/extensions/feishu/src/streaming-card.test.ts @@ -1,5 +1,12 @@ -import { describe, expect, it } from "vitest"; -import { mergeStreamingText } from "./streaming-card.js"; +import { beforeEach, describe, expect, it, vi } from "vitest"; + +const fetchWithSsrFGuardMock = vi.hoisted(() => vi.fn()); + +vi.mock("openclaw/plugin-sdk/feishu", () => ({ + fetchWithSsrFGuard: fetchWithSsrFGuardMock, +})); + +import { FeishuStreamingSession, mergeStreamingText } from "./streaming-card.js"; describe("mergeStreamingText", () => { it("prefers the latest full text when it already includes prior text", () => { @@ -15,4 +22,65 @@ describe("mergeStreamingText", () => { expect(mergeStreamingText("hello wor", "ld")).toBe("hello world"); expect(mergeStreamingText("line1", "line2")).toBe("line1line2"); }); + + it("merges overlap between adjacent partial snapshots", () => { + expect(mergeStreamingText("好的，让我", "让我再读取一遍")).toBe("好的，让我再读取一遍"); + expect(mergeStreamingText("revision_id: 552", "2，一点变化都没有")).toBe( + "revision_id: 552，一点变化都没有", + ); + }); +}); + +describe("FeishuStreamingSession routing", () => { + beforeEach(() => { + vi.clearAllMocks(); + fetchWithSsrFGuardMock.mockReset(); + }); + + it("prefers message.reply when reply target and root id both exist", async () => { + fetchWithSsrFGuardMock + .mockResolvedValueOnce({ + response: { json: async () => ({ code: 0, msg: "ok", tenant_access_token: "token" }) }, + release: async () => {}, + }) + .mockResolvedValueOnce({ + response: { json: async () => ({ code: 0, msg: "ok", data: { card_id: "card_1" } }) }, + release: async () => {}, + }); + + const replyMock = vi.fn(async () => ({ code: 0, data: { message_id: "msg_reply" } })); + const createMock = vi.fn(async () => ({ code: 0, data: { message_id: "msg_create" } })); + + const session = new FeishuStreamingSession( + { + im: { + message: { + reply: replyMock, + create: createMock, + }, + }, + } as never, + { + appId: "app", + appSecret: "secret", + domain: "feishu", + }, + ); + + await session.start("oc_chat", "chat_id", { + replyToMessageId: "om_parent", + replyInThread: true, + rootId: "om_topic_root", + }); + + expect(replyMock).toHaveBeenCalledTimes(1); + expect(replyMock).toHaveBeenCalledWith({ + path: { message_id: "om_parent" }, + data: expect.objectContaining({ + msg_type: "interactive", + reply_in_thread: true, + }), + }); + expect(createMock).not.toHaveBeenCalled(); + }); }); diff --git a/extensions/feishu/src/streaming-card.ts b/extensions/feishu/src/streaming-card.ts index bb92faebf70..a254182614f 100644 --- a/extensions/feishu/src/streaming-card.ts +++ b/extensions/feishu/src/streaming-card.ts @@ -94,7 +94,25 @@ export function mergeStreamingText( if (!next) { return previous; } - if (!previous || next === previous || next.includes(previous)) { + if (!previous || next === previous) { + return next; + } + if (next.startsWith(previous)) { + return next; + } + if (previous.startsWith(next)) { + return previous; + } + + // Merge partial overlaps, e.g. "这" + "这是" => "这是". + const maxOverlap = Math.min(previous.length, next.length); + for (let overlap = maxOverlap; overlap > 0; overlap -= 1) { + if (previous.slice(-overlap) === next.slice(0, overlap)) { + return `${previous}${next.slice(overlap)}`; + } + } + + if (next.includes(previous)) { return next; } if (previous.includes(next)) { @@ -142,7 +160,7 @@ export class FeishuStreamingSession { config: { streaming_mode: true, summary: { content: "[Generating...]" }, - streaming_config: { print_frequency_ms: { default: 50 }, print_step: { default: 2 } }, + streaming_config: { print_frequency_ms: { default: 50 }, print_step: { default: 1 } }, }, body: { elements: [{ tag: "markdown", content: "⏳ Thinking...", element_id: "content" }], @@ -181,20 +199,12 @@ export class FeishuStreamingSession { const cardId = createData.data.card_id; const cardContent = JSON.stringify({ type: "card", data: { card_id: cardId } }); - // Topic-group replies require root_id routing. Prefer create+root_id when available. + // Prefer message.reply when we have a reply target — reply_in_thread + // reliably routes streaming cards into Feishu topics, whereas + // message.create with root_id may silently ignore root_id for card + // references (card_id format). let sendRes; - if (options?.rootId) { - const createData = { - receive_id: receiveId, - msg_type: "interactive", - content: cardContent, - root_id: options.rootId, - }; - sendRes = await this.client.im.message.create({ - params: { receive_id_type: receiveIdType }, - data: createData, - }); - } else if (options?.replyToMessageId) { + if (options?.replyToMessageId) { sendRes = await this.client.im.message.reply({ path: { message_id: options.replyToMessageId }, data: { @@ -203,6 +213,15 @@ export class FeishuStreamingSession { ...(options.replyInThread ? { reply_in_thread: true } : {}), }, }); + } else if (options?.rootId) { + // root_id is undeclared in the SDK types but accepted at runtime + sendRes = await this.client.im.message.create({ + params: { receive_id_type: receiveIdType }, + data: Object.assign( + { receive_id: receiveId, msg_type: "interactive", content: cardContent }, + { root_id: options.rootId }, + ), + }); } else { sendRes = await this.client.im.message.create({ params: { receive_id_type: receiveIdType },