diff --git a/docs/developers/daemon/19-observability.md b/docs/developers/daemon/19-observability.md index b1c8a01b865..701af2a1b24 100644 --- a/docs/developers/daemon/19-observability.md +++ b/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 -> `. 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. diff --git a/docs/developers/qwen-serve-protocol.md b/docs/developers/qwen-serve-protocol.md index 3623dceef73..cb55073f79c 100644 --- a/docs/developers/qwen-serve-protocol.md +++ b/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`, diff --git a/packages/acp-bridge/package.json b/packages/acp-bridge/package.json index e9b28d8bb75..1e1f717e344 100644 --- a/packages/acp-bridge/package.json +++ b/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" diff --git a/packages/acp-bridge/src/index.ts b/packages/acp-bridge/src/index.ts index 316d85e5f28..1bd598d4c37 100644 --- a/packages/acp-bridge/src/index.ts +++ b/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'; diff --git a/packages/acp-bridge/src/ndJsonStream.test.ts b/packages/acp-bridge/src/ndJsonStream.test.ts new file mode 100644 index 00000000000..c9c40f93449 --- /dev/null +++ b/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 = {}) { + return { jsonrpc: '2.0', method, params } satisfies AnyMessage; +} + +function byteStream(chunks: readonly Uint8Array[]): ReadableStream { + return new ReadableStream({ + start(controller) { + for (const chunk of chunks) { + controller.enqueue(chunk); + } + controller.close(); + }, + }); +} + +async function readAll(readable: ReadableStream) { + 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, + msg: AnyMessage, +): Promise { + 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(), + 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(), + 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(), + 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(), + 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(), + 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(), + 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(), + 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({ + 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(), + 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({ + write() { + throw new Error('output closed'); + }, + }), + byteStream([]), + { onMessageSent }, + ); + + await expect(writeOne(stream.writable, sent)).rejects.toThrow( + 'output closed', + ); + expect(onMessageSent).not.toHaveBeenCalled(); + }); +}); diff --git a/packages/acp-bridge/src/ndJsonStream.ts b/packages/acp-bridge/src/ndJsonStream.ts new file mode 100644 index 00000000000..bbb1864b3ba --- /dev/null +++ b/packages/acp-bridge/src/ndJsonStream.ts @@ -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, + input: ReadableStream, + hooks?: NdJsonStreamHooks, +): Stream { + const textEncoder = new TextEncoder(); + const textDecoder = new TextDecoder(); + + const readable = new ReadableStream({ + async start(controller) { + const pending: Uint8Array[] = []; + const reader = input.getReader(); + try { + while (true) { + const { value, done } = await reader.read(); + if (done) break; + if (!value) continue; + readChunk(value, pending, controller, textDecoder, hooks); + } + } finally { + reader.releaseLock(); + controller.close(); + } + }, + }); + + const writable = new WritableStream({ + 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, + 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, + 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 */ + } +} diff --git a/packages/acp-bridge/src/spawnChannel.test.ts b/packages/acp-bridge/src/spawnChannel.test.ts index 6be22c75cf4..178b3b0245f 100644 --- a/packages/acp-bridge/src/spawnChannel.test.ts +++ b/packages/acp-bridge/src/spawnChannel.test.ts @@ -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', () => { diff --git a/packages/acp-bridge/src/spawnChannel.ts b/packages/acp-bridge/src/spawnChannel.ts index b88d076ceb9..f22b0ff209a 100644 --- a/packages/acp-bridge/src/spawnChannel.ts +++ b/packages/acp-bridge/src/spawnChannel.ts @@ -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; const readable = Readable.toWeb(child.stdout) as ReadableStream; - const stream = ndJsonStream(writable, readable); + const stream = ndJsonStream(writable, readable, options.pipeHooks); return { stream, diff --git a/packages/cli/src/acp-integration/acpAgent.test.ts b/packages/cli/src/acp-integration/acpAgent.test.ts index 4d949093554..51ad826bd32 100644 --- a/packages/cli/src/acp-integration/acpAgent.test.ts +++ b/packages/cli/src/acp-integration/acpAgent.test.ts @@ -58,7 +58,6 @@ vi.mock('@agentclientprotocol/sdk', () => ({ return mockConnectionState.promise; }, })), - ndJsonStream: vi.fn().mockReturnValue({}), RequestError: class RequestError extends Error { code: number; data: unknown; @@ -102,6 +101,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(); @@ -120,6 +123,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: { @@ -598,6 +611,8 @@ import { applyProviderInstallPlan, Storage, unregisterGoalHook, + startEventLoopLagMonitor, + registerAcpEventLoopLagGauge, } from '@qwen-code/qwen-code-core'; import type { McpServer } from '@agentclientprotocol/sdk'; import { AgentSideConnection } from '@agentclientprotocol/sdk'; @@ -789,6 +804,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', () => { diff --git a/packages/cli/src/acp-integration/acpAgent.ts b/packages/cli/src/acp-integration/acpAgent.ts index 59a2ddecf26..881cef06dd2 100644 --- a/packages/cli/src/acp-integration/acpAgent.ts +++ b/packages/cli/src/acp-integration/acpAgent.ts @@ -63,6 +63,8 @@ import { FORK_SUBAGENT_TYPE, runManagedRememberByAgent, matchesAnyServerPattern, + registerAcpEventLoopLagGauge, + startEventLoopLagMonitor, } from '@qwen-code/qwen-code-core'; import { randomUUID } from 'node:crypto'; import type { @@ -87,7 +89,6 @@ import type { JSONRPCMessage } from '@modelcontextprotocol/sdk/types.js'; import { AgentSideConnection, RequestError, - ndJsonStream, PROTOCOL_VERSION, } from '@agentclientprotocol/sdk'; import type { Content } from '@google/genai'; @@ -128,6 +129,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'; @@ -2298,23 +2300,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; - - // 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; + + // 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 @@ -2377,23 +2391,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 { - process.stdin.destroy(); - } catch { - // stdin may already be closed - } try { - process.stdout.destroy(); - } catch { - // stdout may already be closed + // 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.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(); } - // 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. @@ -2408,16 +2428,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 { diff --git a/packages/cli/src/acp-integration/acpAgent.worktree.test.ts b/packages/cli/src/acp-integration/acpAgent.worktree.test.ts index c7424816300..04818a8201f 100644 --- a/packages/cli/src/acp-integration/acpAgent.worktree.test.ts +++ b/packages/cli/src/acp-integration/acpAgent.worktree.test.ts @@ -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(); 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, diff --git a/packages/cli/src/serve/daemon-status.test.ts b/packages/cli/src/serve/daemon-status.test.ts index f6fbe3eb970..90fcf793657 100644 --- a/packages/cli/src/serve/daemon-status.test.ts +++ b/packages/cli/src/serve/daemon-status.test.ts @@ -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! } + : {}), }; } diff --git a/packages/cli/src/serve/daemon-status.ts b/packages/cli/src/serve/daemon-status.ts index 098940afb8d..b7098e7a74f 100644 --- a/packages/cli/src/serve/daemon-status.ts +++ b/packages/cli/src/serve/daemon-status.ts @@ -94,6 +94,7 @@ export interface BuildDaemonStatusOptions { sessionShellCommandEnabled: boolean; startup?: DaemonStartupSnapshot; getChannelWorkerSnapshot?: () => ChannelWorkerSnapshot; + getPerfSnapshot?: () => DaemonPerfSnapshot; } interface DaemonStatusSection { @@ -168,9 +169,29 @@ interface DaemonStatusRuntime { enabled: boolean; rejectedSinceStart: Record; }; + 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 } : {}), diff --git a/packages/cli/src/serve/routes/daemon-status.ts b/packages/cli/src/serve/routes/daemon-status.ts index de115e4edb1..a355891e7b5 100644 --- a/packages/cli/src/serve/routes/daemon-status.ts +++ b/packages/cli/src/serve/routes/daemon-status.ts @@ -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) { diff --git a/packages/cli/src/serve/run-qwen-serve.test.ts b/packages/cli/src/serve/run-qwen-serve.test.ts index aeb98f6fa57..cbd66c053f6 100644 --- a/packages/cli/src/serve/run-qwen-serve.test.ts +++ b/packages/cli/src/serve/run-qwen-serve.test.ts @@ -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, + ); + 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', () => { diff --git a/packages/cli/src/serve/run-qwen-serve.ts b/packages/cli/src/serve/run-qwen-serve.ts index 1c529294ce2..4a8510bdfa3 100644 --- a/packages/cli/src/serve/run-qwen-serve.ts +++ b/packages/cli/src/serve/run-qwen-serve.ts @@ -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 + | undefined; let runtimeStartupError: string | undefined; let runtimeStarting: Promise | 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) => { diff --git a/packages/cli/src/serve/server.ts b/packages/cli/src/serve/server.ts index 058dae08fcd..c346e1e7ee0 100644 --- a/packages/cli/src/serve/server.ts +++ b/packages/cli/src/serve/server.ts @@ -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, { diff --git a/packages/cli/vitest.config.ts b/packages/cli/vitest.config.ts index 806344672fa..fe51cd9b18c 100644 --- a/packages/cli/vitest.config.ts +++ b/packages/cli/vitest.config.ts @@ -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', diff --git a/packages/core/src/telemetry/daemon-metrics.test.ts b/packages/core/src/telemetry/daemon-metrics.test.ts index 0a40e168b74..cfb9f0f7b52 100644 --- a/packages/core/src/telemetry/daemon-metrics.test.ts +++ b/packages/core/src/telemetry/daemon-metrics.test.ts @@ -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({ diff --git a/packages/core/src/telemetry/daemon-metrics.ts b/packages/core/src/telemetry/daemon-metrics.ts index e69a0467699..a7828b30145 100644 --- a/packages/core/src/telemetry/daemon-metrics.ts +++ b/packages/core/src/telemetry/daemon-metrics.ts @@ -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 }); +} diff --git a/packages/core/src/telemetry/event-loop-lag-metrics.test.ts b/packages/core/src/telemetry/event-loop-lag-metrics.test.ts new file mode 100644 index 00000000000..85768e30ff9 --- /dev/null +++ b/packages/core/src/telemetry/event-loop-lag-metrics.test.ts @@ -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 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(); + }); +}); diff --git a/packages/core/src/telemetry/event-loop-lag-metrics.ts b/packages/core/src/telemetry/event-loop-lag-metrics.ts new file mode 100644 index 00000000000..4bd7bb06323 --- /dev/null +++ b/packages/core/src/telemetry/event-loop-lag-metrics.ts @@ -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; +} diff --git a/packages/core/src/telemetry/event-loop-lag.test.ts b/packages/core/src/telemetry/event-loop-lag.test.ts new file mode 100644 index 00000000000..5e04f2d0bc1 --- /dev/null +++ b/packages/core/src/telemetry/event-loop-lag.test.ts @@ -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); + }); +}); diff --git a/packages/core/src/telemetry/event-loop-lag.ts b/packages/core/src/telemetry/event-loop-lag.ts new file mode 100644 index 00000000000..2aacdaf84cf --- /dev/null +++ b/packages/core/src/telemetry/event-loop-lag.ts @@ -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; +} diff --git a/packages/core/src/telemetry/index.ts b/packages/core/src/telemetry/index.ts index 9e93aed66c7..c3ed4046fe8 100644 --- a/packages/core/src/telemetry/index.ts +++ b/packages/core/src/telemetry/index.ts @@ -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,