diff --git a/extensions/slack/src/monitor/events/channels.test.ts b/extensions/slack/src/monitor/events/channels.test.ts index 2697a450ca25..1d777fad5480 100644 --- a/extensions/slack/src/monitor/events/channels.test.ts +++ b/extensions/slack/src/monitor/events/channels.test.ts @@ -1,16 +1,28 @@ // Slack tests cover channels plugin behavior. import { beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; -const enqueueSystemEventMock = vi.hoisted(() => vi.fn()); +const { enqueueSystemEventMock, mutateConfigFileMock, readConfigSnapshotMock } = vi.hoisted(() => ({ + enqueueSystemEventMock: vi.fn(), + mutateConfigFileMock: vi.fn(), + readConfigSnapshotMock: vi.fn(), +})); let registerSlackChannelEvents: typeof import("./channels.js").registerSlackChannelEvents; let createSlackSystemEventTestHarness: typeof import("./system-event-test-harness.js").createSlackSystemEventTestHarness; vi.mock("openclaw/plugin-sdk/system-event-runtime", () => ({ enqueueSystemEvent: (...args: unknown[]) => enqueueSystemEventMock(...args), })); +vi.mock("openclaw/plugin-sdk/channel-config-writes", () => ({ + resolveChannelConfigWrites: () => true, +})); +vi.mock("openclaw/plugin-sdk/config-mutation", () => ({ + mutateConfigFile: (...args: unknown[]) => mutateConfigFileMock(...args), + readConfigFileSnapshotForWrite: (...args: unknown[]) => readConfigSnapshotMock(...args), +})); type SlackChannelHandler = (args: { event: Record; body: unknown; + context?: Record; }) => Promise; function createChannelContext(params?: { @@ -23,6 +35,8 @@ function createChannelContext(params?: { } registerSlackChannelEvents({ ctx: harness.ctx, trackEvent: params?.trackEvent }); return { + ctx: harness.ctx, + getHandler: (name: string) => harness.getHandler(name) as SlackChannelHandler | null, getCreatedHandler: () => harness.getHandler("channel_created") as SlackChannelHandler | null, }; } @@ -42,6 +56,8 @@ describe("registerSlackChannelEvents", () => { beforeEach(() => { enqueueSystemEventMock.mockClear(); + mutateConfigFileMock.mockReset(); + readConfigSnapshotMock.mockReset(); }); it("does not track mismatched events", async () => { @@ -81,4 +97,63 @@ describe("registerSlackChannelEvents", () => { contextKey: "slack:channel:created:C1", }); }); + + it("keeps live config unchanged when channel-ID persistence fails, then retries", async () => { + const oldChannelId = "C_OLD"; + const newChannelId = "C_NEW"; + const initialConfig = { + channels: { slack: { channels: { [oldChannelId]: { enabled: true } } } }, + }; + let persistedConfig = structuredClone(initialConfig); + const persistenceError = new Error("disk full"); + readConfigSnapshotMock.mockImplementation(async () => ({ + snapshot: { + hash: "base-hash", + sourceConfig: structuredClone(persistedConfig), + }, + })); + mutateConfigFileMock + .mockRejectedValueOnce(persistenceError) + .mockImplementationOnce( + async (params: { mutate: (draft: typeof initialConfig) => unknown }) => { + const draft = structuredClone(persistedConfig); + const result = params.mutate(draft); + persistedConfig = draft; + return { result, nextConfig: draft }; + }, + ); + const { ctx, getHandler } = createChannelContext(); + ctx.cfg = structuredClone(initialConfig) as never; + ctx.accountId = "default"; + ctx.runtime.error = vi.fn(); + const handler = requireChannelHandler(getHandler("channel_id_changed")); + const turnAdoptionLifecycle = { + admission: "exclusive", + abortSignal: new AbortController().signal, + onAdopted: vi.fn(), + onDeferred: vi.fn(), + onAbandoned: vi.fn(), + }; + const args = { + event: { + type: "channel_id_changed", + old_channel_id: oldChannelId, + new_channel_id: newChannelId, + }, + body: {}, + context: { openclawIngressLifecycle: turnAdoptionLifecycle }, + }; + + await expect(handler(args)).rejects.toBe(persistenceError); + expect(ctx.cfg.channels?.slack?.channels).toHaveProperty(oldChannelId); + expect(ctx.cfg.channels?.slack?.channels).not.toHaveProperty(newChannelId); + expect(persistedConfig.channels.slack.channels).toHaveProperty(oldChannelId); + + await expect(handler(args)).resolves.toBeUndefined(); + expect(ctx.cfg.channels?.slack?.channels).not.toHaveProperty(oldChannelId); + expect(ctx.cfg.channels?.slack?.channels).toHaveProperty(newChannelId); + expect(persistedConfig.channels.slack.channels).not.toHaveProperty(oldChannelId); + expect(persistedConfig.channels.slack.channels).toHaveProperty(newChannelId); + expect(mutateConfigFileMock).toHaveBeenCalledTimes(2); + }); }); diff --git a/extensions/slack/src/monitor/events/channels.ts b/extensions/slack/src/monitor/events/channels.ts index aad419a9d932..7db01a9e1248 100644 --- a/extensions/slack/src/monitor/events/channels.ts +++ b/extensions/slack/src/monitor/events/channels.ts @@ -1,14 +1,17 @@ // Slack plugin module implements channels behavior. -import type { SlackEventMiddlewareArgs } from "@slack/bolt"; +import type { AllMiddlewareArgs, SlackEventMiddlewareArgs } from "@slack/bolt"; import { resolveChannelConfigWrites } from "openclaw/plugin-sdk/channel-config-writes"; -import { mutateConfigFile } from "openclaw/plugin-sdk/config-mutation"; +import { + mutateConfigFile, + readConfigFileSnapshotForWrite, +} from "openclaw/plugin-sdk/config-mutation"; import { formatErrorMessage } from "openclaw/plugin-sdk/error-runtime"; -import { getRuntimeConfig } from "openclaw/plugin-sdk/runtime-config-snapshot"; import { danger, warn } from "openclaw/plugin-sdk/runtime-env"; import { enqueueSystemEvent } from "openclaw/plugin-sdk/system-event-runtime"; import { migrateSlackChannelConfig } from "../../channel-migration.js"; import { resolveSlackChannelLabel } from "../channel-config.js"; import type { SlackMonitorContext } from "../context.js"; +import { resolveSlackIngressTurnLifecycle } from "../ingress.js"; import type { SlackChannelCreatedEvent, SlackChannelIdChangedEvent, @@ -94,7 +97,12 @@ export function registerSlackChannelEvents(params: { ctx.app.event( "channel_id_changed", - async ({ event, body }: SlackEventMiddlewareArgs<"channel_id_changed">) => { + async ({ + event, + body, + context, + }: SlackEventMiddlewareArgs<"channel_id_changed"> & AllMiddlewareArgs) => { + const turnAdoptionLifecycle = resolveSlackIngressTurnLifecycle(context); try { if (ctx.shouldDropMismatchedSlackEvent(body)) { return; @@ -131,34 +139,39 @@ export function registerSlackChannelEvents(params: { return; } - const currentConfig = getRuntimeConfig(); - const migration = migrateSlackChannelConfig({ - cfg: currentConfig, + const { snapshot } = await readConfigFileSnapshotForWrite(); + const previewConfig = structuredClone(snapshot.sourceConfig); + const preview = migrateSlackChannelConfig({ + cfg: previewConfig, accountId: ctx.accountId, oldChannelId, newChannelId, }); - if (migration.migrated) { - migrateSlackChannelConfig({ - cfg: ctx.cfg, - accountId: ctx.accountId, - oldChannelId, - newChannelId, - }); - await mutateConfigFile({ + if (preview.migrated) { + const persisted = await mutateConfigFile({ + baseHash: snapshot.hash ?? undefined, afterWrite: { mode: "auto" }, - mutate: (draft) => { + mutate: (draft) => migrateSlackChannelConfig({ cfg: draft, accountId: ctx.accountId, oldChannelId, newChannelId, - }); - }, + }), }); - ctx.runtime.log?.(warn("[slack] Channel config migrated and saved successfully.")); - } else if (migration.skippedExisting) { + if (persisted.result?.migrated) { + // Persistence owns the migration. Update this monitor's captured + // config only after the durable write succeeds. + migrateSlackChannelConfig({ + cfg: ctx.cfg, + accountId: ctx.accountId, + oldChannelId, + newChannelId, + }); + ctx.runtime.log?.(warn("[slack] Channel config migrated and saved successfully.")); + } + } else if (preview.skippedExisting) { ctx.runtime.log?.( warn( `[slack] Channel config already exists for ${newChannelId}; leaving ${oldChannelId} unchanged`, @@ -175,6 +188,9 @@ export function registerSlackChannelEvents(params: { ctx.runtime.error?.( danger(`slack channel_id_changed handler failed: ${formatErrorMessage(err)}`), ); + if (turnAdoptionLifecycle) { + throw err; + } } }, ); diff --git a/extensions/slack/src/monitor/ingress.test.ts b/extensions/slack/src/monitor/ingress.test.ts index 5eec0233ce53..f2998285f6a2 100644 --- a/extensions/slack/src/monitor/ingress.test.ts +++ b/extensions/slack/src/monitor/ingress.test.ts @@ -33,6 +33,25 @@ function createSlackEnvelope(eventId: string, ts = "1700000000.000100") { }; } +function createChannelIdChangedEnvelope( + eventId: string, + oldChannelId: string, + newChannelId: string, +) { + return { + team_id: "T_TEST", + api_app_id: "A_TEST", + type: "event_callback", + event_id: eventId, + event_time: 1_700_000_000, + event: { + type: "channel_id_changed", + old_channel_id: oldChannelId, + new_channel_id: newChannelId, + }, + }; +} + function createReceiverHarness() { let receive: ((event: ReceiverEvent) => Promise) | undefined; const receiver: Receiver = { @@ -65,6 +84,10 @@ function createReceiverEvent( }; } +function createReceiverEventWithBody(body: Record): ReceiverEvent { + return { body, ack: vi.fn(async () => {}) }; +} + function attachIngress( queue: ChannelIngressQueue, processEvent: (event: ReceiverEvent) => Promise, @@ -158,6 +181,59 @@ describe("Slack durable ingress", () => { }); }); + it("serializes new-channel messages behind channel-ID migration", async () => { + await withQueue(async (queue) => { + let markMigrationStarted: () => void = () => {}; + let releaseMigration: () => void = () => {}; + const migrationStarted = new Promise((resolve) => { + markMigrationStarted = resolve; + }); + const migrationGate = new Promise((resolve) => { + releaseMigration = resolve; + }); + const starts: string[] = []; + const processEvent = vi.fn(async (receiverEvent: ReceiverEvent) => { + const event = (receiverEvent.body as { event?: { type?: string } }).event; + const type = event?.type ?? "unknown"; + starts.push(type); + if (type === "channel_id_changed") { + markMigrationStarted(); + await migrationGate; + } + await resolveSlackIngressTurnLifecycle(receiverEvent.customProperties)?.onAdopted(); + }); + const { ingress, receive } = attachIngress(queue, processEvent); + ingress.start(); + + await receive( + createReceiverEventWithBody( + createChannelIdChangedEnvelope("Ev-channel-migrate", "C_OLD", "C_NEW"), + ), + ); + await receive( + createReceiverEventWithBody({ + ...createSlackEnvelope("Ev-new-channel-message"), + event: { + type: "message", + channel: "C_NEW", + user: "U_TEST", + ts: "1700000000.000200", + text: "after migration", + }, + }), + ); + + await migrationStarted; + await Promise.resolve(); + expect(starts).toEqual(["channel_id_changed"]); + + releaseMigration(); + await ingress.waitForIdle(); + expect(starts).toEqual(["channel_id_changed", "message"]); + await ingress.stop(); + }); + }); + it("drains a durable event when its acknowledgement fails", async () => { await withQueue(async (queue) => { const processEvent = vi.fn(async (event: ReceiverEvent) => { diff --git a/extensions/slack/src/monitor/ingress.ts b/extensions/slack/src/monitor/ingress.ts index c2ae5dcc644b..be65f8814376 100644 --- a/extensions/slack/src/monitor/ingress.ts +++ b/extensions/slack/src/monitor/ingress.ts @@ -126,7 +126,15 @@ function resolveSlackIngressLane(body: unknown, eventId: string): string { .find((value) => typeof value === "string" && value.trim()) ?.toString() .trim() || "workspace"; - const channelId = [event?.channel, event?.channel_id, item?.channel, assistantThread?.channel_id] + // New-channel traffic must stay behind channel_id_changed migration work. + // The new ID owns the post-change conversation lane, not the retired old ID. + const channelId = [ + event?.channel, + event?.channel_id, + event?.new_channel_id, + item?.channel, + assistantThread?.channel_id, + ] .find((value) => typeof value === "string" && value.trim()) ?.toString() .trim(); diff --git a/extensions/slack/src/monitor/media.test.ts b/extensions/slack/src/monitor/media.test.ts index ca0c396a49d0..88f2d65f6b65 100644 --- a/extensions/slack/src/monitor/media.test.ts +++ b/extensions/slack/src/monitor/media.test.ts @@ -28,6 +28,16 @@ type SaveMediaBufferMock = ( originalFilename?: string, ) => Promise; type SlackMediaResult = NonNullable>>; +type ResolveSlackThreadStarterParams = Parameters[0]; + +function resolveTestSlackThreadStarter( + params: Omit, +) { + return resolveSlackThreadStarter({ + ...params, + workspaceScope: { accountId: "test", teamId: "T1" }, + }); +} function expectSlackMediaResult( result: Awaited>, @@ -1633,7 +1643,7 @@ describe("resolveSlackThreadStarter", () => { conversations: { replies }, } as unknown as Parameters[0]["client"]; - const result = await resolveSlackThreadStarter({ + const result = await resolveTestSlackThreadStarter({ channelId: "C1", threadTs: "1.000", client, @@ -1655,7 +1665,7 @@ describe("resolveSlackThreadStarter", () => { conversations: { replies }, } as unknown as Parameters[0]["client"]; - const result = await resolveSlackThreadStarter({ + const result = await resolveTestSlackThreadStarter({ channelId: "C1", threadTs: "1.000", client, @@ -1686,7 +1696,7 @@ describe("resolveSlackThreadStarter", () => { conversations: { replies }, } as unknown as Parameters[0]["client"]; - const result = await resolveSlackThreadStarter({ + const result = await resolveTestSlackThreadStarter({ channelId: "C1", threadTs: "1.000", client, @@ -1727,7 +1737,7 @@ describe("resolveSlackThreadStarter", () => { conversations: { replies }, } as unknown as Parameters[0]["client"]; - const result = await resolveSlackThreadStarter({ + const result = await resolveTestSlackThreadStarter({ channelId: "C1", threadTs: "1.000", client, @@ -1751,7 +1761,7 @@ describe("resolveSlackThreadStarter", () => { conversations: { replies }, } as unknown as Parameters[0]["client"]; - const result = await resolveSlackThreadStarter({ + const result = await resolveTestSlackThreadStarter({ channelId: "C1", threadTs: "1.000", client, @@ -1773,7 +1783,7 @@ describe("resolveSlackThreadStarter", () => { conversations: { replies }, } as unknown as Parameters[0]["client"]; - const result = await resolveSlackThreadStarter({ + const result = await resolveTestSlackThreadStarter({ channelId: "C42", threadTs: "9.999", client, @@ -1792,7 +1802,7 @@ describe("resolveSlackThreadStarter", () => { conversations: { replies }, } as unknown as Parameters[0]["client"]; - const result = await resolveSlackThreadStarter({ + const result = await resolveTestSlackThreadStarter({ channelId: "C1", threadTs: "1.000", client, diff --git a/extensions/slack/src/monitor/message-handler/prepare.ts b/extensions/slack/src/monitor/message-handler/prepare.ts index 2cef12322d97..6e0317245560 100644 --- a/extensions/slack/src/monitor/message-handler/prepare.ts +++ b/extensions/slack/src/monitor/message-handler/prepare.ts @@ -707,9 +707,10 @@ export async function prepareSlackMessage(params: { }): Promise { const { ctx, account, message, opts } = params; const slackClient = opts.eventScope?.client ?? ctx.app.client; - const threadStarterWorkspaceScope = opts.eventScope - ? { accountId: account.accountId, teamId: opts.eventScope.teamId } - : undefined; + const threadStarterWorkspaceScope = { + accountId: account.accountId, + teamId: opts.eventScope?.teamId ?? ctx.teamId, + }; const cfg = ctx.cfg; const conversation = await resolveSlackConversationContext({ ctx, diff --git a/extensions/slack/src/monitor/monitor.media.test.ts b/extensions/slack/src/monitor/monitor.media.test.ts index 73af5ee4dba3..94d7dea08e97 100644 --- a/extensions/slack/src/monitor/monitor.media.test.ts +++ b/extensions/slack/src/monitor/monitor.media.test.ts @@ -2,7 +2,21 @@ import { afterEach, describe, expect, it, vi } from "vitest"; import { resetSlackThreadStarterCacheForTest, resolveSlackThreadStarter } from "./thread.js"; -type ThreadStarterClient = Parameters[0]["client"]; +type ResolveSlackThreadStarterParams = Parameters[0]; +type ThreadStarterClient = ResolveSlackThreadStarterParams["client"]; + +const TEST_WORKSPACE_SCOPE = { accountId: "test", teamId: "T1" }; + +function resolveTestSlackThreadStarter( + params: Omit & { + workspaceScope?: ResolveSlackThreadStarterParams["workspaceScope"]; + }, +) { + return resolveSlackThreadStarter({ + ...params, + workspaceScope: params.workspaceScope ?? TEST_WORKSPACE_SCOPE, + }); +} function createThreadStarterRepliesClient( response: { messages?: Array<{ text?: string; user?: string; ts?: string }> } = { @@ -25,12 +39,12 @@ describe("resolveSlackThreadStarter cache", () => { it("returns cached thread starter without refetching within ttl", async () => { const { replies, client } = createThreadStarterRepliesClient(); - const first = await resolveSlackThreadStarter({ + const first = await resolveTestSlackThreadStarter({ channelId: "C1", threadTs: "1000.1", client, }); - const second = await resolveSlackThreadStarter({ + const second = await resolveTestSlackThreadStarter({ channelId: "C1", threadTs: "1000.1", client, @@ -40,7 +54,7 @@ describe("resolveSlackThreadStarter cache", () => { expect(replies).toHaveBeenCalledTimes(1); }); - it("isolates the same channel and thread across enterprise workspaces", async () => { + it("isolates the same channel and thread across Slack accounts", async () => { const teamOne = createThreadStarterRepliesClient({ messages: [{ text: "team one root", user: "U1", ts: "1000.1" }], }); @@ -48,23 +62,23 @@ describe("resolveSlackThreadStarter cache", () => { messages: [{ text: "team two root", user: "U2", ts: "1000.1" }], }); - const first = await resolveSlackThreadStarter({ + const first = await resolveTestSlackThreadStarter({ channelId: "C1", threadTs: "1000.1", client: teamOne.client, - workspaceScope: { accountId: "enterprise", teamId: "T1" }, + workspaceScope: { accountId: "account-one", teamId: "T1" }, }); - const second = await resolveSlackThreadStarter({ + const second = await resolveTestSlackThreadStarter({ channelId: "C1", threadTs: "1000.1", client: teamTwo.client, - workspaceScope: { accountId: "enterprise", teamId: "T2" }, + workspaceScope: { accountId: "account-two", teamId: "T2" }, }); - const cachedFirst = await resolveSlackThreadStarter({ + const cachedFirst = await resolveTestSlackThreadStarter({ channelId: "C1", threadTs: "1000.1", client: teamOne.client, - workspaceScope: { accountId: "enterprise", teamId: "T1" }, + workspaceScope: { accountId: "account-one", teamId: "T1" }, }); expect(first?.text).toBe("team one root"); @@ -80,14 +94,14 @@ describe("resolveSlackThreadStarter cache", () => { const { replies, client } = createThreadStarterRepliesClient(); - await resolveSlackThreadStarter({ + await resolveTestSlackThreadStarter({ channelId: "C1", threadTs: "1000.1", client, }); vi.setSystemTime(new Date("2026-01-01T07:00:00.000Z")); - await resolveSlackThreadStarter({ + await resolveTestSlackThreadStarter({ channelId: "C1", threadTs: "1000.1", client, @@ -100,13 +114,13 @@ describe("resolveSlackThreadStarter cache", () => { const nowSpy = vi.spyOn(Date, "now").mockReturnValue(1_700_000_000_000); const { replies, client } = createThreadStarterRepliesClient(); - const first = await resolveSlackThreadStarter({ + const first = await resolveTestSlackThreadStarter({ channelId: "C1", threadTs: "1000.1", client, }); nowSpy.mockReturnValue(Number.NaN); - const second = await resolveSlackThreadStarter({ + const second = await resolveTestSlackThreadStarter({ channelId: "C1", threadTs: "1000.1", client, @@ -120,12 +134,12 @@ describe("resolveSlackThreadStarter cache", () => { vi.spyOn(Date, "now").mockReturnValue(8_640_000_000_000_000); const { replies, client } = createThreadStarterRepliesClient(); - const first = await resolveSlackThreadStarter({ + const first = await resolveTestSlackThreadStarter({ channelId: "C1", threadTs: "1000.1", client, }); - const second = await resolveSlackThreadStarter({ + const second = await resolveTestSlackThreadStarter({ channelId: "C1", threadTs: "1000.1", client, @@ -140,12 +154,12 @@ describe("resolveSlackThreadStarter cache", () => { messages: [{ text: " ", user: "U1", ts: "1000.1" }], }); - const first = await resolveSlackThreadStarter({ + const first = await resolveTestSlackThreadStarter({ channelId: "C1", threadTs: "1000.1", client, }); - const second = await resolveSlackThreadStarter({ + const second = await resolveTestSlackThreadStarter({ channelId: "C1", threadTs: "1000.1", client, @@ -160,7 +174,7 @@ describe("resolveSlackThreadStarter cache", () => { const { replies, client } = createThreadStarterRepliesClient(); for (let i = 0; i <= 2000; i += 1) { - await resolveSlackThreadStarter({ + await resolveTestSlackThreadStarter({ channelId: "C1", threadTs: `1000.${i}`, client, @@ -168,7 +182,7 @@ describe("resolveSlackThreadStarter cache", () => { } const callsAfterFill = replies.mock.calls.length; - await resolveSlackThreadStarter({ + await resolveTestSlackThreadStarter({ channelId: "C1", threadTs: "1000.0", client, diff --git a/extensions/slack/src/monitor/thread.ts b/extensions/slack/src/monitor/thread.ts index fbb27136b103..59e5529a840a 100644 --- a/extensions/slack/src/monitor/thread.ts +++ b/extensions/slack/src/monitor/thread.ts @@ -126,18 +126,15 @@ export async function resolveSlackThreadStarter(params: { channelId: string; threadTs: string; client: SlackWebClient; - /** Enterprise cache partition. Omit to preserve workspace-install cache identity. */ - workspaceScope?: { accountId: string; teamId: string }; + workspaceScope: { accountId: string; teamId: string }; }): Promise { evictThreadStarterCache(); - const cacheKey = params.workspaceScope - ? JSON.stringify([ - params.workspaceScope.accountId, - params.workspaceScope.teamId, - params.channelId, - params.threadTs, - ]) - : `${params.channelId}:${params.threadTs}`; + const cacheKey = JSON.stringify([ + params.workspaceScope.accountId, + params.workspaceScope.teamId, + params.channelId, + params.threadTs, + ]); const cached = THREAD_STARTER_CACHE.get(cacheKey); if (cached) { const now = asDateTimestampMs(Date.now());