From 83a462bcdc104ed99cba1b84e5a2884634eeb530 Mon Sep 17 00:00:00 2001 From: reason-bot Date: Thu, 2 Jul 2026 14:41:26 -0400 Subject: [PATCH 1/2] fix(auto-reply): gate sendBlockReply on disableBlockStreaming The channel-side setting `channels..streaming.block.enabled=false` (e.g. Slack's) resolves to `disableBlockStreaming: true` on the get-reply options and correctly disables the streamed block-reply pipeline, but the plain block-delivery path was unaffected: inter-tool assistant-text blocks flowed through the reply dispatcher and became one outbound `chat.postMessage` per block (Slack: one turn producing a burst of same-second channel-root posts under a single traceId/spanId with no parentSpanId). Gate the drop in two places for defense in depth: 1. Reply dispatcher primitive: add `disableBlockStreaming` to `ReplyDispatcherOptions`; `sendBlockReply` returns `false` (matching the existing normalization-skip semantics) when it is set. Tool results and final replies stay untouched. 2. Runtime block-reply handler in `dispatch-from-config.ts`: early-return before the accumulate/TTS/normalize work when `replyOptions.disableBlockStreaming === true`, while preserving the opt-in behaviour of durable reasoning/commentary lanes. Thread the flag from the incoming `replyOptions` through both `dispatchInboundMessageWithBufferedDispatcher` and `dispatchInboundMessageWithDispatcher` so channels that only set it on `replyOptions` (like Slack) get gated at the dispatcher too. Adds `reply-dispatcher.disable-block-streaming.test.ts` covering the disabled drop, the default pass-through, and explicit `false` pass-through. `vitest.auto-reply-reply` project: 2,597 tests pass. Related: openclaw#23791, openclaw#25592. --- src/auto-reply/dispatch.ts | 10 +++ src/auto-reply/reply/dispatch-from-config.ts | 14 +++ ...dispatcher.disable-block-streaming.test.ts | 85 +++++++++++++++++++ src/auto-reply/reply/reply-dispatcher.ts | 26 +++++- 4 files changed, 134 insertions(+), 1 deletion(-) create mode 100644 src/auto-reply/reply/reply-dispatcher.disable-block-streaming.test.ts diff --git a/src/auto-reply/dispatch.ts b/src/auto-reply/dispatch.ts index 907ca9a969b1a..5e5bd9c86bed5 100644 --- a/src/auto-reply/dispatch.ts +++ b/src/auto-reply/dispatch.ts @@ -623,6 +623,12 @@ export async function dispatchInboundMessageWithBufferedDispatcher(params: { deliver, beforeDeliver, silentReplyContext: params.dispatcherOptions.silentReplyContext ?? silentReplyContext, + // AIDEV-NOTE: Hoist `disableBlockStreaming` from the incoming replyOptions + // so the dispatcher can no-op `sendBlockReply`. The channel-provided + // `dispatcherOptions.disableBlockStreaming` (if any) takes precedence. + disableBlockStreaming: + params.dispatcherOptions.disableBlockStreaming ?? + params.replyOptions?.disableBlockStreaming, }); markReplyPayloadSendingBeforeDeliverInstalled(dispatcher, replyPayloadBeforeDeliver); try { @@ -687,6 +693,10 @@ export async function dispatchInboundMessageWithDispatcher(params: { ...params.dispatcherOptions, beforeDeliver: composedBeforeDeliver, silentReplyContext: params.dispatcherOptions.silentReplyContext ?? silentReplyContext, + // AIDEV-NOTE: mirror the hoist done in dispatchInboundMessageWithBufferedDispatcher + // so this non-buffered dispatch path honors the same gate. + disableBlockStreaming: + params.dispatcherOptions.disableBlockStreaming ?? params.replyOptions?.disableBlockStreaming, }); markReplyPayloadSendingBeforeDeliverInstalled(dispatcher, replyPayloadBeforeDeliver); return await dispatchInboundMessage({ diff --git a/src/auto-reply/reply/dispatch-from-config.ts b/src/auto-reply/reply/dispatch-from-config.ts index 8ca482085060a..9b2dd3f9ced68 100644 --- a/src/auto-reply/reply/dispatch-from-config.ts +++ b/src/auto-reply/reply/dispatch-from-config.ts @@ -3352,6 +3352,20 @@ export async function dispatchReplyFromConfig( if (payload.isCommentary === true && !commentaryPayloadsEnabled) { return; } + // AIDEV-NOTE: `disableBlockStreaming` (channel setting e.g. + // `channels.slack.streaming.block.enabled=false`) suppresses + // plain assistant-text block payloads at emission time. This + // avoids the expensive normalize/TTS/accumulate work and + // prevents each inter-tool text block from becoming its own + // outbound message. Durable reasoning/commentary lanes stay on + // their own opt-in switches above so they aren't affected. + if ( + params.replyOptions?.disableBlockStreaming === true && + payload.isReasoning !== true && + payload.isCommentary !== true + ) { + return; + } // Accumulate block text for TTS generation after streaming. // Exclude status notices — they are informational UI signals // and must not be synthesised into the spoken reply. Display diff --git a/src/auto-reply/reply/reply-dispatcher.disable-block-streaming.test.ts b/src/auto-reply/reply/reply-dispatcher.disable-block-streaming.test.ts new file mode 100644 index 0000000000000..213a9ebf6cf6a --- /dev/null +++ b/src/auto-reply/reply/reply-dispatcher.disable-block-streaming.test.ts @@ -0,0 +1,85 @@ +// Tests the dispatcher-level gate for `disableBlockStreaming`. +// +// AIDEV-NOTE: Regression test for the block-payload leak surfaced on +// 2026-07-02: channels that set `channels..streaming.block.enabled=false` +// expect inter-tool assistant-text blocks to be dropped, but the plain +// (non-streamed) block path still enqueued each block through the reply +// dispatcher and produced one outbound `chat.postMessage` per inter-tool +// text block on Slack. The dispatcher now no-ops `sendBlockReply` when +// `disableBlockStreaming` is set, while leaving tool results and final +// replies untouched. +import { describe, expect, it } from "vitest"; +import type { ReplyPayload } from "../types.js"; +import { createReplyDispatcher } from "./reply-dispatcher.js"; + +describe("createReplyDispatcher with disableBlockStreaming", () => { + it("drops block replies without delivering them", async () => { + const delivered: Array<{ kind: string; text: string }> = []; + + const dispatcher = createReplyDispatcher({ + deliver: async (payload, info) => { + delivered.push({ kind: info.kind, text: payload.text ?? "" }); + }, + disableBlockStreaming: true, + }); + + // Block replies are silently dropped and report as not-enqueued. + expect(dispatcher.sendBlockReply({ text: "inter-tool block one" } as ReplyPayload)).toBe(false); + expect(dispatcher.sendBlockReply({ text: "inter-tool block two" } as ReplyPayload)).toBe(false); + + // Tool results and final replies still go through as before. + expect(dispatcher.sendToolResult({ text: "tool result" } as ReplyPayload)).toBe(true); + expect(dispatcher.sendFinalReply({ text: "final answer" } as ReplyPayload)).toBe(true); + + dispatcher.markComplete(); + await dispatcher.waitForIdle(); + + expect(delivered).toEqual([ + { kind: "tool", text: "tool result" }, + { kind: "final", text: "final answer" }, + ]); + // Queued counts reflect actual work performed: no block enqueues. + expect(dispatcher.getQueuedCounts()).toEqual({ tool: 1, block: 0, final: 1 }); + // The drops are not counted as cancellations (they never entered the + // queue), matching the semantics of a normalization skip. + expect(dispatcher.getCancelledCounts?.()).toEqual({ tool: 0, block: 0, final: 0 }); + }); + + it("delivers block replies as usual when disableBlockStreaming is unset", async () => { + const delivered: Array<{ kind: string; text: string }> = []; + + const dispatcher = createReplyDispatcher({ + deliver: async (payload, info) => { + delivered.push({ kind: info.kind, text: payload.text ?? "" }); + }, + }); + + expect(dispatcher.sendBlockReply({ text: "block one" } as ReplyPayload)).toBe(true); + expect(dispatcher.sendBlockReply({ text: "block two" } as ReplyPayload)).toBe(true); + dispatcher.markComplete(); + await dispatcher.waitForIdle(); + + expect(delivered).toEqual([ + { kind: "block", text: "block one" }, + { kind: "block", text: "block two" }, + ]); + expect(dispatcher.getQueuedCounts()).toEqual({ tool: 0, block: 2, final: 0 }); + }); + + it("delivers block replies as usual when disableBlockStreaming is explicitly false", async () => { + const delivered: string[] = []; + + const dispatcher = createReplyDispatcher({ + deliver: async (payload) => { + delivered.push(payload.text ?? ""); + }, + disableBlockStreaming: false, + }); + + expect(dispatcher.sendBlockReply({ text: "keeps flowing" } as ReplyPayload)).toBe(true); + dispatcher.markComplete(); + await dispatcher.waitForIdle(); + + expect(delivered).toEqual(["keeps flowing"]); + }); +}); diff --git a/src/auto-reply/reply/reply-dispatcher.ts b/src/auto-reply/reply/reply-dispatcher.ts index 965089fda6c7c..ed69231aeeeca 100644 --- a/src/auto-reply/reply/reply-dispatcher.ts +++ b/src/auto-reply/reply/reply-dispatcher.ts @@ -110,6 +110,17 @@ export type ReplyDispatcherOptions = { onSkip?: ReplyDispatchSkipHandler; /** Human-like delay between block replies for natural rhythm. */ humanDelay?: HumanDelayConfig; + /** + * When true, `sendBlockReply` is a no-op (returns false without enqueueing). + * Mirrors the channel-side gate for `channels..streaming.block.enabled=false`: + * suppresses inter-tool assistant text blocks from becoming their own outbound + * messages while leaving tool results and final replies untouched. + * + * AIDEV-NOTE: Reasoning/commentary payloads travel their own channel-owned + * lanes and are gated separately by `reasoningPayloadsEnabled` / + * `commentaryPayloadsEnabled`; do not use this flag to suppress them. + */ + disableBlockStreaming?: boolean; beforeDeliver?: ReplyDispatchBeforeDeliver; onBeforeDeliverCancelled?: ReplyDispatchCancelHandler; /** Observe each queued payload settling, including cancellation and delivery failure. */ @@ -317,7 +328,20 @@ export function createReplyDispatcher(options: ReplyDispatcherOptions): ReplyDis return { sendToolResult: (payload) => enqueue("tool", payload), - sendBlockReply: (payload) => enqueue("block", payload), + sendBlockReply: (payload) => { + // AIDEV-NOTE: dispatcher-level gate for `disableBlockStreaming`. + // The runtime already avoids emitting inter-tool block payloads in the + // streamed pipeline when `blockStreamingEnabled=false`, but plain + // (non-streamed) block deliveries reached this dispatcher unconditionally + // and produced one outbound `chat.postMessage` per inter-tool text block + // on Slack (see openclaw#23791 / #25592 for related regressions). Returning + // false here matches the "payload skipped before delivery" semantics that + // enqueue() already uses on normalization failure. + if (options.disableBlockStreaming === true) { + return false; + } + return enqueue("block", payload); + }, sendFinalReply: (payload) => enqueue("final", payload), appendBeforeDeliver: (hook) => { const previousBeforeDeliver = beforeDeliver; From 2a7ad055299ba41b626d188da2bce1e8d57559b5 Mon Sep 17 00:00:00 2001 From: reason-bot Date: Thu, 2 Jul 2026 15:39:14 -0400 Subject: [PATCH 2/2] test(auto-reply): integration tests for disableBlockStreaming block-payload gate Two tests in dispatch-from-config.test.ts mirror the existing isReasoning/isCommentary suppression patterns to cover the new gate: 1. When replyOptions.disableBlockStreaming is true, plain inter-tool text blocks emitted via opts.onBlockReply do not reach dispatcher.sendBlockReply; the final reply still lands via sendFinalReply. This is the Slack-in-production repro. 2. When disableBlockStreaming is unset, the default block-delivery path is unchanged and both blocks reach the dispatcher (defense against accidentally turning the gate on by default). Runs inside the existing dispatch-from-config test harness so the gate is exercised against the real runtime handler, not just the dispatcher primitive. --- .../reply/dispatch-from-config.test.ts | 72 +++++++++++++++++++ 1 file changed, 72 insertions(+) diff --git a/src/auto-reply/reply/dispatch-from-config.test.ts b/src/auto-reply/reply/dispatch-from-config.test.ts index d14cc4b29ad15..9ccd653a3452d 100644 --- a/src/auto-reply/reply/dispatch-from-config.test.ts +++ b/src/auto-reply/reply/dispatch-from-config.test.ts @@ -8112,6 +8112,78 @@ describe("dispatchReplyFromConfig", () => { expect(blockReplySentTexts).toContain("The answer is 42"); }); + it("suppresses inter-tool text-block replies when replyOptions.disableBlockStreaming is true", async () => { + // AIDEV-NOTE: Regression test for the inter-tool block-payload leak on + // channels that set `channels..streaming.block.enabled=false` + // (Slack in production). Before the runtime-side gate in + // dispatch-from-config.ts, plain text blocks emitted between tool calls + // reached `dispatcher.sendBlockReply` even though the channel had opted + // out of block streaming entirely, producing one outbound message per + // block. The gate must let tool results and the final reply through + // unchanged. + setNoAbort(); + const dispatcher = createDispatcher(); + const ctx = buildTestCtx({ Provider: "slack" }); + const blockReplySentTexts: string[] = []; + const replyResolver = async ( + _ctx: MsgContext, + opts?: GetReplyOptions, + ): Promise => { + await opts?.onBlockReply?.({ text: "inter-tool block one" }); + await opts?.onBlockReply?.({ text: "inter-tool block two" }); + return { text: "final answer" }; + }; + (dispatcher.sendBlockReply as ReturnType).mockImplementation( + (payload: ReplyPayload) => { + if (payload.text) { + blockReplySentTexts.push(payload.text); + } + return true; + }, + ); + await dispatchReplyFromConfig({ + ctx, + cfg: emptyConfig, + dispatcher, + replyOptions: { disableBlockStreaming: true }, + replyResolver, + }); + expect(blockReplySentTexts).toEqual([]); + expect(dispatcher.sendFinalReply).toHaveBeenCalledWith({ text: "final answer" }); + }); + + it("still delivers block replies when disableBlockStreaming is not set", async () => { + // Belt-and-suspenders: the default path (no flag) keeps existing behavior. + setNoAbort(); + const dispatcher = createDispatcher(); + const ctx = buildTestCtx({ Provider: "slack" }); + const blockReplySentTexts: string[] = []; + const replyResolver = async ( + _ctx: MsgContext, + opts?: GetReplyOptions, + ): Promise => { + await opts?.onBlockReply?.({ text: "inter-tool block one" }); + await opts?.onBlockReply?.({ text: "inter-tool block two" }); + return { text: "final answer" }; + }; + (dispatcher.sendBlockReply as ReturnType).mockImplementation( + (payload: ReplyPayload) => { + if (payload.text) { + blockReplySentTexts.push(payload.text); + } + return true; + }, + ); + await dispatchReplyFromConfig({ + ctx, + cfg: emptyConfig, + dispatcher, + replyResolver, + }); + expect(blockReplySentTexts).toEqual(["inter-tool block one", "inter-tool block two"]); + expect(dispatcher.sendFinalReply).toHaveBeenCalledWith({ text: "final answer" }); + }); + it("delivers opted-in block reasoning payloads without applying TTS", async () => { setNoAbort(); const dispatcher = createDispatcher();