From 252e06af28fe5ad7f2978db117324cb77514260f Mon Sep 17 00:00:00 2001 From: "jinjing.zzj" Date: Tue, 1 Sep 2026 23:24:16 +0800 Subject: [PATCH 1/6] fix(acp-bridge): backpressure ACP NDJSON queue saturation instead of tearing down the channel When the decoded NDJSON queue saturates (slow consumer plus large session/update frames), the fail-closed guard tears down the whole ACP channel, killing every session multiplexed on it (#10162). Wait up to a bounded grace window (default 10s, overridable via queueSaturationGraceMs) for the consumer to drain before falling back to the original fail-closed guard. The wait pauses the pump, which backpressures the agent's stdout pipe while keeping the memory bound intact. Fire onQueueSaturated once per episode (wired to a daemon WARN in qwen serve) so saturation is visible before any eviction. Co-authored-by: Qwen-Coder --- packages/acp-bridge/src/ndJsonStream.test.ts | 85 ++++++++++ packages/acp-bridge/src/ndJsonStream.ts | 166 +++++++++++++++++-- packages/cli/src/serve/run-qwen-serve.ts | 25 ++- 3 files changed, 263 insertions(+), 13 deletions(-) diff --git a/packages/acp-bridge/src/ndJsonStream.test.ts b/packages/acp-bridge/src/ndJsonStream.test.ts index b3e5bcbe950..5ba6a9a9374 100644 --- a/packages/acp-bridge/src/ndJsonStream.test.ts +++ b/packages/acp-bridge/src/ndJsonStream.test.ts @@ -396,6 +396,7 @@ describe('ndJsonStream', () => { maxFrameBytes: 200, maxQueuedMessages: 2, maxQueuedBytes: 200, + queueSaturationGraceMs: 25, }), ); await vi.waitFor(() => @@ -422,6 +423,7 @@ describe('ndJsonStream', () => { maxFrameBytes: 200, maxQueuedMessages: 100, maxQueuedBytes: firstBytes + 1, + queueSaturationGraceMs: 25, }), ); await vi.waitFor(() => @@ -437,6 +439,88 @@ describe('ndJsonStream', () => { await stream.readable.cancel(); }); + it('backpressures a saturated queue until a transiently slow consumer drains', async () => { + let inputController!: ReadableStreamDefaultController; + const input = new ReadableStream({ + start(controller) { + inputController = controller; + }, + }); + const first = message('first', { text: 'x'.repeat(120) }); + const second = message('second', { text: 'y'.repeat(120) }); + const onQueueSaturated = vi.fn(); + const onTransportError = vi.fn(); + const stream = ndJsonStream( + new WritableStream(), + input, + { onQueueSaturated, onTransportError }, + limits({ + maxFrameBytes: 1024, + maxQueuedMessages: 2, + maxQueuedBytes: 220, + queueSaturationGraceMs: 5_000, + }), + ); + inputController.enqueue( + encoder.encode(`${JSON.stringify(first)}\n${JSON.stringify(second)}\n`), + ); + + // No consumer read yet, so both frames charge the decoded queue and the + // second one saturates it. The pump must wait instead of failing. + await vi.waitFor(() => expect(onQueueSaturated).toHaveBeenCalledOnce()); + expect(onTransportError).not.toHaveBeenCalled(); + + // Draining frees the queue; the waiting frame is then delivered. + const reader = stream.readable.getReader(); + await expect(reader.read()).resolves.toMatchObject({ + value: first, + done: false, + }); + inputController.close(); + await expect(reader.read()).resolves.toMatchObject({ + value: second, + done: false, + }); + await expect(reader.read()).resolves.toMatchObject({ done: true }); + expect(onQueueSaturated).toHaveBeenCalledOnce(); + expect(onTransportError).not.toHaveBeenCalled(); + reader.releaseLock(); + }); + + it('warns once before failing closed when the consumer stays stalled', async () => { + const first = message('first', { text: 'x'.repeat(120) }); + const second = message('second', { text: 'y'.repeat(120) }); + const onQueueSaturated = vi.fn(); + const onTransportError = vi.fn(); + const stream = ndJsonStream( + new WritableStream(), + byteStream([ + encoder.encode(`${JSON.stringify(first)}\n${JSON.stringify(second)}\n`), + ]), + { onQueueSaturated, onTransportError }, + limits({ + maxFrameBytes: 1024, + maxQueuedMessages: 2, + maxQueuedBytes: 220, + queueSaturationGraceMs: 25, + }), + ); + + await vi.waitFor(() => + expect(onTransportError).toHaveBeenCalledWith( + expect.any(NdJsonQueueLimitError), + ), + ); + expect(onQueueSaturated).toHaveBeenCalledOnce(); + expect(onQueueSaturated).toHaveBeenCalledWith( + expect.objectContaining({ + maxQueuedBytes: 220, + graceMs: 25, + }), + ); + await stream.readable.cancel(); + }); + it('bounds requests retained by the ACP SDK while responses are blocked', async () => { const cancel = vi.fn(); let inputController!: ReadableStreamDefaultController; @@ -481,6 +565,7 @@ describe('ndJsonStream', () => { maxFrameBytes: 1024, maxQueuedMessages: 2, maxQueuedBytes: 4096, + queueSaturationGraceMs: 25, }), ); const connection = new ClientSideConnection(() => ({}) as never, stream); diff --git a/packages/acp-bridge/src/ndJsonStream.ts b/packages/acp-bridge/src/ndJsonStream.ts index c0039e33ca8..bc7e5ab711f 100644 --- a/packages/acp-bridge/src/ndJsonStream.ts +++ b/packages/acp-bridge/src/ndJsonStream.ts @@ -14,17 +14,42 @@ export interface NdJsonMessageObservation { message: AnyMessage; } +export interface NdJsonQueueSaturationInfo { + requiredBytes: number; + availableBytes: number; + maxQueuedMessages: number; + maxQueuedBytes: number; + graceMs: number; +} + export interface NdJsonStreamHooks { onMessageReceived?: (bytes: number) => void; onMessageSent?: (bytes: number) => void; onMessageObserved?: (observation: NdJsonMessageObservation) => void; onTransportError?: (error: unknown) => void; + /** + * Fired once per saturation episode, before the bounded backpressure wait + * starts. Warns that the decoded queue is full BEFORE the fail-closed + * guard can fire (issue #10162). + */ + onQueueSaturated?: (info: NdJsonQueueSaturationInfo) => void; } +/** + * How long the bounded reader waits for the consumer to drain the decoded + * queue before falling back to the fail-closed `NdJsonQueueLimitError`. + * Transient slow consumers (channel clients reconnecting, blocked outbound + * SSE) resolve within this window and keep the channel alive; a genuinely + * stalled consumer still tears the transport down afterwards so the memory + * bound remains effective. + */ +export const NDJSON_QUEUE_SATURATION_GRACE_MS = 10_000; + export interface NdJsonStreamLimits { maxFrameBytes: number; maxQueuedMessages: number; maxQueuedBytes: number; + queueSaturationGraceMs?: number; } export type NdJsonInboundMessageValidator = (message: AnyMessage) => boolean; @@ -216,9 +241,23 @@ function createBoundedReadable( const minimumQueueCharge = Math.ceil( limits.maxQueuedBytes / limits.maxQueuedMessages, ); + const graceMs = + limits.queueSaturationGraceMs ?? NDJSON_QUEUE_SATURATION_GRACE_MS; let nextQueueCharge = minimumQueueCharge; let reader: ReadableStreamDefaultReader | undefined; let canceled = false; + // Single waiter: only the pump loop ever waits for queue space. + let wakeQueueWaiter: (() => void) | undefined; + const waitForQueueSpace = (timeoutMs: number): Promise => + new Promise((resolve) => { + const timer = setTimeout(finish, Math.max(0, timeoutMs)); + function finish() { + clearTimeout(timer); + wakeQueueWaiter = undefined; + resolve(); + } + wakeQueueWaiter = finish; + }); return new ReadableStream( { @@ -236,15 +275,23 @@ function createBoundedReadable( validateInboundMessage, fatalCleanEof, minimumQueueCharge, + graceMs, + waitForQueueSpace, (charge) => { nextQueueCharge = charge; }, () => canceled, ); }, + pull() { + // The consumer drained enough to want more: wake a pump that is + // backpressured on a saturated decoded queue. + wakeQueueWaiter?.(); + }, async cancel(reason) { canceled = true; pending.clear(); + wakeQueueWaiter?.(); if (reader) await cancelReader(reader, reason); }, }, @@ -267,6 +314,8 @@ async function pumpBoundedInput( validateInboundMessage: NdJsonInboundMessageValidator | undefined, fatalCleanEof: boolean, minimumQueueCharge: number, + queueSaturationGraceMs: number, + waitForQueueSpace: (timeoutMs: number) => Promise, setNextQueueCharge: (charge: number) => void, isCanceled: () => boolean, ): Promise { @@ -283,7 +332,7 @@ async function pumpBoundedInput( return; } if (!result.value) continue; - readBoundedChunk( + await readBoundedChunk( result.value, pending, controller, @@ -294,8 +343,12 @@ async function pumpBoundedInput( inboundRequests, validateInboundMessage, minimumQueueCharge, + queueSaturationGraceMs, + waitForQueueSpace, setNextQueueCharge, + isCanceled, ); + if (isCanceled()) return; } } catch (error) { if (isCanceled()) return; @@ -337,7 +390,7 @@ function readLegacyChunk( } } -function readBoundedChunk( +async function readBoundedChunk( chunk: Uint8Array, pending: BoundedFrameBuffer, controller: ReadableStreamDefaultController, @@ -348,8 +401,11 @@ function readBoundedChunk( inboundRequests: BoundedInboundRequestLedger, validateInboundMessage: NdJsonInboundMessageValidator | undefined, minimumQueueCharge: number, + queueSaturationGraceMs: number, + waitForQueueSpace: (timeoutMs: number) => Promise, setNextQueueCharge: (charge: number) => void, -): void { + isCanceled: () => boolean, +): Promise { let start = 0; let newline = chunk.indexOf(0x0a, start); while (newline !== -1) { @@ -363,15 +419,16 @@ function readBoundedChunk( continue; } const queueCharge = Math.max(frameBytes, minimumQueueCharge); - const availableBytes = controller.desiredSize; - if (availableBytes === null || queueCharge > availableBytes) { - throw new NdJsonQueueLimitError( - limits.maxQueuedMessages, - limits.maxQueuedBytes, - queueCharge, - Math.max(0, availableBytes ?? 0), - ); - } + await waitForDecodedQueueSpace({ + controller, + limits, + queueCharge, + queueSaturationGraceMs, + waitForQueueSpace, + hooks, + isCanceled, + }); + if (isCanceled()) return; setNextQueueCharge(queueCharge); handleBoundedLine( pending.take(current), @@ -388,6 +445,88 @@ function readBoundedChunk( if (start < chunk.length) pending.append(chunk.subarray(start)); } +interface DecodedQueueSpaceOptions { + controller: ReadableStreamDefaultController; + limits: NdJsonStreamLimits; + queueCharge: number; + queueSaturationGraceMs: number; + waitForQueueSpace: (timeoutMs: number) => Promise; + hooks: NdJsonStreamHooks | undefined; + isCanceled: () => boolean; +} + +/** + * Waits (bounded) for room in the decoded queue instead of failing the + * transport immediately (issue #10162). Saturating the queue means the + * consumer is slow; pausing the producer backpressures the agent's stdout + * pipe and keeps the memory bound intact. Only if the consumer stays slow + * for the whole grace window does the original fail-closed guard fire. + */ +async function waitForDecodedQueueSpace({ + controller, + limits, + queueCharge, + queueSaturationGraceMs, + waitForQueueSpace, + hooks, + isCanceled, +}: DecodedQueueSpaceOptions): Promise { + let availableBytes = controller.desiredSize; + if (availableBytes === null) { + throw new NdJsonQueueLimitError( + limits.maxQueuedMessages, + limits.maxQueuedBytes, + queueCharge, + 0, + ); + } + if (queueCharge <= availableBytes) return; + // A frame that would not fit even in a fully drained queue can never be + // rescued by waiting. + if (queueCharge > limits.maxQueuedBytes) { + throw new NdJsonQueueLimitError( + limits.maxQueuedMessages, + limits.maxQueuedBytes, + queueCharge, + Math.max(0, availableBytes), + ); + } + + callHook(hooks?.onQueueSaturated, { + requiredBytes: queueCharge, + availableBytes: Math.max(0, availableBytes), + maxQueuedMessages: limits.maxQueuedMessages, + maxQueuedBytes: limits.maxQueuedBytes, + graceMs: queueSaturationGraceMs, + }); + + const deadline = Date.now() + queueSaturationGraceMs; + while (queueCharge > availableBytes) { + if (isCanceled()) return; + const remainingMs = deadline - Date.now(); + if (remainingMs <= 0) { + throw new NdJsonQueueLimitError( + limits.maxQueuedMessages, + limits.maxQueuedBytes, + queueCharge, + Math.max(0, availableBytes), + ); + } + await waitForQueueSpace(remainingMs); + if (isCanceled()) return; + const nextAvailableBytes = controller.desiredSize; + if (nextAvailableBytes === null) { + throw new NdJsonQueueLimitError( + limits.maxQueuedMessages, + limits.maxQueuedBytes, + queueCharge, + 0, + ); + } + availableBytes = nextAvailableBytes; + } +} + function takeLegacyLineBytes( pending: Uint8Array[], current: Uint8Array, @@ -721,6 +860,9 @@ export function validateNdJsonStreamLimits(limits: NdJsonStreamLimits): void { ['maxFrameBytes', limits.maxFrameBytes], ['maxQueuedMessages', limits.maxQueuedMessages], ['maxQueuedBytes', limits.maxQueuedBytes], + ...(limits.queueSaturationGraceMs !== undefined + ? ([['queueSaturationGraceMs', limits.queueSaturationGraceMs]] as const) + : []), ] as const; for (const [name, value] of values) { if (!Number.isSafeInteger(value) || value <= 0) { diff --git a/packages/cli/src/serve/run-qwen-serve.ts b/packages/cli/src/serve/run-qwen-serve.ts index 5564d2584bb..3a58ead38a2 100644 --- a/packages/cli/src/serve/run-qwen-serve.ts +++ b/packages/cli/src/serve/run-qwen-serve.ts @@ -45,7 +45,10 @@ import { type BridgeEvent, } from '@qwen-code/acp-bridge/eventBus'; import { resolveSessionRestoreTimeoutMs } from '@qwen-code/acp-bridge/sessionRestoreTimeout'; -import type { NdJsonMessageObservation } from '@qwen-code/acp-bridge/ndJsonStream'; +import type { + NdJsonMessageObservation, + NdJsonQueueSaturationInfo, +} from '@qwen-code/acp-bridge/ndJsonStream'; import { getDeviceFlowRegistry } from './auth/device-flow.js'; import { consumeServeFastPathRejectedLoaderKeys, @@ -4900,6 +4903,23 @@ async function runQwenServeImpl( daemonLog, emitTelemetryLog: core.emitDaemonLog, }); + // Saturation now backpressures the agent pipe for a bounded grace + // window instead of tearing the channel down immediately (#10162). + // Warn at episode start so field diagnosis doesn't begin at the + // `channel exited` breadcrumb. + const warnAcpQueueSaturated = (info: NdJsonQueueSaturationInfo): void => { + try { + daemonLog.warn('ACP NDJSON decoded queue saturated', { + requiredBytes: info.requiredBytes, + availableBytes: info.availableBytes, + maxQueuedMessages: info.maxQueuedMessages, + maxQueuedBytes: info.maxQueuedBytes, + queueSaturationGraceMs: info.graceMs, + }); + } catch { + // Observability must not affect transport behavior. + } + }; const recordPromptQueueWait = (durationMs: number): void => { promptQueueWaitStats.count += 1; promptQueueWaitStats.totalMs += durationMs; @@ -5071,6 +5091,7 @@ async function runQwenServeImpl( bytes, message, }), + onQueueSaturated: warnAcpQueueSaturated, }, ...(acpChildExtraArgs(opts) ? { extraArgs: acpChildExtraArgs(opts) } @@ -5977,6 +5998,7 @@ async function runQwenServeImpl( bytes, message, }), + onQueueSaturated: warnAcpQueueSaturated, }, ...(acpChildExtraArgs(opts) ? { extraArgs: acpChildExtraArgs(opts) } @@ -6635,6 +6657,7 @@ async function runQwenServeImpl( bytes, message, }), + onQueueSaturated: warnAcpQueueSaturated, }, ...(acpChildExtraArgs(opts) ? { extraArgs: acpChildExtraArgs(opts) } From 1b853b14539a82937980f0534b09c053da99a136 Mon Sep 17 00:00:00 2001 From: "jinjing.zzj" Date: Tue, 1 Sep 2026 23:35:14 +0800 Subject: [PATCH 2/6] test(acp-bridge): cover saturation cancel, oversized frame, and grace validation Address review findings on #10162: add a test that cancels the readable while the pump is parked on a saturated queue (a missed cancel wake would expire the wait and fail closed), a test that a frame larger than maxQueuedBytes fails immediately without the saturation warning or the grace delay, a validation test for queueSaturationGraceMs, and tighten the onQueueSaturated payload assertion to all five fields. Co-authored-by: Qwen-Coder --- packages/acp-bridge/src/ndJsonStream.test.ts | 81 +++++++++++++++++++- 1 file changed, 78 insertions(+), 3 deletions(-) diff --git a/packages/acp-bridge/src/ndJsonStream.test.ts b/packages/acp-bridge/src/ndJsonStream.test.ts index 5ba6a9a9374..357e862f0aa 100644 --- a/packages/acp-bridge/src/ndJsonStream.test.ts +++ b/packages/acp-bridge/src/ndJsonStream.test.ts @@ -512,15 +512,90 @@ describe('ndJsonStream', () => { ), ); expect(onQueueSaturated).toHaveBeenCalledOnce(); - expect(onQueueSaturated).toHaveBeenCalledWith( - expect.objectContaining({ + const firstFrameBytes = + encoder.encode(JSON.stringify(first)).byteLength + 1; + const secondFrameBytes = + encoder.encode(JSON.stringify(second)).byteLength + 1; + expect(onQueueSaturated).toHaveBeenCalledWith({ + requiredBytes: secondFrameBytes, + availableBytes: 220 - firstFrameBytes, + maxQueuedMessages: 2, + maxQueuedBytes: 220, + graceMs: 25, + }); + await stream.readable.cancel(); + }); + + it('cancel while backpressured wakes the pump without a transport error', async () => { + let inputController!: ReadableStreamDefaultController; + const input = new ReadableStream({ + start(controller) { + inputController = controller; + }, + }); + const first = message('first', { text: 'x'.repeat(120) }); + const second = message('second', { text: 'y'.repeat(120) }); + const onQueueSaturated = vi.fn(); + const onTransportError = vi.fn(); + const stream = ndJsonStream( + new WritableStream(), + input, + { onQueueSaturated, onTransportError }, + limits({ + maxFrameBytes: 1024, + maxQueuedMessages: 2, + maxQueuedBytes: 220, + queueSaturationGraceMs: 200, + }), + ); + inputController.enqueue( + encoder.encode(`${JSON.stringify(first)}\n${JSON.stringify(second)}\n`), + ); + await vi.waitFor(() => expect(onQueueSaturated).toHaveBeenCalledOnce()); + + await stream.readable.cancel(); + // Outlast the grace window: a pump that missed the cancel wake would expire + // the wait and fail closed on the cancelled stream. + await new Promise((resolve) => setTimeout(resolve, 400)); + expect(onTransportError).not.toHaveBeenCalled(); + }); + + it('fails immediately for a frame that can never fit, without the saturation warning', async () => { + const big = message('big', { text: 'x'.repeat(300) }); + const onQueueSaturated = vi.fn(); + const onTransportError = vi.fn(); + const stream = ndJsonStream( + new WritableStream(), + byteStream([encoder.encode(`${JSON.stringify(big)}\n`)]), + { onQueueSaturated, onTransportError }, + limits({ + maxFrameBytes: 1024, + maxQueuedMessages: 2, maxQueuedBytes: 220, - graceMs: 25, + queueSaturationGraceMs: 5_000, }), ); + + await vi.waitFor(() => + expect(onTransportError).toHaveBeenCalledWith( + expect.any(NdJsonQueueLimitError), + ), + ); + expect(onQueueSaturated).not.toHaveBeenCalled(); await stream.readable.cancel(); }); + it('rejects a non-positive queueSaturationGraceMs', () => { + expect(() => + ndJsonStream( + new WritableStream(), + byteStream([]), + undefined, + limits({ queueSaturationGraceMs: 0 }), + ), + ).toThrow(RangeError); + }); + it('bounds requests retained by the ACP SDK while responses are blocked', async () => { const cancel = vi.fn(); let inputController!: ReadableStreamDefaultController; From 5141bedf198ea899609be2b5906125c50804d622 Mon Sep 17 00:00:00 2001 From: "jinjing.zzj" Date: Tue, 1 Sep 2026 23:43:40 +0800 Subject: [PATCH 3/6] fix(acp-bridge): dedupe queue-saturation warnings per episode, use a monotonic deadline Correctness review findings on #10162: a chronically borderline consumer saturated on every frame and re-fired onQueueSaturated indefinitely, flooding the daemon log the warning exists to make diagnosable. Track an episode: warn once per uninterrupted saturation episode and re-arm only after the queue fully drains. Also compute the grace deadline with performance.now() so wall-clock steps cannot extend or shorten the backpressure window. Adds a test pinning one warning across multiple saturating frames in a single episode. Co-authored-by: Qwen-Coder --- packages/acp-bridge/src/ndJsonStream.test.ts | 48 ++++++++++++++++++++ packages/acp-bridge/src/ndJsonStream.ts | 48 +++++++++++++++----- 2 files changed, 85 insertions(+), 11 deletions(-) diff --git a/packages/acp-bridge/src/ndJsonStream.test.ts b/packages/acp-bridge/src/ndJsonStream.test.ts index 357e862f0aa..37d0e7add3e 100644 --- a/packages/acp-bridge/src/ndJsonStream.test.ts +++ b/packages/acp-bridge/src/ndJsonStream.test.ts @@ -526,6 +526,54 @@ describe('ndJsonStream', () => { await stream.readable.cancel(); }); + it('warns once per saturation episode, not once per saturating frame', async () => { + let inputController!: ReadableStreamDefaultController; + const input = new ReadableStream({ + start(controller) { + inputController = controller; + }, + }); + const frames = [ + message('first', { text: 'x'.repeat(120) }), + message('second', { text: 'y'.repeat(120) }), + message('third', { text: 'z'.repeat(120) }), + ]; + const onQueueSaturated = vi.fn(); + const onTransportError = vi.fn(); + const stream = ndJsonStream( + new WritableStream(), + input, + { onQueueSaturated, onTransportError }, + limits({ + maxFrameBytes: 1024, + maxQueuedMessages: 3, + maxQueuedBytes: 220, + queueSaturationGraceMs: 5_000, + }), + ); + inputController.enqueue( + encoder.encode( + frames.map((frame) => `${JSON.stringify(frame)}\n`).join(''), + ), + ); + + // Frame one fits; frames two and three each saturate before the queue + // fully drains, but only the first saturation of the episode warns. + await vi.waitFor(() => expect(onQueueSaturated).toHaveBeenCalledOnce()); + const reader = stream.readable.getReader(); + for (const frame of frames) { + await expect(reader.read()).resolves.toMatchObject({ + value: frame, + done: false, + }); + } + inputController.close(); + await expect(reader.read()).resolves.toMatchObject({ done: true }); + expect(onQueueSaturated).toHaveBeenCalledOnce(); + expect(onTransportError).not.toHaveBeenCalled(); + reader.releaseLock(); + }); + it('cancel while backpressured wakes the pump without a transport error', async () => { let inputController!: ReadableStreamDefaultController; const input = new ReadableStream({ diff --git a/packages/acp-bridge/src/ndJsonStream.ts b/packages/acp-bridge/src/ndJsonStream.ts index bc7e5ab711f..e9313bc345a 100644 --- a/packages/acp-bridge/src/ndJsonStream.ts +++ b/packages/acp-bridge/src/ndJsonStream.ts @@ -5,6 +5,7 @@ */ import { createHash } from 'node:crypto'; +import { performance } from 'node:perf_hooks'; import { inspect } from 'node:util'; import type { AnyMessage, Stream } from '@agentclientprotocol/sdk'; @@ -29,8 +30,10 @@ export interface NdJsonStreamHooks { onTransportError?: (error: unknown) => void; /** * Fired once per saturation episode, before the bounded backpressure wait - * starts. Warns that the decoded queue is full BEFORE the fail-closed - * guard can fire (issue #10162). + * starts. An episode ends when the queue fully drains, so a chronically + * borderline consumer produces one warning per episode, not one per frame. + * Warns that the decoded queue is full BEFORE the fail-closed guard can + * fire (issue #10162). */ onQueueSaturated?: (info: NdJsonQueueSaturationInfo) => void; } @@ -246,6 +249,9 @@ function createBoundedReadable( let nextQueueCharge = minimumQueueCharge; let reader: ReadableStreamDefaultReader | undefined; let canceled = false; + // Dedup state for onQueueSaturated: one warning per uninterrupted + // saturation episode, reset once the queue fully drains. + const saturationEpisode = { warned: false }; // Single waiter: only the pump loop ever waits for queue space. let wakeQueueWaiter: (() => void) | undefined; const waitForQueueSpace = (timeoutMs: number): Promise => @@ -277,6 +283,7 @@ function createBoundedReadable( minimumQueueCharge, graceMs, waitForQueueSpace, + saturationEpisode, (charge) => { nextQueueCharge = charge; }, @@ -316,6 +323,7 @@ async function pumpBoundedInput( minimumQueueCharge: number, queueSaturationGraceMs: number, waitForQueueSpace: (timeoutMs: number) => Promise, + saturationEpisode: SaturationEpisodeState, setNextQueueCharge: (charge: number) => void, isCanceled: () => boolean, ): Promise { @@ -345,6 +353,7 @@ async function pumpBoundedInput( minimumQueueCharge, queueSaturationGraceMs, waitForQueueSpace, + saturationEpisode, setNextQueueCharge, isCanceled, ); @@ -403,6 +412,7 @@ async function readBoundedChunk( minimumQueueCharge: number, queueSaturationGraceMs: number, waitForQueueSpace: (timeoutMs: number) => Promise, + saturationEpisode: SaturationEpisodeState, setNextQueueCharge: (charge: number) => void, isCanceled: () => boolean, ): Promise { @@ -425,6 +435,7 @@ async function readBoundedChunk( queueCharge, queueSaturationGraceMs, waitForQueueSpace, + saturationEpisode, hooks, isCanceled, }); @@ -445,12 +456,17 @@ async function readBoundedChunk( if (start < chunk.length) pending.append(chunk.subarray(start)); } +interface SaturationEpisodeState { + warned: boolean; +} + interface DecodedQueueSpaceOptions { controller: ReadableStreamDefaultController; limits: NdJsonStreamLimits; queueCharge: number; queueSaturationGraceMs: number; waitForQueueSpace: (timeoutMs: number) => Promise; + saturationEpisode: SaturationEpisodeState; hooks: NdJsonStreamHooks | undefined; isCanceled: () => boolean; } @@ -468,6 +484,7 @@ async function waitForDecodedQueueSpace({ queueCharge, queueSaturationGraceMs, waitForQueueSpace, + saturationEpisode, hooks, isCanceled, }: DecodedQueueSpaceOptions): Promise { @@ -480,6 +497,11 @@ async function waitForDecodedQueueSpace({ 0, ); } + // The queue fully drained: the previous saturation episode is over and a + // new one may warn again. + if (availableBytes === limits.maxQueuedBytes) { + saturationEpisode.warned = false; + } if (queueCharge <= availableBytes) return; // A frame that would not fit even in a fully drained queue can never be // rescued by waiting. @@ -492,18 +514,22 @@ async function waitForDecodedQueueSpace({ ); } - callHook(hooks?.onQueueSaturated, { - requiredBytes: queueCharge, - availableBytes: Math.max(0, availableBytes), - maxQueuedMessages: limits.maxQueuedMessages, - maxQueuedBytes: limits.maxQueuedBytes, - graceMs: queueSaturationGraceMs, - }); + if (!saturationEpisode.warned) { + saturationEpisode.warned = true; + callHook(hooks?.onQueueSaturated, { + requiredBytes: queueCharge, + availableBytes: Math.max(0, availableBytes), + maxQueuedMessages: limits.maxQueuedMessages, + maxQueuedBytes: limits.maxQueuedBytes, + graceMs: queueSaturationGraceMs, + }); + } - const deadline = Date.now() + queueSaturationGraceMs; + // Monotonic: wall-clock steps must not extend or shorten the grace window. + const deadline = performance.now() + queueSaturationGraceMs; while (queueCharge > availableBytes) { if (isCanceled()) return; - const remainingMs = deadline - Date.now(); + const remainingMs = deadline - performance.now(); if (remainingMs <= 0) { throw new NdJsonQueueLimitError( limits.maxQueuedMessages, From 50a05d3e8bf311648f9cd8ea4d072eb7988b8fe7 Mon Sep 17 00:00:00 2001 From: "jinjing.zzj" Date: Wed, 2 Sep 2026 02:58:35 +0800 Subject: [PATCH 4/6] fix(web-shell): drop duplicated language binding in ChatEditor tests A stale merge left two `language` interface members and two destructured defaults in ChatEditor.test.tsx, breaking esbuild transform ("The symbol language has already been declared") in CI. Co-authored-by: Qwen-Coder Patrol-Run: qwen-pr-conflict/jmtj0ojdzar --- packages/web-shell/client/components/ChatEditor.test.tsx | 2 -- 1 file changed, 2 deletions(-) diff --git a/packages/web-shell/client/components/ChatEditor.test.tsx b/packages/web-shell/client/components/ChatEditor.test.tsx index e9e381913f6..3c7c07d8cd2 100644 --- a/packages/web-shell/client/components/ChatEditor.test.tsx +++ b/packages/web-shell/client/components/ChatEditor.test.tsx @@ -404,7 +404,6 @@ interface ChatEditorRenderProps { builtinAtProviders?: WebShellCustomization['builtinAtProviders']; atProviders?: WebShellCustomization['atProviders']; skills?: Array<{ name: string; description: string }>; - language?: WebShellLanguage; reasoning?: DaemonReasoningControls; onSelectReasoningEffort?: (value: ReasoningSelection) => Promise | void; } @@ -422,7 +421,6 @@ function renderChatEditorInto( language = 'en', renderComposerTagTooltip, onComposerTagClick, - language = 'en', ...chatEditorProps } = props; if (composerTags) { From 9a7ae1dee77c6ab7116f454e7fb6b354260ff611 Mon Sep 17 00:00:00 2001 From: yiliang114 Date: Wed, 2 Sep 2026 10:30:56 +0800 Subject: [PATCH 5/6] fix(acp-bridge): address saturation review feedback Clamp oversized timeout scheduling without changing the monotonic grace deadline, strengthen cancellation and episode regression coverage, and verify serve-side saturation hook wiring and log fields. Co-authored-by: Qwen-Coder --- packages/acp-bridge/src/ndJsonStream.test.ts | 59 ++++++++++++++++++- packages/acp-bridge/src/ndJsonStream.ts | 5 +- packages/cli/src/serve/run-qwen-serve.test.ts | 32 +++++++++- 3 files changed, 91 insertions(+), 5 deletions(-) diff --git a/packages/acp-bridge/src/ndJsonStream.test.ts b/packages/acp-bridge/src/ndJsonStream.test.ts index 37d0e7add3e..2d62a858eb9 100644 --- a/packages/acp-bridge/src/ndJsonStream.test.ts +++ b/packages/acp-bridge/src/ndJsonStream.test.ts @@ -567,9 +567,26 @@ describe('ndJsonStream', () => { done: false, }); } + + const nextFrames = [ + message('fourth', { text: 'a'.repeat(120) }), + message('fifth', { text: 'b'.repeat(120) }), + ]; + inputController.enqueue( + encoder.encode( + nextFrames.map((frame) => `${JSON.stringify(frame)}\n`).join(''), + ), + ); + await vi.waitFor(() => expect(onQueueSaturated).toHaveBeenCalledTimes(2)); + for (const frame of nextFrames) { + await expect(reader.read()).resolves.toMatchObject({ + value: frame, + done: false, + }); + } inputController.close(); await expect(reader.read()).resolves.toMatchObject({ done: true }); - expect(onQueueSaturated).toHaveBeenCalledOnce(); + expect(onQueueSaturated).toHaveBeenCalledTimes(2); expect(onTransportError).not.toHaveBeenCalled(); reader.releaseLock(); }); @@ -602,12 +619,48 @@ describe('ndJsonStream', () => { await vi.waitFor(() => expect(onQueueSaturated).toHaveBeenCalledOnce()); await stream.readable.cancel(); - // Outlast the grace window: a pump that missed the cancel wake would expire - // the wait and fail closed on the cancelled stream. + await vi.waitFor(() => expect(input.locked).toBe(false), { timeout: 100 }); + // Keep checking past the grace window for a late transport error. await new Promise((resolve) => setTimeout(resolve, 400)); expect(onTransportError).not.toHaveBeenCalled(); }); + it('caps the queue-space timer at the Node timeout maximum', async () => { + let inputController!: ReadableStreamDefaultController; + const input = new ReadableStream({ + start(controller) { + inputController = controller; + }, + }); + const setTimeoutSpy = vi.spyOn(globalThis, 'setTimeout'); + const onQueueSaturated = vi.fn(); + const stream = ndJsonStream( + new WritableStream(), + input, + { onQueueSaturated }, + limits({ + maxQueuedMessages: 2, + maxQueuedBytes: 220, + queueSaturationGraceMs: 2_147_483_648, + }), + ); + const first = message('first', { text: 'x'.repeat(120) }); + const second = message('second', { text: 'y'.repeat(120) }); + inputController.enqueue( + encoder.encode(`${JSON.stringify(first)}\n${JSON.stringify(second)}\n`), + ); + + try { + await vi.waitFor(() => expect(onQueueSaturated).toHaveBeenCalledOnce()); + expect(setTimeoutSpy.mock.calls.map((call) => call[1])).toContain( + 2_147_483_647, + ); + } finally { + await stream.readable.cancel(); + setTimeoutSpy.mockRestore(); + } + }); + it('fails immediately for a frame that can never fit, without the saturation warning', async () => { const big = message('big', { text: 'x'.repeat(300) }); const onQueueSaturated = vi.fn(); diff --git a/packages/acp-bridge/src/ndJsonStream.ts b/packages/acp-bridge/src/ndJsonStream.ts index e9313bc345a..743696e30a5 100644 --- a/packages/acp-bridge/src/ndJsonStream.ts +++ b/packages/acp-bridge/src/ndJsonStream.ts @@ -256,7 +256,10 @@ function createBoundedReadable( let wakeQueueWaiter: (() => void) | undefined; const waitForQueueSpace = (timeoutMs: number): Promise => new Promise((resolve) => { - const timer = setTimeout(finish, Math.max(0, timeoutMs)); + const timer = setTimeout( + finish, + Math.min(Math.max(0, timeoutMs), 2_147_483_647), + ); function finish() { clearTimeout(timer); wakeQueueWaiter = undefined; diff --git a/packages/cli/src/serve/run-qwen-serve.test.ts b/packages/cli/src/serve/run-qwen-serve.test.ts index 0085214b813..bba72d1b5d7 100644 --- a/packages/cli/src/serve/run-qwen-serve.test.ts +++ b/packages/cli/src/serve/run-qwen-serve.test.ts @@ -11632,7 +11632,10 @@ describe('runQwenServe Web Shell signals on RunHandle', () => { maxSessions: 1, ...extra, }, - { bridge: makeFakeBridge() }, + { + bridge: makeFakeBridge(), + daemonLogBaseDir: path.join(tmpDir, 'debug'), + }, ); } @@ -12143,6 +12146,22 @@ describe('runQwenServe Web Shell signals on RunHandle', () => { const handle = await bootHandle({ serveWebShell: false }); try { await handle.runtimeReady; + const saturationInfo = { + requiredBytes: 200, + availableBytes: 20, + maxQueuedMessages: 2, + maxQueuedBytes: 220, + graceMs: 10_000, + }; + for (const options of mockCreateSpawnChannelFactoryOptions) { + const hooks = options['pipeHooks'] as + | { + onQueueSaturated?: (info: typeof saturationInfo) => void; + } + | undefined; + expect(hooks?.onQueueSaturated).toEqual(expect.any(Function)); + hooks?.onQueueSaturated?.(saturationInfo); + } const pipeHooks = mockCreateSpawnChannelFactoryOptions.at(-1)?.[ 'pipeHooks' ] as @@ -12189,6 +12208,17 @@ describe('runQwenServe Web Shell signals on RunHandle', () => { } finally { await handle.close(); } + const logPath = path.join(tmpDir, 'debug', 'daemon', 'daemon.log'); + let logContent = ''; + await vi.waitFor(() => { + logContent = fs.readFileSync(logPath, 'utf8'); + expect(logContent).toContain('ACP NDJSON decoded queue saturated'); + }); + expect(logContent).toContain('requiredBytes=200'); + expect(logContent).toContain('availableBytes=20'); + expect(logContent).toContain('maxQueuedMessages=2'); + expect(logContent).toContain('maxQueuedBytes=220'); + expect(logContent).toContain('queueSaturationGraceMs=10000'); }); }); From e6106dd2830fc7133164511bea9ef53eaab61c9b Mon Sep 17 00:00:00 2001 From: yiliang114 <1204183885@qq.com> Date: Wed, 2 Sep 2026 12:13:49 +0900 Subject: [PATCH 6/6] fix(acp-bridge): keep saturation grace under the liveness probe timeout MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The grace window defaulted to 10s, exactly CHANNEL_LIVENESS_PROBE_TIMEOUT_MS. The liveness probe's response travels over this same NDJSON stream, so a pump parked on backpressure cannot answer it: at parity every saturation episode burns a probe, and two episodes inside two probe intervals tear the channel down as an acp_channel_liveness_timeout — the multi-session outage the grace window exists to prevent, reported as the wrong cause. Default to 5s and pin the invariant against channel-liveness.ts in ndJsonStream.test.ts. Also scope the saturation state (grace, waiter, episode dedup) to the createBoundedReadable closure instead of threading it through pumpBoundedInput and readBoundedChunk. Both drop from three added parameters to one, and DecodedQueueSpaceOptions / SaturationEpisodeState go away entirely. Drop the try/catch around the serve-side warn as well: callHook already isolates hook throws from the transport. --- packages/acp-bridge/src/ndJsonStream.test.ts | 19 ++ packages/acp-bridge/src/ndJsonStream.ts | 202 ++++++++----------- packages/cli/src/serve/run-qwen-serve.ts | 20 +- 3 files changed, 107 insertions(+), 134 deletions(-) diff --git a/packages/acp-bridge/src/ndJsonStream.test.ts b/packages/acp-bridge/src/ndJsonStream.test.ts index 2d62a858eb9..cca90f8aa30 100644 --- a/packages/acp-bridge/src/ndJsonStream.test.ts +++ b/packages/acp-bridge/src/ndJsonStream.test.ts @@ -11,11 +11,16 @@ import { type AnyMessage, } from '@agentclientprotocol/sdk'; import { + NDJSON_QUEUE_SATURATION_GRACE_MS, NdJsonIncompleteFrameError, NdJsonQueueLimitError, ndJsonStream, type NdJsonStreamLimits, } from './ndJsonStream.js'; +import { + CHANNEL_LIVENESS_INTERVAL_MS, + CHANNEL_LIVENESS_PROBE_TIMEOUT_MS, +} from './channel-liveness.js'; const encoder = new TextEncoder(); @@ -686,6 +691,20 @@ describe('ndJsonStream', () => { await stream.readable.cancel(); }); + it('keeps the default grace window under the liveness probe timeout', () => { + // The channel liveness probe's response travels over this same stream, so + // a pump parked on backpressure cannot answer it. At or above parity a + // saturation episode burns a probe, and two episodes inside two probe + // intervals tear the channel down as a liveness timeout — the outage the + // grace window exists to prevent, reported as the wrong cause. + expect(NDJSON_QUEUE_SATURATION_GRACE_MS).toBeLessThan( + CHANNEL_LIVENESS_PROBE_TIMEOUT_MS, + ); + expect(NDJSON_QUEUE_SATURATION_GRACE_MS).toBeLessThan( + CHANNEL_LIVENESS_INTERVAL_MS, + ); + }); + it('rejects a non-positive queueSaturationGraceMs', () => { expect(() => ndJsonStream( diff --git a/packages/acp-bridge/src/ndJsonStream.ts b/packages/acp-bridge/src/ndJsonStream.ts index 743696e30a5..08317271cc7 100644 --- a/packages/acp-bridge/src/ndJsonStream.ts +++ b/packages/acp-bridge/src/ndJsonStream.ts @@ -45,8 +45,15 @@ export interface NdJsonStreamHooks { * SSE) resolve within this window and keep the channel alive; a genuinely * stalled consumer still tears the transport down afterwards so the memory * bound remains effective. + * + * Must stay below `CHANNEL_LIVENESS_PROBE_TIMEOUT_MS`: the liveness probe's + * response travels over this same stream, so a parked pump cannot answer it. + * At parity the grace window burns a probe every episode, and two episodes + * inside `CHANNEL_LIVENESS_INTERVAL_MS * 2` would tear the channel down as a + * liveness timeout — the outage this backpressure exists to prevent, with a + * misleading cause. `ndJsonStream.test.ts` pins the relation. */ -export const NDJSON_QUEUE_SATURATION_GRACE_MS = 10_000; +export const NDJSON_QUEUE_SATURATION_GRACE_MS = 5_000; export interface NdJsonStreamLimits { maxFrameBytes: number; @@ -249,15 +256,17 @@ function createBoundedReadable( let nextQueueCharge = minimumQueueCharge; let reader: ReadableStreamDefaultReader | undefined; let canceled = false; - // Dedup state for onQueueSaturated: one warning per uninterrupted - // saturation episode, reset once the queue fully drains. - const saturationEpisode = { warned: false }; + // One onQueueSaturated warning per uninterrupted saturation episode; reset + // once the queue fully drains. + let saturationWarned = false; // Single waiter: only the pump loop ever waits for queue space. let wakeQueueWaiter: (() => void) | undefined; const waitForQueueSpace = (timeoutMs: number): Promise => new Promise((resolve) => { const timer = setTimeout( finish, + // Node clamps delays above 2^31-1 ms to 1 ms (and warns), which would + // spin the wait loop; cap so a long grace waits instead. Math.min(Math.max(0, timeoutMs), 2_147_483_647), ); function finish() { @@ -268,6 +277,61 @@ function createBoundedReadable( wakeQueueWaiter = finish; }); + /** + * Waits (bounded) for room in the decoded queue instead of failing the + * transport immediately (issue #10162). Saturating the queue means the + * consumer is slow; pausing the producer backpressures the agent's stdout + * pipe and keeps the memory bound intact. Only if the consumer stays slow + * for the whole grace window does the original fail-closed guard fire. + */ + const ensureQueueSpace = async ( + controller: ReadableStreamDefaultController, + queueCharge: number, + ): Promise => { + const queueLimitError = (available: number) => + new NdJsonQueueLimitError( + limits.maxQueuedMessages, + limits.maxQueuedBytes, + queueCharge, + Math.max(0, available), + ); + let availableBytes = controller.desiredSize; + if (availableBytes === null) throw queueLimitError(0); + // The queue fully drained: the previous saturation episode is over and a + // new one may warn again. + if (availableBytes === limits.maxQueuedBytes) saturationWarned = false; + if (queueCharge <= availableBytes) return; + // A frame that would not fit even in a fully drained queue can never be + // rescued by waiting. + if (queueCharge > limits.maxQueuedBytes) { + throw queueLimitError(availableBytes); + } + + if (!saturationWarned) { + saturationWarned = true; + callHook(hooks?.onQueueSaturated, { + requiredBytes: queueCharge, + availableBytes: Math.max(0, availableBytes), + maxQueuedMessages: limits.maxQueuedMessages, + maxQueuedBytes: limits.maxQueuedBytes, + graceMs, + }); + } + + // Monotonic: wall-clock steps must not extend or shorten the grace window. + const deadline = performance.now() + graceMs; + while (queueCharge > availableBytes) { + if (canceled) return; + const remainingMs = deadline - performance.now(); + if (remainingMs <= 0) throw queueLimitError(availableBytes); + await waitForQueueSpace(remainingMs); + if (canceled) return; + const nextAvailableBytes = controller.desiredSize; + if (nextAvailableBytes === null) throw queueLimitError(0); + availableBytes = nextAvailableBytes; + } + }; + return new ReadableStream( { start(controller) { @@ -284,9 +348,7 @@ function createBoundedReadable( validateInboundMessage, fatalCleanEof, minimumQueueCharge, - graceMs, - waitForQueueSpace, - saturationEpisode, + ensureQueueSpace, (charge) => { nextQueueCharge = charge; }, @@ -324,9 +386,10 @@ async function pumpBoundedInput( validateInboundMessage: NdJsonInboundMessageValidator | undefined, fatalCleanEof: boolean, minimumQueueCharge: number, - queueSaturationGraceMs: number, - waitForQueueSpace: (timeoutMs: number) => Promise, - saturationEpisode: SaturationEpisodeState, + ensureQueueSpace: ( + controller: ReadableStreamDefaultController, + queueCharge: number, + ) => Promise, setNextQueueCharge: (charge: number) => void, isCanceled: () => boolean, ): Promise { @@ -354,9 +417,7 @@ async function pumpBoundedInput( inboundRequests, validateInboundMessage, minimumQueueCharge, - queueSaturationGraceMs, - waitForQueueSpace, - saturationEpisode, + ensureQueueSpace, setNextQueueCharge, isCanceled, ); @@ -413,9 +474,10 @@ async function readBoundedChunk( inboundRequests: BoundedInboundRequestLedger, validateInboundMessage: NdJsonInboundMessageValidator | undefined, minimumQueueCharge: number, - queueSaturationGraceMs: number, - waitForQueueSpace: (timeoutMs: number) => Promise, - saturationEpisode: SaturationEpisodeState, + ensureQueueSpace: ( + controller: ReadableStreamDefaultController, + queueCharge: number, + ) => Promise, setNextQueueCharge: (charge: number) => void, isCanceled: () => boolean, ): Promise { @@ -432,16 +494,7 @@ async function readBoundedChunk( continue; } const queueCharge = Math.max(frameBytes, minimumQueueCharge); - await waitForDecodedQueueSpace({ - controller, - limits, - queueCharge, - queueSaturationGraceMs, - waitForQueueSpace, - saturationEpisode, - hooks, - isCanceled, - }); + await ensureQueueSpace(controller, queueCharge); if (isCanceled()) return; setNextQueueCharge(queueCharge); handleBoundedLine( @@ -459,103 +512,6 @@ async function readBoundedChunk( if (start < chunk.length) pending.append(chunk.subarray(start)); } -interface SaturationEpisodeState { - warned: boolean; -} - -interface DecodedQueueSpaceOptions { - controller: ReadableStreamDefaultController; - limits: NdJsonStreamLimits; - queueCharge: number; - queueSaturationGraceMs: number; - waitForQueueSpace: (timeoutMs: number) => Promise; - saturationEpisode: SaturationEpisodeState; - hooks: NdJsonStreamHooks | undefined; - isCanceled: () => boolean; -} - -/** - * Waits (bounded) for room in the decoded queue instead of failing the - * transport immediately (issue #10162). Saturating the queue means the - * consumer is slow; pausing the producer backpressures the agent's stdout - * pipe and keeps the memory bound intact. Only if the consumer stays slow - * for the whole grace window does the original fail-closed guard fire. - */ -async function waitForDecodedQueueSpace({ - controller, - limits, - queueCharge, - queueSaturationGraceMs, - waitForQueueSpace, - saturationEpisode, - hooks, - isCanceled, -}: DecodedQueueSpaceOptions): Promise { - let availableBytes = controller.desiredSize; - if (availableBytes === null) { - throw new NdJsonQueueLimitError( - limits.maxQueuedMessages, - limits.maxQueuedBytes, - queueCharge, - 0, - ); - } - // The queue fully drained: the previous saturation episode is over and a - // new one may warn again. - if (availableBytes === limits.maxQueuedBytes) { - saturationEpisode.warned = false; - } - if (queueCharge <= availableBytes) return; - // A frame that would not fit even in a fully drained queue can never be - // rescued by waiting. - if (queueCharge > limits.maxQueuedBytes) { - throw new NdJsonQueueLimitError( - limits.maxQueuedMessages, - limits.maxQueuedBytes, - queueCharge, - Math.max(0, availableBytes), - ); - } - - if (!saturationEpisode.warned) { - saturationEpisode.warned = true; - callHook(hooks?.onQueueSaturated, { - requiredBytes: queueCharge, - availableBytes: Math.max(0, availableBytes), - maxQueuedMessages: limits.maxQueuedMessages, - maxQueuedBytes: limits.maxQueuedBytes, - graceMs: queueSaturationGraceMs, - }); - } - - // Monotonic: wall-clock steps must not extend or shorten the grace window. - const deadline = performance.now() + queueSaturationGraceMs; - while (queueCharge > availableBytes) { - if (isCanceled()) return; - const remainingMs = deadline - performance.now(); - if (remainingMs <= 0) { - throw new NdJsonQueueLimitError( - limits.maxQueuedMessages, - limits.maxQueuedBytes, - queueCharge, - Math.max(0, availableBytes), - ); - } - await waitForQueueSpace(remainingMs); - if (isCanceled()) return; - const nextAvailableBytes = controller.desiredSize; - if (nextAvailableBytes === null) { - throw new NdJsonQueueLimitError( - limits.maxQueuedMessages, - limits.maxQueuedBytes, - queueCharge, - 0, - ); - } - availableBytes = nextAvailableBytes; - } -} - function takeLegacyLineBytes( pending: Uint8Array[], current: Uint8Array, diff --git a/packages/cli/src/serve/run-qwen-serve.ts b/packages/cli/src/serve/run-qwen-serve.ts index b9b5fa19416..180a76a2c48 100644 --- a/packages/cli/src/serve/run-qwen-serve.ts +++ b/packages/cli/src/serve/run-qwen-serve.ts @@ -4933,18 +4933,16 @@ async function runQwenServeImpl( // window instead of tearing the channel down immediately (#10162). // Warn at episode start so field diagnosis doesn't begin at the // `channel exited` breadcrumb. + // `callHook` in ndJsonStream already isolates hook throws from the + // transport, so this needs no guard of its own. const warnAcpQueueSaturated = (info: NdJsonQueueSaturationInfo): void => { - try { - daemonLog.warn('ACP NDJSON decoded queue saturated', { - requiredBytes: info.requiredBytes, - availableBytes: info.availableBytes, - maxQueuedMessages: info.maxQueuedMessages, - maxQueuedBytes: info.maxQueuedBytes, - queueSaturationGraceMs: info.graceMs, - }); - } catch { - // Observability must not affect transport behavior. - } + daemonLog.warn('ACP NDJSON decoded queue saturated', { + requiredBytes: info.requiredBytes, + availableBytes: info.availableBytes, + maxQueuedMessages: info.maxQueuedMessages, + maxQueuedBytes: info.maxQueuedBytes, + queueSaturationGraceMs: info.graceMs, + }); }; const recordPromptQueueWait = (durationMs: number): void => { promptQueueWaitStats.count += 1;