From ad3e36ef33148ea5b341ebbbc8f5b0fcd84b0596 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Fri, 31 Jul 2026 23:16:04 -0700 Subject: [PATCH] fix(feishu): unify outbound delivery lifecycle (#117223) Co-authored-by: Peter Steinberger --- extensions/feishu/src/outbound.test.ts | 254 ++++++++++++++++++++++++ extensions/feishu/src/outbound.ts | 264 +++++++++++++------------ 2 files changed, 392 insertions(+), 126 deletions(-) diff --git a/extensions/feishu/src/outbound.test.ts b/extensions/feishu/src/outbound.test.ts index 47d38547562d..ecdfd2606036 100644 --- a/extensions/feishu/src/outbound.test.ts +++ b/extensions/feishu/src/outbound.test.ts @@ -310,16 +310,21 @@ describe("feishuOutbound.sendText local-image auto-convert", () => { adapter, proofs: { text: async () => { + const onDeliveryResult = vi.fn(); const result = await adapterSendText({ cfg: emptyConfig, to: "chat:chat-1", text: "hello", accountId: "default", + onDeliveryResult, }); expect(sendMessageCall()?.to).toBe("chat:chat-1"); expect(sendMessageCall()?.text).toBe("hello"); expect(sendMessageCall()?.accountId).toBe("default"); expect(result.receipt.platformMessageIds).toEqual(["feishu-text-1"]); + expect(onDeliveryResult.mock.calls[0]?.[0]?.receipt.platformMessageIds).toEqual([ + "feishu-text-1", + ]); }, media: async () => { const onDeliveryResult = vi.fn(); @@ -567,6 +572,29 @@ describe("feishuOutbound.sendText local-image auto-convert", () => { } }); + it("does not send fallback text when accepted local-image progress cannot be persisted", async () => { + const { dir, file } = await createTmpImage(); + const onDeliveryResult = vi.fn().mockRejectedValueOnce(new Error("progress write failed")); + + try { + await expect( + sendText({ + cfg: emptyConfig, + to: "chat_1", + text: file, + accountId: "main", + mediaLocalRoots: [dir], + onDeliveryResult, + }), + ).rejects.toThrow("progress write failed"); + expect(sendMediaFeishuMock).toHaveBeenCalledOnce(); + expect(sendMessageFeishuMock).not.toHaveBeenCalled(); + expect(onDeliveryResult).toHaveBeenCalledOnce(); + } finally { + await fs.rm(dir, { recursive: true, force: true }); + } + }); + it("uses markdown cards when renderMode=card", async () => { const result = await sendText({ cfg: cardRenderConfig, @@ -651,6 +679,52 @@ describe("feishuOutbound.sendPayload native cards", () => { return { dir, file }; } + it("records delegated post subchunks once without duplicating parent receipts", async () => { + sendMessageFeishuMock.mockImplementation(async () => ({ + messageId: `chunk_${sendMessageFeishuMock.mock.calls.length}`, + })); + const onDeliveryResult = vi.fn(); + const text = Array.from({ length: 2_200 }, () => "a").join("\n"); + + await feishuOutbound.sendPayload?.({ + cfg: emptyConfig, + to: "chat_1", + text, + accountId: "main", + payload: { text }, + onDeliveryResult, + }); + + expect(sendMessageFeishuMock.mock.calls.length).toBeGreaterThan(1); + expect(onDeliveryResult.mock.calls.map(([result]) => result.messageId)).toEqual( + sendMessageFeishuMock.mock.calls.map((_call, index) => `chunk_${index + 1}`), + ); + }); + + it("records separate fallback media and each expanded text chunk once", async () => { + sendMessageFeishuMock.mockImplementation(async () => ({ + messageId: `chunk_${sendMessageFeishuMock.mock.calls.length}`, + })); + const onDeliveryResult = vi.fn(); + const text = Array.from({ length: 2_200 }, () => "a").join("\n"); + + await feishuOutbound.sendPayload?.({ + cfg: emptyConfig, + to: "chat_1", + text, + accountId: "main", + payload: { text, mediaUrl: "https://example.com/image.png" }, + onDeliveryResult, + }); + + expect(sendMediaFeishuMock).toHaveBeenCalledOnce(); + expect(sendMessageFeishuMock.mock.calls.length).toBeGreaterThan(1); + expect(onDeliveryResult.mock.calls.map(([result]) => result.messageId)).toEqual([ + "media_msg", + ...sendMessageFeishuMock.mock.calls.map((_call, index) => `chunk_${index + 1}`), + ]); + }); + it("renders presentation-only payloads into Feishu channelData cards for core delivery", async () => { const presentation: MessagePresentation = { title: "Approval", @@ -2315,6 +2389,82 @@ describe("feishuOutbound.sendText replyToId forwarding", () => { } }); + it("keeps explicit first-mode replies sticky across expanded post-md chunks", async () => { + await sendText({ + cfg: emptyConfig, + to: "chat_1", + text: Array.from({ length: 2_200 }, () => "a").join("\n"), + replyToId: "om_explicit_reply", + replyToIdSource: "explicit", + replyToMode: "first", + accountId: "main", + }); + + expect(sendMessageFeishuMock.mock.calls.length).toBeGreaterThan(1); + for (const [params] of sendMessageFeishuMock.mock.calls) { + expect(params.replyToMessageId).toBe("om_explicit_reply"); + } + }); + + it("records each accepted expanded text chunk before the next send", async () => { + sendMessageFeishuMock.mockImplementation(async () => ({ + messageId: `chunk_${sendMessageFeishuMock.mock.calls.length}`, + })); + const onDeliveryResult = vi.fn(); + + await sendText({ + cfg: emptyConfig, + to: "chat_1", + text: Array.from({ length: 2_200 }, () => "a").join("\n"), + accountId: "main", + onDeliveryResult, + }); + + expect(sendMessageFeishuMock.mock.calls.length).toBeGreaterThan(1); + expect(onDeliveryResult.mock.calls.map(([result]) => result.messageId)).toEqual( + sendMessageFeishuMock.mock.calls.map((_call, index) => `chunk_${index + 1}`), + ); + }); + + it("preserves the first accepted text chunk when the following send fails", async () => { + sendMessageFeishuMock + .mockResolvedValueOnce({ messageId: "accepted_chunk" }) + .mockRejectedValueOnce(new Error("second chunk failed")); + const onDeliveryResult = vi.fn(); + + await expect( + sendText({ + cfg: emptyConfig, + to: "chat_1", + text: Array.from({ length: 2_200 }, () => "a").join("\n"), + accountId: "main", + onDeliveryResult, + }), + ).rejects.toThrow("second chunk failed"); + + expect(sendMessageFeishuMock).toHaveBeenCalledTimes(2); + expect(onDeliveryResult.mock.calls.map(([result]) => result.messageId)).toEqual([ + "accepted_chunk", + ]); + }); + + it("stops text fanout immediately when accepted delivery cannot be persisted", async () => { + const onDeliveryResult = vi.fn().mockRejectedValueOnce(new Error("progress write failed")); + + await expect( + sendText({ + cfg: emptyConfig, + to: "chat_1", + text: Array.from({ length: 2_200 }, () => "a").join("\n"), + accountId: "main", + onDeliveryResult, + }), + ).rejects.toThrow("progress write failed"); + + expect(sendMessageFeishuMock).toHaveBeenCalledOnce(); + expect(onDeliveryResult).toHaveBeenCalledOnce(); + }); + it("re-chunks expanded post-md text at the selected account limit", async () => { await sendText({ cfg: { @@ -2380,6 +2530,26 @@ describe("feishuOutbound.sendMedia replyToId forwarding", () => { resetOutboundMocks(); }); + it("sends and records text-only media requests exactly once", async () => { + const onDeliveryResult = vi.fn(); + + const result = await feishuOutbound.sendMedia?.({ + cfg: emptyConfig, + to: "chat_1", + text: "text without an attachment", + replyToId: "om_reply_target", + accountId: "main", + onDeliveryResult, + }); + + expect(sendMessageFeishuMock).toHaveBeenCalledOnce(); + expect(sendMediaFeishuMock).not.toHaveBeenCalled(); + expect(onDeliveryResult.mock.calls.map(([delivery]) => delivery.messageId)).toEqual([ + "text_msg", + ]); + expectFeishuResult(result, "text_msg"); + }); + it("forwards replyToId to sendMediaFeishu", async () => { await feishuOutbound.sendMedia?.({ cfg: emptyConfig, @@ -2485,6 +2655,25 @@ describe("feishuOutbound.sendMedia replyToId forwarding", () => { expect(sendMediaCall()?.replyToMessageId).toBe("om_reply_target"); }); + it("keeps explicit first-mode targets sticky across every caption chunk and attachment", async () => { + await feishuOutbound.sendMedia?.({ + cfg: emptyConfig, + to: "chat_1", + text: Array.from({ length: 2_200 }, () => "a").join("\n"), + mediaUrl: "https://example.com/image.png", + replyToId: "om_explicit_reply", + replyToIdSource: "explicit", + replyToMode: "first", + accountId: "main", + }); + + expect(sendMessageFeishuMock.mock.calls.length).toBeGreaterThan(1); + for (const [params] of sendMessageFeishuMock.mock.calls) { + expect(params.replyToMessageId).toBe("om_explicit_reply"); + } + expect(sendMediaCall()?.replyToMessageId).toBe("om_explicit_reply"); + }); + it("keeps native topic roots sticky across captions, attachments, and fallback", async () => { sendMediaFeishuMock.mockRejectedValueOnce(new Error("upload failed")); @@ -2651,6 +2840,71 @@ describe("feishuOutbound.sendMedia replyToId forwarding", () => { expectFeishuResult(result, "fallback_msg"); }); + it("records every accepted caption chunk before recording its attachment", async () => { + sendMessageFeishuMock.mockImplementation(async () => ({ + messageId: `caption_${sendMessageFeishuMock.mock.calls.length}`, + })); + const onDeliveryResult = vi.fn(); + + await feishuOutbound.sendMedia?.({ + cfg: emptyConfig, + to: "chat_1", + text: Array.from({ length: 2_200 }, () => "a").join("\n"), + mediaUrl: "https://example.com/image.png", + accountId: "main", + onDeliveryResult, + }); + + expect(sendMessageFeishuMock.mock.calls.length).toBeGreaterThan(1); + expect(onDeliveryResult.mock.calls.map(([result]) => result.messageId)).toEqual([ + ...sendMessageFeishuMock.mock.calls.map((_call, index) => `caption_${index + 1}`), + "media_msg", + ]); + }); + + it("preserves accepted caption chunks when a later chunk fails before media", async () => { + sendMessageFeishuMock + .mockResolvedValueOnce({ messageId: "accepted_caption" }) + .mockRejectedValueOnce(new Error("second caption failed")); + const onDeliveryResult = vi.fn(); + + await expect( + feishuOutbound.sendMedia?.({ + cfg: emptyConfig, + to: "chat_1", + text: Array.from({ length: 2_200 }, () => "a").join("\n"), + mediaUrl: "https://example.com/image.png", + accountId: "main", + onDeliveryResult, + }), + ).rejects.toThrow("second caption failed"); + + expect(sendMessageFeishuMock).toHaveBeenCalledTimes(2); + expect(sendMediaFeishuMock).not.toHaveBeenCalled(); + expect(onDeliveryResult.mock.calls.map(([result]) => result.messageId)).toEqual([ + "accepted_caption", + ]); + }); + + it("stops before later caption chunks and media when delivery persistence fails", async () => { + const onDeliveryResult = vi.fn().mockRejectedValueOnce(new Error("progress write failed")); + + await expect( + feishuOutbound.sendMedia?.({ + cfg: emptyConfig, + to: "chat_1", + text: Array.from({ length: 2_200 }, () => "a").join("\n"), + mediaUrl: "https://example.com/image.png", + accountId: "main", + onDeliveryResult, + }), + ).rejects.toThrow("progress write failed"); + + expect(sendMessageFeishuMock).toHaveBeenCalledOnce(); + expect(sendMediaFeishuMock).not.toHaveBeenCalled(); + expect(onDeliveryResult).toHaveBeenCalledOnce(); + }); + it("does not send fallback text after an accepted media send loses its receipt", async () => { const acceptedError = createChannelPartialDeliveryError( new Error("Feishu image send failed: no message_id returned"), diff --git a/extensions/feishu/src/outbound.ts b/extensions/feishu/src/outbound.ts index cdb5b08e0c29..d7e90e500fba 100644 --- a/extensions/feishu/src/outbound.ts +++ b/extensions/feishu/src/outbound.ts @@ -150,6 +150,15 @@ type FeishuOutboundPayload = Parameters< NonNullable >[0]["payload"]; type FeishuSendPayloadContext = Parameters>[0]; +type FeishuSendTextContext = Parameters>[0]; + +async function reportFeishuOutboundDelivery( + result: T, + onDeliveryResult: FeishuSendTextContext["onDeliveryResult"], +): Promise { + await onDeliveryResult?.(attachChannelToResult("feishu", result)); + return result; +} function consumeFeishuPresentationFallbackMarker(payload: FeishuOutboundPayload): { payload: FeishuOutboundPayload; @@ -390,8 +399,11 @@ async function sendOutboundText(params: { replyToMessageId?: string; replyInThread?: boolean; accountId?: string; + replyToIdSource?: FeishuSendTextContext["replyToIdSource"]; + replyToMode?: FeishuSendTextContext["replyToMode"]; + onDeliveryResult?: FeishuSendTextContext["onDeliveryResult"]; }) { - const { cfg, to, text, accountId, replyToMessageId, replyInThread } = params; + const { cfg, to, text, accountId, replyToMessageId, replyInThread, onDeliveryResult } = params; const commentResult = await sendCommentThreadReply({ cfg, to, @@ -400,7 +412,7 @@ async function sendOutboundText(params: { accountId, }); if (commentResult) { - return commentResult; + return await reportFeishuOutboundDelivery(commentResult, onDeliveryResult); } const account = resolveFeishuAccount({ cfg, accountId }); @@ -410,14 +422,17 @@ async function sendOutboundText(params: { // modified by post-md newline normalization. Only the post path below // materializes CommonMark soft breaks for Feishu rendering. if (renderMode === "card" || (renderMode === "auto" && shouldUseCard(text))) { - return sendMarkdownCardFeishu({ - cfg, - to, - text, - accountId, - replyToMessageId, - replyInThread, - }); + return await reportFeishuOutboundDelivery( + await sendMarkdownCardFeishu({ + cfg, + to, + text, + accountId, + replyToMessageId, + replyInThread, + }), + onDeliveryResult, + ); } // Tables need contiguous source rows, so convert them before the parser @@ -436,30 +451,26 @@ async function sendOutboundText(params: { limit: postLimit, mode: resolveChunkMode(cfg, "feishu", accountId), }); - if (subChunks.length <= 1) { - return sendMessageFeishu({ - cfg, - to, - text: subChunks[0] ?? normalizedText, - accountId, - replyToMessageId, - replyInThread, - }); - } - let lastResult: Awaited> | undefined; const preserveThread = replyInThread === true; - for (const [i, chunk] of subChunks.entries()) { - // Thread roots must accompany every chunk. Ordinary quoted replies remain - // first-chunk-only so implicit reply ids are consumed once per fanout. - lastResult = await sendMessageFeishu({ - cfg, - to, - text: chunk, - accountId, - replyToMessageId: preserveThread || i === 0 ? replyToMessageId : undefined, - replyInThread: preserveThread ? true : i === 0 ? replyInThread : undefined, - }); + const nextReplyToMessageId = createReplyToFanout({ + replyToId: replyToMessageId, + replyToIdSource: params.replyToIdSource, + replyToMode: params.replyToMode ?? "first", + }); + for (const [i, chunk] of (subChunks.length ? subChunks : [normalizedText]).entries()) { + // Explicit replies and native topic roots stay sticky; implicit first replies do not. + lastResult = await reportFeishuOutboundDelivery( + await sendMessageFeishu({ + cfg, + to, + text: chunk, + accountId, + replyToMessageId: preserveThread ? replyToMessageId : nextReplyToMessageId(), + replyInThread: preserveThread ? true : i === 0 ? replyInThread : undefined, + }), + onDeliveryResult, + ); } return lastResult!; } @@ -508,18 +519,14 @@ async function sendFeishuFallbackPayload(params: { mediaUrl, replyToId: nextReplyToId(), audioAsVoice: params.payload.audioAsVoice ?? ctx.audioAsVoice, - onDeliveryResult: undefined, }); - await ctx.onDeliveryResult?.(lastResult); } for (const chunk of textChunks) { lastResult = await sendText({ ...ctx, text: chunk, replyToId: nextReplyToId(), - onDeliveryResult: undefined, }); - await ctx.onDeliveryResult?.(lastResult); } return lastResult!; } @@ -562,9 +569,7 @@ async function sendFeishuTtsSupplementPayload(params: { ...ctx, text: chunk, replyToId: nextReplyToId(), - onDeliveryResult: undefined, }); - await ctx.onDeliveryResult?.(lastResult); } } @@ -774,21 +779,26 @@ export const feishuOutbound: ChannelOutboundAdapter = { text, accountId, replyToId, + replyToIdSource, + replyToMode, threadId, mediaLocalRoots, identity, + onDeliveryResult, }) => { const { replyToMessageId, replyInThread } = resolveFeishuReplyMode({ replyToId, threadId, }); + const deliveryOptions = { replyToIdSource, replyToMode, onDeliveryResult }; // Scheme A compatibility shim: // when upstream accidentally returns a local image path as plain text, // auto-upload and send as Feishu image message instead of leaking path text. const localImagePath = normalizePossibleLocalImagePath(text); if (localImagePath) { + let mediaResult: Awaited>; try { - return await sendMediaFeishu({ + mediaResult = await sendMediaFeishu({ cfg, to, mediaUrl: localImagePath, @@ -810,8 +820,10 @@ export const feishuOutbound: ChannelOutboundAdapter = { accountId: accountId ?? undefined, replyToMessageId, replyInThread, + ...deliveryOptions, }); } + return await reportFeishuOutboundDelivery(mediaResult, onDeliveryResult); } if (parseFeishuCommentTarget(to)) { @@ -822,20 +834,24 @@ export const feishuOutbound: ChannelOutboundAdapter = { accountId: accountId ?? undefined, replyToMessageId, replyInThread, + ...deliveryOptions, }); } const card = readNativeFeishuCardJson(text); if (card) { assertFeishuCardWithinEnvelope(card, "Feishu native card"); - return await sendCardFeishu({ - cfg, - to, - card: markRenderedFeishuCard(card), - accountId: accountId ?? undefined, - replyToMessageId, - replyInThread, - }); + return await reportFeishuOutboundDelivery( + await sendCardFeishu({ + cfg, + to, + card: markRenderedFeishuCard(card), + accountId: accountId ?? undefined, + replyToMessageId, + replyInThread, + }), + onDeliveryResult, + ); } const account = resolveFeishuAccount({ cfg, accountId: accountId ?? undefined }); @@ -848,15 +864,18 @@ export const feishuOutbound: ChannelOutboundAdapter = { template: "blue" as const, } : undefined; - return await sendStructuredCardFeishu({ - cfg, - to, - text, - replyToMessageId, - replyInThread, - accountId: accountId ?? undefined, - header: header?.title ? header : undefined, - }); + return await reportFeishuOutboundDelivery( + await sendStructuredCardFeishu({ + cfg, + to, + text, + replyToMessageId, + replyInThread, + accountId: accountId ?? undefined, + header: header?.title ? header : undefined, + }), + onDeliveryResult, + ); } return await sendOutboundText({ cfg, @@ -865,6 +884,7 @@ export const feishuOutbound: ChannelOutboundAdapter = { accountId: accountId ?? undefined, replyToMessageId, replyInThread, + ...deliveryOptions, }); }, sendMedia: async ({ @@ -897,8 +917,8 @@ export const feishuOutbound: ChannelOutboundAdapter = { }); return { replyToMessageId, replyInThread }; }; - const commentTarget = parseFeishuCommentTarget(to); - if (commentTarget) { + const deliveryOptions = { replyToIdSource, replyToMode, onDeliveryResult }; + if (parseFeishuCommentTarget(to)) { const commentText = mediaUrl?.trim() ? await buildFeishuMediaFallbackText({ text, @@ -912,94 +932,86 @@ export const feishuOutbound: ChannelOutboundAdapter = { text: commentText, accountId: accountId ?? undefined, ...nextReplyMode(), + ...deliveryOptions, }); } - const suppressTextForVoiceMedia = - mediaUrl !== undefined && - shouldSuppressFeishuTextForVoiceMedia({ - mediaUrl, - audioAsVoice, + if (!mediaUrl) { + return await sendOutboundText({ + cfg, + to, + text: text ?? "", + accountId: accountId ?? undefined, + ...nextReplyMode(), + ...deliveryOptions, }); - const reportDelivery = async (result: Awaited>) => { - await onDeliveryResult?.(attachChannelToResult("feishu", result)); - }; + } + + const suppressTextForVoiceMedia = shouldSuppressFeishuTextForVoiceMedia({ + mediaUrl, + audioAsVoice, + }); let textSent = false; // Send text first if provided, except for Feishu native voice bubbles. if (text?.trim() && !suppressTextForVoiceMedia) { - const textResult = await sendOutboundText({ + await sendOutboundText({ cfg, to, text, accountId: accountId ?? undefined, ...nextReplyMode(), + ...deliveryOptions, }); textSent = true; - await reportDelivery(textResult); } - // Upload and send media if URL or local path provided - if (mediaUrl) { - let mediaResult: Awaited>; - const mediaReplyMode = nextReplyMode(); - try { - mediaResult = await sendMediaFeishu({ - cfg, - to, - mediaUrl, - accountId: accountId ?? undefined, - mediaLocalRoots, - ...mediaReplyMode, - ...(audioAsVoice === true ? { audioAsVoice: true } : {}), - }); - } catch (err) { - if (isChannelPartialDeliveryError(err)) { - // Accepted media is not an upload failure and must never trigger a second send. - throw err; - } - // Log the error for debugging - console.error(`[feishu] sendMediaFeishu failed:`, err); - const fallbackText = await buildFeishuMediaFallbackText({ - text: textSent ? undefined : text, - mediaUrl, - }); - const fallbackResult = await sendOutboundText({ - cfg, - to, - text: fallbackText, - accountId: accountId ?? undefined, - // A rejected upload never delivered its attempted reply target. - ...(textSent ? nextReplyMode() : mediaReplyMode), - }); - await reportDelivery(fallbackResult); - return fallbackResult; + let mediaResult: Awaited>; + const mediaReplyMode = nextReplyMode(); + try { + mediaResult = await sendMediaFeishu({ + cfg, + to, + mediaUrl, + accountId: accountId ?? undefined, + mediaLocalRoots, + ...mediaReplyMode, + ...(audioAsVoice === true ? { audioAsVoice: true } : {}), + }); + } catch (err) { + if (isChannelPartialDeliveryError(err)) { + // Accepted media is not an upload failure and must never trigger a second send. + throw err; } - - // Upload fallback applies only to the platform send. Persistence and - // follow-up failures must not resend an attachment already accepted by Feishu. - await onDeliveryResult?.(attachChannelToResult("feishu", mediaResult)); - if (mediaResult.voiceIntentDegradedToFile && text?.trim()) { - const textResult = await sendOutboundText({ - cfg, - to, - text, - accountId: accountId ?? undefined, - ...nextReplyMode(), - }); - await reportDelivery(textResult); - } - return mediaResult; + console.error(`[feishu] sendMediaFeishu failed:`, err); + const fallbackText = await buildFeishuMediaFallbackText({ + text: textSent ? undefined : text, + mediaUrl, + }); + return await sendOutboundText({ + cfg, + to, + text: fallbackText, + accountId: accountId ?? undefined, + // A rejected upload never delivered its attempted reply target. + ...(textSent ? nextReplyMode() : mediaReplyMode), + ...deliveryOptions, + }); } - // No media URL, just return text result - return await sendOutboundText({ - cfg, - to, - text: text ?? "", - accountId: accountId ?? undefined, - ...nextReplyMode(), - }); + // Persist the accepted attachment before any later fallible text action. + await reportFeishuOutboundDelivery(mediaResult, onDeliveryResult); + if (mediaResult.voiceIntentDegradedToFile && text?.trim()) { + await sendOutboundText({ + cfg, + to, + text, + accountId: accountId ?? undefined, + ...nextReplyMode(), + ...deliveryOptions, + }); + } + return mediaResult; }, }), };