feat(serve): bound daemon ACP NDJSON buffers (#8911)

Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com>

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.
 

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<Uint8Array> {
   });
 }
 
+function limits(
+  overrides: Partial<NdJsonStreamLimits> = {},
+): NdJsonStreamLimits {
+  return {
+    maxFrameBytes: 1024,
+    maxQueuedMessages: 4,
+    maxQueuedBytes: 4096,
+    ...overrides,
+  };
+}
+
 async function readAll(readable: ReadableStream<AnyMessage>) {
   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<Uint8Array>(),
+      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<Uint8Array>(),
+      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<Uint8Array>(),
+      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<Uint8Array>(),
+      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<Uint8Array>(),
+      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<Uint8Array>(),
+      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<Uint8Array>(),
+      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<Uint8Array>(),
+      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<Uint8Array>({
+        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<Uint8Array>(),
+      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<Uint8Array>({
+      start(controller) {
+        controller.enqueue(encoder.encode('{"partial":'));
+      },
+      cancel,
+    });
+    const stream = ndJsonStream(
+      new WritableStream<Uint8Array>(),
+      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<Uint8Array>(),
+      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' }),
+    );
+  });
 });

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<Uint8Array>,
   input: ReadableStream<Uint8Array>,
   hooks?: NdJsonStreamHooks,
+  limits?: NdJsonStreamLimits,
 ): Stream {
   const textEncoder = new TextEncoder();
   const textDecoder = new TextDecoder();
+  if (limits) validateNdJsonStreamLimits(limits);
 
-  const readable = new ReadableStream<AnyMessage>({
-    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<AnyMessage>({
     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<Uint8Array>,
+  textDecoder: TextDecoderLike,
+  hooks?: NdJsonStreamHooks,
+): ReadableStream<AnyMessage> {
+  return new ReadableStream<AnyMessage>({
+    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<Uint8Array>,
+  textDecoder: TextDecoderLike,
+  hooks: NdJsonStreamHooks | undefined,
+  limits: NdJsonStreamLimits,
+): ReadableStream<AnyMessage> {
+  const pending = new BoundedFrameBuffer(limits.maxFrameBytes);
+  const minimumQueueCharge = Math.ceil(
+    limits.maxQueuedBytes / limits.maxQueuedMessages,
+  );
+  let nextQueueCharge = minimumQueueCharge;
+  let reader: ReadableStreamDefaultReader<Uint8Array> | undefined;
+  let canceled = false;
+
+  return new ReadableStream<AnyMessage>(
+    {
+      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<Uint8Array>,
+  pending: BoundedFrameBuffer,
+  controller: ReadableStreamDefaultController<AnyMessage>,
+  textDecoder: TextDecoderLike,
+  hooks: NdJsonStreamHooks | undefined,
+  limits: NdJsonStreamLimits,
+  minimumQueueCharge: number,
+  setNextQueueCharge: (charge: number) => void,
+  isCanceled: () => boolean,
+): Promise<void> {
+  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<AnyMessage>,
@@ -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<AnyMessage>,
+  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<AnyMessage>,
   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<AnyMessage>,
+  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<Uint8Array>,
+  reason: unknown,
+): Promise<void> {
+  try {
+    await reader.cancel(reason);
+  } catch {
+    /* preserve the transport error that caused cancellation */
+  }
+}
+
 function callHook<T>(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;
+}

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 });

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<NdJsonStreamLimits> =
+  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<NodeJS.ProcessEnv>;
   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<Uint8Array>;
     const readable = Readable.toWeb(child.stdout) as ReadableStream<Uint8Array>;
-    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,

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,

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: {