From 1d4dd13750482ef45bf4e455c2a7d21abc600158 Mon Sep 17 00:00:00 2001 From: joshavant <830519+joshavant@users.noreply.github.com> Date: Tue, 11 Aug 2026 18:32:35 -0500 Subject: [PATCH] fix(feishu): keep streaming rejections terminal --- extensions/feishu/src/delivery-trace.test.ts | 101 +++++++++++++++++-- extensions/feishu/src/reply-dispatcher.ts | 38 +++++-- 2 files changed, 123 insertions(+), 16 deletions(-) diff --git a/extensions/feishu/src/delivery-trace.test.ts b/extensions/feishu/src/delivery-trace.test.ts index 6b988a7c9599..213e206373dd 100644 --- a/extensions/feishu/src/delivery-trace.test.ts +++ b/extensions/feishu/src/delivery-trace.test.ts @@ -13,15 +13,9 @@ import { type DeliveryTraceScenarioName, type WireRecorder, } from "openclaw/plugin-sdk/channel-contract-testing"; -import { - dispatchChannelInboundReply, - isChannelPartialDeliveryError, -} from "openclaw/plugin-sdk/channel-inbound"; -import { setReplyPayloadMetadata } from "openclaw/plugin-sdk/reply-payload-testing"; import { withFetchPreconnect } from "openclaw/plugin-sdk/test-env"; import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; import { FeishuConfigSchema } from "./config-schema.js"; -import { sendReplyOrFallbackDirect } from "./send.js"; import type { ResolvedFeishuAccount } from "./types.js"; const settlePendingFinalDeliveryMock = vi.hoisted(() => @@ -37,6 +31,13 @@ vi.mock("../../../src/infra/outbound/delivery-completion.js", async (importOrigi type RecordedWireCall = Parameters[0]; type CreateFeishuReplyDispatcher = typeof import("./reply-dispatcher.js").createFeishuReplyDispatcher; +type DispatchChannelInboundReply = + typeof import("openclaw/plugin-sdk/channel-inbound").dispatchChannelInboundReply; +type IsChannelPartialDeliveryError = + typeof import("openclaw/plugin-sdk/channel-inbound").isChannelPartialDeliveryError; +type SendReplyOrFallbackDirect = typeof import("./send.js").sendReplyOrFallbackDirect; +type SetReplyPayloadMetadata = + typeof import("openclaw/plugin-sdk/reply-payload-testing").setReplyPayloadMetadata; type StreamingStartBackoffMap = typeof import("./reply-dispatcher-state.js").streamingStartBackoffUntilByAccount; @@ -50,7 +51,7 @@ type FeishuTraceState = { cardCount: number; setupCount: number; wireFaults: Array<{ fault: "rate-limit"; retryAfterMs: number }>; - messageResult: "identified" | "no-id"; + messageResult: "identified" | "no-id" | "rejected"; failNextReplace: boolean; }; @@ -145,13 +146,21 @@ vi.mock("./streaming-card.js", async (importOriginal) => { }); let createFeishuReplyDispatcher: CreateFeishuReplyDispatcher; +let dispatchChannelInboundReply: DispatchChannelInboundReply; +let isChannelPartialDeliveryError: IsChannelPartialDeliveryError; +let sendReplyOrFallbackDirect: SendReplyOrFallbackDirect; +let setReplyPayloadMetadata: SetReplyPayloadMetadata; let streamingStartBackoffUntilByAccount: StreamingStartBackoffMap; beforeAll(async () => { // Collection can share a worker with suites that mock the same Feishu modules. // Reload only after this file's hoisted mocks are registered. vi.resetModules(); + ({ dispatchChannelInboundReply, isChannelPartialDeliveryError } = + await import("openclaw/plugin-sdk/channel-inbound")); + ({ setReplyPayloadMetadata } = await import("openclaw/plugin-sdk/reply-payload-testing")); ({ createFeishuReplyDispatcher } = await import("./reply-dispatcher.js")); + ({ sendReplyOrFallbackDirect } = await import("./send.js")); ({ streamingStartBackoffUntilByAccount } = await import("./reply-dispatcher-state.js")); }); @@ -195,10 +204,14 @@ function nextMessageId(): string { } function createRecordingLarkClient() { - const messageSendResult = (messageId: string) => - traceState.messageResult === "identified" + const messageSendResult = (messageId: string) => { + if (traceState.messageResult === "rejected") { + return { code: 230099, msg: "card table number over limit", data: {} }; + } + return traceState.messageResult === "identified" ? { code: 0, msg: "ok", data: { message_id: messageId } } : { code: 0, msg: "ok", data: {} }; + }; return { im: { message: { @@ -555,6 +568,76 @@ describe("feishu producer custody boundaries", () => { expect(client.im.message.create).not.toHaveBeenCalled(); }); + it.each([ + { mode: "reply", dispatcherParams: { replyToMessageId: "om-inbound" } }, + { mode: "root-create", dispatcherParams: { rootId: "om-root" } }, + { mode: "ordinary-create", dispatcherParams: {} }, + ])( + "keeps a permanent $mode streaming rejection terminal without static replay", + async ({ dispatcherParams }) => { + const wireCalls: RecordedWireCall[] = []; + traceState.recordWireCall = (call) => wireCalls.push(call); + traceState.messageCount = 0; + traceState.cardCount = 0; + traceState.messageResult = "rejected"; + traceState.account = { + ...makeTraceAccount("final-only"), + config: FeishuConfigSchema.parse({ renderMode: "card", streaming: { mode: "partial" } }), + }; + traceState.larkClient = createRecordingLarkClient(); + traceState.cardKitFetch = createRecordingCardKitFetch(); + const created = createFeishuReplyDispatcher({ + cfg: {} as never, + agentId: "agent", + runtime: { log: () => {}, error: () => {} } as never, + chatId: "oc-trace-chat", + sendTarget: "oc-trace-chat", + ...dispatcherParams, + }); + const sourcePayload = setReplyPayloadMetadata( + { text: "the final answer" }, + { pendingFinalDeliveryCompletion: pendingFinalCompletion }, + ); + + const error = await dispatchChannelInboundReply({ + cfg: {}, + channel: "feishu", + accountId: "main", + agentId: "agent", + routeSessionKey: pendingFinalCompletion.sessionKey, + storePath: pendingFinalCompletion.storePath, + ctxPayload: createTraceContext(), + recordInboundSession: async () => undefined, + dispatchReplyWithBufferedBlockDispatcher: async ({ dispatcherOptions }) => { + await dispatcherOptions.deliver?.(sourcePayload, { kind: "final" }); + return { queuedFinal: true, counts: { tool: 0, block: 0, final: 1 } }; + }, + dispatcherOptions: created.dispatcherOptions, + replyOptions: created.replyOptions, + delivery: created.delivery, + }).catch((caught: unknown) => caught); + + expect(error).toMatchObject({ + name: "PlatformMessageNotDispatchedError", + retryable: false, + }); + expect(settlePendingFinalDeliveryMock).toHaveBeenNthCalledWith( + 2, + { kind: "pending-final", ...pendingFinalCompletion }, + "suppressed", + ["prepared", "queued", "unknown"], + ); + expect( + wireCalls.filter( + (call) => call.method === "im.message.reply" || call.method === "im.message.create", + ), + ).toHaveLength(1); + await created.dispatcherOptions.onIdle?.(); + await created.dispatcherOptions.onIdle?.(); + expect(streamingStartBackoffUntilByAccount.has("main")).toBe(false); + }, + ); + it("retains visible custody when no-ID preview disposition fails", async () => { const wireCalls: RecordedWireCall[] = []; traceState.recordWireCall = (call) => wireCalls.push(call); diff --git a/extensions/feishu/src/reply-dispatcher.ts b/extensions/feishu/src/reply-dispatcher.ts index eeafa26ff3b5..3f82de97011d 100644 --- a/extensions/feishu/src/reply-dispatcher.ts +++ b/extensions/feishu/src/reply-dispatcher.ts @@ -12,7 +12,10 @@ import { resolveChannelPreviewStreamMode, resolveChannelStreamingBlockEnabled, } from "openclaw/plugin-sdk/channel-outbound"; -import { toStringifiedError as toFeishuError } from "openclaw/plugin-sdk/error-runtime"; +import { + PlatformMessageNotDispatchedError, + toStringifiedError as toFeishuError, +} from "openclaw/plugin-sdk/error-runtime"; import { getGlobalHookRunner } from "openclaw/plugin-sdk/plugin-runtime"; import { getReplyPayloadTtsSupplement, @@ -318,6 +321,7 @@ export function createFeishuReplyDispatcher(params: CreateFeishuReplyDispatcherP let sentIndependentBlockText = false; let partialUpdateQueue: Promise = Promise.resolve(); let streamingStartPromise: Promise | null = null; + let terminalStreamingStartError: PlatformMessageNotDispatchedError | undefined; let streamingGeneration = 0; let activeStreamingGeneration: number | undefined; let inFlightStreamingClose: @@ -345,6 +349,12 @@ export function createFeishuReplyDispatcher(params: CreateFeishuReplyDispatcherP visibleReplySent = true; }; + const throwIfStreamingStartRejected = () => { + if (terminalStreamingStartError) { + throw terminalStreamingStartError; + } + }; + const normalizeStreamingFinalizationFailure = ( error: unknown, ): { result: FeishuReplyDeliveryResult; error: Error } | undefined => { @@ -460,6 +470,7 @@ export function createFeishuReplyDispatcher(params: CreateFeishuReplyDispatcherP !streamingEnabled || streamingStartPromise || streaming || + terminalStreamingStartError || isStreamingStartBackedOff(account.accountId) ) { return; @@ -500,12 +511,21 @@ export function createFeishuReplyDispatcher(params: CreateFeishuReplyDispatcherP }); streamingStartBackoffUntilByAccount.delete(account.accountId); } catch (error) { - rememberStreamingStartFailure(account.accountId); - params.runtime.error?.( - `feishu[${account.accountId}]: streaming start failed; using non-streaming card fallback for ${ - STREAMING_START_FAILURE_BACKOFF_MS / 1000 - }s: ${String(error)}`, - ); + if (error instanceof PlatformMessageNotDispatchedError && !error.retryable) { + // A provider-declared no-send is terminal for this logical payload. + // Keep the fact until delivery consumes it instead of replaying statically. + terminalStreamingStartError = error; + params.runtime.error?.( + `feishu[${account.accountId}]: streaming start permanently rejected: ${String(error)}`, + ); + } else { + rememberStreamingStartFailure(account.accountId); + params.runtime.error?.( + `feishu[${account.accountId}]: streaming start failed; using non-streaming card fallback for ${ + STREAMING_START_FAILURE_BACKOFF_MS / 1000 + }s: ${String(error)}`, + ); + } if (streaming === session) { streaming = null; streamingStartPromise = null; @@ -518,6 +538,7 @@ export function createFeishuReplyDispatcher(params: CreateFeishuReplyDispatcherP const resetStreamingState = () => { streaming = null; streamingStartPromise = null; + terminalStreamingStartError = undefined; activeStreamingGeneration = undefined; partialUpdateQueue = Promise.resolve(); streamText = ""; @@ -699,6 +720,7 @@ export function createFeishuReplyDispatcher(params: CreateFeishuReplyDispatcherP await streamingStartPromise; } await partialUpdateQueue; + throwIfStreamingStartRejected(); if (streaming?.isActive()) { try { await streaming.discard(); @@ -1398,6 +1420,7 @@ export function createFeishuReplyDispatcher(params: CreateFeishuReplyDispatcherP if (streamingStartPromise) { await streamingStartPromise; } + throwIfStreamingStartRejected(); } if (info?.kind === "final" && useStreamingCard) { @@ -1405,6 +1428,7 @@ export function createFeishuReplyDispatcher(params: CreateFeishuReplyDispatcherP if (streamingStartPromise) { await streamingStartPromise; } + throwIfStreamingStartRejected(); } const shouldStreamText = info?.kind === "block" || info?.kind === "final";