diff --git a/packages/acp-bridge/src/ndJsonStream.test.ts b/packages/acp-bridge/src/ndJsonStream.test.ts index b3e5bcbe950..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(); @@ -396,6 +401,7 @@ describe('ndJsonStream', () => { maxFrameBytes: 200, maxQueuedMessages: 2, maxQueuedBytes: 200, + queueSaturationGraceMs: 25, }), ); await vi.waitFor(() => @@ -422,6 +428,7 @@ describe('ndJsonStream', () => { maxFrameBytes: 200, maxQueuedMessages: 100, maxQueuedBytes: firstBytes + 1, + queueSaturationGraceMs: 25, }), ); await vi.waitFor(() => @@ -437,6 +444,278 @@ 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(); + 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('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, + }); + } + + 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).toHaveBeenCalledTimes(2); + 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({ + 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(); + 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(); + 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, + queueSaturationGraceMs: 5_000, + }), + ); + + await vi.waitFor(() => + expect(onTransportError).toHaveBeenCalledWith( + expect.any(NdJsonQueueLimitError), + ), + ); + expect(onQueueSaturated).not.toHaveBeenCalled(); + 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( + 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; @@ -481,6 +760,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..08317271cc7 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'; @@ -14,17 +15,51 @@ 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. 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; } +/** + * 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. + * + * 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 = 5_000; + export interface NdJsonStreamLimits { maxFrameBytes: number; maxQueuedMessages: number; maxQueuedBytes: number; + queueSaturationGraceMs?: number; } export type NdJsonInboundMessageValidator = (message: AnyMessage) => boolean; @@ -216,9 +251,86 @@ 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; + // 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() { + clearTimeout(timer); + wakeQueueWaiter = undefined; + resolve(); + } + 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( { @@ -236,15 +348,22 @@ function createBoundedReadable( validateInboundMessage, fatalCleanEof, minimumQueueCharge, + ensureQueueSpace, (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 +386,10 @@ async function pumpBoundedInput( validateInboundMessage: NdJsonInboundMessageValidator | undefined, fatalCleanEof: boolean, minimumQueueCharge: number, + ensureQueueSpace: ( + controller: ReadableStreamDefaultController, + queueCharge: number, + ) => Promise, setNextQueueCharge: (charge: number) => void, isCanceled: () => boolean, ): Promise { @@ -283,7 +406,7 @@ async function pumpBoundedInput( return; } if (!result.value) continue; - readBoundedChunk( + await readBoundedChunk( result.value, pending, controller, @@ -294,8 +417,11 @@ async function pumpBoundedInput( inboundRequests, validateInboundMessage, minimumQueueCharge, + ensureQueueSpace, setNextQueueCharge, + isCanceled, ); + if (isCanceled()) return; } } catch (error) { if (isCanceled()) return; @@ -337,7 +463,7 @@ function readLegacyChunk( } } -function readBoundedChunk( +async function readBoundedChunk( chunk: Uint8Array, pending: BoundedFrameBuffer, controller: ReadableStreamDefaultController, @@ -348,8 +474,13 @@ function readBoundedChunk( inboundRequests: BoundedInboundRequestLedger, validateInboundMessage: NdJsonInboundMessageValidator | undefined, minimumQueueCharge: number, + ensureQueueSpace: ( + controller: ReadableStreamDefaultController, + queueCharge: number, + ) => Promise, setNextQueueCharge: (charge: number) => void, -): void { + isCanceled: () => boolean, +): Promise { let start = 0; let newline = chunk.indexOf(0x0a, start); while (newline !== -1) { @@ -363,15 +494,8 @@ 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 ensureQueueSpace(controller, queueCharge); + if (isCanceled()) return; setNextQueueCharge(queueCharge); handleBoundedLine( pending.take(current), @@ -721,6 +845,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.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'); }); }); diff --git a/packages/cli/src/serve/run-qwen-serve.ts b/packages/cli/src/serve/run-qwen-serve.ts index 0019a52a6c2..180a76a2c48 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, @@ -4926,6 +4929,21 @@ 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. + // `callHook` in ndJsonStream already isolates hook throws from the + // transport, so this needs no guard of its own. + const warnAcpQueueSaturated = (info: NdJsonQueueSaturationInfo): void => { + 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; promptQueueWaitStats.totalMs += durationMs; @@ -5097,6 +5115,7 @@ async function runQwenServeImpl( bytes, message, }), + onQueueSaturated: warnAcpQueueSaturated, }, ...(acpChildExtraArgs(opts) ? { extraArgs: acpChildExtraArgs(opts) } @@ -6003,6 +6022,7 @@ async function runQwenServeImpl( bytes, message, }), + onQueueSaturated: warnAcpQueueSaturated, }, ...(acpChildExtraArgs(opts) ? { extraArgs: acpChildExtraArgs(opts) } @@ -6661,6 +6681,7 @@ async function runQwenServeImpl( bytes, message, }), + onQueueSaturated: warnAcpQueueSaturated, }, ...(acpChildExtraArgs(opts) ? { extraArgs: acpChildExtraArgs(opts) }