feat(serve): bound daemon ACP NDJSON buffers

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' }),
+    );
+  });
 });