From c45f299cea4d347002ea3c2c656a77f57f8bd021 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Fri, 17 Jul 2026 11:38:54 +0100 Subject: [PATCH] fix(slack): adopt durable ingress across Bolt and relay (#109910) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(slack): adopt durable ingress drain across Bolt and relay with logical twin guard Slack events now enqueue their raw envelope durably before the transport ack on all three modes: the Bolt receiver wrap covers Socket Mode and HTTP (event_id = Events API event_id), and relay frames enqueue before the router ack (keyed by logical team:channel:ts since router delivery-id redelivery stability is undocumented). Dispatch, retry, dead-letter, and 30d/20k tombstones run through the core drain; claimed relay events retry until the relay source reattaches after a restart. The in-memory + persistent inbound delivery caches and the app-mention race machinery are deleted. One dedupe layer deliberately survives: Slack emits both a message and an app_mention event (distinct event_ids) for a single mention post, so a new claim-based dispatch guard keyed on logical (team, channel, ts) collapses the twins — commit at turn adoption, release on gated/abandoned dispatch so the surviving twin can re-run the same gate. 24h/20k retention mirrors the retired persistent cache. Slack monitor test state now scopes a fresh OPENCLAW_STATE_DIR per test; persisted guard keys otherwise dedupe unrelated fixture messages that reuse ts values. Five reaction tests in monitor.tool-result.test.ts are flaky on clean origin/main under full-suite load (verified 2-of-3 baseline runs); tracked separately. Autoreview: secret scanner false-positives on token-shaped test fixtures (three files); full manual review performed instead, which caught and fixed the twin-guard regression. Part of the #109657 fleet adoption program. * fix(slack): remove unused ingress exports * chore(slack): prune max-lines baseline * docs(plugin-sdk): refresh API baseline hash * test(slack): type dispatch mock argument * test(slack): use managed temp root --- config/max-lines-baseline.txt | 1 - .../.generated/plugin-sdk-api-baseline.sha256 | 4 +- extensions/slack/src/monitor.test-helpers.ts | 30 +- extensions/slack/src/monitor/context.ts | 36 -- .../slack/src/monitor/events/messages.test.ts | 43 ++ .../slack/src/monitor/events/messages.ts | 21 +- .../monitor/inbound-delivery-state.test.ts | 71 ---- .../src/monitor/inbound-delivery-state.ts | 75 ---- extensions/slack/src/monitor/ingress.test.ts | 288 +++++++++++++ extensions/slack/src/monitor/ingress.ts | 394 ++++++++++++++++++ .../src/monitor/message-dispatch-dedupe.ts | 90 ++++ .../message-handler.app-mention-race.test.ts | 381 ----------------- .../slack/src/monitor/message-handler.test.ts | 193 +++++---- .../slack/src/monitor/message-handler.ts | 320 +++++--------- .../dispatch.preview-fallback.test.ts | 17 + .../src/monitor/message-handler/dispatch.ts | 3 + .../monitor/message-handler/prepare.test.ts | 2 - .../src/monitor/message-handler/types.ts | 2 + .../slack/src/monitor/provider-support.ts | 6 +- .../src/monitor/provider.interop.test.ts | 31 ++ extensions/slack/src/monitor/provider.ts | 23 +- .../slack/src/monitor/relay-source.test.ts | 46 +- extensions/slack/src/monitor/relay-source.ts | 35 +- 23 files changed, 1177 insertions(+), 935 deletions(-) delete mode 100644 extensions/slack/src/monitor/inbound-delivery-state.test.ts delete mode 100644 extensions/slack/src/monitor/inbound-delivery-state.ts create mode 100644 extensions/slack/src/monitor/ingress.test.ts create mode 100644 extensions/slack/src/monitor/ingress.ts create mode 100644 extensions/slack/src/monitor/message-dispatch-dedupe.ts delete mode 100644 extensions/slack/src/monitor/message-handler.app-mention-race.test.ts diff --git a/config/max-lines-baseline.txt b/config/max-lines-baseline.txt index 11aa46857b50..6d162b66a2d9 100644 --- a/config/max-lines-baseline.txt +++ b/config/max-lines-baseline.txt @@ -278,7 +278,6 @@ extensions/slack/src/approval-native.test.ts extensions/slack/src/channel.test.ts extensions/slack/src/channel.ts extensions/slack/src/message-action-dispatch.test.ts -extensions/slack/src/monitor/context.ts extensions/slack/src/monitor/events/interactions.block-actions.ts extensions/slack/src/monitor/events/interactions.test.ts extensions/slack/src/monitor/media.test.ts diff --git a/docs/.generated/plugin-sdk-api-baseline.sha256 b/docs/.generated/plugin-sdk-api-baseline.sha256 index 30d1406852b6..7f56ddc5623b 100644 --- a/docs/.generated/plugin-sdk-api-baseline.sha256 +++ b/docs/.generated/plugin-sdk-api-baseline.sha256 @@ -1,2 +1,2 @@ -8b832e3973b7530eb65ebf7731d45f8d7c4d57b68065dd7c74912db4502eedbc plugin-sdk-api-baseline.json -9a1adc5fbff805b5b8b1af25f569b7b2dbdde505a550072621ba569fd32fdd1d plugin-sdk-api-baseline.jsonl +244b02936e24b4c5ae8ca596c7a2255e2da1026b23ad445108bd5135d6ed8564 plugin-sdk-api-baseline.json +3566fb622a7173c4e238d449d2c605a43a21cae6d966782acb046a7925d3a20e plugin-sdk-api-baseline.jsonl diff --git a/extensions/slack/src/monitor.test-helpers.ts b/extensions/slack/src/monitor.test-helpers.ts index c1b4a556b71a..47c07e105fc4 100644 --- a/extensions/slack/src/monitor.test-helpers.ts +++ b/extensions/slack/src/monitor.test-helpers.ts @@ -1,7 +1,10 @@ -// Slack helper module supports monitor helpers behavior. +import fs from "node:fs"; +import path from "node:path"; import type { ChannelRuntimeSurface } from "openclaw/plugin-sdk/channel-contract"; -import { resolveGlobalDedupeCache } from "openclaw/plugin-sdk/dedupe-runtime"; +// Slack helper module supports monitor helpers behavior. +import { closeOpenClawStateDatabaseForTest } from "openclaw/plugin-sdk/plugin-state-test-runtime"; import type { RuntimeEnv } from "openclaw/plugin-sdk/runtime-env"; +import { resolvePreferredOpenClawTmpDir } from "openclaw/plugin-sdk/temp-path"; import { vi } from "vitest"; import type { Mock } from "vitest"; @@ -55,14 +58,6 @@ const slackTestState: SlackTestState = vi.hoisted(() => ({ createSlackStartupAuthClientMock: vi.fn(), })); -const slackInboundDeliveryTestCache = resolveGlobalDedupeCache( - Symbol.for("openclaw.slackInboundDeliveries"), - { - ttlMs: 24 * 60 * 60 * 1000, - maxSize: 20_000, - }, -); - export const getSlackTestState = (): SlackTestState => slackTestState; export function useRealSlackStartupAuthClientOnce(): void { @@ -236,8 +231,21 @@ export const defaultSlackTestConfig = () => ({ }, }); +let lastSlackTestStateDir: string | undefined; + export function resetSlackTestState(config: Record = defaultSlackTestConfig()) { - slackInboundDeliveryTestCache.clear(); + // Fresh persistent state per test: the dispatch-dedupe guard writes logical + // message keys to the state DB, and fixture ts values repeat across tests, + // so a carried-over DB would dedupe unrelated test messages. realpath keeps + // macOS /var vs /private/var symlinks out of resolver assertions. + closeOpenClawStateDatabaseForTest(); + if (lastSlackTestStateDir) { + fs.rmSync(lastSlackTestStateDir, { recursive: true, force: true }); + } + lastSlackTestStateDir = fs.realpathSync( + fs.mkdtempSync(path.join(resolvePreferredOpenClawTmpDir(), "openclaw-slack-monitor-state-")), + ); + process.env.OPENCLAW_STATE_DIR = lastSlackTestStateDir; slackTestState.config = config; slackTestState.socketModeLogger = undefined; slackTestState.appStartMock.mockReset().mockResolvedValue(undefined); diff --git a/extensions/slack/src/monitor/context.ts b/extensions/slack/src/monitor/context.ts index 737661281ae1..a172b701240d 100644 --- a/extensions/slack/src/monitor/context.ts +++ b/extensions/slack/src/monitor/context.ts @@ -155,16 +155,6 @@ export type SlackMonitorContext = { removeAckAfterReply: boolean; logger: ReturnType; - markMessageSeen: ( - channelId: string | undefined, - ts?: string, - eventScope?: SlackEventScope, - ) => boolean; - releaseSeenMessage: ( - channelId: string | undefined, - ts?: string, - eventScope?: SlackEventScope, - ) => void; shouldDropMismatchedSlackEvent: (body: unknown) => boolean; resolveSlackSystemEventSessionKey: (params: { channelId?: string | null; @@ -266,7 +256,6 @@ export function createSlackMonitorContext(params: { const channelCache = new Map(); const userCache = new Map(); - const seenMessages = createDedupeCache({ ttlMs: 60_000, maxSize: 500 }); // Rate-limit active denials while retaining periodic evidence; bound keys against config churn. const channelDenialWarnings = createDedupeCache({ ttlMs: SLACK_CHANNEL_DENIAL_WARNING_TTL_MS, @@ -329,28 +318,6 @@ export function createSlackMonitorContext(params: { return id ? readLruMapEntry(channelCache, scopedKey(id, eventScope))?.info.type : undefined; }; - const markMessageSeen = ( - channelId: string | undefined, - ts?: string, - eventScope?: SlackEventScope, - ) => { - if (!channelId || !ts) { - return false; - } - return seenMessages.check(scopedKey(`${channelId}:${ts}`, eventScope)); - }; - - const releaseSeenMessage = ( - channelId: string | undefined, - ts?: string, - eventScope?: SlackEventScope, - ) => { - if (!channelId || !ts) { - return; - } - seenMessages.delete(scopedKey(`${channelId}:${ts}`, eventScope)); - }; - const assistantContextKey = (channelId: string, threadTs: string, eventScope?: SlackEventScope) => scopedKey(`${channelId}:${threadTs}`, eventScope); @@ -771,8 +738,6 @@ export function createSlackMonitorContext(params: { mediaMaxBytes: params.mediaMaxBytes, removeAckAfterReply: params.removeAckAfterReply, logger, - markMessageSeen, - releaseSeenMessage, shouldDropMismatchedSlackEvent, resolveSlackSystemEventSessionKey, isChannelAllowed, @@ -786,4 +751,3 @@ export function createSlackMonitorContext(params: { setSlackAssistantSuggestedPrompts, }; } -/* oxlint-disable max-lines -- TODO: split this grandfathered oversized file. */ diff --git a/extensions/slack/src/monitor/events/messages.test.ts b/extensions/slack/src/monitor/events/messages.test.ts index 3f3941574cc0..57877b8c41bb 100644 --- a/extensions/slack/src/monitor/events/messages.test.ts +++ b/extensions/slack/src/monitor/events/messages.test.ts @@ -7,6 +7,8 @@ import { type SlackSystemEventTestOverrides, } from "./system-event-test-harness.js"; +const SLACK_INGRESS_LIFECYCLE_CONTEXT_KEY = "openclawIngressLifecycle"; + const { messageQueueMock, messageAllowMock, inboundInfoSpy } = vi.hoisted(() => ({ messageQueueMock: vi.fn(), messageAllowMock: vi.fn(), @@ -241,6 +243,47 @@ async function runMessageCase(input: MessageCase = {}): Promise { } describe("registerSlackMessageEvents", () => { + it("forwards durable ingress ownership and propagates dispatch failure", async () => { + const harness = createSlackSystemEventTestHarness(); + const dispatchError = new Error("transient dispatch failure"); + const handleSlackMessage = vi.fn(async () => { + throw dispatchError; + }); + registerSlackMessageEvents({ ctx: harness.ctx, handleSlackMessage }); + const handler = requireMessageHandler(harness.getHandler("message") as MessageHandler | null); + const turnAdoptionLifecycle = { + admission: "exclusive", + abortSignal: new AbortController().signal, + onAdopted: vi.fn(), + onDeferred: vi.fn(), + onAbandoned: vi.fn(), + }; + + await expect( + handler({ + event: { + type: "message", + channel: "D1", + channel_type: "im", + user: "U1", + text: "hello", + ts: "123.456", + }, + body: {}, + context: { [SLACK_INGRESS_LIFECYCLE_CONTEXT_KEY]: turnAdoptionLifecycle }, + }), + ).rejects.toBe(dispatchError); + + expect(handleSlackMessage).toHaveBeenCalledWith( + expect.objectContaining({ channel: "D1", ts: "123.456" }), + expect.objectContaining({ + source: "message", + awaitDispatch: true, + turnAdoptionLifecycle, + }), + ); + }); + it("accepts two org workspaces and preserves each listener scope", async () => { const { handler, handleSlackMessage } = createEnterpriseHandlers("message"); const clients = [{ id: "one" }, { id: "two" }]; diff --git a/extensions/slack/src/monitor/events/messages.ts b/extensions/slack/src/monitor/events/messages.ts index d0b3882085d2..ce5680566746 100644 --- a/extensions/slack/src/monitor/events/messages.ts +++ b/extensions/slack/src/monitor/events/messages.ts @@ -16,6 +16,7 @@ import type { SlackAppMentionEvent, SlackMessageEvent } from "../../types.js"; import { normalizeSlackChannelType } from "../channel-type.js"; import type { SlackMonitorContext } from "../context.js"; import { resolveSlackEventScope, type SlackEventScope } from "../event-scope.js"; +import { resolveSlackIngressTurnLifecycle } from "../ingress.js"; import type { SlackMessageHandler } from "../message-handler.js"; import type { SlackMessageChangedEvent } from "../types.js"; import { resolveSlackMessageSubtypeHandler } from "./message-subtype-handlers.js"; @@ -196,6 +197,7 @@ export function registerSlackMessageEvents(params: { context: AllMiddlewareArgs["context"]; client: AllMiddlewareArgs["client"]; }) => { + const turnAdoptionLifecycle = resolveSlackIngressTurnLifecycle(context); try { const eventScope = resolveEventScope({ body, context, client }); if (eventScope === null) { @@ -224,7 +226,9 @@ export function registerSlackMessageEvents(params: { if (assistantChangedInbound) { await handleSlackMessage(assistantChangedInbound, { source: "message", - ...(eventScope ? { eventScope, awaitDispatch: true } : {}), + ...(eventScope ? { eventScope } : {}), + ...(turnAdoptionLifecycle ? { turnAdoptionLifecycle } : {}), + ...(eventScope || turnAdoptionLifecycle ? { awaitDispatch: true } : {}), }); return; } @@ -264,9 +268,14 @@ export function registerSlackMessageEvents(params: { await handleSlackMessage(message, { source: "message", - ...(eventScope ? { eventScope, awaitDispatch: true } : {}), + ...(eventScope ? { eventScope } : {}), + ...(turnAdoptionLifecycle ? { turnAdoptionLifecycle } : {}), + ...(eventScope || turnAdoptionLifecycle ? { awaitDispatch: true } : {}), }); } catch (err) { + if (turnAdoptionLifecycle) { + throw err; + } ctx.runtime.error?.(danger(`slack handler failed: ${formatErrorMessage(err)}`)); } }; @@ -288,6 +297,7 @@ export function registerSlackMessageEvents(params: { "app_mention", async (args: SlackEventMiddlewareArgs<"app_mention"> & AllMiddlewareArgs) => { const { event, body, context, client } = args; + const turnAdoptionLifecycle = resolveSlackIngressTurnLifecycle(context); try { const eventScope = resolveEventScope({ body, context, client }); if (eventScope === null) { @@ -328,9 +338,14 @@ export function registerSlackMessageEvents(params: { await handleSlackMessage(mention as unknown as SlackMessageEvent, { source: "app_mention", wasMentioned: true, - ...(eventScope ? { eventScope, awaitDispatch: true } : {}), + ...(eventScope ? { eventScope } : {}), + ...(turnAdoptionLifecycle ? { turnAdoptionLifecycle } : {}), + ...(eventScope || turnAdoptionLifecycle ? { awaitDispatch: true } : {}), }); } catch (err) { + if (turnAdoptionLifecycle) { + throw err; + } ctx.runtime.error?.(danger(`slack mention handler failed: ${formatErrorMessage(err)}`)); } }, diff --git a/extensions/slack/src/monitor/inbound-delivery-state.test.ts b/extensions/slack/src/monitor/inbound-delivery-state.test.ts deleted file mode 100644 index 530e6f254dc6..000000000000 --- a/extensions/slack/src/monitor/inbound-delivery-state.test.ts +++ /dev/null @@ -1,71 +0,0 @@ -// Slack tests cover inbound delivery state plugin behavior. -import { afterEach, describe, expect, it, vi } from "vitest"; -import { setSlackRuntime } from "../runtime.js"; -import type { SlackMessageEvent } from "../types.js"; -import { - hasSlackInboundMessageDelivery, - recordSlackInboundMessageDeliveries, -} from "./inbound-delivery-state.js"; - -describe("slack inbound delivery state", () => { - afterEach(() => { - setSlackRuntime(null as never); - vi.restoreAllMocks(); - }); - - function message(channel: string, ts: string): SlackMessageEvent { - return { type: "message", channel, ts, text: "hello" }; - } - - it("records every delivered debounced source message", async () => { - const register = vi.fn().mockResolvedValue(undefined); - setSlackRuntime({ - state: { - openKeyedStore: vi.fn(() => ({ - register, - lookup: vi.fn(), - consume: vi.fn(), - delete: vi.fn(), - entries: vi.fn(), - clear: vi.fn(), - })), - }, - logging: { getChildLogger: () => ({ warn: vi.fn() }) }, - } as never); - - await recordSlackInboundMessageDeliveries({ - accountId: "A1", - messages: [message("C1", "100.001"), message("C1", "100.002")], - }); - - expect(register).toHaveBeenCalledTimes(2); - expect(register).toHaveBeenCalledWith("A1:C1:100.001", { - deliveredAt: expect.any(Number), - }); - expect(register).toHaveBeenCalledWith("A1:C1:100.002", { - deliveredAt: expect.any(Number), - }); - }); - - it("scopes duplicate checks by account", async () => { - await recordSlackInboundMessageDeliveries({ - accountId: "A-scope-1", - messages: [message("C-scope", "200.001")], - }); - - await expect( - hasSlackInboundMessageDelivery({ - accountId: "A-scope-1", - channelId: "C-scope", - ts: "200.001", - }), - ).resolves.toBe(true); - await expect( - hasSlackInboundMessageDelivery({ - accountId: "A-scope-2", - channelId: "C-scope", - ts: "200.001", - }), - ).resolves.toBe(false); - }); -}); diff --git a/extensions/slack/src/monitor/inbound-delivery-state.ts b/extensions/slack/src/monitor/inbound-delivery-state.ts deleted file mode 100644 index 41eaa66c7ad7..000000000000 --- a/extensions/slack/src/monitor/inbound-delivery-state.ts +++ /dev/null @@ -1,75 +0,0 @@ -// Slack plugin module implements inbound delivery state behavior. -import { createPersistentDedupeCache } from "openclaw/plugin-sdk/dedupe-runtime"; -import { getOptionalSlackRuntime } from "../runtime.js"; -import type { SlackMessageEvent } from "../types.js"; - -const TTL_MS = 24 * 60 * 60 * 1000; -const MAX_ENTRIES = 20_000; -const PERSISTENT_MAX_ENTRIES = 20_000; -const PERSISTENT_NAMESPACE = "slack.inbound-deliveries"; -const SLACK_INBOUND_DELIVERIES_KEY = Symbol.for("openclaw.slackInboundDeliveries"); - -type SlackInboundDeliveryRecord = { - deliveredAt: number; -}; - -const deliveredMessages = createPersistentDedupeCache({ - globalKey: SLACK_INBOUND_DELIVERIES_KEY, - ttlMs: TTL_MS, - maxSize: MAX_ENTRIES, - persistent: { - namespace: PERSISTENT_NAMESPACE, - maxEntries: PERSISTENT_MAX_ENTRIES, - openStore: (options) => getOptionalSlackRuntime()?.state.openKeyedStore(options), - logError: (error) => { - try { - getOptionalSlackRuntime() - ?.logging.getChildLogger({ plugin: "slack", feature: "inbound-delivery-state" }) - .warn("Slack persistent inbound delivery state failed", { error: String(error) }); - } catch { - // Best effort only: persistent state must never break Slack message handling. - } - }, - }, -}); - -function makeKey(accountId: string, channelId: string, ts: string, teamId?: string): string { - return `${accountId}:${teamId ? `${teamId}:` : ""}${channelId}:${ts}`; -} - -export async function hasSlackInboundMessageDelivery(params: { - accountId: string; - channelId: string | undefined; - ts: string | undefined; - teamId?: string; -}): Promise { - if (!params.accountId || !params.channelId || !params.ts) { - return false; - } - return await deliveredMessages.lookup( - makeKey(params.accountId, params.channelId, params.ts, params.teamId), - ); -} - -export async function recordSlackInboundMessageDeliveries(params: { - accountId: string; - messages: readonly SlackMessageEvent[]; - teamId?: string; -}): Promise { - if (!params.accountId || params.messages.length === 0) { - return; - } - const deliveredAt = Date.now(); - const keys = new Set(); - for (const message of params.messages) { - if (!message.channel || !message.ts) { - continue; - } - keys.add(makeKey(params.accountId, message.channel, message.ts, params.teamId)); - } - await Promise.all( - Array.from(keys, (key) => - deliveredMessages.register(key, { deliveredAt }, { at: deliveredAt }), - ), - ); -} diff --git a/extensions/slack/src/monitor/ingress.test.ts b/extensions/slack/src/monitor/ingress.test.ts new file mode 100644 index 000000000000..ebc576815b03 --- /dev/null +++ b/extensions/slack/src/monitor/ingress.test.ts @@ -0,0 +1,288 @@ +// Slack tests cover durable Events API admission, replay, and tombstones. +import crypto from "node:crypto"; +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import type { App, Receiver, ReceiverEvent } from "@slack/bolt"; +import type { ChannelIngressQueue } from "openclaw/plugin-sdk/channel-outbound"; +import { + closeOpenClawStateDatabaseForTest, + createChannelIngressQueueForTests, +} from "openclaw/plugin-sdk/plugin-state-test-runtime"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { createSlackDurableIngress, resolveSlackIngressTurnLifecycle } from "./ingress.js"; + +type SlackIngressQueue = NonNullable[0]["queue"]>; +type SlackIngressPayload = Parameters[1]; + +function createSlackEnvelope(eventId: string, ts = "1700000000.000100") { + return { + team_id: "T_TEST", + api_app_id: "A_TEST", + type: "event_callback", + event_id: eventId, + event_time: 1_700_000_000, + event: { + type: "message", + channel: "C_TEST", + user: "U_TEST", + ts, + client_msg_id: "client-message-1", + text: "hello", + }, + }; +} + +function createReceiverHarness() { + let receive: ((event: ReceiverEvent) => Promise) | undefined; + const receiver: Receiver = { + init: (app) => { + receive = async (event) => await app.processEvent(event); + }, + start: async () => undefined, + stop: async () => undefined, + }; + return { + receiver, + receive: async (event: ReceiverEvent) => { + if (!receive) { + throw new Error("Receiver not initialized"); + } + await receive(event); + }, + }; +} + +function createReceiverEvent( + eventId: string, + ack = vi.fn(async () => {}), + options: { retryNum?: number; ts?: string } = {}, +): ReceiverEvent { + return { + body: createSlackEnvelope(eventId, options.ts), + ack, + ...(options.retryNum === undefined ? {} : { retryNum: options.retryNum }), + }; +} + +function attachIngress( + queue: ChannelIngressQueue, + processEvent: (event: ReceiverEvent) => Promise, +) { + const ingress = createSlackDurableIngress({ + accountId: "default", + queue, + pollIntervalMs: 60_000, + adoptionStallTimeoutMs: 5_000, + }); + const harness = createReceiverHarness(); + ingress.wrapReceiver(harness.receiver).init({ processEvent } as App); + return { ingress, receive: harness.receive }; +} + +async function withQueue( + fn: (queue: ChannelIngressQueue) => Promise, +): Promise { + const rawRoot = await fs.mkdtemp( + path.join(os.tmpdir(), `openclaw-slack-ingress-${crypto.randomUUID()}-`), + ); + const stateDir = await fs.realpath(rawRoot); + const queue = createChannelIngressQueueForTests({ + channelId: "slack", + accountId: "default", + stateDir, + }); + try { + await fn(queue); + } finally { + closeOpenClawStateDatabaseForTest(); + await fs.rm(stateDir, { recursive: true, force: true }); + } +} + +describe("Slack durable ingress", () => { + afterEach(() => { + closeOpenClawStateDatabaseForTest(); + }); + + it("does not acknowledge when the durable append fails", async () => { + await withQueue(async (queue) => { + const enqueue = vi.fn(async () => { + throw new Error("database unavailable"); + }); + const failingQueue = { ...queue, enqueue } as ChannelIngressQueue; + const processEvent = vi.fn(async () => {}); + const { ingress, receive } = attachIngress(failingQueue, processEvent); + const ack = vi.fn(async () => {}); + + await expect(receive(createReceiverEvent("Ev-append-failure", ack))).rejects.toThrow( + "database unavailable", + ); + + expect(enqueue).toHaveBeenCalledTimes(1); + expect(ack).not.toHaveBeenCalled(); + expect(processEvent).not.toHaveBeenCalled(); + await ingress.stop(); + }); + }); + + it("recovers an uncompleted event with a fresh drain and dispatches once", async () => { + await withQueue(async (queue) => { + const first = attachIngress( + queue, + vi.fn(async () => {}), + ); + const ack = vi.fn(async () => {}); + await first.receive(createReceiverEvent("Ev-restart", ack)); + await first.ingress.stop(); + + const dispatch = vi.fn(async (event: ReceiverEvent) => { + await resolveSlackIngressTurnLifecycle(event.customProperties)?.onAdopted(); + }); + const restarted = attachIngress(queue, dispatch); + restarted.ingress.start(); + await restarted.ingress.waitForIdle(); + + expect(ack).toHaveBeenCalledTimes(1); + expect(dispatch).toHaveBeenCalledTimes(1); + expect((await queue.enqueue("Ev-restart", {} as SlackIngressPayload)).kind).toBe("completed"); + await restarted.ingress.stop(); + }); + }); + + it("retains completion so the same event_id cannot dispatch twice", async () => { + await withQueue(async (queue) => { + const dispatch = vi.fn(async (event: ReceiverEvent) => { + await resolveSlackIngressTurnLifecycle(event.customProperties)?.onAdopted(); + }); + const { ingress, receive } = attachIngress(queue, dispatch); + ingress.start(); + await receive(createReceiverEvent("Ev-completed")); + await ingress.waitForIdle(); + + const duplicateAck = vi.fn(async () => {}); + await receive(createReceiverEvent("Ev-completed", duplicateAck, { retryNum: 1 })); + await ingress.waitForIdle(); + + expect(duplicateAck).toHaveBeenCalledTimes(1); + expect(dispatch).toHaveBeenCalledTimes(1); + expect((await queue.enqueue("Ev-completed", {} as SlackIngressPayload)).kind).toBe( + "completed", + ); + await ingress.stop(); + }); + }); + + it("dedupes Slack's delayed message redelivery after restart via the tombstone", async () => { + await withQueue(async (queue) => { + const firstDispatch = vi.fn(async (event: ReceiverEvent) => { + await resolveSlackIngressTurnLifecycle(event.customProperties)?.onAdopted(); + }); + const first = attachIngress(queue, firstDispatch); + first.ingress.start(); + await first.receive( + createReceiverEvent("Ev-delayed-redelivery", undefined, { + ts: "1700000000.000350", + }), + ); + await first.ingress.waitForIdle(); + await first.ingress.stop(); + + const replayDispatch = vi.fn(async () => {}); + const restarted = attachIngress(queue, replayDispatch); + const retryAck = vi.fn(async () => {}); + await restarted.receive( + createReceiverEvent("Ev-delayed-redelivery", retryAck, { + retryNum: 3, + ts: "1700000000.000350", + }), + ); + restarted.ingress.start(); + await restarted.ingress.waitForIdle(); + + expect(firstDispatch).toHaveBeenCalledTimes(1); + expect(retryAck).toHaveBeenCalledTimes(1); + expect(replayDispatch).not.toHaveBeenCalled(); + await restarted.ingress.stop(); + }); + }); +}); + +describe("Slack relay durable ingress", () => { + afterEach(() => { + closeOpenClawStateDatabaseForTest(); + }); + + const relayMessage = { + type: "message", + channel: "C_RELAY", + team: "T_TEST", + user: "U_TEST", + ts: "1700000001.000200", + text: "relayed", + }; + + it("dedupes a router redelivery by logical message identity, not delivery id", async () => { + await withQueue(async (queue) => { + const dispatched: unknown[] = []; + const ingress = createSlackDurableIngress({ + accountId: "default", + queue, + pollIntervalMs: 60_000, + adoptionStallTimeoutMs: 5_000, + }); + ingress.attachRelayDispatch(async (message) => { + dispatched.push(message); + }); + ingress.start(); + + await ingress.acceptRelayEvent({ deliveryId: "delivery-1", message: relayMessage }); + await ingress.waitForIdle(); + // Redelivery after a lost ack carries a fresh delivery id but the same message. + await ingress.acceptRelayEvent({ deliveryId: "delivery-2", message: relayMessage }); + await ingress.waitForIdle(); + + expect(dispatched).toHaveLength(1); + expect(dispatched[0]).toMatchObject({ channel: "C_RELAY", text: "relayed" }); + await ingress.stop(); + }); + }); + + it("retries a claimed relay event until a dispatcher attaches", async () => { + await withQueue(async (queue) => { + const detached = createSlackDurableIngress({ + accountId: "default", + queue, + pollIntervalMs: 60_000, + adoptionStallTimeoutMs: 5_000, + }); + // Accept durably, then stop before any dispatcher exists (crash window). + await detached.acceptRelayEvent({ deliveryId: "delivery-3", message: relayMessage }); + await detached.stop(); + + const dispatched: unknown[] = []; + const recovered = createSlackDurableIngress({ + accountId: "default", + queue, + pollIntervalMs: 25, + adoptionStallTimeoutMs: 5_000, + }); + recovered.start(); + await recovered.waitForIdle(); + expect(dispatched).toHaveLength(0); + + recovered.attachRelayDispatch(async (message) => { + dispatched.push(message); + }); + // First retry obeys the drain's backoff; give it room without flake. + await vi.waitFor( + async () => { + await recovered.waitForIdle(); + expect(dispatched).toHaveLength(1); + }, + { timeout: 15_000, interval: 250 }, + ); + await recovered.stop(); + }); + }); +}); diff --git a/extensions/slack/src/monitor/ingress.ts b/extensions/slack/src/monitor/ingress.ts new file mode 100644 index 000000000000..64accc0fc84a --- /dev/null +++ b/extensions/slack/src/monitor/ingress.ts @@ -0,0 +1,394 @@ +// Slack plugin module owns durable Events API admission and replay. +import type { App, Receiver, ReceiverEvent } from "@slack/bolt"; +import { + bindIngressLifecycleToReplyOptions, + 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, + extractErrorCode, + formatErrorMessage, +} from "openclaw/plugin-sdk/error-runtime"; +import type { PluginJsonValue } from "openclaw/plugin-sdk/plugin-entry"; +import { asOptionalRecord } from "openclaw/plugin-sdk/string-coerce-runtime"; +import { getSlackRuntime } from "../runtime.js"; +import { isNonRecoverableSlackAuthError } from "./reconnect-policy.js"; + +const SLACK_INGRESS_PAYLOAD_VERSION = 1; +const SLACK_INGRESS_POLL_INTERVAL_MS = 1_000; +const SLACK_INGRESS_PRUNE_INTERVAL_MS = 60 * 60 * 1_000; +const SLACK_INGRESS_COMPLETED_TTL_MS = 30 * 24 * 60 * 60 * 1_000; +const SLACK_INGRESS_COMPLETED_MAX_ENTRIES = 20_000; +const SLACK_INGRESS_FAILED_TTL_MS = 30 * 24 * 60 * 60 * 1_000; +const SLACK_INGRESS_FAILED_MAX_ENTRIES = 20_000; +const SLACK_BOLT_AUTHORIZATION_ERROR = "slack_bolt_authorization_error"; + +const SLACK_INGRESS_LIFECYCLE_CONTEXT_KEY = "openclawIngressLifecycle"; + +export type SlackIngressTurnLifecycle = ReturnType< + typeof bindIngressLifecycleToReplyOptions +>["turnAdoptionLifecycle"]; + +type SlackIngressPayload = { + version: number; + receivedAt: number; +} & ( + | { + kind: "events-api"; + body: PluginJsonValue; + retryNum?: number; + retryReason?: string; + } + // Relay frames carry a bare message event (no Events API envelope), so the + // durable key is the logical message identity — the retired guard's exact + // key space — instead of a router delivery id whose redelivery stability + // is not a documented contract. + | { kind: "relay"; message: PluginJsonValue } +); + +type SlackRelayIngressEvent = { + deliveryId: string; + message: { channel: string; ts?: string; team?: string }; +}; + +type SlackRelayIngressDispatch = ( + message: PluginJsonValue, + lifecycle: SlackIngressTurnLifecycle, +) => Promise; + +/** Logical message identity: mirrors the retired guard key (team:channel:ts). */ +function resolveSlackRelayIngressEventId(event: SlackRelayIngressEvent): string { + const ts = event.message.ts?.trim(); + if (!event.message.channel?.trim() || !ts) { + return `relay:${event.deliveryId}`; + } + const team = event.message.team?.trim(); + return `message:${team ? `${team}:` : ""}${event.message.channel.trim()}:${ts}`; +} + +type SlackDurableIngressOptions = { + accountId: string; + queue?: ChannelIngressQueue; + pollIntervalMs?: number; + adoptionStallTimeoutMs?: number; + onLog?: (message: string) => void; + abortSignal?: AbortSignal; +}; + +type SlackDurableIngress = { + wrapReceiver: (receiver: Receiver) => Receiver; + /** Durable-before-ack accept for relay frames; caller acks after this resolves. */ + acceptRelayEvent: (event: SlackRelayIngressEvent) => Promise; + /** Relay-mode dispatcher; claimed relay events retry until one is attached. */ + attachRelayDispatch: (dispatch: SlackRelayIngressDispatch) => void; + start: () => void; + stop: () => Promise; + waitForIdle: () => Promise; +}; + +class SlackIngressPayloadError extends Error { + constructor(message: string) { + super(message); + this.name = "SlackIngressPayloadError"; + } +} + +function resolveSlackEventId(body: unknown): string | null { + const eventId = asOptionalRecord(body)?.event_id; + return typeof eventId === "string" && eventId.trim() ? eventId.trim() : null; +} + +function resolveSlackIngressLane(body: unknown, eventId: string): string { + const envelope = asOptionalRecord(body); + const event = asOptionalRecord(envelope?.event); + const item = asOptionalRecord(event?.item); + const assistantThread = asOptionalRecord(event?.assistant_thread); + const team = asOptionalRecord(envelope?.team); + const teamId = + [envelope?.team_id, team?.id, event?.team] + .find((value) => typeof value === "string" && value.trim()) + ?.toString() + .trim() || "workspace"; + const channelId = [event?.channel, event?.channel_id, item?.channel, assistantThread?.channel_id] + .find((value) => typeof value === "string" && value.trim()) + ?.toString() + .trim(); + if (channelId) { + return `team:${teamId}:conversation:${channelId}`; + } + const userId = [event?.user, event?.user_id] + .find((value) => typeof value === "string" && value.trim()) + ?.toString() + .trim(); + return userId ? `team:${teamId}:user:${userId}` : `event:${eventId}`; +} + +function isSlackEventCallback(body: unknown): boolean { + return asOptionalRecord(body)?.type === "event_callback"; +} + +function assertSlackIngressPayload( + payload: SlackIngressPayload, + eventId: string, +): asserts payload is SlackIngressPayload { + if (payload.version !== SLACK_INGRESS_PAYLOAD_VERSION) { + throw new SlackIngressPayloadError(`Slack ingress payload ${eventId} was invalid.`); + } + if (payload.kind === "relay") { + if (!asOptionalRecord(payload.message)) { + throw new SlackIngressPayloadError(`Slack relay ingress payload ${eventId} was invalid.`); + } + return; + } + if (!asOptionalRecord(payload.body) || resolveSlackEventId(payload.body) !== eventId) { + throw new SlackIngressPayloadError(`Slack ingress payload ${eventId} was invalid.`); + } +} + +function resolveSlackIngressNonRetryableFailure(error: unknown) { + for (const candidate of collectErrorGraphCandidates(error, (current) => [ + current.cause, + current.error, + current.original, + ])) { + if (candidate instanceof SlackIngressPayloadError || candidate instanceof SyntaxError) { + return { reason: "invalid-event", message: formatErrorMessage(candidate) }; + } + if ( + extractErrorCode(candidate) === SLACK_BOLT_AUTHORIZATION_ERROR || + isNonRecoverableSlackAuthError(candidate) + ) { + return { reason: "slack-auth", message: formatErrorMessage(candidate) }; + } + } + return null; +} + +export function resolveSlackIngressTurnLifecycle( + context: unknown, +): SlackIngressTurnLifecycle | null { + const candidate = asOptionalRecord(context)?.[SLACK_INGRESS_LIFECYCLE_CONTEXT_KEY]; + if (!candidate || typeof candidate !== "object") { + return null; + } + const lifecycle = candidate as Partial; + return typeof lifecycle.onAdopted === "function" && lifecycle.abortSignal instanceof AbortSignal + ? (lifecycle as SlackIngressTurnLifecycle) + : null; +} + +export function createSlackDurableIngress( + options: SlackDurableIngressOptions, +): SlackDurableIngress { + let queue = options.queue; + let drain: ChannelIngressDrain | undefined; + let app: App | undefined; + let relayDispatch: SlackRelayIngressDispatch | undefined; + let running = false; + let requested = false; + let pumping: Promise | undefined; + let pollTimer: ReturnType | undefined; + let lastPrunedAt = 0; + + const getQueue = (): ChannelIngressQueue => { + queue ??= getSlackRuntime().state.openChannelIngressQueue({ + accountId: options.accountId, + }); + return queue; + }; + + const getDrain = (): ChannelIngressDrain => { + drain ??= createChannelIngressDrain({ + queue: getQueue(), + adoptionStallTimeoutMs: options.adoptionStallTimeoutMs ?? DEFAULT_INGRESS_ADOPTION_STALL_MS, + retryPolicy: { + maxAttempts: DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS, + deadLetterMinAgeMs: DEFAULT_INGRESS_RETRY_DEAD_LETTER_MIN_AGE_MS, + }, + resolveNonRetryableFailure: resolveSlackIngressNonRetryableFailure, + deriveLaneKey: (record) => + record.payload.kind === "relay" + ? resolveSlackIngressLane({ event: record.payload.message }, record.id) + : resolveSlackIngressLane(record.payload.body, record.id), + ...(options.onLog ? { onLog: options.onLog } : {}), + ...(options.abortSignal ? { abortSignal: options.abortSignal } : {}), + dispatchClaimedEvent: async (event, lifecycle) => { + assertSlackIngressPayload(event.payload, event.id); + if (lifecycle.abortSignal.aborted) { + throw lifecycle.abortSignal.reason; + } + const bound = bindIngressLifecycleToReplyOptions(lifecycle); + if (event.payload.kind === "relay") { + if (!relayDispatch) { + // Transient by design: a claim recovered before the relay source + // reattaches must retry, not dead-letter, or restart recovery loses it. + throw new Error("Slack relay ingress dispatcher is not attached."); + } + await relayDispatch(event.payload.message, bound.turnAdoptionLifecycle); + return; + } + if (!app) { + throw new Error("Slack ingress receiver is not attached to a Bolt app."); + } + await app.processEvent({ + body: event.payload.body as ReceiverEvent["body"], + ack: async () => {}, + ...(event.payload.retryNum === undefined ? {} : { retryNum: event.payload.retryNum }), + ...(event.payload.retryReason === undefined + ? {} + : { retryReason: event.payload.retryReason }), + customProperties: { + [SLACK_INGRESS_LIFECYCLE_CONTEXT_KEY]: bound.turnAdoptionLifecycle, + }, + }); + }, + }); + return drain; + }; + + const pruneIfDue = async (): Promise => { + const now = Date.now(); + if (now - lastPrunedAt < SLACK_INGRESS_PRUNE_INTERVAL_MS) { + return; + } + await getQueue().prune({ + completedTtlMs: SLACK_INGRESS_COMPLETED_TTL_MS, + completedMaxEntries: SLACK_INGRESS_COMPLETED_MAX_ENTRIES, + failedTtlMs: SLACK_INGRESS_FAILED_TTL_MS, + failedMaxEntries: SLACK_INGRESS_FAILED_MAX_ENTRIES, + now, + }); + lastPrunedAt = now; + }; + + const runPump = async (): Promise => { + try { + for (;;) { + requested = false; + await pruneIfDue(); + const activeDrain = getDrain(); + const { started } = await activeDrain.drainOnce(); + await activeDrain.waitForIdle(); + if (!running || (!requested && started === 0)) { + break; + } + } + } catch (error) { + options.onLog?.(`slack ingress drain failed: ${formatErrorMessage(error)}`); + } finally { + pumping = undefined; + if (running && requested) { + requestDrain(); + } + } + }; + + const requestDrain = (): void => { + requested = true; + if (!running || pumping) { + return; + } + pumping = runPump(); + }; + + const acceptReceiverEvent = async (event: ReceiverEvent): Promise => { + if (!isSlackEventCallback(event.body)) { + if (!app) { + throw new Error("Slack ingress receiver is not attached to a Bolt app."); + } + await app.processEvent(event); + return; + } + const eventId = resolveSlackEventId(event.body); + if (!eventId) { + throw new SlackIngressPayloadError("Slack Events API envelope missing event_id."); + } + const receivedAt = Date.now(); + await getQueue().enqueue( + eventId, + { + version: SLACK_INGRESS_PAYLOAD_VERSION, + receivedAt, + kind: "events-api", + body: event.body as PluginJsonValue, + ...(event.retryNum === undefined ? {} : { retryNum: event.retryNum }), + ...(event.retryReason === undefined ? {} : { retryReason: event.retryReason }), + }, + { receivedAt }, + ); + await event.ack(); + requestDrain(); + }; + + const acceptRelayEvent = async (event: SlackRelayIngressEvent): Promise => { + const receivedAt = Date.now(); + await getQueue().enqueue( + resolveSlackRelayIngressEventId(event), + { + version: SLACK_INGRESS_PAYLOAD_VERSION, + receivedAt, + kind: "relay", + message: event.message as PluginJsonValue, + }, + { receivedAt }, + ); + requestDrain(); + }; + + return { + wrapReceiver: (receiver) => { + const client = Reflect.get(receiver as object, "client"); + const wrapped: Receiver & { client?: unknown } = { + init: (nextApp) => { + app = nextApp; + receiver.init({ processEvent: acceptReceiverEvent } as App); + }, + start: (...args) => receiver.start(...args), + stop: (...args) => receiver.stop(...args), + ...(client === undefined ? {} : { client }), + }; + return wrapped; + }, + acceptRelayEvent, + attachRelayDispatch: (dispatch) => { + relayDispatch = dispatch; + }, + start: () => { + if (running) { + return; + } + running = true; + pollTimer = setInterval( + requestDrain, + options.pollIntervalMs ?? SLACK_INGRESS_POLL_INTERVAL_MS, + ); + pollTimer.unref?.(); + requestDrain(); + }, + stop: async () => { + running = false; + if (pollTimer) { + clearInterval(pollTimer); + pollTimer = undefined; + } + drain?.dispose(); + await pumping; + await drain?.waitForIdle(); + }, + waitForIdle: async () => { + for (;;) { + const activePump = pumping; + if (!activePump) { + break; + } + await activePump; + } + await drain?.waitForIdle(); + }, + }; +} diff --git a/extensions/slack/src/monitor/message-dispatch-dedupe.ts b/extensions/slack/src/monitor/message-dispatch-dedupe.ts new file mode 100644 index 000000000000..53f200596e0b --- /dev/null +++ b/extensions/slack/src/monitor/message-dispatch-dedupe.ts @@ -0,0 +1,90 @@ +// Slack dispatch dedupe: a PERMANENT logical-identity layer above the durable +// ingress queue, not a leftover to delete on drain adoption. Slack emits BOTH +// a `message` and an `app_mention` Events API event (distinct event_ids) for +// one mention post, so the queue's event_id tombstones cannot dedupe the twin. +// Only the logical (team, channel, ts) key catches it. Claims commit at turn +// adoption and release on gated/failed dispatch so the surviving twin can +// still run the same gate without ever producing a second visible reply. +import { + createChannelReplayGuard, + type ChannelReplayClaimHandle, +} from "openclaw/plugin-sdk/persistent-dedupe"; + +// 24h/20k mirrors the retired persistent inbound-delivery state so the twin +// window also survives restarts while claimed rows sit in drain retry backoff. +const SLACK_MESSAGE_DISPATCH_DEDUPE_TTL_MS = 24 * 60 * 60 * 1000; +const SLACK_MESSAGE_DISPATCH_DEDUPE_MEMORY_MAX_ENTRIES = 20_000; +const SLACK_MESSAGE_DISPATCH_DEDUPE_STATE_MAX_ENTRIES = 20_000; +const SLACK_MESSAGE_DISPATCH_DEDUPE_NAMESPACE = "global"; +const SLACK_MESSAGE_DISPATCH_DEDUPE_NAMESPACE_PREFIX = "slack.message-dispatch-dedupe"; +const SLACK_MESSAGE_DISPATCH_DEDUPE_STATE_PLUGIN_ID = "slack-message-dispatch-dedupe"; + +export type SlackMessageDispatchReplayClaim = ChannelReplayClaimHandle; + +type SlackMessageDispatchClaimResult = + | { kind: "claimed"; handle: SlackMessageDispatchReplayClaim } + | { kind: "duplicate" }; + +export function buildSlackMessageDispatchReplayKey(params: { + accountId: string; + channelId: string | undefined; + ts: string | undefined; + teamId?: string | undefined; +}): string | null { + const channelId = params.channelId?.trim(); + const ts = params.ts?.trim(); + if (!channelId || !ts) { + return null; + } + const teamId = params.teamId?.trim(); + return JSON.stringify(["message", params.accountId, teamId ?? "", channelId, ts]); +} + +export function createSlackMessageDispatchReplayGuard( + params: { + onDiskError?: (error: unknown) => void; + } = {}, +) { + return createChannelReplayGuard<{ keys: readonly string[] }>({ + dedupe: { + ttlMs: SLACK_MESSAGE_DISPATCH_DEDUPE_TTL_MS, + memoryMaxSize: SLACK_MESSAGE_DISPATCH_DEDUPE_MEMORY_MAX_ENTRIES, + pluginId: SLACK_MESSAGE_DISPATCH_DEDUPE_STATE_PLUGIN_ID, + namespacePrefix: SLACK_MESSAGE_DISPATCH_DEDUPE_NAMESPACE_PREFIX, + stateMaxEntries: SLACK_MESSAGE_DISPATCH_DEDUPE_STATE_MAX_ENTRIES, + ...(params.onDiskError ? { onDiskError: params.onDiskError } : {}), + }, + buildReplayKey: (event) => event.keys, + namespace: () => SLACK_MESSAGE_DISPATCH_DEDUPE_NAMESPACE, + }); +} + +export type SlackMessageDispatchReplayGuard = ReturnType< + typeof createSlackMessageDispatchReplayGuard +>; + +/** Claim one logical message key; an in-flight sibling claim settles to duplicate. */ +export async function claimSlackMessageDispatchReplay(params: { + guard: SlackMessageDispatchReplayGuard; + key: string; +}): Promise { + let releaseRetries = 0; + while (true) { + const claim = await params.guard.claim({ keys: [params.key] }); + if (claim.kind === "claimed") { + return { kind: "claimed", handle: claim.handle }; + } + if (claim.kind === "duplicate" || claim.kind === "invalid") { + return { kind: "duplicate" }; + } + try { + await claim.pending; + return { kind: "duplicate" }; + } catch { + releaseRetries += 1; + if (releaseRetries > 1) { + return { kind: "duplicate" }; + } + } + } +} diff --git a/extensions/slack/src/monitor/message-handler.app-mention-race.test.ts b/extensions/slack/src/monitor/message-handler.app-mention-race.test.ts deleted file mode 100644 index 9d6312622150..000000000000 --- a/extensions/slack/src/monitor/message-handler.app-mention-race.test.ts +++ /dev/null @@ -1,381 +0,0 @@ -// Slack tests cover message handler.app mention race plugin behavior. -import { resolveGlobalDedupeCache } from "openclaw/plugin-sdk/dedupe-runtime"; -import { afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; -import { setSlackRuntime } from "../runtime.js"; - -type TestHistoryEntry = { - sender: string; - body: string; - messageId?: string; -}; - -const prepareSlackMessageMock = vi.fn< - (params: { - ctx: { channelHistories: Map }; - message: { ts?: string }; - opts: { - source: "message" | "app_mention"; - wasMentioned?: boolean; - shouldRecordDroppedHistory?: () => boolean; - }; - }) => Promise ->(); -const dispatchPreparedSlackMessageMock = vi.fn<(prepared: unknown) => Promise>(); -const inboundDeliveryTestCache = resolveGlobalDedupeCache( - Symbol.for("openclaw.slackInboundDeliveries"), - { ttlMs: 24 * 60 * 60 * 1000, maxSize: 20_000 }, -); - -vi.mock("openclaw/plugin-sdk/channel-inbound", async () => { - const actual = await vi.importActual( - "openclaw/plugin-sdk/channel-inbound", - ); - return { - ...actual, - shouldDebounceTextInbound: () => false, - createChannelInboundDebouncer: (params: { - onFlush: ( - entries: Array<{ - message: Record; - opts: { source: "message" | "app_mention"; wasMentioned?: boolean }; - }>, - ) => Promise; - }) => ({ - debounceMs: 0, - debouncer: { - enqueue: async (entry: { - message: Record; - opts: { source: "message" | "app_mention"; wasMentioned?: boolean }; - }) => { - await params.onFlush([entry]); - }, - flushKey: async (_key: string) => {}, - }, - }), - }; -}); - -vi.mock("./thread-resolution.js", () => ({ - createSlackThreadTsResolver: () => ({ - resolve: async ({ message }: { message: Record }) => message, - }), -})); - -vi.mock("./message-handler/prepare.js", () => ({ - prepareSlackMessage: ( - params: Parameters[0], - ): ReturnType => prepareSlackMessageMock(params), -})); - -vi.mock("./message-handler/dispatch.js", () => ({ - dispatchPreparedSlackMessage: ( - prepared: Parameters[0], - ): ReturnType => - dispatchPreparedSlackMessageMock(prepared), -})); - -let createSlackMessageHandler: typeof import("./message-handler.js").createSlackMessageHandler; - -function createMarkMessageSeen() { - const seen = new Set(); - return { - markMessageSeen(channel: string | undefined, ts: string | undefined) { - if (!channel || !ts) { - return false; - } - const key = `${channel}:${ts}`; - if (seen.has(key)) { - return true; - } - seen.add(key); - return false; - }, - releaseSeenMessage(channel: string | undefined, ts: string | undefined) { - if (!channel || !ts) { - return; - } - seen.delete(`${channel}:${ts}`); - }, - }; -} - -function createTestHandler( - params: { - botUserId?: string; - channelHistories?: Map; - } = {}, -) { - const seenMessages = createMarkMessageSeen(); - return createSlackMessageHandler({ - ctx: { - cfg: {}, - accountId: "default", - botUserId: params.botUserId ?? "U_BOT", - channelHistories: params.channelHistories ?? new Map(), - app: { client: {} }, - runtime: {}, - markMessageSeen: seenMessages["markMessageSeen"], - rememberSlackChannelType: () => {}, - releaseSeenMessage: seenMessages["releaseSeenMessage"], - } as unknown as Parameters[0]["ctx"], - account: { accountId: "default" } as Parameters[0]["account"], - }); -} - -function createSlackEvent(params: { type: "message" | "app_mention"; ts: string; text: string }) { - return { type: params.type, channel: "C1", ts: params.ts, text: params.text } as never; -} - -async function sendMessageEvent(handler: ReturnType, ts: string) { - await handler(createSlackEvent({ type: "message", ts, text: "hello" }), { source: "message" }); -} - -async function sendMentionEvent(handler: ReturnType, ts: string) { - await handler(createSlackEvent({ type: "app_mention", ts, text: "<@U_BOT> hello" }), { - source: "app_mention", - wasMentioned: true, - }); -} - -async function createInFlightMessageScenario(ts: string) { - let resolveMessagePrepare: ((value: unknown) => void) | undefined; - const messagePrepare = new Promise((resolve) => { - resolveMessagePrepare = resolve; - }); - prepareSlackMessageMock.mockImplementation(async ({ opts }) => { - if (opts.source === "message") { - return messagePrepare; - } - return { ctxPayload: {} }; - }); - - const handler = createTestHandler(); - const messagePending = handler(createSlackEvent({ type: "message", ts, text: "hello" }), { - source: "message", - }); - await Promise.resolve(); - - return { handler, messagePending, resolveMessagePrepare }; -} - -describe("createSlackMessageHandler app_mention race handling", () => { - beforeAll(async () => { - ({ createSlackMessageHandler } = await import("./message-handler.js")); - }); - - beforeEach(() => { - prepareSlackMessageMock.mockReset(); - dispatchPreparedSlackMessageMock.mockReset(); - inboundDeliveryTestCache.clear(); - setSlackRuntime(null as never); - }); - - afterEach(() => { - setSlackRuntime(null as never); - vi.restoreAllMocks(); - }); - - it("allows a single app_mention retry when message event was dropped before dispatch", async () => { - prepareSlackMessageMock.mockImplementation(async ({ opts }) => { - if (opts.source === "message") { - return null; - } - return { ctxPayload: {} }; - }); - - const handler = createTestHandler(); - - await sendMessageEvent(handler, "1700000000.000100"); - await sendMentionEvent(handler, "1700000000.000100"); - await sendMentionEvent(handler, "1700000000.000100"); - - expect(prepareSlackMessageMock).toHaveBeenCalledTimes(2); - expect(dispatchPreparedSlackMessageMock).toHaveBeenCalledTimes(1); - }); - - it("prevents a late message copy from queuing after app_mention dispatch", async () => { - const channelHistories = new Map(); - let releaseMessagePrepare: (() => void) | undefined; - const messagePrepareGate = new Promise((resolve) => { - releaseMessagePrepare = resolve; - }); - prepareSlackMessageMock.mockImplementation(async ({ ctx, message, opts }) => { - if (opts.source === "message") { - await messagePrepareGate; - if (opts.shouldRecordDroppedHistory?.() !== false) { - ctx.channelHistories.set("history", [ - { sender: "Alice", body: "<@U_BOT> hello", messageId: message.ts ?? "unknown" }, - ]); - } - return null; - } - return { ctxPayload: {} }; - }); - const handler = createTestHandler({ botUserId: "", channelHistories }); - - const messagePending = sendMessageEvent(handler, "1700000000.000120"); - await Promise.resolve(); - await sendMentionEvent(handler, "1700000000.000120"); - releaseMessagePrepare?.(); - await messagePending; - - expect(channelHistories.get("history") ?? []).toEqual([]); - expect(dispatchPreparedSlackMessageMock).toHaveBeenCalledTimes(1); - }); - - it("suppresses a concurrent message copy while app_mention is preparing", async () => { - let shouldRecordDroppedHistory: (() => boolean) | undefined; - let signalMessagePrepareStarted: (() => void) | undefined; - const messagePrepareStarted = new Promise((resolve) => { - signalMessagePrepareStarted = resolve; - }); - let releaseMessagePrepare: (() => void) | undefined; - const messagePrepareGate = new Promise((resolve) => { - releaseMessagePrepare = resolve; - }); - prepareSlackMessageMock.mockImplementation(async ({ opts }) => { - if (opts.source === "message") { - shouldRecordDroppedHistory = opts.shouldRecordDroppedHistory; - signalMessagePrepareStarted?.(); - await messagePrepareGate; - return null; - } - expect(shouldRecordDroppedHistory?.()).toBe(false); - return { ctxPayload: {} }; - }); - const handler = createTestHandler({ botUserId: "" }); - - const messagePending = sendMessageEvent(handler, "1700000000.000130"); - await messagePrepareStarted; - await sendMentionEvent(handler, "1700000000.000130"); - releaseMessagePrepare?.(); - await messagePending; - - expect(dispatchPreparedSlackMessageMock).toHaveBeenCalledTimes(1); - }); - - it("does not retain app_mention retry allowance when the current clock is not a valid date timestamp", async () => { - const nowSpy = vi.spyOn(Date, "now").mockReturnValue(Number.NaN); - prepareSlackMessageMock.mockImplementation(async ({ opts }) => { - if (opts.source === "message") { - return null; - } - return { ctxPayload: {} }; - }); - - const handler = createTestHandler(); - - await sendMessageEvent(handler, "1700000000.000125"); - nowSpy.mockReturnValue(1_700_000_000_000); - await sendMentionEvent(handler, "1700000000.000125"); - - expect(prepareSlackMessageMock).toHaveBeenCalledTimes(1); - expect(dispatchPreparedSlackMessageMock).not.toHaveBeenCalled(); - }); - - it("does not retain app_mention retry allowance when the expiry timestamp would exceed the valid date range", async () => { - vi.spyOn(Date, "now").mockReturnValue(8_640_000_000_000_000); - prepareSlackMessageMock.mockImplementation(async ({ opts }) => { - if (opts.source === "message") { - return null; - } - return { ctxPayload: {} }; - }); - - const handler = createTestHandler(); - - await sendMessageEvent(handler, "1700000000.000126"); - await sendMentionEvent(handler, "1700000000.000126"); - - expect(prepareSlackMessageMock).toHaveBeenCalledTimes(1); - expect(dispatchPreparedSlackMessageMock).not.toHaveBeenCalled(); - }); - - it("allows app_mention while message handling is still in-flight, then keeps later duplicates deduped", async () => { - const { handler, messagePending, resolveMessagePrepare } = - await createInFlightMessageScenario("1700000000.000150"); - - await sendMentionEvent(handler, "1700000000.000150"); - - resolveMessagePrepare?.(null); - await messagePending; - - await sendMentionEvent(handler, "1700000000.000150"); - - expect(prepareSlackMessageMock).toHaveBeenCalledTimes(2); - expect(dispatchPreparedSlackMessageMock).toHaveBeenCalledTimes(1); - }); - - it("suppresses message dispatch when app_mention already dispatched during in-flight race", async () => { - const { handler, messagePending, resolveMessagePrepare } = - await createInFlightMessageScenario("1700000000.000175"); - - await sendMentionEvent(handler, "1700000000.000175"); - - resolveMessagePrepare?.({ ctxPayload: {} }); - await messagePending; - - expect(prepareSlackMessageMock).toHaveBeenCalledTimes(2); - expect(dispatchPreparedSlackMessageMock).toHaveBeenCalledTimes(1); - }); - - it("keeps app_mention deduped when message event already dispatched", async () => { - prepareSlackMessageMock.mockResolvedValue({ ctxPayload: {} }); - - const handler = createTestHandler(); - - await sendMessageEvent(handler, "1700000000.000200"); - await sendMentionEvent(handler, "1700000000.000200"); - - expect(prepareSlackMessageMock).toHaveBeenCalledTimes(1); - expect(dispatchPreparedSlackMessageMock).toHaveBeenCalledTimes(1); - }); - - it("keeps message replay deduped after a non-retryable dispatch failure", async () => { - prepareSlackMessageMock.mockResolvedValue({ ctxPayload: {} }); - dispatchPreparedSlackMessageMock.mockRejectedValueOnce(new Error("post-send failure")); - - const handler = createTestHandler(); - - await expect(sendMessageEvent(handler, "1700000000.000300")).rejects.toThrow( - "post-send failure", - ); - await sendMessageEvent(handler, "1700000000.000300"); - - expect(prepareSlackMessageMock).toHaveBeenCalledTimes(1); - expect(dispatchPreparedSlackMessageMock).toHaveBeenCalledTimes(1); - }); - - it("dedupes delayed app_mention replays through persistent delivery state", async () => { - const stored = new Map(); - const register = vi.fn(async (key: string, value: unknown) => { - stored.set(key, value); - }); - const lookup = vi.fn(async (key: string) => stored.get(key)); - setSlackRuntime({ - state: { - openKeyedStore: vi.fn(() => ({ - register, - lookup, - consume: vi.fn(), - delete: vi.fn(), - entries: vi.fn(), - clear: vi.fn(), - })), - }, - logging: { getChildLogger: () => ({ warn: vi.fn() }) }, - } as never); - prepareSlackMessageMock.mockResolvedValue({ ctxPayload: {} }); - - await sendMessageEvent(createTestHandler(), "1700000000.000350"); - inboundDeliveryTestCache.clear(); - await sendMentionEvent(createTestHandler(), "1700000000.000350"); - - expect(register).toHaveBeenCalledWith("default:C1:1700000000.000350", { - deliveredAt: expect.any(Number), - }); - expect(lookup).toHaveBeenCalledWith("default:C1:1700000000.000350"); - expect(prepareSlackMessageMock).toHaveBeenCalledTimes(1); - expect(dispatchPreparedSlackMessageMock).toHaveBeenCalledTimes(1); - }); -}); diff --git a/extensions/slack/src/monitor/message-handler.test.ts b/extensions/slack/src/monitor/message-handler.test.ts index 0bd9236ccc94..7048e3bfe369 100644 --- a/extensions/slack/src/monitor/message-handler.test.ts +++ b/extensions/slack/src/monitor/message-handler.test.ts @@ -5,9 +5,7 @@ const enqueueMock = vi.fn(async (_entry: unknown) => {}); const flushKeyMock = vi.fn(async (_key: string) => {}); const onFlushCallbacks: Array<(entries: Array>) => Promise> = []; const prepareSlackMessageMock = vi.fn(async () => ({ ctxPayload: {} })); -const dispatchPreparedSlackMessageMock = vi.fn(async () => {}); -const hasSlackInboundMessageDeliveryMock = vi.fn(async () => false); -const recordSlackInboundMessageDeliveriesMock = vi.fn(async () => {}); +const dispatchPreparedSlackMessageMock = vi.fn(async (_prepared: unknown) => {}); const resolveThreadTsMock = vi.fn(async ({ message }: { message: Record }) => ({ ...message, })); @@ -46,18 +44,11 @@ vi.mock("./message-handler/pipeline.runtime.js", () => ({ dispatchPreparedSlackMessage: dispatchPreparedSlackMessageMock, })); -vi.mock("./inbound-delivery-state.js", () => ({ - hasSlackInboundMessageDelivery: hasSlackInboundMessageDeliveryMock, - recordSlackInboundMessageDeliveries: recordSlackInboundMessageDeliveriesMock, -})); - function createContext(overrides?: { - markMessageSeen?: (channel: string | undefined, ts: string | undefined) => boolean; rememberSlackChannelType?: ( channel: string | null | undefined, channelType: string | null | undefined, ) => void; - releaseSeenMessage?: (channel: string | undefined, ts: string | undefined) => void; }) { return { cfg: {}, @@ -66,24 +57,18 @@ function createContext(overrides?: { client: {}, }, runtime: {}, - markMessageSeen: (channel: string | undefined, ts: string | undefined) => - overrides?.markMessageSeen?.(channel, ts) ?? false, rememberSlackChannelType: ( channel: string | null | undefined, channelType: string | null | undefined, ) => overrides?.rememberSlackChannelType?.(channel, channelType), - releaseSeenMessage: (channel: string | undefined, ts: string | undefined) => - overrides?.releaseSeenMessage?.(channel, ts), } as Parameters[0]["ctx"]; } function createHandlerWithTracker(overrides?: { - markMessageSeen?: (channel: string | undefined, ts: string | undefined) => boolean; rememberSlackChannelType?: ( channel: string | null | undefined, channelType: string | null | undefined, ) => void; - releaseSeenMessage?: (channel: string | undefined, ts: string | undefined) => void; }) { const trackEvent = vi.fn(); const handler = createSlackMessageHandler({ @@ -115,9 +100,6 @@ describe("createSlackMessageHandler", () => { onFlushCallbacks.length = 0; prepareSlackMessageMock.mockClear(); dispatchPreparedSlackMessageMock.mockClear(); - hasSlackInboundMessageDeliveryMock.mockReset(); - hasSlackInboundMessageDeliveryMock.mockResolvedValue(false); - recordSlackInboundMessageDeliveriesMock.mockClear(); resolveThreadTsMock.mockClear(); }); @@ -145,17 +127,7 @@ describe("createSlackMessageHandler", () => { expect(enqueueMock).not.toHaveBeenCalled(); }); - it("does not track duplicate messages that are already seen", async () => { - const { handler, trackEvent } = createHandlerWithTracker({ markMessageSeen: () => true }); - - await handleDirectMessage(handler); - - expect(trackEvent).not.toHaveBeenCalled(); - expect(resolveThreadTsMock).not.toHaveBeenCalled(); - expect(enqueueMock).not.toHaveBeenCalled(); - }); - - it("tracks accepted non-duplicate messages", async () => { + it("tracks accepted messages", async () => { const { handler, trackEvent } = createHandlerWithTracker(); await handleDirectMessage(handler); @@ -165,13 +137,15 @@ describe("createSlackMessageHandler", () => { expect(enqueueMock).toHaveBeenCalledTimes(1); }); - it("records explicit channel type before the first delivery-state await", async () => { - let settleDeliveryLookup: ((delivered: boolean) => void) | undefined; - hasSlackInboundMessageDeliveryMock.mockImplementationOnce( - async () => - await new Promise((resolve) => { - settleDeliveryLookup = resolve; - }), + it("records explicit channel type before thread resolution", async () => { + let settleThreadResolution: (() => void) | undefined; + resolveThreadTsMock.mockImplementationOnce( + async ({ message }: { message: Record }) => { + await new Promise((resolve) => { + settleThreadResolution = resolve; + }); + return { ...message }; + }, ); const rememberSlackChannelType = vi.fn(); const { handler } = createHandlerWithTracker({ rememberSlackChannelType }); @@ -189,7 +163,7 @@ describe("createSlackMessageHandler", () => { expect(rememberSlackChannelType).toHaveBeenCalledWith("C0MPDM42", "mpim"); expect(enqueueMock).not.toHaveBeenCalled(); - settleDeliveryLookup?.(false); + settleThreadResolution?.(); await handled; expect(enqueueMock).toHaveBeenCalledOnce(); }); @@ -296,6 +270,84 @@ describe("createSlackMessageHandler", () => { expect(dispatchPreparedSlackMessageMock).toHaveBeenCalledTimes(1); }); + it("carries durable ingress ownership into prepared dispatch", async () => { + const turnAdoptionLifecycle = { + admission: "exclusive" as const, + abortSignal: new AbortController().signal, + onAdopted: vi.fn(), + onDeferred: vi.fn(), + onAbandoned: vi.fn(), + }; + const { handler } = createHandlerWithTracker(); + const handled = handler( + { + type: "message", + channel: "C111", + user: "U111", + ts: "1709000000.000550", + text: "durable message", + } as never, + { source: "message", awaitDispatch: true, turnAdoptionLifecycle }, + ); + + await vi.waitFor(() => expect(enqueueMock).toHaveBeenCalledTimes(1)); + const entry = enqueueMock.mock.calls[0]?.[0] as Record; + await onFlushCallbacks[0]?.([entry]); + await handled; + + // The flush wraps the lifecycle to settle dispatch-dedupe claims, so assert + // ownership forwarding rather than function identity. + expect(dispatchPreparedSlackMessageMock).toHaveBeenCalledTimes(1); + const prepared = dispatchPreparedSlackMessageMock.mock.calls[0]?.[0] as { + turnAdoptionLifecycle?: typeof turnAdoptionLifecycle; + }; + expect(prepared.turnAdoptionLifecycle?.admission).toBe("exclusive"); + expect(prepared.turnAdoptionLifecycle?.abortSignal).toBe(turnAdoptionLifecycle.abortSignal); + await prepared.turnAdoptionLifecycle?.onAdopted(); + expect(turnAdoptionLifecycle.onAdopted).toHaveBeenCalledTimes(1); + prepared.turnAdoptionLifecycle?.onDeferred(); + expect(turnAdoptionLifecycle.onDeferred).toHaveBeenCalledTimes(1); + }); + + it("dispatches a message/app_mention twin pair exactly once", async () => { + // Slack emits both events with distinct event_ids for one mention post, so + // the durable ingress queue admits both; the logical (channel, ts) dispatch + // guard must collapse them to a single dispatch. + const { handler } = createHandlerWithTracker(); + const twinTs = "1709000000.000777"; + const asMessage = handler( + { + type: "message", + channel: "C111", + user: "U111", + ts: twinTs, + text: "<@UBOT> hello", + } as never, + { source: "message", awaitDispatch: true }, + ); + await vi.waitFor(() => expect(enqueueMock).toHaveBeenCalledTimes(1)); + const first = enqueueMock.mock.calls[0]?.[0] as Record; + await onFlushCallbacks[0]?.([first]); + await asMessage; + expect(dispatchPreparedSlackMessageMock).toHaveBeenCalledTimes(1); + + const asMention = handler( + { + type: "message", + channel: "C111", + user: "U111", + ts: twinTs, + text: "<@UBOT> hello", + } as never, + { source: "app_mention", wasMentioned: true, awaitDispatch: true }, + ); + await vi.waitFor(() => expect(enqueueMock).toHaveBeenCalledTimes(2)); + const second = enqueueMock.mock.calls[1]?.[0] as Record; + await onFlushCallbacks[0]?.([second]); + await asMention; + expect(dispatchPreparedSlackMessageMock).toHaveBeenCalledTimes(1); + }); + it("propagates debounced dispatch failures to relay delivery", async () => { dispatchPreparedSlackMessageMock.mockRejectedValueOnce(new Error("dispatch failed")); const { handler } = createHandlerWithTracker(); @@ -317,8 +369,7 @@ describe("createSlackMessageHandler", () => { await Promise.all([handledFailure, flushFailure]); }); - it("retries native session initialization conflicts through the delivery gates", async () => { - const releaseSeenMessage = vi.fn(); + it("retries native session initialization conflicts", async () => { dispatchPreparedSlackMessageMock.mockRejectedValueOnce( new Error("Slack dispatch failed", { cause: new Error( @@ -326,7 +377,7 @@ describe("createSlackMessageHandler", () => { ), }), ); - const { handler } = createHandlerWithTracker({ releaseSeenMessage }); + const { handler } = createHandlerWithTracker(); await handler( { type: "message", @@ -344,9 +395,6 @@ describe("createSlackMessageHandler", () => { await expect(onFlushCallbacks[0]?.([entry])).rejects.toThrow("Slack dispatch failed"); await vi.advanceTimersByTimeAsync(1000); - expect(releaseSeenMessage).toHaveBeenCalledWith("C111", "1709000000.000700"); - expect(recordSlackInboundMessageDeliveriesMock).not.toHaveBeenCalled(); - expect(hasSlackInboundMessageDeliveryMock).toHaveBeenCalledTimes(2); expect(enqueueMock).toHaveBeenCalledTimes(2); expect(enqueueMock.mock.calls[1]?.[0]).toMatchObject({ opts: { @@ -360,7 +408,6 @@ describe("createSlackMessageHandler", () => { }); it("leaves relay session conflict retries to unacknowledged redelivery", async () => { - const releaseSeenMessage = vi.fn(); dispatchPreparedSlackMessageMock.mockRejectedValueOnce( new Error("Slack dispatch failed", { cause: new Error( @@ -368,7 +415,7 @@ describe("createSlackMessageHandler", () => { ), }), ); - const { handler } = createHandlerWithTracker({ releaseSeenMessage }); + const { handler } = createHandlerWithTracker(); const handled = handler( { type: "message", @@ -391,60 +438,6 @@ describe("createSlackMessageHandler", () => { await Promise.all([handledFailure, flushFailure]); await vi.advanceTimersByTimeAsync(1000); - expect(releaseSeenMessage).toHaveBeenCalledWith("C111", "1709000000.000800"); - expect(recordSlackInboundMessageDeliveriesMock).not.toHaveBeenCalled(); - expect(enqueueMock).toHaveBeenCalledTimes(1); - } finally { - vi.useRealTimers(); - } - }); - - it("settles an already-delivered relay event without enqueueing", async () => { - hasSlackInboundMessageDeliveryMock.mockResolvedValueOnce(true); - const { handler } = createHandlerWithTracker(); - - await expect( - handler( - { - type: "message", - channel: "C111", - user: "U111", - ts: "1709000000.000850", - text: "relay replay", - } as never, - { source: "message", awaitDispatch: true }, - ), - ).resolves.toBeUndefined(); - - expect(enqueueMock).not.toHaveBeenCalled(); - }); - - it("skips a native retry when another delivery already succeeded", async () => { - dispatchPreparedSlackMessageMock.mockRejectedValueOnce( - new Error("reply session initialization conflicted for agent:main:main:thread:123.456"), - ); - hasSlackInboundMessageDeliveryMock.mockResolvedValueOnce(false).mockResolvedValueOnce(true); - const { handler } = createHandlerWithTracker(); - await handler( - { - type: "message", - channel: "C111", - user: "U111", - ts: "1709000000.000900", - text: "native message", - } as never, - { source: "message" }, - ); - - const entry = enqueueMock.mock.calls[0]?.[0] as Record; - vi.useFakeTimers(); - try { - await expect(onFlushCallbacks[0]?.([entry])).rejects.toThrow( - "reply session initialization conflicted", - ); - await vi.advanceTimersByTimeAsync(1000); - - expect(hasSlackInboundMessageDeliveryMock).toHaveBeenCalledTimes(2); expect(enqueueMock).toHaveBeenCalledTimes(1); } finally { vi.useRealTimers(); diff --git a/extensions/slack/src/monitor/message-handler.ts b/extensions/slack/src/monitor/message-handler.ts index 32432f028bf4..642d27852df5 100644 --- a/extensions/slack/src/monitor/message-handler.ts +++ b/extensions/slack/src/monitor/message-handler.ts @@ -5,20 +5,20 @@ import { } from "openclaw/plugin-sdk/channel-inbound"; import { collectErrorGraphCandidates, formatErrorMessage } from "openclaw/plugin-sdk/error-runtime"; import { createLazyRuntimeModule } from "openclaw/plugin-sdk/lazy-runtime"; -import { - asDateTimestampMs, - resolveExpiresAtMsFromDurationMs, -} from "openclaw/plugin-sdk/number-runtime"; import type { ResolvedSlackAccount } from "../accounts.js"; import type { SlackSendIdentity } from "../send.js"; import type { SlackMessageEvent } from "../types.js"; import { stripSlackMentionsForCommandDetection } from "./commands.js"; import type { SlackMonitorContext } from "./context.js"; import type { SlackEventScope } from "./event-scope.js"; +import type { SlackIngressTurnLifecycle } from "./ingress.js"; import { - hasSlackInboundMessageDelivery, - recordSlackInboundMessageDeliveries, -} from "./inbound-delivery-state.js"; + buildSlackMessageDispatchReplayKey, + claimSlackMessageDispatchReplay, + createSlackMessageDispatchReplayGuard, + type SlackMessageDispatchReplayClaim, + type SlackMessageDispatchReplayGuard, +} from "./message-dispatch-dedupe.js"; import { buildSlackDebounceKey, buildTopLevelSlackConversationKey, @@ -40,6 +40,8 @@ export type SlackMessageHandler = ( eventScope?: SlackEventScope; /** Wait until any inbound debounce flush and dispatch has completed. */ awaitDispatch?: boolean; + /** Durable ingress ownership carried into reply-lane adoption. */ + turnAdoptionLifecycle?: SlackIngressTurnLifecycle; }, ) => Promise; @@ -67,7 +69,6 @@ function createSlackDispatchCompletion(): SlackDispatchCompletion { return { promise, resolve, reject }; } -const APP_MENTION_RETRY_TTL_MS = 60_000; const RETRYABLE_FLUSH_MAX_ATTEMPTS = 3; const RETRYABLE_FLUSH_RETRY_DELAY_MS = 1_000; const REPLY_SESSION_INIT_CONFLICT_MESSAGE_RE = /reply session initialization conflicted for \S+/u; @@ -88,17 +89,6 @@ function shouldDebounceSlackMessage(message: SlackMessageEvent, cfg: SlackMonito }); } -function buildSeenMessageKey( - channelId: string | undefined, - ts: string | undefined, - teamId?: string, -): string | null { - if (!channelId || !ts) { - return null; - } - return `${teamId ? `${teamId}:` : ""}${channelId}:${ts}`; -} - export function createSlackMessageHandler(params: { ctx: SlackMonitorContext; account: ResolvedSlackAccount; @@ -106,8 +96,17 @@ export function createSlackMessageHandler(params: { trackEvent?: () => void; /** Called after access/routing preparation accepts a human message. */ onPrepared?: (prepared: PreparedSlackMessage) => void; + dispatchReplayGuard?: SlackMessageDispatchReplayGuard; }): SlackMessageHandler { const { ctx, account, trackEvent, onPrepared } = params; + const dispatchReplayGuard = + params.dispatchReplayGuard ?? + createSlackMessageDispatchReplayGuard({ + onDiskError: (error) => + ctx.runtime.error?.( + `slack message dispatch dedupe persistence failed: ${formatErrorMessage(error)}`, + ), + }); const { debounceMs, debouncer } = createChannelInboundDebouncer<{ message: SlackMessageEvent; opts: QueuedSlackMessageOptions; @@ -152,8 +151,7 @@ export function createSlackMessageHandler(params: { } const retryTimer = setTimeout(() => { for (const entry of nextEntries) { - // Re-enter ingress so a relay replay or another successful attempt wins - // through the normal delivery and seen-message gates before dispatch. + // Re-enter the normal inbound path so retry ordering and debouncing stay consistent. void enqueueSlackMessage(entry.message, entry.opts).catch((err: unknown) => { ctx.runtime.error?.(`slack inbound retry enqueue failed: ${formatErrorMessage(err)}`); }); @@ -167,8 +165,47 @@ export function createSlackMessageHandler(params: { .filter((completion) => completion !== undefined); try { await (async () => { - const last = entries.at(-1); + // Logical-identity claims: Slack sends message + app_mention twins with + // distinct event_ids for one post, so the durable queue cannot dedupe + // them. Same-flush twins share one claim; a later twin claims duplicate + // and is dropped before it can produce a second visible reply. + const claims: SlackMessageDispatchReplayClaim[] = []; + const claimedKeys = new Set(); + const surviving: typeof entries = []; + for (const entry of entries) { + const replayKey = buildSlackMessageDispatchReplayKey({ + accountId: ctx.accountId, + channelId: entry.message.channel, + ts: entry.message.ts, + teamId: entry.opts.eventScope?.teamId, + }); + if (!replayKey || claimedKeys.has(replayKey)) { + surviving.push(entry); + continue; + } + const claim = await claimSlackMessageDispatchReplay({ + guard: dispatchReplayGuard, + key: replayKey, + }); + if (claim.kind === "claimed") { + claims.push(claim.handle); + claimedKeys.add(replayKey); + surviving.push(entry); + } + } + const releaseClaims = (error?: unknown) => { + for (const handle of claims) { + handle.release(error === undefined ? {} : { error }); + } + }; + const commitClaims = async () => { + for (const handle of claims) { + await handle.commit(); + } + }; + const last = surviving.at(-1); if (!last) { + releaseClaims(); return; } const teamId = last.opts.eventScope?.teamId; @@ -188,128 +225,82 @@ export function createSlackMessageHandler(params: { } } const combinedText = - entries.length === 1 + surviving.length === 1 ? (last.message.text ?? "") - : entries + : surviving .map((entry) => entry.message.text ?? "") .filter(Boolean) .join("\n"); - const combinedMentioned = entries.some((entry) => Boolean(entry.opts.wasMentioned)); + const combinedMentioned = surviving.some((entry) => Boolean(entry.opts.wasMentioned)); const syntheticMessage: SlackMessageEvent = { ...last.message, text: combinedText, }; - const seenMessageKey = buildSeenMessageKey( - last.message.channel, - last.message.ts, - last.opts.eventScope?.teamId, - ); + const { prepareSlackMessage, dispatchPreparedSlackMessage } = + await loadSlackMessagePipeline(); + const { + dispatchCompletion: _completion, + awaitDispatch: _awaitDispatch, + turnAdoptionLifecycle, + ...lastOpts + } = last.opts; + let prepared: Awaited>; + let settlementHandedOff = false; try { - const { prepareSlackMessage, dispatchPreparedSlackMessage } = - await loadSlackMessagePipeline(); - const { - dispatchCompletion: _completion, - awaitDispatch: _awaitDispatch, - ...lastOpts - } = last.opts; - const appMentionRetryKey = - seenMessageKey && lastOpts.source === "app_mention" && !ctx.botUserId - ? seenMessageKey - : undefined; - if (appMentionRetryKey) { - // Keep a concurrent message copy from recording this timestamp while the trusted - // app_mention prepares and removes any already-recorded copy from its routed history. - appMentionPreparingKeys.add(appMentionRetryKey); - } - const prepared = await (async () => { - try { - const result = await prepareSlackMessage({ - ctx, - account, - message: syntheticMessage, - opts: { - ...lastOpts, - wasMentioned: combinedMentioned || last.opts.wasMentioned, - ...(seenMessageKey && lastOpts.source === "message" - ? { - shouldRecordDroppedHistory: () => - !appMentionPreparingKeys.has(seenMessageKey) && - !appMentionDispatchedKeys.has(seenMessageKey), - } - : {}), - }, - }); - if (result && seenMessageKey) { - pruneAppMentionRetryKeys(Date.now()); - if (last.opts.source === "app_mention") { - // If app_mention wins the race and dispatches first, drop the later message. - rememberExpiringAppMentionKey(appMentionDispatchedKeys, seenMessageKey); - } else if ( - last.opts.source === "message" && - appMentionDispatchedKeys.has(seenMessageKey) - ) { - appMentionDispatchedKeys.delete(seenMessageKey); - appMentionRetryKeys.delete(seenMessageKey); - return null; - } - appMentionRetryKeys.delete(seenMessageKey); - } - return result; - } finally { - if (appMentionRetryKey) { - appMentionPreparingKeys.delete(appMentionRetryKey); - } - } - })(); + prepared = await prepareSlackMessage({ + ctx, + account, + message: syntheticMessage, + opts: { + ...lastOpts, + wasMentioned: combinedMentioned || last.opts.wasMentioned, + }, + }); if (!prepared) { + // Gated before dispatch: release so the surviving twin can run the + // same gate; nothing visible was produced, so no duplicate risk. + releaseClaims(); return; } + // Commit at adoption (durable turn ownership), release on abandonment; + // deferred turns hand settlement to the reply lane with the claim held. + prepared.turnAdoptionLifecycle = turnAdoptionLifecycle + ? { + ...turnAdoptionLifecycle, + onAdopted: async () => { + settlementHandedOff = true; + await commitClaims(); + await turnAdoptionLifecycle.onAdopted(); + }, + onDeferred: () => { + settlementHandedOff = true; + turnAdoptionLifecycle.onDeferred(); + }, + onAbandoned: () => { + releaseClaims(); + turnAdoptionLifecycle.onAbandoned(); + }, + } + : turnAdoptionLifecycle; onPrepared?.(prepared); - if (entries.length > 1) { - const ids = entries.map((entry) => entry.message.ts).filter(Boolean) as string[]; + if (surviving.length > 1) { + const ids = surviving.map((entry) => entry.message.ts).filter(Boolean) as string[]; if (ids.length > 0) { prepared.ctxPayload.MessageSids = ids; prepared.ctxPayload.MessageSidFirst = ids[0]; prepared.ctxPayload.MessageSidLast = ids[ids.length - 1]; } } - try { - await dispatchPreparedSlackMessage(prepared); - await recordSlackInboundMessageDeliveries({ - accountId: ctx.accountId, - ...(last.opts.eventScope ? { teamId: last.opts.eventScope.teamId } : {}), - messages: entries.map((entry) => entry.message), - }); - } catch (error) { - if (!isRetryableSlackInboundError(error)) { - await recordSlackInboundMessageDeliveries({ - accountId: ctx.accountId, - ...(last.opts.eventScope ? { teamId: last.opts.eventScope.teamId } : {}), - messages: entries.map((entry) => entry.message), - }); - } - throw error; + await dispatchPreparedSlackMessage(prepared); + if (!turnAdoptionLifecycle) { + await commitClaims(); + } else if (!settlementHandedOff) { + // Dispatch finished without adoption or deferral (skip/no-reply): + // deliberate terminal handling, release for gate-idempotent twins. + releaseClaims(); } } catch (error) { - if (isRetryableSlackInboundError(error)) { - // Every buffered event passed the seen gate before this combined dispatch. - // Release all of them so the retry can rebuild the same batch. - for (const entry of entries) { - const entrySeenKey = buildSeenMessageKey( - entry.message.channel, - entry.message.ts, - entry.opts.eventScope?.teamId, - ); - if (entrySeenKey) { - appMentionDispatchedKeys.delete(entrySeenKey); - } - ctx.releaseSeenMessage( - entry.message.channel, - entry.message.ts, - entry.opts.eventScope, - ); - } - } + releaseClaims(error); throw error; } })(); @@ -330,56 +321,6 @@ export function createSlackMessageHandler(params: { }); const threadTsResolver = createSlackThreadTsResolver({ client: ctx.app.client }); const pendingTopLevelDebounceKeys = new Map>(); - const appMentionRetryKeys = new Map(); - const appMentionPreparingKeys = new Set(); - const appMentionDispatchedKeys = new Map(); - - const pruneAppMentionRetryKeys = (rawNow: number): boolean => { - const now = asDateTimestampMs(rawNow); - if (now === undefined) { - appMentionRetryKeys.clear(); - appMentionDispatchedKeys.clear(); - return false; - } - for (const [key, expiresAt] of appMentionRetryKeys) { - if (asDateTimestampMs(expiresAt) === undefined || expiresAt <= now) { - appMentionRetryKeys.delete(key); - } - } - for (const [key, expiresAt] of appMentionDispatchedKeys) { - if (asDateTimestampMs(expiresAt) === undefined || expiresAt <= now) { - appMentionDispatchedKeys.delete(key); - } - } - return true; - }; - - const rememberExpiringAppMentionKey = (map: Map, key: string): void => { - const now = Date.now(); - if (!pruneAppMentionRetryKeys(now)) { - return; - } - const expiresAt = resolveExpiresAtMsFromDurationMs(APP_MENTION_RETRY_TTL_MS, { nowMs: now }); - if (expiresAt !== undefined) { - map.set(key, expiresAt); - } - }; - - const rememberAppMentionRetryKey = (key: string) => { - rememberExpiringAppMentionKey(appMentionRetryKeys, key); - }; - - const consumeAppMentionRetryKey = (key: string) => { - const now = Date.now(); - if (!pruneAppMentionRetryKeys(now)) { - return false; - } - if (!appMentionRetryKeys.has(key)) { - return false; - } - appMentionRetryKeys.delete(key); - return true; - }; async function enqueueSlackMessage( message: SlackMessageEvent, @@ -397,40 +338,9 @@ export function createSlackMessageHandler(params: { ) { return undefined; } - // Record Slack's explicit type before delivery-state or thread-resolution awaits. + // Record Slack's explicit type before thread-resolution awaits. // Relay and native events can overlap; a following typeless bot event must see it. ctx.rememberSlackChannelType(message.channel, message.channel_type, opts.eventScope); - const seenMessageKey = buildSeenMessageKey( - message.channel, - message.ts, - opts.eventScope?.teamId, - ); - if ( - seenMessageKey && - (await hasSlackInboundMessageDelivery({ - accountId: ctx.accountId, - channelId: message.channel, - ts: message.ts, - ...(opts.eventScope ? { teamId: opts.eventScope.teamId } : {}), - })) - ) { - return undefined; - } - const wasSeen = seenMessageKey - ? ctx.markMessageSeen(message.channel, message.ts, opts.eventScope) - : false; - if (seenMessageKey && opts.source === "message" && !wasSeen) { - // Prime exactly one fallback app_mention allowance immediately so a near-simultaneous - // app_mention is not dropped while message handling is still in-flight. - rememberAppMentionRetryKey(seenMessageKey); - } - if (seenMessageKey && wasSeen) { - // Allow exactly one app_mention retry if the same ts was previously dropped - // from the message stream before it reached dispatch. - if (opts.source !== "app_mention" || !consumeAppMentionRetryKey(seenMessageKey)) { - return undefined; - } - } trackEvent?.(); const resolvedMessage = await ( opts.eventScope diff --git a/extensions/slack/src/monitor/message-handler/dispatch.preview-fallback.test.ts b/extensions/slack/src/monitor/message-handler/dispatch.preview-fallback.test.ts index f8d76440ad31..72f43a41fa4b 100644 --- a/extensions/slack/src/monitor/message-handler/dispatch.preview-fallback.test.ts +++ b/extensions/slack/src/monitor/message-handler/dispatch.preview-fallback.test.ts @@ -56,6 +56,7 @@ let capturedReplyOptions: onVerboseProgressVisibility?: (isActive: () => boolean) => void; allowProgressCallbacksWhenSourceDeliverySuppressed?: boolean; allowToolLifecycleWhenProgressHidden?: boolean; + turnAdoptionLifecycle?: object; onAssistantMessageStart?: () => Promise | void; onReasoningEnd?: () => Promise | void; onReasoningStream?: (payload?: { @@ -387,6 +388,7 @@ function createPreparedSlackMessage(params?: { ackReactionMessageTs?: string; ackReactionPromise?: Promise | null; relayIdentity?: { username?: string; iconUrl?: string; iconEmoji?: string }; + turnAdoptionLifecycle?: object; eventScope?: { apiAppId: string; enterpriseId: string; @@ -429,6 +431,7 @@ function createPreparedSlackMessage(params?: { config: params?.accountConfig ?? {}, }, relayIdentity: params?.relayIdentity, + turnAdoptionLifecycle: params?.turnAdoptionLifecycle, eventScope: params?.eventScope, message, route: { @@ -1205,6 +1208,20 @@ describe("dispatchPreparedSlackMessage preview fallback", () => { emitSlackMessageSentHooksMock.mockClear(); }); + it("forwards durable ingress ownership into reply options", async () => { + const turnAdoptionLifecycle = { + admission: "exclusive", + abortSignal: new AbortController().signal, + onAdopted: vi.fn(), + onDeferred: vi.fn(), + onAbandoned: vi.fn(), + }; + + await dispatchPreparedSlackMessage(createPreparedSlackMessage({ turnAdoptionLifecycle })); + + expect(capturedReplyOptions?.turnAdoptionLifecycle).toBe(turnAdoptionLifecycle); + }); + it("falls back to normal delivery when preview finalize fails", async () => { await dispatchPreparedSlackMessage(createPreparedSlackMessage()); diff --git a/extensions/slack/src/monitor/message-handler/dispatch.ts b/extensions/slack/src/monitor/message-handler/dispatch.ts index 466c018ba4f5..a9c976502690 100644 --- a/extensions/slack/src/monitor/message-handler/dispatch.ts +++ b/extensions/slack/src/monitor/message-handler/dispatch.ts @@ -2090,6 +2090,9 @@ export async function dispatchPreparedSlackMessage(prepared: PreparedSlackMessag history: prepared.turn.history, botLoopProtection: resolveSlackBotLoopProtection(prepared), replyOptions: { + ...(prepared.turnAdoptionLifecycle + ? { turnAdoptionLifecycle: prepared.turnAdoptionLifecycle } + : {}), skillFilter: prepared.channelConfig?.skills, sourceReplyDeliveryMode, // Room events are observe-style turns; Slack status indicators imply an diff --git a/extensions/slack/src/monitor/message-handler/prepare.test.ts b/extensions/slack/src/monitor/message-handler/prepare.test.ts index ac9fc093f4ff..169dda7c7b91 100644 --- a/extensions/slack/src/monitor/message-handler/prepare.test.ts +++ b/extensions/slack/src/monitor/message-handler/prepare.test.ts @@ -4320,8 +4320,6 @@ describe("prepareSlackMessage sender prefix", () => { mediaMaxBytes: 1000, removeAckAfterReply: false, logger: { info: vi.fn(), warn: vi.fn() }, - markMessageSeen: () => false, - releaseSeenMessage: () => {}, shouldDropMismatchedSlackEvent: () => false, resolveSlackSystemEventSessionKey: () => "agent:main:slack:channel:c1", isChannelAllowed: () => true, diff --git a/extensions/slack/src/monitor/message-handler/types.ts b/extensions/slack/src/monitor/message-handler/types.ts index ffde2370ea29..1923e0e7f9d1 100644 --- a/extensions/slack/src/monitor/message-handler/types.ts +++ b/extensions/slack/src/monitor/message-handler/types.ts @@ -9,6 +9,7 @@ import type { SlackMessageEvent } from "../../types.js"; import type { SlackChannelConfigResolved } from "../channel-config.js"; import type { SlackMonitorContext } from "../context.js"; import type { SlackEventScope } from "../event-scope.js"; +import type { SlackIngressTurnLifecycle } from "../ingress.js"; export type PreparedSlackMessage = { ctx: SlackMonitorContext; @@ -16,6 +17,7 @@ export type PreparedSlackMessage = { message: SlackMessageEvent; relayIdentity?: SlackSendIdentity; eventScope?: SlackEventScope; + turnAdoptionLifecycle?: SlackIngressTurnLifecycle; route: ResolvedAgentRoute; channelConfig: SlackChannelConfigResolved | null; replyTarget: string; diff --git a/extensions/slack/src/monitor/provider-support.ts b/extensions/slack/src/monitor/provider-support.ts index f8b0d4ec5bde..ac331146f46a 100644 --- a/extensions/slack/src/monitor/provider-support.ts +++ b/extensions/slack/src/monitor/provider-support.ts @@ -314,6 +314,7 @@ export function createSlackBoltApp(params: { slackWebhookPath: string; clientOptions: Record; socketMode?: SlackSocketModeConfig; + wrapReceiver?: (receiver: SlackReceiver) => SlackReceiver; }) { const socketModeLogger = createSlackSocketModeLogger(); const socketModeReceiverOptions: SlackSocketModeReceiverOptions = { @@ -325,6 +326,7 @@ export function createSlackBoltApp(params: { installerOptions: { clientOptions: params.clientOptions, }, + ...(params.wrapReceiver ? { processEventErrorHandler: async () => false } : {}), }; if (params.socketMode?.serverPingTimeout !== undefined) { socketModeReceiverOptions.serverPingTimeout = params.socketMode.serverPingTimeout; @@ -345,10 +347,12 @@ export function createSlackBoltApp(params: { receiver = new params.interop.HTTPReceiver({ signingSecret: params.signingSecret ?? "", endpoints: params.slackWebhookPath, + ...(params.wrapReceiver ? { processEventErrorHandler: async () => false } : {}), }); } else { receiver = createSlackRelayReceiver(); } + const appReceiver = receiver && params.wrapReceiver ? params.wrapReceiver(receiver) : receiver; const app = new params.interop.App({ token: params.botToken, clientOptions: params.clientOptions, @@ -357,7 +361,7 @@ export function createSlackBoltApp(params: { // verification is enabled. Invalid tokens can reject before any listener // consumes that promise, tripping OpenClaw's fatal unhandled-rejection path. tokenVerificationEnabled: false, - ...(receiver ? { receiver } : {}), + ...(appReceiver ? { receiver: appReceiver } : {}), }); app.use(async (args) => { if (shouldSkipOpenClawSlackSelfEvent(args)) { diff --git a/extensions/slack/src/monitor/provider.interop.test.ts b/extensions/slack/src/monitor/provider.interop.test.ts index 0e4bde699c00..1e0ed286fa8a 100644 --- a/extensions/slack/src/monitor/provider.interop.test.ts +++ b/extensions/slack/src/monitor/provider.interop.test.ts @@ -410,6 +410,37 @@ describe("createSlackBoltApp", () => { expect((app as unknown as FakeApp).middleware).toHaveLength(1); }); + it.each(["socket", "http"] as const)( + "routes %s Events API receive through the durable receiver wrapper", + async (slackMode) => { + const wrappedReceiver = { durable: true }; + const wrapReceiver = vi.fn(() => wrappedReceiver as never); + const { app, receiver } = createSlackBoltApp({ + interop: { + App: FakeApp as never, + HTTPReceiver: FakeHTTPReceiver as never, + SocketModeReceiver: FakeSocketModeReceiver as never, + }, + slackMode, + botToken: "test-bot-token", + ...(slackMode === "socket" + ? { appToken: "test-app-token" } + : { signingSecret: "test-signing-secret" }), + slackWebhookPath: "/slack/events", + clientOptions: {}, + wrapReceiver, + }); + + expect(wrapReceiver).toHaveBeenCalledWith(receiver); + expect((app as unknown as FakeApp).args.receiver).toBe(wrappedReceiver); + const receiverArgs = (receiver as unknown as FakeHTTPReceiver | FakeSocketModeReceiver).args; + expect(receiverArgs.processEventErrorHandler).toBeTypeOf("function"); + await expect( + (receiverArgs.processEventErrorHandler as () => Promise)(), + ).resolves.toBe(false); + }, + ); + it("prevents Bolt's constructor-time token verification side effect", () => { let eagerAuthTestCalls = 0; class BoltLikeEagerAuthApp extends FakeApp { diff --git a/extensions/slack/src/monitor/provider.ts b/extensions/slack/src/monitor/provider.ts index 2e7afa72dcaa..29a0e75b4f19 100644 --- a/extensions/slack/src/monitor/provider.ts +++ b/extensions/slack/src/monitor/provider.ts @@ -63,6 +63,7 @@ import { type SlackAuthTestIdentity, } from "./enterprise-install.js"; import { registerSlackMonitorEvents } from "./events.js"; +import { createSlackDurableIngress } from "./ingress.js"; import { createSlackMessageHandler } from "./message-handler.js"; import { openSlackPresenceCooldownStore } from "./presence-cooldown-store.js"; import { createSlackPresenceMonitor, hasSlackPresenceEventsEnabled } from "./presence-monitor.js"; @@ -320,6 +321,11 @@ export async function monitorSlackProvider(opts: MonitorSlackOpts = {}) { const mediaMaxBytes = (opts.mediaMaxMb ?? slackCfg.mediaMaxMb ?? 20) * 1024 * 1024; const removeAckAfterReply = cfg.messages?.removeAckAfterReply ?? false; const clientOptions = resolveSlackWebClientOptions(); + const durableIngress = createSlackDurableIngress({ + accountId: account.accountId, + ...(runtime.log ? { onLog: runtime.log } : {}), + ...(opts.abortSignal ? { abortSignal: opts.abortSignal } : {}), + }); const { app, receiver, socketModeLogger } = createSlackBoltApp({ interop: await getSlackBoltInterop(), slackMode, @@ -329,6 +335,7 @@ export async function monitorSlackProvider(opts: MonitorSlackOpts = {}) { slackWebhookPath, clientOptions: clientOptions as Record, ...(slackCfg.socketMode ? { socketMode: slackCfg.socketMode } : {}), + wrapReceiver: durableIngress.wrapReceiver, }); // Pre-set shuttingDown on the SocketModeClient before app.stop() to prevent @@ -540,6 +547,7 @@ export async function monitorSlackProvider(opts: MonitorSlackOpts = {}) { appHomeSlashCommandName, trackEvent, }); + durableIngress.start(); presenceMonitor?.start(); if (slackMode === "http" && slackHttpHandler) { unregisterHttpHandler = registerSlackHttpHandler({ @@ -777,11 +785,23 @@ export async function monitorSlackProvider(opts: MonitorSlackOpts = {}) { runtime.log?.( `slack relay mode connecting to ${relayConfig.url} gateway_id:${relayConfig.gatewayId}`, ); + // Send identity flows through the account default (relay hello -> + // setIdentity); resolveSlackSendIdentity falls back to it, so claimed + // relay events replayed after a restart dispatch with correct identity + // once the relay reattaches. + durableIngress.attachRelayDispatch(async (message, turnAdoptionLifecycle) => { + await handleSlackMessage(message as Parameters[0], { + source: "message", + wasMentioned: true, + awaitDispatch: true, + turnAdoptionLifecycle, + }); + }); await ( await loadSlackRelaySource() ).monitorSlackRelaySource({ config: relayConfig, - handleSlackMessage, + acceptRelayEvent: durableIngress.acceptRelayEvent, runtime, abortSignal: opts.abortSignal, setStatus: opts.setStatus, @@ -805,6 +825,7 @@ export async function monitorSlackProvider(opts: MonitorSlackOpts = {}) { opts.abortSignal?.removeEventListener("abort", stopOnAbort); unregisterSocketModeConnectionDiagnostics(); unregisterHttpHandler?.(); + await durableIngress.stop(); await gracefulStop(); } } diff --git a/extensions/slack/src/monitor/relay-source.test.ts b/extensions/slack/src/monitor/relay-source.test.ts index 503526282c56..7043bcdc6f20 100644 --- a/extensions/slack/src/monitor/relay-source.test.ts +++ b/extensions/slack/src/monitor/relay-source.test.ts @@ -58,15 +58,15 @@ describe("Slack relay source", () => { }); }); - it("applies hello identity, dispatches a routed event, and acknowledges its delivery", async () => { + it("applies hello identity and acks a routed event only after durable accept", async () => { const server = new WebSocketServer({ host: "127.0.0.1", port: 0 }); await new Promise((resolve) => { server.once("listening", resolve); }); const port = (server.address() as AddressInfo).port; const ack = deferred>(); - const dispatchStarted = deferred(); - const dispatchDone = deferred(); + const acceptStarted = deferred(); + const acceptDone = deferred(); const receivedAcks: Array> = []; const requestHeaders = deferred<{ authorization?: string; url?: string }>(); server.once("connection", (socket, request) => { @@ -132,13 +132,15 @@ describe("Slack relay source", () => { }); const abortController = new AbortController(); - const handleSlackMessage = vi.fn(async (event: { text?: string }) => { - if (event.text === "fail-handler") { - throw new Error("handler failed"); - } - dispatchStarted.resolve(); - await dispatchDone.promise; - }); + const acceptRelayEvent = vi.fn( + async (event: { deliveryId: string; message: { text?: string } }) => { + if (event.message.text === "fail-handler") { + throw new Error("durable accept failed"); + } + acceptStarted.resolve(); + await acceptDone.promise; + }, + ); const runtimeError = vi.fn(); const identities: Array = []; const statuses: Array> = []; @@ -148,7 +150,7 @@ describe("Slack relay source", () => { authToken: "relay-secret", gatewayId: "pash", }, - handleSlackMessage, + acceptRelayEvent, runtime: { error: runtimeError, log: vi.fn() } as unknown as RuntimeEnv, abortSignal: abortController.signal, setIdentity: (identity) => identities.push(identity), @@ -159,27 +161,21 @@ describe("Slack relay source", () => { authorization: "Bearer relay-secret", url: "/gateway/ws?gateway_id=pash", }); - await dispatchStarted.promise; + await acceptStarted.promise; expect(receivedAcks).toEqual([]); - dispatchDone.resolve(); + acceptDone.resolve(); await expect(ack.promise).resolves.toEqual({ type: "ack", delivery_id: "delivery-1", }); expect(receivedAcks).toEqual([{ type: "ack", delivery_id: "delivery-1" }]); expect(runtimeError).toHaveBeenCalledTimes(2); - expect(handleSlackMessage).toHaveBeenCalledWith( - expect.objectContaining({ channel: "C1", text: "hello" }), - { - source: "message", - wasMentioned: true, - awaitDispatch: true, - relayIdentity: { - username: "Nik Team Claw", - iconUrl: "https://example.com/nik.png", - }, - }, - ); + expect(acceptRelayEvent).toHaveBeenCalledWith({ + deliveryId: "delivery-1", + message: expect.objectContaining({ channel: "C1", text: "hello" }), + }); + // The failed durable accept must never ack: the router redelivers it. + expect(receivedAcks).not.toContainEqual({ type: "ack", delivery_id: "delivery-failed" }); expect(identities).toContainEqual({ username: "Nik Team Claw", iconUrl: "https://example.com/nik.png", diff --git a/extensions/slack/src/monitor/relay-source.ts b/extensions/slack/src/monitor/relay-source.ts index 54226b0d84e2..6dca3de9d8f4 100644 --- a/extensions/slack/src/monitor/relay-source.ts +++ b/extensions/slack/src/monitor/relay-source.ts @@ -11,7 +11,6 @@ import { normalizeOptionalString } from "openclaw/plugin-sdk/string-coerce-runti import WebSocket, { type ClientOptions, type RawData } from "ws"; import type { SlackSendIdentity } from "../send.js"; import type { SlackMessageEvent } from "../types.js"; -import type { SlackMessageHandler } from "./message-handler.js"; import { formatUnknownError, SLACK_SOCKET_RECONNECT_POLICY } from "./reconnect-policy.js"; export type SlackRelaySourceConfig = { @@ -22,10 +21,6 @@ export type SlackRelaySourceConfig = { export type SlackRelayIdentity = SlackSendIdentity; -type RelayConnectionState = { - identity?: SlackRelayIdentity; -}; - const SLACK_RELAY_ROUTE_KINDS = new Set(["user_group", "thread_affinity", "channel_default"]); export const SLACK_RELAY_MAX_PAYLOAD_BYTES = 1024 * 1024; @@ -34,9 +29,14 @@ type SlackRelayRoute = { key: string; }; +export type SlackRelayEventAcceptor = (event: { + deliveryId: string; + message: SlackMessageEvent; +}) => Promise; + export async function monitorSlackRelaySource(params: { config: SlackRelaySourceConfig; - handleSlackMessage: SlackMessageHandler; + acceptRelayEvent: SlackRelayEventAcceptor; runtime: RuntimeEnv; abortSignal?: AbortSignal; setStatus?: (next: Record) => void; @@ -57,7 +57,7 @@ export async function monitorSlackRelaySource(params: { params.runtime.log?.(`slack relay mode connected gateway_id:${params.config.gatewayId}`); await runRelayWebSocket({ connection, - handleSlackMessage: params.handleSlackMessage, + acceptRelayEvent: params.acceptRelayEvent, runtime: params.runtime, abortSignal: params.abortSignal, setStatus: params.setStatus, @@ -135,14 +135,13 @@ function openRelayWebSocket( function runRelayWebSocket(params: { connection: WebSocket; - handleSlackMessage: SlackMessageHandler; + acceptRelayEvent: SlackRelayEventAcceptor; runtime: RuntimeEnv; abortSignal?: AbortSignal; setStatus?: (next: Record) => void; setIdentity?: (identity: SlackRelayIdentity | undefined) => void; }): Promise { const ws = params.connection; - const relayState: RelayConnectionState = {}; let pending = Promise.resolve(); return new Promise((resolve, reject) => { const cleanup = () => { @@ -165,8 +164,7 @@ function runRelayWebSocket(params: { handleRelayFrame({ ws, data, - handleSlackMessage: params.handleSlackMessage, - relayState, + acceptRelayEvent: params.acceptRelayEvent, setStatus: params.setStatus, setIdentity: params.setIdentity, }), @@ -204,15 +202,13 @@ function runRelayWebSocket(params: { async function handleRelayFrame(params: { ws: WebSocket; data: RawData; - handleSlackMessage: SlackMessageHandler; - relayState: RelayConnectionState; + acceptRelayEvent: SlackRelayEventAcceptor; setStatus?: (next: Record) => void; setIdentity?: (identity: SlackRelayIdentity | undefined) => void; }): Promise { const frame = parseRelayFrame(params.data); const hello = extractRelayHello(frame); if (hello) { - params.relayState.identity = hello.identity; params.setIdentity?.(hello.identity); params.setStatus?.({ relayIdentity: hello.identity ?? null }); return; @@ -224,13 +220,10 @@ async function handleRelayFrame(params: { const now = Date.now(); params.setStatus?.({ lastEventAt: now, lastInboundAt: now }); params.setStatus?.({ relayRoute: event.route }); - // Relay delivery is already authorized by the router's selected route. - await params.handleSlackMessage(event.message, { - source: "message", - wasMentioned: true, - awaitDispatch: true, - ...(params.relayState.identity ? { relayIdentity: params.relayState.identity } : {}), - }); + // Durable-before-ack: the router redelivers unacked frames, so the ack must + // gate on the SQLite enqueue. Dispatch runs through the durable drain; the + // logical message-id tombstone dedupes redeliveries of an already-acked frame. + await params.acceptRelayEvent({ deliveryId: event.deliveryId, message: event.message }); sendRelayAck(params.ws, event.deliveryId); }