diff --git a/docs/design/2026-07-31-daemon-capacity-model-and-memory-bounds.md b/docs/design/2026-07-31-daemon-capacity-model-and-memory-bounds.md index 7af90f826d..a01c55ee48 100644 --- a/docs/design/2026-07-31-daemon-capacity-model-and-memory-bounds.md +++ b/docs/design/2026-07-31-daemon-capacity-model-and-memory-bounds.md @@ -137,7 +137,13 @@ This is deliberately promoted ahead of the byte-cap work. It is the only piece w Ordered by measured risk, each independently shippable. -**The NDJSON frame reader has no bound of any kind.** `packages/acp-bridge/src/ndJsonStream.ts:35` declares `pending: Uint8Array[]`, pushes unterminated tail bytes at `:92`, and never checks a count or a byte total. `takeLineBytes` (`:96-111`) then allocates one contiguous copy of the accumulated total, `TextDecoder.decode` produces a UTF-16 string at roughly twice that, and `JSON.parse` builds objects again — about fivefold amplification over a frame that has no upper bound. This is the read side of every spawned ACP child's stdout, and `packages/cli/src/serve/large-pipe-frame-observer.ts:10` only logs frames above 256 KiB. The fix is a frame byte cap checked on every chunk, a typed fatal error on daemon-managed streams, and a queuing strategy on the decoded-message `ReadableStream` at `:33`, which never consults `desiredSize` and is a second unbounded buffer behind a slow consumer. `createStderrForwarder` (`spawnChannel.ts:58-72`, 64 KiB with a `[truncated]` marker) and the channel worker's log buffer (`channel-worker-supervisor.ts:67-69`) are the in-repo templates. +**The NDJSON frame reader is the first bounded-container increment.** Before this change, `packages/acp-bridge/src/ndJsonStream.ts` retained every unterminated tail chunk without a count or byte check, then allocated one contiguous copy, a UTF-16 string, and a parsed object — about fivefold amplification over a frame with no upper bound. Its decoded-message `ReadableStream` also ignored `desiredSize`, creating a second unbounded buffer behind a slow consumer. This is the read side of every spawned ACP child's stdout, while `packages/cli/src/serve/large-pipe-frame-observer.ts` only observes frames after parse and enqueue. `createStderrForwarder` (64 KiB with a `[truncated]` marker) and the channel worker's log buffer are the in-repo templates for bounding at the container. + +The first Part 3 increment applies that protection only to ACP streams created by `qwen serve`. A complete inbound or outbound frame is limited to 64 MiB including its newline. The decoded inbound queue is limited to 256 messages and 64 MiB of retained wire bytes. `ReadableStream` exposes one scalar queuing cost rather than independent count and byte watermarks, so each message is charged `max(frameBytes, ceil(64 MiB / 256))`. This is deliberately conservative: it proves both upper bounds, although a mixed queue can be rejected before either independent limit is exactly full. Admission is checked against `desiredSize` before decode and parse, so the message that would exceed the queue is never materialized. + +Crossing either bound is transport-fatal. The reader cancels the child stdout, reports a typed cause through the transport lifecycle hook, and closes its decoded readable; it does not error that readable because the ACP SDK's internal receive loop does not catch `reader.read()` rejection. The spawn channel terminates that exact tracked child, and the bridge's existing channel-exit path tears down only the sessions multiplexed on that workspace generation. An unterminated final frame is also fatal on this daemon-owned path. Parse-error logs on the bounded path contain only an error code, byte length, and SHA-256 digest; they never echo the frame or the parser error, whose message may itself contain input. The public `ndJsonStream` default, in-memory channels, direct embeds, interactive CLI, and IDE companion do not opt in, so they retain the existing eager queue, parse logging, and unterminated-EOF behavior. + +The outbound check happens after `JSON.stringify` and UTF-8 encoding. It prevents an oversized frame from entering the child pipe, but it is not a pre-allocation encoder budget; bounded/canonical JSON encoding remains a separate container change rather than being hidden inside this transport PR. **The EventBus replay ring bounds by frame count only.** `packages/acp-bridge/src/eventBus.ts:473` evicts on `ring.length > ringSize`, default 8000 frames, per session, tunable to a million. This is conspicuous because everything around the ring is already byte-bounded: per-subscriber queues at 2 MiB, replay burst at 8 MiB, journal at 8 MiB, compacted replay at 4 MiB. The ring is the gap, and it multiplies the unbounded frames above by 8000. The serialized size is **already computed and in scope** at `:459`, where it is handed to the compaction engine; applying it to the ring is a running total, an eviction loop over both bounds, and the retain-at-least-one guarantee the compaction engine already implements. diff --git a/packages/acp-bridge/src/ndJsonStream.test.ts b/packages/acp-bridge/src/ndJsonStream.test.ts index c70128760c..593dd3a08c 100644 --- a/packages/acp-bridge/src/ndJsonStream.test.ts +++ b/packages/acp-bridge/src/ndJsonStream.test.ts @@ -5,8 +5,16 @@ */ import { describe, expect, it, vi } from 'vitest'; -import type { AnyMessage } from '@agentclientprotocol/sdk'; -import { ndJsonStream } from './ndJsonStream.js'; +import { + ClientSideConnection, + type AnyMessage, +} from '@agentclientprotocol/sdk'; +import { + NdJsonIncompleteFrameError, + NdJsonQueueLimitError, + ndJsonStream, + type NdJsonStreamLimits, +} from './ndJsonStream.js'; const encoder = new TextEncoder(); @@ -25,6 +33,17 @@ function byteStream(chunks: readonly Uint8Array[]): ReadableStream { }); } +function limits( + overrides: Partial = {}, +): NdJsonStreamLimits { + return { + maxFrameBytes: 1024, + maxQueuedMessages: 4, + maxQueuedBytes: 4096, + ...overrides, + }; +} + async function readAll(readable: ReadableStream) { const reader = readable.getReader(); const out: AnyMessage[] = []; @@ -250,4 +269,262 @@ describe('ndJsonStream', () => { expect(onMessageSent).not.toHaveBeenCalled(); expect(onMessageObserved).not.toHaveBeenCalled(); }); + + it('accepts an inbound frame exactly at the configured byte limit', async () => { + const sent = message('exact-limit'); + const frame = encoder.encode(`${JSON.stringify(sent)}\n`); + const stream = ndJsonStream( + new WritableStream(), + byteStream([frame.slice(0, 3), frame.slice(3)]), + undefined, + limits({ + maxFrameBytes: frame.byteLength, + maxQueuedMessages: 2, + maxQueuedBytes: frame.byteLength * 2, + }), + ); + + await expect(readAll(stream.readable)).resolves.toEqual([sent]); + }); + + it('counts CRLF and resets the bounded accumulator between frames', async () => { + const first = message('first-crlf'); + const second = message('second-crlf'); + const firstFrame = encoder.encode(`${JSON.stringify(first)}\r\n`); + const secondFrame = encoder.encode(`${JSON.stringify(second)}\r\n`); + const maxFrameBytes = Math.max( + firstFrame.byteLength, + secondFrame.byteLength, + ); + const stream = ndJsonStream( + new WritableStream(), + byteStream([ + firstFrame.slice(0, firstFrame.byteLength - 1), + new Uint8Array([ + firstFrame[firstFrame.byteLength - 1]!, + ...secondFrame, + ]), + ]), + undefined, + limits({ + maxFrameBytes, + maxQueuedMessages: 2, + maxQueuedBytes: maxFrameBytes * 2, + }), + ); + + await expect(readAll(stream.readable)).resolves.toEqual([first, second]); + + const onTransportError = vi.fn(); + const rejected = ndJsonStream( + new WritableStream(), + byteStream([firstFrame]), + { onTransportError }, + limits({ maxFrameBytes: firstFrame.byteLength - 1 }), + ); + await expect(readAll(rejected.readable)).resolves.toEqual([]); + expect(onTransportError).toHaveBeenCalledWith( + expect.objectContaining({ + code: 'ndjson_frame_too_large', + observedBytes: firstFrame.byteLength, + }), + ); + expect(onTransportError).toHaveBeenCalledOnce(); + }); + + it('rejects an oversized inbound frame before parsing or reporting it', async () => { + const sent = message('over-limit', { secret: 'do-not-log' }); + const frame = encoder.encode(`${JSON.stringify(sent)}\n`); + const onMessageReceived = vi.fn(); + const onTransportError = vi.fn(); + const stderr = vi.spyOn(console, 'error').mockImplementation(() => {}); + const stream = ndJsonStream( + new WritableStream(), + byteStream([frame.slice(0, 5), frame.slice(5)]), + { onMessageReceived, onTransportError }, + limits({ maxFrameBytes: frame.byteLength - 1 }), + ); + + await expect(readAll(stream.readable)).resolves.toEqual([]); + expect(onMessageReceived).not.toHaveBeenCalled(); + expect(onTransportError).toHaveBeenCalledWith( + expect.objectContaining({ + code: 'ndjson_frame_too_large', + direction: 'received', + limitBytes: frame.byteLength - 1, + observedBytes: frame.byteLength, + }), + ); + expect(onTransportError).toHaveBeenCalledOnce(); + expect(stderr).not.toHaveBeenCalled(); + stderr.mockRestore(); + }); + + it('rejects an incomplete final frame only on the bounded path', async () => { + const partial = encoder.encode(JSON.stringify(message('partial'))); + const onTransportError = vi.fn(); + const stream = ndJsonStream( + new WritableStream(), + byteStream([partial]), + { onTransportError }, + limits(), + ); + + await expect(readAll(stream.readable)).resolves.toEqual([]); + expect(onTransportError).toHaveBeenCalledWith( + expect.objectContaining({ + name: NdJsonIncompleteFrameError.name, + code: 'ndjson_incomplete_frame', + observedBytes: partial.byteLength, + }), + ); + expect(onTransportError).toHaveBeenCalledOnce(); + }); + + it('bounds the decoded queue by message count for a stalled consumer', async () => { + const frames = ['one', 'two', 'three'] + .map((method) => `${JSON.stringify(message(method))}\n`) + .join(''); + const onMessageReceived = vi.fn(); + const onTransportError = vi.fn(); + const stream = ndJsonStream( + new WritableStream(), + byteStream([encoder.encode(frames)]), + { onMessageReceived, onTransportError }, + limits({ + maxFrameBytes: 200, + maxQueuedMessages: 2, + maxQueuedBytes: 200, + }), + ); + await vi.waitFor(() => + expect(onTransportError).toHaveBeenCalledWith( + expect.any(NdJsonQueueLimitError), + ), + ); + expect(onTransportError).toHaveBeenCalledOnce(); + expect(onMessageReceived).toHaveBeenCalledTimes(2); + await stream.readable.cancel(); + }); + + it('bounds the decoded queue by retained wire bytes', async () => { + const first = `${JSON.stringify(message('first', { text: 'x'.repeat(40) }))}\n`; + const second = `${JSON.stringify(message('second', { text: 'y'.repeat(40) }))}\n`; + const firstBytes = encoder.encode(first).byteLength; + const onMessageReceived = vi.fn(); + const onTransportError = vi.fn(); + const stream = ndJsonStream( + new WritableStream(), + byteStream([encoder.encode(first + second)]), + { onMessageReceived, onTransportError }, + limits({ + maxFrameBytes: 200, + maxQueuedMessages: 100, + maxQueuedBytes: firstBytes + 1, + }), + ); + await vi.waitFor(() => + expect(onTransportError).toHaveBeenCalledWith( + expect.objectContaining({ + code: 'ndjson_queue_limit_exceeded', + maxQueuedBytes: firstBytes + 1, + }), + ), + ); + expect(onTransportError).toHaveBeenCalledOnce(); + expect(onMessageReceived).toHaveBeenCalledOnce(); + await stream.readable.cancel(); + }); + + it('keeps bounded parse-error logs free of input and parser text', async () => { + const payload = '{"secret":"do-not-echo"'; + const stderr = vi.spyOn(console, 'error').mockImplementation(() => {}); + const stream = ndJsonStream( + new WritableStream(), + byteStream([encoder.encode(`${payload}\n`)]), + undefined, + limits(), + ); + + await expect(readAll(stream.readable)).resolves.toEqual([]); + expect(stderr).toHaveBeenCalledWith('Failed to parse JSON message:', { + errorKind: 'ndjson_parse_error', + bytes: encoder.encode(payload).byteLength, + sha256: expect.stringMatching(/^[a-f0-9]{64}$/u), + payloadOmitted: true, + }); + expect(JSON.stringify(stderr.mock.calls)).not.toContain('do-not-echo'); + stderr.mockRestore(); + }); + + it('checks outbound frame bytes including the newline', async () => { + const sent = message('outbound-exact'); + const payloadBytes = encoder.encode(JSON.stringify(sent)).byteLength; + const outputChunks: Uint8Array[] = []; + const exact = ndJsonStream( + new WritableStream({ + write(chunk) { + outputChunks.push(chunk); + }, + }), + byteStream([]), + undefined, + limits({ maxFrameBytes: payloadBytes + 1 }), + ); + await expect(writeOne(exact.writable, sent)).resolves.toBeUndefined(); + expect(outputChunks[0]?.byteLength).toBe(payloadBytes + 1); + + const onTransportError = vi.fn(); + const rejected = ndJsonStream( + new WritableStream(), + byteStream([]), + { onTransportError }, + limits({ maxFrameBytes: payloadBytes }), + ); + await expect(writeOne(rejected.writable, sent)).rejects.toMatchObject({ + code: 'ndjson_frame_too_large', + direction: 'sent', + observedBytes: payloadBytes + 1, + }); + expect(onTransportError).toHaveBeenCalledOnce(); + }); + + it('cancels and unlocks bounded input during frame assembly', async () => { + const cancel = vi.fn(); + const input = new ReadableStream({ + start(controller) { + controller.enqueue(encoder.encode('{"partial":')); + }, + cancel, + }); + const stream = ndJsonStream( + new WritableStream(), + input, + undefined, + limits(), + ); + + await stream.readable.cancel('test cancellation'); + + expect(cancel).toHaveBeenCalledWith('test cancellation'); + await vi.waitFor(() => expect(input.locked).toBe(false)); + }); + + it('closes the ACP SDK connection without rejecting on inbound fatal', async () => { + const onTransportError = vi.fn(); + const stream = ndJsonStream( + new WritableStream(), + byteStream([encoder.encode('x'.repeat(17))]), + { onTransportError }, + limits({ maxFrameBytes: 16 }), + ); + const connection = new ClientSideConnection(() => ({}) as never, stream); + + await expect(connection.closed).resolves.toBeUndefined(); + expect(connection.signal.aborted).toBe(true); + expect(onTransportError).toHaveBeenCalledOnce(); + expect(onTransportError).toHaveBeenCalledWith( + expect.objectContaining({ code: 'ndjson_frame_too_large' }), + ); + }); }); diff --git a/packages/acp-bridge/src/ndJsonStream.ts b/packages/acp-bridge/src/ndJsonStream.ts index 2822431c47..eb5391c4fa 100644 --- a/packages/acp-bridge/src/ndJsonStream.ts +++ b/packages/acp-bridge/src/ndJsonStream.ts @@ -4,6 +4,7 @@ * SPDX-License-Identifier: Apache-2.0 */ +import { createHash } from 'node:crypto'; import type { AnyMessage, Stream } from '@agentclientprotocol/sdk'; export interface NdJsonMessageObservation { @@ -16,6 +17,55 @@ export interface NdJsonStreamHooks { onMessageReceived?: (bytes: number) => void; onMessageSent?: (bytes: number) => void; onMessageObserved?: (observation: NdJsonMessageObservation) => void; + onTransportError?: (error: unknown) => void; +} + +export interface NdJsonStreamLimits { + maxFrameBytes: number; + maxQueuedMessages: number; + maxQueuedBytes: number; +} + +export class NdJsonFrameTooLargeError extends Error { + readonly code = 'ndjson_frame_too_large'; + + constructor( + readonly direction: 'sent' | 'received', + readonly limitBytes: number, + readonly observedBytes: number, + ) { + super( + `NDJSON ${direction} frame exceeds ${limitBytes} bytes ` + + `(observed ${observedBytes} bytes)`, + ); + this.name = 'NdJsonFrameTooLargeError'; + } +} + +export class NdJsonQueueLimitError extends Error { + readonly code = 'ndjson_queue_limit_exceeded'; + + constructor( + readonly maxQueuedMessages: number, + readonly maxQueuedBytes: number, + readonly requiredBytes: number, + readonly availableBytes: number, + ) { + super( + `NDJSON decoded queue is full ` + + `(required ${requiredBytes} bytes, available ${availableBytes} bytes)`, + ); + this.name = 'NdJsonQueueLimitError'; + } +} + +export class NdJsonIncompleteFrameError extends Error { + readonly code = 'ndjson_incomplete_frame'; + + constructor(readonly observedBytes: number) { + super(`NDJSON input ended with an incomplete ${observedBytes}-byte frame`); + this.name = 'NdJsonIncompleteFrameError'; + } } interface TextDecoderLike { @@ -26,33 +76,31 @@ export function ndJsonStream( output: WritableStream, input: ReadableStream, hooks?: NdJsonStreamHooks, + limits?: NdJsonStreamLimits, ): Stream { const textEncoder = new TextEncoder(); const textDecoder = new TextDecoder(); + if (limits) validateNdJsonStreamLimits(limits); - const readable = new ReadableStream({ - async start(controller) { - const pending: Uint8Array[] = []; - const reader = input.getReader(); - try { - while (true) { - const { value, done } = await reader.read(); - if (done) break; - if (!value) continue; - readChunk(value, pending, controller, textDecoder, hooks); - } - } finally { - reader.releaseLock(); - controller.close(); - } - }, - }); + const readable = limits + ? createBoundedReadable(input, textDecoder, hooks, limits) + : createLegacyReadable(input, textDecoder, hooks); const writable = new WritableStream({ async write(message) { const content = JSON.stringify(message); const payload = textEncoder.encode(content); - const frame = new Uint8Array(payload.byteLength + 1); + const frameBytes = payload.byteLength + 1; + if (limits && frameBytes > limits.maxFrameBytes) { + const error = new NdJsonFrameTooLargeError( + 'sent', + limits.maxFrameBytes, + frameBytes, + ); + callHook(hooks?.onTransportError, error); + throw error; + } + const frame = new Uint8Array(frameBytes); frame.set(payload); frame[payload.byteLength] = 0x0a; const writer = output.getWriter(); @@ -64,6 +112,9 @@ export function ndJsonStream( bytes: payload.byteLength, message, }); + } catch (error) { + if (limits) callHook(hooks?.onTransportError, error); + throw error; } finally { writer.releaseLock(); } @@ -73,7 +124,125 @@ export function ndJsonStream( return { readable, writable }; } -function readChunk( +function createLegacyReadable( + input: ReadableStream, + textDecoder: TextDecoderLike, + hooks?: NdJsonStreamHooks, +): ReadableStream { + return new ReadableStream({ + async start(controller) { + const pending: Uint8Array[] = []; + const reader = input.getReader(); + try { + while (true) { + const { value, done } = await reader.read(); + if (done) break; + if (!value) continue; + readLegacyChunk(value, pending, controller, textDecoder, hooks); + } + } finally { + reader.releaseLock(); + controller.close(); + } + }, + }); +} + +function createBoundedReadable( + input: ReadableStream, + textDecoder: TextDecoderLike, + hooks: NdJsonStreamHooks | undefined, + limits: NdJsonStreamLimits, +): ReadableStream { + const pending = new BoundedFrameBuffer(limits.maxFrameBytes); + const minimumQueueCharge = Math.ceil( + limits.maxQueuedBytes / limits.maxQueuedMessages, + ); + let nextQueueCharge = minimumQueueCharge; + let reader: ReadableStreamDefaultReader | undefined; + let canceled = false; + + return new ReadableStream( + { + start(controller) { + reader = input.getReader(); + void pumpBoundedInput( + reader, + pending, + controller, + textDecoder, + hooks, + limits, + minimumQueueCharge, + (charge) => { + nextQueueCharge = charge; + }, + () => canceled, + ); + }, + async cancel(reason) { + canceled = true; + pending.clear(); + if (reader) await cancelReader(reader, reason); + }, + }, + { + highWaterMark: limits.maxQueuedBytes, + size: () => nextQueueCharge, + }, + ); +} + +async function pumpBoundedInput( + reader: ReadableStreamDefaultReader, + pending: BoundedFrameBuffer, + controller: ReadableStreamDefaultController, + textDecoder: TextDecoderLike, + hooks: NdJsonStreamHooks | undefined, + limits: NdJsonStreamLimits, + minimumQueueCharge: number, + setNextQueueCharge: (charge: number) => void, + isCanceled: () => boolean, +): Promise { + try { + while (true) { + const result = await reader.read(); + if (result.done) { + if (isCanceled()) return; + if (pending.byteLength > 0) { + throw new NdJsonIncompleteFrameError(pending.byteLength); + } + controller.close(); + return; + } + if (!result.value) continue; + readBoundedChunk( + result.value, + pending, + controller, + textDecoder, + hooks, + limits, + minimumQueueCharge, + setNextQueueCharge, + ); + } + } catch (error) { + if (isCanceled()) return; + pending.clear(); + callHook(hooks?.onTransportError, error); + await cancelReader(reader, error); + // ACP SDK's receive loop closes in `finally` but does not catch a rejected + // `reader.read()`. Report the typed cause through the lifecycle hook and + // close here so a transport guard cannot become an unhandled rejection. + if (!isCanceled()) controller.close(); + } finally { + pending.clear(); + reader.releaseLock(); + } +} + +function readLegacyChunk( chunk: Uint8Array, pending: Uint8Array[], controller: ReadableStreamDefaultController, @@ -83,8 +252,11 @@ function readChunk( let start = 0; let newline = chunk.indexOf(0x0a, start); while (newline !== -1) { - const lineBytes = takeLineBytes(pending, chunk.subarray(start, newline)); - handleLine(lineBytes, controller, textDecoder, hooks); + const lineBytes = takeLegacyLineBytes( + pending, + chunk.subarray(start, newline), + ); + handleLegacyLine(lineBytes, controller, textDecoder, hooks); start = newline + 1; newline = chunk.indexOf(0x0a, start); } @@ -93,7 +265,50 @@ function readChunk( } } -function takeLineBytes(pending: Uint8Array[], current: Uint8Array): Uint8Array { +function readBoundedChunk( + chunk: Uint8Array, + pending: BoundedFrameBuffer, + controller: ReadableStreamDefaultController, + textDecoder: TextDecoderLike, + hooks: NdJsonStreamHooks | undefined, + limits: NdJsonStreamLimits, + minimumQueueCharge: number, + setNextQueueCharge: (charge: number) => void, +): void { + let start = 0; + let newline = chunk.indexOf(0x0a, start); + while (newline !== -1) { + const current = chunk.subarray(start, newline); + const frameBytes = pending.byteLength + current.byteLength + 1; + assertFrameSize('received', limits.maxFrameBytes, frameBytes); + if (pending.isJsonWhitespaceLine(current)) { + pending.clear(); + start = newline + 1; + newline = chunk.indexOf(0x0a, start); + 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), + ); + } + setNextQueueCharge(queueCharge); + handleBoundedLine(pending.take(current), controller, textDecoder, hooks); + start = newline + 1; + newline = chunk.indexOf(0x0a, start); + } + if (start < chunk.length) pending.append(chunk.subarray(start)); +} + +function takeLegacyLineBytes( + pending: Uint8Array[], + current: Uint8Array, +): Uint8Array { if (pending.length === 0) return current; const totalLength = @@ -110,7 +325,7 @@ function takeLineBytes(pending: Uint8Array[], current: Uint8Array): Uint8Array { return line; } -function handleLine( +function handleLegacyLine( lineBytes: Uint8Array, controller: ReadableStreamDefaultController, textDecoder: TextDecoderLike, @@ -123,29 +338,163 @@ function handleLine( try { const message = JSON.parse(trimmedLine) as AnyMessage; controller.enqueue(message); - const bytes = jsonPayloadByteLength(lineBytes); - callHook(hooks?.onMessageReceived, bytes); - callHook(hooks?.onMessageObserved, { - direction: 'received', - bytes, - message, - }); + reportReceivedMessage(lineBytes, message, hooks); } catch (err) { // eslint-disable-next-line no-console -- match ACP SDK parse-error behavior console.error('Failed to parse JSON message:', trimmedLine, err); } } +function handleBoundedLine( + lineBytes: Uint8Array, + controller: ReadableStreamDefaultController, + textDecoder: TextDecoderLike, + hooks?: NdJsonStreamHooks, +): void { + const line = textDecoder.decode(lineBytes); + const trimmedLine = line.trim(); + if (!trimmedLine) return; + + let message: AnyMessage; + try { + message = JSON.parse(trimmedLine) as AnyMessage; + } catch { + const bytes = jsonPayloadByteLength(lineBytes); + const digest = createHash('sha256') + .update(lineBytes.subarray(0, bytes)) + .digest('hex'); + // eslint-disable-next-line no-console -- bounded metadata only + console.error('Failed to parse JSON message:', { + errorKind: 'ndjson_parse_error', + bytes, + sha256: digest, + payloadOmitted: true, + }); + return; + } + + controller.enqueue(message); + reportReceivedMessage(lineBytes, message, hooks); +} + +function reportReceivedMessage( + lineBytes: Uint8Array, + message: AnyMessage, + hooks?: NdJsonStreamHooks, +): void { + const bytes = jsonPayloadByteLength(lineBytes); + callHook(hooks?.onMessageReceived, bytes); + callHook(hooks?.onMessageObserved, { + direction: 'received', + bytes, + message, + }); +} + function jsonPayloadByteLength(lineBytes: Uint8Array): number { return lineBytes[lineBytes.byteLength - 1] === 0x0d ? lineBytes.byteLength - 1 : lineBytes.byteLength; } +export function validateNdJsonStreamLimits(limits: NdJsonStreamLimits): void { + const values = [ + ['maxFrameBytes', limits.maxFrameBytes], + ['maxQueuedMessages', limits.maxQueuedMessages], + ['maxQueuedBytes', limits.maxQueuedBytes], + ] as const; + for (const [name, value] of values) { + if (!Number.isSafeInteger(value) || value <= 0) { + throw new RangeError(`${name} must be a positive safe integer`); + } + } +} + +function assertFrameSize( + direction: 'sent' | 'received', + limitBytes: number, + observedBytes: number, +): void { + if (observedBytes > limitBytes) { + throw new NdJsonFrameTooLargeError(direction, limitBytes, observedBytes); + } +} + +async function cancelReader( + reader: ReadableStreamDefaultReader, + reason: unknown, +): Promise { + try { + await reader.cancel(reason); + } catch { + /* preserve the transport error that caused cancellation */ + } +} + function callHook(hook: ((value: T) => void) | undefined, value: T): void { try { hook?.(value); } catch { - /* metrics hooks must not break the transport */ + /* metrics and lifecycle hooks must not break the transport */ } } + +class BoundedFrameBuffer { + private buffer: Uint8Array | undefined; + private length = 0; + + constructor(private readonly maxFrameBytes: number) {} + + get byteLength(): number { + return this.length; + } + + append(bytes: Uint8Array): void { + const requiredBytes = this.length + bytes.byteLength; + assertFrameSize('received', this.maxFrameBytes, requiredBytes); + if (requiredBytes === 0) return; + + if (!this.buffer || this.buffer.byteLength < requiredBytes) { + const doubledCapacity = Math.min( + this.maxFrameBytes, + Math.max(1024, (this.buffer?.byteLength ?? 0) * 2), + ); + const next = new Uint8Array(Math.max(requiredBytes, doubledCapacity)); + if (this.buffer) next.set(this.buffer.subarray(0, this.length)); + this.buffer = next; + } + this.buffer.set(bytes, this.length); + this.length = requiredBytes; + } + + take(current: Uint8Array): Uint8Array { + if (this.length === 0) return current; + + const line = new Uint8Array(this.length + current.byteLength); + line.set(this.buffer!.subarray(0, this.length)); + line.set(current, this.length); + this.clear(); + return line; + } + + isJsonWhitespaceLine(current: Uint8Array): boolean { + if (this.buffer) { + for (let index = 0; index < this.length; index++) { + if (!isJsonWhitespaceByte(this.buffer[index]!)) return false; + } + } + for (const byte of current) { + if (!isJsonWhitespaceByte(byte)) return false; + } + return true; + } + + clear(): void { + this.buffer = undefined; + this.length = 0; + } +} + +function isJsonWhitespaceByte(byte: number): boolean { + return byte === 0x20 || byte === 0x09 || byte === 0x0d; +} diff --git a/packages/acp-bridge/src/spawnChannel.test.ts b/packages/acp-bridge/src/spawnChannel.test.ts index d4f733d079..25be6fd2b1 100644 --- a/packages/acp-bridge/src/spawnChannel.test.ts +++ b/packages/acp-bridge/src/spawnChannel.test.ts @@ -50,6 +50,7 @@ vi.mock('node:child_process', async (importOriginal) => { }); import { + DAEMON_ACP_NDJSON_LIMITS, createSpawnChannelFactory, createStderrForwarder, getAcpMemoryArgs, @@ -227,6 +228,45 @@ describe('createSpawnChannelFactory env policy', () => { writer.releaseLock(); }); + it('terminates the tracked child when a bounded pipe fails', async () => { + const child = createFakeChildProcess(); + mockSpawn.mockReturnValue(child); + const factory = createSpawnChannelFactory({ + pipeLimits: { + maxFrameBytes: 16, + maxQueuedMessages: 2, + maxQueuedBytes: 32, + }, + }); + const channel = await factory('/tmp/project'); + const reader = channel.stream.readable.getReader(); + + (child.stdout as PassThrough).write('x'.repeat(17)); + + await expect(reader.closed).resolves.toBeUndefined(); + await vi.waitFor(() => expect(child.kill).toHaveBeenCalledWith('SIGTERM')); + reader.releaseLock(); + }); + + it('keeps the default factory unbounded and validates opt-in limits early', () => { + expect(DAEMON_ACP_NDJSON_LIMITS).toEqual({ + maxFrameBytes: 64 * 1024 * 1024, + maxQueuedMessages: 256, + maxQueuedBytes: 64 * 1024 * 1024, + }); + expect(() => createSpawnChannelFactory()).not.toThrow(); + expect(() => + createSpawnChannelFactory({ + pipeLimits: { + maxFrameBytes: 0, + maxQueuedMessages: 1, + maxQueuedBytes: 1, + }, + }), + ).toThrow('maxFrameBytes must be a positive safe integer'); + expect(mockSpawn).not.toHaveBeenCalled(); + }); + it('settles exited on an async spawn error only when no process exists', async () => { const child = createFakeChildProcess(); Object.defineProperty(child, 'pid', { value: undefined }); diff --git a/packages/acp-bridge/src/spawnChannel.ts b/packages/acp-bridge/src/spawnChannel.ts index 48846061aa..64be7b6675 100644 --- a/packages/acp-bridge/src/spawnChannel.ts +++ b/packages/acp-bridge/src/spawnChannel.ts @@ -10,13 +10,25 @@ import { Readable, Writable } from 'node:stream'; import { getHeapStatistics } from 'node:v8'; import type { ChannelFactory } from './channel.js'; import { redactLogCredentials } from './logRedaction.js'; -import { ndJsonStream, type NdJsonStreamHooks } from './ndJsonStream.js'; +import { + ndJsonStream, + type NdJsonStreamHooks, + type NdJsonStreamLimits, + validateNdJsonStreamLimits, +} from './ndJsonStream.js'; import { MissingCliEntryError } from './status.js'; import { EXTERNAL_TOOL_GUARD_TOKEN_ENV } from './externalToolGuard.js'; import { ProcessRegistry } from './process-registry.js'; import type { ChildHeapPolicy } from './child-heap-policy.js'; let cachedMemoryArgs: string[] | undefined; +export const DAEMON_ACP_NDJSON_LIMITS: Readonly = + Object.freeze({ + maxFrameBytes: 64 * 1024 * 1024, + maxQueuedMessages: 256, + maxQueuedBytes: 64 * 1024 * 1024, + }); + export function getAcpMemoryArgs(): string[] { if (cachedMemoryArgs) return cachedMemoryArgs; const constrainedMemory = (process as { constrainedMemory?: () => number }) @@ -107,6 +119,7 @@ export interface SpawnChannelFactoryOptions { onDiagnosticLine?: (line: string, level?: 'info' | 'warn' | 'error') => void; extraArgs?: string[]; pipeHooks?: NdJsonStreamHooks; + pipeLimits?: NdJsonStreamLimits; sourceEnv?: Readonly; processRegistry?: ProcessRegistry; /** @@ -134,6 +147,7 @@ export interface SpawnChannelFactoryOptions { export function createSpawnChannelFactory( options: SpawnChannelFactoryOptions = {}, ): ChannelFactory { + if (options.pipeLimits) validateNdJsonStreamLimits(options.pipeLimits); const processRegistry = options.processRegistry ?? new ProcessRegistry(); return async (workspaceCwd, childEnvOverrides) => { const sourceEnv = options.sourceEnv ?? process.env; @@ -218,7 +232,21 @@ export function createSpawnChannelFactory( const writable = Writable.toWeb(child.stdin) as WritableStream; const readable = Readable.toWeb(child.stdout) as ReadableStream; - const stream = ndJsonStream(writable, readable, options.pipeHooks); + const pipeHooks = options.pipeLimits + ? { + ...options.pipeHooks, + onTransportError: (error: unknown) => { + void trackedChild.terminate().catch(() => {}); + options.pipeHooks?.onTransportError?.(error); + }, + } + : options.pipeHooks; + const stream = ndJsonStream( + writable, + readable, + pipeHooks, + options.pipeLimits, + ); return { stream, diff --git a/packages/cli/src/serve/run-qwen-serve.test.ts b/packages/cli/src/serve/run-qwen-serve.test.ts index f44cafa41c..7fb0a3f595 100644 --- a/packages/cli/src/serve/run-qwen-serve.test.ts +++ b/packages/cli/src/serve/run-qwen-serve.test.ts @@ -1599,6 +1599,7 @@ describe('runQwenServe telemetry validation', () => { }); it('adds, advertises, and hot-removes a dynamic workspace runtime', async () => { + mockCreateSpawnChannelFactoryOptions.length = 0; tmpDir = fs.realpathSync( fs.mkdtempSync(path.join(os.tmpdir(), 'qws-hot-remove-')), ); @@ -1671,6 +1672,14 @@ describe('runQwenServe telemetry validation', () => { body: JSON.stringify({ cwd: secondary, persist: true }), }); expect(added.status).toBe(201); + expect(mockCreateSpawnChannelFactoryOptions).toHaveLength(2); + for (const options of mockCreateSpawnChannelFactoryOptions) { + expect(options['pipeLimits']).toEqual({ + maxFrameBytes: 64 * 1024 * 1024, + maxQueuedMessages: 256, + maxQueuedBytes: 64 * 1024 * 1024, + }); + } expect(createBridge.mock.calls[0]?.[0].onChannelDelivery).toBeTypeOf( 'function', ); @@ -1914,6 +1923,7 @@ describe('runQwenServe telemetry validation', () => { }); it('uses the daemon-wide policy and limits when constructing workspace bridges', async () => { + mockCreateSpawnChannelFactoryOptions.length = 0; tmpDir = fs.realpathSync(fs.mkdtempSync(path.join(os.tmpdir(), 'qws-ws-'))); const primary = path.join(tmpDir, 'primary'); const secondary = path.join(tmpDir, 'secondary'); @@ -1990,6 +2000,14 @@ describe('runQwenServe telemetry validation', () => { try { await handle.runtimeReady; expect(createBridge).toHaveBeenCalledTimes(2); + expect(mockCreateSpawnChannelFactoryOptions).toHaveLength(2); + for (const options of mockCreateSpawnChannelFactoryOptions) { + expect(options['pipeLimits']).toEqual({ + maxFrameBytes: 64 * 1024 * 1024, + maxQueuedMessages: 256, + maxQueuedBytes: 64 * 1024 * 1024, + }); + } expect(createBridge.mock.calls[0]?.[0]).toMatchObject({ compactedReplayMaxBytes: 1024, eventRingSize: 1234, diff --git a/packages/cli/src/serve/run-qwen-serve.ts b/packages/cli/src/serve/run-qwen-serve.ts index 148b4832bb..92858dd2f3 100644 --- a/packages/cli/src/serve/run-qwen-serve.ts +++ b/packages/cli/src/serve/run-qwen-serve.ts @@ -1171,6 +1171,7 @@ async function loadServeRuntimeModules() { resolveBridgeFsFactory: serverModule.resolveBridgeFsFactory, createAcpSessionBridge: bridgeModule.createAcpSessionBridge, createSpawnChannelFactory: spawnChannelModule.createSpawnChannelFactory, + daemonAcpNdJsonLimits: spawnChannelModule.DAEMON_ACP_NDJSON_LIMITS, ProcessRegistry: processRegistryModule.ProcessRegistry, createDaemonWorkspaceService: workspaceModule.createDaemonWorkspaceService, WorkspaceSettingsPartialPersistError: @@ -3731,6 +3732,7 @@ async function runQwenServeImpl( const channelFactory = runtime.createSpawnChannelFactory({ processRegistry, childHeapPolicy, + pipeLimits: runtime.daemonAcpNdJsonLimits, sourceEnv: runtimeEffectiveEnv, onDiagnosticLine: diagnosticSink, pipeHooks: { @@ -4437,6 +4439,7 @@ async function runQwenServeImpl( const secondaryChannelFactory = runtime.createSpawnChannelFactory({ processRegistry, childHeapPolicy, + pipeLimits: runtime.daemonAcpNdJsonLimits, sourceEnv: secondaryEnv.effectiveEnv, onDiagnosticLine: diagnosticSink, pipeHooks: { @@ -4978,6 +4981,7 @@ async function runQwenServeImpl( const wsChannelFactory = runtime.createSpawnChannelFactory({ processRegistry, childHeapPolicy, + pipeLimits: runtime.daemonAcpNdJsonLimits, sourceEnv: wsEnv.effectiveEnv, onDiagnosticLine: diagnosticSink, pipeHooks: {