diff --git a/extensions/slack/src/monitor/message-handler.test.ts b/extensions/slack/src/monitor/message-handler.test.ts index 77efa16554be..ea58d560a4f1 100644 --- a/extensions/slack/src/monitor/message-handler.test.ts +++ b/extensions/slack/src/monitor/message-handler.test.ts @@ -1,6 +1,11 @@ // Slack tests cover message handler plugin behavior. import { createTestInboundDebounceFlush } from "openclaw/plugin-sdk/channel-test-helpers"; -import { beforeEach, describe, expect, it, vi } from "vitest"; +import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts"; +import { + clearRuntimeConfigSnapshot, + setRuntimeConfigSnapshot, +} from "openclaw/plugin-sdk/runtime-config-snapshot"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; type InboundDebounceFlush = { admission: Promise; completion: Promise }; @@ -70,13 +75,14 @@ vi.mock("./message-handler/pipeline.runtime.js", () => ({ })); function createContext(overrides?: { + cfg?: OpenClawConfig; rememberSlackChannelType?: ( channel: string | null | undefined, channelType: string | null | undefined, ) => void; }) { return { - cfg: {}, + cfg: overrides?.cfg ?? {}, accountId: "default", app: { client: {}, @@ -120,6 +126,7 @@ async function handleDirectMessage( describe("createSlackMessageHandler", () => { beforeEach(() => { + clearRuntimeConfigSnapshot(); enqueueMock.mockClear(); flushKeyMock.mockClear(); onFlushCallbacks.length = 0; @@ -128,6 +135,197 @@ describe("createSlackMessageHandler", () => { resolveThreadTsMock.mockClear(); }); + afterEach(() => { + clearRuntimeConfigSnapshot(); + }); + + it("uses the latest runtime config for messages without restarting the monitor", async () => { + const startupConfig: OpenClawConfig = { agents: { defaults: { thinkingDefault: "max" } } }; + const updatedConfig: OpenClawConfig = { + agents: { defaults: { thinkingDefault: "ultra", fastModeDefault: true } }, + }; + const context = createContext({ cfg: startupConfig }); + const handler = createSlackMessageHandler({ + ctx: context, + account: { accountId: "default" } as Parameters< + typeof createSlackMessageHandler + >[0]["account"], + }); + + setRuntimeConfigSnapshot(updatedConfig, updatedConfig); + await handler( + { + type: "message", + channel: "D1", + user: "U1", + ts: "1709000000.009001", + text: "hello", + } as never, + { source: "message" }, + ); + const entry = enqueueMock.mock.calls[0]?.[0] as Record; + await runOnFlush([entry]); + + expect(prepareSlackMessageMock).toHaveBeenCalledWith( + expect.objectContaining({ + ctx: expect.objectContaining({ cfg: updatedConfig }), + }), + ); + expect(context.cfg).toBe(startupConfig); + }); + + it.each([ + { + label: "without a source snapshot", + includeSourceSnapshot: false, + messageTs: "1709000000.009004", + }, + { + label: "with an unrelated source snapshot", + includeSourceSnapshot: true, + messageTs: "1709000000.009005", + }, + ])("preserves explicit monitor config $label", async ({ includeSourceSnapshot, messageTs }) => { + const explicitConfig: OpenClawConfig = { + agents: { defaults: { thinkingDefault: "ultra" } }, + messages: { responsePrefix: "scoped" }, + }; + const unrelatedRuntimeConfig: OpenClawConfig = { + agents: { defaults: { thinkingDefault: "low" } }, + }; + setRuntimeConfigSnapshot( + unrelatedRuntimeConfig, + includeSourceSnapshot ? unrelatedRuntimeConfig : undefined, + ); + const context = createContext({ cfg: explicitConfig }); + const handler = createSlackMessageHandler({ + ctx: context, + account: { accountId: "default" } as Parameters< + typeof createSlackMessageHandler + >[0]["account"], + }); + + setRuntimeConfigSnapshot({ agents: { defaults: { thinkingDefault: "high" } } }); + await handler( + { + type: "message", + channel: "D1", + user: "U1", + ts: messageTs, + text: "hello", + } as never, + { source: "message" }, + ); + const entry = enqueueMock.mock.calls[0]?.[0] as Record; + await runOnFlush([entry]); + + expect(prepareSlackMessageMock).toHaveBeenCalledWith(expect.objectContaining({ ctx: context })); + expect(context.cfg).toBe(explicitConfig); + }); + + it("follows runtime updates when the monitor config matches the runtime source", async () => { + const startupSourceConfig: OpenClawConfig = { + agents: { defaults: { thinkingDefault: "max" } }, + }; + const startupRuntimeConfig: OpenClawConfig = { + agents: { defaults: { thinkingDefault: "max", fastModeDefault: false } }, + }; + const updatedRuntimeConfig: OpenClawConfig = { + agents: { defaults: { thinkingDefault: "ultra", fastModeDefault: true } }, + }; + setRuntimeConfigSnapshot(startupRuntimeConfig, startupSourceConfig); + const context = createContext({ cfg: structuredClone(startupSourceConfig) }); + const handler = createSlackMessageHandler({ + ctx: context, + account: { accountId: "default" } as Parameters< + typeof createSlackMessageHandler + >[0]["account"], + }); + + setRuntimeConfigSnapshot(updatedRuntimeConfig, updatedRuntimeConfig); + await handler( + { + type: "message", + channel: "D1", + user: "U1", + ts: "1709000000.009006", + text: "hello", + } as never, + { source: "message" }, + ); + const entry = enqueueMock.mock.calls[0]?.[0] as Record; + await runOnFlush([entry]); + + expect(prepareSlackMessageMock).toHaveBeenCalledWith( + expect.objectContaining({ + ctx: expect.objectContaining({ cfg: updatedRuntimeConfig }), + }), + ); + expect(context.cfg).toEqual(startupSourceConfig); + }); + + it("keeps each in-flight message on its captured config snapshot", async () => { + const startupConfig: OpenClawConfig = { agents: { defaults: { thinkingDefault: "max" } } }; + const firstConfig: OpenClawConfig = { agents: { defaults: { thinkingDefault: "high" } } }; + const secondConfig: OpenClawConfig = { agents: { defaults: { thinkingDefault: "ultra" } } }; + const context = createContext({ cfg: startupConfig }); + const handler = createSlackMessageHandler({ + ctx: context, + account: { accountId: "default" } as Parameters< + typeof createSlackMessageHandler + >[0]["account"], + }); + let releaseFirstPreparation!: () => void; + const firstPreparation = new Promise((resolve) => { + releaseFirstPreparation = resolve; + }); + prepareSlackMessageMock.mockImplementationOnce(async () => { + await firstPreparation; + return { ctxPayload: {} }; + }); + + setRuntimeConfigSnapshot(firstConfig, firstConfig); + await handler( + { + type: "message", + channel: "D1", + user: "U1", + ts: "1709000000.009002", + text: "first", + } as never, + { source: "message" }, + ); + const firstEntry = enqueueMock.mock.calls[0]?.[0] as Record; + const firstFlush = runOnFlush([firstEntry]); + await vi.waitFor(() => expect(prepareSlackMessageMock).toHaveBeenCalledTimes(1)); + + setRuntimeConfigSnapshot(secondConfig, secondConfig); + await handler( + { + type: "message", + channel: "D2", + user: "U2", + ts: "1709000000.009003", + text: "second", + } as never, + { source: "message" }, + ); + const secondEntry = enqueueMock.mock.calls[1]?.[0] as Record; + await runOnFlush([secondEntry]); + releaseFirstPreparation(); + await firstFlush; + + expect(prepareSlackMessageMock).toHaveBeenNthCalledWith( + 1, + expect.objectContaining({ ctx: expect.objectContaining({ cfg: firstConfig }) }), + ); + expect(prepareSlackMessageMock).toHaveBeenNthCalledWith( + 2, + expect.objectContaining({ ctx: expect.objectContaining({ cfg: secondConfig }) }), + ); + expect(context.cfg).toBe(startupConfig); + }); + it("does not track invalid non-message events from the message stream", async () => { const trackEvent = vi.fn(); const handler = createSlackMessageHandler({ diff --git a/extensions/slack/src/monitor/message-handler.ts b/extensions/slack/src/monitor/message-handler.ts index 6253942944d7..555a8c58c908 100644 --- a/extensions/slack/src/monitor/message-handler.ts +++ b/extensions/slack/src/monitor/message-handler.ts @@ -5,6 +5,11 @@ 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 { + getRuntimeConfigSnapshot, + getRuntimeConfigSourceSnapshot, + selectApplicableRuntimeConfig, +} from "openclaw/plugin-sdk/runtime-config-snapshot"; import type { ResolvedSlackAccount } from "../accounts.js"; import type { SlackSendIdentity } from "../send.js"; import type { SlackMessageEvent } from "../types.js"; @@ -101,6 +106,36 @@ export function createSlackMessageHandler(params: { dispatchReplayGuard?: SlackMessageDispatchReplayGuard; }): SlackMessageHandler { const { ctx, account, trackEvent, onPrepared } = params; + const startupRuntimeConfig = getRuntimeConfigSnapshot(); + const startupRuntimeSourceConfig = getRuntimeConfigSourceSnapshot(); + // Bind snapshot ownership once so unrelated process-global config cannot replace scoped monitors. + const followsRuntimeConfig = + !startupRuntimeConfig || + startupRuntimeConfig === ctx.cfg || + (startupRuntimeSourceConfig !== null && + selectApplicableRuntimeConfig({ + inputConfig: ctx.cfg, + runtimeConfig: startupRuntimeConfig, + runtimeSourceConfig: startupRuntimeSourceConfig, + }) === startupRuntimeConfig); + const runtimeContexts = new WeakMap< + NonNullable, + SlackMonitorContext + >(); + const resolveRuntimeContext = (): SlackMonitorContext => { + // Channel monitors outlive config reloads; pin one live snapshot per turn without reconnecting. + const runtimeConfig = getRuntimeConfigSnapshot(); + if (!followsRuntimeConfig || !runtimeConfig || runtimeConfig === ctx.cfg) { + return ctx; + } + const cached = runtimeContexts.get(runtimeConfig); + if (cached) { + return cached; + } + const runtimeContext = { ...ctx, cfg: runtimeConfig }; + runtimeContexts.set(runtimeConfig, runtimeContext); + return runtimeContext; + }; const dispatchReplayGuard = params.dispatchReplayGuard ?? createSlackMessageDispatchReplayGuard({ @@ -274,8 +309,9 @@ export function createSlackMessageHandler(params: { let visibleDrop = false; let settlementHandedOff = false; try { + const runtimeContext = resolveRuntimeContext(); prepared = await prepareSlackMessage({ - ctx, + ctx: runtimeContext, account, message: syntheticMessage, opts: {