fix(slack): isolate queued reply drafts (#107820)

extensions/slack/src/monitor/message-handler/dispatch.preview-fallback.test.ts

@@ -101,6 +101,7 @@ let capturedReplyOptions:
         summary?: string;
       }) => Promise<void> | void;
       onPartialReply?: (payload: { text: string }) => Promise<void> | void;
+      onQueuedFollowupAdmitted?: () => Promise<void> | void;
     }
   | undefined;
 let capturedStatusReactionOptions: { enabled?: boolean; initialEmoji?: string } | undefined;
@@ -3286,6 +3287,24 @@ describe("dispatchPreparedSlackMessage preview fallback", () => {
     expect(draftStream.forceNewMessage).toHaveBeenCalledTimes(1);
   });
 
+  it("starts a new draft delivery target when a queued followup is admitted", async () => {
+    const draftStream = {
+      ...createDraftStreamStub(),
+      flush: vi.fn(noopAsync),
+    };
+    createSlackDraftStreamMock.mockReturnValueOnce(draftStream);
+    mockedSlackStreamingMode = "partial";
+    mockedSlackDraftMode = "replace";
+    mockedDispatchSequence = [];
+    mockedReplyOptionEvents = [{ kind: "partial", text: "first reply" }];
+
+    await dispatchPreparedSlackMessage(createPreparedSlackMessage({}));
+    await capturedReplyOptions?.onQueuedFollowupAdmitted?.();
+
+    expect(draftStream.flush).toHaveBeenCalledTimes(1);
+    expect(draftStream.forceNewMessage).toHaveBeenCalledTimes(1);
+  });
+
   it("can hide raw Slack command progress text by config", async () => {
     const draftStream = createDraftStreamStub();
     createSlackDraftStreamMock.mockReturnValueOnce(draftStream);

extensions/slack/src/monitor/message-handler/dispatch.ts

@@ -813,7 +813,7 @@ export async function dispatchPreparedSlackMessage(prepared: PreparedSlackMessag
       delivery.outcome = "success";
     }
   };
-  const deliveryTracker = createSlackEventDeliveryTracker();
+  let deliveryTracker = createSlackEventDeliveryTracker();
   const markPreviewPayloadDelivered = (params: {
     kind: ReplyDispatchKind;
     payload: ReplyPayload;
@@ -1869,6 +1869,17 @@ export async function dispatchPreparedSlackMessage(prepared: PreparedSlackMessag
       label: "Reasoning",
     });
   };
+  const resetDraftDeliveryState = () => {
+    hasStreamedMessage = false;
+    appendRenderedText = "";
+    appendSourceText = "";
+    statusUpdateCount = 0;
+  };
+  const resetDraftProgressState = () => {
+    reasoningProgressRawText = "";
+    previewToolProgressSuppressed = false;
+    previewToolProgressLines = [];
+  };
   const onDraftBoundary = !shouldUseDraftStream
     ? undefined
     : async () => {
@@ -1877,14 +1888,23 @@ export async function dispatchPreparedSlackMessage(prepared: PreparedSlackMessag
         // posts a new draft instead of editing the existing preview.
         if (hasStreamedMessage && streamMode !== "status_final") {
           draftStream?.forceNewMessage();
-          hasStreamedMessage = false;
-          appendRenderedText = "";
-          appendSourceText = "";
-          statusUpdateCount = 0;
+          resetDraftDeliveryState();
         }
-        reasoningProgressRawText = "";
-        previewToolProgressSuppressed = false;
-        previewToolProgressLines = [];
+        resetDraftProgressState();
+      };
+
+  const onQueuedFollowupAdmitted = !shouldUseDraftStream
+    ? undefined
+    : async () => {
+        // A queued input is a new visible reply even though it drains through
+        // this turn's callbacks. Do not let it edit or dedupe against this run.
+        await draftStream?.flush();
+        draftStream?.forceNewMessage();
+        draftPreviewCommitted = false;
+        observedFinalReplyDelivery = false;
+        deliveryTracker = createSlackEventDeliveryTracker();
+        resetDraftDeliveryState();
+        resetDraftProgressState();
       };
 
   let dispatchError: unknown;
@@ -1942,6 +1962,7 @@ export async function dispatchPreparedSlackMessage(prepared: PreparedSlackMessag
               },
         onAssistantMessageStart: onDraftBoundary,
         onReasoningEnd: onDraftBoundary,
+        onQueuedFollowupAdmitted,
         onReasoningStream:
           statusReactionsEnabled || previewToolProgressEnabled
             ? async (payload) => {

src/auto-reply/get-reply-options.types.ts

@@ -288,6 +288,8 @@ export type GetReplyOptions = {
   queuedDeliveryCorrelations?: QueuedReplyDeliveryCorrelation[];
   /** Tracks ownership transfer when this turn later drains as a queued followup. */
   queuedFollowupLifecycle?: QueuedReplyLifecycle;
+  /** Called after a queued followup owns the reply lane, before its model run starts. */
+  onQueuedFollowupAdmitted?: () => Promise<void> | void;
   /** Allow channel-owned progress UI while final/source reply delivery remains message-tool-only. */
   allowProgressCallbacksWhenSourceDeliverySuppressed?: boolean;
   /** Called when a suppressed source reply mode observes visible delivery through another path. */

src/auto-reply/reply/followup-runner.test.ts

@@ -3124,6 +3124,28 @@ describe("createFollowupRunner runtime config", () => {
     expect(events).toEqual(["begin", "run", "end"]);
   });
 
+  it("notifies the active dispatcher after queued followup admission", async () => {
+    const events: string[] = [];
+    runEmbeddedAgentMock.mockImplementationOnce(async () => {
+      events.push("run");
+      return { payloads: [], meta: {} };
+    });
+    const runner = createFollowupRunner({
+      typing: createMockTypingController(),
+      typingMode: "instant",
+      defaultModel: "openai/gpt-5.4",
+      opts: {
+        onQueuedFollowupAdmitted: () => {
+          events.push("admitted");
+        },
+      },
+    });
+
+    await runner(createQueuedRun());
+
+    expect(events).toEqual(["admitted", "run"]);
+  });
+
   it("resolves queued embedded followups before preflight helpers read config", async () => {
     const sourceConfig: OpenClawConfig = {
       skills: {

src/auto-reply/reply/followup-runner.ts

@@ -717,6 +717,9 @@ export function createFollowupRunner(params: {
       if (isFollowupRunAborted(effectiveQueued)) {
         return;
       }
+      // Channel delivery state belongs to one admitted run. Give the active
+      // dispatcher a boundary before callbacks from this followup can reuse it.
+      await opts?.onQueuedFollowupAdmitted?.();
       if (replyOperation.sessionId !== run.sessionId) {
         run = { ...run, sessionId: replyOperation.sessionId };
         effectiveQueued = { ...effectiveQueued, run };