From 736dddb96a6e731ddc440e2dafd8f147a2dfe7d2 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Sat, 18 Jul 2026 18:49:38 +0100 Subject: [PATCH] fix(feishu): preserve inbound messages across restarts (#110864) * fix(feishu): adopt durable inbound ingress * fix(feishu): preserve failed broadcast ingress * fix(feishu): commit broadcast lanes at adoption * test(feishu): fix lifecycle fixture typing --- extensions/feishu/src/bot-broadcast.ts | 235 +++++++ extensions/feishu/src/bot.broadcast.test.ts | 378 ++++++++++- extensions/feishu/src/bot.ts | 393 +++++++---- extensions/feishu/src/comment-handler.ts | 6 + extensions/feishu/src/dedup-state.ts | 1 + extensions/feishu/src/dedup.test.ts | 29 +- extensions/feishu/src/dedup.ts | 24 +- extensions/feishu/src/delivery-trace.test.ts | 4 +- extensions/feishu/src/feishu-ingress.test.ts | 519 ++++++++++++++ extensions/feishu/src/feishu-ingress.ts | 632 ++++++++++++++++++ .../feishu/src/lifecycle.test-support.ts | 5 + extensions/feishu/src/monitor.account.ts | 49 +- ...dcast.reply-once.lifecycle.test-support.ts | 17 +- .../src/monitor.comment-notice-handler.ts | 112 ++-- extensions/feishu/src/monitor.comment.test.ts | 91 +-- .../monitor.message-handler.ingress.test.ts | 296 ++++++++ .../feishu/src/monitor.message-handler.ts | 215 ++++-- .../feishu/src/monitor.reaction.test.ts | 3 - extensions/feishu/src/monitor.transport.ts | 7 + ...monitor.vc-meeting-invited-handler.test.ts | 3 - .../test-support/lifecycle-test-support.ts | 36 +- 21 files changed, 2668 insertions(+), 387 deletions(-) create mode 100644 extensions/feishu/src/feishu-ingress.test.ts create mode 100644 extensions/feishu/src/feishu-ingress.ts create mode 100644 extensions/feishu/src/monitor.message-handler.ingress.test.ts diff --git a/extensions/feishu/src/bot-broadcast.ts b/extensions/feishu/src/bot-broadcast.ts index 7433489099c3..c30db54e1505 100644 --- a/extensions/feishu/src/bot-broadcast.ts +++ b/extensions/feishu/src/bot-broadcast.ts @@ -1,4 +1,239 @@ +import type { ChannelReplayClaimHandle } from "openclaw/plugin-sdk/persistent-dedupe"; import type { ClawdbotConfig } from "./bot-runtime-api.js"; +import type { FeishuIngressLifecycle } from "./feishu-ingress.js"; + +export function createFeishuBroadcastIngressSettlement(params: { + lifecycle?: FeishuIngressLifecycle; + replayClaim?: ChannelReplayClaimHandle; + onReplayCommitError?: (error: unknown) => void; + onAdopted?: () => void; +}): { + createLane: (replayClaim?: ChannelReplayClaimHandle) => { + lifecycle: FeishuIngressLifecycle; + onDispatchComplete: (dispatched: boolean) => Promise; + onDispatchFailed: (error: unknown) => Promise; + }; + onLanePending: () => void; + onDispatchComplete: () => Promise; + onDispatchFailed: (error: unknown) => Promise; +} { + type LaneState = { + replayClaim?: ChannelReplayClaimHandle; + status: "pending" | "deferred" | "adopted" | "completed" | "failed" | "abandoned"; + }; + + const lanes = new Set(); + const failures: unknown[] = []; + const fallbackAbortSignal = new AbortController().signal; + let fanoutSettled = false; + let terminal: "adopted" | "abandoned" | undefined; + let adoption: Promise | undefined; + let abandonment: Promise | undefined; + let finalizing = false; + let deferred = false; + let replayReleased = false; + + const beginFinalizing = () => { + if (finalizing) { + return; + } + finalizing = true; + params.lifecycle?.onAdoptionFinalizing(); + }; + const defer = () => { + if (deferred) { + return; + } + deferred = true; + params.lifecycle?.onDeferred(); + }; + const reportReplayCommitError = (error: unknown) => { + try { + params.onReplayCommitError?.(error); + } catch { + // Reporting cannot undo an already adopted durable turn. + } + }; + const releaseReplayClaim = (error: unknown) => { + if (replayReleased || terminal === "adopted") { + return; + } + replayReleased = true; + params.replayClaim?.release({ error }); + }; + const runAbandonment = async (error: unknown) => { + if (terminal) { + return; + } + releaseReplayClaim(error); + try { + await params.lifecycle?.onAbandoned(); + } finally { + terminal = "abandoned"; + } + }; + const abandon = async (error: unknown) => { + if (terminal) { + return; + } + if (adoption) { + await adoption.catch(() => undefined); + if (terminal) { + return; + } + } + const activeAbandonment = abandonment ?? runAbandonment(error); + abandonment = activeAbandonment; + await activeAbandonment; + }; + const runAdoption = async () => { + beginFinalizing(); + try { + await params.lifecycle?.onAdopted(); + terminal = "adopted"; + try { + params.onAdopted?.(); + } catch { + // Local cleanup cannot reopen an already adopted durable turn. + } + try { + await params.replayClaim?.commit(); + } catch (error) { + reportReplayCommitError(error); + } + } catch (error) { + await runAbandonment(error).catch(() => undefined); + throw error; + } + }; + const adopt = async () => { + if (terminal) { + return; + } + if (abandonment) { + await abandonment.catch(() => undefined); + if (terminal) { + return; + } + } + const activeAdoption = adoption ?? runAdoption(); + adoption = activeAdoption; + await activeAdoption; + }; + const maybeSettle = async () => { + if (!fanoutSettled || terminal) { + return; + } + if ( + failures.length > 0 || + [...lanes].some((lane) => lane.status === "failed" || lane.status === "abandoned") + ) { + await abandon( + failures.length === 1 + ? failures[0] + : new AggregateError(failures, "Feishu broadcast dispatch failed"), + ); + return; + } + if ( + [...lanes].some( + (lane) => + lane.status === "pending" || lane.status === "deferred" || lane.status === "adopted", + ) + ) { + return; + } + await adopt(); + }; + + return { + createLane: (replayClaim) => { + const lane: LaneState = { replayClaim, status: "pending" }; + lanes.add(lane); + const releaseLane = (error: unknown) => { + lane.replayClaim?.release({ error }); + }; + return { + lifecycle: { + abortSignal: params.lifecycle?.abortSignal ?? fallbackAbortSignal, + onAdopted: async () => { + if ( + lane.status === "adopted" || + lane.status === "completed" || + lane.status === "failed" || + lane.status === "abandoned" + ) { + return; + } + lane.status = "adopted"; + beginFinalizing(); + try { + await lane.replayClaim?.commit(); + } catch (error) { + reportReplayCommitError(error); + } + lane.status = "completed"; + await maybeSettle(); + }, + onDeferred: () => { + if (lane.status !== "pending") { + return; + } + lane.status = "deferred"; + defer(); + }, + onAdoptionFinalizing: beginFinalizing, + onAbandoned: async () => { + if ( + lane.status === "completed" || + lane.status === "failed" || + lane.status === "abandoned" + ) { + return; + } + lane.status = "abandoned"; + releaseLane(new Error("feishu-broadcast-turn-abandoned")); + await maybeSettle(); + }, + }, + onDispatchComplete: async (dispatched) => { + if (!dispatched && lane.status === "pending") { + const error = new Error("feishu broadcast lane was not dispatched"); + lane.status = "failed"; + failures.push(error); + releaseLane(error); + return; + } + if (lane.status !== "pending") { + return; + } + const error = new Error("feishu broadcast dispatch returned before turn adoption"); + lane.status = "failed"; + failures.push(error); + releaseLane(error); + }, + onDispatchFailed: async (error) => { + failures.push(error); + if (lane.status !== "completed") { + lane.status = "failed"; + releaseLane(error); + } + await maybeSettle(); + }, + }; + }, + onLanePending: defer, + onDispatchComplete: async () => { + fanoutSettled = true; + await maybeSettle(); + }, + onDispatchFailed: async (error) => { + failures.push(error); + fanoutSettled = true; + await maybeSettle(); + }, + }; +} export function resolveBroadcastAgents(cfg: ClawdbotConfig, peerId: string): string[] | null { const broadcast = (cfg as Record).broadcast; diff --git a/extensions/feishu/src/bot.broadcast.test.ts b/extensions/feishu/src/bot.broadcast.test.ts index b62f8036d173..9bb51de20d9f 100644 --- a/extensions/feishu/src/bot.broadcast.test.ts +++ b/extensions/feishu/src/bot.broadcast.test.ts @@ -1,9 +1,12 @@ // Feishu tests cover bot.broadcast plugin behavior. -import { afterAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { afterAll, afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import type { ClawdbotConfig, PluginRuntime } from "../runtime-api.js"; import { feishuGroupNameCache } from "./bot-group-name-state.js"; import type { FeishuMessageEvent } from "./bot.js"; import { handleFeishuMessage } from "./bot.js"; +import { feishuDedupeState } from "./dedup-state.js"; +import type { FeishuMessageProcessingClaim } from "./dedup.js"; +import type { FeishuIngressLifecycle } from "./feishu-ingress.js"; import { setFeishuRuntime } from "./runtime.js"; const { @@ -26,7 +29,9 @@ const { mockDispatchReply: vi.fn().mockResolvedValue({ queuedFinal: false, counts: { final: 1 } }), mockRecordInboundSession: vi.fn().mockResolvedValue(undefined), mockResolveAgentRoute: vi.fn(), - mockResolveStorePath: vi.fn(() => "/tmp/feishu-session-store.json"), + mockResolveStorePath: vi.fn( + (_store?: unknown, _options?: { agentId?: string }) => "/tmp/feishu-session-store.json", + ), })); vi.mock("openclaw/plugin-sdk/channel-inbound", async () => { @@ -75,6 +80,31 @@ function createRuntimeEnv() { }; } +function createIngressLifecycle() { + const calls = { + adopted: vi.fn(async () => {}), + deferred: vi.fn(), + finalizing: vi.fn(), + abandoned: vi.fn(async () => {}), + }; + const lifecycle: FeishuIngressLifecycle = { + abortSignal: new AbortController().signal, + onAdopted: calls.adopted, + onDeferred: calls.deferred, + onAdoptionFinalizing: calls.finalizing, + onAbandoned: calls.abandoned, + }; + return { calls, lifecycle }; +} + +function createReplayClaim(key: string): FeishuMessageProcessingClaim { + return { + keys: [key], + commit: vi.fn(async () => true), + release: vi.fn(), + }; +} + describe("broadcast dispatch", () => { const mockGetChatInfo = vi.fn(); const mockShouldComputeCommandAuthorized = vi.fn(() => false); @@ -129,16 +159,21 @@ describe("broadcast dispatch", () => { updateLastRoute: turn.record?.updateLastRoute, onRecordError: turn.record?.onRecordError ?? (() => undefined), }); + const dispatchResult = await mockDispatchReply({ + ctx: turn.ctxPayload, + cfg: turn.cfg, + replyOptions: turn.replyOptions, + }); + const dispatched = !(dispatchResult as { undispatched?: boolean }).undispatched; + if (dispatched && !(dispatchResult as { deferAdoption?: boolean }).deferAdoption) { + await turn.replyOptions?.turnAdoptionLifecycle?.onAdopted?.(); + } return { admission: turn.admission ?? { kind: "dispatch" as const }, - dispatched: true, + dispatched, ctxPayload: turn.ctxPayload, routeSessionKey, - dispatchResult: await mockDispatchReply({ - ctx: turn.ctxPayload, - cfg: turn.cfg, - replyOptions: turn.replyOptions, - }), + ...(dispatched ? { dispatchResult } : {}), }; }), }, @@ -208,10 +243,12 @@ describe("broadcast dispatch", () => { beforeEach(() => { vi.clearAllMocks(); + feishuDedupeState.reset(); mockDispatchReply.mockReset().mockResolvedValue({ queuedFinal: false, counts: { final: 1 }, }); + mockResolveStorePath.mockReset().mockReturnValue("/tmp/feishu-session-store.json"); feishuGroupNameCache.clear(); builtInboundContextCalls.length = 0; mockResolveAgentRoute.mockReturnValue({ @@ -247,6 +284,11 @@ describe("broadcast dispatch", () => { setFeishuRuntime(runtimeStub); }); + afterEach(() => { + vi.restoreAllMocks(); + feishuDedupeState.reset(); + }); + it("dispatches to all broadcast agents when bot is mentioned", async () => { const cfg = createBroadcastConfig(); const event = createBroadcastEvent({ @@ -562,6 +604,326 @@ describe("broadcast dispatch", () => { expect(mockGetChatInfo).not.toHaveBeenCalled(); }); + it("abandons a failed broadcast claim and re-dispatches it on redelivery", async () => { + const firstClaim = createReplayClaim("broadcast-first-attempt"); + const retryClaim = createReplayClaim("broadcast-retry"); + const firstSusanClaim = createReplayClaim("broadcast-susan-first-attempt"); + const retrySusanClaim = createReplayClaim("broadcast-susan-retry"); + const mainClaim = createReplayClaim("broadcast-main"); + let broadcastAttempt = 0; + let susanAttempt = 0; + let mainAttempt = 0; + vi.spyOn(feishuDedupeState.guard, "claim").mockImplementation(async (_messageId, options) => { + if (options?.namespace === "broadcast") { + broadcastAttempt += 1; + return { + kind: "claimed", + handle: broadcastAttempt === 1 ? firstClaim : retryClaim, + }; + } + if (options?.namespace === "broadcast:susan") { + susanAttempt += 1; + return { + kind: "claimed", + handle: susanAttempt === 1 ? firstSusanClaim : retrySusanClaim, + }; + } + if (options?.namespace === "broadcast:main") { + mainAttempt += 1; + return mainAttempt === 1 ? { kind: "claimed", handle: mainClaim } : { kind: "duplicate" }; + } + return { kind: "invalid" }; + }); + mockDispatchReply + .mockRejectedValueOnce(new Error("observer dispatch failed")) + .mockResolvedValue({ queuedFinal: false, counts: { final: 1 } }); + const event = createBroadcastEvent({ + messageId: "msg-broadcast-redelivery", + text: "retry me", + botMentioned: true, + }); + const firstTransport = createIngressLifecycle(); + const cfg = createBroadcastConfig(); + (cfg.broadcast as Record).strategy = "sequential"; + + await expect( + handleFeishuMessage({ + cfg, + event, + botOpenId: "bot-open-id", + runtime: createRuntimeEnv(), + turnAdoptionLifecycle: firstTransport.lifecycle, + }), + ).rejects.toThrow("observer dispatch failed"); + + expect(mockDispatchReply).toHaveBeenCalledTimes(2); + expect(firstClaim.commit).not.toHaveBeenCalled(); + expect(firstClaim.release).toHaveBeenCalledTimes(1); + expect(firstSusanClaim.commit).not.toHaveBeenCalled(); + expect(firstSusanClaim.release).toHaveBeenCalledTimes(1); + expect(mainClaim.commit).toHaveBeenCalledTimes(1); + expect(mainClaim.release).not.toHaveBeenCalled(); + expect(firstTransport.calls.adopted).not.toHaveBeenCalled(); + expect(firstTransport.calls.abandoned).toHaveBeenCalledTimes(1); + + mockDispatchReply.mockClear(); + const retryTransport = createIngressLifecycle(); + await handleFeishuMessage({ + cfg, + event, + botOpenId: "bot-open-id", + runtime: createRuntimeEnv(), + turnAdoptionLifecycle: retryTransport.lifecycle, + }); + + // The adopted main lane stays committed; only failed Susan re-dispatches. + expect(mockDispatchReply).toHaveBeenCalledTimes(1); + expect(retryTransport.calls.adopted).toHaveBeenCalledTimes(1); + expect(retryTransport.calls.abandoned).not.toHaveBeenCalled(); + expect(retryClaim.release).not.toHaveBeenCalled(); + expect(retryClaim.commit).toHaveBeenCalledTimes(1); + expect(retrySusanClaim.commit).toHaveBeenCalledTimes(1); + expect(retrySusanClaim.release).not.toHaveBeenCalled(); + }); + + it("keeps an adopted active lane committed when its no-visible fallback fails", async () => { + const broadcastClaim = createReplayClaim("broadcast-fallback-failure"); + const susanClaim = createReplayClaim("broadcast-fallback-failure-susan"); + const mainClaim = createReplayClaim("broadcast-fallback-failure-main"); + vi.spyOn(feishuDedupeState.guard, "claim").mockImplementation(async (_messageId, options) => ({ + kind: "claimed", + handle: + options?.namespace === "broadcast:susan" + ? susanClaim + : options?.namespace === "broadcast:main" + ? mainClaim + : broadcastClaim, + })); + mockDispatchReply.mockImplementation(async ({ ctx }) => + String(ctx.SessionKey).startsWith("agent:main:") + ? { + queuedFinal: false, + counts: { final: 0 }, + noVisibleReplyFallbackEligible: true, + } + : { queuedFinal: false, counts: { final: 1 } }, + ); + mockCreateFeishuReplyDispatcher.mockReturnValueOnce({ + dispatcherOptions: {}, + delivery: { deliver: vi.fn(async () => undefined) }, + replyOptions: {}, + ensureNoVisibleReplyFallback: vi.fn(async () => { + throw new Error("fallback send failed"); + }), + }); + const transport = createIngressLifecycle(); + + await expect( + handleFeishuMessage({ + cfg: createBroadcastConfig(), + event: createBroadcastEvent({ + messageId: "msg-broadcast-fallback-failure", + text: "fallback must retry", + botMentioned: true, + }), + botOpenId: "bot-open-id", + runtime: createRuntimeEnv(), + turnAdoptionLifecycle: transport.lifecycle, + }), + ).rejects.toThrow("fallback send failed"); + + expect(susanClaim.commit).toHaveBeenCalledTimes(1); + expect(mainClaim.commit).toHaveBeenCalledTimes(1); + expect(mainClaim.release).not.toHaveBeenCalled(); + expect(broadcastClaim.commit).not.toHaveBeenCalled(); + expect(broadcastClaim.release).toHaveBeenCalledTimes(1); + expect(transport.calls.abandoned).toHaveBeenCalledTimes(1); + }); + + it("releases an agent claim when broadcast lane setup fails", async () => { + const broadcastClaim = createReplayClaim("broadcast-setup-failure"); + const susanClaim = createReplayClaim("broadcast-setup-failure-susan"); + const mainClaim = createReplayClaim("broadcast-setup-failure-main"); + vi.spyOn(feishuDedupeState.guard, "claim").mockImplementation(async (_messageId, options) => ({ + kind: "claimed", + handle: + options?.namespace === "broadcast:susan" + ? susanClaim + : options?.namespace === "broadcast:main" + ? mainClaim + : broadcastClaim, + })); + mockResolveStorePath.mockImplementation((_store, options?: { agentId?: string }) => { + if (options?.agentId === "susan") { + throw new Error("session path failed"); + } + return "/tmp/feishu-session-store.json"; + }); + const cfg = createBroadcastConfig(); + (cfg.broadcast as Record).strategy = "sequential"; + const transport = createIngressLifecycle(); + + await expect( + handleFeishuMessage({ + cfg, + event: createBroadcastEvent({ + messageId: "msg-broadcast-setup-failure", + text: "setup must release", + botMentioned: true, + }), + botOpenId: "bot-open-id", + runtime: createRuntimeEnv(), + turnAdoptionLifecycle: transport.lifecycle, + }), + ).rejects.toThrow("session path failed"); + + expect(susanClaim.commit).not.toHaveBeenCalled(); + expect(susanClaim.release).toHaveBeenCalledTimes(1); + expect(mainClaim.commit).toHaveBeenCalledTimes(1); + expect(broadcastClaim.release).toHaveBeenCalledTimes(1); + expect(transport.calls.abandoned).toHaveBeenCalledTimes(1); + }); + + it("abandons the shared claim when an agent lane is not dispatched", async () => { + const broadcastClaim = createReplayClaim("broadcast-undispatched"); + const susanClaim = createReplayClaim("broadcast-undispatched-susan"); + const mainClaim = createReplayClaim("broadcast-undispatched-main"); + vi.spyOn(feishuDedupeState.guard, "claim").mockImplementation(async (_messageId, options) => ({ + kind: "claimed", + handle: + options?.namespace === "broadcast:susan" + ? susanClaim + : options?.namespace === "broadcast:main" + ? mainClaim + : broadcastClaim, + })); + mockDispatchReply.mockImplementation(async ({ ctx }) => + String(ctx.SessionKey).startsWith("agent:susan:") + ? { queuedFinal: false, counts: { final: 0 }, undispatched: true } + : { queuedFinal: false, counts: { final: 1 } }, + ); + const transport = createIngressLifecycle(); + + await handleFeishuMessage({ + cfg: createBroadcastConfig(), + event: createBroadcastEvent({ + messageId: "msg-broadcast-undispatched", + text: "do not tombstone", + botMentioned: true, + }), + botOpenId: "bot-open-id", + runtime: createRuntimeEnv(), + turnAdoptionLifecycle: transport.lifecycle, + }); + + expect(susanClaim.commit).not.toHaveBeenCalled(); + expect(susanClaim.release).toHaveBeenCalledTimes(1); + expect(mainClaim.commit).toHaveBeenCalledTimes(1); + expect(broadcastClaim.commit).not.toHaveBeenCalled(); + expect(broadcastClaim.release).toHaveBeenCalledTimes(1); + expect(transport.calls.adopted).not.toHaveBeenCalled(); + expect(transport.calls.abandoned).toHaveBeenCalledTimes(1); + }); + + it("commits the shared broadcast claim only after transport adoption", async () => { + const broadcastClaim = createReplayClaim("broadcast-adoption-order"); + const susanClaim = createReplayClaim("broadcast-adoption-order-susan"); + const mainClaim = createReplayClaim("broadcast-adoption-order-main"); + vi.spyOn(feishuDedupeState.guard, "claim").mockImplementation(async (_messageId, options) => ({ + kind: "claimed", + handle: + options?.namespace === "broadcast:susan" + ? susanClaim + : options?.namespace === "broadcast:main" + ? mainClaim + : broadcastClaim, + })); + const transport = createIngressLifecycle(); + let finishAdoption!: () => void; + const adoptionGate = new Promise((resolve) => { + finishAdoption = resolve; + }); + transport.calls.adopted.mockImplementationOnce(async () => await adoptionGate); + + const handling = handleFeishuMessage({ + cfg: createBroadcastConfig(), + event: createBroadcastEvent({ + messageId: "msg-broadcast-adoption-order", + text: "adopt before dedupe", + botMentioned: true, + }), + botOpenId: "bot-open-id", + runtime: createRuntimeEnv(), + turnAdoptionLifecycle: transport.lifecycle, + }); + + await vi.waitFor(() => expect(transport.calls.adopted).toHaveBeenCalledTimes(1)); + expect(transport.calls.finalizing).toHaveBeenCalledTimes(1); + expect(broadcastClaim.commit).not.toHaveBeenCalled(); + expect(susanClaim.commit).toHaveBeenCalledTimes(1); + expect(mainClaim.commit).toHaveBeenCalledTimes(1); + + finishAdoption(); + await handling; + + expect(broadcastClaim.commit).toHaveBeenCalledTimes(1); + expect(transport.calls.adopted.mock.invocationCallOrder[0]).toBeLessThan( + vi.mocked(broadcastClaim.commit).mock.invocationCallOrder[0] ?? Number.POSITIVE_INFINITY, + ); + expect(broadcastClaim.release).not.toHaveBeenCalled(); + }); + + it("waits for every independently deferred broadcast lane before adoption", async () => { + const broadcastClaim = createReplayClaim("broadcast-deferred"); + const susanClaim = createReplayClaim("broadcast-deferred-susan"); + const mainClaim = createReplayClaim("broadcast-deferred-main"); + vi.spyOn(feishuDedupeState.guard, "claim").mockImplementation(async (_messageId, options) => ({ + kind: "claimed", + handle: + options?.namespace === "broadcast:susan" + ? susanClaim + : options?.namespace === "broadcast:main" + ? mainClaim + : broadcastClaim, + })); + let deferredLifecycle: + | Pick + | undefined; + mockDispatchReply.mockImplementation(async ({ ctx, replyOptions }) => { + if (String(ctx.SessionKey).startsWith("agent:susan:")) { + deferredLifecycle = replyOptions?.turnAdoptionLifecycle; + deferredLifecycle?.onDeferred(); + return { queuedFinal: false, counts: { final: 1 }, deferAdoption: true }; + } + return { queuedFinal: false, counts: { final: 1 } }; + }); + const transport = createIngressLifecycle(); + + await handleFeishuMessage({ + cfg: createBroadcastConfig(), + event: createBroadcastEvent({ + messageId: "msg-broadcast-deferred", + text: "wait for every lane", + botMentioned: true, + }), + botOpenId: "bot-open-id", + runtime: createRuntimeEnv(), + turnAdoptionLifecycle: transport.lifecycle, + }); + + expect(transport.calls.deferred).toHaveBeenCalledTimes(1); + expect(transport.calls.adopted).not.toHaveBeenCalled(); + expect(broadcastClaim.commit).not.toHaveBeenCalled(); + expect(mainClaim.commit).toHaveBeenCalledTimes(1); + expect(susanClaim.commit).not.toHaveBeenCalled(); + + await deferredLifecycle?.onAdopted(); + + expect(susanClaim.commit).toHaveBeenCalledTimes(1); + expect(transport.calls.adopted).toHaveBeenCalledTimes(1); + expect(broadcastClaim.commit).toHaveBeenCalledTimes(1); + }); + it("skips unknown agents not in agents.list", async () => { const cfg: ClawdbotConfig = { broadcast: { "oc-broadcast-group": ["susan", "unknown-agent"] }, diff --git a/extensions/feishu/src/bot.ts b/extensions/feishu/src/bot.ts index 79151d701e35..a85c8242abee 100644 --- a/extensions/feishu/src/bot.ts +++ b/extensions/feishu/src/bot.ts @@ -6,7 +6,10 @@ import { resolveEnvelopeFormatOptions, toInboundMediaFacts, } from "openclaw/plugin-sdk/channel-inbound"; -import { resolveAgentOutboundIdentity } from "openclaw/plugin-sdk/channel-outbound"; +import { + bindIngressLifecycleToReplyOptions, + resolveAgentOutboundIdentity, +} from "openclaw/plugin-sdk/channel-outbound"; import { createChannelPairingController } from "openclaw/plugin-sdk/channel-pairing"; import { ensureConfiguredBindingRouteReady, @@ -31,7 +34,11 @@ import { normalizeOptionalString, uniqueStrings } from "openclaw/plugin-sdk/stri import { truncateUtf16Safe } from "openclaw/plugin-sdk/text-utility-runtime"; import { resolveFeishuRuntimeAccount } from "./accounts.js"; import { buildFeishuAgentBody } from "./bot-agent-body.js"; -import { buildBroadcastSessionKey, resolveBroadcastAgents } from "./bot-broadcast.js"; +import { + buildBroadcastSessionKey, + createFeishuBroadcastIngressSettlement, + resolveBroadcastAgents, +} from "./bot-broadcast.js"; import { checkBotMentioned, normalizeFeishuCommandProbeBody, @@ -54,8 +61,8 @@ import { resolveFeishuSenderName, type FeishuPermissionError } from "./bot-sende import { createFeishuClient } from "./client.js"; import { resolveConfiguredFeishuGroupSessionScope } from "./conversation-id.js"; import { + claimUnprocessedFeishuMessage, finalizeFeishuMessageProcessing, - recordProcessedFeishuMessage, type FeishuMessageProcessingClaim, } from "./dedup.js"; import { resolveFeishuMessageDedupeKey } from "./dedupe-key.js"; @@ -81,6 +88,7 @@ import { getMessageFeishu, listFeishuThreadMessages, sendMessageFeishu } from ". import { getFeishuSyntheticDirectPreDispatchTarget } from "./synthetic-event-target.js"; export type { FeishuBotAddedEvent, FeishuMessageEvent } from "./event-types.js"; import type { FeishuMessageEvent } from "./event-types.js"; +import type { FeishuIngressLifecycle } from "./feishu-ingress.js"; import { isFeishuGroupChatType, type FeishuMessageContext, @@ -286,6 +294,7 @@ export async function handleFeishuMessage(params: { accountId?: string; processingClaim?: FeishuMessageProcessingClaim; messageDedupeKey?: string; + turnAdoptionLifecycle?: FeishuIngressLifecycle; }): Promise { const { cfg, @@ -298,6 +307,7 @@ export async function handleFeishuMessage(params: { accountId, processingClaim, messageDedupeKey: messageDedupeKeyOverride, + turnAdoptionLifecycle, } = params; // Resolve account with merged config @@ -310,6 +320,7 @@ export async function handleFeishuMessage(params: { const messageId = event.message.message_id; const messageDedupeKey = messageDedupeKeyOverride ?? resolveFeishuMessageDedupeKey(event); if ( + !turnAdoptionLifecycle && !(await finalizeFeishuMessageProcessing({ messageId: messageDedupeKey, namespace: account.accountId, @@ -1492,16 +1503,45 @@ export async function handleFeishuMessage(params: { // Cross-account dedup: in multi-account setups, Feishu delivers the same // event to every bot account in the group. Only one account should handle // broadcast dispatch to avoid duplicate agent sessions and race conditions. - // Uses a shared "broadcast" namespace (not per-account) so the first handler - // to reach this point claims the message; subsequent accounts skip. - if ( - !(await recordProcessedFeishuMessage(messageDedupeKey ?? ctx.messageId, "broadcast", log)) - ) { + // Hold the shared claim until the complete fan-out adopts. Failed fan-out + // releases it so a transport retry can dispatch the broadcast again. + const broadcastDedupeKey = messageDedupeKey ?? ctx.messageId; + const broadcastClaim = await claimUnprocessedFeishuMessage({ + messageId: broadcastDedupeKey, + namespace: "broadcast", + log, + }); + if (broadcastClaim.kind === "duplicate" || broadcastClaim.kind === "inflight") { log( `feishu[${account.accountId}]: broadcast already claimed by another account for message ${ctx.messageId}; skipping`, ); return; } + const broadcastSettlement = createFeishuBroadcastIngressSettlement({ + lifecycle: turnAdoptionLifecycle, + replayClaim: broadcastClaim.kind === "claimed" ? broadcastClaim.handle : undefined, + onReplayCommitError: (err) => + error( + `feishu[${account.accountId}]: failed to commit broadcast replay guard: ${String(err)}`, + ), + onAdopted: () => { + if (isGroup && historyKey && chatHistories) { + createChannelHistoryWindow({ historyMap: chatHistories }).clear({ + historyKey, + limit: historyLimit, + }); + } + }, + }); + const abandonBroadcast = async (err: unknown) => { + try { + await broadcastSettlement.onDispatchFailed(err); + } catch (abandonError) { + error( + `feishu[${account.accountId}]: failed to abandon broadcast ingress: ${String(abandonError)}`, + ); + } + }; // --- Broadcast dispatch: send message to all configured agents --- const rawStrategy = ( @@ -1518,169 +1558,222 @@ export async function handleFeishuMessage(params: { ); const dispatchForAgent = async (agentId: string) => { - if (hasKnownAgents && !agentIds.includes(normalizeAgentId(agentId))) { + const normalizedAgentId = normalizeAgentId(agentId); + if (hasKnownAgents && !agentIds.includes(normalizedAgentId)) { log( `feishu[${account.accountId}]: broadcast agent ${agentId} not found in agents.list; skipping`, ); return; } - const agentSessionKey = buildBroadcastSessionKey(route.sessionKey, route.agentId, agentId); - const agentStorePath = resolveStorePath(cfg.session?.store, { - agentId, - }); - const agentRecord = { - updateLastRoute: buildFeishuInboundLastRouteUpdate({ - sessionKey: agentSessionKey, - accountId: route.accountId, - }), - onRecordError: (err: unknown) => { - log( - `feishu[${account.accountId}]: failed to record broadcast inbound session ${agentSessionKey}: ${String(err)}`, - ); - }, - }; - const allowReasoningPreview = resolveFeishuReasoningPreviewEnabled({ - cfg, - agentId, - storePath: agentStorePath, - sessionKey: agentSessionKey, - }); - const agentCtx = await buildCtxPayloadForAgent( - agentId, - agentSessionKey, - route.accountId, - ctx.mentionedBot && agentId === activeAgentId, + let agentClaim: Awaited> | undefined; + for (let attempt = 0; attempt < 2; attempt += 1) { + agentClaim = await claimUnprocessedFeishuMessage({ + messageId: broadcastDedupeKey, + namespace: `broadcast:${normalizedAgentId}`, + log, + }); + if (agentClaim.kind === "duplicate") { + return; + } + if (agentClaim.kind !== "inflight") { + break; + } + broadcastSettlement.onLanePending(); + try { + await agentClaim.pending; + return; + } catch (err) { + if (attempt === 1) { + throw err; + } + } + } + const lane = broadcastSettlement.createLane( + agentClaim?.kind === "claimed" ? agentClaim.handle : undefined, ); - if (agentId === activeAgentId) { - // Active agent: real Feishu dispatcher (responds on Feishu) - const identity = resolveAgentOutboundIdentity(cfg, agentId); - const { dispatcherOptions, delivery, replyOptions, ensureNoVisibleReplyFallback } = - createFeishuReplyDispatcher({ - cfg, - agentId, - runtime: runtime as RuntimeEnv, - chatId: ctx.chatId, - sendTarget: feishuTo, - allowReasoningPreview, - replyToMessageId: replyTargetMessageId, - typingTargetMessageId, - skipReplyToInMessages: !isGroup && !directThreadReply, - replyInThread, - rootId: ctx.rootId, - threadReply, - accountId: account.accountId, - identity, - mentionTargets: ctx.mentionTargets, - requiredMentionTargets, - messageCreateTimeMs, + try { + const agentSessionKey = buildBroadcastSessionKey( + route.sessionKey, + route.agentId, + agentId, + ); + const agentStorePath = resolveStorePath(cfg.session?.store, { + agentId, + }); + const agentRecord = { + updateLastRoute: buildFeishuInboundLastRouteUpdate({ sessionKey: agentSessionKey, - }); + accountId: route.accountId, + }), + onRecordError: (err: unknown) => { + log( + `feishu[${account.accountId}]: failed to record broadcast inbound session ${agentSessionKey}: ${String(err)}`, + ); + }, + }; + const allowReasoningPreview = resolveFeishuReasoningPreviewEnabled({ + cfg, + agentId, + storePath: agentStorePath, + sessionKey: agentSessionKey, + }); + const agentCtx = await buildCtxPayloadForAgent( + agentId, + agentSessionKey, + route.accountId, + ctx.mentionedBot && agentId === activeAgentId, + ); - log( - `feishu[${account.accountId}]: broadcast active dispatch agent=${agentId} (session=${agentSessionKey})`, - ); - const turnResult = await core.channel.inbound.run({ - channel: "feishu", - accountId: route.accountId, - raw: ctx, - adapter: { - ingest: () => ({ - id: ctx.messageId, - timestamp: messageCreateTimeMs, - rawText: ctx.content, - textForAgent: agentCtx.BodyForAgent, - textForCommands: agentCtx.CommandBody, - raw: ctx, - }), - resolveTurn: () => ({ + if (agentId === activeAgentId) { + // Active agent: real Feishu dispatcher (responds on Feishu) + const identity = resolveAgentOutboundIdentity(cfg, agentId); + const { dispatcherOptions, delivery, replyOptions, ensureNoVisibleReplyFallback } = + createFeishuReplyDispatcher({ cfg, - channel: "feishu", - accountId: route.accountId, - route: { agentId, sessionKey: agentSessionKey }, - ctxPayload: agentCtx, - record: agentRecord, - dispatcherOptions, - delivery, - replyOptions, - }), - }, - }); - if ( - turnResult.dispatched && - shouldSendNoVisibleReplyFallback(turnResult.dispatchResult) - ) { - await ensureNoVisibleReplyFallback("broadcast-dispatch-complete-no-visible-reply"); + agentId, + runtime: runtime as RuntimeEnv, + chatId: ctx.chatId, + sendTarget: feishuTo, + allowReasoningPreview, + replyToMessageId: replyTargetMessageId, + typingTargetMessageId, + skipReplyToInMessages: !isGroup && !directThreadReply, + replyInThread, + rootId: ctx.rootId, + threadReply, + accountId: account.accountId, + identity, + mentionTargets: ctx.mentionTargets, + requiredMentionTargets, + messageCreateTimeMs, + sessionKey: agentSessionKey, + }); + + log( + `feishu[${account.accountId}]: broadcast active dispatch agent=${agentId} (session=${agentSessionKey})`, + ); + const turnResult = await core.channel.inbound.run({ + channel: "feishu", + accountId: route.accountId, + raw: ctx, + adapter: { + ingest: () => ({ + id: ctx.messageId, + timestamp: messageCreateTimeMs, + rawText: ctx.content, + textForAgent: agentCtx.BodyForAgent, + textForCommands: agentCtx.CommandBody, + raw: ctx, + }), + resolveTurn: () => ({ + cfg, + channel: "feishu", + accountId: route.accountId, + route: { agentId, sessionKey: agentSessionKey }, + ctxPayload: agentCtx, + record: agentRecord, + dispatcherOptions, + delivery, + replyOptions: { + ...replyOptions, + ...bindIngressLifecycleToReplyOptions(lane.lifecycle), + }, + }), + }, + }); + if ( + turnResult.dispatched && + shouldSendNoVisibleReplyFallback(turnResult.dispatchResult) + ) { + await ensureNoVisibleReplyFallback("broadcast-dispatch-complete-no-visible-reply"); + } + await lane.onDispatchComplete(turnResult.dispatched); + } else { + // Observer agent: no-op dispatcher (session entry + inference, no Feishu reply). + // Strip CommandAuthorized so slash commands (e.g. /reset) don't silently + // mutate observer sessions — only the active agent should execute commands. + delete (agentCtx as Record).CommandAuthorized; + log( + `feishu[${account.accountId}]: broadcast observer dispatch agent=${agentId} (session=${agentSessionKey})`, + ); + const turnResult = await core.channel.inbound.run({ + channel: "feishu", + accountId: route.accountId, + raw: ctx, + adapter: { + ingest: () => ({ + id: ctx.messageId, + timestamp: messageCreateTimeMs, + rawText: ctx.content, + textForAgent: agentCtx.BodyForAgent, + textForCommands: agentCtx.CommandBody, + raw: ctx, + }), + resolveTurn: () => ({ + cfg, + channel: "feishu", + accountId: route.accountId, + route: { agentId, sessionKey: agentSessionKey }, + ctxPayload: agentCtx, + record: agentRecord, + admission: { kind: "observeOnly", reason: "broadcast-observer" }, + delivery: { + deliver: async () => ({ visibleReplySent: false }), + }, + replyOptions: bindIngressLifecycleToReplyOptions(lane.lifecycle), + }), + }, + }); + await lane.onDispatchComplete(turnResult.dispatched); } - } else { - // Observer agent: no-op dispatcher (session entry + inference, no Feishu reply). - // Strip CommandAuthorized so slash commands (e.g. /reset) don't silently - // mutate observer sessions — only the active agent should execute commands. - delete (agentCtx as Record).CommandAuthorized; - log( - `feishu[${account.accountId}]: broadcast observer dispatch agent=${agentId} (session=${agentSessionKey})`, - ); - await core.channel.inbound.run({ - channel: "feishu", - accountId: route.accountId, - raw: ctx, - adapter: { - ingest: () => ({ - id: ctx.messageId, - timestamp: messageCreateTimeMs, - rawText: ctx.content, - textForAgent: agentCtx.BodyForAgent, - textForCommands: agentCtx.CommandBody, - raw: ctx, - }), - resolveTurn: () => ({ - cfg, - channel: "feishu", - accountId: route.accountId, - route: { agentId, sessionKey: agentSessionKey }, - ctxPayload: agentCtx, - record: agentRecord, - admission: { kind: "observeOnly", reason: "broadcast-observer" }, - delivery: { - deliver: async () => ({ visibleReplySent: false }), - }, - }), - }, - }); + } catch (err) { + await lane.onDispatchFailed(err); + throw err; } }; + const results: PromiseSettledResult[] = []; if (strategy === "sequential") { for (const agentId of broadcastAgents) { try { await dispatchForAgent(agentId); - } catch (err) { - log( - `feishu[${account.accountId}]: broadcast dispatch failed for agent=${agentId}: ${String(err)}`, - ); + results.push({ status: "fulfilled", value: undefined }); + } catch (reason) { + results.push({ status: "rejected", reason }); } } } else { - const results = await Promise.allSettled(broadcastAgents.map(dispatchForAgent)); - for (const [i, result] of results.entries()) { - if (result.status === "rejected") { - const agentId = broadcastAgents.at(i); - if (agentId === undefined) { - continue; - } - log( - `feishu[${account.accountId}]: broadcast dispatch failed for agent=${agentId}: ${String(result.reason)}`, - ); + results.push(...(await Promise.allSettled(broadcastAgents.map(dispatchForAgent)))); + } + const failures: unknown[] = []; + for (const [i, result] of results.entries()) { + if (result.status === "rejected") { + const agentId = broadcastAgents.at(i); + if (agentId === undefined) { + continue; } + log( + `feishu[${account.accountId}]: broadcast dispatch failed for agent=${agentId}: ${String(result.reason)}`, + ); + failures.push(result.reason); } } + if (failures.length > 0) { + const failure = + failures.length === 1 + ? failures[0] + : new AggregateError(failures, "Feishu broadcast dispatch failed"); + await abandonBroadcast(failure); + throw failure; + } - if (isGroup && historyKey && chatHistories) { - createChannelHistoryWindow({ historyMap: chatHistories }).clear({ - historyKey, - limit: historyLimit, - }); + try { + await broadcastSettlement.onDispatchComplete(); + } catch (err) { + await abandonBroadcast(err); + throw err; } log( @@ -1766,7 +1859,12 @@ export async function handleFeishuMessage(params: { }, dispatcherOptions, delivery, - replyOptions, + replyOptions: { + ...replyOptions, + ...(turnAdoptionLifecycle + ? bindIngressLifecycleToReplyOptions(turnAdoptionLifecycle) + : {}), + }, }), }, }); @@ -1785,6 +1883,9 @@ export async function handleFeishuMessage(params: { } } catch (err) { error(`feishu[${account.accountId}]: failed to dispatch message: ${String(err)}`); + if (turnAdoptionLifecycle) { + throw err; + } } } /* oxlint-disable max-lines -- TODO: split this grandfathered oversized file. */ diff --git a/extensions/feishu/src/comment-handler.ts b/extensions/feishu/src/comment-handler.ts index 6714c1456d52..1a9a60212c1f 100644 --- a/extensions/feishu/src/comment-handler.ts +++ b/extensions/feishu/src/comment-handler.ts @@ -1,5 +1,6 @@ // Feishu plugin module implements comment handler behavior. import { buildChannelInboundEventContext } from "openclaw/plugin-sdk/channel-inbound"; +import { bindIngressLifecycleToReplyOptions } from "openclaw/plugin-sdk/channel-outbound"; import { parseStrictNonNegativeInteger } from "openclaw/plugin-sdk/number-runtime"; import type { ResolvedAgentRoute } from "openclaw/plugin-sdk/routing"; import { resolveFeishuRuntimeAccount } from "./accounts.js"; @@ -13,6 +14,7 @@ import { import { buildFeishuCommentTarget } from "./comment-target.js"; import { deliverCommentThreadText } from "./drive.js"; import { maybeCreateDynamicAgent } from "./dynamic-agent.js"; +import type { FeishuIngressLifecycle } from "./feishu-ingress.js"; import { resolveDriveCommentEventTurn, type FeishuDriveCommentNoticeEvent, @@ -27,6 +29,7 @@ type HandleFeishuCommentEventParams = { event: FeishuDriveCommentNoticeEvent; botOpenId?: string; abortSignal?: AbortSignal; + turnAdoptionLifecycle?: FeishuIngressLifecycle; }; function buildCommentSessionKey(params: { @@ -295,6 +298,9 @@ export async function handleFeishuCommentEvent( }, dispatcherOptions, delivery, + ...(params.turnAdoptionLifecycle + ? { replyOptions: bindIngressLifecycleToReplyOptions(params.turnAdoptionLifecycle) } + : {}), }), }, }); diff --git a/extensions/feishu/src/dedup-state.ts b/extensions/feishu/src/dedup-state.ts index c64c70db796d..b75ff1ad905a 100644 --- a/extensions/feishu/src/dedup-state.ts +++ b/extensions/feishu/src/dedup-state.ts @@ -1,5 +1,6 @@ import { createChannelReplayGuard } from "openclaw/plugin-sdk/persistent-dedupe"; +// 24h/10k preserves the pre-drain logical-twin guard window across restarts. const DEDUPE_NAMESPACE_PREFIX = "feishu.dedup"; const DEDUP_TTL_MS = 24 * 60 * 60 * 1000; const MEMORY_MAX_SIZE = 1_000; diff --git a/extensions/feishu/src/dedup.test.ts b/extensions/feishu/src/dedup.test.ts index aa3fa1de04b3..3f760cf5ff49 100644 --- a/extensions/feishu/src/dedup.test.ts +++ b/extensions/feishu/src/dedup.test.ts @@ -9,7 +9,6 @@ import { claimUnprocessedFeishuMessage, finalizeFeishuMessageProcessing, hasProcessedFeishuMessage, - recordProcessedFeishuMessage, warmupDedupFromPluginState, } from "./dedup.js"; @@ -85,25 +84,37 @@ describe("Feishu claimable dedupe", () => { it("dedupes cross-account broadcast claims through the shared namespace", async () => { // Multi-account groups deliver the same event once per bot account; the // shared "broadcast" namespace lets the first account claim dispatch. - await expect(recordProcessedFeishuMessage("msg-6", "broadcast")).resolves.toBe(true); - await expect(recordProcessedFeishuMessage("msg-6", "broadcast")).resolves.toBe(false); + await expect( + finalizeFeishuMessageProcessing({ messageId: "msg-6", namespace: "broadcast" }), + ).resolves.toBe(true); + await expect( + finalizeFeishuMessageProcessing({ messageId: "msg-6", namespace: "broadcast" }), + ).resolves.toBe(false); await restartFeishuDedup(); - await expect(recordProcessedFeishuMessage("msg-6", "broadcast")).resolves.toBe(false); + await expect( + finalizeFeishuMessageProcessing({ messageId: "msg-6", namespace: "broadcast" }), + ).resolves.toBe(false); }); it("warms memory from persisted plugin state", async () => { - await expect(recordProcessedFeishuMessage("msg-7", "account-a")).resolves.toBe(true); + await expect( + finalizeFeishuMessageProcessing({ messageId: "msg-7", namespace: "account-a" }), + ).resolves.toBe(true); await restartFeishuDedup(); await expect(warmupDedupFromPluginState("account-a")).resolves.toBe(1); - await expect(recordProcessedFeishuMessage("msg-7", "account-a")).resolves.toBe(false); + await expect( + finalizeFeishuMessageProcessing({ messageId: "msg-7", namespace: "account-a" }), + ).resolves.toBe(false); }); it("ignores committed messages after the TTL expires", async () => { vi.useFakeTimers(); vi.setSystemTime(1_000); - await expect(recordProcessedFeishuMessage("msg-8", "account-a")).resolves.toBe(true); + await expect( + finalizeFeishuMessageProcessing({ messageId: "msg-8", namespace: "account-a" }), + ).resolves.toBe(true); await restartFeishuDedup(); vi.setSystemTime(1_000 + 24 * 60 * 60 * 1000 + 1); @@ -117,7 +128,9 @@ describe("Feishu claimable dedupe", () => { process.env.OPENCLAW_STATE_DIR = path.join(blockedPath, "nested"); const log = vi.fn(); - await expect(recordProcessedFeishuMessage("msg-9", "account-a", log)).resolves.toBe(true); + await expect( + finalizeFeishuMessageProcessing({ messageId: "msg-9", namespace: "account-a", log }), + ).resolves.toBe(true); await expect( claimUnprocessedFeishuMessage({ messageId: "msg-9", namespace: "account-a", log }), ).resolves.toEqual({ kind: "duplicate" }); diff --git a/extensions/feishu/src/dedup.ts b/extensions/feishu/src/dedup.ts index f990f3b69581..56d276284f71 100644 --- a/extensions/feishu/src/dedup.ts +++ b/extensions/feishu/src/dedup.ts @@ -1,8 +1,7 @@ -// Feishu inbound replay protection rides the core claimable dedupe: Feishu -// redelivers events after reconnects/restarts and multi-account groups receive -// the same event once per bot, so handlers claim a dedupe key before -// processing, commit once handling is dispatched, and release on retryable -// failure so the event can be redelivered. +// PERMANENT logical-identity guard above durable event_id tombstones. Feishu +// can redeliver one text message with a fresh message_id/event_id (#46778), +// and multi-account groups receive one logical broadcast per bot account. +// Queue tombstones cannot cover either twin; claims commit at turn adoption. import type { ChannelReplayClaimHandle } from "openclaw/plugin-sdk/persistent-dedupe"; import { feishuDedupeState } from "./dedup-state.js"; @@ -13,7 +12,7 @@ export type FeishuMessageProcessingClaim = ChannelReplayClaimHandle; type FeishuMessageClaim = | { kind: "claimed"; handle: FeishuMessageProcessingClaim } | { kind: "duplicate" } - | { kind: "inflight" } + | { kind: "inflight"; pending: Promise } | { kind: "invalid" }; function dedupeKey(messageId: string | undefined | null): string { @@ -49,9 +48,6 @@ export async function claimUnprocessedFeishuMessage(params: { params.messageId, dedupeOptions(params.namespace, params.log), ); - if (claim.kind === "inflight") { - return { kind: "inflight" }; - } return claim; } @@ -78,16 +74,6 @@ export async function finalizeFeishuMessageProcessing(params: { return await ("kind" in claim ? claim.handle : claim).commit(); } -/** Records a handled message so restart/replay cannot dispatch it again; false when already recorded. */ -export async function recordProcessedFeishuMessage( - messageId: string | undefined | null, - namespace = "global", - log?: FeishuDedupeLog, -): Promise { - const claim = await feishuDedupeState.guard.claim(messageId, dedupeOptions(namespace, log)); - return claim.kind === "claimed" ? await claim.handle.commit() : false; -} - /** Forgets a recorded message so a retryable synthetic event can be handled on redelivery. */ export async function forgetProcessedFeishuMessage( messageId: string | undefined | null, diff --git a/extensions/feishu/src/delivery-trace.test.ts b/extensions/feishu/src/delivery-trace.test.ts index f4314d07aacd..c1499a005adc 100644 --- a/extensions/feishu/src/delivery-trace.test.ts +++ b/extensions/feishu/src/delivery-trace.test.ts @@ -1,7 +1,7 @@ // Feishu delivery trace goldens: replayable wire-level lifecycle recordings. // -// IN events are fed straight into the reply-dispatcher wiring (the options -// createReplyDispatcherWithTyping receives plus replyOptions callbacks); OUT +// IN events are fed straight into the reply-dispatcher plan (dispatcher, +// delivery, and replyOptions callbacks); OUT // events are recorded at the mocked Lark SDK client and the mocked CardKit // HTTP fetch, so streaming-card entity calls are captured at the wire seam. // Refresh goldens with OPENCLAW_TRACE_UPDATE=1 (see delivery-trace harness docs). diff --git a/extensions/feishu/src/feishu-ingress.test.ts b/extensions/feishu/src/feishu-ingress.test.ts new file mode 100644 index 000000000000..1d653c8d57a8 --- /dev/null +++ b/extensions/feishu/src/feishu-ingress.test.ts @@ -0,0 +1,519 @@ +// Feishu durable ingress tests cover ack gating, recovery, tombstones, and logical twins. +import crypto from "node:crypto"; +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import type * as Lark from "@larksuiteoapi/node-sdk"; +import { + closeOpenClawStateDatabaseForTest, + createChannelIngressQueueForTests, +} from "openclaw/plugin-sdk/plugin-state-test-runtime"; +import { createNonExitingRuntimeEnv } from "openclaw/plugin-sdk/plugin-test-runtime"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { feishuDedupeState } from "./dedup-state.js"; +import { claimUnprocessedFeishuMessage } from "./dedup.js"; +import { resolveFeishuMessageDedupeKey } from "./dedupe-key.js"; +import type { FeishuMessageEvent } from "./event-types.js"; +import { + buildFeishuFlushIngressLifecycle, + createFeishuDurableIngress, + type FeishuIngressLifecycle, +} from "./feishu-ingress.js"; +import { monitorWebhook } from "./monitor.transport.js"; +import { getFreePort, waitUntilServerReady } from "./monitor.webhook.test-helpers.js"; +import type { ResolvedFeishuAccount } from "./types.js"; + +type FeishuIngressQueue = NonNullable[0]["queue"]>; +type FeishuIngressPayload = Parameters[1]; + +function messageEnvelope(params: { + eventId: string; + messageId?: string; + chatId?: string; + createTime?: string; + text?: string; +}) { + return { + schema: "2.0", + header: { + event_id: params.eventId, + event_type: "im.message.receive_v1", + create_time: "1710000000000", + }, + event: { + sender: { + sender_id: { open_id: "ou-user" }, + sender_type: "user", + }, + message: { + message_id: params.messageId ?? "om-message", + chat_id: params.chatId ?? "oc-chat", + chat_type: "p2p", + message_type: "text", + content: JSON.stringify({ text: params.text ?? "hello" }), + create_time: params.createTime ?? "1710000000000", + }, + }, + }; +} + +function flattenEnvelope(envelope: ReturnType) { + return { ...envelope.header, ...envelope.event }; +} + +function createLifecycle() { + const calls = { + adopted: vi.fn(async () => {}), + deferred: vi.fn(), + finalizing: vi.fn(), + abandoned: vi.fn(async () => {}), + }; + const lifecycle: FeishuIngressLifecycle = { + abortSignal: new AbortController().signal, + onAdopted: calls.adopted, + onDeferred: calls.deferred, + onAdoptionFinalizing: calls.finalizing, + onAbandoned: calls.abandoned, + }; + return { calls, lifecycle }; +} + +function createDispatcher( + run: (data: ReturnType) => unknown = async () => undefined, +): Pick { + return { + invoke: async (data) => await run(data as ReturnType), + }; +} + +function startIngress(params: { + queue: FeishuIngressQueue; + dispatcher: Pick; +}) { + return createFeishuDurableIngress({ + accountId: "default", + queue: params.queue, + dispatcher: params.dispatcher, + runtime: { error: vi.fn(), log: vi.fn() }, + pollIntervalMs: 60_000, + adoptionStallTimeoutMs: 5_000, + }); +} + +async function withQueue(fn: (queue: FeishuIngressQueue, stateDir: string) => Promise) { + const created = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-feishu-ingress-")); + const stateDir = await fs.realpath(created); + const previousStateDir = process.env.OPENCLAW_STATE_DIR; + process.env.OPENCLAW_STATE_DIR = stateDir; + const queue = createChannelIngressQueueForTests({ + channelId: "feishu", + accountId: "default", + stateDir, + }); + try { + return await fn(queue, stateDir); + } finally { + feishuDedupeState.reset(); + closeOpenClawStateDatabaseForTest(); + if (previousStateDir === undefined) { + delete process.env.OPENCLAW_STATE_DIR; + } else { + process.env.OPENCLAW_STATE_DIR = previousStateDir; + } + await fs.rm(stateDir, { recursive: true, force: true }); + } +} + +function signWebhookBody(rawBody: string, encryptKey: string): Record { + const timestamp = "1711111111"; + const nonce = "feishu-ingress-test"; + return { + "content-type": "application/json", + "x-lark-request-timestamp": timestamp, + "x-lark-request-nonce": nonce, + "x-lark-signature": crypto + .createHash("sha256") + .update(timestamp + nonce + encryptKey + rawBody) + .digest("hex"), + }; +} + +async function withWebhook( + eventDispatcher: Pick, + run: (url: string) => Promise, +) { + const port = await getFreePort(); + const webhookPath = "/feishu-ingress-test"; + const encryptKey = "feishu-ingress-test-key"; + const account = { + accountId: "default", + encryptKey, + config: { webhookHost: "127.0.0.1", webhookPort: port, webhookPath }, + } as ResolvedFeishuAccount; + const abortController = new AbortController(); + const monitor = monitorWebhook({ + account, + accountId: account.accountId, + eventDispatcher: eventDispatcher as Lark.EventDispatcher, + abortSignal: abortController.signal, + runtime: createNonExitingRuntimeEnv(), + }); + const url = `http://127.0.0.1:${port}${webhookPath}`; + await waitUntilServerReady(url); + try { + await run(url); + } finally { + abortController.abort(); + await monitor; + } +} + +async function postWebhook(url: string, envelope: ReturnType) { + const rawBody = JSON.stringify(envelope); + return await fetch(url, { + method: "POST", + headers: signWebhookBody(rawBody, "feishu-ingress-test-key"), + body: rawBody, + }); +} + +afterEach(() => { + feishuDedupeState.reset(); + closeOpenClawStateDatabaseForTest(); + vi.restoreAllMocks(); +}); + +describe("Feishu durable ingress", () => { + it("waits for durable append before acknowledging the webhook", async () => { + await withQueue(async (queue) => { + let releaseAppend!: () => void; + const appendGate = new Promise((resolve) => { + releaseAppend = resolve; + }); + const enqueue = vi.fn(async (...args: Parameters) => { + await appendGate; + return await queue.enqueue(...args); + }); + const gatedQueue = { ...queue, enqueue } as FeishuIngressQueue; + const ingress = startIngress({ queue: gatedQueue, dispatcher: createDispatcher() }); + + await withWebhook({ invoke: ingress.invoke }, async (url) => { + let responseSettled = false; + const responsePromise = postWebhook( + url, + messageEnvelope({ eventId: "evt-append-gate" }), + ).then((response) => { + responseSettled = true; + return response; + }); + await vi.waitFor(() => expect(enqueue).toHaveBeenCalledTimes(1)); + expect(responseSettled).toBe(false); + + releaseAppend(); + await expect(responsePromise.then((response) => response.status)).resolves.toBe(200); + }); + await ingress.stop(); + }); + }); + + it("returns non-2xx when durable append fails", async () => { + await withQueue(async (queue) => { + const enqueue = vi.fn(async () => { + throw new Error("sqlite unavailable"); + }); + const failingQueue = { ...queue, enqueue } as FeishuIngressQueue; + const dispatch = vi.fn(async () => undefined); + const ingress = startIngress({ queue: failingQueue, dispatcher: createDispatcher(dispatch) }); + + await withWebhook({ invoke: ingress.invoke }, async (url) => { + const response = await postWebhook(url, messageEnvelope({ eventId: "evt-append-fail" })); + expect(response.status).toBe(500); + }); + + expect(enqueue).toHaveBeenCalledTimes(3); + expect(dispatch).not.toHaveBeenCalled(); + await ingress.stop(); + }); + }); + + it("recovers an uncompleted envelope with a fresh drain and dispatches exactly once", async () => { + await withQueue(async (queue) => { + const interrupted = startIngress({ queue, dispatcher: createDispatcher() }); + await interrupted.invoke(messageEnvelope({ eventId: "evt-restart" }), { needCheck: false }); + await interrupted.stop(); + + const dispatch = vi.fn(async (data: ReturnType) => { + await recovered.resolveLifecycle(flattenEnvelope(data))?.onAdopted(); + }); + const recovered = startIngress({ queue, dispatcher: createDispatcher(dispatch) }); + recovered.start(); + await recovered.waitForIdle(); + + expect(dispatch).toHaveBeenCalledTimes(1); + expect((await queue.enqueue("evt-restart", {} as FeishuIngressPayload)).kind).toBe( + "completed", + ); + await recovered.stop(); + }); + }); + + it("retains completion so one event_id dispatches only once", async () => { + await withQueue(async (queue) => { + const dispatch = vi.fn(async (data: ReturnType) => { + await ingress.resolveLifecycle(flattenEnvelope(data))?.onAdopted(); + }); + const ingress = startIngress({ queue, dispatcher: createDispatcher(dispatch) }); + ingress.start(); + const envelope = messageEnvelope({ eventId: "evt-completed" }); + + await ingress.invoke(envelope, { needCheck: false }); + await ingress.waitForIdle(); + await ingress.invoke(envelope, { needCheck: false }); + await ingress.waitForIdle(); + + expect(dispatch).toHaveBeenCalledTimes(1); + await ingress.stop(); + }); + }); + + it("dead-letters authentication failures without retrying", async () => { + await withQueue(async (queue) => { + const dispatch = vi.fn(async () => { + throw Object.assign(new Error("unauthorized"), { status: 401 }); + }); + const ingress = startIngress({ queue, dispatcher: createDispatcher(dispatch) }); + ingress.start(); + const envelope = messageEnvelope({ eventId: "evt-auth-failure" }); + + await ingress.invoke(envelope, { needCheck: false }); + await ingress.waitForIdle(); + + expect(dispatch).toHaveBeenCalledTimes(1); + const duplicate = await queue.enqueue("evt-auth-failure", {} as FeishuIngressPayload); + expect(duplicate.kind).toBe("failed"); + if (duplicate.kind === "failed") { + expect(duplicate.record.reason).toBe("authentication-failed"); + } + await ingress.stop(); + }); + }); + + it("durably dead-letters recognized envelopes without conversation identity", async () => { + await withQueue(async (queue) => { + const dispatch = vi.fn(async () => undefined); + const ingress = startIngress({ queue, dispatcher: createDispatcher(dispatch) }); + ingress.start(); + const envelope = messageEnvelope({ eventId: "evt-missing-chat" }); + delete (envelope.event.message as { chat_id?: string }).chat_id; + + await expect(ingress.invoke(envelope, { needCheck: false })).resolves.toBeUndefined(); + await ingress.waitForIdle(); + + expect(dispatch).not.toHaveBeenCalled(); + const duplicate = await queue.enqueue("evt-missing-chat", {} as FeishuIngressPayload); + expect(duplicate.kind).toBe("failed"); + if (duplicate.kind === "failed") { + expect(duplicate.record.reason).toBe("invalid-event"); + } + await ingress.stop(); + }); + }); + + it("keeps transient dispatch failures retryable", async () => { + await withQueue(async (queue) => { + const dispatch = vi.fn(async () => { + throw new Error("temporary network failure"); + }); + const ingress = startIngress({ queue, dispatcher: createDispatcher(dispatch) }); + ingress.start(); + const envelope = messageEnvelope({ eventId: "evt-transient-failure" }); + + await ingress.invoke(envelope, { needCheck: false }); + await ingress.waitForIdle(); + + expect(dispatch).toHaveBeenCalledTimes(1); + await expect( + queue.enqueue("evt-transient-failure", {} as FeishuIngressPayload), + ).resolves.toMatchObject({ kind: "pending", duplicate: true }); + await ingress.stop(); + }); + }); + + it("keeps unrelated downstream syntax failures retryable", async () => { + await withQueue(async (queue) => { + const dispatch = vi.fn(async () => { + throw new SyntaxError("temporary API response parse failure"); + }); + const ingress = startIngress({ queue, dispatcher: createDispatcher(dispatch) }); + ingress.start(); + const envelope = messageEnvelope({ eventId: "evt-downstream-syntax" }); + + await ingress.invoke(envelope, { needCheck: false }); + await ingress.waitForIdle(); + + expect(dispatch).toHaveBeenCalledTimes(1); + await expect( + queue.enqueue("evt-downstream-syntax", {} as FeishuIngressPayload), + ).resolves.toMatchObject({ kind: "pending", duplicate: true }); + await ingress.stop(); + }); + }); + + it("keeps the permanent logical guard for different event_id and message_id twins", async () => { + await withQueue(async () => { + const firstEnvelope = messageEnvelope({ + eventId: "evt-twin-a", + messageId: "om-twin-a", + }); + const secondEnvelope = messageEnvelope({ + eventId: "evt-twin-b", + messageId: "om-twin-b", + }); + const first = firstEnvelope.event as FeishuMessageEvent; + const second = secondEnvelope.event as FeishuMessageEvent; + const firstKey = resolveFeishuMessageDedupeKey(first); + const secondKey = resolveFeishuMessageDedupeKey(second); + expect(firstEnvelope.header.event_id).not.toBe(secondEnvelope.header.event_id); + expect(firstKey).toBe(secondKey); + + const claim = await claimUnprocessedFeishuMessage({ + messageId: firstKey, + namespace: "default", + }); + expect(claim.kind).toBe("claimed"); + if (claim.kind !== "claimed") { + throw new Error(`expected claimed logical twin, received ${claim.kind}`); + } + const transport = createLifecycle(); + const { lifecycle } = buildFeishuFlushIngressLifecycle([ + { lifecycle: transport.lifecycle, replayClaim: claim.handle }, + ]); + lifecycle?.onAdoptionFinalizing(); + await lifecycle?.onAdopted(); + + await expect( + claimUnprocessedFeishuMessage({ messageId: secondKey, namespace: "default" }), + ).resolves.toEqual({ kind: "duplicate" }); + expect(transport.calls.finalizing).toHaveBeenCalledTimes(1); + expect(transport.calls.adopted).toHaveBeenCalledTimes(1); + }); + }); + + it("does not reopen adopted transport claims when a logical guard commit fails", async () => { + const transport = createLifecycle(); + const onReplayCommitError = vi.fn(); + const replayClaim = { + keys: ["failing-adoption"] as const, + commit: vi.fn(async () => { + throw new Error("dedupe persistence failed"); + }), + release: vi.fn(), + }; + const { lifecycle } = buildFeishuFlushIngressLifecycle( + [{ lifecycle: transport.lifecycle, replayClaim }], + { onReplayCommitError }, + ); + + await expect(lifecycle?.onAdopted()).resolves.toBeUndefined(); + + expect(transport.calls.adopted).toHaveBeenCalledTimes(1); + expect(transport.calls.abandoned).not.toHaveBeenCalled(); + expect(onReplayCommitError).toHaveBeenCalledWith( + expect.objectContaining({ message: "dedupe persistence failed" }), + ); + }); + + it("abandons logical claims when transport adoption persistence fails", async () => { + const transport = createLifecycle(); + transport.lifecycle.onAdopted = vi.fn(async () => { + throw new Error("queue completion failed"); + }); + const replayClaim = { + keys: ["transport-adoption-failure"] as const, + commit: vi.fn(async () => true), + release: vi.fn(), + }; + const { lifecycle } = buildFeishuFlushIngressLifecycle([ + { lifecycle: transport.lifecycle, replayClaim }, + ]); + + await expect(lifecycle?.onAdopted()).rejects.toThrow("queue completion failed"); + + expect(replayClaim.commit).not.toHaveBeenCalled(); + expect(replayClaim.release).toHaveBeenCalledTimes(1); + expect(transport.calls.abandoned).toHaveBeenCalledTimes(1); + }); + + it("does not adopt while constituent abandonment is in progress", async () => { + const transport = createLifecycle(); + let finishAbandonment!: () => void; + const abandonmentGate = new Promise((resolve) => { + finishAbandonment = resolve; + }); + transport.lifecycle.onAbandoned = vi.fn(async () => await abandonmentGate); + const replayClaim = { + keys: ["concurrent-terminal-state"] as const, + commit: vi.fn(async () => true), + release: vi.fn(), + }; + const { lifecycle } = buildFeishuFlushIngressLifecycle([ + { lifecycle: transport.lifecycle, replayClaim }, + ]); + + const abandonment = lifecycle?.onAbandoned(); + await vi.waitFor(() => { + expect(transport.lifecycle.onAbandoned).toHaveBeenCalledTimes(1); + }); + const adoption = lifecycle?.onAdopted(); + expect(transport.calls.adopted).not.toHaveBeenCalled(); + + finishAbandonment(); + await Promise.all([abandonment, adoption]); + + expect(transport.calls.adopted).not.toHaveBeenCalled(); + expect(replayClaim.commit).not.toHaveBeenCalled(); + expect(replayClaim.release).toHaveBeenCalledTimes(1); + }); + + it("abandons a gated claim when terminal completion persistence fails", async () => { + const transport = createLifecycle(); + transport.lifecycle.onAdopted = vi.fn(async () => { + throw new Error("queue completion failed"); + }); + const { settle } = buildFeishuFlushIngressLifecycle([{ lifecycle: transport.lifecycle }]); + + await expect(settle()).rejects.toThrow("queue completion failed"); + + expect(transport.calls.finalizing).toHaveBeenCalledTimes(1); + expect(transport.calls.abandoned).toHaveBeenCalledTimes(1); + }); + + it("finalizes gated claims before persisting terminal completion", async () => { + const transport = createLifecycle(); + const { settle } = buildFeishuFlushIngressLifecycle([{ lifecycle: transport.lifecycle }]); + + await settle(); + + expect(transport.calls.finalizing).toHaveBeenCalledTimes(1); + expect(transport.calls.adopted).toHaveBeenCalledTimes(1); + expect(transport.calls.finalizing.mock.invocationCallOrder[0]).toBeLessThan( + transport.calls.adopted.mock.invocationCallOrder[0] ?? Number.POSITIVE_INFINITY, + ); + }); + + it("does not settle an observer fallback after an active reply lane defers", async () => { + const transport = createLifecycle(); + const { lifecycle, settle } = buildFeishuFlushIngressLifecycle([ + { lifecycle: transport.lifecycle }, + ]); + + lifecycle?.onDeferred(); + await settle(); + + expect(transport.calls.deferred).toHaveBeenCalledTimes(1); + expect(transport.calls.adopted).not.toHaveBeenCalled(); + + await lifecycle?.onAdopted(); + expect(transport.calls.adopted).toHaveBeenCalledTimes(1); + }); +}); diff --git a/extensions/feishu/src/feishu-ingress.ts b/extensions/feishu/src/feishu-ingress.ts new file mode 100644 index 000000000000..b0ad5f21d655 --- /dev/null +++ b/extensions/feishu/src/feishu-ingress.ts @@ -0,0 +1,632 @@ +// Feishu plugin owns raw Lark event admission, replay, and turn adoption. +import * as Lark from "@larksuiteoapi/node-sdk"; +import { + createChannelIngressDrain, + DEFAULT_INGRESS_ADOPTION_STALL_MS, + DEFAULT_INGRESS_RETRY_DEAD_LETTER_MIN_AGE_MS, + DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS, + type ChannelIngressDrain, + type ChannelIngressQueue, +} from "openclaw/plugin-sdk/channel-outbound"; +import { collectErrorGraphCandidates, formatErrorMessage } from "openclaw/plugin-sdk/error-runtime"; +import type { ChannelReplayClaimHandle } from "openclaw/plugin-sdk/persistent-dedupe"; +import { getFeishuRuntime } from "./runtime.js"; + +const FEISHU_INGRESS_PAYLOAD_VERSION = 1; +const FEISHU_INGRESS_POLL_INTERVAL_MS = 500; +const FEISHU_INGRESS_PRUNE_INTERVAL_MS = 60 * 60 * 1_000; +const FEISHU_INGRESS_COMPLETED_TTL_MS = 30 * 24 * 60 * 60 * 1_000; +const FEISHU_INGRESS_COMPLETED_MAX_ENTRIES = 20_000; +const FEISHU_INGRESS_FAILED_TTL_MS = 30 * 24 * 60 * 60 * 1_000; +const FEISHU_INGRESS_FAILED_MAX_ENTRIES = 20_000; +const FEISHU_DURABLE_EVENT_TYPES = new Set([ + "drive.notice.comment_add_v1", + "im.message.receive_v1", +]); + +export type FeishuIngressLifecycle = { + abortSignal: AbortSignal; + onAdopted: () => void | Promise; + onDeferred: () => void; + onAdoptionFinalizing: () => void; + onAbandoned: () => void | Promise; + registerAbandonHandler?: (handler: () => void | Promise) => () => void; +}; + +type FeishuIngressPayload = { + version: 1; + receivedAt: number; + rawEnvelope: string; +}; + +type FeishuIngressFacts = { + eventId: string; + eventType: string; + laneKey: string; +}; + +type FeishuIngressOptions = { + accountId: string; + dispatcher: Pick; + encryptKey?: string; + runtime: { + error?: (message: string) => void; + log?: (message: string) => void; + }; + queue?: ChannelIngressQueue; + pollIntervalMs?: number; + adoptionStallTimeoutMs?: number; +}; + +type FeishuDurableIngress = { + invoke: Lark.EventDispatcher["invoke"]; + resolveLifecycle: (data: unknown) => FeishuIngressLifecycle | undefined; + setSocketTerminator: (terminate: (() => void) | undefined) => void; + start: () => void; + stop: () => Promise; + waitForIdle: () => Promise; +}; + +type FeishuLifecycleSource = { + lifecycle?: FeishuIngressLifecycle; + replayClaim?: ChannelReplayClaimHandle; +}; + +export class FeishuIngressPermanentError extends Error { + constructor( + readonly reason: "authentication-failed" | "invalid-event", + message: string, + options?: ErrorOptions, + ) { + super(message, options); + this.name = "FeishuIngressPermanentError"; + } +} + +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value); +} + +function readString(value: unknown): string | null { + return typeof value === "string" && value.trim() ? value.trim() : null; +} + +function parseRawEnvelope(rawEnvelope: string): Record { + let parsed: unknown; + try { + parsed = JSON.parse(rawEnvelope); + } catch (error) { + throw new FeishuIngressPermanentError( + "invalid-event", + "Feishu ingress envelope contains invalid JSON.", + { cause: error }, + ); + } + if (!isRecord(parsed)) { + throw new FeishuIngressPermanentError( + "invalid-event", + "Feishu ingress envelope must be a JSON object.", + ); + } + return parsed; +} + +function decryptEnvelope( + envelope: Record, + encryptKey: string | undefined, +): Record { + const encrypted = readString(envelope.encrypt); + if (!encrypted) { + return envelope; + } + const key = encryptKey?.trim(); + if (!key) { + throw new FeishuIngressPermanentError( + "authentication-failed", + "Feishu encrypted ingress envelope has no configured encrypt key.", + ); + } + try { + return parseRawEnvelope(new Lark.AESCipher(key).decrypt(encrypted)); + } catch (error) { + if (error instanceof FeishuIngressPermanentError) { + throw error; + } + throw new FeishuIngressPermanentError( + "authentication-failed", + "Feishu ingress envelope decryption failed.", + { cause: error }, + ); + } +} + +function inspectFeishuIngressEnvelope( + rawEnvelope: string, + encryptKey: string | undefined, + allowInvalidLane = false, +): FeishuIngressFacts | null { + const envelope = decryptEnvelope(parseRawEnvelope(rawEnvelope), encryptKey); + const nestedHeader = isRecord(envelope.header) ? envelope.header : null; + const nestedEvent = isRecord(envelope.event) ? envelope.event : null; + const eventType = readString(nestedHeader?.event_type) ?? readString(envelope.event_type); + if (!eventType || !FEISHU_DURABLE_EVENT_TYPES.has(eventType)) { + return null; + } + // Lark v2 carries delivery identity in header.event_id. The flattened shape + // is also accepted because EventDispatcher hands that exact shape to handlers. + const eventId = readString(nestedHeader?.event_id) ?? readString(envelope.event_id); + if (!eventId) { + throw new FeishuIngressPermanentError( + "invalid-event", + `Feishu ${eventType} envelope is missing header.event_id.`, + ); + } + const event = nestedEvent ?? envelope; + if (eventType === "im.message.receive_v1") { + const message = isRecord(event.message) ? event.message : null; + const chatId = readString(message?.chat_id); + if (!chatId) { + if (allowInvalidLane) { + return { eventId, eventType, laneKey: `invalid:${eventType}:${eventId}` }; + } + throw new FeishuIngressPermanentError( + "invalid-event", + "Feishu message ingress envelope is missing event.message.chat_id.", + ); + } + return { eventId, eventType, laneKey: `chat:${chatId}` }; + } + const noticeMeta = isRecord(event.notice_meta) ? event.notice_meta : null; + const fileType = readString(noticeMeta?.file_type); + const documentId = readString(noticeMeta?.file_token); + if (!fileType || !documentId) { + if (allowInvalidLane) { + return { eventId, eventType, laneKey: `invalid:${eventType}:${eventId}` }; + } + throw new FeishuIngressPermanentError( + "invalid-event", + "Feishu comment ingress envelope is missing its document identity.", + ); + } + return { eventId, eventType, laneKey: `comment-doc:${fileType}:${documentId}` }; +} + +function resolveFeishuIngressNonRetryableFailure(error: unknown) { + if (error instanceof FeishuIngressPermanentError) { + return { reason: error.reason, message: error.message }; + } + for (const candidate of collectErrorGraphCandidates(error, (current) => [ + current.cause, + current.response, + ])) { + const status = isRecord(candidate) ? candidate.status : undefined; + if (status === 401 || status === 403) { + return { reason: "authentication-failed", message: formatErrorMessage(error) }; + } + } + return null; +} + +/** Fan one merged Feishu turn's adoption across every transport and logical claim. */ +export function buildFeishuFlushIngressLifecycle( + sources: readonly FeishuLifecycleSource[], + options?: { onReplayCommitError?: (error: unknown) => void }, +): { + lifecycle: FeishuIngressLifecycle | undefined; + settle: () => Promise; +} { + const durableSources = sources.filter( + ( + source, + ): source is Required> & + Pick => source.lifecycle !== undefined, + ); + const lifecycles = durableSources.map((source) => source.lifecycle); + const replayClaims = durableSources + .map((source) => source.replayClaim) + .filter((claim) => claim !== undefined); + const [firstLifecycle] = lifecycles; + if (!firstLifecycle) { + return { lifecycle: undefined, settle: async () => {} }; + } + let handedOff = false; + let terminal: "adopted" | "abandoned" | undefined; + let adopting: Promise | undefined; + let abandoning: Promise | undefined; + const releaseReplayClaims = () => { + for (const claim of replayClaims) { + claim.release({ error: new Error("feishu-ingress-not-adopted") }); + } + }; + const runAbandon = async () => { + if (terminal) { + return; + } + releaseReplayClaims(); + await Promise.all(lifecycles.map(async (lifecycle) => await lifecycle.onAbandoned())); + terminal = "abandoned"; + }; + const ensureAbandoned = async () => { + if (terminal) { + return; + } + const activeAbandonment = abandoning ?? runAbandon(); + abandoning = activeAbandonment; + try { + await activeAbandonment; + } finally { + if (abandoning === activeAbandonment && terminal !== "abandoned") { + abandoning = undefined; + } + } + }; + const abandonAll = async () => { + if (terminal) { + return; + } + if (adopting) { + await adopting.catch(() => undefined); + if (terminal) { + return; + } + } + await ensureAbandoned(); + }; + const adoptAll = async () => { + if (terminal) { + return; + } + if (abandoning) { + await abandoning.catch(() => undefined); + if (terminal) { + return; + } + } + const activeAdoption = + adopting ?? + (async () => { + try { + for (const lifecycle of lifecycles) { + await lifecycle.onAdopted(); + } + terminal = "adopted"; + // Queue adoption is authoritative. Logical twin guards commit only + // afterward: partial best-effort guard writes may admit a duplicate, + // but can never split or suppress recovery of an unadopted turn. + const results = await Promise.allSettled( + replayClaims.map(async (claim) => claim.commit()), + ); + for (const result of results) { + if (result.status === "rejected") { + try { + options?.onReplayCommitError?.(result.reason); + } catch { + // Reporting cannot undo an already adopted durable turn. + } + } + } + } catch (error) { + await ensureAbandoned().catch(() => undefined); + throw error; + } + })(); + adopting = activeAdoption; + try { + await activeAdoption; + } finally { + if (adopting === activeAdoption && terminal !== "adopted") { + adopting = undefined; + } + } + }; + return { + lifecycle: { + abortSignal: + lifecycles.length === 1 + ? firstLifecycle.abortSignal + : AbortSignal.any(lifecycles.map((lifecycle) => lifecycle.abortSignal)), + onAdopted: async () => { + handedOff = true; + await adoptAll(); + }, + onDeferred: () => { + handedOff = true; + for (const lifecycle of lifecycles) { + lifecycle.onDeferred(); + } + }, + onAdoptionFinalizing: () => { + for (const lifecycle of lifecycles) { + lifecycle.onAdoptionFinalizing(); + } + }, + onAbandoned: async () => { + handedOff = true; + await abandonAll(); + }, + }, + // A gated/no-turn envelope is terminal for transport replay, but its + // logical claim releases so a different event_id twin can run the gate. + settle: async () => { + if (handedOff) { + return; + } + handedOff = true; + releaseReplayClaims(); + try { + for (const lifecycle of lifecycles) { + lifecycle.onAdoptionFinalizing(); + } + for (const lifecycle of lifecycles) { + await lifecycle.onAdopted(); + } + terminal = "adopted"; + } catch (error) { + await ensureAbandoned().catch(() => undefined); + throw error; + } + }, + }; +} + +export function createFeishuDurableIngress(options: FeishuIngressOptions): FeishuDurableIngress { + let queue = options.queue; + let drain: ChannelIngressDrain | undefined; + let running = false; + let requested = false; + let pumping: Promise | undefined; + let pollTimer: ReturnType | undefined; + let admissionTail: Promise = Promise.resolve(); + let socketTerminator: (() => void) | undefined; + let lastPrunedAt = 0; + const activeLifecycles = new Map(); + const activeDeliveries = new Set>(); + const deferredClaims = new Map>(); + + const getQueue = (): ChannelIngressQueue => { + queue ??= getFeishuRuntime().state.openChannelIngressQueue({ + accountId: options.accountId, + }); + return queue; + }; + + const getDrain = (): ChannelIngressDrain => { + drain ??= createChannelIngressDrain({ + queue: getQueue(), + adoptionStallTimeoutMs: options.adoptionStallTimeoutMs ?? DEFAULT_INGRESS_ADOPTION_STALL_MS, + retryPolicy: { + maxAttempts: DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS, + deadLetterMinAgeMs: DEFAULT_INGRESS_RETRY_DEAD_LETTER_MIN_AGE_MS, + }, + resolveNonRetryableFailure: resolveFeishuIngressNonRetryableFailure, + onLog: (message) => options.runtime.error?.(`feishu ingress: ${message}`), + dispatchClaimedEvent: async (claimed, lifecycle) => { + if (claimed.payload.version !== FEISHU_INGRESS_PAYLOAD_VERSION) { + throw new FeishuIngressPermanentError( + "invalid-event", + `Feishu ingress row ${claimed.id} has an unsupported version.`, + ); + } + const facts = inspectFeishuIngressEnvelope(claimed.payload.rawEnvelope, options.encryptKey); + if (!facts || facts.eventId !== claimed.id) { + throw new FeishuIngressPermanentError( + "invalid-event", + `Feishu ingress row ${claimed.id} has invalid delivery identity.`, + ); + } + if (lifecycle.abortSignal.aborted) { + throw lifecycle.abortSignal.reason; + } + let resolveDeferred!: () => void; + const deferred = new Promise((resolve) => { + resolveDeferred = resolve; + }); + let deferredSettled = false; + const settleDeferred = () => { + if (deferredSettled) { + return; + } + deferredSettled = true; + if (deferredClaims.get(claimed.id) === deferred) { + deferredClaims.delete(claimed.id); + } + resolveDeferred(); + }; + const abandonHandlers = new Set<() => void | Promise>(); + const wrappedLifecycle: FeishuIngressLifecycle = { + ...lifecycle, + onAdopted: async () => { + try { + await lifecycle.onAdopted(); + } finally { + settleDeferred(); + } + }, + onAbandoned: async () => { + try { + await Promise.allSettled( + [...abandonHandlers].map(async (handler) => await handler()), + ); + await lifecycle.onAbandoned(); + } finally { + settleDeferred(); + } + }, + registerAbandonHandler: (handler) => { + abandonHandlers.add(handler); + return () => abandonHandlers.delete(handler); + }, + }; + activeLifecycles.set(claimed.id, wrappedLifecycle); + const delivery = options.dispatcher.invoke(parseRawEnvelope(claimed.payload.rawEnvelope), { + needCheck: false, + }); + activeDeliveries.add(delivery); + try { + const result = await delivery; + if (isRecord(result) && result.kind === "deferred" && !deferredSettled) { + deferredClaims.set(claimed.id, deferred); + } + return result; + } finally { + if (activeLifecycles.get(claimed.id) === wrappedLifecycle) { + activeLifecycles.delete(claimed.id); + } + activeDeliveries.delete(delivery); + } + }, + }); + return drain; + }; + + const pruneIfDue = async () => { + const now = Date.now(); + if (now - lastPrunedAt < FEISHU_INGRESS_PRUNE_INTERVAL_MS) { + return; + } + await getQueue().prune({ + completedTtlMs: FEISHU_INGRESS_COMPLETED_TTL_MS, + completedMaxEntries: FEISHU_INGRESS_COMPLETED_MAX_ENTRIES, + failedTtlMs: FEISHU_INGRESS_FAILED_TTL_MS, + failedMaxEntries: FEISHU_INGRESS_FAILED_MAX_ENTRIES, + now, + }); + lastPrunedAt = now; + }; + + const runPump = async () => { + try { + for (;;) { + requested = false; + await pruneIfDue(); + if (!running) { + break; + } + const activeDrain = getDrain(); + const { started } = await activeDrain.drainOnce(); + await activeDrain.waitForIdle(); + if (!running || (!requested && started === 0)) { + break; + } + } + } catch (error) { + options.runtime.error?.(`feishu ingress drain failed: ${formatErrorMessage(error)}`); + } finally { + pumping = undefined; + if (running && requested) { + requestDrain(); + } + } + }; + + const requestDrain = () => { + requested = true; + if (!running || pumping) { + return; + } + pumping = runPump(); + }; + + const admitOnce = async (rawEnvelope: string, facts: FeishuIngressFacts) => { + const receivedAt = Date.now(); + let lastError: unknown; + for (const delayMs of [0, 100, 300]) { + if (delayMs > 0) { + await new Promise((resolve) => { + setTimeout(resolve, delayMs); + }); + } + try { + await getQueue().enqueue( + facts.eventId, + { version: FEISHU_INGRESS_PAYLOAD_VERSION, receivedAt, rawEnvelope }, + { receivedAt, laneKey: facts.laneKey }, + ); + requestDrain(); + return; + } catch (error) { + lastError = error; + } + } + socketTerminator?.(); + throw lastError; + }; + + const invoke: Lark.EventDispatcher["invoke"] = async (data, params) => { + let rawEnvelope: string; + try { + const serialized = JSON.stringify(data); + if (serialized === undefined) { + throw new TypeError("Feishu ingress envelope has no JSON representation."); + } + rawEnvelope = serialized; + } catch (error) { + throw new FeishuIngressPermanentError( + "invalid-event", + "Feishu ingress envelope is not serializable.", + { cause: error }, + ); + } + // Webhook transport verifies the raw-body signature before invoke; the + // Lark WS client invokes from its authenticated socket with needCheck=false. + // Keep malformed recognized deliveries durable when their stable event id + // exists. Claim-side validation then dead-letters them without retry. + const facts = inspectFeishuIngressEnvelope(rawEnvelope, options.encryptKey, true); + if (!facts) { + return await options.dispatcher.invoke(data, params); + } + const admission = admissionTail.then(async () => await admitOnce(rawEnvelope, facts)); + admissionTail = admission.catch(() => undefined); + await admission; + return undefined; + }; + + return { + invoke, + resolveLifecycle: (data) => { + const eventId = isRecord(data) ? readString(data.event_id) : null; + return eventId ? activeLifecycles.get(eventId) : undefined; + }, + setSocketTerminator: (terminate) => { + socketTerminator = terminate; + }, + start: () => { + if (running) { + return; + } + running = true; + pollTimer = setInterval( + requestDrain, + options.pollIntervalMs ?? FEISHU_INGRESS_POLL_INTERVAL_MS, + ); + pollTimer.unref?.(); + requestDrain(); + }, + stop: async () => { + running = false; + if (pollTimer) { + clearInterval(pollTimer); + pollTimer = undefined; + } + await admissionTail; + drain?.dispose(); + await pumping; + drain?.dispose(); + await Promise.allSettled(activeDeliveries); + await Promise.allSettled(deferredClaims.values()); + await drain?.waitForIdle(); + activeLifecycles.clear(); + socketTerminator = undefined; + }, + waitForIdle: async () => { + for (;;) { + const activePump = pumping; + if (!activePump) { + break; + } + await activePump; + } + await drain?.waitForIdle(); + }, + }; +} diff --git a/extensions/feishu/src/lifecycle.test-support.ts b/extensions/feishu/src/lifecycle.test-support.ts index a355a2b9133f..380c3035b99c 100644 --- a/extensions/feishu/src/lifecycle.test-support.ts +++ b/extensions/feishu/src/lifecycle.test-support.ts @@ -33,6 +33,11 @@ type DispatchReplyFromConfigMock = Mock< (params: { ctx: DispatchReplyContext; dispatcher: DispatchReplyDispatcher; + replyOptions?: { + turnAdoptionLifecycle?: { + onAdopted: () => void | Promise; + }; + }; }) => Promise<{ queuedFinal: boolean; counts: DispatchReplyCounts }> >; type WithReplyDispatcherMock = Mock< diff --git a/extensions/feishu/src/monitor.account.ts b/extensions/feishu/src/monitor.account.ts index 56b563b6a724..b969682ca5cf 100644 --- a/extensions/feishu/src/monitor.account.ts +++ b/extensions/feishu/src/monitor.account.ts @@ -13,6 +13,7 @@ import { handleFeishuCardAction, type FeishuCardActionEvent } from "./card-actio import { createEventDispatcher } from "./client.js"; import { isRecord, readString } from "./comment-shared.js"; import { hasProcessedFeishuMessage, warmupDedupFromPluginState } from "./dedup.js"; +import { createFeishuDurableIngress, type FeishuIngressLifecycle } from "./feishu-ingress.js"; import { applyBotIdentityState, startBotIdentityRecovery } from "./monitor.bot-identity.js"; import { createFeishuBotMenuHandler } from "./monitor.bot-menu-handler.js"; import { createFeishuDriveCommentNoticeHandler } from "./monitor.comment-notice-handler.js"; @@ -177,6 +178,7 @@ type RegisterEventHandlersContext = { * liveness is published by the transport layer. */ statusSink?: FeishuStatusSink; + resolveIngressLifecycle?: (data: unknown) => FeishuIngressLifecycle | undefined; }; function parseFeishuBotAddedEventPayload(value: unknown): FeishuBotAddedEvent | null { @@ -311,6 +313,7 @@ function registerEventHandlers( getBotOpenId: (id) => botOpenIds.get(id), getBotName: (id) => botNames.get(id), resolveSequentialKey: getFeishuSequentialKey, + resolveIngressLifecycle: context.resolveIngressLifecycle, ...(context.statusSink ? { statusSink: context.statusSink } : {}), }), "im.message.message_read_v1": async () => { @@ -347,6 +350,7 @@ function registerEventHandlers( runtime, fireAndForget, abortSignal, + resolveIngressLifecycle: context.resolveIngressLifecycle, }), "vc.bot.meeting_invited_v1": createFeishuVcMeetingInvitedHandler({ cfg, @@ -507,6 +511,22 @@ export async function monitorSingleAccount(params: MonitorSingleAccountParams): let threadBindingManager: ReturnType | null | undefined; try { const eventDispatcher = createEventDispatcher(account); + // Minimal dispatcher doubles exercise handlers directly and intentionally + // omit invoke; production SDK dispatchers always provide the receive seam. + const durableIngress = + typeof (eventDispatcher as { invoke?: unknown }).invoke === "function" + ? createFeishuDurableIngress({ + accountId, + dispatcher: eventDispatcher, + ...(account.encryptKey ? { encryptKey: account.encryptKey } : {}), + runtime: runtime ?? {}, + }) + : undefined; + const durableEventDispatcher = durableIngress + ? (Object.assign(Object.create(eventDispatcher), { + invoke: durableIngress.invoke, + }) as Lark.EventDispatcher) + : eventDispatcher; const chatHistories = new Map(); threadBindingManager = createFeishuThreadBindingManager({ accountId, cfg }); const channelRuntime = params.channelRuntime?.inbound @@ -522,27 +542,34 @@ export async function monitorSingleAccount(params: MonitorSingleAccountParams): fireAndForget: params.fireAndForget ?? true, vcAutoJoin: account.config.vcAutoJoin === true, abortSignal, + ...(durableIngress ? { resolveIngressLifecycle: durableIngress.resolveLifecycle } : {}), ...(params.statusSink ? { statusSink: params.statusSink } : {}), }); - if (connectionMode === "webhook") { - return await monitorWebhook({ + durableIngress?.start(); + try { + if (connectionMode === "webhook") { + return await monitorWebhook({ + account, + accountId, + runtime, + abortSignal, + eventDispatcher: durableEventDispatcher, + ...(params.statusSink ? { statusSink: params.statusSink } : {}), + }); + } + return await monitorWebSocket({ account, accountId, runtime, abortSignal, - eventDispatcher, + eventDispatcher: durableEventDispatcher, + ...(durableIngress ? { setSocketTerminator: durableIngress.setSocketTerminator } : {}), ...(params.statusSink ? { statusSink: params.statusSink } : {}), }); + } finally { + await durableIngress?.stop(); } - return await monitorWebSocket({ - account, - accountId, - runtime, - abortSignal, - eventDispatcher, - ...(params.statusSink ? { statusSink: params.statusSink } : {}), - }); } finally { threadBindingManager?.stop(); } diff --git a/extensions/feishu/src/monitor.broadcast.reply-once.lifecycle.test-support.ts b/extensions/feishu/src/monitor.broadcast.reply-once.lifecycle.test-support.ts index bad2a9005bdb..73a3ba548775 100644 --- a/extensions/feishu/src/monitor.broadcast.reply-once.lifecycle.test-support.ts +++ b/extensions/feishu/src/monitor.broadcast.reply-once.lifecycle.test-support.ts @@ -229,13 +229,16 @@ describe("Feishu broadcast reply-once lifecycle", () => { text: "hello broadcast", }); - dispatchReplyFromConfigMock.mockImplementationOnce(async ({ ctx, dispatcher }) => { - if (typeof ctx?.SessionKey === "string" && ctx.SessionKey.includes("agent:susan:")) { - return { queuedFinal: false, counts: { final: 0 } }; - } - await dispatcher.sendFinalReply({ text: "broadcast reply once" }); - throw new Error("post-send failure"); - }); + dispatchReplyFromConfigMock.mockImplementationOnce( + async ({ ctx, dispatcher, replyOptions }) => { + await replyOptions?.turnAdoptionLifecycle?.onAdopted(); + if (typeof ctx?.SessionKey === "string" && ctx.SessionKey.includes("agent:susan:")) { + return { queuedFinal: false, counts: { final: 0 } }; + } + await dispatcher.sendFinalReply({ text: "broadcast reply once" }); + throw new Error("post-send failure"); + }, + ); await runFeishuLifecycleSequence( [() => onMessageA(event), () => onMessageB(event)], diff --git a/extensions/feishu/src/monitor.comment-notice-handler.ts b/extensions/feishu/src/monitor.comment-notice-handler.ts index 8dbb95cc17ad..64bf85cfafb2 100644 --- a/extensions/feishu/src/monitor.comment-notice-handler.ts +++ b/extensions/feishu/src/monitor.comment-notice-handler.ts @@ -1,10 +1,13 @@ // Feishu plugin module implements monitor.comment notice handler behavior. import type { ClawdbotConfig, RuntimeEnv } from "../runtime-api.js"; import { handleFeishuCommentEvent } from "./comment-handler.js"; -import { claimUnprocessedFeishuMessage, type FeishuMessageProcessingClaim } from "./dedup.js"; +import { + buildFeishuFlushIngressLifecycle, + FeishuIngressPermanentError, + type FeishuIngressLifecycle, +} from "./feishu-ingress.js"; import { parseFeishuDriveCommentNoticeEventPayload } from "./monitor.comment.js"; import { botOpenIds } from "./monitor.state.js"; -import { isFeishuRetryableSyntheticEventError } from "./monitor.synthetic-error.js"; import { createSequentialQueue } from "./sequential-queue.js"; function buildCommentNoticeQueueKey(event: { @@ -25,6 +28,7 @@ export function createFeishuDriveCommentNoticeHandler(params: { fireAndForget?: boolean; getBotOpenId?: (accountId: string) => string | undefined; abortSignal?: AbortSignal; + resolveIngressLifecycle?: (data: unknown) => FeishuIngressLifecycle | undefined; }): (data: unknown) => Promise { const { cfg, accountId, runtime, fireAndForget, abortSignal } = params; const log = runtime?.log ?? console.log; @@ -41,62 +45,58 @@ export function createFeishuDriveCommentNoticeHandler(params: { } }; - return async (data: unknown) => { - await runFeishuHandler(async () => { - const event = parseFeishuDriveCommentNoticeEventPayload(data); - if (!event) { - error(`feishu[${accountId}]: ignoring malformed drive comment notice payload`); + const handleNotice = async ( + data: unknown, + turnAdoptionLifecycle?: FeishuIngressLifecycle, + ): Promise => { + const event = parseFeishuDriveCommentNoticeEventPayload(data); + if (!event) { + if (turnAdoptionLifecycle) { + throw new FeishuIngressPermanentError( + "invalid-event", + "Feishu durable comment event payload is malformed.", + ); + } + error(`feishu[${accountId}]: ignoring malformed drive comment notice payload`); + return; + } + log( + `feishu[${accountId}]: received drive comment notice ` + + `event=${event.event_id ?? "unknown"} ` + + `type=${event.notice_meta?.notice_type ?? "unknown"} ` + + `file=${event.notice_meta?.file_type ?? "unknown"}:${event.notice_meta?.file_token ?? "unknown"} ` + + `comment=${event.comment_id ?? "unknown"} ` + + `reply=${event.reply_id ?? "none"} ` + + `from=${event.notice_meta?.from_user_id?.open_id ?? "unknown"} ` + + `mentioned=${event.is_mentioned === true ? "yes" : "no"}`, + ); + await enqueue(buildCommentNoticeQueueKey(event), async () => { + if (turnAdoptionLifecycle?.abortSignal.aborted) { + await turnAdoptionLifecycle.onAbandoned(); return; } - const eventId = event.event_id?.trim(); - const syntheticMessageId = eventId ? `drive-comment:${eventId}` : undefined; - let processingClaim: FeishuMessageProcessingClaim | undefined; - if (syntheticMessageId) { - const claim = await claimUnprocessedFeishuMessage({ - messageId: syntheticMessageId, - namespace: accountId, - log, - }); - if (claim.kind === "duplicate") { - log(`feishu[${accountId}]: dropping duplicate comment event ${syntheticMessageId}`); - return; - } - if (claim.kind === "inflight") { - log(`feishu[${accountId}]: dropping in-flight comment event ${syntheticMessageId}`); - return; - } - processingClaim = claim.kind === "claimed" ? claim.handle : undefined; - } - log( - `feishu[${accountId}]: received drive comment notice ` + - `event=${event.event_id ?? "unknown"} ` + - `type=${event.notice_meta?.notice_type ?? "unknown"} ` + - `file=${event.notice_meta?.file_type ?? "unknown"}:${event.notice_meta?.file_token ?? "unknown"} ` + - `comment=${event.comment_id ?? "unknown"} ` + - `reply=${event.reply_id ?? "none"} ` + - `from=${event.notice_meta?.from_user_id?.open_id ?? "unknown"} ` + - `mentioned=${event.is_mentioned === true ? "yes" : "no"}`, - ); - try { - await enqueue(buildCommentNoticeQueueKey(event), async () => { - await handleFeishuCommentEvent({ - cfg, - accountId, - event, - botOpenId: getBotOpenId(accountId), - runtime, - abortSignal, - }); - }); - await processingClaim?.commit(); - } catch (err) { - if (isFeishuRetryableSyntheticEventError(err)) { - processingClaim?.release({ error: err }); - } else { - await processingClaim?.commit(); - } - throw err; - } + await handleFeishuCommentEvent({ + cfg, + accountId, + event, + botOpenId: getBotOpenId(accountId), + runtime, + abortSignal, + turnAdoptionLifecycle, + }); }); }; + + return async (data: unknown) => { + const ingressLifecycle = params.resolveIngressLifecycle?.(data); + if (!ingressLifecycle) { + await runFeishuHandler(async () => await handleNotice(data)); + return; + } + const { lifecycle, settle } = buildFeishuFlushIngressLifecycle([ + { lifecycle: ingressLifecycle }, + ]); + await handleNotice(data, lifecycle); + await settle(); + }; } diff --git a/extensions/feishu/src/monitor.comment.test.ts b/extensions/feishu/src/monitor.comment.test.ts index 25f267d808fd..def450f9d18f 100644 --- a/extensions/feishu/src/monitor.comment.test.ts +++ b/extensions/feishu/src/monitor.comment.test.ts @@ -2,7 +2,7 @@ import { createNonExitingRuntimeEnv } from "openclaw/plugin-sdk/plugin-test-runtime"; import { afterAll, afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import type { ClawdbotConfig } from "../runtime-api.js"; -import * as dedup from "./dedup.js"; +import type { FeishuIngressLifecycle } from "./feishu-ingress.js"; import { createFeishuDriveCommentNoticeHandler } from "./monitor.comment-notice-handler.js"; import { resolveDriveCommentEventTurn, @@ -872,20 +872,10 @@ describe("resolveDriveCommentEventTurn", () => { }); describe("drive.notice.comment_add_v1 monitor handler", () => { - let processingClaim: dedup.FeishuMessageProcessingClaim; beforeEach(() => { lastRuntime = createNonExitingRuntimeEnv(); handleFeishuCommentEventMock.mockClear(); createFeishuClientMock.mockReset().mockReturnValue(makeOpenApiClient({}) as never); - processingClaim = { - keys: ["drive-comment:test"], - commit: vi.fn(async () => true), - release: vi.fn(), - }; - vi.spyOn(dedup, "claimUnprocessedFeishuMessage").mockResolvedValue({ - kind: "claimed", - handle: processingClaim, - }); }); afterEach(() => { @@ -942,10 +932,6 @@ describe("drive.notice.comment_add_v1 monitor handler", () => { reply_id: "reply_2", }), ); - await vi.waitFor(() => { - expect(dedup.claimUnprocessedFeishuMessage).toHaveBeenCalledTimes(2); - }); - expect(handleFeishuCommentEventMock).toHaveBeenCalledTimes(1); resolveFirst?.(); @@ -969,48 +955,41 @@ describe("drive.notice.comment_add_v1 monitor handler", () => { expect(secondCall?.event?.event_id).toBe("evt_2"); }); - it("drops duplicate comment events before dispatch", async () => { - vi.spyOn(dedup, "claimUnprocessedFeishuMessage").mockResolvedValue({ kind: "duplicate" }); - const onComment = await setupCommentMonitorHandler(); - - await onComment(makeDriveCommentEvent()); - - expect(handleFeishuCommentEventMock).not.toHaveBeenCalled(); - }); - - it("records generic comment-handler failures so replay stays closed", async () => { - const onComment = await setupCommentMonitorHandler(); - handleFeishuCommentEventMock.mockRejectedValueOnce(new Error("post-send failure")); - - await onComment(makeDriveCommentEvent()); - - await vi.waitFor(() => { - expect(processingClaim.commit).toHaveBeenCalledTimes(1); - expect(processingClaim.release).not.toHaveBeenCalled(); - expect(lastRuntime?.error).toHaveBeenCalledWith( - "feishu[default]: error handling drive comment notice: Error: post-send failure", - ); - }); - }); - - it("releases comment replay without recording when failure is explicitly retryable", async () => { - const onComment = await setupCommentMonitorHandler(); - handleFeishuCommentEventMock.mockRejectedValueOnce( - Object.assign(new Error("retry me"), { - name: "FeishuRetryableSyntheticEventError", - }), + it("does not execute a queued durable comment after its claim aborts", async () => { + let resolveFirst!: () => void; + handleFeishuCommentEventMock.mockImplementationOnce( + () => + new Promise((resolve) => { + resolveFirst = resolve; + }), ); - - await onComment(makeDriveCommentEvent()); - - await vi.waitFor(() => { - expect(processingClaim.commit).not.toHaveBeenCalled(); - expect(processingClaim.release).toHaveBeenCalledWith({ - error: expect.objectContaining({ message: "retry me" }), - }); - expect(lastRuntime?.error).toHaveBeenCalledWith( - "feishu[default]: error handling drive comment notice: FeishuRetryableSyntheticEventError: retry me", - ); + const controller = new AbortController(); + const abandoned = vi.fn(async () => {}); + const lifecycle: FeishuIngressLifecycle = { + abortSignal: controller.signal, + onAdopted: vi.fn(async () => {}), + onDeferred: vi.fn(), + onAdoptionFinalizing: vi.fn(), + onAbandoned: abandoned, + }; + const onComment = createFeishuDriveCommentNoticeHandler({ + cfg: buildMonitorConfig(), + accountId: "default", + runtime: createNonExitingRuntimeEnv(), + fireAndForget: true, + getBotOpenId: () => "ou_bot", + resolveIngressLifecycle: (data) => + (data as { event_id?: string }).event_id === "evt_queued" ? lifecycle : undefined, }); + + await onComment(makeDriveCommentEvent({ event_id: "evt_blocking" })); + await vi.waitFor(() => expect(handleFeishuCommentEventMock).toHaveBeenCalledTimes(1)); + const queued = onComment(makeDriveCommentEvent({ event_id: "evt_queued" })); + controller.abort(new Error("adoption timeout")); + resolveFirst(); + await queued; + + expect(handleFeishuCommentEventMock).toHaveBeenCalledTimes(1); + expect(abandoned).toHaveBeenCalledTimes(1); }); }); diff --git a/extensions/feishu/src/monitor.message-handler.ingress.test.ts b/extensions/feishu/src/monitor.message-handler.ingress.test.ts new file mode 100644 index 000000000000..d08bd9dad333 --- /dev/null +++ b/extensions/feishu/src/monitor.message-handler.ingress.test.ts @@ -0,0 +1,296 @@ +import { createNonExitingRuntimeEnv } from "openclaw/plugin-sdk/plugin-test-runtime"; +// Feishu ingress tests cover debounce ownership and constituent claim settlement. +import { afterEach, describe, expect, it, vi } from "vitest"; +import type { ClawdbotConfig, PluginRuntime, RuntimeEnv } from "../runtime-api.js"; +import * as dedup from "./dedup.js"; +import type { FeishuMessageEvent } from "./event-types.js"; +import type { FeishuIngressLifecycle } from "./feishu-ingress.js"; +import { createFeishuMessageReceiveHandler } from "./monitor.message-handler.js"; + +type MessageReceiveHandlerContext = Parameters[0]; +type HandleMessageParams = Parameters[0]; +type DebounceEntry = Parameters< + Parameters[0]["onFlush"] +>[0][number]; + +function createTextEvent( + eventId: string, + messageId: string, + text: string, +): FeishuMessageEvent & { event_id: string } { + return { + event_id: eventId, + sender: { + sender_id: { open_id: "ou-user" }, + sender_type: "user", + }, + message: { + message_id: messageId, + chat_id: "oc-chat", + chat_type: "p2p", + message_type: "text", + content: JSON.stringify({ text }), + create_time: "1710000000000", + }, + }; +} + +function createLifecycle() { + const controller = new AbortController(); + const abandonHandlers = new Set<() => void | Promise>(); + const calls = { + adopted: vi.fn(async () => {}), + deferred: vi.fn(), + finalizing: vi.fn(), + abandoned: vi.fn(async () => {}), + }; + const lifecycle: FeishuIngressLifecycle = { + abortSignal: controller.signal, + onAdopted: calls.adopted, + onDeferred: calls.deferred, + onAdoptionFinalizing: calls.finalizing, + onAbandoned: async () => { + await Promise.all([...abandonHandlers].map(async (handler) => await handler())); + await calls.abandoned(); + }, + registerAbandonHandler: (handler) => { + abandonHandlers.add(handler); + return () => abandonHandlers.delete(handler); + }, + }; + return { calls, controller, lifecycle }; +} + +function createClaim(name: string): dedup.FeishuMessageProcessingClaim { + return { + keys: [name], + commit: vi.fn(async () => true), + release: vi.fn(), + }; +} + +function createHarness(params: { + lifecycles: ReadonlyMap; + claims: readonly dedup.FeishuMessageProcessingClaim[]; + adoptTurn: boolean; +}) { + let onFlush: ((entries: DebounceEntry[]) => Promise) | undefined; + let onError: ((err: unknown, entries: DebounceEntry[]) => void) | undefined; + const entries: DebounceEntry[] = []; + const runtimeError = vi.fn(); + const channelRuntime = { + commands: { isControlCommandMessage: () => false }, + debounce: { + resolveInboundDebounceMs: () => 25, + createInboundDebouncer: vi.fn( + (options: { + onFlush: (entries: DebounceEntry[]) => Promise; + onError: (err: unknown, entries: DebounceEntry[]) => void; + }) => { + onFlush = options.onFlush; + onError = options.onError; + return { + enqueue: async (entry: DebounceEntry) => { + entries.push(entry); + }, + }; + }, + ), + }, + } as unknown as PluginRuntime["channel"]; + const handleMessage = vi.fn(async (turn: HandleMessageParams) => { + if (params.adoptTurn) { + turn.turnAdoptionLifecycle?.onAdoptionFinalizing(); + await turn.turnAdoptionLifecycle?.onAdopted(); + } + }); + const claim = vi.spyOn(dedup, "claimUnprocessedFeishuMessage"); + for (const handle of params.claims) { + claim.mockResolvedValueOnce({ kind: "claimed", handle }); + } + const handler = createFeishuMessageReceiveHandler({ + cfg: {} as ClawdbotConfig, + channelRuntime, + accountId: "default", + runtime: { ...createNonExitingRuntimeEnv(), error: runtimeError } satisfies RuntimeEnv, + chatHistories: new Map(), + handleMessage, + resolveDebounceText: () => "hello", + hasProcessedMessage: vi.fn(async () => false), + getBotOpenId: () => "ou-bot", + resolveIngressLifecycle: (data) => { + const eventId = (data as { event_id?: string }).event_id; + return eventId ? params.lifecycles.get(eventId) : undefined; + }, + }); + return { + claim, + entries, + handler, + handleMessage, + flush: async () => { + if (!onFlush) { + throw new Error("debouncer flush callback missing"); + } + await onFlush(entries.splice(0)); + }, + failFlush: (err: unknown) => { + if (!onError) { + throw new Error("debouncer error callback missing"); + } + onError(err, entries); + }, + runtimeError, + }; +} + +afterEach(() => { + vi.restoreAllMocks(); +}); + +describe("Feishu durable ingress debounce lifecycle", () => { + it("returns deferred and fans merged adoption to every constituent claim", async () => { + const first = createLifecycle(); + const second = createLifecycle(); + const firstClaim = createClaim("first"); + const secondClaim = createClaim("second"); + const harness = createHarness({ + lifecycles: new Map([ + ["evt-a", first.lifecycle], + ["evt-b", second.lifecycle], + ]), + claims: [firstClaim, secondClaim], + adoptTurn: true, + }); + + await expect(harness.handler(createTextEvent("evt-a", "om-a", "alpha"))).resolves.toEqual({ + kind: "deferred", + }); + await expect(harness.handler(createTextEvent("evt-b", "om-b", "beta"))).resolves.toEqual({ + kind: "deferred", + }); + await harness.flush(); + + expect(harness.handleMessage).toHaveBeenCalledTimes(1); + expect(firstClaim.commit).toHaveBeenCalledTimes(1); + expect(secondClaim.commit).toHaveBeenCalledTimes(1); + expect(first.calls.finalizing).toHaveBeenCalledTimes(1); + expect(second.calls.finalizing).toHaveBeenCalledTimes(1); + expect(first.calls.adopted).toHaveBeenCalledTimes(1); + expect(second.calls.adopted).toHaveBeenCalledTimes(1); + }); + + it("completes gated no-dispatch transport claims and releases the logical guard", async () => { + const transport = createLifecycle(); + const logicalClaim = createClaim("gated"); + const harness = createHarness({ + lifecycles: new Map([["evt-gated", transport.lifecycle]]), + claims: [logicalClaim], + adoptTurn: false, + }); + + await harness.handler(createTextEvent("evt-gated", "om-gated", "gated")); + await harness.flush(); + + expect(logicalClaim.commit).not.toHaveBeenCalled(); + expect(logicalClaim.release).toHaveBeenCalledTimes(1); + expect(transport.calls.adopted).toHaveBeenCalledTimes(1); + expect(transport.calls.abandoned).not.toHaveBeenCalled(); + }); + + it("completes the transport claim when the permanent logical guard suppresses a twin", async () => { + const transport = createLifecycle(); + const harness = createHarness({ + lifecycles: new Map([["evt-twin", transport.lifecycle]]), + claims: [], + adoptTurn: false, + }); + harness.claim.mockResolvedValueOnce({ kind: "duplicate" }); + + await expect(harness.handler(createTextEvent("evt-twin", "om-twin", "twin"))).resolves.toBe( + undefined, + ); + + expect(harness.entries).toHaveLength(0); + expect(transport.calls.finalizing).toHaveBeenCalledTimes(1); + expect(transport.calls.adopted).toHaveBeenCalledTimes(1); + expect(transport.calls.abandoned).not.toHaveBeenCalled(); + }); + + it("releases a deferred logical claim when the drain abandons before debounce flush", async () => { + const transport = createLifecycle(); + const logicalClaim = createClaim("pre-flush-abandonment"); + const harness = createHarness({ + lifecycles: new Map([["evt-abandoned", transport.lifecycle]]), + claims: [logicalClaim], + adoptTurn: false, + }); + + await expect( + harness.handler(createTextEvent("evt-abandoned", "om-abandoned", "queued")), + ).resolves.toEqual({ kind: "deferred" }); + await transport.lifecycle.onAbandoned(); + await harness.flush(); + + expect(logicalClaim.release).toHaveBeenCalledTimes(1); + expect(transport.calls.abandoned).toHaveBeenCalledTimes(1); + expect(harness.handleMessage).not.toHaveBeenCalled(); + }); + + it("reports rejected durable abandonment after a debounce flush error", async () => { + const transport = createLifecycle(); + transport.calls.abandoned.mockRejectedValueOnce(new Error("state store unavailable")); + const logicalClaim = createClaim("flush-error"); + const harness = createHarness({ + lifecycles: new Map([["evt-flush-error", transport.lifecycle]]), + claims: [logicalClaim], + adoptTurn: false, + }); + + await harness.handler(createTextEvent("evt-flush-error", "om-flush-error", "queued")); + harness.failFlush(new Error("flush failed")); + + await vi.waitFor(() => { + expect(harness.runtimeError).toHaveBeenCalledWith( + expect.stringContaining("failed to abandon durable ingress after debounce error"), + ); + }); + expect(logicalClaim.release).toHaveBeenCalled(); + }); + + it("does not dispatch a queued turn after its ingress claim aborts", async () => { + const first = createLifecycle(); + const second = createLifecycle(); + let finishFirst!: () => void; + const firstGate = new Promise((resolve) => { + finishFirst = resolve; + }); + const harness = createHarness({ + lifecycles: new Map([ + ["evt-first", first.lifecycle], + ["evt-second", second.lifecycle], + ]), + claims: [createClaim("first-queued"), createClaim("second-queued")], + adoptTurn: true, + }); + harness.handleMessage.mockImplementationOnce(async (turn) => { + await firstGate; + turn.turnAdoptionLifecycle?.onAdoptionFinalizing(); + await turn.turnAdoptionLifecycle?.onAdopted(); + }); + + await harness.handler(createTextEvent("evt-first", "om-first", "first")); + const firstFlush = harness.flush(); + await vi.waitFor(() => expect(harness.handleMessage).toHaveBeenCalledTimes(1)); + await harness.handler(createTextEvent("evt-second", "om-second", "second")); + const secondFlush = harness.flush(); + await Promise.resolve(); + second.controller.abort(new Error("adoption timeout")); + await second.lifecycle.onAbandoned(); + finishFirst(); + await Promise.all([firstFlush, secondFlush]); + + expect(harness.handleMessage).toHaveBeenCalledTimes(1); + expect(second.calls.adopted).not.toHaveBeenCalled(); + }); +}); diff --git a/extensions/feishu/src/monitor.message-handler.ts b/extensions/feishu/src/monitor.message-handler.ts index fab556c0dbd2..33236a08edfe 100644 --- a/extensions/feishu/src/monitor.message-handler.ts +++ b/extensions/feishu/src/monitor.message-handler.ts @@ -4,6 +4,11 @@ import type { ClawdbotConfig, HistoryEntry, PluginRuntime, RuntimeEnv } from ".. import { claimUnprocessedFeishuMessage, type FeishuMessageProcessingClaim } from "./dedup.js"; import { resolveFeishuMessageDedupeKey } from "./dedupe-key.js"; import type { FeishuMessageEvent } from "./event-types.js"; +import { + buildFeishuFlushIngressLifecycle, + FeishuIngressPermanentError, + type FeishuIngressLifecycle, +} from "./feishu-ingress.js"; import { isMentionForwardRequest } from "./mention.js"; import { createSequentialQueue } from "./sequential-queue.js"; import type { FeishuChatType } from "./types.js"; @@ -26,6 +31,7 @@ type FeishuMessageReceiveHandlerContext = { accountId?: string; processingClaim?: FeishuMessageProcessingClaim; messageDedupeKey?: string; + turnAdoptionLifecycle?: FeishuIngressLifecycle; }) => Promise; resolveDebounceText: (params: { event: FeishuMessageEvent; @@ -51,6 +57,7 @@ type FeishuMessageReceiveHandlerContext = { * published by the transport layer. */ statusSink?: import("./monitor.js").FeishuStatusSink; + resolveIngressLifecycle?: (data: unknown) => FeishuIngressLifecycle | undefined; }; function normalizeFeishuChatType(value: unknown): FeishuChatType | undefined { @@ -108,6 +115,8 @@ function mergeFeishuDebounceMentions( type FeishuMessageDebounceEntry = { event: FeishuMessageEvent; processingClaim?: FeishuMessageProcessingClaim; + turnAdoptionLifecycle?: FeishuIngressLifecycle; + abandoned?: boolean; }; function dedupeFeishuDebounceEntriesByDedupeKey( @@ -172,7 +181,10 @@ export function createFeishuMessageReceiveHandler({ resolveSequentialKey = ({ accountId: accountIdLocal, event }) => `feishu:${accountIdLocal}:${event.message.chat_id?.trim() || "unknown"}`, statusSink, -}: FeishuMessageReceiveHandlerContext): (data: unknown) => Promise { + resolveIngressLifecycle, +}: FeishuMessageReceiveHandlerContext): ( + data: unknown, +) => Promise<{ kind: "deferred" } | { kind: "failed-retryable"; error: unknown } | void> { const inboundDebounceMs = channelRuntime.debounce.resolveInboundDebounceMs({ cfg, channel: "feishu", @@ -191,6 +203,7 @@ export function createFeishuMessageReceiveHandler({ event: FeishuMessageEvent, messageDedupeKey?: string, processingClaim?: FeishuMessageProcessingClaim, + turnAdoptionLifecycle?: FeishuIngressLifecycle, ) => { const sequentialKey = resolveSequentialKey({ accountId, @@ -198,8 +211,12 @@ export function createFeishuMessageReceiveHandler({ botOpenId: getBotOpenId(accountId), botName: getBotName(accountId), }); - const task = () => - handleMessage({ + const task = async () => { + if (turnAdoptionLifecycle?.abortSignal.aborted) { + await turnAdoptionLifecycle.onAbandoned(); + return; + } + await handleMessage({ cfg, event, botOpenId: getBotOpenId(accountId), @@ -210,7 +227,9 @@ export function createFeishuMessageReceiveHandler({ accountId, processingClaim, messageDedupeKey, + turnAdoptionLifecycle, }); + }; await enqueue(sequentialKey, task); }; @@ -273,68 +292,120 @@ export function createFeishuMessageReceiveHandler({ return Boolean(text) && !channelRuntime.commands.isControlCommandMessage(text, cfg); }, onFlush: async (entries) => { - const last = entries.at(-1); + const activeEntries = entries.filter((entry) => !entry.abandoned); + const last = activeEntries.at(-1); if (!last) { return; } - if (entries.length === 1) { - await dispatchFeishuMessage( - last.event, - resolveFeishuMessageDedupeKey(last.event), - last.processingClaim, - ); - return; - } - const dedupedEntries = dedupeFeishuDebounceEntriesByDedupeKey(entries); - const freshEntries: FeishuMessageDebounceEntry[] = []; - for (const entry of dedupedEntries) { - if ( - !(await hasProcessedMessage(resolveFeishuMessageDedupeKey(entry.event), accountId, log)) - ) { - freshEntries.push(entry); - } - } - const dispatchEntry = freshEntries.at(-1); - if (!dispatchEntry) { - return; - } - const dispatchDedupeKey = resolveFeishuMessageDedupeKey(dispatchEntry.event); - await recordSuppressedMessageIds(dedupedEntries, dispatchDedupeKey); - const combinedText = freshEntries - .map((entry) => resolveDebounceText(entry.event)) - .filter(Boolean) - .join("\n"); - const mergedMentions = resolveFeishuDebounceMentions({ - entries: freshEntries.map((entry) => entry.event), - botOpenId: getBotOpenId(accountId), - }); - await dispatchFeishuMessage( + const { lifecycle, settle } = buildFeishuFlushIngressLifecycle( + activeEntries.map((entry) => ({ + lifecycle: entry.turnAdoptionLifecycle, + replayClaim: entry.processingClaim, + })), { - ...dispatchEntry.event, - message: { - ...dispatchEntry.event.message, - ...(combinedText.trim() - ? { - message_type: "text", - content: JSON.stringify({ text: combinedText }), - } - : {}), - mentions: mergedMentions ?? dispatchEntry.event.message.mentions, - }, + onReplayCommitError: (err) => + error(`feishu[${accountId}]: failed to commit logical replay guard: ${String(err)}`), }, - dispatchDedupeKey, - dispatchEntry.processingClaim, ); + if (lifecycle?.abortSignal.aborted) { + await lifecycle.onAbandoned(); + return; + } + try { + if (activeEntries.length === 1) { + await dispatchFeishuMessage( + last.event, + resolveFeishuMessageDedupeKey(last.event), + last.processingClaim, + lifecycle, + ); + await settle(); + return; + } + const dedupedEntries = dedupeFeishuDebounceEntriesByDedupeKey(activeEntries); + const freshEntries: FeishuMessageDebounceEntry[] = []; + for (const entry of dedupedEntries) { + if ( + !(await hasProcessedMessage( + resolveFeishuMessageDedupeKey(entry.event), + accountId, + log, + )) + ) { + freshEntries.push(entry); + } + } + const dispatchEntry = freshEntries.at(-1); + if (!dispatchEntry) { + await settle(); + return; + } + const dispatchDedupeKey = resolveFeishuMessageDedupeKey(dispatchEntry.event); + if (!lifecycle) { + await recordSuppressedMessageIds(dedupedEntries, dispatchDedupeKey); + } + const combinedText = freshEntries + .map((entry) => resolveDebounceText(entry.event)) + .filter(Boolean) + .join("\n"); + const mergedMentions = resolveFeishuDebounceMentions({ + entries: freshEntries.map((entry) => entry.event), + botOpenId: getBotOpenId(accountId), + }); + await dispatchFeishuMessage( + { + ...dispatchEntry.event, + message: { + ...dispatchEntry.event.message, + ...(combinedText.trim() + ? { + message_type: "text", + content: JSON.stringify({ text: combinedText }), + } + : {}), + mentions: mergedMentions ?? dispatchEntry.event.message.mentions, + }, + }, + dispatchDedupeKey, + dispatchEntry.processingClaim, + lifecycle, + ); + await settle(); + } catch (err) { + await lifecycle?.onAbandoned(); + throw err; + } }, onError: (err, entries) => { for (const entry of entries) { entry.processingClaim?.release({ error: err }); + try { + void Promise.resolve(entry.turnAdoptionLifecycle?.onAbandoned()).catch( + (abandonError: unknown) => { + error( + `feishu[${accountId}]: failed to abandon durable ingress after debounce error: ${String(abandonError)}`, + ); + }, + ); + } catch (abandonError) { + error( + `feishu[${accountId}]: failed to abandon durable ingress after debounce error: ${String(abandonError)}`, + ); + } } error(`feishu[${accountId}]: inbound debounce flush failed: ${String(err)}`); }, }); return async (data) => { + const turnAdoptionLifecycle = resolveIngressLifecycle?.(data); + const completeSuppressedIngress = async () => { + if (!turnAdoptionLifecycle) { + return; + } + turnAdoptionLifecycle.onAdoptionFinalizing(); + await turnAdoptionLifecycle.onAdopted(); + }; // Publish message recency before dedupe/debounce; transport liveness is // owned by the WebSocket/Webhook lifecycle monitors. const inboundAt = Date.now(); @@ -344,8 +415,14 @@ export function createFeishuMessageReceiveHandler({ const event = parseFeishuMessageEventPayload(data); if (!event) { + if (turnAdoptionLifecycle) { + throw new FeishuIngressPermanentError( + "invalid-event", + "Feishu durable message event payload is malformed.", + ); + } error(`feishu[${accountId}]: ignoring malformed message event payload`); - return; + return undefined; } const messageId = event.message?.message_id?.trim(); const botOpenId = getBotOpenId(accountId)?.trim(); @@ -354,7 +431,8 @@ export function createFeishuMessageReceiveHandler({ // Feishu bot receive events identify their sender by open_id. Drop this // account's bot before it can consume a claim or debounce slot. log(`feishu[${accountId}]: dropping self-authored message ${messageId ?? "unknown"}`); - return; + await completeSuppressedIngress(); + return undefined; } const messageDedupeKey = resolveFeishuMessageDedupeKey(event); const claim = await claimUnprocessedFeishuMessage({ @@ -364,14 +442,36 @@ export function createFeishuMessageReceiveHandler({ }); if (claim.kind === "duplicate" || claim.kind === "inflight") { log(`feishu[${accountId}]: dropping ${claim.kind} event for message ${messageId}`); - return; + await completeSuppressedIngress(); + return undefined; + } + const debounceEntry: FeishuMessageDebounceEntry = { + event, + ...(claim.kind === "claimed" ? { processingClaim: claim.handle } : {}), + ...(turnAdoptionLifecycle ? { turnAdoptionLifecycle } : {}), + }; + if (claim.kind === "claimed" && turnAdoptionLifecycle) { + // The durable drain can abandon before the debounce timer flushes. Tie + // the logical claim and queued entry to that earlier lifecycle. + turnAdoptionLifecycle.registerAbandonHandler?.(() => { + debounceEntry.abandoned = true; + claim.handle.release({ error: new Error("feishu-ingress-abandoned-before-flush") }); + }); } const processMessage = async () => { - await inboundDebouncer.enqueue({ - event, - ...(claim.kind === "claimed" ? { processingClaim: claim.handle } : {}), - }); + await inboundDebouncer.enqueue(debounceEntry); }; + if (turnAdoptionLifecycle) { + try { + await processMessage(); + return { kind: "deferred" }; + } catch (err) { + if (claim.kind === "claimed") { + claim.handle.release({ error: err }); + } + return { kind: "failed-retryable", error: err }; + } + } if (fireAndForget) { void processMessage().catch((err: unknown) => { if (claim.kind === "claimed") { @@ -379,7 +479,7 @@ export function createFeishuMessageReceiveHandler({ } error(`feishu[${accountId}]: error handling message: ${String(err)}`); }); - return; + return undefined; } try { await processMessage(); @@ -389,5 +489,6 @@ export function createFeishuMessageReceiveHandler({ } error(`feishu[${accountId}]: error handling message: ${String(err)}`); } + return undefined; }; } diff --git a/extensions/feishu/src/monitor.reaction.test.ts b/extensions/feishu/src/monitor.reaction.test.ts index c02fe40e9393..9dfe72b20d53 100644 --- a/extensions/feishu/src/monitor.reaction.test.ts +++ b/extensions/feishu/src/monitor.reaction.test.ts @@ -248,7 +248,6 @@ function setDedupPassThroughMocks(): void { vi.spyOn(dedup, "claimUnprocessedFeishuMessage").mockResolvedValue( createClaimedFeishuDedupeResult(), ); - vi.spyOn(dedup, "recordProcessedFeishuMessage").mockResolvedValue(true); vi.spyOn(dedup, "hasProcessedFeishuMessage").mockResolvedValue(false); } @@ -611,7 +610,6 @@ describe("Feishu inbound debounce regressions", () => { vi.spyOn(dedup, "claimUnprocessedFeishuMessage").mockResolvedValue( createClaimedFeishuDedupeResult(), ); - vi.spyOn(dedup, "recordProcessedFeishuMessage").mockResolvedValue(true); vi.spyOn(dedup, "hasProcessedFeishuMessage").mockResolvedValue(false); const onMessage = await setupDebounceMonitor({ botName: "OpenClaw Bot" }); @@ -696,7 +694,6 @@ describe("Feishu inbound debounce regressions", () => { vi.spyOn(dedup, "claimUnprocessedFeishuMessage").mockResolvedValue( createClaimedFeishuDedupeResult(), ); - vi.spyOn(dedup, "recordProcessedFeishuMessage").mockResolvedValue(true); setStaleRetryMocks(); const onMessage = await setupDebounceMonitor(); diff --git a/extensions/feishu/src/monitor.transport.ts b/extensions/feishu/src/monitor.transport.ts index 65b9cfc1cf0c..e43a0707f241 100644 --- a/extensions/feishu/src/monitor.transport.ts +++ b/extensions/feishu/src/monitor.transport.ts @@ -33,6 +33,7 @@ type MonitorTransportParams = { runtime?: RuntimeEnv; abortSignal?: AbortSignal; eventDispatcher: Lark.EventDispatcher; + setSocketTerminator?: (terminate: (() => void) | undefined) => void; /** * Optional status sink for Feishu health tracking. Lifecycle callbacks * publish connected state; validated inbound webhook requests publish @@ -203,6 +204,7 @@ export async function monitorWebSocket({ runtime, abortSignal, eventDispatcher, + setSocketTerminator, statusSink, }: MonitorTransportParams): Promise { const log = runtime?.log ?? console.log; @@ -253,6 +255,7 @@ export async function monitorWebSocket({ onReconnected: publishWsConnected, onReconnecting: publishWsReconnecting, }); + setSocketTerminator?.(() => wsClient?.close({ force: true })); if (abortSignal?.aborted) { cleanupFeishuWsClient({ accountId, wsClient, error, clearIdentity: true }); break; @@ -265,10 +268,12 @@ export async function monitorWebSocket({ if (cycleEnd === "abort") { log(`feishu[${accountId}]: abort signal received, stopping`); cleanupFeishuWsClient({ accountId, wsClient, error, clearIdentity: true }); + setSocketTerminator?.(undefined); return; } cleanupFeishuWsClient({ accountId, wsClient, error, clearIdentity: false }); + setSocketTerminator?.(undefined); if (abortSignal?.aborted) { break; } @@ -292,6 +297,7 @@ export async function monitorWebSocket({ } } catch (err) { cleanupFeishuWsClient({ accountId, wsClient, error, clearIdentity: false }); + setSocketTerminator?.(undefined); if (abortSignal?.aborted) { break; } @@ -315,6 +321,7 @@ export async function monitorWebSocket({ } } cleanupFeishuWsClient({ accountId, wsClient: undefined, error, clearIdentity: true }); + setSocketTerminator?.(undefined); } export async function monitorWebhook({ diff --git a/extensions/feishu/src/monitor.vc-meeting-invited-handler.test.ts b/extensions/feishu/src/monitor.vc-meeting-invited-handler.test.ts index e49ce812396f..3464877ac90e 100644 --- a/extensions/feishu/src/monitor.vc-meeting-invited-handler.test.ts +++ b/extensions/feishu/src/monitor.vc-meeting-invited-handler.test.ts @@ -14,7 +14,6 @@ const createFeishuThreadBindingManagerMock = vi.hoisted(() => vi.fn(() => ({ sto const dedupMocks = vi.hoisted(() => ({ warmupDedupFromPluginState: vi.fn(async () => 0), hasProcessedFeishuMessage: vi.fn(async () => false), - recordProcessedFeishuMessage: vi.fn(async () => true), })); let handlers: Record Promise> = {}; @@ -46,7 +45,6 @@ vi.mock("./dedup.js", async () => { ...actual, warmupDedupFromPluginState: dedupMocks.warmupDedupFromPluginState, hasProcessedFeishuMessage: dedupMocks.hasProcessedFeishuMessage, - recordProcessedFeishuMessage: dedupMocks.recordProcessedFeishuMessage, }; }); @@ -126,7 +124,6 @@ describe("createFeishuVcMeetingInvitedHandler", () => { handleFeishuMessageMock.mockResolvedValue(undefined); dedupMocks.warmupDedupFromPluginState.mockResolvedValue(0); dedupMocks.hasProcessedFeishuMessage.mockResolvedValue(false); - dedupMocks.recordProcessedFeishuMessage.mockResolvedValue(true); }); it("ignores invitations unless VC auto-join is enabled", async () => { diff --git a/extensions/feishu/src/test-support/lifecycle-test-support.ts b/extensions/feishu/src/test-support/lifecycle-test-support.ts index 08b139b91370..384c2769dd9b 100644 --- a/extensions/feishu/src/test-support/lifecycle-test-support.ts +++ b/extensions/feishu/src/test-support/lifecycle-test-support.ts @@ -28,6 +28,11 @@ type FeishuDispatchReplyMock = Mock< (args: { ctx: FeishuDispatchReplyContext; dispatcher: FeishuDispatchReplyDispatcher; + replyOptions?: { + turnAdoptionLifecycle?: { + onAdopted: () => void | Promise; + }; + }; }) => Promise<{ queuedFinal: boolean; counts: FeishuDispatchReplyCounts }> >; type RuntimeReplyDispatcher = NonNullable< @@ -111,7 +116,12 @@ function installFeishuLifecycleRuntime(params: { reply: { resolveEnvelopeFormatOptions: vi.fn(() => ({})), formatAgentEnvelope: vi.fn((value: { body: string }) => value.body), - dispatchReplyWithBufferedBlockDispatcher: async ({ cfg, ctx, dispatcherOptions }) => { + dispatchReplyWithBufferedBlockDispatcher: async ({ + cfg, + ctx, + dispatcherOptions, + replyOptions, + }) => { // ReplyDispatcher enqueue methods are synchronous; settlement owns async delivery. const pendingDeliveries: Promise[] = []; const dispatcher: RuntimeReplyDispatcher = { @@ -137,6 +147,7 @@ function installFeishuLifecycleRuntime(params: { cfg, ctx: ctx as Parameters[0]["ctx"], dispatcher, + replyOptions, }), }); }, @@ -188,16 +199,19 @@ export function mockFeishuReplyOnceDispatch(params: { replyText: string; shouldSendFinalReply?: (ctx: unknown) => boolean; }) { - params.dispatchReplyFromConfigMock.mockImplementation(async ({ ctx, dispatcher }) => { - const shouldSendFinalReply = params.shouldSendFinalReply?.(ctx) ?? true; - if (shouldSendFinalReply && typeof dispatcher?.sendFinalReply === "function") { - await dispatcher.sendFinalReply({ text: params.replyText }); - } - return { - queuedFinal: false, - counts: { final: shouldSendFinalReply ? 1 : 0 }, - }; - }); + params.dispatchReplyFromConfigMock.mockImplementation( + async ({ ctx, dispatcher, replyOptions }) => { + await replyOptions?.turnAdoptionLifecycle?.onAdopted(); + const shouldSendFinalReply = params.shouldSendFinalReply?.(ctx) ?? true; + if (shouldSendFinalReply && typeof dispatcher?.sendFinalReply === "function") { + await dispatcher.sendFinalReply({ text: params.replyText }); + } + return { + queuedFinal: false, + counts: { final: shouldSendFinalReply ? 1 : 0 }, + }; + }, + ); } export function createFeishuLifecycleConfig(params: {