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
This commit is contained in:
Peter Steinberger
2026-07-18 18:49:38 +01:00
committed by GitHub
parent 897ac2d4b0
commit 736dddb96a
21 changed files with 2668 additions and 387 deletions

View File

@@ -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<void>;
onDispatchFailed: (error: unknown) => Promise<void>;
};
onLanePending: () => void;
onDispatchComplete: () => Promise<void>;
onDispatchFailed: (error: unknown) => Promise<void>;
} {
type LaneState = {
replayClaim?: ChannelReplayClaimHandle;
status: "pending" | "deferred" | "adopted" | "completed" | "failed" | "abandoned";
};
const lanes = new Set<LaneState>();
const failures: unknown[] = [];
const fallbackAbortSignal = new AbortController().signal;
let fanoutSettled = false;
let terminal: "adopted" | "abandoned" | undefined;
let adoption: Promise<void> | undefined;
let abandonment: Promise<void> | 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<string, unknown>).broadcast;

View File

@@ -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<string, unknown>).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<string, unknown>).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<void>((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<FeishuIngressLifecycle, "onAdopted" | "onDeferred" | "onAbandoned">
| 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"] },

View File

@@ -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<void> {
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<ReturnType<typeof claimUnprocessedFeishuMessage>> | 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<string, unknown>).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<string, unknown>).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<void>[] = [];
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. */

View File

@@ -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) }
: {}),
}),
},
});

View File

@@ -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;

View File

@@ -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" });

View File

@@ -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<boolean> }
| { 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<boolean> {
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,

View File

@@ -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).

View File

@@ -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<Parameters<typeof createFeishuDurableIngress>[0]["queue"]>;
type FeishuIngressPayload = Parameters<FeishuIngressQueue["enqueue"]>[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<typeof messageEnvelope>) {
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<typeof messageEnvelope>) => unknown = async () => undefined,
): Pick<Lark.EventDispatcher, "invoke"> {
return {
invoke: async (data) => await run(data as ReturnType<typeof messageEnvelope>),
};
}
function startIngress(params: {
queue: FeishuIngressQueue;
dispatcher: Pick<Lark.EventDispatcher, "invoke">;
}) {
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<T>(fn: (queue: FeishuIngressQueue, stateDir: string) => Promise<T>) {
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<FeishuIngressPayload>({
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<string, string> {
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<Lark.EventDispatcher, "invoke">,
run: (url: string) => Promise<void>,
) {
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<typeof messageEnvelope>) {
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<void>((resolve) => {
releaseAppend = resolve;
});
const enqueue = vi.fn(async (...args: Parameters<FeishuIngressQueue["enqueue"]>) => {
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<typeof messageEnvelope>) => {
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<typeof messageEnvelope>) => {
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<void>((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);
});
});

View File

@@ -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<void>;
onDeferred: () => void;
onAdoptionFinalizing: () => void;
onAbandoned: () => void | Promise<void>;
registerAbandonHandler?: (handler: () => void | Promise<void>) => () => void;
};
type FeishuIngressPayload = {
version: 1;
receivedAt: number;
rawEnvelope: string;
};
type FeishuIngressFacts = {
eventId: string;
eventType: string;
laneKey: string;
};
type FeishuIngressOptions = {
accountId: string;
dispatcher: Pick<Lark.EventDispatcher, "invoke">;
encryptKey?: string;
runtime: {
error?: (message: string) => void;
log?: (message: string) => void;
};
queue?: ChannelIngressQueue<FeishuIngressPayload>;
pollIntervalMs?: number;
adoptionStallTimeoutMs?: number;
};
type FeishuDurableIngress = {
invoke: Lark.EventDispatcher["invoke"];
resolveLifecycle: (data: unknown) => FeishuIngressLifecycle | undefined;
setSocketTerminator: (terminate: (() => void) | undefined) => void;
start: () => void;
stop: () => Promise<void>;
waitForIdle: () => Promise<void>;
};
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<string, unknown> {
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<string, unknown> {
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<string, unknown>,
encryptKey: string | undefined,
): Record<string, unknown> {
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<void>;
} {
const durableSources = sources.filter(
(
source,
): source is Required<Pick<FeishuLifecycleSource, "lifecycle">> &
Pick<FeishuLifecycleSource, "replayClaim"> => 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<void> | undefined;
let abandoning: Promise<void> | 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<void> | undefined;
let pollTimer: ReturnType<typeof setInterval> | undefined;
let admissionTail: Promise<void> = Promise.resolve();
let socketTerminator: (() => void) | undefined;
let lastPrunedAt = 0;
const activeLifecycles = new Map<string, FeishuIngressLifecycle>();
const activeDeliveries = new Set<Promise<unknown>>();
const deferredClaims = new Map<string, Promise<void>>();
const getQueue = (): ChannelIngressQueue<FeishuIngressPayload> => {
queue ??= getFeishuRuntime().state.openChannelIngressQueue<FeishuIngressPayload>({
accountId: options.accountId,
});
return queue;
};
const getDrain = (): ChannelIngressDrain => {
drain ??= createChannelIngressDrain<FeishuIngressPayload>({
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<void>((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<void>>();
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();
},
};
}

View File

@@ -33,6 +33,11 @@ type DispatchReplyFromConfigMock = Mock<
(params: {
ctx: DispatchReplyContext;
dispatcher: DispatchReplyDispatcher;
replyOptions?: {
turnAdoptionLifecycle?: {
onAdopted: () => void | Promise<void>;
};
};
}) => Promise<{ queuedFinal: boolean; counts: DispatchReplyCounts }>
>;
type WithReplyDispatcherMock = Mock<

View File

@@ -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<typeof createFeishuThreadBindingManager> | 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<string, HistoryEntry[]>();
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();
}

View File

@@ -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)],

View File

@@ -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<void> {
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<void> => {
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();
};
}

View File

@@ -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<void>((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);
});
});

View File

@@ -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<typeof createFeishuMessageReceiveHandler>[0];
type HandleMessageParams = Parameters<MessageReceiveHandlerContext["handleMessage"]>[0];
type DebounceEntry = Parameters<
Parameters<PluginRuntime["channel"]["debounce"]["createInboundDebouncer"]>[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<void>>();
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<string, FeishuIngressLifecycle>;
claims: readonly dedup.FeishuMessageProcessingClaim[];
adoptTurn: boolean;
}) {
let onFlush: ((entries: DebounceEntry[]) => Promise<void>) | 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<void>;
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<void>((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();
});
});

View File

@@ -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<void>;
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<void> {
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;
};
}

View File

@@ -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();

View File

@@ -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<void> {
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({

View File

@@ -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<string, (data: unknown) => Promise<void>> = {};
@@ -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 () => {

View File

@@ -28,6 +28,11 @@ type FeishuDispatchReplyMock = Mock<
(args: {
ctx: FeishuDispatchReplyContext;
dispatcher: FeishuDispatchReplyDispatcher;
replyOptions?: {
turnAdoptionLifecycle?: {
onAdopted: () => void | Promise<void>;
};
};
}) => 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<unknown>[] = [];
const dispatcher: RuntimeReplyDispatcher = {
@@ -137,6 +147,7 @@ function installFeishuLifecycleRuntime(params: {
cfg,
ctx: ctx as Parameters<typeof params.dispatchReplyFromConfig>[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: {