fix(serve): Optimize daemon NDJSON stream handling

docs/developers/daemon/19-observability.md

@@ -10,6 +10,7 @@
 | ------------------------------------------- | ---------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
 | `QWEN_SERVE_DEBUG` stderr logs              | `bridge.ts` and call sites                     | Env values `1` / `true` / `on` / `yes` (case-insensitive) print `qwen serve debug: ...` lines to stderr.                                                                                                                                                                                  |
 | OpenTelemetry span instrumentation          | `server.ts` `daemonTelemetryMiddleware`        | Each HTTP request is wrapped in `withDaemonRequestSpan`; attributes include route, sessionId, clientId, and status code. Permission routes have dedicated spans. Prompt lifecycle is traced end-to-end. Configuration lives in `settings.json` `telemetry`.                               |
+| OpenTelemetry daemon perf metrics           | `telemetry/*event-loop-lag*`, `daemon-metrics` | Event loop lag gauges for daemon and ACP child processes, plus daemon-child pipe message byte histograms.                                                                                                                                                                                 |
 | `DaemonLogger` structured file logs         | `serve/daemon-logger.ts`                       | Structured JSON-like log lines are written to a file. Boot prints `daemon log -> <path>`. Supports `info` / `warn` / `error` levels, with structured fields such as `route`, `sessionId`, `clientId`, `childPid`, and `channelId`.                                                        |
 | Per-request access-log middleware           | `server.ts`, registered before `bearerAuth`    | Logs `method`, `path`, `status`, `durationMs`, `sessionId`, and `clientId` after each request. Skips `GET /health` and heartbeat. 4xx+ uses `warn`; success uses `info`.                                                                                                                  |
 | `/health`                                   | `server.ts` route                              | Liveness probe; `?deep=1` returns extended details.                                                                                                                                                                                                                                       |
@@ -25,7 +26,7 @@
 
 ## What does not exist today
 
-- **No Prometheus / metrics endpoint.** There is no `process_cpu_seconds_total`, `http_requests_total`, or `event_bus_queue_depth`.
+- **No Prometheus / metrics endpoint.** OTel metrics can be exported, but the daemon does not expose a Prometheus scrape endpoint.
 - **No external audit sink for `PermissionAuditRing`.** The ring exists, but fan-out hooks to SIEM or external storage are not wired.
 
 ## Debugging recipes
@@ -91,6 +92,32 @@ The first signal triggers graceful shutdown (see [`02-serve-runtime.md`](./02-se
 
 A **second** SIGTERM/SIGINT intentionally triggers `bridge.killAllSync()` + `process.exit(1)`.
 
+### 9. Is the daemon event loop or ACP pipe overloaded?
+
+`GET /daemon/status` may include `runtime.perf` when the production daemon runtime injects the perf snapshot provider:
+
+```json
+{
+  "runtime": {
+    "perf": {
+      "eventLoop": { "meanMs": 1.2, "p50Ms": 1.0, "p99Ms": 9.5, "maxMs": 25 },
+      "pipe": {
+        "inbound": { "count": 42, "totalBytes": 100000, "maxBytes": 12000 },
+        "outbound": { "count": 41, "totalBytes": 90000, "maxBytes": 11000 }
+      }
+    }
+  }
+}
+```
+
+The status payload is daemon-only. ACP child event loop lag is intentionally not aggregated into `/daemon/status`; it is visible through OTel gauge `qwen-code.acp.event_loop.lag` and through stderr stall lines forwarded into daemon logs.
+
+New OTel metric names:
+
+- `qwen-code.daemon.event_loop.lag`, gauge in milliseconds with `stat=mean|p50|p99|max`.
+- `qwen-code.acp.event_loop.lag`, gauge in milliseconds with `stat=mean|p50|p99|max`.
+- `qwen-code.daemon.pipe.message_bytes`, histogram in bytes with `direction=inbound|outbound`.
+
 ## Flow
 
 ### Typical triage flow
@@ -119,6 +146,7 @@ flowchart TD
 - `process.stderr.write` for debug stderr.
 - `DaemonLogger` for structured file logs.
 - OpenTelemetry SDK through `initializeTelemetry` and `createDaemonBridgeTelemetry`.
+- `node:perf_hooks.monitorEventLoopDelay` for daemon and ACP event loop lag gauges.
 - `node:process` for env and signal inspection.
 
 ## Configuration
@@ -135,6 +163,7 @@ flowchart TD
 
 - **DaemonLogger file logs are structured** and can be filtered by `route`, `sessionId`, and `clientId`. `QWEN_SERVE_DEBUG` stderr logs remain unstructured text.
 - **OpenTelemetry spans include per-request correlation.** Each HTTP request span carries route, sessionId, and clientId attributes that can be joined in a tracing backend.
+- **`runtime.perf` is daemon-only.** Child event loop lag is not reported there by design; use OTel or forwarded stderr stall warnings for ACP child stalls.
 - **ACP-level `/workspace/preflight` cells require a live session.** On an idle daemon, auth / MCP / skills / providers may show `status: 'not_started'`; this is expected.
 - **`/workspace/env` only reports secret presence, not values.** Do not expose the response where the mere presence of a secret is sensitive.
 - **The audit ring is process-local** and history is lost on daemon restart.

docs/developers/qwen-serve-protocol.md

@@ -319,11 +319,22 @@ Response shape:
         "wsStreams": 0,
         "pendingClientRequests": 0
       }
+    },
+    "perf": {
+      "eventLoop": { "meanMs": 0, "p50Ms": 0, "p99Ms": 0, "maxMs": 0 },
+      "pipe": {
+        "inbound": { "count": 0, "totalBytes": 0, "maxBytes": 0 },
+        "outbound": { "count": 0, "totalBytes": 0, "maxBytes": 0 }
+      }
     }
   }
 }
 ```
 
+`runtime.perf` is optional. When present, it reports daemon-process event loop
+lag and daemon-child pipe byte counters only; ACP child event loop lag is not
+included in `/daemon/status`.
+
 `status` is `error` if any issue has error severity, `warning` if any issue has
 warning severity, otherwise `ok`. Issue codes are stable and include
 `session_capacity_high`, `connection_capacity_high`, `pending_permissions`,

packages/acp-bridge/package.json

@@ -59,6 +59,10 @@
       "types": "./dist/spawnChannel.d.ts",
       "import": "./dist/spawnChannel.js"
     },
+    "./ndJsonStream": {
+      "types": "./dist/ndJsonStream.d.ts",
+      "import": "./dist/ndJsonStream.js"
+    },
     "./logRedaction": {
       "types": "./dist/logRedaction.d.ts",
       "import": "./dist/logRedaction.js"

packages/acp-bridge/src/index.ts

@@ -15,6 +15,7 @@ export * from './bridgeErrors.js';
 export * from './bridgeTypes.js';
 export * from './bridgeOptions.js';
 export * from './spawnChannel.js';
+export * from './ndJsonStream.js';
 export * from './bridgeClient.js';
 export * from './bridge.js';
 export * from './bridgeFileSystem.js';

packages/acp-bridge/src/ndJsonStream.test.ts

@@ -0,0 +1,215 @@
+/**
+ * @license
+ * Copyright 2026 Qwen Team
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+import { describe, expect, it, vi } from 'vitest';
+import type { AnyMessage } from '@agentclientprotocol/sdk';
+import { ndJsonStream } from './ndJsonStream.js';
+
+const encoder = new TextEncoder();
+
+function message(method: string, params: Record<string, unknown> = {}) {
+  return { jsonrpc: '2.0', method, params } satisfies AnyMessage;
+}
+
+function byteStream(chunks: readonly Uint8Array[]): ReadableStream<Uint8Array> {
+  return new ReadableStream<Uint8Array>({
+    start(controller) {
+      for (const chunk of chunks) {
+        controller.enqueue(chunk);
+      }
+      controller.close();
+    },
+  });
+}
+
+async function readAll(readable: ReadableStream<AnyMessage>) {
+  const reader = readable.getReader();
+  const out: AnyMessage[] = [];
+  try {
+    while (true) {
+      const { value, done } = await reader.read();
+      if (done) break;
+      out.push(value);
+    }
+  } finally {
+    reader.releaseLock();
+  }
+  return out;
+}
+
+async function writeOne(
+  writable: WritableStream<AnyMessage>,
+  msg: AnyMessage,
+): Promise<void> {
+  const writer = writable.getWriter();
+  try {
+    await writer.write(msg);
+  } finally {
+    writer.releaseLock();
+  }
+}
+
+describe('ndJsonStream', () => {
+  it('round-trips one message', async () => {
+    const sent = message('hello', { n: 1 });
+    const line = `${JSON.stringify(sent)}\n`;
+    const stream = ndJsonStream(
+      new WritableStream<Uint8Array>(),
+      byteStream([encoder.encode(line)]),
+    );
+
+    await expect(readAll(stream.readable)).resolves.toEqual([sent]);
+  });
+
+  it('parses multiple messages from one chunk', async () => {
+    const first = message('first');
+    const second = message('second', { ok: true });
+    const stream = ndJsonStream(
+      new WritableStream<Uint8Array>(),
+      byteStream([
+        encoder.encode(`${JSON.stringify(first)}\n${JSON.stringify(second)}\n`),
+      ]),
+    );
+
+    await expect(readAll(stream.readable)).resolves.toEqual([first, second]);
+  });
+
+  it('parses a large message split across many chunks', async () => {
+    const sent = message('large', { text: 'x'.repeat(1024 * 1024) });
+    const bytes = encoder.encode(`${JSON.stringify(sent)}\n`);
+    const chunks: Uint8Array[] = [];
+    for (let offset = 0; offset < bytes.length; offset += 64 * 1024) {
+      chunks.push(bytes.slice(offset, offset + 64 * 1024));
+    }
+    const stream = ndJsonStream(
+      new WritableStream<Uint8Array>(),
+      byteStream(chunks),
+    );
+
+    await expect(readAll(stream.readable)).resolves.toEqual([sent]);
+  });
+
+  it('preserves multibyte UTF-8 characters across chunk boundaries', async () => {
+    const sent = message('unicode', { text: 'a中b' });
+    const bytes = encoder.encode(`${JSON.stringify(sent)}\n`);
+    const split = bytes.indexOf(encoder.encode('中')[1]!);
+    const stream = ndJsonStream(
+      new WritableStream<Uint8Array>(),
+      byteStream([bytes.slice(0, split), bytes.slice(split)]),
+    );
+
+    await expect(readAll(stream.readable)).resolves.toEqual([sent]);
+  });
+
+  it('skips empty and CRLF lines', async () => {
+    const sent = message('crlf');
+    const stream = ndJsonStream(
+      new WritableStream<Uint8Array>(),
+      byteStream([encoder.encode(`\n\r\n${JSON.stringify(sent)}\r\n`)]),
+    );
+
+    await expect(readAll(stream.readable)).resolves.toEqual([sent]);
+  });
+
+  it('logs invalid JSON and continues with later messages', async () => {
+    const stderr = vi.spyOn(console, 'error').mockImplementation(() => {});
+    const sent = message('after-error');
+    const stream = ndJsonStream(
+      new WritableStream<Uint8Array>(),
+      byteStream([encoder.encode(`{bad json}\n${JSON.stringify(sent)}\n`)]),
+    );
+
+    await expect(readAll(stream.readable)).resolves.toEqual([sent]);
+    expect(stderr).toHaveBeenCalledWith(
+      'Failed to parse JSON message:',
+      '{bad json}',
+      expect.any(SyntaxError),
+    );
+    stderr.mockRestore();
+  });
+
+  it('drops an unterminated final line at EOF', async () => {
+    const complete = message('complete');
+    const partial = message('partial');
+    const stream = ndJsonStream(
+      new WritableStream<Uint8Array>(),
+      byteStream([
+        encoder.encode(
+          `${JSON.stringify(complete)}\n${JSON.stringify(partial)}`,
+        ),
+      ]),
+    );
+
+    await expect(readAll(stream.readable)).resolves.toEqual([complete]);
+  });
+
+  it('reports received and sent payload byte counts without newlines', async () => {
+    const received = message('received');
+    const sent = message('sent', { value: 'ok' });
+    const receivedBytes = encoder.encode(JSON.stringify(received)).byteLength;
+    const sentBytes = encoder.encode(JSON.stringify(sent)).byteLength;
+    const onMessageReceived = vi.fn();
+    const onMessageSent = vi.fn();
+    const outputChunks: Uint8Array[] = [];
+    const stream = ndJsonStream(
+      new WritableStream<Uint8Array>({
+        write(chunk) {
+          outputChunks.push(chunk);
+        },
+      }),
+      byteStream([encoder.encode(`${JSON.stringify(received)}\r\n`)]),
+      { onMessageReceived, onMessageSent },
+    );
+
+    await expect(readAll(stream.readable)).resolves.toEqual([received]);
+    await writeOne(stream.writable, sent);
+
+    expect(onMessageReceived).toHaveBeenCalledWith(receivedBytes);
+    expect(onMessageSent).toHaveBeenCalledWith(sentBytes);
+    expect(new TextDecoder().decode(outputChunks[0])).toBe(
+      `${JSON.stringify(sent)}\n`,
+    );
+  });
+
+  it('does not let hook errors break transport', async () => {
+    const received = message('received');
+    const sent = message('sent');
+    const stream = ndJsonStream(
+      new WritableStream<Uint8Array>(),
+      byteStream([encoder.encode(`${JSON.stringify(received)}\n`)]),
+      {
+        onMessageReceived: () => {
+          throw new Error('received hook failed');
+        },
+        onMessageSent: () => {
+          throw new Error('sent hook failed');
+        },
+      },
+    );
+
+    await expect(readAll(stream.readable)).resolves.toEqual([received]);
+    await expect(writeOne(stream.writable, sent)).resolves.toBeUndefined();
+  });
+
+  it('propagates output write errors without reporting sent bytes', async () => {
+    const sent = message('write-error');
+    const onMessageSent = vi.fn();
+    const stream = ndJsonStream(
+      new WritableStream<Uint8Array>({
+        write() {
+          throw new Error('output closed');
+        },
+      }),
+      byteStream([]),
+      { onMessageSent },
+    );
+
+    await expect(writeOne(stream.writable, sent)).rejects.toThrow(
+      'output closed',
+    );
+    expect(onMessageSent).not.toHaveBeenCalled();
+  });
+});