From ddcc3fbd80bfa163a458f1d8e8318d3d911bf424 Mon Sep 17 00:00:00 2001 From: pash-openai Date: Wed, 5 Aug 2026 21:42:38 -0700 Subject: [PATCH] fix(slack): keep agent replies below newer messages (#119832) * fix(slack): keep replies below messages sent while an agent is working * fix(slack): preserve conversation order through final delivery --- .../slack/src/draft-message-boundaries.ts | 115 ++++++ extensions/slack/src/draft-stream.test.ts | 327 ++++++++++++++++++ extensions/slack/src/draft-stream.ts | 90 ++++- .../slack/src/monitor/events/messages.test.ts | 38 +- .../slack/src/monitor/events/messages.ts | 21 ++ .../message-handler/dispatch-progress.ts | 1 + .../dispatch.preview-fallback.test.ts | 85 +++++ .../src/monitor/message-handler/dispatch.ts | 48 ++- 8 files changed, 698 insertions(+), 27 deletions(-) create mode 100644 extensions/slack/src/draft-message-boundaries.ts diff --git a/extensions/slack/src/draft-message-boundaries.ts b/extensions/slack/src/draft-message-boundaries.ts new file mode 100644 index 000000000000..73055193748a --- /dev/null +++ b/extensions/slack/src/draft-message-boundaries.ts @@ -0,0 +1,115 @@ +type SlackDraftConversation = { + accountId?: string; + teamId?: string; + channelId: string; + threadTs?: string; +}; + +type ActiveSlackDraft = { + messageTs?: string; + latestHumanMessageTs?: string; + onInterveningMessage: () => void; +}; + +type SlackDraftMessageTracker = { + setMessageTs: (messageTs: string) => void; + stop: () => void; +}; + +const activeDraftsByConversation = new Map>(); + +function conversationKey(conversation: SlackDraftConversation): string { + return [ + conversation.accountId ?? "default", + conversation.teamId ?? "", + conversation.channelId, + conversation.threadTs ?? "", + ].join(":"); +} + +function isLaterSlackMessage(candidate: string, current: string): boolean { + const candidateTimestamp = Number(candidate); + const currentTimestamp = Number(current); + return ( + Number.isFinite(candidateTimestamp) && + Number.isFinite(currentTimestamp) && + candidateTimestamp > currentTimestamp + ); +} + +/** Keeps a live preview attached to its actual place in the Slack conversation. */ +export function trackSlackDraftMessage( + conversation: SlackDraftConversation & ActiveSlackDraft, +): SlackDraftMessageTracker { + const key = conversationKey(conversation); + const activeDraft: ActiveSlackDraft = { + messageTs: conversation.messageTs, + onInterveningMessage: conversation.onInterveningMessage, + }; + const drafts = activeDraftsByConversation.get(key) ?? new Set(); + drafts.add(activeDraft); + activeDraftsByConversation.set(key, drafts); + + const stop = () => { + const currentDrafts = activeDraftsByConversation.get(key); + currentDrafts?.delete(activeDraft); + if (currentDrafts?.size === 0) { + activeDraftsByConversation.delete(key); + } + }; + + return { + setMessageTs: (messageTs) => { + activeDraft.messageTs = messageTs; + if ( + activeDraft.latestHumanMessageTs && + isLaterSlackMessage(activeDraft.latestHumanMessageTs, messageTs) + ) { + activeDraft.onInterveningMessage(); + } + }, + stop, + }; +} + +/** A later human message means subsequent assistant output belongs below it. */ +export function noteSlackDraftConversationMessage( + conversation: SlackDraftConversation & { + messageTs?: string; + userId?: string; + botUserId?: string; + botId?: string; + subtype?: string; + }, +): void { + if ( + !conversation.messageTs || + !conversation.userId || + conversation.userId === conversation.botUserId || + conversation.botId || + conversation.subtype === "bot_message" + ) { + return; + } + + const drafts = activeDraftsByConversation.get(conversationKey(conversation)); + if (!drafts) { + return; + } + + for (const draft of drafts) { + if (!draft.messageTs) { + if ( + !draft.latestHumanMessageTs || + isLaterSlackMessage(conversation.messageTs, draft.latestHumanMessageTs) + ) { + // Slack can deliver the next message before chat.postMessage returns its timestamp. + draft.latestHumanMessageTs = conversation.messageTs; + } + continue; + } + if (isLaterSlackMessage(conversation.messageTs, draft.messageTs)) { + draft.onInterveningMessage(); + } + } +} diff --git a/extensions/slack/src/draft-stream.test.ts b/extensions/slack/src/draft-stream.test.ts index 7f24992f3af0..39c661c1ee30 100644 --- a/extensions/slack/src/draft-stream.test.ts +++ b/extensions/slack/src/draft-stream.test.ts @@ -1,6 +1,7 @@ // Slack tests cover draft stream plugin behavior. import { createMessageReceiptFromOutboundResults } from "openclaw/plugin-sdk/channel-outbound"; import { describe, expect, it, vi } from "vitest"; +import { noteSlackDraftConversationMessage } from "./draft-message-boundaries.js"; import { createSlackDraftStream } from "./draft-stream.js"; type DraftStreamParams = Parameters[0]; @@ -29,7 +30,9 @@ function slackDraftSendResult(messageId: string, channelId = "C123") { function createDraftStreamHarness( params: { + accountId?: string; maxChars?: number; + threadTs?: string; send?: DraftSendFn; edit?: DraftEditFn; eventScope?: DraftStreamParams["eventScope"]; @@ -45,9 +48,12 @@ function createDraftStreamHarness( target: "channel:C123", cfg: TEST_CFG, token: "xoxb-test", + accountId: params.accountId, + conversationChannelId: "C123", throttleMs: 250, maxChars: params.maxChars, eventScope: params.eventScope, + resolveThreadTs: params.threadTs ? () => params.threadTs : undefined, send, edit, remove, @@ -194,6 +200,327 @@ describe("createSlackDraftStream", () => { expect(stream.messageId()).toBe("333.444"); }); + it("continues below a human message that interrupts an in-progress Slack reply", async () => { + const accountId = "interrupted-reply"; + const send = vi + .fn() + .mockResolvedValueOnce(slackDraftSendResult("100.100")) + .mockResolvedValueOnce(slackDraftSendResult("100.300")); + const { stream, edit, remove } = createDraftStreamHarness({ + accountId, + threadTs: "100.000", + send, + }); + + stream.update("_looking into the original question_"); + await stream.flush(); + + noteSlackDraftConversationMessage({ + accountId, + channelId: "C123", + threadTs: "100.000", + messageTs: "100.200", + userId: "U_OWNER", + botUserId: "U_BOT", + }); + + expect(stream.messageId()).toBeUndefined(); + expect(remove).not.toHaveBeenCalled(); + + stream.update("_incorporating your clarification_"); + await stream.flush(); + stream.update("_checking one last detail_"); + await stream.flush(); + + expect(send).toHaveBeenCalledTimes(2); + expect(edit).toHaveBeenCalledWith( + "C123", + "100.300", + "_checking one last detail_", + expect.objectContaining({ accountId }), + ); + expect(stream.messageId()).toBe("100.300"); + }); + + it("keeps moving below repeated interruptions from different participants", async () => { + const accountId = "multiple-participants"; + const send = vi + .fn() + .mockResolvedValueOnce(slackDraftSendResult("100.100")) + .mockResolvedValueOnce(slackDraftSendResult("100.300")) + .mockResolvedValueOnce(slackDraftSendResult("100.500")); + const { stream, edit, remove } = createDraftStreamHarness({ + accountId, + threadTs: "100.000", + send, + }); + + stream.update("_first update_"); + await stream.flush(); + noteSlackDraftConversationMessage({ + accountId, + channelId: "C123", + threadTs: "100.000", + messageTs: "100.200", + userId: "U_OWNER", + }); + + stream.update("_second update_"); + await stream.flush(); + noteSlackDraftConversationMessage({ + accountId, + channelId: "C123", + threadTs: "100.000", + messageTs: "100.400", + userId: "U_COLLEAGUE", + }); + + stream.update("_third update_"); + await stream.flush(); + + expect(send).toHaveBeenCalledTimes(3); + expect(edit).not.toHaveBeenCalled(); + expect(remove).not.toHaveBeenCalled(); + expect(stream.messageId()).toBe("100.500"); + }); + + it("reconciles an interruption received before Slack returns the first preview id", async () => { + const accountId = "interruption-during-send"; + let finishFirstSend: ((value: ReturnType) => void) | undefined; + const firstSend = new Promise>((resolve) => { + finishFirstSend = resolve; + }); + const send = vi + .fn() + .mockImplementationOnce(async () => await firstSend) + .mockResolvedValueOnce(slackDraftSendResult("100.300")); + const { stream, edit, remove } = createDraftStreamHarness({ + accountId, + threadTs: "100.000", + send, + }); + + stream.update("_checking the original request_"); + const firstFlush = stream.flush(); + await vi.waitFor(() => { + expect(send).toHaveBeenCalledOnce(); + }); + + noteSlackDraftConversationMessage({ + accountId, + channelId: "C123", + threadTs: "100.000", + messageTs: "100.200", + userId: "U_OWNER", + }); + finishFirstSend?.(slackDraftSendResult("100.100")); + await firstFlush; + + expect(stream.messageId()).toBeUndefined(); + expect(remove).not.toHaveBeenCalled(); + + stream.update("_incorporating the newer clarification_"); + await stream.flush(); + + expect(send).toHaveBeenCalledTimes(2); + expect(edit).not.toHaveBeenCalled(); + expect(stream.messageId()).toBe("100.300"); + }); + + it("keeps direct-message previews after the latest unthreaded human message", async () => { + const accountId = "unthreaded-direct-message"; + const send = vi + .fn() + .mockResolvedValueOnce(slackDraftSendResult("100.100")) + .mockResolvedValueOnce(slackDraftSendResult("100.300")); + const { stream } = createDraftStreamHarness({ accountId, send }); + + stream.update("_looking into this_"); + await stream.flush(); + noteSlackDraftConversationMessage({ + accountId, + channelId: "C123", + messageTs: "100.200", + userId: "U_OWNER", + }); + stream.update("_looking into this_"); + await stream.flush(); + + expect(send).toHaveBeenCalledTimes(2); + expect(stream.messageId()).toBe("100.300"); + }); + + it("keeps simultaneous Enterprise Grid conversations isolated by workspace", async () => { + const accountId = "enterprise-grid"; + const eventScope = { + apiAppId: "A_TEST", + enterpriseId: "E_TEST", + isEnterpriseInstall: true as const, + teamId: "T_FIRST", + client: {} as NonNullable["client"], + }; + const send = vi + .fn() + .mockResolvedValueOnce(slackDraftSendResult("100.100")) + .mockResolvedValueOnce(slackDraftSendResult("100.300")); + const { stream, edit } = createDraftStreamHarness({ + accountId, + eventScope, + threadTs: "100.000", + send, + }); + + stream.update("_first workspace_"); + await stream.flush(); + noteSlackDraftConversationMessage({ + accountId, + teamId: "T_SECOND", + channelId: "C123", + threadTs: "100.000", + messageTs: "100.200", + userId: "U_OTHER_WORKSPACE", + }); + stream.update("_still in the first workspace_"); + await stream.flush(); + + expect(send).toHaveBeenCalledOnce(); + expect(edit).toHaveBeenCalledOnce(); + + noteSlackDraftConversationMessage({ + accountId, + teamId: "T_FIRST", + channelId: "C123", + threadTs: "100.000", + messageTs: "100.200", + userId: "U_OWNER", + }); + stream.update("_after the real interruption_"); + await stream.flush(); + + expect(send).toHaveBeenCalledTimes(2); + expect(stream.messageId()).toBe("100.300"); + }); + + it("ignores older, duplicate, unrelated, and bot-authored conversation events", async () => { + const accountId = "irrelevant-events"; + const { stream, send, edit } = createDraftStreamHarness({ + accountId, + threadTs: "100.000", + }); + + stream.update("_still working_"); + await stream.flush(); + + for (const event of [ + { channelId: "C123", threadTs: "100.000", messageTs: "111.111", userId: "U_OWNER" }, + { channelId: "C123", threadTs: "100.000", messageTs: "111.222", userId: "U_OWNER" }, + { channelId: "C123", threadTs: "200.000", messageTs: "111.333", userId: "U_OWNER" }, + { channelId: "C_OTHER", threadTs: "100.000", messageTs: "111.333", userId: "U_OWNER" }, + { + channelId: "C123", + threadTs: "100.000", + messageTs: "111.333", + userId: "U_BOT", + botUserId: "U_BOT", + }, + { + channelId: "C123", + threadTs: "100.000", + messageTs: "111.333", + userId: "U_OTHER_BOT", + botId: "B_OTHER", + }, + ]) { + noteSlackDraftConversationMessage({ accountId, ...event }); + } + + stream.update("_latest update_"); + await stream.flush(); + + expect(send).toHaveBeenCalledTimes(1); + expect(edit).toHaveBeenCalledOnce(); + expect(stream.messageId()).toBe("111.222"); + }); + + it("continues observing conversation boundaries while the final preview edit is in flight", async () => { + const accountId = "interrupted-final-edit"; + let finishFinalEdit: (() => void) | undefined; + const finalEdit = new Promise((resolve) => { + finishFinalEdit = resolve; + }); + const { stream, edit, remove } = createDraftStreamHarness({ + accountId, + threadTs: "100.000", + }); + + stream.update("_checking the last detail_"); + await stream.flush(); + await stream.seal(); + const finalizing = stream.finalizeMessage("111.222", async () => { + await finalEdit; + }); + + noteSlackDraftConversationMessage({ + accountId, + channelId: "C123", + threadTs: "100.000", + messageTs: "111.333", + userId: "U_OWNER", + }); + finishFinalEdit?.(); + + await expect(finalizing).resolves.toBe(false); + expect(stream.messageId()).toBeUndefined(); + expect(remove).not.toHaveBeenCalled(); + expect(edit).toHaveBeenCalledWith( + "C123", + "111.222", + "_checking the last detail_", + expect.objectContaining({ accountId }), + ); + }); + + it("does not finalize a preview invalidated while the stream was being sealed", async () => { + const accountId = "interrupted-sealed-preview"; + const { stream, edit } = createDraftStreamHarness({ accountId, threadTs: "100.000" }); + const finalize = vi.fn(async () => {}); + + stream.update("_nearly finished_"); + await stream.flush(); + await stream.seal(); + noteSlackDraftConversationMessage({ + accountId, + channelId: "C123", + threadTs: "100.000", + messageTs: "111.333", + userId: "U_OWNER", + }); + + await expect(stream.finalizeMessage("111.222", finalize)).resolves.toBe(false); + expect(finalize).not.toHaveBeenCalled(); + expect(edit).not.toHaveBeenCalled(); + }); + + it("stops observing conversation boundaries once the preview is finalized", async () => { + const accountId = "finalized-preview"; + const { stream } = createDraftStreamHarness({ accountId, threadTs: "100.000" }); + + stream.update("_finished_"); + await stream.flush(); + await stream.seal(); + await stream.finalizeMessage("111.222", async () => {}); + + noteSlackDraftConversationMessage({ + accountId, + channelId: "C123", + threadTs: "100.000", + messageTs: "111.333", + userId: "U_OWNER", + }); + + expect(stream.messageId()).toBe("111.222"); + }); + it("stops when text exceeds max chars", async () => { const { stream, send, edit, warn } = createDraftStreamHarness({ maxChars: 5 }); diff --git a/extensions/slack/src/draft-stream.ts b/extensions/slack/src/draft-stream.ts index b926f381db7e..6e46fbebf424 100644 --- a/extensions/slack/src/draft-stream.ts +++ b/extensions/slack/src/draft-stream.ts @@ -4,6 +4,7 @@ import type { Block, KnownBlock } from "@slack/web-api"; import { createFinalizableDraftStreamControlsForState } from "openclaw/plugin-sdk/channel-outbound"; import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts"; import { deleteSlackMessage, editSlackMessage } from "./actions.js"; +import { trackSlackDraftMessage } from "./draft-message-boundaries.js"; import { formatSlackError } from "./errors.js"; import { SLACK_TEXT_LIMIT } from "./limits.js"; import type { SlackEventScope } from "./monitor/event-scope.js"; @@ -20,6 +21,7 @@ type SlackDraftStream = { seal: () => Promise; stop: () => void; forceNewMessage: () => void; + finalizeMessage: (messageId: string, editFinal: () => Promise) => Promise; messageId: () => string | undefined; channelId: () => string | undefined; }; @@ -36,6 +38,7 @@ export function createSlackDraftStream(params: { cfg: OpenClawConfig; token: string; accountId?: string; + conversationChannelId?: string; eventScope?: SlackEventScope; identity?: SlackSendIdentity; maxChars?: number; @@ -57,6 +60,8 @@ export function createSlackDraftStream(params: { let streamMessageId: string | undefined; let streamChannelId: string | undefined; + let untrackConversationBoundary: (() => void) | undefined; + let lastVisibleUpdate: { text: string; blocks?: (Block | KnownBlock)[] } | undefined; let lastSentKey = ""; const streamState = { stopped: false, final: false }; @@ -92,13 +97,25 @@ export function createSlackDraftStream(params: { ...(params.eventScope ? { client: params.eventScope.client } : {}), ...(blocks ? { blocks } : {}), }); + lastVisibleUpdate = { text: trimmed, ...(blocks ? { blocks } : {}) }; return; } + const threadTs = params.resolveThreadTs?.(); + const pendingBoundary = params.conversationChannelId + ? trackSlackDraftMessage({ + accountId: params.accountId, + teamId: params.eventScope?.teamId, + channelId: params.conversationChannelId, + threadTs, + onInterveningMessage: forceNewMessage, + }) + : undefined; + untrackConversationBoundary = pendingBoundary?.stop; const sent = await send(params.target, trimmed, { cfg: params.cfg, token: params.token, accountId: params.accountId, - threadTs: params.resolveThreadTs?.(), + threadTs, identity: params.identity, ...(params.eventScope ? { client: params.eventScope.client, enterpriseEventScope: params.eventScope } @@ -109,17 +126,34 @@ export function createSlackDraftStream(params: { streamChannelId = sent.channelId || streamChannelId; streamMessageId = sent.messageId || streamMessageId; if (!streamChannelId || !streamMessageId) { + stopTrackingConversationBoundary(); streamState.stopped = true; params.warn?.("slack stream preview stopped (missing identifiers from sendMessage)"); return; } + lastVisibleUpdate = { text: trimmed, ...(blocks ? { blocks } : {}) }; + if (pendingBoundary && params.conversationChannelId === streamChannelId) { + pendingBoundary.setMessageTs(streamMessageId); + } else { + stopTrackingConversationBoundary(); + const tracker = trackSlackDraftMessage({ + accountId: params.accountId, + teamId: params.eventScope?.teamId, + channelId: streamChannelId, + threadTs, + messageTs: streamMessageId, + onInterveningMessage: forceNewMessage, + }); + untrackConversationBoundary = tracker.stop; + } params.onMessageSent?.(); } catch (err) { + stopTrackingConversationBoundary(); streamState.stopped = true; params.warn?.(`slack stream preview failed: ${formatSlackError(err)}`); } }; - const { loop, update, discardPending } = + const { loop, update, discardPending, seal } = createFinalizableDraftStreamControlsForState({ throttleMs, state: streamState, @@ -128,17 +162,25 @@ export function createSlackDraftStream(params: { isEmpty: (value) => !normalizeUpdate(value).text.trim(), }); + const stopTrackingConversationBoundary = () => { + untrackConversationBoundary?.(); + untrackConversationBoundary = undefined; + }; + const stop = () => { + stopTrackingConversationBoundary(); streamState.stopped = true; loop.stop(); }; const clear = async () => { + stopTrackingConversationBoundary(); await discardPending(); const channelId = streamChannelId; const messageId = streamMessageId; streamChannelId = undefined; streamMessageId = undefined; + lastVisibleUpdate = undefined; lastSentKey = ""; if (!channelId || !messageId) { return; @@ -155,22 +197,62 @@ export function createSlackDraftStream(params: { }; const forceNewMessage = () => { + stopTrackingConversationBoundary(); streamMessageId = undefined; streamChannelId = undefined; + lastVisibleUpdate = undefined; lastSentKey = ""; loop.resetPending(); }; + const discardPendingAndStopTracking = async () => { + stopTrackingConversationBoundary(); + await discardPending(); + }; + + const finalizeMessage = async ( + messageId: string, + editFinal: () => Promise, + ): Promise => { + const channelId = streamChannelId; + const previousUpdate = lastVisibleUpdate; + if (!channelId || streamMessageId !== messageId || !previousUpdate) { + return false; + } + + await editFinal(); + if (streamChannelId === channelId && streamMessageId === messageId) { + stopTrackingConversationBoundary(); + return true; + } + + // A human spoke while the final edit was in flight. Preserve the earlier + // progress they responded to and let the final answer land below them. + try { + await edit(channelId, messageId, previousUpdate.text, { + cfg: params.cfg, + token: params.token, + accountId: params.accountId, + ...(params.eventScope ? { client: params.eventScope.client } : {}), + ...(previousUpdate.blocks ? { blocks: previousUpdate.blocks } : {}), + }); + } catch (err) { + params.warn?.(`slack stream preview restore failed: ${formatSlackError(err)}`); + } + return false; + }; + params.log?.(`slack stream preview ready (maxChars=${maxChars}, throttleMs=${throttleMs})`); return { update, flush: loop.flush, clear, - discardPending, - seal: discardPending, + discardPending: discardPendingAndStopTracking, + seal, stop, forceNewMessage, + finalizeMessage, messageId: () => streamMessageId, channelId: () => streamChannelId, }; diff --git a/extensions/slack/src/monitor/events/messages.test.ts b/extensions/slack/src/monitor/events/messages.test.ts index cb14f8ebfe45..0b026a9f9218 100644 --- a/extensions/slack/src/monitor/events/messages.test.ts +++ b/extensions/slack/src/monitor/events/messages.test.ts @@ -9,10 +9,16 @@ import { const SLACK_INGRESS_LIFECYCLE_CONTEXT_KEY = "openclawIngressLifecycle"; -const { messageQueueMock, messageAllowMock, inboundInfoSpy } = vi.hoisted(() => ({ - messageQueueMock: vi.fn(), - messageAllowMock: vi.fn(), - inboundInfoSpy: vi.fn(), +const { messageQueueMock, messageAllowMock, inboundInfoSpy, noteConversationMessageMock } = + vi.hoisted(() => ({ + messageQueueMock: vi.fn(), + messageAllowMock: vi.fn(), + inboundInfoSpy: vi.fn(), + noteConversationMessageMock: vi.fn(), + })); + +vi.mock("../../draft-message-boundaries.js", () => ({ + noteSlackDraftConversationMessage: (...args: unknown[]) => noteConversationMessageMock(...args), })); vi.mock("openclaw/plugin-sdk/runtime-env", async (importOriginal) => { @@ -115,6 +121,7 @@ function requireMessageHandler(handler: MessageHandler | null): MessageHandler { function resetMessageMocks(): void { messageQueueMock.mockClear(); messageAllowMock.mockReset().mockResolvedValue([]); + noteConversationMessageMock.mockClear(); } beforeAll(async () => { @@ -465,6 +472,14 @@ describe("registerSlackMessageEvents", () => { expect(handleSlackMessage).toHaveBeenCalledTimes(1); expect(messageQueueMock).not.toHaveBeenCalled(); + expect(noteConversationMessageMock).toHaveBeenCalledWith( + expect.objectContaining({ + channelId: "D1", + messageTs: "123.456", + userId: "U1", + botUserId: "U_BOT", + }), + ); }); it("passes thread_broadcast events to the message handler", async () => { @@ -515,6 +530,14 @@ describe("registerSlackMessageEvents", () => { expect(message?.text).toBe("assistant wrapped user text"); expect(message?.ts).toBe("123.456"); expect(message?.thread_ts).toBe("123.000"); + expect(noteConversationMessageMock).toHaveBeenCalledWith( + expect.objectContaining({ + channelId: "D1", + threadTs: "123.000", + messageTs: "123.456", + userId: "UREAL123", + }), + ); expect(message?.assistant_thread).toEqual({ channel_id: "D1", thread_ts: "123.000", @@ -794,6 +817,13 @@ describe("registerSlackMessageEvents", () => { expect(inboundLogLines()).toEqual([ "Inbound app_mention slack:T_TEST:channel:C123:user:U1 -> bot:U_BOT (channel, 14 chars)", ]); + expect(noteConversationMessageMock).toHaveBeenCalledWith( + expect.objectContaining({ + channelId: "C123", + messageTs: "123.789", + userId: "U1", + }), + ); }); it("logs channel app_mention receipts with zero chars when text is absent", async () => { diff --git a/extensions/slack/src/monitor/events/messages.ts b/extensions/slack/src/monitor/events/messages.ts index 1c04489c426e..4a37b376c072 100644 --- a/extensions/slack/src/monitor/events/messages.ts +++ b/extensions/slack/src/monitor/events/messages.ts @@ -12,6 +12,7 @@ import { normalizeOptionalString as asString, } from "openclaw/plugin-sdk/string-coerce-runtime"; import { enqueueSystemEvent } from "openclaw/plugin-sdk/system-event-runtime"; +import { noteSlackDraftConversationMessage } from "../../draft-message-boundaries.js"; import type { SlackAppMentionEvent, SlackMessageEvent } from "../../types.js"; import { normalizeSlackChannelType } from "../channel-type.js"; import type { SlackMonitorContext } from "../context.js"; @@ -193,6 +194,23 @@ export function registerSlackMessageEvents(params: { }) { const { ctx, handleSlackMessage } = params; + const noteConversationMessage = ( + message: SlackMessageEvent | SlackAppMentionEvent, + eventScope?: SlackEventScope, + ) => { + noteSlackDraftConversationMessage({ + accountId: ctx.accountId, + teamId: eventScope?.teamId, + channelId: message.channel, + threadTs: message.thread_ts, + messageTs: message.ts ?? message.event_ts, + userId: asString(message.user), + botUserId: ctx.botUserId, + botId: asString(message.bot_id), + subtype: "subtype" in message ? asString(message.subtype) : undefined, + }); + }; + const resolveEventScope = (args: { body: unknown; context: AllMiddlewareArgs["context"]; @@ -250,6 +268,7 @@ export function registerSlackMessageEvents(params: { ctx, }); if (assistantChangedInbound) { + noteConversationMessage(assistantChangedInbound, eventScope); await handleSlackMessage(assistantChangedInbound, { source: "message", ...(eventScope ? { eventScope } : {}), @@ -292,6 +311,7 @@ export function registerSlackMessageEvents(params: { return; } + noteConversationMessage(message, eventScope); await handleSlackMessage(message, { source: "message", ...(eventScope ? { eventScope } : {}), @@ -373,6 +393,7 @@ export function registerSlackMessageEvents(params: { }), ); + noteConversationMessage(mention, eventScope); await handleSlackMessage(mention as unknown as SlackMessageEvent, { source: "app_mention", wasMentioned: true, diff --git a/extensions/slack/src/monitor/message-handler/dispatch-progress.ts b/extensions/slack/src/monitor/message-handler/dispatch-progress.ts index 220fd7c5a27a..8c31a94b3593 100644 --- a/extensions/slack/src/monitor/message-handler/dispatch-progress.ts +++ b/extensions/slack/src/monitor/message-handler/dispatch-progress.ts @@ -78,6 +78,7 @@ export function createSlackProgressRuntime(runtimeParams: { cfg, token: ctx.botToken, accountId: account.accountId, + conversationChannelId: message.channel, ...(prepared.eventScope ? { eventScope: prepared.eventScope } : {}), // Impersonated Slack messages cannot be deleted. Keep the temporary // preview app-authored and apply custom identity only to final delivery. 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 908275dfc772..fbe4c63d95ca 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 @@ -282,6 +282,10 @@ function createDraftStreamStub() { seal: vi.fn(noopAsync), stop: vi.fn(noop), forceNewMessage: vi.fn(), + finalizeMessage: vi.fn(async (_messageId: string, editFinal: () => Promise) => { + await editFinal(); + return true; + }), messageId: () => "171234.567", channelId: () => "C123", }; @@ -1230,6 +1234,45 @@ describe("dispatchPreparedSlackMessage preview fallback", () => { expectDeliverReplyCall(0, FINAL_REPLY_TEXT); }); + it("posts the final below a human message that interrupted the live preview", async () => { + let messageId: string | undefined = "171234.567"; + const draftStream = { + ...createDraftStreamStub(), + flush: vi.fn(async () => { + messageId = undefined; + }), + messageId: () => messageId, + channelId: () => (messageId ? "C123" : undefined), + }; + createSlackDraftStreamMock.mockReturnValueOnce(draftStream); + + await dispatchPreparedSlackMessage(createPreparedSlackMessage()); + + expect(finalizeSlackPreviewEditMock).not.toHaveBeenCalled(); + expect(deliverRepliesMock).toHaveBeenCalledOnce(); + expectDeliverReplyCall(0, FINAL_REPLY_TEXT); + }); + + it("posts the final below a human message received while the preview was sealing", async () => { + let messageId: string | undefined = "171234.567"; + const draftStream = { + ...createDraftStreamStub(), + seal: vi.fn(async () => { + messageId = undefined; + }), + finalizeMessage: vi.fn(async () => false), + messageId: () => messageId, + channelId: () => (messageId ? "C123" : undefined), + }; + createSlackDraftStreamMock.mockReturnValueOnce(draftStream); + + await dispatchPreparedSlackMessage(createPreparedSlackMessage()); + + expect(finalizeSlackPreviewEditMock).not.toHaveBeenCalled(); + expect(deliverRepliesMock).toHaveBeenCalledOnce(); + expectDeliverReplyCall(0, FINAL_REPLY_TEXT); + }); + it.each([ { name: "ASCII", text: "x".repeat(4_001) }, { name: "UTF-8", text: "界".repeat(1_334) }, @@ -4446,6 +4489,48 @@ describe("dispatchPreparedSlackMessage preview fallback", () => { ]); }); + it("delivers TTS below a human interruption received while its preview was flushing", async () => { + let messageId: string | undefined = "171234.567"; + const draftStream = { + ...createDraftStreamStub(), + flush: vi.fn(async () => { + messageId = undefined; + }), + messageId: () => messageId, + channelId: () => (messageId ? "C123" : undefined), + }; + createSlackDraftStreamMock.mockReturnValueOnce(draftStream); + mockedDispatchSequence = [ + { + kind: "final", + payload: { + mediaUrl: "https://example.com/tts.mp3", + audioAsVoice: true, + spokenText: "Spoken answer", + ttsSupplement: { spokenText: "Spoken answer" }, + }, + }, + ]; + + await dispatchPreparedSlackMessage(createPreparedSlackMessage()); + + expect(finalizeSlackPreviewEditMock).not.toHaveBeenCalled(); + expect(deliverRepliesMock).toHaveBeenCalledOnce(); + const delivered = requireRecord( + requireMockCall(deliverRepliesMock, 0, "deliver replies")[0], + "deliver replies params", + ); + expect(delivered.replies).toEqual([ + { + text: "Spoken answer", + mediaUrl: "https://example.com/tts.mp3", + audioAsVoice: true, + spokenText: "Spoken answer", + ttsSupplement: { spokenText: "Spoken answer" }, + }, + ]); + }); + it("delivers complete oversized TTS text together with its media", async () => { const spokenText = "x".repeat(4_001); finalizeSlackPreviewEditMock.mockResolvedValueOnce(undefined); diff --git a/extensions/slack/src/monitor/message-handler/dispatch.ts b/extensions/slack/src/monitor/message-handler/dispatch.ts index ff0a3f477b8d..3ac3e8fde402 100644 --- a/extensions/slack/src/monitor/message-handler/dispatch.ts +++ b/extensions/slack/src/monitor/message-handler/dispatch.ts @@ -166,23 +166,28 @@ export async function dispatchPreparedSlackMessage(prepared: PreparedSlackMessag previewFinalTextFitsEdit && trimmedFinalText.length > 0 ) { + await draftStream.flush(); const channelId = draftStream.channelId(); const messageId = draftStream.messageId(); if (channelId && messageId) { const finalThreadTs = delivery.usedReplyThreadTs ?? statusThreadTs; - await draftStream.flush(); await draftStream.seal(); try { - await finalizeSlackPreviewEdit({ - client: slackClient, - token: ctx.botToken, - accountId: account.accountId, - channelId, - messageId, - text: previewFinalText, - ...(slackBlocks?.length ? { blocks: slackBlocks } : {}), - threadTs: finalThreadTs, + const finalized = await draftStream.finalizeMessage(messageId, async () => { + await finalizeSlackPreviewEdit({ + client: slackClient, + token: ctx.botToken, + accountId: account.accountId, + channelId, + messageId, + text: previewFinalText, + ...(slackBlocks?.length ? { blocks: slackBlocks } : {}), + threadTs: finalThreadTs, + }); }); + if (!finalized) { + throw new Error("Slack preview moved below a newer conversation message"); + } } catch (err) { logVerbose( `slack: preview final edit failed; falling back to standard send (${formatSlackError(err)})`, @@ -265,16 +270,21 @@ export async function dispatchPreparedSlackMessage(prepared: PreparedSlackMessag if (delivery.hasDelivered({ kind: info.kind, payload, threadTs: edit.threadTs })) { return; } - await finalizeSlackPreviewEdit({ - client: slackClient, - token: ctx.botToken, - accountId: account.accountId, - channelId: preview.channelId, - messageId: preview.messageId, - text: edit.text, - ...(edit.blocks?.length ? { blocks: edit.blocks } : {}), - threadTs: edit.threadTs, + const finalized = await draftStream?.finalizeMessage(preview.messageId, async () => { + await finalizeSlackPreviewEdit({ + client: slackClient, + token: ctx.botToken, + accountId: account.accountId, + channelId: preview.channelId, + messageId: preview.messageId, + text: edit.text, + ...(edit.blocks?.length ? { blocks: edit.blocks } : {}), + threadTs: edit.threadTs, + }); }); + if (!finalized) { + throw new Error("Slack preview moved below a newer conversation message"); + } if (!ttsSupplement) { emitSlackMessageSentHooks({ ...messageSentHookContext,