From 3883d7365e6281ae1f120e076872cc91ecd28780 Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Thu, 11 Jun 2026 20:09:38 +0900 Subject: [PATCH] fix(qqbot): guard silent-final tool flushing --- CHANGELOG.md | 1 - .../engine/gateway/outbound-dispatch.test.ts | 96 ++++++++++++++ .../src/engine/gateway/outbound-dispatch.ts | 120 ++++++++++-------- 3 files changed, 162 insertions(+), 55 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 1f9ee715a65c..49dc13d729b4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -34,7 +34,6 @@ Docs: https://docs.openclaw.ai - UI: require explicit user intent before opening chat sessions and drain restored chat queues after session switches. (#91480) Thanks @TurboTheTurtle. - Android: avoid the `dataSync` foreground-service type for persistent nodes. (#80082) Thanks @davelutztx. - Native hooks: bound relay lifetimes so abandoned native hook connections cannot linger indefinitely. (#91550) Thanks @joshavant. -- QQBot: flush buffered tool-visible output before suppressing a silent non-streaming final block (empty, `NO_REPLY`, or `[SKIP]`), so tool progress visible with streaming enabled is no longer dropped when streaming is disabled. (#92074) Thanks @sliverp. ## 2026.6.5 diff --git a/extensions/qqbot/src/engine/gateway/outbound-dispatch.test.ts b/extensions/qqbot/src/engine/gateway/outbound-dispatch.test.ts index 67dde63e2e71..04faa2e24940 100644 --- a/extensions/qqbot/src/engine/gateway/outbound-dispatch.test.ts +++ b/extensions/qqbot/src/engine/gateway/outbound-dispatch.test.ts @@ -109,6 +109,7 @@ function makeInboundRuntime(): GatewayPluginRuntime["channel"]["inbound"] { function makeRuntime(params: { onFinalize?: (ctx: Record) => void; isControlCommandMessage?: (text?: string, cfg?: unknown) => boolean; + skipFreshSettledDelivery?: boolean; onDispatch?: (dispatcherOptions: { deliver: ( payload: { text?: string; mediaUrl?: string; mediaUrls?: string[]; audioAsVoice?: boolean }, @@ -119,6 +120,7 @@ function makeRuntime(params: { info: { kind: string; reason: "empty" | "silent" | "heartbeat" }, ) => void; onSettled?: () => unknown; + onFreshSettledDelivery?: () => unknown; }) => Promise; onDeliver?: ( deliver: ( @@ -160,6 +162,7 @@ function makeRuntime(params: { info: { kind: string; reason: "empty" | "silent" | "heartbeat" }, ) => void; onSettled?: () => unknown; + onFreshSettledDelivery?: () => unknown; }; } ).dispatcherOptions; @@ -169,6 +172,9 @@ function makeRuntime(params: { await params.onDeliver?.(dispatcherOptions.deliver); } await dispatcherOptions.onSettled?.(); + if (!params.skipFreshSettledDelivery) { + await dispatcherOptions.onFreshSettledDelivery?.(); + } }), finalizeInboundContext: vi.fn((rawCtx: Record) => { params.onFinalize?.(rawCtx); @@ -478,6 +484,96 @@ describe("dispatchOutbound", () => { expect(sendMediaMock).not.toHaveBeenCalled(); }); + it("waits for fresh settled delivery after a skipped silent block", async () => { + vi.useFakeTimers(); + const runtime = makeRuntime({ + onDispatch: async ({ deliver, onSkip }) => { + await deliver({ text: "visible tool message" }, { kind: "tool" }); + onSkip?.({ text: "NO_REPLY" }, { kind: "block", reason: "silent" }); + await vi.advanceTimersByTimeAsync(60_000); + expect(sendTextMock).not.toHaveBeenCalled(); + }, + }); + + await dispatchOutbound(makeInbound(), { + runtime, + cfg: {}, + account: { ...account, config: { streaming: false } }, + }); + + expect(sendTextMock.mock.calls.map((call) => call[1])).toEqual(["visible tool message"]); + expect(sendMediaMock).not.toHaveBeenCalled(); + }); + + it("does not send stale tool fallback when fresh settled delivery is suppressed", async () => { + vi.useFakeTimers(); + const runtime = makeRuntime({ + skipFreshSettledDelivery: true, + onDispatch: async ({ deliver, onSkip }) => { + await deliver({ text: "stale visible tool message" }, { kind: "tool" }); + onSkip?.({ text: "NO_REPLY" }, { kind: "block", reason: "silent" }); + }, + }); + + await dispatchOutbound(makeInbound(), { + runtime, + cfg: {}, + account: { ...account, config: { streaming: false } }, + }); + + expect(sendTextMock).not.toHaveBeenCalled(); + expect(sendMediaMock).not.toHaveBeenCalled(); + expect(vi.getTimerCount()).toBe(0); + }); + + it("bounds tool media flushes without racing the fallback timer", async () => { + vi.useFakeTimers(); + sendMediaMock.mockImplementationOnce(() => new Promise(() => {})); + sendMediaMock.mockImplementationOnce(() => new Promise(() => {})); + const firstMediaUrl = "https://example.com/progress-1.png"; + const secondMediaUrl = "https://example.com/progress-2.png"; + const runtime = makeRuntime({ + onDispatch: async ({ deliver, onSkip }) => { + await deliver({ mediaUrl: firstMediaUrl }, { kind: "tool" }); + await deliver({ mediaUrl: secondMediaUrl }, { kind: "tool" }); + await deliver({ text: "visible tool message" }, { kind: "tool" }); + onSkip?.({ text: "NO_REPLY" }, { kind: "block", reason: "silent" }); + }, + }); + + const dispatchPromise = dispatchOutbound(makeInbound(), { + runtime, + cfg: {}, + account: { ...account, config: { streaming: false } }, + }); + + await vi.advanceTimersByTimeAsync(90_000); + await dispatchPromise; + + expect(sendMediaMock).toHaveBeenCalledTimes(2); + expect(sendTextMock.mock.calls.map((call) => call[1])).toEqual(["visible tool message"]); + }); + + it("clears the media timeout after a successful silent-final flush", async () => { + vi.useFakeTimers(); + const mediaUrl = "https://example.com/progress.png"; + const runtime = makeRuntime({ + onDispatch: async ({ deliver, onSkip }) => { + await deliver({ mediaUrl }, { kind: "tool" }); + onSkip?.({ text: "NO_REPLY" }, { kind: "block", reason: "silent" }); + }, + }); + + await dispatchOutbound(makeInbound(), { + runtime, + cfg: {}, + account: { ...account, config: { streaming: false } }, + }); + + expect(sendMediaMock).toHaveBeenCalledTimes(1); + expect(vi.getTimerCount()).toBe(0); + }); + it.each([ { name: "empty text", payload: {} }, { name: "silent token", payload: { text: "NO_REPLY" } }, diff --git a/extensions/qqbot/src/engine/gateway/outbound-dispatch.ts b/extensions/qqbot/src/engine/gateway/outbound-dispatch.ts index 7869085258c2..50ef819ea4d4 100644 --- a/extensions/qqbot/src/engine/gateway/outbound-dispatch.ts +++ b/extensions/qqbot/src/engine/gateway/outbound-dispatch.ts @@ -175,38 +175,53 @@ export async function dispatchOutbound( }; // ---- Tool fallback ---- + const sendToolMediaWithTimeout = async ( + mediaUrl: string, + labels: { resultError: string; thrownError: string }, + ): Promise => { + const ac = new AbortController(); + let mediaTimeoutId: ReturnType | null = null; + try { + const result = await Promise.race([ + sendMedia({ + to: qualifiedTarget, + text: "", + mediaUrl, + accountId: account.accountId, + replyToId: event.messageId, + account, + }).then((r) => { + if (ac.signal.aborted) { + return { channel: "qqbot", error: "suppressed" } as OutboundResult; + } + return r; + }), + new Promise((resolve) => { + mediaTimeoutId = setTimeout(() => { + ac.abort(); + resolve({ channel: "qqbot", error: "timeout" }); + }, TOOL_MEDIA_SEND_TIMEOUT); + }), + ]); + if (result.error) { + log?.error(`${labels.resultError}: ${result.error}`); + } + } catch (err) { + log?.error(`${labels.thrownError}: ${String(err)}`); + } finally { + if (mediaTimeoutId) { + clearTimeout(mediaTimeoutId); + } + } + }; + const sendToolFallback = async (): Promise => { if (toolMediaUrls.length > 0) { for (const mediaUrl of toolMediaUrls) { - const ac = new AbortController(); - try { - const result = await Promise.race([ - sendMedia({ - to: qualifiedTarget, - text: "", - mediaUrl, - accountId: account.accountId, - replyToId: event.messageId, - account, - }).then((r) => { - if (ac.signal.aborted) { - return { channel: "qqbot", error: "suppressed" } as OutboundResult; - } - return r; - }), - new Promise((resolve) => { - setTimeout(() => { - ac.abort(); - resolve({ channel: "qqbot", error: "timeout" }); - }, TOOL_MEDIA_SEND_TIMEOUT); - }), - ]); - if (result.error) { - log?.error(`Tool fallback error: ${result.error}`); - } - } catch (err) { - log?.error(`Tool fallback failed: ${String(err)}`); - } + await sendToolMediaWithTimeout(mediaUrl, { + resultError: "Tool fallback error", + thrownError: "Tool fallback failed", + }); } return; } @@ -240,7 +255,7 @@ export async function dispatchOutbound( toolRenewalCount++; } toolOnlyTimeoutId = setTimeout(() => { - if (!hasBlockResponse && !toolFallbackSent) { + if (!hasBlockResponse && !toolFallbackSent && !skippedSilentBlockResponse) { toolFallbackSent = true; void sendToolFallback().catch(() => {}); } @@ -286,21 +301,10 @@ export async function dispatchOutbound( const urlsToSend = [...toolMediaUrls]; toolMediaUrls.length = 0; for (const mediaUrl of urlsToSend) { - try { - const result = await sendMedia({ - to: qualifiedTarget, - text: "", - mediaUrl, - accountId: account.accountId, - replyToId: event.messageId, - account, - }); - if (result.error) { - log?.error(`Tool media forward error: ${result.error}`); - } - } catch (err) { - log?.error(`Tool media forward failed: ${String(err)}`); - } + await sendToolMediaWithTimeout(mediaUrl, { + resultError: "Tool media forward error", + thrownError: "Tool media forward failed", + }); } } @@ -620,16 +624,14 @@ export async function dispatchOutbound( (info.reason === "silent" || info.reason === "empty") ) { skippedSilentBlockResponse = true; - markBlockResponse(); } }, - onSettled: async () => { - if ( - skippedSilentBlockResponse && - !hasVisibleBlockResponse && - (await flushPendingToolDeliveriesOnce()) - ) { - return { visibleReplySent: true }; + onFreshSettledDelivery: async () => { + if (skippedSilentBlockResponse && !hasVisibleBlockResponse) { + markBlockResponse(); + if (await flushPendingToolDeliveriesOnce()) { + return { visibleReplySent: true }; + } } return undefined; }, @@ -668,13 +670,23 @@ export async function dispatchOutbound( } catch { if (timeoutId) { clearTimeout(timeoutId); + timeoutId = null; } } finally { + if (timeoutId) { + clearTimeout(timeoutId); + timeoutId = null; + } if (toolOnlyTimeoutId) { clearTimeout(toolOnlyTimeoutId); toolOnlyTimeoutId = null; } - if (toolDeliverCount > 0 && !hasBlockResponse && !toolFallbackSent) { + if ( + toolDeliverCount > 0 && + !hasBlockResponse && + !toolFallbackSent && + !skippedSilentBlockResponse + ) { toolFallbackSent = true; await sendToolFallback(); }