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

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<string, Set<ActiveSlackDraft>>();
+
+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<ActiveSlackDraft>();
+  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();
+    }
+  }
+}

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<typeof createSlackDraftStream>[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<DraftSendFn>()
+      .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<DraftSendFn>()
+      .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<typeof slackDraftSendResult>) => void) | undefined;
+    const firstSend = new Promise<ReturnType<typeof slackDraftSendResult>>((resolve) => {
+      finishFirstSend = resolve;
+    });
+    const send = vi
+      .fn<DraftSendFn>()
+      .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<DraftSendFn>()
+      .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<DraftStreamParams["eventScope"]>["client"],
+    };
+    const send = vi
+      .fn<DraftSendFn>()
+      .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<void>((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 });
 

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<void>;
   stop: () => void;
   forceNewMessage: () => void;
+  finalizeMessage: (messageId: string, editFinal: () => Promise<void>) => Promise<boolean>;
   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<SlackDraftStreamUpdate>({
       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<void>,
+  ): Promise<boolean> => {
+    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,
   };

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 () => {

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,

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.

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<void>) => {
+      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);

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,