vrr/qwen-code
2
0
Fork 0
mirror of https://github.com/QwenLM/qwen-code.git synced 2026-07-10 01:29:17 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 12

fix(serve): optimize daemon NDJSON stream handling (#6263)

Browse source 
Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com>
This commit is contained in:
jinye 2026-07-03 23:27:28 +08:00 • committed by GitHub
parent 68ff698cd2
commit 3911b1dc34
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
25 changed files with 1209 additions and 48 deletions
Download patch file Download diff file Expand all files Collapse all files

31
docs/developers/daemon/19-observability.md
View file

@ -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.

11
docs/developers/qwen-serve-protocol.md
View file

@ -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`,

4
packages/acp-bridge/package.json
View file

@ -63,6 +63,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"

1
packages/acp-bridge/src/index.ts
View file

@ -16,6 +16,7 @@ export * from './sessionArtifacts.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';

215
packages/acp-bridge/src/ndJsonStream.test.ts Normal file
View file

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

136
packages/acp-bridge/src/ndJsonStream.ts Normal file
View file

@ -0,0 +1,136 @@
/**
* @license
* Copyright 2026 Qwen Team
* SPDX-License-Identifier: Apache-2.0
*/
import type { AnyMessage, Stream } from '@agentclientprotocol/sdk';
export interface NdJsonStreamHooks {
onMessageReceived?: (bytes: number) => void;
onMessageSent?: (bytes: number) => void;
}
interface TextDecoderLike {
decode(input?: Uint8Array): string;
}
export function ndJsonStream(
output: WritableStream<Uint8Array>,
input: ReadableStream<Uint8Array>,
hooks?: NdJsonStreamHooks,
): Stream {
const textEncoder = new TextEncoder();
const textDecoder = new TextDecoder();
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 writable = new WritableStream<AnyMessage>({
async write(message) {
const content = JSON.stringify(message);
const payload = textEncoder.encode(content);
const frame = new Uint8Array(payload.byteLength + 1);
frame.set(payload);
frame[payload.byteLength] = 0x0a;
const writer = output.getWriter();
try {
await writer.write(frame);
callHook(hooks?.onMessageSent, payload.byteLength);
} finally {
writer.releaseLock();
}
},
});
return { readable, writable };
}
function readChunk(
chunk: Uint8Array,
pending: Uint8Array[],
controller: ReadableStreamDefaultController<AnyMessage>,
textDecoder: TextDecoderLike,
hooks?: NdJsonStreamHooks,
): void {
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);
start = newline + 1;
newline = chunk.indexOf(0x0a, start);
}
if (start < chunk.length) {
pending.push(chunk.subarray(start));
}
}
function takeLineBytes(pending: Uint8Array[], current: Uint8Array): Uint8Array {
if (pending.length === 0) return current;
const totalLength =
pending.reduce((sum, part) => sum + part.byteLength, 0) +
current.byteLength;
const line = new Uint8Array(totalLength);
let offset = 0;
for (const part of pending) {
line.set(part, offset);
offset += part.byteLength;
}
line.set(current, offset);
pending.length = 0;
return line;
}
function handleLine(
lineBytes: Uint8Array,
controller: ReadableStreamDefaultController<AnyMessage>,
textDecoder: TextDecoderLike,
hooks?: NdJsonStreamHooks,
): void {
const line = textDecoder.decode(lineBytes);
const trimmedLine = line.trim();
if (!trimmedLine) return;
try {
const message = JSON.parse(trimmedLine) as AnyMessage;
controller.enqueue(message);
callHook(hooks?.onMessageReceived, jsonPayloadByteLength(lineBytes));
} catch (err) {
// eslint-disable-next-line no-console -- match ACP SDK parse-error behavior
console.error('Failed to parse JSON message:', trimmedLine, err);
}
}
function jsonPayloadByteLength(lineBytes: Uint8Array): number {
return lineBytes[lineBytes.byteLength - 1] === 0x0d
? lineBytes.byteLength - 1
: lineBytes.byteLength;
}
function callHook(
hook: ((bytes: number) => void) | undefined,
bytes: number,
): void {
try {
hook?.(bytes);
} catch {
/* metrics hooks must not break the transport */
}
}

36
packages/acp-bridge/src/spawnChannel.test.ts
View file

@ -121,6 +121,42 @@ describe('createSpawnChannelFactory env policy', () => {
const args = mockSpawn.mock.calls[0]?.[1] as string[] | undefined;
expect(args?.slice(-2)).toEqual(['--acp', '--experimental-lsp']);
});
it('threads NDJSON pipe hooks through daemon-side spawned channels', async () => {
const child = createFakeChildProcess();
mockSpawn.mockReturnValue(child);
const onMessageSent = vi.fn();
const onMessageReceived = vi.fn();
const factory = createSpawnChannelFactory({
pipeHooks: { onMessageSent, onMessageReceived },
});
const channel = await factory('/tmp/project');
const writer = channel.stream.writable.getWriter();
const reader = channel.stream.readable.getReader();
await writer.write({ jsonrpc: '2.0', method: 'daemon-to-child' });
(child.stdout as PassThrough).write(
`${JSON.stringify({ jsonrpc: '2.0', method: 'child-to-daemon' })}\n`,
);
await expect(reader.read()).resolves.toMatchObject({
value: { jsonrpc: '2.0', method: 'child-to-daemon' },
done: false,
});
expect(onMessageSent).toHaveBeenCalledWith(
Buffer.byteLength(
JSON.stringify({ jsonrpc: '2.0', method: 'daemon-to-child' }),
),
);
expect(onMessageReceived).toHaveBeenCalledWith(
Buffer.byteLength(
JSON.stringify({ jsonrpc: '2.0', method: 'child-to-daemon' }),
),
);
reader.releaseLock();
writer.releaseLock();
});
});
describe('createStderrForwarder', () => {

5
packages/acp-bridge/src/spawnChannel.ts
View file

@ -8,9 +8,9 @@ import { spawn, type ChildProcess } from 'node:child_process';
import * as os from 'node:os';
import { Readable, Writable } from 'node:stream';
import { getHeapStatistics } from 'node:v8';
import { ndJsonStream } from '@agentclientprotocol/sdk';
import type { AcpChannelExitInfo, ChannelFactory } from './channel.js';
import { redactLogCredentials } from './logRedaction.js';
import { ndJsonStream, type NdJsonStreamHooks } from './ndJsonStream.js';
import { MissingCliEntryError } from './status.js';
let cachedMemoryArgs: string[] | undefined;
@ -103,6 +103,7 @@ export function createStderrForwarder(opts: StderrForwarderOptions): {
export interface SpawnChannelFactoryOptions {
onDiagnosticLine?: (line: string, level?: 'info' | 'warn' | 'error') => void;
extraArgs?: string[];
pipeHooks?: NdJsonStreamHooks;
}
/**
@ -189,7 +190,7 @@ 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);
const stream = ndJsonStream(writable, readable, options.pipeHooks);
return {
stream,

75
packages/cli/src/acp-integration/acpAgent.test.ts
View file

@ -60,7 +60,6 @@ vi.mock('@agentclientprotocol/sdk', () => ({
return mockConnectionState.promise;
},
})),
ndJsonStream: vi.fn().mockReturnValue({}),
RequestError: class RequestError extends Error {
code: number;
data: unknown;
@ -104,6 +103,10 @@ vi.mock('@agentclientprotocol/sdk', () => ({
PROTOCOL_VERSION: '1.0.0',
}));
vi.mock('@qwen-code/acp-bridge/ndJsonStream', () => ({
ndJsonStream: vi.fn().mockReturnValue({}),
}));
// Mock stream conversion
vi.mock('node:stream', async (importOriginal) => {
const actual = await importOriginal<typeof import('node:stream')>();
@ -122,6 +125,16 @@ vi.mock('@qwen-code/qwen-code-core', () => ({
warn: vi.fn(),
info: vi.fn(),
}),
registerAcpEventLoopLagGauge: vi.fn(),
startEventLoopLagMonitor: vi.fn(() => ({
snapshot: vi.fn(() => ({
meanMs: 0,
p50Ms: 0,
p99Ms: 0,
maxMs: 0,
})),
dispose: vi.fn(),
})),
APPROVAL_MODE_INFO: {},
APPROVAL_MODES: [],
AuthType: {
@ -601,6 +614,8 @@ import {
applyProviderInstallPlan,
Storage,
unregisterGoalHook,
startEventLoopLagMonitor,
registerAcpEventLoopLagGauge,
} from '@qwen-code/qwen-code-core';
import type { McpServer } from '@agentclientprotocol/sdk';
import { AgentSideConnection } from '@agentclientprotocol/sdk';
@ -792,6 +807,64 @@ describe('runAcpAgent shutdown cleanup', () => {
mockConnectionState.resolve();
await agentPromise;
});
it('registers and disposes the event loop monitor on normal connection close', async () => {
const snapshot = vi.fn(() => ({
meanMs: 0,
p50Ms: 0,
p99Ms: 0,
maxMs: 0,
}));
const dispose = vi.fn();
vi.mocked(startEventLoopLagMonitor).mockReturnValueOnce({
snapshot,
dispose,
});
const agentPromise = runAcpAgent(mockConfig, mockSettings, mockArgv);
await vi.waitFor(() => {
expect(registerAcpEventLoopLagGauge).toHaveBeenCalledWith(
expect.any(Function),
);
});
const readGauge = vi.mocked(registerAcpEventLoopLagGauge).mock.calls[0]![0];
expect(readGauge()).toEqual({
meanMs: 0,
p50Ms: 0,
p99Ms: 0,
maxMs: 0,
});
expect(snapshot).toHaveBeenCalledTimes(1);
mockConnectionState.resolve();
await agentPromise;
expect(dispose).toHaveBeenCalledTimes(1);
});
it('disposes the event loop monitor when connection setup fails', async () => {
const dispose = vi.fn();
vi.mocked(startEventLoopLagMonitor).mockReturnValueOnce({
snapshot: vi.fn(() => ({
meanMs: 0,
p50Ms: 0,
p99Ms: 0,
maxMs: 0,
})),
dispose,
});
vi.mocked(AgentSideConnection).mockImplementationOnce(() => {
throw new Error('connection setup failed');
});
await expect(
runAcpAgent(mockConfig, mockSettings, mockArgv),
).rejects.toThrow('connection setup failed');
expect(dispose).toHaveBeenCalledTimes(1);
expect(registerAcpEventLoopLagGauge).not.toHaveBeenCalled();
});
});
describe('runAcpAgent SessionEnd hooks', () => {

105
packages/cli/src/acp-integration/acpAgent.ts
View file

@ -64,6 +64,8 @@ import {
runManagedAutoMemoryDream,
runManagedRememberByAgent,
matchesAnyServerPattern,
registerAcpEventLoopLagGauge,
startEventLoopLagMonitor,
} from '@qwen-code/qwen-code-core';
import { randomUUID } from 'node:crypto';
import type {
@ -88,7 +90,6 @@ import type { JSONRPCMessage } from '@modelcontextprotocol/sdk/types.js';
import {
AgentSideConnection,
RequestError,
ndJsonStream,
PROTOCOL_VERSION,
} from '@agentclientprotocol/sdk';
import type { Content } from '@google/genai';
@ -129,6 +130,7 @@ import {
pickAuthMethodsForAuthRequired,
} from './authMethods.js';
import { AcpFileSystemService } from './service/filesystem.js';
import { ndJsonStream } from '@qwen-code/acp-bridge/ndJsonStream';
import { Readable, Writable } from 'node:stream';
import { normalizeDisabledToolList } from '../config/normalizeDisabledTools.js';
import { pipeline } from 'node:stream/promises';
@ -2315,23 +2317,35 @@ export async function runAcpAgent(
// client-hosted MCP server (#5626) round-trips over the parent WS.
sendSdkMcpMessage: bootstrapClientMcpSender,
});
const eventLoopMonitor = startEventLoopLagMonitor({
onNewMaxStall: (maxMs) => {
console.error(`[perf] acp agent event loop stall: max=${maxMs}ms`);
},
});
const stdout = Writable.toWeb(process.stdout) as WritableStream;
const stdin = Readable.toWeb(process.stdin) as ReadableStream<Uint8Array>;
// Stdout is used to send messages to the client, so console.log/console.info
// messages to stderr so that they don't interfere with ACP.
console.log = console.error;
console.info = console.error;
console.debug = console.error;
const stream = ndJsonStream(stdout, stdin);
let agentInstance: QwenAgent | undefined;
const connection = new AgentSideConnection((conn) => {
acpConnection = conn;
agentInstance = new QwenAgent(config, settings, argv, conn);
return agentInstance;
}, stream);
let connection: AgentSideConnection;
try {
const stdout = Writable.toWeb(process.stdout) as WritableStream;
const stdin = Readable.toWeb(process.stdin) as ReadableStream<Uint8Array>;
// Stdout is used to send messages to the client, so console.log/console.info
// messages to stderr so that they don't interfere with ACP.
console.log = console.error;
console.info = console.error;
console.debug = console.error;
const stream = ndJsonStream(stdout, stdin);
connection = new AgentSideConnection((conn) => {
acpConnection = conn;
agentInstance = new QwenAgent(config, settings, argv, conn);
return agentInstance;
}, stream);
} catch (err) {
eventLoopMonitor.dispose();
throw err;
}
registerAcpEventLoopLagGauge(() => eventLoopMonitor.snapshot());
// Both the SIGTERM handler and the IDE-initiated close path need
// to drain the MCP pool before runExitCleanup. Single helper
@ -2394,23 +2408,29 @@ export async function runAcpAgent(
shuttingDown = true;
debugLogger.debug('[ACP] Shutdown signal received, closing streams');
// Fire SessionEnd hook for all active sessions (aligned with core path)
await fireSessionEndOnce(SessionEndReason.Other);
agentInstance?.disposeSessions();
try {
// Fire SessionEnd hook for all active sessions (aligned with core path)
await fireSessionEndOnce(SessionEndReason.Other);
agentInstance?.disposeSessions();
try {
process.stdin.destroy();
} catch {
// stdin may already be closed
try {
process.stdin.destroy();
} catch {
// stdin may already be closed
}
try {
process.stdout.destroy();
} catch {
// stdout may already be closed
}
// Drain the workspace MCP pool BEFORE runExitCleanup so the
// descendant pid sweep can SIGTERM wrapper grandchildren.
await drainPoolBeforeExit('signal');
} catch (err) {
debugLogger.error('[ACP] Shutdown error:', err);
} finally {
eventLoopMonitor.dispose();
}
try {
process.stdout.destroy();
} catch {
// stdout may already be closed
}
// Drain the workspace MCP pool BEFORE runExitCleanup so the
// descendant pid sweep can SIGTERM wrapper grandchildren.
await drainPoolBeforeExit('signal');
// Clean up child processes (MCP servers, etc.) and force exit.
// Without this, orphan subprocesses keep the Node.js event loop alive
// and the CLI process never terminates after the IDE disconnects.
@ -2425,16 +2445,19 @@ export async function runAcpAgent(
process.on('SIGTERM', shutdownHandler);
process.on('SIGINT', shutdownHandler);
await connection.closed;
// Connection closed by IDE - fire SessionEnd hook (aligned with core path)
await fireSessionEndOnce(SessionEndReason.PromptInputExit);
// Mirror the SIGTERM handler's pool drain on the IDE-initiated
// normal close path to avoid leaking shared MCP entries.
await drainPoolBeforeExit('ide_close');
agentInstance?.disposeSessions();
process.off('SIGTERM', shutdownHandler);
process.off('SIGINT', shutdownHandler);
try {
await connection.closed;
// Connection closed by IDE - fire SessionEnd hook (aligned with core path)
await fireSessionEndOnce(SessionEndReason.PromptInputExit);
// Mirror the SIGTERM handler's pool drain on the IDE-initiated
// normal close path to avoid leaking shared MCP entries.
await drainPoolBeforeExit('ide_close');
agentInstance?.disposeSessions();
} finally {
process.off('SIGTERM', shutdownHandler);
process.off('SIGINT', shutdownHandler);
eventLoopMonitor.dispose();
}
}
export function toStdioServer(server: McpServer): McpServerStdio | undefined {

15
packages/cli/src/acp-integration/acpAgent.worktree.test.ts
View file

@ -55,7 +55,6 @@ vi.mock('@agentclientprotocol/sdk', () => ({
return mockConnectionState.promise;
},
})),
ndJsonStream: vi.fn().mockReturnValue({}),
RequestError: class RequestError extends Error {
static authRequired = vi
.fn()
@ -75,6 +74,10 @@ vi.mock('@agentclientprotocol/sdk', () => ({
PROTOCOL_VERSION: '1.0.0',
}));
vi.mock('@qwen-code/acp-bridge/ndJsonStream', () => ({
ndJsonStream: vi.fn().mockReturnValue({}),
}));
vi.mock('node:stream', async (importOriginal) => {
const actual = await importOriginal<typeof import('node:stream')>();
return {
@ -98,6 +101,16 @@ vi.mock('@qwen-code/qwen-code-core', () => ({
warn: vi.fn(),
info: vi.fn(),
}),
registerAcpEventLoopLagGauge: vi.fn(),
startEventLoopLagMonitor: vi.fn(() => ({
snapshot: vi.fn(() => ({
meanMs: 0,
p50Ms: 0,
p99Ms: 0,
maxMs: 0,
})),
dispose: vi.fn(),
})),
APPROVAL_MODE_INFO: {},
APPROVAL_MODES: [],
DEFAULT_STOP_HOOK_BLOCK_CAP: 8,

39
packages/cli/src/serve/daemon-status.test.ts
View file

@ -370,6 +370,35 @@ describe('buildDaemonStatusResponse', () => {
},
});
});
it('includes additive daemon performance data when provided', async () => {
const response = await buildDaemonStatusResponse(
'summary',
makeOptions({
perfSnapshot: {
eventLoop: { meanMs: 1, p50Ms: 2, p99Ms: 3, maxMs: 4 },
pipe: {
inbound: { count: 5, totalBytes: 600, maxBytes: 300 },
outbound: { count: 7, totalBytes: 800, maxBytes: 400 },
},
},
}),
);
expect(response.runtime.perf).toEqual({
eventLoop: { meanMs: 1, p50Ms: 2, p99Ms: 3, maxMs: 4 },
pipe: {
inbound: { count: 5, totalBytes: 600, maxBytes: 300 },
outbound: { count: 7, totalBytes: 800, maxBytes: 400 },
},
});
});
it('omits daemon performance data when no provider is injected', async () => {
const response = await buildDaemonStatusResponse('summary', makeOptions());
expect(response.runtime).not.toHaveProperty('perf');
});
});
interface MakeOptionsInput {
@ -382,6 +411,13 @@ interface MakeOptionsInput {
hooksStatus?: unknown;
extensionsStatus?: unknown;
channelWorkerSnapshot?: ChannelWorkerSnapshot;
perfSnapshot?: {
eventLoop: { meanMs: number; p50Ms: number; p99Ms: number; maxMs: number };
pipe: {
inbound: { count: number; totalBytes: number; maxBytes: number };
outbound: { count: number; totalBytes: number; maxBytes: number };
};
};
}
function makeOptions(input: MakeOptionsInput = {}): BuildDaemonStatusOptions {
@ -439,6 +475,9 @@ function makeOptions(input: MakeOptionsInput = {}): BuildDaemonStatusOptions {
...(input.channelWorkerSnapshot
? { getChannelWorkerSnapshot: () => input.channelWorkerSnapshot! }
: {}),
...(input.perfSnapshot
? { getPerfSnapshot: () => input.perfSnapshot! }
: {}),
};
}

22
packages/cli/src/serve/daemon-status.ts
View file

@ -94,6 +94,7 @@ export interface BuildDaemonStatusOptions {
sessionShellCommandEnabled: boolean;
startup?: DaemonStartupSnapshot;
getChannelWorkerSnapshot?: () => ChannelWorkerSnapshot;
getPerfSnapshot?: () => DaemonPerfSnapshot;
}
interface DaemonStatusSection<T> {
@ -168,9 +169,29 @@ interface DaemonStatusRuntime {
enabled: boolean;
rejectedSinceStart: Record<RateLimitTier, number>;
};
perf?: DaemonPerfSnapshot;
process: NodeJS.MemoryUsage;
}
export interface DaemonPipeStatsSnapshot {
count: number;
totalBytes: number;
maxBytes: number;
}
export interface DaemonPerfSnapshot {
eventLoop: {
meanMs: number;
p50Ms: number;
p99Ms: number;
maxMs: number;
};
pipe: {
inbound: DaemonPipeStatsSnapshot;
outbound: DaemonPipeStatsSnapshot;
};
}
export interface DaemonStatusResponse {
v: 1;
detail: DaemonStatusDetail;
@ -314,6 +335,7 @@ export async function buildDaemonStatusResponse(
enabled: input.opts.rateLimit === true,
rejectedSinceStart: rateLimitHits,
},
...(input.getPerfSnapshot ? { perf: input.getPerfSnapshot() } : {}),
process: process.memoryUsage(),
},
...(full ? { full } : {}),

3
packages/cli/src/serve/routes/daemon-status.ts
View file

@ -15,6 +15,7 @@ import type {
import type { DaemonLogger } from '../daemon-logger.js';
import {
buildDaemonStatusResponse,
type DaemonPerfSnapshot,
type DaemonStartupSnapshot,
parseDaemonStatusDetail,
} from '../daemon-status.js';
@ -42,6 +43,7 @@ interface RegisterDaemonStatusRoutesDeps {
deviceFlowRegistry: DeviceFlowRegistry;
sessionShellCommandEnabled: boolean;
getChannelWorkerSnapshot?: () => ChannelWorkerSnapshot;
getPerfSnapshot?: () => DaemonPerfSnapshot;
}
export function registerDaemonStatusRoutes(
@ -76,6 +78,7 @@ export function registerDaemonStatusRoutes(
deviceFlowRegistry: deps.deviceFlowRegistry,
sessionShellCommandEnabled: deps.sessionShellCommandEnabled,
getChannelWorkerSnapshot: deps.getChannelWorkerSnapshot,
getPerfSnapshot: deps.getPerfSnapshot,
}),
);
} catch (err) {

47
packages/cli/src/serve/run-qwen-serve.test.ts
View file

@ -2431,6 +2431,53 @@ describe('runQwenServe runtime startup failures', () => {
expect(bridge.shutdown).toHaveBeenCalledTimes(1);
expect(dispose).toHaveBeenCalledTimes(1);
});
it('disposes the daemon event loop monitor when closed after listening', async () => {
tmpDir = fs.realpathSync(
fs.mkdtempSync(path.join(os.tmpdir(), 'qws-runtime-monitor-close-')),
);
const bridge = {
spawnOrAttach: vi.fn(),
shutdown: vi.fn().mockResolvedValue(undefined),
killAllSync: vi.fn(),
getSession: vi.fn(),
getAllSessions: vi.fn().mockReturnValue([]),
publishWorkspaceEvent: vi.fn(),
getEventRing: vi.fn().mockReturnValue({ getAll: () => [] }),
resume: vi.fn(),
preheat: vi.fn().mockResolvedValue(undefined),
getDaemonStatusSnapshot: vi.fn().mockReturnValue(BASE_BRIDGE_SNAPSHOT),
} as unknown as HttpAcpBridge;
vi.spyOn(acpBridge, 'createAcpSessionBridge').mockReturnValue(
bridge as ReturnType<typeof acpBridge.createAcpSessionBridge>,
);
const dispose = vi.fn();
vi.spyOn(qwenCore, 'startEventLoopLagMonitor').mockReturnValueOnce({
snapshot: () => ({
meanMs: 0,
p50Ms: 0,
p99Ms: 0,
maxMs: 0,
}),
dispose,
});
const handle = await runQwenServe(
{
port: 0,
hostname: '127.0.0.1',
mode: 'http-bridge',
workspace: tmpDir,
maxSessions: 1,
serveWebShell: false,
},
{ resolveOnListen: true },
);
await handle.close();
expect(dispose).toHaveBeenCalledTimes(1);
});
});
describe('runQwenServe Web Shell signals on RunHandle', () => {

45
packages/cli/src/serve/run-qwen-serve.ts
View file

@ -77,6 +77,7 @@ import {
parseDaemonStatusDetail,
positiveFiniteOrNull,
type DaemonStatusIssue,
type DaemonPerfSnapshot,
type DaemonStartupSnapshot,
type DaemonStatusResponse,
} from './daemon-status.js';
@ -1858,6 +1859,9 @@ export async function runQwenServe(
let runtimeApp: Application | undefined;
let runtimeAppForCleanup: Application | undefined;
let bridgeRef: AcpSessionBridge | undefined = deps.bridge;
let daemonEventLoopMonitor:
| ReturnType<CoreRuntime['startEventLoopLagMonitor']>
| undefined;
let runtimeStartupError: string | undefined;
let runtimeStarting: Promise<void> | undefined;
let markRuntimeReady!: () => void;
@ -1872,6 +1876,10 @@ export async function runQwenServe(
markRuntimeFailed = reject;
});
void runtimeReady.catch(() => {});
const disposeDaemonEventLoopMonitor = (): void => {
daemonEventLoopMonitor?.dispose();
daemonEventLoopMonitor = undefined;
};
let channelWorker: ChannelWorkerSupervisor =
createDisabledChannelWorkerSupervisor();
const getChannelWorkerSnapshot = () => channelWorker.snapshot();
@ -1973,6 +1981,30 @@ export async function runQwenServe(
),
);
core.initializeDaemonMetrics();
daemonEventLoopMonitor?.dispose();
daemonEventLoopMonitor = core.startEventLoopLagMonitor({
onNewMaxStall: (maxMs) => {
daemonLog.warn('daemon event loop stall detected', { maxMs });
},
});
const currentDaemonEventLoopMonitor = daemonEventLoopMonitor;
core.registerDaemonEventLoopLagGauge(() =>
currentDaemonEventLoopMonitor.snapshot(),
);
const pipeStats: DaemonPerfSnapshot['pipe'] = {
inbound: { count: 0, totalBytes: 0, maxBytes: 0 },
outbound: { count: 0, totalBytes: 0, maxBytes: 0 },
};
const recordPipeMessage = (
direction: keyof DaemonPerfSnapshot['pipe'],
bytes: number,
): void => {
const stats = pipeStats[direction];
stats.count += 1;
stats.totalBytes += bytes;
stats.maxBytes = Math.max(stats.maxBytes, bytes);
core.recordDaemonPipeMessage(direction, bytes);
};
const daemonTelemetry = core.createDaemonBridgeTelemetry();
daemonTelemetry.metrics = {
sessionLifecycle(action) {
@ -2028,6 +2060,10 @@ export async function runQwenServe(
});
const channelFactory = runtime.createSpawnChannelFactory({
onDiagnosticLine: diagnosticSink,
pipeHooks: {
onMessageSent: (bytes) => recordPipeMessage('outbound', bytes),
onMessageReceived: (bytes) => recordPipeMessage('inbound', bytes),
},
...(opts.experimentalLsp === true
? { extraArgs: ['--experimental-lsp'] }
: {}),
@ -2206,6 +2242,13 @@ export async function runQwenServe(
fsFactory,
daemonLog,
getChannelWorkerSnapshot,
getPerfSnapshot: () => ({
eventLoop: currentDaemonEventLoopMonitor.snapshot(),
pipe: {
inbound: { ...pipeStats.inbound },
outbound: { ...pipeStats.outbound },
},
}),
workspace: workspaceService,
// Reverse tool channel (#5626): the SAME registry wired into `bridge` above,
// so the WS provider and the child-answering bridge share one sender map.
@ -2576,6 +2619,7 @@ export async function runQwenServe(
writeStderrLine(`qwen serve: runtime startup failed: ${message}`);
daemonLog.error('runtime startup failed', error);
await stopChannelWorkerAfterFailedStartup();
disposeDaemonEventLoopMonitor();
removeCurrentServePidfile();
await shutdownBridgeAfterFailedStartup(bridgeForCleanup ?? bridgeRef);
markRuntimeFailed(error);
@ -2884,6 +2928,7 @@ export async function runQwenServe(
rl.setDraining(true);
rl.dispose();
}
disposeDaemonEventLoopMonitor();
// The worker owns daemon-backed sessions; disconnect it before
// tearing down the ACP bridge it is attached to.
await channelWorker.stop().catch((err) => {

7
packages/cli/src/serve/server.ts
View file

@ -7,7 +7,10 @@
import express from 'express';
import type { Application } from 'express';
import type { DaemonLogger } from './daemon-logger.js';
import type { DaemonStartupSnapshot } from './daemon-status.js';
import type {
DaemonPerfSnapshot,
DaemonStartupSnapshot,
} from './daemon-status.js';
import type { ChannelWorkerSnapshot } from './channel-worker-supervisor.js';
import {
allowOriginCors,
@ -210,6 +213,7 @@ export interface ServeAppDeps {
daemonLog?: DaemonLogger;
startup?: DaemonStartupSnapshot;
getChannelWorkerSnapshot?: () => ChannelWorkerSnapshot;
getPerfSnapshot?: () => DaemonPerfSnapshot;
workspace?: DaemonWorkspaceService;
statusProvider?: DaemonStatusProvider;
persistDisabledTools?: (
@ -563,6 +567,7 @@ export function createServeApp(
deviceFlowRegistry,
sessionShellCommandEnabled,
getChannelWorkerSnapshot: deps.getChannelWorkerSnapshot,
getPerfSnapshot: deps.getPerfSnapshot,
});
registerCapabilitiesRoutes(app, {

4
packages/cli/vitest.config.ts
View file

@ -41,6 +41,10 @@ export default defineConfig({
__dirname,
'../acp-bridge/src/spawnChannel.ts',
),
'@qwen-code/acp-bridge/ndJsonStream': path.resolve(
__dirname,
'../acp-bridge/src/ndJsonStream.ts',
),
'@qwen-code/acp-bridge/logRedaction': path.resolve(
__dirname,
'../acp-bridge/src/logRedaction.ts',

18
packages/core/src/telemetry/daemon-metrics.test.ts
View file

@ -81,6 +81,7 @@ describe('Daemon Metrics', () => {
let recordDaemonPromptDuration: typeof import('./daemon-metrics.js').recordDaemonPromptDuration;
let recordDaemonBridgeError: typeof import('./daemon-metrics.js').recordDaemonBridgeError;
let recordDaemonCancel: typeof import('./daemon-metrics.js').recordDaemonCancel;
let recordDaemonPipeMessage: typeof import('./daemon-metrics.js').recordDaemonPipeMessage;
beforeEach(async () => {
vi.resetModules();
@ -100,6 +101,7 @@ describe('Daemon Metrics', () => {
recordDaemonPromptDuration = mod.recordDaemonPromptDuration;
recordDaemonBridgeError = mod.recordDaemonBridgeError;
recordDaemonCancel = mod.recordDaemonCancel;
recordDaemonPipeMessage = mod.recordDaemonPipeMessage;
});
describe('initializeDaemonMetrics', () => {
@ -138,6 +140,10 @@ describe('Daemon Metrics', () => {
'qwen-code.daemon.cancel.count',
expect.any(Object),
);
expect(mockCreateHistogramFn).toHaveBeenCalledWith(
'qwen-code.daemon.pipe.message_bytes',
expect.objectContaining({ unit: 'By' }),
);
});
it('does not re-initialize on second call', () => {
@ -153,6 +159,7 @@ describe('Daemon Metrics', () => {
recordDaemonHttpRequest(100, 'POST /session/:id/prompt', 200);
recordDaemonSessionLifecycle('spawn');
recordDaemonCancel();
recordDaemonPipeMessage('inbound', 1024);
expect(mockCounterAddFn).not.toHaveBeenCalled();
expect(mockHistogramRecordFn).not.toHaveBeenCalled();
});
@ -279,6 +286,17 @@ describe('Daemon Metrics', () => {
});
});
describe('recordDaemonPipeMessage', () => {
it('records pipe message bytes with direction', () => {
initializeDaemonMetrics();
mockHistogramRecordFn.mockClear();
recordDaemonPipeMessage('outbound', 2048);
expect(mockHistogramRecordFn).toHaveBeenCalledWith(2048, {
direction: 'outbound',
});
});
});
describe('registerDaemonGaugeCallbacks', () => {
it('creates 3 observable gauges', () => {
registerDaemonGaugeCallbacks({

24
packages/core/src/telemetry/daemon-metrics.ts
View file

@ -18,6 +18,7 @@ const DAEMON_PROMPT_QUEUE_WAIT = `${SERVICE_NAME}.daemon.prompt.queue_wait`;
const DAEMON_PROMPT_DURATION = `${SERVICE_NAME}.daemon.prompt.duration`;
const DAEMON_BRIDGE_ERROR_COUNT = `${SERVICE_NAME}.daemon.bridge.error.count`;
const DAEMON_CANCEL_COUNT = `${SERVICE_NAME}.daemon.cancel.count`;
const DAEMON_PIPE_MESSAGE_BYTES = `${SERVICE_NAME}.daemon.pipe.message_bytes`;
const DAEMON_SSE_ACTIVE = `${SERVICE_NAME}.daemon.sse.active`;
const DAEMON_PROCESS_HEAP_USED = `${SERVICE_NAME}.daemon.process.heap_used`;
@ -53,6 +54,7 @@ let promptQueueWaitHistogram: Histogram | undefined;
let promptDurationHistogram: Histogram | undefined;
let bridgeErrorCounter: Counter | undefined;
let cancelCounter: Counter | undefined;
let pipeMessageBytesHistogram: Histogram | undefined;
function normalizeErrorType(err: unknown): string {
const name = err instanceof Error ? err.name : typeof err;
@ -125,6 +127,18 @@ export function initializeDaemonMetrics(): void {
valueType: ValueType.INT,
});
pipeMessageBytesHistogram = meter.createHistogram(DAEMON_PIPE_MESSAGE_BYTES, {
description: 'Daemon ACP child pipe message payload size in bytes.',
unit: 'By',
valueType: ValueType.INT,
advice: {
explicitBucketBoundaries: [
256, 1024, 4096, 16_384, 65_536, 262_144, 1_048_576, 4_194_304,
16_777_216,
],
},
});
initialized = true;
}
@ -234,3 +248,13 @@ export function recordDaemonCancel(): void {
if (!initialized) return;
cancelCounter?.add(1);
}
export type DaemonPipeDirection = 'inbound' | 'outbound';
export function recordDaemonPipeMessage(
direction: DaemonPipeDirection,
bytes: number,
): void {
if (!initialized) return;
pipeMessageBytesHistogram?.record(bytes, { direction });
}

126
packages/core/src/telemetry/event-loop-lag-metrics.test.ts Normal file
View file

@ -0,0 +1,126 @@
/**
* @license
* Copyright 2026 Qwen Team
* SPDX-License-Identifier: Apache-2.0
*/
import { beforeEach, describe, expect, it, vi, type Mock } from 'vitest';
import type {
Meter,
ObservableGauge,
ObservableResult,
} from '@opentelemetry/api';
const gaugeCallbacks: Array<(result: ObservableResult) => void> = [];
const mockObservableGaugeAddCallback = vi.fn(
(cb: (result: ObservableResult) => void) => {
gaugeCallbacks.push(cb);
},
);
const mockCreateObservableGaugeFn: Mock<
(name: string, options?: unknown) => ObservableGauge
> = vi.fn().mockReturnValue({
addCallback: mockObservableGaugeAddCallback,
});
const mockMeterInstance: Meter = {
createObservableGauge: mockCreateObservableGaugeFn,
} as Partial<Meter> as Meter;
vi.mock('@opentelemetry/api', () => ({
metrics: {
getMeter: vi.fn().mockReturnValue(mockMeterInstance),
},
ValueType: {
DOUBLE: 2,
},
diag: {
setLogger: vi.fn(),
warn: vi.fn(),
error: vi.fn(),
info: vi.fn(),
debug: vi.fn(),
},
}));
describe('event loop lag metrics', () => {
let registerDaemonEventLoopLagGauge: typeof import('./event-loop-lag-metrics.js').registerDaemonEventLoopLagGauge;
let registerAcpEventLoopLagGauge: typeof import('./event-loop-lag-metrics.js').registerAcpEventLoopLagGauge;
beforeEach(async () => {
vi.resetModules();
vi.clearAllMocks();
gaugeCallbacks.length = 0;
const mod = await import('./event-loop-lag-metrics.js');
registerDaemonEventLoopLagGauge = mod.registerDaemonEventLoopLagGauge;
registerAcpEventLoopLagGauge = mod.registerAcpEventLoopLagGauge;
});
it('registers daemon and ACP event loop lag gauges', () => {
registerDaemonEventLoopLagGauge(() => ({
meanMs: 1,
p50Ms: 2,
p99Ms: 3,
maxMs: 4,
}));
registerAcpEventLoopLagGauge(() => ({
meanMs: 5,
p50Ms: 6,
p99Ms: 7,
maxMs: 8,
}));
expect(mockCreateObservableGaugeFn).toHaveBeenCalledWith(
'qwen-code.daemon.event_loop.lag',
expect.objectContaining({ unit: 'ms' }),
);
expect(mockCreateObservableGaugeFn).toHaveBeenCalledWith(
'qwen-code.acp.event_loop.lag',
expect.objectContaining({ unit: 'ms' }),
);
});
it('observes all four stats with stat attributes', () => {
registerDaemonEventLoopLagGauge(() => ({
meanMs: 1,
p50Ms: 2,
p99Ms: 3,
maxMs: 4,
}));
const result = { observe: vi.fn() };
gaugeCallbacks[0]!(result as unknown as ObservableResult);
expect(result.observe).toHaveBeenCalledWith(1, { stat: 'mean' });
expect(result.observe).toHaveBeenCalledWith(2, { stat: 'p50' });
expect(result.observe).toHaveBeenCalledWith(3, { stat: 'p99' });
expect(result.observe).toHaveBeenCalledWith(4, { stat: 'max' });
});
it('does not re-register the same process role twice', () => {
registerAcpEventLoopLagGauge(() => ({
meanMs: 1,
p50Ms: 1,
p99Ms: 1,
maxMs: 1,
}));
registerAcpEventLoopLagGauge(() => ({
meanMs: 2,
p50Ms: 2,
p99Ms: 2,
maxMs: 2,
}));
expect(mockCreateObservableGaugeFn).toHaveBeenCalledTimes(1);
});
it('swallows snapshot errors from observable callbacks', () => {
registerDaemonEventLoopLagGauge(() => {
throw new Error('snapshot failed');
});
expect(() =>
gaugeCallbacks[0]!({ observe: vi.fn() } as unknown as ObservableResult),
).not.toThrow();
});
});

60
packages/core/src/telemetry/event-loop-lag-metrics.ts Normal file
View file

@ -0,0 +1,60 @@
/**
* @license
* Copyright 2026 Qwen Team
* SPDX-License-Identifier: Apache-2.0
*/
import { ValueType } from '@opentelemetry/api';
import { SERVICE_NAME } from './constants.js';
import { getMeter } from './metrics.js';
import type { EventLoopLagSnapshot } from './event-loop-lag.js';
const DAEMON_EVENT_LOOP_LAG = `${SERVICE_NAME}.daemon.event_loop.lag`;
const ACP_EVENT_LOOP_LAG = `${SERVICE_NAME}.acp.event_loop.lag`;
let daemonGaugeRegistered = false;
let acpGaugeRegistered = false;
export function registerDaemonEventLoopLagGauge(
read: () => EventLoopLagSnapshot,
): void {
if (daemonGaugeRegistered) return;
daemonGaugeRegistered = registerEventLoopLagGauge(
DAEMON_EVENT_LOOP_LAG,
read,
);
}
export function registerAcpEventLoopLagGauge(
read: () => EventLoopLagSnapshot,
): void {
if (acpGaugeRegistered) return;
acpGaugeRegistered = registerEventLoopLagGauge(ACP_EVENT_LOOP_LAG, read);
}
function registerEventLoopLagGauge(
name: string,
read: () => EventLoopLagSnapshot,
): boolean {
const meter = getMeter();
if (!meter) return false;
meter
.createObservableGauge(name, {
description: 'Event loop lag in milliseconds.',
unit: 'ms',
valueType: ValueType.DOUBLE,
})
.addCallback((result) => {
try {
const snapshot = read();
result.observe(snapshot.meanMs, { stat: 'mean' });
result.observe(snapshot.p50Ms, { stat: 'p50' });
result.observe(snapshot.p99Ms, { stat: 'p99' });
result.observe(snapshot.maxMs, { stat: 'max' });
} catch {
/* no-op */
}
});
return true;
}

118
packages/core/src/telemetry/event-loop-lag.test.ts Normal file
View file

@ -0,0 +1,118 @@
/**
* @license
* Copyright 2026 Qwen Team
* SPDX-License-Identifier: Apache-2.0
*/
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
const histogram = vi.hoisted(() => ({
enable: vi.fn(),
disable: vi.fn(),
mean: Number.NaN,
max: Number.NaN,
percentile: vi.fn((_percentile: number) => Number.NaN),
}));
vi.mock('node:perf_hooks', () => ({
monitorEventLoopDelay: vi.fn(() => histogram),
}));
describe('startEventLoopLagMonitor', () => {
let startEventLoopLagMonitor: typeof import('./event-loop-lag.js').startEventLoopLagMonitor;
beforeEach(async () => {
vi.resetModules();
vi.clearAllMocks();
vi.useFakeTimers();
histogram.mean = Number.NaN;
histogram.max = Number.NaN;
histogram.percentile.mockReturnValue(Number.NaN);
({ startEventLoopLagMonitor } = await import('./event-loop-lag.js'));
});
afterEach(() => {
vi.useRealTimers();
});
it('returns finite zeroes before the monitor has samples', () => {
const monitor = startEventLoopLagMonitor({ resolutionMs: 10 });
expect(monitor.snapshot()).toEqual({
meanMs: 0,
p50Ms: 0,
p99Ms: 0,
maxMs: 0,
});
monitor.dispose();
});
it('converts nanosecond histogram values to milliseconds', () => {
histogram.mean = 12_000_000;
histogram.max = 50_000_000;
histogram.percentile.mockImplementation((percentile: number) =>
percentile === 50 ? 20_000_000 : 45_000_000,
);
const monitor = startEventLoopLagMonitor({ resolutionMs: 10 });
expect(monitor.snapshot()).toEqual({
meanMs: 12,
p50Ms: 20,
p99Ms: 45,
maxMs: 50,
});
monitor.dispose();
});
it('actively reports only new max stalls above threshold', async () => {
const onNewMaxStall = vi.fn();
histogram.max = 15_000_000;
const monitor = startEventLoopLagMonitor({
resolutionMs: 10,
stallThresholdMs: 10,
onNewMaxStall,
});
await vi.advanceTimersByTimeAsync(10);
histogram.max = 12_000_000;
await vi.advanceTimersByTimeAsync(10);
histogram.max = 20_000_000;
await vi.advanceTimersByTimeAsync(10);
expect(onNewMaxStall).toHaveBeenCalledTimes(2);
expect(onNewMaxStall).toHaveBeenNthCalledWith(1, 15);
expect(onNewMaxStall).toHaveBeenNthCalledWith(2, 20);
monitor.dispose();
histogram.max = 30_000_000;
await vi.advanceTimersByTimeAsync(10);
expect(onNewMaxStall).toHaveBeenCalledTimes(2);
});
it('swallows stall callback errors', async () => {
const onNewMaxStall = vi.fn(() => {
throw new Error('callback failed');
});
histogram.max = 15_000_000;
const monitor = startEventLoopLagMonitor({
resolutionMs: 10,
stallThresholdMs: 10,
onNewMaxStall,
});
await vi.advanceTimersByTimeAsync(10);
expect(onNewMaxStall).toHaveBeenCalledWith(15);
monitor.dispose();
});
it('enables and disables the underlying histogram', () => {
const monitor = startEventLoopLagMonitor();
expect(histogram.enable).toHaveBeenCalledTimes(1);
monitor.dispose();
expect(histogram.disable).toHaveBeenCalledTimes(1);
});
});

94
packages/core/src/telemetry/event-loop-lag.ts Normal file
View file

@ -0,0 +1,94 @@
/**
* @license
* Copyright 2026 Qwen Team
* SPDX-License-Identifier: Apache-2.0
*/
import { monitorEventLoopDelay } from 'node:perf_hooks';
export interface EventLoopLagSnapshot {
meanMs: number;
p50Ms: number;
p99Ms: number;
maxMs: number;
}
export interface EventLoopLagMonitor {
snapshot(): EventLoopLagSnapshot;
dispose(): void;
}
export interface EventLoopLagMonitorOptions {
resolutionMs?: number;
stallThresholdMs?: number;
onNewMaxStall?: (maxMs: number) => void;
}
const DEFAULT_RESOLUTION_MS = 20;
const DEFAULT_STALL_THRESHOLD_MS = 1_000;
const NS_PER_MS = 1_000_000;
export function startEventLoopLagMonitor(
options: EventLoopLagMonitorOptions = {},
): EventLoopLagMonitor {
const resolutionMs = positiveFiniteOrDefault(
options.resolutionMs,
DEFAULT_RESOLUTION_MS,
);
const stallThresholdMs = positiveFiniteOrDefault(
options.stallThresholdMs,
DEFAULT_STALL_THRESHOLD_MS,
);
const histogram = monitorEventLoopDelay({ resolution: resolutionMs });
histogram.enable();
let disposed = false;
let lastReportedMaxMs = 0;
const readMaxMs = () => nsToMs(histogram.max);
const checkForNewMaxStall = () => {
if (disposed || !options.onNewMaxStall) return;
const maxMs = readMaxMs();
if (maxMs >= stallThresholdMs && maxMs > lastReportedMaxMs) {
lastReportedMaxMs = maxMs;
try {
options.onNewMaxStall(maxMs);
} catch {
/* event loop monitoring must not break the process */
}
}
};
const interval =
options.onNewMaxStall !== undefined
? setInterval(checkForNewMaxStall, resolutionMs)
: undefined;
interval?.unref();
return {
snapshot(): EventLoopLagSnapshot {
return {
meanMs: nsToMs(histogram.mean),
p50Ms: nsToMs(histogram.percentile(50)),
p99Ms: nsToMs(histogram.percentile(99)),
maxMs: readMaxMs(),
};
},
dispose(): void {
if (disposed) return;
disposed = true;
if (interval) {
clearInterval(interval);
}
histogram.disable();
},
};
}
function nsToMs(value: number): number {
return Number.isFinite(value) ? value / NS_PER_MS : 0;
}
function positiveFiniteOrDefault(value: number | undefined, fallback: number) {
return value !== undefined && Number.isFinite(value) && value > 0
? value
: fallback;
}

16
packages/core/src/telemetry/index.ts
View file

@ -210,8 +210,22 @@ export {
recordDaemonPromptDuration,
recordDaemonBridgeError,
recordDaemonCancel,
recordDaemonPipeMessage,
} from './daemon-metrics.js';
export type { DaemonGaugeCallbacks } from './daemon-metrics.js';
export type {
DaemonGaugeCallbacks,
DaemonPipeDirection,
} from './daemon-metrics.js';
export {
startEventLoopLagMonitor,
type EventLoopLagMonitor,
type EventLoopLagMonitorOptions,
type EventLoopLagSnapshot,
} from './event-loop-lag.js';
export {
registerDaemonEventLoopLagGauge,
registerAcpEventLoopLagGauge,
} from './event-loop-lag-metrics.js';
export {
addUserPromptAttributes,
addSystemPromptAttributes,