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 daf17a1b4208..0f25357948bc 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 @@ -101,6 +101,7 @@ let capturedReplyOptions: summary?: string; }) => Promise | void; onPartialReply?: (payload: { text: string }) => Promise | void; + onQueuedFollowupAdmitted?: () => Promise | 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); diff --git a/extensions/slack/src/monitor/message-handler/dispatch.ts b/extensions/slack/src/monitor/message-handler/dispatch.ts index d1e7d4c19d15..a54e74106f84 100644 --- a/extensions/slack/src/monitor/message-handler/dispatch.ts +++ b/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) => { diff --git a/src/auto-reply/get-reply-options.types.ts b/src/auto-reply/get-reply-options.types.ts index bf160da55562..ac7854afe09a 100644 --- a/src/auto-reply/get-reply-options.types.ts +++ b/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; /** 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. */ diff --git a/src/auto-reply/reply/followup-runner.test.ts b/src/auto-reply/reply/followup-runner.test.ts index be0d51843e46..95afc07876f2 100644 --- a/src/auto-reply/reply/followup-runner.test.ts +++ b/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: { diff --git a/src/auto-reply/reply/followup-runner.ts b/src/auto-reply/reply/followup-runner.ts index c6f9fd4fc3eb..1ce564025e8b 100644 --- a/src/auto-reply/reply/followup-runner.ts +++ b/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 };